refactor(webapp): move reconciliation logic to separate file for better testability

Author: bharathkumar39293

apps/webapp/app/presenters/v3/RunPresenter.server.ts

@@ -124,7 +124,6 @@ export class RunPresenter {
       isFinished: isFinalRunStatus(run.status),
       startedAt: run.startedAt,
       completedAt: run.completedAt,
-      createdAt: run.createdAt,
       logsDeletedAt: showDeletedLogs ? null : run.logsDeletedAt,
       rootTaskRun: run.rootTaskRun,
       parentTaskRun: run.parentTaskRun,
@@ -209,41 +208,16 @@ export class RunPresenter {
 
     //we need the start offset for each item, and the total duration of the entire tree
     const treeRootStartTimeMs = tree ? tree?.data.startTime.getTime() : 0;
-
-    const postgresRunDuration =
-      runData.isFinished && run.completedAt
-        ? millisecondsToNanoseconds(
-          run.completedAt.getTime() -
-          (run.rootTaskRun?.createdAt ?? run.createdAt).getTime()
-        )
-        : 0;
-
     let totalDuration = tree?.data.duration ?? 0;
     const events = tree
-      ? flattenTree(tree).map((n, index) => {
-        const isRoot = index === 0;
-        const offset = millisecondsToNanoseconds(n.data.startTime.getTime() - treeRootStartTimeMs);
-
-        let nIsPartial = n.data.isPartial;
-        let nDuration = n.data.duration;
-        let nIsError = n.data.isError;
-
-        // NOTE: Clickhouse trace ingestion is eventually consistent.
-        // When a run is marked finished in Postgres, we reconcile the
-        // root span to reflect completion even if telemetry is still partial.
-        // This is a deliberate UI-layer tradeoff to prevent stale or "stuck"
-        // run states in the dashboard.
-        if (isRoot && runData.isFinished && nIsPartial) {
-          nIsPartial = false;
-          nDuration = Math.max(nDuration ?? 0, postgresRunDuration);
-          nIsError = isFailedRunStatus(runData.status);
-        }
-
+      ? flattenTree(tree).map((n) => {
+        const offset = millisecondsToNanoseconds(
+          n.data.startTime.getTime() - treeRootStartTimeMs
+        );
         //only let non-debug events extend the total duration
         if (!n.data.isDebug) {
-          totalDuration = Math.max(totalDuration, offset + (nIsPartial ? 0 : nDuration));
+          totalDuration = Math.max(totalDuration, offset + n.data.duration);
         }
-
         return {
           ...n,
           data: {
@@ -254,24 +228,23 @@ export class RunPresenter {
               treeRootStartTimeMs
             ),
             //set partial nodes to null duration
-            duration: nIsPartial ? null : nDuration,
-            isPartial: nIsPartial,
-            isError: nIsError,
+            duration: n.data.isPartial ? null : n.data.duration,
             offset,
-            isRoot,
+            isRoot: n.id === traceSummary.rootSpan.id,
           },
         };
       })
       : [];
 
-    if (runData.isFinished) {
-      totalDuration = Math.max(totalDuration, postgresRunDuration);
-    }
-
     //total duration should be a minimum of 1ms
     totalDuration = Math.max(totalDuration, millisecondsToNanoseconds(1));
 
-    const reconciled = reconcileTraceWithRunLifecycle(runData, traceSummary.rootSpan.id, events, totalDuration);
+    const reconciled = reconcileTraceWithRunLifecycle(
+      runData,
+      traceSummary.rootSpan.id,
+      events,
+      totalDuration
+    );
 
     return {
       run: runData,
@@ -312,17 +285,14 @@ export function reconcileTraceWithRunLifecycle(
   totalDuration: number;
   rootSpanStatus: "executing" | "completed" | "failed";
 } {
-  const rootEvent = events[0];
-  const isActualRoot = rootEvent?.id === rootSpanId;
-
-  const currentStatus: "executing" | "completed" | "failed" =
-    isActualRoot && rootEvent
-      ? rootEvent.data.isError
-        ? "failed"
-        : !rootEvent.data.isPartial
-          ? "completed"
-          : "executing"
-      : "executing";
+  const rootEvent = events.find((e) => e.id === rootSpanId);
+  const currentStatus: "executing" | "completed" | "failed" = rootEvent
+    ? rootEvent.data.isError
+      ? "failed"
+      : !rootEvent.data.isPartial
+        ? "completed"
+        : "executing"
+    : "executing";
 
   if (!runData.isFinished) {
     return { events, totalDuration, rootSpanStatus: currentStatus };
@@ -337,28 +307,23 @@ export function reconcileTraceWithRunLifecycle(
 
   const updatedTotalDuration = Math.max(totalDuration, postgresRunDuration);
 
-  // We only need to potentially update the root event (the first one) if it matches our ID
-  if (isActualRoot && rootEvent && rootEvent.data.isPartial) {
-    const updatedEvents = [...events];
-    updatedEvents[0] = {
-      ...rootEvent,
-      data: {
-        ...rootEvent.data,
-        isPartial: false,
-        duration: Math.max(rootEvent.data.duration ?? 0, postgresRunDuration),
-        isError: isFailedRunStatus(runData.status),
-      },
-    };
-
-    return {
-      events: updatedEvents,
-      totalDuration: updatedTotalDuration,
-      rootSpanStatus: isFailedRunStatus(runData.status) ? "failed" : "completed",
-    };
-  }
+  const updatedEvents = events.map((e) => {
+    if (e.id === rootSpanId && e.data.isPartial) {
+      return {
+        ...e,
+        data: {
+          ...e.data,
+          isPartial: false,
+          duration: Math.max(e.data.duration ?? 0, postgresRunDuration),
+          isError: isFailedRunStatus(runData.status),
+        },
+      };
+    }
+    return e;
+  });
 
   return {
-    events,
+    events: updatedEvents,
     totalDuration: updatedTotalDuration,
     rootSpanStatus: isFailedRunStatus(runData.status) ? "failed" : "completed",
   };

apps/webapp/app/presenters/v3/reconcileTrace.server.ts

@@ -0,0 +1,89 @@
+import { millisecondsToNanoseconds } from "@trigger.dev/core/v3";
+import { isFailedRunStatus } from "~/v3/taskStatus";
+import type { TaskRunStatus } from "@trigger.dev/database";
+
+export type ReconcileRunData = {
+    isFinished: boolean;
+    status: TaskRunStatus;
+    createdAt: Date;
+    completedAt: Date | null;
+    rootTaskRun: { createdAt: Date } | null;
+};
+
+export type ReconcileEvent = {
+    id: string;
+    data: {
+        isPartial: boolean;
+        isError: boolean;
+        duration?: number | null;
+    };
+};
+
+export type ReconcileResult = {
+    events: any[];
+    totalDuration: number;
+    rootSpanStatus: "executing" | "completed" | "failed";
+};
+
+// NOTE: Clickhouse trace ingestion is eventually consistent.
+// When a run is marked finished in Postgres, we reconcile the
+// root span to reflect completion even if telemetry is still partial.
+// This is a deliberate UI-layer tradeoff to prevent stale or "stuck"
+// run states in the dashboard.
+export function reconcileTraceWithRunLifecycle(
+    runData: ReconcileRunData,
+    rootSpanId: string,
+    events: any[],
+    totalDuration: number
+): ReconcileResult {
+    const rootEvent = events[0];
+    const isActualRoot = rootEvent?.id === rootSpanId;
+
+    const currentStatus: "executing" | "completed" | "failed" =
+        isActualRoot && rootEvent
+            ? rootEvent.data.isError
+                ? "failed"
+                : !rootEvent.data.isPartial
+                    ? "completed"
+                    : "executing"
+            : "executing";
+
+    if (!runData.isFinished) {
+        return { events, totalDuration, rootSpanStatus: currentStatus };
+    }
+
+    const postgresRunDuration = runData.completedAt
+        ? millisecondsToNanoseconds(
+            runData.completedAt.getTime() -
+            (runData.rootTaskRun?.createdAt ?? runData.createdAt).getTime()
+        )
+        : 0;
+
+    const updatedTotalDuration = Math.max(totalDuration, postgresRunDuration);
+
+    // We only need to potentially update the root event (the first one) if it matches our ID
+    if (isActualRoot && rootEvent && rootEvent.data.isPartial) {
+        const updatedEvents = [...events];
+        updatedEvents[0] = {
+            ...rootEvent,
+            data: {
+                ...rootEvent.data,
+                isPartial: false,
+                duration: Math.max(rootEvent.data.duration ?? 0, postgresRunDuration),
+                isError: isFailedRunStatus(runData.status),
+            },
+        };
+
+        return {
+            events: updatedEvents,
+            totalDuration: updatedTotalDuration,
+            rootSpanStatus: isFailedRunStatus(runData.status) ? "failed" : "completed",
+        };
+    }
+
+    return {
+        events,
+        totalDuration: updatedTotalDuration,
+        rootSpanStatus: isFailedRunStatus(runData.status) ? "failed" : "completed",
+    };
+}