From 20e14bfd57be2df64382049cfddec6baccf1c35b Mon Sep 17 00:00:00 2001 From: Krishna Date: Wed, 7 Oct 2026 16:51:51 +0530 Subject: [PATCH] fix(executor): honor stop-after over pending pauses and carry it on resume Three changes for issue #8661: 1. The engine now lets stop-after win over a pause still pending on another branch. Reaching the stop target completes the run and drops the pause points instead of returning 'paused' and persisting a resume that can never finish. (executor/execution/engine.ts) 2. stopAfterBlockId is rejected upfront when it targets a block that pauses (human-in-the-loop or wait). Such a target cannot be honored: the run pauses there and the resume path prunes its outgoing edges, so the stop target would be lost. (executor/utils/stop-after.ts, execution-core.ts) 3. stopAfterBlockId is persisted in the pause snapshot and passed back into the resumed run, so a run that pauses on another branch still ends after its original stop target. (execution/types.ts, snapshot-serializer.ts, human-in-the-loop-manager.ts) Tests: an engine case for stop-after vs a pending pause, a serializer case for the carried stop target, and unit coverage for the pausing-target check. --- apps/sim/executor/execution/engine.test.ts | 44 +++++++++++++++++++ apps/sim/executor/execution/engine.ts | 16 ++++++- .../execution/snapshot-serializer.test.ts | 8 ++++ .../executor/execution/snapshot-serializer.ts | 2 + apps/sim/executor/execution/types.ts | 5 +++ apps/sim/executor/utils/stop-after.test.ts | 35 +++++++++++++++ apps/sim/executor/utils/stop-after.ts | 19 ++++++++ .../lib/workflows/executor/execution-core.ts | 6 +++ .../executor/human-in-the-loop-manager.ts | 3 ++ 9 files changed, 137 insertions(+), 1 deletion(-) create mode 100644 apps/sim/executor/utils/stop-after.test.ts create mode 100644 apps/sim/executor/utils/stop-after.ts diff --git a/apps/sim/executor/execution/engine.test.ts b/apps/sim/executor/execution/engine.test.ts index b36e88bffcf..97efa442b39 100644 --- a/apps/sim/executor/execution/engine.test.ts +++ b/apps/sim/executor/execution/engine.test.ts @@ -443,6 +443,50 @@ describe('ExecutionEngine', () => { pauseOutput ) }) + + it('completes at the stop target instead of pausing when another branch pauses', async () => { + const stopNode = createMockNode('stop', 'function') + const pauseNode = createMockNode('pause', 'function') + const dag = createMockDAG([stopNode, pauseNode]) + const context = createMockContext({ + decisions: { router: new Map(), condition: new Map() }, + stopAfterBlockId: 'stop', + metadata: { + executionId: 'test-execution', + startTime: new Date().toISOString(), + pendingBlocks: ['stop', 'pause'], + }, + }) + const edgeManager = createMockEdgeManager() + const nodeOrchestrator = createMockNodeOrchestrator() + vi.mocked(nodeOrchestrator.executeNode).mockImplementation(async (_ctx, nodeId) => { + if (nodeId === 'stop') { + return { nodeId: 'stop', output: { done: true }, isFinalOutput: true } + } + return { + nodeId: 'pause', + output: { + response: { status: 'paused' }, + _pauseMetadata: { + contextId: 'pause-1', + blockId: 'pause', + response: { status: 'paused' }, + timestamp: new Date().toISOString(), + pauseKind: 'hitl', + }, + }, + isFinalOutput: false, + } + }) + + const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator) + const result = await engine.run() + + expect(result.success).toBe(true) + expect(result.status).not.toBe('paused') + expect(result.pausePoints).toBeUndefined() + expect(context.metadata.pausePoints).toEqual([]) + }) }) describe('Cancellation via AbortSignal', () => { diff --git a/apps/sim/executor/execution/engine.ts b/apps/sim/executor/execution/engine.ts index b6091f395e7..7db1160b47b 100644 --- a/apps/sim/executor/execution/engine.ts +++ b/apps/sim/executor/execution/engine.ts @@ -32,6 +32,8 @@ export class ExecutionEngine { private cancelledFlag = false private errorFlag = false private stoppedEarlyFlag = false + /** Set when the run ends because it reached `stopAfterBlockId`. */ + private stopAfterReached = false private executionError: Error | null = null private abortPromise!: Promise private abortResolve!: () => void @@ -122,7 +124,18 @@ export class ExecutionEngine { throw this.executionError } - if (this.pausedBlocks.size > 0) { + if (this.stopAfterReached) { + /** + * Stop-after wins over a pause still pending on another branch. The run + * completed at the stop target, so nothing downstream will run and no + * pause point should survive for a later resume. + */ + this.pausedBlocks.clear() + this.context.metadata.pausePoints = [] + if (this.context.metadata.status === 'paused') { + this.context.metadata.status = 'completed' + } + } else if (this.pausedBlocks.size > 0) { return this.buildPausedResult(startTime) } @@ -495,6 +508,7 @@ export class ExecutionEngine { output.shouldContinue === true || output.selectedRoute === EDGE.PARALLEL_CONTINUE if (!shouldContinue) { this.execLogger.info('Stopping execution after target block', { nodeId }) + this.stopAfterReached = true this.stoppedEarlyFlag = true return } diff --git a/apps/sim/executor/execution/snapshot-serializer.test.ts b/apps/sim/executor/execution/snapshot-serializer.test.ts index f2c855fd661..26693aa7ec2 100644 --- a/apps/sim/executor/execution/snapshot-serializer.test.ts +++ b/apps/sim/executor/execution/snapshot-serializer.test.ts @@ -58,6 +58,14 @@ describe('serializePauseSnapshot', () => { expect(snapshot.snapshot).not.toContain('raw-secret') }) + it('carries the stop target across pause and resume', () => { + const context = createContext({ stopAfterBlockId: 'stop-block' }) + + const serialized = JSON.parse(serializePauseSnapshot(context, ['next-block']).snapshot) + + expect(serialized.metadata.stopAfterBlockId).toBe('stop-block') + }) + it('persists only encrypted value-adjacent provenance across pause and resume', () => { const provenance = { version: 1 as const, diff --git a/apps/sim/executor/execution/snapshot-serializer.ts b/apps/sim/executor/execution/snapshot-serializer.ts index 784502a1bc5..e7efd4a72eb 100644 --- a/apps/sim/executor/execution/snapshot-serializer.ts +++ b/apps/sim/executor/execution/snapshot-serializer.ts @@ -306,6 +306,8 @@ export function serializePauseSnapshot( : undefined, /** Preserve the run-level agent-events opt-in across HITL pause/resume. */ agentEvents: metadataFromContext?.agentEvents === true ? true : undefined, + /** A resumed run must still stop after the block its original run targeted. */ + stopAfterBlockId: context.stopAfterBlockId, } const snapshot = new ExecutionSnapshot( diff --git a/apps/sim/executor/execution/types.ts b/apps/sim/executor/execution/types.ts index 59e1679209c..10f0a2866b0 100644 --- a/apps/sim/executor/execution/types.ts +++ b/apps/sim/executor/execution/types.ts @@ -59,6 +59,11 @@ export interface ExecutionMetadata { pendingBlocks?: string[] resumeFromSnapshot?: boolean resumeTerminalNoop?: boolean + /** + * Stop target carried across a human-in-the-loop pause. Without it a resumed + * run continues past the target block the original run was told to stop at. + */ + stopAfterBlockId?: string credentialAccountUserId?: string workflowStateOverride?: { blocks: Record diff --git a/apps/sim/executor/utils/stop-after.test.ts b/apps/sim/executor/utils/stop-after.test.ts new file mode 100644 index 00000000000..ff4b2f0b6f5 --- /dev/null +++ b/apps/sim/executor/utils/stop-after.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from 'vitest' +import { isPausingStopTarget } from '@/executor/utils/stop-after' +import type { SerializedBlock } from '@/serializer/types' + +function block(id: string, type: string): SerializedBlock { + return { + id, + position: { x: 0, y: 0 }, + config: { tool: type, params: {} }, + inputs: {}, + outputs: {}, + metadata: { id: type, name: id }, + enabled: true, + } +} + +describe('isPausingStopTarget', () => { + it.each(['human_in_the_loop', 'human_in_the_loop_v2', 'wait'])( + 'rejects a %s stop target', + (type) => { + expect(isPausingStopTarget([block('b1', type)], 'b1')).toBe(true) + } + ) + + it.each(['function', 'agent', 'condition', 'loop', 'parallel'])( + 'allows a %s stop target', + (type) => { + expect(isPausingStopTarget([block('b1', type)], 'b1')).toBe(false) + } + ) + + it('allows an unknown block id', () => { + expect(isPausingStopTarget([block('b1', 'function')], 'missing')).toBe(false) + }) +}) diff --git a/apps/sim/executor/utils/stop-after.ts b/apps/sim/executor/utils/stop-after.ts new file mode 100644 index 00000000000..9de41cb8416 --- /dev/null +++ b/apps/sim/executor/utils/stop-after.ts @@ -0,0 +1,19 @@ +import { BlockType, isHumanInTheLoopBlock } from '@/executor/constants' +import type { SerializedBlock } from '@/serializer/types' + +/** + * Whether `stopAfterBlockId` targets a block that pauses the run — a + * human-in-the-loop or wait block. + * + * Such a target cannot be honored. The run pauses at that block, and the resume + * path prunes the paused block's outgoing edges, so the stop target would be lost + * and execution would continue past it. Callers reject it upfront instead of + * starting a run whose stop target resume cannot keep. + */ +export function isPausingStopTarget( + blocks: readonly SerializedBlock[], + stopAfterBlockId: string +): boolean { + const blockType = blocks.find((block) => block.id === stopAfterBlockId)?.metadata?.id + return isHumanInTheLoopBlock(blockType) || blockType === BlockType.WAIT +} diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts index 0f994a481f6..fa41af03fa3 100644 --- a/apps/sim/lib/workflows/executor/execution-core.ts +++ b/apps/sim/lib/workflows/executor/execution-core.ts @@ -67,6 +67,7 @@ import { type ResolvedSecretTraceRegistry, } from '@/executor/utils/resolved-secret-trace-registry' import { isRunMetadataEnabled } from '@/executor/utils/start-block' +import { isPausingStopTarget } from '@/executor/utils/stop-after' import { buildLoopSentinelEndId, buildParallelSentinelEndId, @@ -903,6 +904,11 @@ async function executeWorkflowCoreImpl( // Resolve stopAfterBlockId for loop/parallel containers to their sentinel-end IDs let resolvedStopAfterBlockId = stopAfterBlockId if (stopAfterBlockId) { + if (isPausingStopTarget(serializedWorkflow.blocks, stopAfterBlockId)) { + throw new Error( + `Cannot stop after a block that pauses ("${stopAfterBlockId}"); choose a non-pausing block.` + ) + } if (serializedWorkflow.loops?.[stopAfterBlockId]) { resolvedStopAfterBlockId = buildLoopSentinelEndId(stopAfterBlockId) } else if (serializedWorkflow.parallels?.[stopAfterBlockId]) { diff --git a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts index ec9c0bd60c7..3a55347e1f2 100644 --- a/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts +++ b/apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts @@ -1885,6 +1885,9 @@ export class PauseResumeManager { includeFileBase64: true, base64MaxBytes: undefined, abortSignal: timeoutController.signal, + ...(baseSnapshot.metadata.stopAfterBlockId + ? { stopAfterBlockId: baseSnapshot.metadata.stopAfterBlockId } + : {}), ...(resumeDeploymentVersionId ? { resumeDeploymentVersionId } : {}), })