fix(execution-core): Make failed runs close as Failed in Temporal

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<AiStudioNode>): Promise<void> {
+  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<AiStudioNode>):
     }
     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');
+  }
 }

packages/execution-core/README.md

@@ -106,32 +106,34 @@ The registry's mapped type — `{ [K in TNode['type']]: NodeExecutor<Extract<TNo
 
 Each node can declare an `errorPolicy` on its `BaseNode` (sibling to `config`). The runner consults it after catching a node error and decides whether to propagate, absorb, or route the failure.
 
-| Policy       | When the node throws                                                                                                                                                                         | Use case                                                |
-| ------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------- |
-| `'fail'`     | (default) Emit `node_failed`, then abort the workflow with `execution_failed`.                                                                                                               | Unrecoverable infra / programming bugs.                 |
-| `'continue'` | Emit `node_failed`, set `nodeOutputs[id] = { error: { message, code? } }`, schedule downstream nodes through every outgoing edge **except** those tagged with the reserved `'error'` handle. | Best-effort steps; downstream inspects the error.       |
-| `'route'`    | Emit `node_failed`, set the same `{ error }` output, but only follow outgoing edges whose `sourceHandle === 'error'`. The success branch is pruned by the standard skip-propagation path.    | Retry-with-fallback, send-to-DLQ, compensating actions. |
+| Policy         | When the node throws                                                                                                                                                                                       | Use case                                                |
+| -------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------- |
+| `'fail'`       | (default) Emit `node_failed`, then abort the workflow with `execution_failed` and end the run as `{ status: 'failed' }`, which the engine adapter surfaces as a failed run (Temporal closes it as Failed). | Unrecoverable infra / programming bugs.                 |
+| `'continue'`   | Emit `node_failed`, set `nodeOutputs[id] = { error: { message, code? } }`, schedule downstream nodes through every outgoing edge **except** those tagged with the reserved `'errorRoute'` handle.          | Best-effort steps; downstream inspects the error.       |
+| `'errorRoute'` | Emit `node_failed`, set the same `{ error }` output, but only follow outgoing edges whose `sourceHandle === 'errorRoute'`. The success branch is pruned by the standard skip-propagation path.             | Retry-with-fallback, send-to-DLQ, compensating actions. |
 
-`'route'` piggybacks on the same `nextPort` mechanism decision nodes use — non-`'error'` edges are pruned through the standard skip-propagation path, so deep dead branches stay dormant.
+`'errorRoute'` piggybacks on the same `nextPort` mechanism decision nodes use — non-`'errorRoute'` edges are pruned through the standard skip-propagation path, so deep dead branches stay dormant.
 
-### `'error'` is a reserved `sourceHandle`
+Only `'fail'` ends the run as failed. A node that fails under `'continue'` or `'errorRoute'` is absorbed by the graph: `node_failed` is still emitted, so the failure stays visible to anyone tailing events, but the run itself completes — `runGraph` returns `{ status: 'completed' }` and the engine reports a successful run.
 
-The string `'error'` is reserved as the runner's error-routing port name. Edges tagged with `sourceHandle === 'error'` fire **only** when the upstream node failed with policy `'route'`. Every other propagation path — success, `'continue'` on error, decision branching — prunes them. That means:
+### `'errorRoute'` is a reserved `sourceHandle`
+
+The string `'errorRoute'` is reserved as the runner's error-routing port name — deliberately the same literal as the policy, so the wiring reads consistently from schema to runner. Edges tagged with `sourceHandle === 'errorRoute'` fire **only** when the upstream node failed with policy `'errorRoute'`. Every other propagation path — success, `'continue'` on error, decision branching — prunes them. That means:
 
 - A successful node with an unconnected error branch never fires it.
 - A `'continue'` failure flows the error output to **regular** downstream edges only; the dedicated error branch stays dormant.
-- Decision nodes must not use `'error'` as a branch handle.
+- Decision nodes must not use `'errorRoute'` as a branch handle.
 
 ```ts
 const node: MyNode = {
   id: 'fetch-customer',
   type: 'my/http-call',
   config: { url: '…' },
-  errorPolicy: 'route',
+  errorPolicy: 'errorRoute',
 };
 ```
 
-If a node with `'route'` policy fails but has no outgoing edge tagged `'error'`, the run completes cleanly — the failure is recorded as `node_failed` and nothing else fires. That makes `'route'` usable as a silent DLQ when paired with downstream observability on `node_failed` events.
+If a node with `'errorRoute'` policy fails but has no outgoing edge tagged `'errorRoute'`, the run completes cleanly — the failure is recorded as `node_failed` and nothing else fires. That makes `'errorRoute'` usable as a silent DLQ when paired with downstream observability on `node_failed` events.
 
 ## Template references
 
@@ -154,6 +156,7 @@ Authors typing references in the workflow builder UI: see the [variable picker g
 1. Implement `WorkflowEnginePort<TNode>` (`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
 

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.                                                                                                                              |

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 () => {