fix(logs): stop loading execution data and snapshots that reads do not use (#9349)

* fix(logs): stop loading execution data and snapshots that reads do not use

- Status polls select execution_data through a CASE only for the statuses the poll
  reads it for (failed; completed when output is requested; always for selected
  outputs), from one shared status rule.
- The execution snapshot view no longer loads execution_data while resolving its
  context; it is read only after authorization, so a refused caller never fetches it.
- The paused execution list projects execution_snapshot->'triggerIds' instead of
  loading whole rows with the serialized executor snapshot.
- One shared terminal-status predicate replaces the copies in the HITL manager and
  run-file reads.

* test(logs): give the recording client a connection option so its options type-check

* fix(logs): read a snapshot view's execution data with the timing and cost it pairs with

Author: waleedlatif1

apps/sim/lib/logs/application/read-execution-snapshot.test.ts

@@ -66,6 +66,13 @@ const workspaceContext = {
   billedAccountUserId: 'billing-owner-1',
 }
 
+const completedRun = {
+  executionData: null,
+  endedAt: new Date('2026-08-05T12:00:01.000Z'),
+  totalDurationMs: 1000,
+  costTotal: '0.75',
+}
+
 const workflowRecord = {
   id: 'log-1',
   workflowId: 'workflow-1',
@@ -77,17 +84,19 @@ const workflowRecord = {
   endedAt: new Date('2026-08-05T12:00:01.000Z'),
   totalDurationMs: 1000,
   costTotal: '0.75',
-  executionData: null,
 }
 
 /**
  * Answers `db.select(...)` calls in order. The snapshot read walks the workflow
- * log, then (only when that missed) the job log, then the state snapshot.
+ * log, then (only when that missed) the job log, then the state snapshot, then
+ * the run's execution data.
  */
 function queueSelects(...results: unknown[][]): void {
   for (const rows of results) {
     mocks.select.mockReturnValueOnce({
-      from: () => ({ where: () => ({ limit: () => Promise.resolve(rows) }) }),
+      from: () => ({
+        where: () => Object.assign(Promise.resolve(rows), { limit: () => Promise.resolve(rows) }),
+      }),
     })
   }
 }
@@ -135,7 +144,11 @@ describe('readExecutionSnapshot spend projection', () => {
       ...DEFAULT_PERMISSION_GROUP_CONFIG,
       hideCostInfo: true,
     })
-    queueSelects([workflowRecord], [{ id: 'snapshot-1', stateData: { blocks: {} } }])
+    queueSelects(
+      [workflowRecord],
+      [{ id: 'snapshot-1', stateData: { blocks: {} } }],
+      [completedRun]
+    )
 
     const result = await read(principal, 'run-1')
 
@@ -153,11 +166,31 @@ describe('readExecutionSnapshot spend projection', () => {
       ...DEFAULT_PERMISSION_GROUP_CONFIG,
       hideCostInfo: true,
     })
-    queueSelects([workflowRecord], [{ id: 'snapshot-1', stateData: { blocks: {} } }])
+    queueSelects(
+      [workflowRecord],
+      [{ id: 'snapshot-1', stateData: { blocks: {} } }],
+      [completedRun]
+    )
 
     const result = await read(executorPrincipal, 'run-1')
 
     expect(result.executionMetadata.cost).toEqual({ total: 0.75 })
     expect(permissionGroupScopeMockFns.mockResolvePermissionGroupConfig).not.toHaveBeenCalled()
   })
+
+  it('reports the timing and cost read with the execution data, not the earlier context', async () => {
+    queueSelects(
+      [{ ...workflowRecord, endedAt: null, totalDurationMs: null, costTotal: null }],
+      [{ id: 'snapshot-1', stateData: { blocks: {} } }],
+      [completedRun]
+    )
+
+    const result = await read(executorPrincipal, 'run-1')
+
+    expect(result.executionMetadata).toMatchObject({
+      endedAt: '2026-08-05T12:00:01.000Z',
+      totalDurationMs: 1000,
+      cost: { total: 0.75 },
+    })
+  })
 })

apps/sim/lib/logs/application/read-execution-snapshot.ts

@@ -23,7 +23,7 @@ import {
   resolveActiveWorkspaceApplicationContext,
 } from '@/lib/workspaces/application/context'
 
-const CHILD_SNAPSHOT_QUERY_CHUNK_SIZE = 1_000
+const SNAPSHOT_QUERY_CHUNK_SIZE = 1_000
 
 interface WorkflowExecutionRecord {
   kind: 'workflow'
@@ -37,7 +37,6 @@ interface WorkflowExecutionRecord {
   endedAt: Date | null
   totalDurationMs: number | null
   costTotal: string | null
-  executionData: unknown
 }
 
 interface JobExecutionRecord {
@@ -79,7 +78,6 @@ async function resolveExecutionSnapshotContext(
       endedAt: workflowExecutionLogs.endedAt,
       totalDurationMs: workflowExecutionLogs.totalDurationMs,
       costTotal: workflowExecutionLogs.costTotal,
-      executionData: workflowExecutionLogs.executionData,
     })
     .from(workflowExecutionLogs)
     .where(eq(workflowExecutionLogs.executionId, executionId))
@@ -118,6 +116,47 @@ async function resolveExecutionSnapshotContext(
   return { ...workspace, executionId, record: { kind: 'job', ...jobRecord } }
 }
 
+async function readSnapshotStates(
+  ids: string[],
+  signal: AbortSignal | undefined
+): Promise<Map<string, unknown>> {
+  const states = new Map<string, unknown>()
+  for (let index = 0; index < ids.length; index += SNAPSHOT_QUERY_CHUNK_SIZE) {
+    signal?.throwIfAborted()
+    const rows = await db
+      .select({
+        id: workflowExecutionSnapshots.id,
+        stateData: workflowExecutionSnapshots.stateData,
+      })
+      .from(workflowExecutionSnapshots)
+      .where(
+        inArray(workflowExecutionSnapshots.id, ids.slice(index, index + SNAPSHOT_QUERY_CHUNK_SIZE))
+      )
+    for (const row of rows) states.set(row.id, row.stateData)
+  }
+  return states
+}
+
+/**
+ * The run's execution data with the fields a completion write changes alongside it,
+ * read in one statement after authorization: a caller the use case refuses never
+ * fetches the payload, and a run finishing mid-request cannot pair new data with
+ * stale timing or cost.
+ */
+async function readRunCompletion(executionId: string) {
+  const [row] = await db
+    .select({
+      executionData: workflowExecutionLogs.executionData,
+      endedAt: workflowExecutionLogs.endedAt,
+      totalDurationMs: workflowExecutionLogs.totalDurationMs,
+      costTotal: workflowExecutionLogs.costTotal,
+    })
+    .from(workflowExecutionLogs)
+    .where(eq(workflowExecutionLogs.executionId, executionId))
+    .limit(1)
+  return row
+}
+
 function collectChildSnapshotIds(traceSpans: TraceSpan[]): string[] {
   const ids = new Set<string>()
   const pending = [...traceSpans]
@@ -175,17 +214,18 @@ const authorizedReadExecutionSnapshotUseCase = defineAuthorizedWorkspaceUseCase(
       }
     }
 
-    const [snapshot] = await db
-      .select()
-      .from(workflowExecutionSnapshots)
-      .where(eq(workflowExecutionSnapshots.id, record.stateSnapshotId))
-      .limit(1)
-    if (!snapshot) {
+    const workflowState = (await readSnapshotStates([record.stateSnapshotId], input.signal)).get(
+      record.stateSnapshotId
+    )
+    if (workflowState === undefined) {
       throw new OrchestrationError('not_found', 'Workflow state snapshot not found')
     }
 
+    input.signal?.throwIfAborted()
+    const completion = await readRunCompletion(record.executionId)
+    if (!completion) throw new OrchestrationError('not_found', 'Workflow execution not found')
     const executionData = (await materializeExecutionData(
-      record.executionData as Record<string, unknown> | null,
+      completion.executionData as Record<string, unknown> | null,
       {
         workspaceId: context.workspaceId,
         workflowId: record.workflowId,
@@ -201,40 +241,23 @@ const authorizedReadExecutionSnapshotUseCase = defineAuthorizedWorkspaceUseCase(
       })
     }
 
-    const childSnapshotIds = collectChildSnapshotIds(traceSpans)
-    const childWorkflowSnapshots: Array<{ id: string; stateData: unknown }> = []
-    for (let index = 0; index < childSnapshotIds.length; index += CHILD_SNAPSHOT_QUERY_CHUNK_SIZE) {
-      input.signal?.throwIfAborted()
-      childWorkflowSnapshots.push(
-        ...(await db
-          .select({
-            id: workflowExecutionSnapshots.id,
-            stateData: workflowExecutionSnapshots.stateData,
-          })
-          .from(workflowExecutionSnapshots)
-          .where(
-            inArray(
-              workflowExecutionSnapshots.id,
-              childSnapshotIds.slice(index, index + CHILD_SNAPSHOT_QUERY_CHUNK_SIZE)
-            )
-          ))
-      )
-    }
+    const childWorkflowSnapshots = await readSnapshotStates(
+      collectChildSnapshotIds(traceSpans),
+      input.signal
+    )
 
     input.signal?.throwIfAborted()
     return {
       executionId: record.executionId,
       workflowId: record.workflowId,
-      workflowState: snapshot.stateData as Record<string, unknown>,
-      childWorkflowSnapshots: Object.fromEntries(
-        childWorkflowSnapshots.map((child) => [child.id, child.stateData])
-      ),
+      workflowState: workflowState as Record<string, unknown>,
+      childWorkflowSnapshots: Object.fromEntries(childWorkflowSnapshots),
       executionMetadata: {
         trigger: record.trigger,
         startedAt: record.startedAt.toISOString(),
-        endedAt: record.endedAt?.toISOString(),
-        totalDurationMs: record.totalDurationMs,
-        cost: projectCostTotal(record.costTotal, projection),
+        endedAt: completion.endedAt?.toISOString(),
+        totalDurationMs: completion.totalDurationMs,
+        cost: projectCostTotal(completion.costTotal, projection),
       },
     }
   },

apps/sim/lib/logs/types.ts

@@ -225,6 +225,21 @@ export function isPersistedWorkflowExecutionStatus(
 ): value is PersistedWorkflowExecutionStatus {
   return (PERSISTED_WORKFLOW_EXECUTION_STATUSES as readonly string[]).includes(value)
 }
+
+/**
+ * Statuses of a finished run's log. A resume can never claim a log in one of them (the
+ * same set the resume attempt-failure `CASE` preserves).
+ */
+const TERMINAL_WORKFLOW_EXECUTION_STATUSES = [
+  'cancelled',
+  'failed',
+  'completed',
+] as const satisfies readonly PersistedWorkflowExecutionStatus[]
+
+export function isTerminalWorkflowExecutionStatus(status: string): boolean {
+  return (TERMINAL_WORKFLOW_EXECUTION_STATUSES as readonly string[]).includes(status)
+}
+
 /**
  * In-flight statuses a crashed worker can strand, which the stale-execution
  * cron terminalizes. `pending` and `paused` are excluded: both are written as