diff --git a/apps/execution-worker/src/engines/temporal/workflows/run-workflow.ts b/apps/execution-worker/src/engines/temporal/workflows/run-workflow.ts index 53c70ceae..a574830d0 100644 --- a/apps/execution-worker/src/engines/temporal/workflows/run-workflow.ts +++ b/apps/execution-worker/src/engines/temporal/workflows/run-workflow.ts @@ -1,11 +1,12 @@ // Temporal workflow entry. Runs inside V8 sandbox — only deterministic code // + proxyActivities allowed. Delegates graph traversal to the pure runGraph // from execution-core, wiring Temporal proxyActivities as port implementations. -import { CancellationScope, isCancellation, proxyActivities } from '@temporalio/workflow'; +import { ApplicationFailure, CancellationScope, isCancellation, proxyActivities } from '@temporalio/workflow'; import { type ActivityRunnerPort, type EventEmitterPort, + type RunGraphOutcome, type WorkflowExecutionInput, runGraph, } from '@workflow-builder/execution-core/workflow'; @@ -36,8 +37,10 @@ const events: EventEmitterPort = { }; export async function runWorkflow(input: WorkflowExecutionInput): Promise { + let outcome: RunGraphOutcome; + try { - await runGraph(input, runner, events); + outcome = await runGraph(input, runner, events); } catch (error) { if (isCancellation(error)) { // Root scope is cancelled — shield cleanup so these activities aren't @@ -49,4 +52,12 @@ export async function runWorkflow(input: WorkflowExecutionInput): } throw error; } + + // runGraph already emitted execution_failed and wrote the 'failed' status — this check + // tells Temporal to close the run as Failed rather than Completed. It has to be a + // TemporalFailure (anything else fails the workflow *task* and retries forever), and + // non-retryable since replaying a deterministic graph failure would re-run LLM activities. + if (outcome.status === 'failed') { + throw ApplicationFailure.nonRetryable(outcome.error.message, outcome.error.code ?? 'WorkflowExecutionFailed'); + } } diff --git a/packages/execution-core/README.md b/packages/execution-core/README.md index 441a9fd62..dc7610b94 100644 --- a/packages/execution-core/README.md +++ b/packages/execution-core/README.md @@ -106,32 +106,34 @@ The registry's mapped type — `{ [K in TNode['type']]: NodeExecutor` (`submit`, `cancel`). 2. Wire it up in `apps/backend/src/engine/index.ts` (swap `TemporalEngine` for the new adapter). 3. Make sure your engine wires `runGraph` (or equivalent traversal) to its activity primitives. +4. Translate a `{ status: 'failed' }` outcome from `runGraph` into your engine's own failure vocabulary. `runGraph` never throws for node failures — it reports them by return value — so an adapter that ignores the outcome will close failed runs as successful. See `run-workflow.ts` for the Temporal case, which raises `ApplicationFailure.nonRetryable` (only a `TemporalFailure` fails a Workflow Execution; anything else fails the workflow _task_ and retries it forever). ## Replay determinism diff --git a/packages/execution-core/replay-audit.md b/packages/execution-core/replay-audit.md index fe86ed89f..77db68eb0 100644 --- a/packages/execution-core/replay-audit.md +++ b/packages/execution-core/replay-audit.md @@ -42,7 +42,7 @@ The audit therefore focuses on `graph-runner.ts` + `errors.ts`. Activities and a | `Promise.all` completion order | Yes | ✅ Safe — `Promise.all` resolves with results in **input order**, regardless of completion order. The runner reads `results[i]` positionally, never branches on which promise resolved first. | | Async/await scheduling | Yes | ✅ Safe — Temporal patches the JS event loop microtask queue inside the sandbox; the order in which awaits resume is deterministic across replay. | | `Array.prototype.shift` on BFS queue | Yes | ✅ Safe — FIFO order is deterministic given a deterministic push order. The push order in `propagate` comes from iterating `successors` (a `Map` value), which is insertion-deterministic. | -| Throwing for control flow | Yes | ✅ Safe — `throw new Error('Workflow has no entrypoint node')` is synchronous and deterministic. `NodeExecutionError` is a plain `Error` subclass with no side effects in its constructor. | +| Throwing for control flow | No | ✅ Safe — the runner does not throw for control flow; failures are reported by return value (`RunGraphOutcome`), which is fully determined by the input. `NodeExecutionError` is a plain `Error` subclass with no side effects in its constructor. | | External clock / wall time | No | ✅ Safe — runner does not read time. `events.emitEvent('execution_started', ...)` etc. are activities; the timestamp is recorded by the activity outside the sandbox. | | Iteration over `Object.keys`/`values` | No | ✅ Safe — runner uses `Map` for stateful collections; `nodeOutputs` is an object but never iterated for control flow (only `{ ...nodeOutputs }` for context cloning, which preserves order). | | Module-level initialization side effects | No | ✅ Safe — `graph-runner.ts` exports only function declarations; no top-level statements that read environment or instantiate stateful objects. | diff --git a/packages/execution-core/src/graph-runner.test.ts b/packages/execution-core/src/graph-runner.test.ts index f16f65060..169d1d3a6 100644 --- a/packages/execution-core/src/graph-runner.test.ts +++ b/packages/execution-core/src/graph-runner.test.ts @@ -103,7 +103,7 @@ describe('runGraph — topological scheduling', () => { }); const events = makeEvents(); - await runGraph( + const outcome = await runGraph( makeInput([trigger('A'), trigger('B'), trigger('C')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'C')]), runner.port, events.port, @@ -113,6 +113,7 @@ describe('runGraph — topological scheduling', () => { expect(runner.contexts.B).toEqual({ A: 'a-result' }); expect(runner.contexts.C).toEqual({ A: 'a-result', B: 'b-result' }); expect(events.statuses.at(-1)).toEqual({ status: 'completed', errorMessage: undefined }); + expect(outcome).toEqual({ status: 'completed' }); }); it('fan-out A→{B,C} runs B and C in same wave', async () => { @@ -270,7 +271,7 @@ describe('runGraph — topological scheduling', () => { }); const events = makeEvents(); - await runGraph( + const outcome = await runGraph( makeInput([trigger('A'), trigger('B'), trigger('C')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'C')]), runner.port, events.port, @@ -281,20 +282,24 @@ describe('runGraph — topological scheduling', () => { const failedEvent = events.events.find((event) => event.type === 'execution_failed'); expect(failedEvent).toBeDefined(); expect(events.statuses.at(-1)).toEqual({ status: 'failed', errorMessage: 'boom' }); + expect(outcome).toEqual({ status: 'failed', error: { message: 'boom' } }); }); - it('throws when there is no entrypoint', async () => { + it('fails the run when there is no entrypoint', async () => { const runner = makeRunner(); const events = makeEvents(); // Cycle with no in-degree-zero node - await expect( - runGraph( - makeInput([trigger('A'), trigger('B')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'A')]), - runner.port, - events.port, - ), - ).rejects.toThrow('Workflow has no entrypoint node'); + const outcome = await runGraph( + makeInput([trigger('A'), trigger('B')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'A')]), + runner.port, + events.port, + ); + + expect(outcome).toEqual({ status: 'failed', error: { message: 'Workflow has no entrypoint node' } }); + expect(runner.callOrder).toEqual([]); + expect(events.events.map((event) => event.type)).toEqual(['execution_started', 'execution_failed']); + expect(events.statuses.at(-1)).toEqual({ status: 'failed', errorMessage: 'Workflow has no entrypoint node' }); }); it('cycle reachable from an entrypoint fails the workflow with a stalled-node message', async () => { @@ -306,7 +311,7 @@ describe('runGraph — topological scheduling', () => { const runner = makeRunner(); const events = makeEvents(); - await runGraph( + const outcome = await runGraph( makeInput( [trigger('A'), trigger('B'), trigger('C')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'C'), edge('e3', 'C', 'B')], @@ -320,6 +325,7 @@ describe('runGraph — topological scheduling', () => { expect(failedEvent?.payload).toEqual({ error: { message: expect.stringContaining('Workflow stalled') }, }); + expect(outcome.status).toBe('failed'); expect(events.statuses.at(-1)?.status).toBe('failed'); expect(events.statuses.at(-1)?.errorMessage).toContain('B'); expect(events.statuses.at(-1)?.errorMessage).toContain('C'); @@ -556,20 +562,19 @@ describe('runGraph — replay safety (sandbox-safe)', () => { expectNoConsoleWrites(); }); - it('a missing-entrypoint throw writes nothing to console — error surfaces by throw, not by log', async () => { - // The no-entrypoint path is the only one that throws synchronously instead - // of routing through events. The throw is the signal; no console fallback. + it('a missing entrypoint writes nothing to console — surfaces via execution_failed event', async () => { const runner = makeRunner(); const events = makeEvents(); - await expect( - runGraph( - makeInput([trigger('A'), trigger('B')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'A')]), - runner.port, - events.port, - ), - ).rejects.toThrow('Workflow has no entrypoint node'); + const outcome = await runGraph( + makeInput([trigger('A'), trigger('B')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'A')]), + runner.port, + events.port, + ); + expect(outcome.status).toBe('failed'); + const failedEvent = events.events.find((event) => event.type === 'execution_failed'); + expect(failedEvent?.payload).toEqual({ error: { message: 'Workflow has no entrypoint node' } }); expectNoConsoleWrites(); }); }); @@ -594,7 +599,7 @@ describe('runGraph — errorPolicy', () => { const runner = makeRunner({ B: { throws: 'boom' } }); const events = makeEvents(); - await runGraph( + const outcome = await runGraph( makeInput([trigger('A'), trigger('B', 'continue'), trigger('C')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'C')]), runner.port, events.port, @@ -605,6 +610,7 @@ describe('runGraph — errorPolicy', () => { // node_failed for B was emitted, execution itself completed. expect(events.events.some((event) => event.type === 'node_failed' && event.nodeId === 'B')).toBe(true); expect(events.statuses.at(-1)?.status).toBe('completed'); + expect(outcome).toEqual({ status: 'completed' }); }); it("'continue' preserves NodeExecutionError code in the absorbed output", async () => { @@ -635,7 +641,7 @@ describe('runGraph — errorPolicy', () => { const runner = makeRunner({ B: { throws: 'boom' } }); const events = makeEvents(); - await runGraph( + const outcome = await runGraph( makeInput( [trigger('A'), trigger('B', 'errorRoute'), trigger('Success'), trigger('Recovery')], [edge('e1', 'A', 'B'), edge('e2', 'B', 'Success', 'success'), edge('e3', 'B', 'Recovery', 'errorRoute')], @@ -648,6 +654,7 @@ describe('runGraph — errorPolicy', () => { expect(events.events.some((event) => event.type === 'node_started' && event.nodeId === 'Success')).toBe(false); expect(runner.contexts.Recovery).toEqual({ A: 'out-A', B: { error: { message: 'boom' } } }); expect(events.statuses.at(-1)?.status).toBe('completed'); + expect(outcome).toEqual({ status: 'completed' }); }); it("'errorRoute' — skip propagates transitively through the dead success branch", async () => { diff --git a/packages/execution-core/src/graph-runner.ts b/packages/execution-core/src/graph-runner.ts index 8a9cbbac4..9501b29bc 100644 --- a/packages/execution-core/src/graph-runner.ts +++ b/packages/execution-core/src/graph-runner.ts @@ -19,6 +19,14 @@ import type { WorkflowExecutionInput } from './ports/workflow-engine.port'; // policy literal so the wiring reads consistently from schema to runner. const RESERVED_ERROR_HANDLE = 'errorRoute'; +// How the run ended. Returned rather than thrown so `runGraph` stays engine-agnostic: +// each engine adapter decides how a failed outcome maps onto its own vocabulary (the +// Temporal adapter raises an ApplicationFailure so the Workflow Execution shows as +// Failed rather than Completed). Note a node failing under errorPolicy 'continue' or +// 'errorRoute' is absorbed by the graph and still yields `{ status: 'completed' }` — +// only an unhandled node failure, a stall, or a missing entrypoint fails the run. +export type RunGraphOutcome = { status: 'completed' } | { status: 'failed'; error: { message: string; code?: string } }; + // Topological scheduler. A node becomes ready only when ALL of its incoming // edges are resolved (predecessor either completed via a live route, or was // pruned by a decision node's nextPort). Ready nodes within the same wave @@ -36,17 +44,17 @@ export async function runGraph( input: WorkflowExecutionInput, runner: ActivityRunnerPort, events: EventEmitterPort, -): Promise { +): Promise { const adjacency = buildAdjacencyMap(input.definition.nodes, input.definition.edges); const inDegree = computeInDegrees(input.definition.nodes, input.definition.edges); + await events.emitEvent(input.executionId, 'execution_started', { workflowId: input.workflowId }); + const entrypoints = input.definition.nodes.filter((node) => (inDegree.get(node.id) ?? 0) === 0); if (entrypoints.length === 0) { - throw new Error('Workflow has no entrypoint node'); + return await failExecution(input.executionId, events, { message: 'Workflow has no entrypoint node' }); } - await events.emitEvent(input.executionId, 'execution_started', { workflowId: input.workflowId }); - // pendingPredecessors counts incoming edges not yet resolved (completed OR pruned). // liveIncoming counts incoming edges that resolved via a non-pruned route. // A node becomes ready when pending hits 0 AND liveIncoming > 0; @@ -77,9 +85,7 @@ export async function runGraph( // one in deterministic node order, just like the previous behavior. const fatal = results.find((r) => r.failed && resolveErrorPolicy(r.node) === 'fail'); if (fatal && fatal.failed) { - await events.emitEvent(input.executionId, 'execution_failed', { error: { message: fatal.message } }); - await events.updateStatus(input.executionId, 'failed', fatal.message); - return; + return await failExecution(input.executionId, events, { message: fatal.message, code: fatal.code }); } const newlyReady: TNode[] = []; @@ -115,13 +121,25 @@ export async function runGraph( } if (stalled.length > 0) { const message = `Workflow stalled: nodes never became ready: ${stalled.join(', ')}`; - await events.emitEvent(input.executionId, 'execution_failed', { error: { message } }); - await events.updateStatus(input.executionId, 'failed', message); - return; + return await failExecution(input.executionId, events, { message }); } await events.emitEvent(input.executionId, 'execution_completed'); await events.updateStatus(input.executionId, 'completed'); + return { status: 'completed' }; +} + +// Emits the terminal failure signals and shapes the outcome. Every failure path routes +// through here so the event, the engine status, and the returned outcome can never drift. +async function failExecution( + executionId: string, + events: EventEmitterPort, + error: { message: string; code?: string }, +): Promise { + const payload = error.code === undefined ? { message: error.message } : { message: error.message, code: error.code }; + await events.emitEvent(executionId, 'execution_failed', { error: payload }); + await events.updateStatus(executionId, 'failed', error.message); + return { status: 'failed', error: payload }; } type NodeStatus = 'pending' | 'completed' | 'skipped'; diff --git a/packages/execution-core/src/index.ts b/packages/execution-core/src/index.ts index a7707401f..8338ba1e0 100644 --- a/packages/execution-core/src/index.ts +++ b/packages/execution-core/src/index.ts @@ -1,6 +1,7 @@ export type { BaseNode, NodeErrorPolicy } from '@workflow-builder/types/workflow-execution/execution-model'; export { runGraph } from './graph-runner'; +export type { RunGraphOutcome } from './graph-runner'; export { NodeExecutionError } from './errors'; diff --git a/packages/execution-core/src/workflow.ts b/packages/execution-core/src/workflow.ts index 1c2b3faff..89613b6d9 100644 --- a/packages/execution-core/src/workflow.ts +++ b/packages/execution-core/src/workflow.ts @@ -7,6 +7,7 @@ export type { BaseNode, NodeErrorPolicy } from '@workflow-builder/types/workflow-execution/execution-model'; export { runGraph } from './graph-runner'; +export type { RunGraphOutcome } from './graph-runner'; export { NodeExecutionError } from './errors';