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 bebdd37db72..df117356688 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 { + 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, diff --git a/apps/sim/executor/execution/snapshot-serializer.test.ts b/apps/sim/executor/execution/snapshot-serializer.test.ts index 2c44d6c2a0c..8245ddba95e 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,129 @@ describe('serializePauseSnapshot', () => { expect(serialized.metadata.includeToolCalls).toBeUndefined() }) }) + +describe('compactPauseSnapshotScopes', () => { + 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', + { + iteration: 1, + loopType: 'forEach', + currentIterationOutputs: new Map(), + allIterationOutputs: [], + ...overrides, + }, + ], + ]), + } as Partial) + } + + 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) + + 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) + + 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) + }) +}) diff --git a/apps/sim/executor/execution/snapshot-serializer.ts b/apps/sim/executor/execution/snapshot-serializer.ts index fe8721a4875..4af148df09c 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,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, 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. + * + * 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 { + 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 (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) + } + 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) {