fix(executor): stop HITL error edges from firing on successful resume (#5152)

* fix(executor): stop HITL error edges from firing on successful resume

* add comments
This commit is contained in:
Vikhyath Mondreti
2026-06-20 12:41:44 -07:00
committed by GitHub
parent 3ebb9a5029
commit 35a7bf61d2
4 changed files with 315 additions and 8 deletions
@@ -875,6 +875,117 @@ describe('EdgeManager', () => {
})
})
describe('deactivateResumedEdge', () => {
it('prunes a resumed pause block error edge without firing the error target', () => {
// Models the HITL-resume bug: a pause block and a regular block both feed
// an error-notifier via `error` handles. On a fully successful run the
// notifier must never run.
const pauseId = 'pause-block'
const regularId = 'regular-block'
const notifyId = 'error-notify'
const pauseNode = createMockNode(
pauseId,
[
{ target: 'next', sourceHandle: EDGE.SOURCE },
{ target: notifyId, sourceHandle: EDGE.ERROR },
],
[]
)
const regularNode = createMockNode(regularId, [
{ target: notifyId, sourceHandle: EDGE.ERROR },
])
const notifyNode = createMockNode(notifyId, [], [pauseId, regularId])
const dag = createMockDAG(
new Map<string, DAGNode>([
[pauseId, pauseNode],
[regularId, regularNode],
[notifyId, notifyNode],
])
)
const edgeManager = new EdgeManager(dag)
// Resume releases the pause block's error edge as deactivated.
edgeManager.deactivateResumedEdge(pauseId, notifyId, EDGE.ERROR)
expect(edgeManager.getDeactivatedEdges()).toContain(
JSON.stringify([pauseId, notifyId, EDGE.ERROR])
)
expect(edgeManager.getNodesWithActivatedEdge()).not.toContain(notifyId)
// The regular block then completes successfully → its error edge deactivates too.
const readyNodes = edgeManager.processOutgoingEdges(regularNode, { result: 'ok' })
// With no error edge ever activated, the notifier is never scheduled.
expect(readyNodes).not.toContain(notifyId)
expect(edgeManager.getNodesWithActivatedEdge()).not.toContain(notifyId)
})
it('still fires the error target when a real upstream block errors', () => {
// Same topology, but here the regular block genuinely errors — the notifier
// must fire even though the pause block's error edge was pruned on resume.
const pauseId = 'pause-block'
const regularId = 'regular-block'
const notifyId = 'error-notify'
const regularNode = createMockNode(regularId, [
{ target: notifyId, sourceHandle: EDGE.ERROR },
])
const notifyNode = createMockNode(notifyId, [], [pauseId, regularId])
const dag = createMockDAG(
new Map<string, DAGNode>([
[regularId, regularNode],
[notifyId, notifyNode],
])
)
const edgeManager = new EdgeManager(dag)
edgeManager.deactivateResumedEdge(pauseId, notifyId, EDGE.ERROR)
const readyNodes = edgeManager.processOutgoingEdges(regularNode, { error: 'boom' })
expect(readyNodes).toContain(notifyId)
expect(edgeManager.getNodesWithActivatedEdge()).toContain(notifyId)
})
it('leaves a convergence join ready+activated after pruning a pause error edge', () => {
// Join `C` is fed by `B.source` (succeeds) and `P.error` (pause block).
// After B activates and P's error edge is pruned on resume, C must be both
// activated and ready so the engine re-queues it.
const sourceId = 'block-b'
const pauseId = 'pause-block'
const joinId = 'join-c'
const sourceNode = createMockNode(sourceId, [{ target: joinId, sourceHandle: EDGE.SOURCE }])
const pauseNode = createMockNode(pauseId, [{ target: joinId, sourceHandle: EDGE.ERROR }])
const joinNode = createMockNode(joinId, [], [sourceId, pauseId])
const edgeManager = new EdgeManager(
createMockDAG(
new Map<string, DAGNode>([
[sourceId, sourceNode],
[pauseId, pauseNode],
[joinId, joinNode],
])
)
)
// Phase 1: B succeeds → activates B→C, but C still waits on the pause edge.
const readyAfterB = edgeManager.processOutgoingEdges(sourceNode, { result: 'ok' })
expect(readyAfterB).not.toContain(joinId)
expect(edgeManager.hasActivatedEdge(joinId)).toBe(true)
expect(edgeManager.isNodeReady(joinNode)).toBe(false)
// Resume: pause block's error edge is pruned → C is now ready (and stays
// activated), which is exactly the state the engine uses to re-queue it.
edgeManager.deactivateResumedEdge(pauseId, joinId, EDGE.ERROR)
expect(edgeManager.hasActivatedEdge(joinId)).toBe(true)
expect(edgeManager.isNodeReady(joinNode)).toBe(true)
})
})
describe('Diamond pattern (convergent paths)', () => {
it('should handle diamond: condition splits then converges at merge point', () => {
const conditionId = 'condition-1'
@@ -134,6 +134,10 @@ export class EdgeManager {
return Array.from(this.nodesWithActivatedEdge)
}
hasActivatedEdge(nodeId: string): boolean {
return this.nodesWithActivatedEdge.has(nodeId)
}
restoreDeactivatedEdges(edgeKeys?: string[], activatedNodeIds?: string[]): void {
this.deactivatedEdges = new Set(
(edgeKeys ?? []).map((edgeKey) => this.normalizeSerializedEdgeKey(edgeKey))
@@ -145,6 +149,21 @@ export class EdgeManager {
this.nodesWithActivatedEdge.add(nodeId)
}
/**
* Deactivates the `error` edge of a successfully-resumed pause block instead of
* firing it: the block completed normally, so its error path is pruned (and any
* now-dead descendants cascaded), mirroring how a normally-succeeding block's
* error edge is handled in {@link processOutgoingEdges}.
*
* `cascadeTargets` is intentionally left undefined here (unlike the
* {@link processOutgoingEdges} call site, which passes an explicit Set): the
* resume path has no loop/parallel sentinels to queue — the pause block's
* `source` edge drives continuation — so cascade-target collection is omitted.
*/
deactivateResumedEdge(sourceId: string, targetId: string, sourceHandle?: string): void {
this.deactivateEdgeAndDescendants(sourceId, targetId, sourceHandle)
}
/**
* Clear deactivated edges for a set of nodes (used when restoring loop state for next iteration).
*
+132
View File
@@ -112,6 +112,8 @@ function createMockEdgeManager(
getDeactivatedEdges: vi.fn(() => []),
getNodesWithActivatedEdge: vi.fn(() => []),
markNodeWithActivatedEdge: vi.fn(),
deactivateResumedEdge: vi.fn(),
hasActivatedEdge: vi.fn(() => false),
} as unknown as MockEdgeManager
}
@@ -231,6 +233,136 @@ describe('ExecutionEngine', () => {
expect(nodeOrchestrator.executeNode).not.toHaveBeenCalled()
})
it('deactivates a resumed pause block error edge instead of firing it', async () => {
// Pause block has two outgoing edges: a `source` continuation edge and an
// `error` edge to an error-notifier. On a successful resume the error edge
// must be deactivated (never marked/queued); only the continuation fires.
const pauseNode = createMockNode('pause-block', 'human_in_the_loop')
pauseNode.outgoingEdges.set('pause-block→next-source', {
target: 'next',
sourceHandle: EDGE.SOURCE,
})
pauseNode.outgoingEdges.set('pause-block→notify-error', {
target: 'error-notify',
sourceHandle: EDGE.ERROR,
})
const nextNode = createMockNode('next', 'function')
nextNode.incomingEdges.add('pause-block')
const errorNotifyNode = createMockNode('error-notify', 'gmail')
errorNotifyNode.incomingEdges.add('pause-block')
const dag = createMockDAG([pauseNode, nextNode, errorNotifyNode])
const context = createMockContext({
metadata: {
executionId: 'test-execution',
startTime: new Date().toISOString(),
pendingBlocks: [],
// remainingEdges omit sourceHandle (as persisted snapshots do), forcing
// the engine to resolve the handle from the live DAG.
remainingEdges: [
{ source: 'pause-block', target: 'next' },
{ source: 'pause-block', target: 'error-notify' },
],
} as any,
})
const edgeManager = createMockEdgeManager(() => [])
const nodeOrchestrator = createMockNodeOrchestrator()
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await engine.run()
// Continuation edge fires; error edge is deactivated, not activated/queued.
expect(edgeManager.markNodeWithActivatedEdge).toHaveBeenCalledWith('next')
expect(edgeManager.markNodeWithActivatedEdge).not.toHaveBeenCalledWith('error-notify')
expect(edgeManager.deactivateResumedEdge).toHaveBeenCalledWith(
'pause-block',
'error-notify',
EDGE.ERROR
)
// A pure error-handler target (never activated) must not be executed.
expect(nodeOrchestrator.executeNode).not.toHaveBeenCalledWith(context, 'error-notify')
})
it('re-queues a convergence target when a resumed pause error edge is pruned', async () => {
// The join is fed by a succeeding block's `source` edge (activated in
// phase 1) AND the pause block's `error` edge. On resume the error edge is
// pruned, but the join must still run because it already had a genuine
// activation — otherwise it would be silently stranded.
const pauseNode = createMockNode('pause-block', 'human_in_the_loop')
pauseNode.outgoingEdges.set('pause-block→join-error', {
target: 'join',
sourceHandle: EDGE.ERROR,
})
const joinNode = createMockNode('join', 'function')
joinNode.incomingEdges.add('pause-block')
const dag = createMockDAG([pauseNode, joinNode])
const context = createMockContext({
metadata: {
executionId: 'test-execution',
startTime: new Date().toISOString(),
pendingBlocks: [],
remainingEdges: [{ source: 'pause-block', target: 'join' }],
} as any,
})
const edgeManager = createMockEdgeManager(() => [])
// Join already received a genuine activation in phase 1 and is now ready.
vi.mocked(edgeManager.hasActivatedEdge).mockReturnValue(true)
vi.mocked(edgeManager.isNodeReady).mockReturnValue(true)
const nodeOrchestrator = createMockNodeOrchestrator()
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await engine.run()
expect(edgeManager.deactivateResumedEdge).toHaveBeenCalledWith(
'pause-block',
'join',
EDGE.ERROR
)
// Pruning is via deactivation, never force-activation...
expect(edgeManager.markNodeWithActivatedEdge).not.toHaveBeenCalledWith('join')
// ...but the already-activated convergence node still runs.
expect(nodeOrchestrator.executeNode).toHaveBeenCalledWith(context, 'join')
})
it('prefers the continuation handle when a pause block also errors into the same target', async () => {
// Pause block wires BOTH source and error into the same target. The error
// edge is registered first, but on a successful resume the continuation
// (source) handle must win, so the target is activated, not pruned.
const pauseNode = createMockNode('pause-block', 'human_in_the_loop')
pauseNode.outgoingEdges.set('pause-block→both-error', {
target: 'both',
sourceHandle: EDGE.ERROR,
})
pauseNode.outgoingEdges.set('pause-block→both-source', {
target: 'both',
sourceHandle: EDGE.SOURCE,
})
const bothNode = createMockNode('both', 'function')
bothNode.incomingEdges.add('pause-block')
const dag = createMockDAG([pauseNode, bothNode])
const context = createMockContext({
metadata: {
executionId: 'test-execution',
startTime: new Date().toISOString(),
pendingBlocks: [],
remainingEdges: [{ source: 'pause-block', target: 'both' }],
} as any,
})
const edgeManager = createMockEdgeManager(() => [])
const nodeOrchestrator = createMockNodeOrchestrator()
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await engine.run()
expect(edgeManager.markNodeWithActivatedEdge).toHaveBeenCalledWith('both')
expect(edgeManager.deactivateResumedEdge).not.toHaveBeenCalled()
expect(nodeOrchestrator.executeNode).toHaveBeenCalledWith(context, 'both')
})
it('should execute all nodes in a multi-node workflow', async () => {
const nodes = [
createMockNode('start', 'starter'),
+53 -8
View File
@@ -287,17 +287,33 @@ export class ExecutionEngine {
for (const edge of remainingEdges) {
const targetNode = this.dag.nodes.get(edge.target)
if (targetNode) {
const hadEdge = targetNode.incomingEdges.has(edge.source)
targetNode.incomingEdges.delete(edge.source)
if (hadEdge) {
this.edgeManager.markNodeWithActivatedEdge(targetNode.id)
}
if (!targetNode) continue
if (this.edgeManager.isNodeReady(targetNode)) {
this.execLogger.info('Node became ready after edge removal', { nodeId: targetNode.id })
const sourceHandle = this.resolveRemainingEdgeHandle(edge)
if (sourceHandle === EDGE.ERROR) {
this.edgeManager.deactivateResumedEdge(edge.source, targetNode.id, sourceHandle)
if (
this.edgeManager.hasActivatedEdge(targetNode.id) &&
this.edgeManager.isNodeReady(targetNode)
) {
this.execLogger.info('Convergence node ready after pruning resumed error edge', {
nodeId: targetNode.id,
})
this.addToQueue(targetNode.id)
}
continue
}
const hadEdge = targetNode.incomingEdges.has(edge.source)
targetNode.incomingEdges.delete(edge.source)
if (hadEdge) {
this.edgeManager.markNodeWithActivatedEdge(targetNode.id)
}
if (this.edgeManager.isNodeReady(targetNode)) {
this.execLogger.info('Node became ready after edge removal', { nodeId: targetNode.id })
this.addToQueue(targetNode.id)
}
}
@@ -351,6 +367,35 @@ export class ExecutionEngine {
}
}
/**
* Resolves the source handle for an edge released during pause/resume.
* Persisted `remainingEdges` may omit the handle, so fall back to the live DAG
* edge. When a source has both a continuation and an `error` edge to the same
* target, the continuation handle wins — a successful resume must not prune it.
*/
private resolveRemainingEdgeHandle(edge: {
source: string
target: string
sourceHandle?: string
}): string | undefined {
if (edge.sourceHandle !== undefined) return edge.sourceHandle
const sourceNode = this.dag.nodes.get(edge.source)
if (!sourceNode) return undefined
let hasErrorEdge = false
for (const [, outgoing] of sourceNode.outgoingEdges) {
if (outgoing.target !== edge.target) continue
if (outgoing.sourceHandle === EDGE.ERROR) {
hasErrorEdge = true
continue
}
return outgoing.sourceHandle
}
return hasErrorEdge ? EDGE.ERROR : undefined
}
private async processQueue(): Promise<void> {
while (this.readyQueue.length > 0) {
if (this.checkCancellation() || this.errorFlag) {