diff --git a/src/workflow/executor/dag/index.test.ts b/src/workflow/executor/dag/index.test.ts index ab5919d3cf..86297a5f15 100644 --- a/src/workflow/executor/dag/index.test.ts +++ b/src/workflow/executor/dag/index.test.ts @@ -12,12 +12,18 @@ import "#veryfront/schemas/_test-setup.ts"; * @module ai/workflow/executor/dag/index.test */ -import { assertEquals, assertExists, assertRejects } from "#veryfront/testing/assert.ts"; +import { + assertEquals, + assertExists, + assertRejects, + assertStringIncludes, +} from "#veryfront/testing/assert.ts"; import { beforeEach, describe, it } from "#veryfront/testing/bdd.ts"; import { DAGExecutor } from "./index.ts"; import type { Checkpoint, LoopExecutionContext, + NodeState, WorkflowContext, WorkflowNode, WorkflowRun, @@ -1628,6 +1634,214 @@ describe("DAGExecutor", () => { }); }); + describe("wait resume inside a nested composite", () => { + /** + * A composite runs its children against a synthetic run, so the child graph + * cannot read the real run's status. Resuming an approval used to look, one + * level down, exactly like recovering from a dead worker: the enclosing + * composite was re-entered off the recovery budget that exists for crashes. + * + * It only became visible once a re-entered sibling was queued behind the + * concurrency limit. Executing the node overwrites the bumped attempt; + * a node that never gets its turn keeps it, so ordinary approvals raised + * the recorded attempt until the budget was declared spent. + */ + const waitingParallel = (id: string): WorkflowNode => ({ + id, + dependsOn: [], + config: { + type: "parallel", + nodes: [ + { + id: `${id}-wait`, + dependsOn: [], + config: { type: "wait", waitType: "approval", message: "m" } as any, + }, + ], + } as any, + }); + + const approve = ( + states: Record, + waitId: string, + ): Record => ({ + ...states, + [waitId]: { ...states[waitId]!, status: "completed", completedAt: new Date() }, + }); + + it("spends no recovery budget when an approval resumes a nested composite", async () => { + const exec = new DAGExecutor({ + stepExecutor: new MockStepExecutor(), + // Forces the second inner composite to queue behind the first. + maxConcurrency: 1, + }); + const nodes: WorkflowNode[] = [ + { + id: "outer", + dependsOn: [], + config: { + type: "parallel", + nodes: [waitingParallel("inner-a"), waitingParallel("inner-b")], + } as any, + }, + ]; + + const first = await exec.execute(nodes, createTestRun()); + assertEquals(first.waiting, true); + assertEquals(first.waitingNode, "inner-a-wait"); + + // Approving the first wait re-enters "outer", which re-enters "inner-a" + // and reaches "inner-b" -- which parks on its own approval. + const second = await exec.execute( + nodes, + createTestRun({ + status: "waiting", + nodeStates: approve(first.nodeStates, "inner-a-wait"), + context: first.context, + }), + ); + assertEquals(second.waiting, true); + assertEquals(second.waitingNode, "inner-b-wait"); + assertEquals( + second.nodeStates["inner-a"]!.attempt, + 1, + "an approval resume must not consume the recovery budget of a nested composite", + ); + + // Approving the second wait must finish the run. Nothing was ever + // interrupted, so nothing may be reported as interrupted. + const third = await exec.execute( + nodes, + createTestRun({ + status: "waiting", + nodeStates: approve(second.nodeStates, "inner-b-wait"), + context: second.context, + }), + ); + assertEquals(third.error, undefined); + assertEquals(third.completed, true); + }); + + it("recovers an interrupted step on a wait resume instead of skipping it", async () => { + // Parked and interrupted are not exclusive. A worker can die with a step + // in flight while a sibling wait is parked, leaving the run "waiting" + // with that step still recorded running. Treating the whole resume as + // "nothing to recover" strands it -- and with nothing left ready the + // graph reports completion, so the workflow finishes having silently + // skipped a side effect. + const executed: string[] = []; + const exec = new DAGExecutor({ + stepExecutor: new MockStepExecutor(new Map(), (node) => { + executed.push(node.id); + return { success: true, output: node.id, executionTime: 1 }; + }), + }); + const nodes: WorkflowNode[] = [ + { + id: "gate", + dependsOn: [], + config: { type: "wait", waitType: "approval", message: "m" } as any, + }, + { id: "side-effect", dependsOn: [], config: { type: "step" } as any }, + ]; + + const result = await exec.execute( + nodes, + createTestRun({ + status: "waiting", + nodeStates: { + gate: { nodeId: "gate", status: "completed", attempt: 1, completedAt: new Date() }, + "side-effect": { + nodeId: "side-effect", + status: "running", + attempt: 1, + startedAt: new Date(), + }, + }, + }), + ); + + assertEquals(result.completed, true); + assertEquals(executed, ["side-effect"]); + assertEquals(result.nodeStates["side-effect"]!.status, "completed"); + }); + + it("never reports completion while an interrupted step is out of budget", async () => { + // The same shape once the budget is gone. Failing loudly is the only + // honest answer; reporting success would drop the step on the floor. + const executed: string[] = []; + const exec = new DAGExecutor({ + stepExecutor: new MockStepExecutor(new Map(), (node) => { + executed.push(node.id); + return { success: true, output: node.id, executionTime: 1 }; + }), + }); + const nodes: WorkflowNode[] = [ + { + id: "gate", + dependsOn: [], + config: { type: "wait", waitType: "approval", message: "m" } as any, + }, + { id: "side-effect", dependsOn: [], config: { type: "step" } as any }, + ]; + + const result = await exec.execute( + nodes, + createTestRun({ + status: "waiting", + nodeStates: { + gate: { nodeId: "gate", status: "completed", attempt: 1, completedAt: new Date() }, + "side-effect": { + nodeId: "side-effect", + status: "running", + attempt: 2, + startedAt: new Date(), + }, + }, + }), + ); + + assertEquals(result.completed, false); + assertEquals(executed, []); + assertStringIncludes(result.error ?? "", "retry budget exhausted"); + }); + + it("still recovers a nested node when the worker died mid-run", async () => { + const executed: string[] = []; + const exec = new DAGExecutor({ + stepExecutor: new MockStepExecutor(new Map(), (node) => { + executed.push(node.id); + return { success: true, output: node.id, executionTime: 1 }; + }), + }); + const nodes: WorkflowNode[] = [ + { + id: "outer", + dependsOn: [], + config: { + type: "parallel", + nodes: [{ id: "child", dependsOn: [], config: { type: "step" } as any }], + } as any, + }, + ]; + + // A dead worker leaves a "running" run, not a "waiting" one. + const result = await exec.execute( + nodes, + createTestRun({ + status: "running", + nodeStates: { + outer: { nodeId: "outer", status: "running", attempt: 1, startedAt: new Date() }, + child: { nodeId: "child", status: "running", attempt: 1, startedAt: new Date() }, + }, + }), + ); + + assertEquals(result.completed, true); + assertEquals(executed, ["child"]); + }); + }); + describe("loop resume (H9)", () => { it("should not re-run completed steps of an in-flight loop iteration on resume", async () => { let incrRuns = 0; diff --git a/src/workflow/executor/dag/index.ts b/src/workflow/executor/dag/index.ts index b65e98479b..487783cfcf 100644 --- a/src/workflow/executor/dag/index.ts +++ b/src/workflow/executor/dag/index.ts @@ -33,6 +33,7 @@ import type { DAGExecutorConfig, DAGExecutorInternalConfig, DAGInternalExecutionResult, + ExecutionScope, NodeExecutionResult, } from "./types.ts"; import { deriveNodeStatus, shouldCheckpoint } from "./utils.ts"; @@ -76,9 +77,18 @@ export class DAGExecutor { abortSignal?: AbortSignal, ownership?: CheckpointOwnership, ): Promise { + const scope: ExecutionScope = { + rootRunId: run.id, + executionRunId: run.id, + // Read the reason execution stopped once, here, from the only run record + // that carries it. Every child graph below runs against a synthetic run + // whose status is always "running" and would otherwise read a crash. + resumingWait: run.status === "waiting", + ownership, + }; const { contextPatch: _contextPatch, ...result } = await runWithWorkflowSourceIntegrationPolicy( run, - () => this.executeUnwrapped(nodes, run, run.id, startFromNode, abortSignal, ownership), + () => this.executeUnwrapped(nodes, run, scope, startFromNode, abortSignal), ); return result; } @@ -86,17 +96,9 @@ export class DAGExecutor { private async executeUnwrapped( nodes: WorkflowNode[], run: WorkflowRun, - /** - * The root, backend-persisted run id. Composite and sub-workflow nodes execute - * against synthetic run records with generated ids; span correlation must always - * use the id callers can actually look up, so it is threaded rather than read - * from `run`. - */ - rootRunId: string, + scope: ExecutionScope, startFromNode?: string, abortSignal?: AbortSignal, - ownership?: CheckpointOwnership, - executionRunId = run.id, ): Promise { abortSignal?.throwIfAborted(); const context = cloneExecutionState(run.context, "Workflow context"); @@ -132,7 +134,12 @@ export class DAGExecutor { // run. Re-run it. That matches what already happens when a worker dies // before writing any state at all -- the node looks untouched and runs // again -- except that the recorded attempt now bounds the retries. - const resumingWait = run.status === "waiting"; + // + // This reads the reason off the scope, never off `run`: a child graph's + // run is synthetic and permanently "running", so inferring it here would + // charge every nested composite re-entry to the crash budget on an + // ordinary approval. + const { resumingWait } = scope; // Only the top-level run has a row in the backend to write. Composites // execute their children against synthetic runs (`${node.id}_parallel`, // `_branch`, `_iter_N`) whose node states are a different keyspace: a loop @@ -142,7 +149,7 @@ export class DAGExecutor { // top-level node as pending and re-running the workflow from the start -- // the duplicate side effect this recovery path exists to prevent. // Child recoveries are persisted by the parent when it returns. - const isDurableRun = run.id === rootRunId; + const isDurableRun = run.id === scope.rootRunId; const exhausted: Array<{ nodeId: string; attempts: number; maxAttempts: number }> = []; for (const [nodeId, degree] of inDegree) { if (degree !== 0 || ready.includes(nodeId)) continue; @@ -150,8 +157,11 @@ export class DAGExecutor { if (state?.status !== "running") continue; const node = nodeMap.get(nodeId); if (!node) continue; - if (resumingWait) { - if (RESUMABLE_COMPOSITE_TYPES.has(node.config.type)) ready.push(nodeId); + // A composite recorded running on a wait resume encloses the decision. + // Re-enter it so the child resumes; nothing crashed, so it spends no + // recovery budget. + if (resumingWait && RESUMABLE_COMPOSITE_TYPES.has(node.config.type)) { + ready.push(nodeId); continue; } // A wait recorded as running is parked on its decision, never a dead @@ -161,6 +171,14 @@ export class DAGExecutor { // run is always "running" even while its wait is parked. if (node.config.type === "wait") continue; + // Anything else recorded running falls through to recovery even on a + // wait resume. Parked and interrupted are not exclusive: a worker can + // die with a step in flight, leave a sibling wait parked, and the run + // then reaches "waiting" with that step still marked running. Treating + // the whole resume as "nothing to recover" strands it -- and because + // nothing is left ready, the graph reports completion and the workflow + // finishes having silently skipped it. + // The step executor restarts its own retry loop at 1 and overwrites the // recorded attempt, so it cannot bound anything across worker deaths. // Count them here instead, or repeated crashes re-run the node forever @@ -180,10 +198,10 @@ export class DAGExecutor { nodeStates[nodeId] = { ...state, attempt: attempts + 1 }; if (isDurableRun) { const recovered = await this.config.onRecoveryScheduled?.({ - runId: rootRunId, + runId: scope.rootRunId, nodeId, nodeStates: structuredClone(nodeStates), - ownership, + ownership: scope.ownership, }); if (recovered === false) { throw ORCHESTRATION_ERROR.create({ @@ -242,10 +260,8 @@ export class DAGExecutor { nodeMap.get(nodeId)!, contextSnapshots[i]!, nodeStateSnapshots[i]!, - rootRunId, - executionRunId, + scope, abortSignal, - ownership, ) ), ); @@ -314,7 +330,7 @@ export class DAGExecutor { const nodeConfig = nodeMap.get(nodeId); if (nodeResult.state.status === "completed" && nodeConfig && shouldCheckpoint(nodeConfig)) { - await this.checkpoint(run.id, nodeId, context, nodeStates, ownership); + await this.checkpoint(run.id, nodeId, context, nodeStates, scope.ownership); } if (nodeResult.state.status === "failed") { @@ -384,10 +400,8 @@ export class DAGExecutor { node: WorkflowNode, context: WorkflowContext, nodeStates: Record, - rootRunId: string, - executionRunId: string, + scope: ExecutionScope, abortSignal?: AbortSignal, - ownership?: CheckpointOwnership, ): Promise { abortSignal?.throwIfAborted(); const nodeId = node.id; @@ -405,10 +419,8 @@ export class DAGExecutor { node, context, nodeStates, - rootRunId, - executionRunId, + scope, abortSignal, - ownership, ); // A failing node returns a failed state rather than throwing, so the span's own // catch never runs. Without this the span stays UNSET and a failed run is @@ -424,7 +436,7 @@ export class DAGExecutor { return result; }, { - "workflow.run_id": rootRunId, + "workflow.run_id": scope.rootRunId, "workflow.node.id": nodeId, "workflow.node.type": node.config.type, }, @@ -439,10 +451,8 @@ export class DAGExecutor { node: WorkflowNode, context: WorkflowContext, nodeStates: Record, - rootRunId: string, - executionRunId: string, + scope: ExecutionScope, abortSignal?: AbortSignal, - ownership?: CheckpointOwnership, ): Promise { const nodeId = node.id; this.config.onNodeStart?.(nodeId); @@ -462,23 +472,14 @@ export class DAGExecutor { switch (config.type) { case "step": - return this.executeStepNode(node, context, executionRunId, abortSignal); + return this.executeStepNode(node, context, scope.executionRunId, abortSignal); case "parallel": return executeCompositeNodeWithPolicy({ node, parentSignal: abortSignal, cancellationGracePeriod: this.config.cancellationGracePeriod, execute: (attemptSignal) => - this.executeParallelNode( - node, - config, - context, - nodeStates, - rootRunId, - executionRunId, - attemptSignal, - ownership, - ), + this.executeParallelNode(node, config, context, nodeStates, scope, attemptSignal), }); case "map": return executeCompositeNodeWithPolicy({ @@ -493,15 +494,7 @@ export class DAGExecutor { nodeStates, runtime: { executeChildGraph: (nodes, run, options) => - this.executeChildGraph( - nodes, - run, - rootRunId, - executionRunId, - options, - attemptSignal, - ownership, - ), + this.executeChildGraph(nodes, run, scope, options, attemptSignal), onNodeComplete: this.config.onNodeComplete, abortSignal: attemptSignal, }, @@ -529,10 +522,8 @@ export class DAGExecutor { selectedBranch, context, nodeStates, - rootRunId, - executionRunId, + scope, attemptSignal, - ownership, ); }, }); @@ -545,16 +536,7 @@ export class DAGExecutor { parentSignal: abortSignal, cancellationGracePeriod: this.config.cancellationGracePeriod, execute: (attemptSignal) => - this.executeSubWorkflowNode( - node, - config, - context, - rootRunId, - executionRunId, - nodeStates, - attemptSignal, - ownership, - ), + this.executeSubWorkflowNode(node, config, context, nodeStates, scope, attemptSignal), }); case "loop": return executeCompositeNodeWithPolicy({ @@ -569,15 +551,7 @@ export class DAGExecutor { nodeStates, runtime: { executeChildGraph: (nodes, run) => - this.executeChildGraph( - nodes, - run, - rootRunId, - executionRunId, - undefined, - attemptSignal, - ownership, - ), + this.executeChildGraph(nodes, run, scope, undefined, attemptSignal), onNodeComplete: this.config.onNodeComplete, abortSignal: attemptSignal, }, @@ -631,10 +605,8 @@ export class DAGExecutor { config: ParallelNodeConfig, context: WorkflowContext, nodeStates: Record, - rootRunId: string, - executionRunId: string, + scope: ExecutionScope, abortSignal?: AbortSignal, - ownership?: CheckpointOwnership, ): Promise { abortSignal?.throwIfAborted(); const startTime = Date.now(); @@ -656,11 +628,9 @@ export class DAGExecutor { createdAt: new Date(), sourceIntegrationPolicy: captureWorkflowSourceIntegrationPolicy(), }, - rootRunId, + scope, undefined, abortSignal, - ownership, - executionRunId, ); abortSignal?.throwIfAborted(); @@ -697,10 +667,8 @@ export class DAGExecutor { conditionResult: boolean, context: WorkflowContext, nodeStates: Record, - rootRunId: string, - executionRunId: string, + scope: ExecutionScope, abortSignal?: AbortSignal, - ownership?: CheckpointOwnership, ): Promise { abortSignal?.throwIfAborted(); const startTime = Date.now(); @@ -737,11 +705,9 @@ export class DAGExecutor { createdAt: new Date(), sourceIntegrationPolicy: captureWorkflowSourceIntegrationPolicy(), }, - rootRunId, + scope, undefined, abortSignal, - ownership, - executionRunId, ); abortSignal?.throwIfAborted(); @@ -807,11 +773,9 @@ export class DAGExecutor { node: WorkflowNode, config: SubWorkflowNodeConfig, context: WorkflowContext, - rootRunId: string, - executionRunId: string, nodeStates: Record, + scope: ExecutionScope, abortSignal?: AbortSignal, - ownership?: CheckpointOwnership, ): Promise { abortSignal?.throwIfAborted(); const startTime = Date.now(); @@ -858,11 +822,9 @@ export class DAGExecutor { createdAt: new Date(), sourceIntegrationPolicy: captureWorkflowSourceIntegrationPolicy(), }, - rootRunId, + scope, undefined, abortSignal, - ownership, - executionRunId, ); abortSignal?.throwIfAborted(); @@ -926,22 +888,12 @@ export class DAGExecutor { private async executeChildGraph( nodes: WorkflowNode[], run: WorkflowRun, - rootRunId: string, - executionRunId: string, + scope: ExecutionScope, options?: ChildGraphExecutionOptions, abortSignal?: AbortSignal, - ownership?: CheckpointOwnership, ): Promise { if (!options?.maxConcurrency) { - return await this.executeUnwrapped( - nodes, - run, - rootRunId, - undefined, - abortSignal, - ownership, - executionRunId, - ); + return await this.executeUnwrapped(nodes, run, scope, undefined, abortSignal); } // Run the child graph on a scoped executor rather than mutating @@ -952,14 +904,6 @@ export class DAGExecutor { ...this.config, maxConcurrency: options.maxConcurrency, }); - return await childExecutor.executeUnwrapped( - nodes, - run, - rootRunId, - undefined, - abortSignal, - ownership, - executionRunId, - ); + return await childExecutor.executeUnwrapped(nodes, run, scope, undefined, abortSignal); } } diff --git a/src/workflow/executor/dag/types.ts b/src/workflow/executor/dag/types.ts index d65270d685..69dc4f8c5c 100644 --- a/src/workflow/executor/dag/types.ts +++ b/src/workflow/executor/dag/types.ts @@ -16,6 +16,33 @@ export interface ContextPatch { delete: string[]; } +/** + * Facts that belong to the execution as a whole rather than to one graph. + * + * Composite nodes run their children against synthetic `WorkflowRun` records + * that are never persisted, so a child graph can learn nothing about the real + * run from the run it is handed. Anything a child graph must agree with the + * root run about is threaded here instead of inferred from that record. + */ +export interface ExecutionScope { + /** + * The root, backend-persisted run id. Synthetic child runs carry generated + * ids, so this is the only id a caller can actually look up -- span + * correlation and recovery persistence must both use it. + */ + rootRunId: string; + /** Run id handed to step execution for run-scoped hooks. */ + executionRunId: string; + /** + * True when the root run is resuming from a decision it parked on, false when + * it is recovering from a worker that died mid-node. A node recorded + * `running` means something different in each case, and only the root run + * record can tell them apart. + */ + resumingWait: boolean; + ownership?: CheckpointOwnership; +} + export interface DAGExecutorConfig { stepExecutor: StepExecutor; checkpointManager?: CheckpointManager; diff --git a/src/workflow/executor/workflow-tracing.test.ts b/src/workflow/executor/workflow-tracing.test.ts index c2896eafda..fe6237e539 100644 --- a/src/workflow/executor/workflow-tracing.test.ts +++ b/src/workflow/executor/workflow-tracing.test.ts @@ -245,21 +245,20 @@ describe("workflow/executor tracing", () => { executeUnwrapped( nodes: WorkflowNode[], run: WorkflowRun, - rootRunId: string, + scope: { + rootRunId: string; + executionRunId: string; + resumingWait: boolean; + ownership?: unknown; + }, startFromNode?: string, abortSignal?: AbortSignal, - ownership?: unknown, - executionRunId?: string, ): Promise; - }).executeUnwrapped( - nodes, - syntheticRun, - "durable-root-run", - undefined, - undefined, - undefined, - "durable-hook-run", - ); + }).executeUnwrapped(nodes, syntheticRun, { + rootRunId: "durable-root-run", + executionRunId: "durable-hook-run", + resumingWait: false, + }); await tracing.provider.forceFlush(); const nodeSpan = byName(tracing.exporter.getFinishedSpans(), "workflow.node child");