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

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

@@ -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> {
+    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
-    this.context.metadata.status = 'paused'
-
     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,129 @@ describe('serializePauseSnapshot', () => {
     expect(serialized.metadata.includeToolCalls).toBeUndefined()
   })
 })
+
+describe('compactPauseSnapshotScopes', () => {
+  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',
+          {
+            iteration: 1,
+            loopType: 'forEach',
+            currentIterationOutputs: new Map(),
+            allIterationOutputs: [],
+            ...overrides,
+          },
+        ],
+      ]),
+    } as Partial<ExecutionContext>)
+  }
+
+  const serialize = (context: ExecutionContext) =>
+    serializePauseSnapshot(context, [], dag, edgeManager)
+
+  it('lets a pause inside a long-running loop serialize', async () => {
+    const context = loopContext({ allIterationOutputs: fatIterations(40, 300_000) })
+
+    expect(() => serialize(context)).toThrow('oversized loop execution state')
+    await compactPauseSnapshotScopes(context)
+    expect(serialize(context).snapshot).toBeTruthy()
+  })
+
+  /**
+   * `items` is consumed structurally — the orchestrator indexes it to derive
+   * `item`, and the loop resolver asserts no refs reach it — so it stays inline
+   * even though that leaves an oversized collection unhandled. Offloading it
+   * would trade a failed pause for a broken resume.
+   */
+  it('leaves the forEach collection inline rather than breaking iteration', async () => {
+    const context = loopContext({ items: fatIterations(40, 300_000) })
+
+    await compactPauseSnapshotScopes(context)
+
+    const items = context.loopExecutions?.get('loop-1')?.items as unknown[]
+    expect(JSON.stringify(items)).not.toContain('__simLargeValueRef')
+  })
+
+  /** 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()
+  })
+
+  /** 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()
+  })
+
+  /** Refs are unusable unless the resumed run is authorized to read them. */
+  it('authorizes the offloaded values for the resumed run', async () => {
+    const context = loopContext({ allIterationOutputs: fatIterations(40, 300_000) })
+
+    await compactPauseSnapshotScopes(context)
+    const parsed = JSON.parse(serialize(context).snapshot) as {
+      state?: { trustedLargeValueAccess?: { largeValueKeys?: string[] } }
+    }
+
+    expect(parsed.state?.trustedLargeValueAccess?.largeValueKeys?.length ?? 0).toBeGreaterThan(0)
+  })
+
+  /** 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).toBe(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,104 @@ function serializeParallelExecutions(
   return result
 }
 
+/**
+ * Per-value offload ceiling applied once the subflow state is already oversized.
+ *
+ * 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.
+ *
+ * Covers exactly the accumulators the orchestrators themselves compact when a
+ * subflow exits — a loop's iteration outputs and a parallel's branch and
+ * accumulated outputs. Deliberately excluded: `items`, which the loop consumes
+ * structurally (`orchestrators/loop.ts` indexes it to derive `item`, and the
+ * loop resolver asserts no refs reach it), and `currentIterationOutputs`, whose
+ * entries the block executor has already compacted and which resolve through
+ * the reference path that materializes refs. Offloading either would trade this
+ * failure for a broken resume.
+ *
+ * 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> {
+  const loops = context.loopExecutions
+  const parallels = context.parallelExecutions
+  if (!loops?.size && !parallels?.size) return
+  if (!isSubflowStateOversized(loops, parallels)) return
+
+  const buildOptions = () => ({
+    workspaceId: context.workspaceId,
+    workflowId: context.workflowId,
+    executionId: context.executionId,
+    largeValueExecutionIds: context.largeValueExecutionIds,
+    largeValueKeys: context.largeValueKeys,
+    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)
+    }
+    recordMaterializedAccessKeys(context, scope)
+  }
+
+  for (const scope of parallels?.values() ?? []) {
+    if (scope.branchOutputs instanceof Map) {
+      await compactMapValues(scope.branchOutputs)
+    }
+    if (scope.accumulatedOutputs instanceof Map) {
+      await compactMapValues(scope.accumulatedOutputs)
+    }
+    recordMaterializedAccessKeys(context, scope)
+  }
+}
+
 export function serializePauseSnapshot(
   context: ExecutionContext,
   triggerBlockIds: string[],
@@ -241,6 +341,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) {