Skip to content

Commit 20e14bf

Browse files
committed
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.
1 parent 825e7e7 commit 20e14bf

9 files changed

Lines changed: 137 additions & 1 deletion

File tree

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

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -443,6 +443,50 @@ describe('ExecutionEngine', () => {
443443
pauseOutput
444444
)
445445
})
446+
447+
it('completes at the stop target instead of pausing when another branch pauses', async () => {
448+
const stopNode = createMockNode('stop', 'function')
449+
const pauseNode = createMockNode('pause', 'function')
450+
const dag = createMockDAG([stopNode, pauseNode])
451+
const context = createMockContext({
452+
decisions: { router: new Map(), condition: new Map() },
453+
stopAfterBlockId: 'stop',
454+
metadata: {
455+
executionId: 'test-execution',
456+
startTime: new Date().toISOString(),
457+
pendingBlocks: ['stop', 'pause'],
458+
},
459+
})
460+
const edgeManager = createMockEdgeManager()
461+
const nodeOrchestrator = createMockNodeOrchestrator()
462+
vi.mocked(nodeOrchestrator.executeNode).mockImplementation(async (_ctx, nodeId) => {
463+
if (nodeId === 'stop') {
464+
return { nodeId: 'stop', output: { done: true }, isFinalOutput: true }
465+
}
466+
return {
467+
nodeId: 'pause',
468+
output: {
469+
response: { status: 'paused' },
470+
_pauseMetadata: {
471+
contextId: 'pause-1',
472+
blockId: 'pause',
473+
response: { status: 'paused' },
474+
timestamp: new Date().toISOString(),
475+
pauseKind: 'hitl',
476+
},
477+
},
478+
isFinalOutput: false,
479+
}
480+
})
481+
482+
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
483+
const result = await engine.run()
484+
485+
expect(result.success).toBe(true)
486+
expect(result.status).not.toBe('paused')
487+
expect(result.pausePoints).toBeUndefined()
488+
expect(context.metadata.pausePoints).toEqual([])
489+
})
446490
})
447491

448492
describe('Cancellation via AbortSignal', () => {

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

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@ export class ExecutionEngine {
3232
private cancelledFlag = false
3333
private errorFlag = false
3434
private stoppedEarlyFlag = false
35+
/** Set when the run ends because it reached `stopAfterBlockId`. */
36+
private stopAfterReached = false
3537
private executionError: Error | null = null
3638
private abortPromise!: Promise<void>
3739
private abortResolve!: () => void
@@ -122,7 +124,18 @@ export class ExecutionEngine {
122124
throw this.executionError
123125
}
124126

125-
if (this.pausedBlocks.size > 0) {
127+
if (this.stopAfterReached) {
128+
/**
129+
* Stop-after wins over a pause still pending on another branch. The run
130+
* completed at the stop target, so nothing downstream will run and no
131+
* pause point should survive for a later resume.
132+
*/
133+
this.pausedBlocks.clear()
134+
this.context.metadata.pausePoints = []
135+
if (this.context.metadata.status === 'paused') {
136+
this.context.metadata.status = 'completed'
137+
}
138+
} else if (this.pausedBlocks.size > 0) {
126139
return this.buildPausedResult(startTime)
127140
}
128141

@@ -495,6 +508,7 @@ export class ExecutionEngine {
495508
output.shouldContinue === true || output.selectedRoute === EDGE.PARALLEL_CONTINUE
496509
if (!shouldContinue) {
497510
this.execLogger.info('Stopping execution after target block', { nodeId })
511+
this.stopAfterReached = true
498512
this.stoppedEarlyFlag = true
499513
return
500514
}

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

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,14 @@ describe('serializePauseSnapshot', () => {
5858
expect(snapshot.snapshot).not.toContain('raw-secret')
5959
})
6060

61+
it('carries the stop target across pause and resume', () => {
62+
const context = createContext({ stopAfterBlockId: 'stop-block' })
63+
64+
const serialized = JSON.parse(serializePauseSnapshot(context, ['next-block']).snapshot)
65+
66+
expect(serialized.metadata.stopAfterBlockId).toBe('stop-block')
67+
})
68+
6169
it('persists only encrypted value-adjacent provenance across pause and resume', () => {
6270
const provenance = {
6371
version: 1 as const,

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -306,6 +306,8 @@ export function serializePauseSnapshot(
306306
: undefined,
307307
/** Preserve the run-level agent-events opt-in across HITL pause/resume. */
308308
agentEvents: metadataFromContext?.agentEvents === true ? true : undefined,
309+
/** A resumed run must still stop after the block its original run targeted. */
310+
stopAfterBlockId: context.stopAfterBlockId,
309311
}
310312

311313
const snapshot = new ExecutionSnapshot(

‎apps/sim/executor/execution/types.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,11 @@ export interface ExecutionMetadata {
5959
pendingBlocks?: string[]
6060
resumeFromSnapshot?: boolean
6161
resumeTerminalNoop?: boolean
62+
/**
63+
* Stop target carried across a human-in-the-loop pause. Without it a resumed
64+
* run continues past the target block the original run was told to stop at.
65+
*/
66+
stopAfterBlockId?: string
6267
credentialAccountUserId?: string
6368
workflowStateOverride?: {
6469
blocks: Record<string, any>
Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
import { describe, expect, it } from 'vitest'
2+
import { isPausingStopTarget } from '@/executor/utils/stop-after'
3+
import type { SerializedBlock } from '@/serializer/types'
4+
5+
function block(id: string, type: string): SerializedBlock {
6+
return {
7+
id,
8+
position: { x: 0, y: 0 },
9+
config: { tool: type, params: {} },
10+
inputs: {},
11+
outputs: {},
12+
metadata: { id: type, name: id },
13+
enabled: true,
14+
}
15+
}
16+
17+
describe('isPausingStopTarget', () => {
18+
it.each(['human_in_the_loop', 'human_in_the_loop_v2', 'wait'])(
19+
'rejects a %s stop target',
20+
(type) => {
21+
expect(isPausingStopTarget([block('b1', type)], 'b1')).toBe(true)
22+
}
23+
)
24+
25+
it.each(['function', 'agent', 'condition', 'loop', 'parallel'])(
26+
'allows a %s stop target',
27+
(type) => {
28+
expect(isPausingStopTarget([block('b1', type)], 'b1')).toBe(false)
29+
}
30+
)
31+
32+
it('allows an unknown block id', () => {
33+
expect(isPausingStopTarget([block('b1', 'function')], 'missing')).toBe(false)
34+
})
35+
})
Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
import { BlockType, isHumanInTheLoopBlock } from '@/executor/constants'
2+
import type { SerializedBlock } from '@/serializer/types'
3+
4+
/**
5+
* Whether `stopAfterBlockId` targets a block that pauses the run — a
6+
* human-in-the-loop or wait block.
7+
*
8+
* Such a target cannot be honored. The run pauses at that block, and the resume
9+
* path prunes the paused block's outgoing edges, so the stop target would be lost
10+
* and execution would continue past it. Callers reject it upfront instead of
11+
* starting a run whose stop target resume cannot keep.
12+
*/
13+
export function isPausingStopTarget(
14+
blocks: readonly SerializedBlock[],
15+
stopAfterBlockId: string
16+
): boolean {
17+
const blockType = blocks.find((block) => block.id === stopAfterBlockId)?.metadata?.id
18+
return isHumanInTheLoopBlock(blockType) || blockType === BlockType.WAIT
19+
}

‎apps/sim/lib/workflows/executor/execution-core.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ import {
6767
type ResolvedSecretTraceRegistry,
6868
} from '@/executor/utils/resolved-secret-trace-registry'
6969
import { isRunMetadataEnabled } from '@/executor/utils/start-block'
70+
import { isPausingStopTarget } from '@/executor/utils/stop-after'
7071
import {
7172
buildLoopSentinelEndId,
7273
buildParallelSentinelEndId,
@@ -903,6 +904,11 @@ async function executeWorkflowCoreImpl(
903904
// Resolve stopAfterBlockId for loop/parallel containers to their sentinel-end IDs
904905
let resolvedStopAfterBlockId = stopAfterBlockId
905906
if (stopAfterBlockId) {
907+
if (isPausingStopTarget(serializedWorkflow.blocks, stopAfterBlockId)) {
908+
throw new Error(
909+
`Cannot stop after a block that pauses ("${stopAfterBlockId}"); choose a non-pausing block.`
910+
)
911+
}
906912
if (serializedWorkflow.loops?.[stopAfterBlockId]) {
907913
resolvedStopAfterBlockId = buildLoopSentinelEndId(stopAfterBlockId)
908914
} else if (serializedWorkflow.parallels?.[stopAfterBlockId]) {

‎apps/sim/lib/workflows/executor/human-in-the-loop-manager.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1885,6 +1885,9 @@ export class PauseResumeManager {
18851885
includeFileBase64: true,
18861886
base64MaxBytes: undefined,
18871887
abortSignal: timeoutController.signal,
1888+
...(baseSnapshot.metadata.stopAfterBlockId
1889+
? { stopAfterBlockId: baseSnapshot.metadata.stopAfterBlockId }
1890+
: {}),
18881891
...(resumeDeploymentVersionId ? { resumeDeploymentVersionId } : {}),
18891892
})
18901893

0 commit comments

Comments
 (0)