fix(execution): cover every subflow field that grows without a bound

The first pass only compacted a loop's completed iteration outputs, which left
the same pause failure reachable by four other routes: a forEach collection, an
in-flight iteration output, two loops each individually under the limit but
oversized together, and a parallel's accumulated branch outputs. Parallel state
was not even asserted, so it shipped an oversized snapshot silently rather than
failing.

Compact every field that accumulates, assert parallel state alongside loop
state, and offload at a threshold far below the snapshot's own — the assertion
measures the combined record, so compacting at its ceiling is a no-op in
exactly the case that needs it.

Skip the pass entirely when the state already fits, so a pause per iteration
inside a modest loop pays one bounded measurement rather than a structural
rebuild each time, and count the compaction against the recorded duration
instead of stopping the clock before it runs.

Add an engine-level test: the serializer tests all passed with the call removed,
leaving the wiring itself undefended.

Author: waleedlatif1

apps/sim/executor/execution/engine.test.ts

@@ -427,6 +427,52 @@ describe('ExecutionEngine', () => {
       )
     })
 
+    /**
+     * The compaction pass is what keeps an oversized loop from failing the pause
+     * outright. Asserting it from the engine keeps the wiring defended: without
+     * this, removing the call leaves every serializer test still green.
+     */
+    it('compacts oversized loop state before building the paused result', async () => {
+      const node = createMockNode('hitl', 'function')
+      const dag = createMockDAG([node])
+      const payload = 'x'.repeat(300_000)
+      const context = createMockContext({
+        decisions: { router: new Map(), condition: new Map() },
+        loopExecutions: new Map([
+          [
+            'loop-1',
+            {
+              iteration: 1,
+              currentIterationOutputs: new Map(),
+              allIterationOutputs: Array.from({ length: 40 }, () => [{ payload }]),
+            },
+          ],
+        ]),
+      } as Partial<ExecutionContext>)
+      const edgeManager = createMockEdgeManager()
+      const nodeOrchestrator = createMockNodeOrchestrator()
+      vi.mocked(nodeOrchestrator.executeNode).mockResolvedValue({
+        nodeId: 'hitl',
+        output: {
+          response: { status: 'paused' },
+          _pauseMetadata: {
+            contextId: 'pause-1',
+            blockId: 'hitl',
+            response: { status: 'paused' },
+            timestamp: new Date().toISOString(),
+            pauseKind: 'hitl',
+          },
+        },
+        isFinalOutput: false,
+      })
+
+      const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
+      const result = await engine.run('hitl')
+
+      expect(result.status).toBe('paused')
+      expect(result.snapshotSeed?.snapshot).toBeTruthy()
+    })
+
     it('does not stop run-until execution on parallel batch continuation', async () => {
       const parallelEnd = createMockNode('parallel-end', 'parallel')
       const nextNode = createMockNode('next', 'function')

apps/sim/executor/execution/engine.ts

@@ -495,12 +495,12 @@ export class ExecutionEngine {
   }
 
   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 endTime = performance.now()
+    this.context.metadata.endTime = new Date().toISOString()
+    this.context.metadata.duration = endTime - startTime
     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

@@ -261,69 +261,132 @@ describe('serializePauseSnapshot', () => {
 })
 
 describe('compactPauseSnapshotScopes', () => {
-  function createLoopContext(iterations: number, bytesPerIteration: number): ExecutionContext {
-    const payload = 'x'.repeat(bytesPerIteration)
+  const dag = { nodes: new Map<string, DAGNode>() } as unknown as DAG
+  const edgeManager = new EdgeManager(dag)
+
+  function fatIterations(count: number, bytes: number): any[][] {
+    const payload = 'x'.repeat(bytes)
+    // Real shape is an array per iteration, which routes oversized entries
+    // through the chunked-manifest path rather than a single ref.
+    return Array.from({ length: count }, () => [{ payload }])
+  }
+
+  function loopContext(overrides: Record<string, unknown>): ExecutionContext {
     return createContext({
       loopExecutions: new Map([
         [
           'loop-1',
           {
-            loopId: 'loop-1',
-            iteration: iterations,
-            maxIterations: iterations,
+            iteration: 1,
+            loopType: 'forEach',
             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 })),
+            allIterationOutputs: [],
+            ...overrides,
           },
         ],
       ]),
     } 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.
-   */
+  const serialize = (context: ExecutionContext) =>
+    serializePauseSnapshot(context, [], dag, edgeManager)
+
   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)
+    const context = loopContext({ allIterationOutputs: fatIterations(40, 300_000) })
 
-    expect(() => serializePauseSnapshot(context, [], dag, edgeManager)).toThrow(
-      'oversized loop execution state'
-    )
+    expect(() => serialize(context)).toThrow('oversized loop execution state')
+    await compactPauseSnapshotScopes(context)
+    expect(serialize(context).snapshot).toBeTruthy()
+  })
+
+  /** A forEach collection is the most common way a loop's state gets large. */
+  it('compacts the forEach collection, not just the iteration outputs', async () => {
+    const context = loopContext({ items: fatIterations(40, 300_000) })
+
+    expect(() => serialize(context)).toThrow('oversized loop execution state')
+    await compactPauseSnapshotScopes(context)
+    expect(serialize(context).snapshot).toBeTruthy()
+  })
+
+  /** A single fat mid-flight block output reaches the limit on its own. */
+  it('compacts in-flight iteration outputs', async () => {
+    const context = loopContext({
+      currentIterationOutputs: new Map([['block-1', { payload: 'x'.repeat(9_000_000) }]]),
+    })
+
+    expect(() => serialize(context)).toThrow('oversized loop execution state')
+    await compactPauseSnapshotScopes(context)
+    expect(serialize(context).snapshot).toBeTruthy()
+  })
+
+  /** The assertion is on the whole record, so per-scope headroom is not enough. */
+  it('compacts across multiple loops whose combined state is oversized', async () => {
+    const context = createContext({
+      loopExecutions: new Map([
+        [
+          'loop-1',
+          {
+            iteration: 1,
+            currentIterationOutputs: new Map(),
+            allIterationOutputs: fatIterations(15, 300_000),
+          },
+        ],
+        [
+          'loop-2',
+          {
+            iteration: 1,
+            currentIterationOutputs: new Map(),
+            allIterationOutputs: fatIterations(15, 300_000),
+          },
+        ],
+      ]),
+    } as Partial<ExecutionContext>)
 
+    expect(() => serialize(context)).toThrow('oversized loop execution state')
     await compactPauseSnapshotScopes(context)
+    expect(serialize(context).snapshot).toBeTruthy()
+  })
 
-    const seed = serializePauseSnapshot(context, [], dag, edgeManager)
-    expect(seed.snapshot).toBeTruthy()
+  /** Parallels accumulate the same way and were previously not even asserted. */
+  it('compacts accumulated parallel branch outputs', async () => {
+    const context = createContext({
+      parallelExecutions: new Map([
+        [
+          'parallel-1',
+          {
+            parallelId: 'parallel-1',
+            totalBranches: 40,
+            branchOutputs: new Map([[0, fatIterations(40, 300_000).flat()]]),
+            accumulatedOutputs: new Map(),
+          },
+        ],
+      ]),
+    } as Partial<ExecutionContext>)
+
+    expect(() => serialize(context)).toThrow('oversized parallel execution state')
+    await compactPauseSnapshotScopes(context)
+    expect(serialize(context).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.
-   */
+  /** Refs are unusable unless the resumed run is authorized to read them. */
   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)
+    const context = loopContext({ allIterationOutputs: fatIterations(40, 300_000) })
 
     await compactPauseSnapshotScopes(context)
-    const parsed = JSON.parse(serializePauseSnapshot(context, [], dag, edgeManager).snapshot) as {
+    const parsed = JSON.parse(serialize(context).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)
+  /** Compaction is a structural rebuild, so a modest loop must not pay for it. */
+  it('skips the rebuild entirely when the state already fits', async () => {
+    const context = loopContext({ allIterationOutputs: fatIterations(2, 100) })
+    const before = context.loopExecutions?.get('loop-1')?.allIterationOutputs
 
     await compactPauseSnapshotScopes(context)
 
-    expect(context.loopExecutions?.get('loop-1')?.allIterationOutputs).toEqual(before)
+    expect(context.loopExecutions?.get('loop-1')?.allIterationOutputs).toBe(before)
   })
 })

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

@@ -1,6 +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 { compactExecutionPayload, 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'
@@ -185,23 +185,59 @@ function serializeParallelExecutions(
 }
 
 /**
- * Offload accumulated loop iteration outputs so a pause snapshot stays compact.
+ * Per-value offload ceiling applied once the subflow state is already oversized.
  *
- * 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.
+ * Deliberately far below the snapshot's own limit. The assertion measures the
+ * *combined* record, so scopes that are each individually under it still fail
+ * together — compacting at the snapshot ceiling would be a no-op in exactly the
+ * case that needs it. Only reached when the state is already too large, so the
+ * fidelity cost lands on runs that would otherwise fail outright.
+ */
+const PAUSE_SNAPSHOT_COMPACT_VALUE_BYTES = 64 * 1024
+
+/**
+ * Whether the serialized subflow state is already past what the snapshot allows.
+ *
+ * Measured on the serialized shape because that is what the assertions read,
+ * and bounded so an oversized structure short-circuits instead of being walked
+ * in full.
+ */
+function isSubflowStateOversized(loops?: Map<string, any>, parallels?: Map<string, any>): boolean {
+  const limit = LARGE_VALUE_THRESHOLD_BYTES
+  const loopBytes = getBoundedJsonByteLength(serializeLoopExecutions(loops), limit)
+  if (loopBytes !== undefined && loopBytes > limit) return true
+  const parallelBytes = getBoundedJsonByteLength(serializeParallelExecutions(parallels), limit)
+  return parallelBytes !== undefined && parallelBytes > limit
+}
+
+/**
+ * Offload accumulated subflow state so a pause snapshot stays under the size
+ * assertions below.
+ *
+ * A loop or parallel compacts its accumulated outputs when it *exits*, but a
+ * pause is by definition mid-flight and never reaches that point. The running
+ * total therefore arrives here uncompacted and trips the assertion, which
+ * throws rather than degrades — turning the pause into a failed run, so no
+ * paused-execution row is ever written. The approval notification has already
+ * gone out by then, leaving the approver holding a link to something 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.
+ * Every field that grows without an aggregate bound is covered: a loop's
+ * iteration outputs, its in-flight iteration outputs and its `forEach`
+ * collection, and the parallel equivalents. Compacting only one of them would
+ * leave the same failure reachable by a different route.
+ *
+ * Skipped entirely when the state already serializes small enough, so the
+ * common case — a pause per iteration inside a modest loop — pays one bounded
+ * measurement rather than a full structural rebuild each time.
  */
 export async function compactPauseSnapshotScopes(context: ExecutionContext): Promise<void> {
-  if (!context.loopExecutions?.size) return
+  const loops = context.loopExecutions
+  const parallels = context.parallelExecutions
+  if (!loops?.size && !parallels?.size) return
+  if (!isSubflowStateOversized(loops, parallels)) return
 
-  const options = {
+  const buildOptions = () => ({
     workspaceId: context.workspaceId,
     workflowId: context.workflowId,
     executionId: context.executionId,
@@ -210,14 +246,49 @@ export async function compactPauseSnapshotScopes(context: ExecutionContext): Pro
     allowLargeValueWorkflowScope: context.allowLargeValueWorkflowScope,
     userId: context.userId,
     requireDurable: true,
+    thresholdBytes: PAUSE_SNAPSHOT_COMPACT_VALUE_BYTES,
+  })
+
+  const compactList = async <T>(values: T[]): Promise<T[]> =>
+    compactSubflowResults(values, buildOptions())
+
+  const compactMapValues = async (map: Map<unknown, unknown[]>): Promise<void> => {
+    for (const [key, value] of map) {
+      if (Array.isArray(value) && value.length > 0) {
+        map.set(key, await compactList(value))
+      }
+    }
+  }
+
+  for (const scope of loops?.values() ?? []) {
+    if (scope.allIterationOutputs?.length) {
+      scope.allIterationOutputs = await compactList(scope.allIterationOutputs)
+    }
+    if (scope.items?.length) {
+      scope.items = await compactList(scope.items)
+    }
+    if (scope.currentIterationOutputs instanceof Map && scope.currentIterationOutputs.size > 0) {
+      for (const [blockId, output] of scope.currentIterationOutputs) {
+        scope.currentIterationOutputs.set(
+          blockId,
+          await compactExecutionPayload(output, { ...buildOptions(), preserveRoot: false })
+        )
+      }
+    }
+    recordMaterializedAccessKeys(context, scope)
   }
 
-  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)
+  for (const scope of parallels?.values() ?? []) {
+    if (scope.items?.length) {
+      scope.items = await compactList(scope.items)
+    }
+    if (scope.branchOutputs instanceof Map) {
+      await compactMapValues(scope.branchOutputs)
+    }
+    if (scope.accumulatedOutputs instanceof Map) {
+      await compactMapValues(scope.accumulatedOutputs)
+    }
+    recordMaterializedAccessKeys(context, scope)
   }
 }
 
@@ -280,6 +351,7 @@ export function serializePauseSnapshot(
 
   assertSnapshotValueIsCompact(context.workflowVariables, 'workflow variables')
   assertSnapshotValueIsCompact(state.loopExecutions, 'loop execution state')
+  assertSnapshotValueIsCompact(state.parallelExecutions, 'parallel execution state')
 
   const workspaceId = metadataFromContext?.workspaceId ?? context.workspaceId
   if (!workspaceId) {