From af500aad2d933b957a4fdc4edebdcfbaa777f5fc Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 4 Aug 2026 11:47:26 -0700 Subject: [PATCH 1/3] fix(execution): compact loop state before serializing a pause snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/executor/execution/engine.ts | 10 ++- .../execution/snapshot-serializer.test.ts | 73 ++++++++++++++++++- .../executor/execution/snapshot-serializer.ts | 39 ++++++++++ 3 files changed, 118 insertions(+), 4 deletions(-) diff --git a/apps/sim/executor/execution/engine.ts b/apps/sim/executor/execution/engine.ts index bebdd37db72..c1f86a55f05 100644 --- a/apps/sim/executor/execution/engine.ts +++ b/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 { 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, diff --git a/apps/sim/executor/execution/snapshot-serializer.test.ts b/apps/sim/executor/execution/snapshot-serializer.test.ts index 2c44d6c2a0c..4203c998e6b 100644 --- a/apps/sim/executor/execution/snapshot-serializer.test.ts +++ b/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) + } + + /** + * 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() } 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() } 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) + }) +}) diff --git a/apps/sim/executor/execution/snapshot-serializer.ts b/apps/sim/executor/execution/snapshot-serializer.ts index fe8721a4875..0a691f32d1b 100644 --- a/apps/sim/executor/execution/snapshot-serializer.ts +++ b/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 { + 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[], From 1f833ee74f61191c449a3bd83af6aaa2bb5978af Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 4 Aug 2026 12:03:11 -0700 Subject: [PATCH 2/3] fix(execution): cover every subflow field that grows without a bound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- apps/sim/executor/execution/engine.test.ts | 46 +++++++ apps/sim/executor/execution/engine.ts | 6 +- .../execution/snapshot-serializer.test.ts | 129 +++++++++++++----- .../executor/execution/snapshot-serializer.ts | 110 ++++++++++++--- 4 files changed, 236 insertions(+), 55 deletions(-) diff --git a/apps/sim/executor/execution/engine.test.ts b/apps/sim/executor/execution/engine.test.ts index 904fc5f22bb..451b373dc8b 100644 --- a/apps/sim/executor/execution/engine.test.ts +++ b/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) + 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') diff --git a/apps/sim/executor/execution/engine.ts b/apps/sim/executor/execution/engine.ts index c1f86a55f05..df117356688 100644 --- a/apps/sim/executor/execution/engine.ts +++ b/apps/sim/executor/execution/engine.ts @@ -495,12 +495,12 @@ export class ExecutionEngine { } private async buildPausedResult(startTime: number): Promise { - 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, diff --git a/apps/sim/executor/execution/snapshot-serializer.test.ts b/apps/sim/executor/execution/snapshot-serializer.test.ts index 4203c998e6b..597cc4421d4 100644 --- a/apps/sim/executor/execution/snapshot-serializer.test.ts +++ b/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() } 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): 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) } - /** - * 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() } 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) + 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) + + 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() } 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) }) }) diff --git a/apps/sim/executor/execution/snapshot-serializer.ts b/apps/sim/executor/execution/snapshot-serializer.ts index 0a691f32d1b..bf208fe8040 100644 --- a/apps/sim/executor/execution/snapshot-serializer.ts +++ b/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, parallels?: Map): 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 { - 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 (values: T[]): Promise => + compactSubflowResults(values, buildOptions()) + + const compactMapValues = async (map: Map): Promise => { + 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) { From e3671e84af19a417e1a14e47e63d4bfcc6da9e02 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 4 Aug 2026 12:10:08 -0700 Subject: [PATCH 3/3] fix(execution): stop offloading state the engine consumes structurally MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Compacting a loop's `items` was a regression: the orchestrator indexes that collection to derive the current `item`, the resume path rebuilds the scope verbatim without materializing anything, and the loop resolver asserts no refs reach it — so an oversized forEach would have traded a failed pause for a broken resume. `currentIterationOutputs` is excluded for the same reason: the block executor has already compacted its entries, and they resolve through the reference path rather than being read raw. Offloading is now limited to exactly the accumulators the orchestrators themselves compact when a subflow exits. An oversized `items` collection therefore still fails the pause; that is the honest outcome until it can be handled without breaking iteration. --- .../execution/snapshot-serializer.test.ts | 23 ++++++--------- .../executor/execution/snapshot-serializer.ts | 28 ++++++------------- 2 files changed, 18 insertions(+), 33 deletions(-) diff --git a/apps/sim/executor/execution/snapshot-serializer.test.ts b/apps/sim/executor/execution/snapshot-serializer.test.ts index 597cc4421d4..8245ddba95e 100644 --- a/apps/sim/executor/execution/snapshot-serializer.test.ts +++ b/apps/sim/executor/execution/snapshot-serializer.test.ts @@ -299,24 +299,19 @@ describe('compactPauseSnapshotScopes', () => { 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 () => { + /** + * `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) }) - 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() + 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. */ diff --git a/apps/sim/executor/execution/snapshot-serializer.ts b/apps/sim/executor/execution/snapshot-serializer.ts index bf208fe8040..4af148df09c 100644 --- a/apps/sim/executor/execution/snapshot-serializer.ts +++ b/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 { compactExecutionPayload, compactSubflowResults } from '@/lib/execution/payloads/serializer' +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' @@ -222,10 +222,14 @@ function isSubflowStateOversized(loops?: Map, parallels?: Map 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 parallels?.values() ?? []) { - if (scope.items?.length) { - scope.items = await compactList(scope.items) - } if (scope.branchOutputs instanceof Map) { await compactMapValues(scope.branchOutputs) }