fix(execution): compact loop state before serializing a pause snapshot

A loop compacts its accumulated iteration outputs when it exits, but a pause is
by definition mid-flight and never reaches that point. The running total
therefore arrived at the serializer uncompacted and tripped its size assertion,
which throws rather than degrades — turning the pause into a failed run, so no
paused_executions row was ever written.

The approval notification goes out during block execution, well before the
engine builds the paused result, so the approver was left holding a working
looking resume link pointing at a row that never existed, and the run reported
a generic failure with no hint that a byte budget caused it.

Run the same compaction the loop performs on exit, and register the keys it
mints: reads are gated on the context's key list, so a resumed run could not
materialize the offloaded values otherwise. The assertion stays — it is a valid
post-condition, and an unstorable snapshot means an unresumable pause.

Author: waleedlatif1

apps/sim/executor/execution/engine.ts

@@ -8,7 +8,10 @@ import {
 import { BlockType, EDGE } from '@/executor/constants'
 import type { DAG } from '@/executor/dag/builder'
 import type { EdgeManager } from '@/executor/execution/edge-manager'
-import { serializePauseSnapshot } from '@/executor/execution/snapshot-serializer'
+import {
+  compactPauseSnapshotScopes,
+  serializePauseSnapshot,
+} from '@/executor/execution/snapshot-serializer'
 import type { SerializableExecutionState } from '@/executor/execution/types'
 import type { NodeExecutionOrchestrator } from '@/executor/orchestrators/node'
 import type {
@@ -127,7 +130,7 @@ export class ExecutionEngine {
       }
 
       if (this.pausedBlocks.size > 0) {
-        return this.buildPausedResult(startTime)
+        return await this.buildPausedResult(startTime)
       }
 
       const endTime = performance.now()
@@ -491,12 +494,13 @@ export class ExecutionEngine {
     this.addMultipleToQueue(readyNodes)
   }
 
-  private buildPausedResult(startTime: number): ExecutionResult {
+  private async buildPausedResult(startTime: number): Promise<ExecutionResult> {
     const endTime = performance.now()
     this.context.metadata.endTime = new Date().toISOString()
     this.context.metadata.duration = endTime - startTime
     this.context.metadata.status = 'paused'
 
+    await compactPauseSnapshotScopes(this.context)
     const snapshotSeed = serializePauseSnapshot(this.context, [], this.dag, this.edgeManager)
     const pausePoints: PausePoint[] = Array.from(this.pausedBlocks.values()).map((pause) => ({
       contextId: pause.contextId,

apps/sim/executor/execution/snapshot-serializer.test.ts

@@ -4,7 +4,10 @@
 import { describe, expect, it, vi } from 'vitest'
 import type { DAG, DAGNode } from '@/executor/dag/builder'
 import { EdgeManager } from '@/executor/execution/edge-manager'
-import { serializePauseSnapshot } from '@/executor/execution/snapshot-serializer'
+import {
+  compactPauseSnapshotScopes,
+  serializePauseSnapshot,
+} from '@/executor/execution/snapshot-serializer'
 import type { ExecutionContext } from '@/executor/types'
 import { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
 
@@ -256,3 +259,71 @@ describe('serializePauseSnapshot', () => {
     expect(serialized.metadata.includeToolCalls).toBeUndefined()
   })
 })
+
+describe('compactPauseSnapshotScopes', () => {
+  function createLoopContext(iterations: number, bytesPerIteration: number): ExecutionContext {
+    const payload = 'x'.repeat(bytesPerIteration)
+    return createContext({
+      loopExecutions: new Map([
+        [
+          'loop-1',
+          {
+            loopId: 'loop-1',
+            iteration: iterations,
+            maxIterations: iterations,
+            currentIterationOutputs: new Map(),
+            // Each entry is well under the per-value cap; only the running total is oversized,
+            // which is exactly what per-iteration compaction cannot catch.
+            allIterationOutputs: Array.from({ length: iterations }, () => ({ payload })),
+          },
+        ],
+      ]),
+    } as Partial<ExecutionContext>)
+  }
+
+  /**
+   * A loop compacts its accumulated outputs when it exits, but a pause is
+   * mid-flight and never gets there — so without this pass the running total
+   * trips the serializer's size assertion and the pause fails outright.
+   */
+  it('lets a pause inside a long-running loop serialize', async () => {
+    const context = createLoopContext(40, 300_000)
+    const dag = { nodes: new Map<string, DAGNode>() } as unknown as DAG
+    const edgeManager = new EdgeManager(dag)
+
+    expect(() => serializePauseSnapshot(context, [], dag, edgeManager)).toThrow(
+      'oversized loop execution state'
+    )
+
+    await compactPauseSnapshotScopes(context)
+
+    const seed = serializePauseSnapshot(context, [], dag, edgeManager)
+    expect(seed.snapshot).toBeTruthy()
+  })
+
+  /**
+   * The refs are only usable if the resumed run is authorized to read them, so
+   * the keys compaction registers must reach the snapshot's trusted-access list.
+   */
+  it('authorizes the offloaded values for the resumed run', async () => {
+    const context = createLoopContext(40, 300_000)
+    const dag = { nodes: new Map<string, DAGNode>() } as unknown as DAG
+    const edgeManager = new EdgeManager(dag)
+
+    await compactPauseSnapshotScopes(context)
+    const parsed = JSON.parse(serializePauseSnapshot(context, [], dag, edgeManager).snapshot) as {
+      state?: { trustedLargeValueAccess?: { largeValueKeys?: string[] } }
+    }
+
+    expect(parsed.state?.trustedLargeValueAccess?.largeValueKeys?.length ?? 0).toBeGreaterThan(0)
+  })
+
+  it('leaves a loop whose accumulated output already fits untouched', async () => {
+    const context = createLoopContext(2, 100)
+    const before = structuredClone(context.loopExecutions?.get('loop-1')?.allIterationOutputs)
+
+    await compactPauseSnapshotScopes(context)
+
+    expect(context.loopExecutions?.get('loop-1')?.allIterationOutputs).toEqual(before)
+  })
+})

apps/sim/executor/execution/snapshot-serializer.ts

@@ -1,4 +1,6 @@
+import { recordMaterializedAccessKeys } from '@/lib/execution/payloads/access-keys'
 import { LARGE_VALUE_THRESHOLD_BYTES } from '@/lib/execution/payloads/large-value-ref'
+import { compactSubflowResults } from '@/lib/execution/payloads/serializer'
 import type { DAG } from '@/executor/dag/builder'
 import type { EdgeManager } from '@/executor/execution/edge-manager'
 import { ExecutionSnapshot } from '@/executor/execution/snapshot'
@@ -182,6 +184,43 @@ function serializeParallelExecutions(
   return result
 }
 
+/**
+ * Offload accumulated loop iteration outputs so a pause snapshot stays compact.
+ *
+ * A loop compacts `allIterationOutputs` when it exits, but a pause is by
+ * definition mid-flight and never reaches that point — so the running total
+ * arrives at the serializer uncompacted and trips its size assertion, failing
+ * the pause outright. The approval notification has already gone out by then,
+ * leaving the approver holding a link to a paused execution that was never
+ * recorded.
+ *
+ * Mirrors the loop-exit pass: entries move to large-value storage and the
+ * snapshot keeps refs. The keys they register are picked up by the snapshot's
+ * `trustedLargeValueAccess`, so the resumed run can still read them.
+ */
+export async function compactPauseSnapshotScopes(context: ExecutionContext): Promise<void> {
+  if (!context.loopExecutions?.size) return
+
+  const options = {
+    workspaceId: context.workspaceId,
+    workflowId: context.workflowId,
+    executionId: context.executionId,
+    largeValueExecutionIds: context.largeValueExecutionIds,
+    largeValueKeys: context.largeValueKeys,
+    allowLargeValueWorkflowScope: context.allowLargeValueWorkflowScope,
+    userId: context.userId,
+    requireDurable: true,
+  }
+
+  for (const scope of context.loopExecutions.values()) {
+    if (!scope.allIterationOutputs?.length) continue
+    scope.allIterationOutputs = await compactSubflowResults(scope.allIterationOutputs, options)
+    // Authorize the refs this pass just minted. Reads are gated on the context's
+    // key list, so a resumed run cannot materialize them otherwise.
+    recordMaterializedAccessKeys(context, scope.allIterationOutputs)
+  }
+}
+
 export function serializePauseSnapshot(
   context: ExecutionContext,
   triggerBlockIds: string[],