diff --git a/apps/ai-studio/src/data/support-triage-flow.ts b/apps/ai-studio/src/data/support-triage-flow.ts index f611ceccf..9569a46e1 100644 --- a/apps/ai-studio/src/data/support-triage-flow.ts +++ b/apps/ai-studio/src/data/support-triage-flow.ts @@ -108,7 +108,8 @@ Use the exact lowercase keyword on the Type line - it drives downstream routing. id: 'branch-general', sourceHandle: 'source:inner:general', label: 'How-to / Other', - conditions: [], + // A branch with no conditions never matches; an always-true condition is the catch-all. + conditions: [{ x: 'always', y: 'always', comparisonOperator: 'isEqual', logicalOperator: 'AND' }], }, ], }, diff --git a/apps/execution-worker/README.md b/apps/execution-worker/README.md index b2140ba7e..ab3e5054d 100644 --- a/apps/execution-worker/README.md +++ b/apps/execution-worker/README.md @@ -80,11 +80,28 @@ own: one executor per node type and the database as the store port. - **Namespace:** `TEMPORAL_NAMESPACE`, default `default`. Unlike the task queue this is _not_ shared through the plugin: both apps read it through `@workflow-builder/temporal-connection`, but each environment has to set the same value — a mismatch is silent, the worker simply never sees the backend's submissions. - **Workflow ID:** `execution-` — deterministic, lets the backend cancel by execution ID. Also owned by the package. - **Activity timeouts:** DB activities get 30s / 5 retries; node activities (may call LLMs) get 10m / 2 retries. Exported as `DEFAULT_DATABASE_ACTIVITY_PROFILE` and `DEFAULT_NODE_ACTIVITY_PROFILE`. -- **Retries per failure:** an executor throwing `PermanentNodeExecutionError` stops on its first attempt; `TransientNodeExecutionError` retries within the profile's limit. An unclassified throw keeps today's behavior. Of the reference executors, only the AI Agent's `ai_not_configured` is classified (permanent) so far; the rest are still unclassified. +- **Retries per failure:** an executor throwing `PermanentNodeExecutionError` stops on its first attempt; `TransientNodeExecutionError` retries within the profile's limit. An unclassified throw keeps the profile's uniform retry. Every failure the reference executors make a judgment on is in the table below. - **Sandbox constraint:** `workflows.ts` is bundled into V8 with no Web APIs. It may only re-export from `@workflowbuilder/temporal/workflow`, never from the package root. - **Editing the package:** the worker imports its built `dist`, so run `pnpm build:temporal` after changing `packages/temporal/src`. - **Deploys that change the emitted event set:** drain in-flight runs first. Replaying an old run's history against a new emit sequence diverges — see [`replay-audit.md`](../../packages/execution-core/replay-audit.md) rule 9. +### Failure classification + +Each judgment is made at the throw site that owns the error. The runner and the adapter never infer a class from a status code, so a consumer's own executors are unaffected by this table. + +| Failure | Class | Code | +| -------------------------------------------------------------- | --------- | ------------------------------------------- | +| AI Agent: provider answered 401 or 403 | permanent | `provider_auth_rejected` | +| AI Agent: provider answered any other 4xx except 408 and 429 | permanent | `provider_rejected_request` | +| AI Agent: provider answered 429 | transient | `provider_rate_limited` | +| AI Agent: provider answered 5xx | transient | `provider_unavailable` | +| AI Agent: provider answered 408, or the connection failed | transient | `provider_unreachable` | +| AI Agent: `AI_*` variables missing | permanent | `ai_not_configured` | +| AI Agent, Decision: template reference malformed or unresolved | permanent | `template_malformed`, `template_unresolved` | +| Decision: no branch matched | permanent | `no_branch_matched` | + +The provider's own error is attached as `cause`, and `node_failed` reports the deepest non-empty cause's text, so the provider's message reaches the UI as it did before classification. A refused connection is the exception: the SDK reports it as `Cannot connect to API:` with nothing after the colon, because the reason sits in an `AggregateError` it wraps — one entry per address tried. Only messages survive the activity boundary, so the classifier attaches the first entry (`connect ECONNREFUSED ::1:11434`) as the cause instead of the SDK error. The classifier's own message, which names the HTTP status, is one level up and visible only in Temporal's failure record. 409 is permanent on purpose, unlike the AI SDK's own retry default: no chat provider is known to answer 409 for a condition a retry would clear. Two kinds of SDK error stay unclassified and keep the profile's uniform retry: a response the SDK could not parse (a 2xx with a non-JSON body, typically a proxy answering with HTML) and errors raised without any provider response (a malformed tool call from the model, no output generated), which describe model behaviour a retry can change. Marking a failure transient does not buy extra attempts — the node profile still caps them. + ## Adding a new engine 1. Create `src/engines//` with: diff --git a/apps/execution-worker/src/activities/ai-agent.test.ts b/apps/execution-worker/src/activities/ai-agent.test.ts index 277f987b9..74ccdb9b7 100644 --- a/apps/execution-worker/src/activities/ai-agent.test.ts +++ b/apps/execution-worker/src/activities/ai-agent.test.ts @@ -2,10 +2,15 @@ import { APICallError } from 'ai'; import { MockLanguageModelV3 } from 'ai/test'; import { describe, expect, it } from 'vitest'; -import type { ExecutionContext } from '@workflow-builder/execution-core'; +import { + type ExecutionContext, + PermanentNodeExecutionError, + TransientNodeExecutionError, +} from '@workflow-builder/execution-core'; import type { AiAgentNode } from '../domain/ai-studio-nodes'; import { executeAiAgent } from './ai-agent'; +import { apiCallError } from './api-call-error.fixture'; function context(): ExecutionContext { return { @@ -26,6 +31,14 @@ function aiAgentNode(): AiAgentNode { }; } +function failingModel(statusCode: number, message: string): MockLanguageModelV3 { + return new MockLanguageModelV3({ + doGenerate: () => { + throw apiCallError(statusCode, message); + }, + }); +} + describe('executeAiAgent', () => { it('returns the model text as the node output', async () => { const model = new MockLanguageModelV3({ @@ -46,21 +59,40 @@ describe('executeAiAgent', () => { }); it('calls the model exactly once on a retryable failure (retries belong to the Temporal activity policy)', async () => { + // statusCode 500 makes isRetryable default to true — a failure the SDK itself would + // retry, so this assertion fails if client retries ever come back on. + const model = failingModel(500, 'Internal Server Error'); + + await expect(executeAiAgent(aiAgentNode(), context(), { model })).rejects.toThrow(TransientNodeExecutionError); + + expect(model.doGenerateCalls).toHaveLength(1); + }); + + it('surfaces a 5xx as a transient failure that keeps the provider error as its cause', async () => { + const model = failingModel(503, 'upstream overloaded'); + + await expect(executeAiAgent(aiAgentNode(), context(), { model })).rejects.toMatchObject({ + code: 'provider_unavailable', + cause: expect.any(APICallError), + }); + }); + + it('surfaces a rejected API key as a permanent failure', async () => { + const model = failingModel(401, 'Incorrect API key provided'); + + await expect(executeAiAgent(aiAgentNode(), context(), { model })).rejects.toBeInstanceOf( + PermanentNodeExecutionError, + ); + }); + + it('rethrows an error that is not a provider response unchanged', async () => { + const thrown = new Error('mock exploded'); const model = new MockLanguageModelV3({ doGenerate: () => { - // statusCode 500 makes isRetryable default to true — the error must be one - // the SDK would retry, or this test passes even with retries enabled. - throw new APICallError({ - message: 'Internal Server Error', - url: 'https://model.invalid/chat/completions', - requestBodyValues: {}, - statusCode: 500, - }); + throw thrown; }, }); - await expect(executeAiAgent(aiAgentNode(), context(), { model })).rejects.toThrow(APICallError); - - expect(model.doGenerateCalls).toHaveLength(1); + await expect(executeAiAgent(aiAgentNode(), context(), { model })).rejects.toBe(thrown); }); }); diff --git a/apps/execution-worker/src/activities/ai-agent.ts b/apps/execution-worker/src/activities/ai-agent.ts index 313b3f424..02b7b8afc 100644 --- a/apps/execution-worker/src/activities/ai-agent.ts +++ b/apps/execution-worker/src/activities/ai-agent.ts @@ -1,9 +1,15 @@ import { generateText, stepCountIs } from 'ai'; -import { type ExecutionContext, type LoggerPort, resolveTemplate } from '@workflow-builder/execution-core'; +import { + type ExecutionContext, + type LoggerPort, + NodeExecutionError, + resolveTemplate, +} from '@workflow-builder/execution-core'; import type { AiAgentNode } from '../domain/ai-studio-nodes'; import { createWebSearchTool } from '../tools/web-search'; +import { classifyProviderError } from './provider-error'; // Bounds the agentic tool loop so a misbehaving model can't run up cost. const MAX_TOOL_STEPS = 4; @@ -58,14 +64,15 @@ export async function executeAiAgent(node: AiAgentNode, context: ExecutionContex return { output: { response: result.text } }; } catch (error) { - // Mirror the `node_failed` SSE payload shape so a log line and the event line up by executionId. + const failure = classifyProviderError(error); + // executionId joins this line to its node_failed event. const message = error instanceof Error ? error.message : String(error); deps.logger?.error('llm call failed', { workflowId: context.workflowId, executionId: context.executionId, nodeId: node.id, - error: { message }, + error: { message, ...(failure instanceof NodeExecutionError ? { code: failure.code } : {}) }, }); - throw error; + throw failure; } } diff --git a/apps/execution-worker/src/activities/api-call-error.fixture.ts b/apps/execution-worker/src/activities/api-call-error.fixture.ts new file mode 100644 index 000000000..fbafa9888 --- /dev/null +++ b/apps/execution-worker/src/activities/api-call-error.fixture.ts @@ -0,0 +1,10 @@ +import { APICallError } from 'ai'; + +export function apiCallError(statusCode?: number, message = 'provider said no'): APICallError { + return new APICallError({ + message, + url: 'https://model.invalid/chat/completions', + requestBodyValues: {}, + statusCode, + }); +} diff --git a/apps/execution-worker/src/activities/provider-error.test.ts b/apps/execution-worker/src/activities/provider-error.test.ts new file mode 100644 index 000000000..9b42f472c --- /dev/null +++ b/apps/execution-worker/src/activities/provider-error.test.ts @@ -0,0 +1,133 @@ +import { APICallError, InvalidToolInputError, NoOutputGeneratedError, RetryError } from 'ai'; +import { describe, expect, it } from 'vitest'; + +import { PermanentNodeExecutionError, TransientNodeExecutionError } from '@workflow-builder/execution-core'; + +import { apiCallError } from './api-call-error.fixture'; +import { classifyProviderError } from './provider-error'; + +function retryError(errors: Error[]): RetryError { + return new RetryError({ message: 'Failed after 3 attempts', reason: 'maxRetriesExceeded', errors }); +} + +// Mirrors how the SDK renders a failed fetch: its own text plus the caught error's +// message — which is where the dangling colon comes from when there is none. +function unreachable(cause: Error): APICallError { + return new APICallError({ + message: `Cannot connect to API: ${cause.message}`, + url: 'http://localhost:11434/v1/chat/completions', + requestBodyValues: {}, + cause, + }); +} + +describe('classifyProviderError', () => { + it.each([ + [400, PermanentNodeExecutionError, 'provider_rejected_request'], + [401, PermanentNodeExecutionError, 'provider_auth_rejected'], + [402, PermanentNodeExecutionError, 'provider_rejected_request'], + [403, PermanentNodeExecutionError, 'provider_auth_rejected'], + [404, PermanentNodeExecutionError, 'provider_rejected_request'], + [409, PermanentNodeExecutionError, 'provider_rejected_request'], + [413, PermanentNodeExecutionError, 'provider_rejected_request'], + [422, PermanentNodeExecutionError, 'provider_rejected_request'], + [408, TransientNodeExecutionError, 'provider_unreachable'], + [429, TransientNodeExecutionError, 'provider_rate_limited'], + [500, TransientNodeExecutionError, 'provider_unavailable'], + [502, TransientNodeExecutionError, 'provider_unavailable'], + [503, TransientNodeExecutionError, 'provider_unavailable'], + [529, TransientNodeExecutionError, 'provider_unavailable'], + ])('HTTP %i becomes a %o with code %s', (status, ErrorClass, code) => { + const classified = classifyProviderError(apiCallError(status)); + + expect(classified).toBeInstanceOf(ErrorClass); + expect(classified).toMatchObject({ code, message: expect.stringContaining(`HTTP ${status}`) }); + }); + + it('a provider error without a status code (the connection failed) is transient', () => { + const original = apiCallError(); + const classified = classifyProviderError(original); + + expect(classified).toBeInstanceOf(TransientNodeExecutionError); + expect(classified).toMatchObject({ code: 'provider_unreachable', cause: original }); + }); + + it('reports the refused address rather than the SDK text that has none', () => { + // A local provider being down: fetch tries both addresses and reports them in an + // AggregateError of its own, which has no message. + // eslint-disable-next-line unicorn/error-message -- the empty message is the shape under test + const refused = new AggregateError([ + new Error('connect ECONNREFUSED ::1:11434'), + new Error('connect ECONNREFUSED 127.0.0.1:11434'), + ]); + + const classified = classifyProviderError(unreachable(refused)); + + // The cause is what node_failed shows: the runner reports the deepest non-empty + // message, and "Cannot connect to API: " would otherwise be the last one standing. + expect(classified).toMatchObject({ code: 'provider_unreachable', cause: refused.errors[0] }); + }); + + it('keeps the provider error when the connection failure named a reason itself', () => { + const named = unreachable(new AggregateError([new Error('read ECONNRESET')], 'socket hang up')); + + expect(classifyProviderError(named)).toMatchObject({ cause: named }); + }); + + it('keeps the provider error when no entry named a reason either', () => { + // eslint-disable-next-line unicorn/error-message -- the empty messages are the shape under test + const blank = unreachable(new AggregateError([new Error()])); + + expect(classifyProviderError(blank)).toMatchObject({ cause: blank }); + }); + + it('classifies the provider error inside a RetryError, so SDK retries do not hide the status', () => { + const original = apiCallError(401); + + expect(classifyProviderError(retryError([original]))).toMatchObject({ + code: 'provider_auth_rejected', + cause: original, + }); + }); + + it("keeps the provider's own error as the cause, so node_failed still shows the provider's text", () => { + const original = apiCallError(401, 'Incorrect API key provided'); + + expect(classifyProviderError(original)).toMatchObject({ cause: original }); + }); + + it('passes a 2xx the SDK could not parse through unclassified', () => { + // The SDK reports a proxy answering with HTML as an APICallError carrying the 200. + const unparsable = apiCallError(200, 'Failed to process successful response'); + + expect(classifyProviderError(unparsable)).toBe(unparsable); + }); + + // Raised without any provider response, so they describe model behaviour a retry can + // change. Built from the real SDK classes the README names, not from look-alikes. + it.each([ + ['produced no output', new NoOutputGeneratedError({ message: 'No output generated.' })], + [ + 'called a tool with malformed input', + new InvalidToolInputError({ + toolName: 'webSearch', + toolInput: '{"query":', + cause: new Error('Unexpected end of JSON input'), + }), + ], + ])('passes an SDK error saying the model %s through unclassified', (_label, error) => { + expect(classifyProviderError(error)).toBe(error); + }); + + it.each([ + ['a plain Error', new Error('boom')], + ['a non-error value', 'boom'], + ['an object that merely looks like an API error', { statusCode: 500, message: 'not from the SDK' }], + ['a status below the 4xx floor', apiCallError(0)], + ['a redirect the SDK could not follow', apiCallError(302)], + ['a RetryError whose last error is not a provider response', retryError([new Error('mock exploded')])], + ['a RetryError that captured no error at all', retryError([])], + ])('passes %s through unclassified', (_label, error) => { + expect(classifyProviderError(error)).toBe(error); + }); +}); diff --git a/apps/execution-worker/src/activities/provider-error.ts b/apps/execution-worker/src/activities/provider-error.ts new file mode 100644 index 000000000..f286fcb27 --- /dev/null +++ b/apps/execution-worker/src/activities/provider-error.ts @@ -0,0 +1,59 @@ +import { APICallError, RetryError } from 'ai'; + +import { PermanentNodeExecutionError, TransientNodeExecutionError } from '@workflow-builder/execution-core'; + +export function classifyProviderError(error: unknown): unknown { + // With SDK retries enabled the provider error arrives wrapped in a RetryError; + // unwrap it so the status stays readable if maxRetries ever leaves 0. + const providerError = RetryError.isInstance(error) ? error.lastError : error; + if (!APICallError.isInstance(providerError)) return error; + + const status = providerError.statusCode; + const options = { cause: providerError }; + + if (status === undefined) { + return new TransientNodeExecutionError('provider_unreachable', 'Could not reach the provider', { + cause: connectionFailureCause(providerError), + }); + } + if (status === 408) { + return new TransientNodeExecutionError('provider_unreachable', 'Provider timed out (HTTP 408)', options); + } + if (status === 429) { + return new TransientNodeExecutionError('provider_rate_limited', 'Provider rate limit hit (HTTP 429)', options); + } + if (status >= 500) { + return new TransientNodeExecutionError( + 'provider_unavailable', + `Provider failed to serve the request (HTTP ${status})`, + options, + ); + } + if (status === 401 || status === 403) { + return new PermanentNodeExecutionError( + 'provider_auth_rejected', + `Provider rejected the API key (HTTP ${status})`, + options, + ); + } + if (status >= 400) { + return new PermanentNodeExecutionError( + 'provider_rejected_request', + `Provider rejected the request (HTTP ${status})`, + options, + ); + } + return error; +} + +// A refused connection reaches the SDK as an AggregateError with no message of its +// own — one entry per address tried — so APICallError renders it as "Cannot connect +// to API: ". Only messages cross the activity boundary, so the entry is picked here, +// at the throw site, while the array still exists. +function connectionFailureCause(error: APICallError): unknown { + const { cause } = error; + if (!(cause instanceof AggregateError) || cause.message !== '') return error; + + const entries: unknown[] = cause.errors; + return entries.find((entry) => entry instanceof Error && entry.message !== '') ?? error; +} diff --git a/apps/execution-worker/src/executors/decision.test.ts b/apps/execution-worker/src/executors/decision.test.ts index ac1f98de2..8dc7ed26b 100644 --- a/apps/execution-worker/src/executors/decision.test.ts +++ b/apps/execution-worker/src/executors/decision.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from 'vitest'; -import { type ExecutionContext, NodeExecutionError } from '@workflow-builder/execution-core'; +import { + type ExecutionContext, + NodeExecutionError, + PermanentNodeExecutionError, +} from '@workflow-builder/execution-core'; import type { DecisionNode } from '../domain/ai-studio-nodes'; import { executeDecision } from './decision'; @@ -59,7 +63,7 @@ describe('executeDecision', () => { expect(result.nextPort).toBe('yes'); }); - it('throws NodeExecutionError with code "no_branch_matched" when nothing matches', () => { + it('throws a permanent NodeExecutionError with code "no_branch_matched" when nothing matches', () => { const node = decisionNode([ { sourceHandle: 'b1', @@ -71,15 +75,14 @@ describe('executeDecision', () => { }, ]); - expect(() => executeDecision(node, context())).toThrowError(NodeExecutionError); + const decide = () => executeDecision(node, context()); - try { - executeDecision(node, context()); - } catch (error) { - expect(error).toBeInstanceOf(NodeExecutionError); - expect((error as NodeExecutionError).code).toBe('no_branch_matched'); - expect((error as NodeExecutionError).message).toMatch(/no matching branch/i); - } + expect(decide).toThrow(PermanentNodeExecutionError); + expect(decide).toThrow( + expect.objectContaining({ code: 'no_branch_matched', message: expect.stringMatching(/no matching branch/i) }), + ); + // A branch with no conditions never matches, so the remediation must not suggest one. + expect(decide).toThrow(expect.objectContaining({ message: expect.stringContaining('always true') })); }); it('treats a branch with no conditions as non-matching (so callers must throw or use explicit operators)', () => { diff --git a/apps/execution-worker/src/executors/decision.ts b/apps/execution-worker/src/executors/decision.ts index 847fa2db5..a5df4ef8a 100644 --- a/apps/execution-worker/src/executors/decision.ts +++ b/apps/execution-worker/src/executors/decision.ts @@ -1,5 +1,5 @@ // Decision executor — picks the first matching branch. -import { type ExecutionContext, NodeExecutionError, resolveTemplate } from '@workflow-builder/execution-core'; +import { type ExecutionContext, PermanentNodeExecutionError, resolveTemplate } from '@workflow-builder/execution-core'; import type { DecisionBranchCondition, DecisionNode } from '../domain/ai-studio-nodes'; @@ -18,12 +18,12 @@ export function executeDecision(node: DecisionNode, context: ExecutionContext): } } - // No silent fallback — surface misconfigured decisions as node_failed. - // Authors must design an explicit catch-all branch (one with no conditions, - // or whose conditions are tautologically true). - throw new NodeExecutionError( + // No silent fallback — surface misconfigured decisions as node_failed. A branch + // with no conditions never matches, so a catch-all needs an always-true condition + // until a branch can be marked as the default (follow-up: decision-default-branch). + throw new PermanentNodeExecutionError( 'no_branch_matched', - `Decision node has no matching branch (evaluated ${node.config.decisionBranches.length} branch(es)) and no default. Add an explicit catch-all branch with no conditions, or fix the existing conditions to cover every input.`, + `Decision node has no matching branch (evaluated ${node.config.decisionBranches.length} branch(es)) and no default. Add a catch-all branch whose condition is always true (for example "isEqual" with the same value on both sides), or fix the existing conditions to cover every input.`, ); } diff --git a/packages/execution-core/README.md b/packages/execution-core/README.md index 81f4587d2..c05ac2d17 100644 --- a/packages/execution-core/README.md +++ b/packages/execution-core/README.md @@ -223,6 +223,8 @@ A modifier triggers **only when the resolved value is strictly `undefined`** - t The strict default is deliberate: a typo in a prompt template should fail the run, not silently leak a broken token into an LLM. The opt-in modifiers exist for fields where the absence of a value is a legitimate runtime state (an optional trigger field, an output that only exists on one branch of a decision). +Both failures are thrown as `PermanentNodeExecutionError` (codes `template_malformed` and `template_unresolved`), so the engine does not retry the node: a retried node receives the same context, and a reference that failed once fails identically every time. + Authors typing references in the workflow builder UI: see the [variable picker guide](https://www.workflowbuilder.io/docs/guides/use-variable-picker/). ## Adding a new workflow engine diff --git a/packages/execution-core/decision-no-match.decision-log.md b/packages/execution-core/decision-no-match.decision-log.md index f0538d188..01de32845 100644 --- a/packages/execution-core/decision-no-match.decision-log.md +++ b/packages/execution-core/decision-no-match.decision-log.md @@ -4,6 +4,8 @@ ### Date: 29.04.2026 +> **Update 15.09.2026.** The throw is now a `PermanentNodeExecutionError` (a subclass of the `NodeExecutionError` described below), so the engine adapter stops the node on its first attempt instead of retrying a verdict that cannot change. The executor itself moved to `apps/execution-worker/src/executors/decision.ts`. The "Explicit default" alternative rejected below is the tracked follow-up `decision-default-branch`. + ## Context The decision executor at `packages/execution-core/src/executors/decision.ts:21–29` was routing execution down `decisionBranches[0]` whenever no branch's conditions matched — silently. No log, no error event, no `node_failed`. The `matchedBranch` in the output reflected the silent fallback, so an event-log audit looked identical to a successful match. diff --git a/packages/execution-core/replay-audit.md b/packages/execution-core/replay-audit.md index 98c758024..8930f3eeb 100644 --- a/packages/execution-core/replay-audit.md +++ b/packages/execution-core/replay-audit.md @@ -28,28 +28,28 @@ The audit therefore focuses on `graph-runner.ts` + `errors.ts` + `redact.ts` (a ## Sources of non-determinism reviewed -| Source | Used in runner? | Verdict | -| ---------------------------------------- | --------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `new Date()` / `Date.now()` | No | ✅ Safe — runner takes no `LoggerPort` precisely so timestamps cannot leak in. Temporal patches `Date` in the sandbox anyway, but we avoid it altogether to keep intent obvious. | -| `Math.random()` / `crypto.randomUUID` | No | ✅ Safe — IDs are passed in via `WorkflowExecutionInput`; the runner never generates one. | -| `setTimeout` / `setInterval` | No | ✅ Safe — runner is pure scheduling logic; any delay primitive would have to go through Temporal's `sleep()` activity-equivalent. | -| `Promise.race` / `Promise.any` | No | ✅ Safe — only `Promise.all` is used, whose result array is positional (matches input order), so it is deterministic given a deterministic input. | -| `Map` iteration order | Yes | ✅ Safe — ES2015+ guarantees insertion order for `Map` and `Set`. Every `Map` in the runner (`adjacency`, `pendingPredecessors`, `liveIncoming`, `status`, `livePruneKind`, `nodeOutputs` object) is keyed by node id and populated in `definition.nodes` order, which the caller controls. | -| `Set` iteration order | No | ✅ Safe — runner uses no `Set`. (`Array.prototype.find` on a small `successors` list replaces what would otherwise be set membership.) | -| `JSON.stringify` key order | No | ✅ Safe — runner never serializes. Events/payloads handed to ports are plain objects; ports serialize outside the sandbox. | -| `process.env`, `fs`, `fetch` | No | ✅ Safe — runner has no I/O. All side effects flow through `EventEmitterPort` / `ActivityRunnerPort`, which are activity proxies in the Temporal adapter. | -| Object property iteration | Yes (`for…of`) | ✅ Safe — `for…of` on a `Map` follows insertion order. Object literal spread (`{ ...nodeOutputs }`) preserves own-property order per ECMA-262 §7.3.21 since 2015; both V8 and Temporal's sandboxed V8 respect this. | -| `Array.prototype.sort` | No | ✅ Safe — no sorting in the runner. If sorting becomes necessary, it must be stable AND use a deterministic comparator based on input data (not, e.g., insertion time). | -| `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 | 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. | -| `errors.ts` `NodeExecutionError` | Yes | ✅ Safe — constructor only calls `super(message, { cause })` and sets `this.name`. No `Date.now()` in the message, no UUID minting, no env reads. The `Permanent`/`Transient` subclasses add a literal field and their own name, nothing else. | -| `errors.ts` `classifyNodeError` | Yes | ✅ Safe — reads two fields off the error and returns a literal. Shape-based on purpose (`instanceof` cannot work across bundled copies of this module), so it never depends on which copy of the class the value came from. | -| `Error.cause` chain traversal | Yes | ✅ Safe — `extractDeepestError` walks the chain to surface the deepest message, and reads a `NodeErrorEnvelope` off `details[0]` where an adapter left one. Both are plain-data reads at a fixed check order. Walk is bounded by `MAX_CAUSE_DEPTH = 16`, so a cyclic chain produced by a buggy adapter (`a.cause = b; b.cause = a`) cannot spin the workflow. Bound is part of the contract — see rule 8. | +| Source | Used in runner? | Verdict | +| ---------------------------------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `new Date()` / `Date.now()` | No | ✅ Safe — runner takes no `LoggerPort` precisely so timestamps cannot leak in. Temporal patches `Date` in the sandbox anyway, but we avoid it altogether to keep intent obvious. | +| `Math.random()` / `crypto.randomUUID` | No | ✅ Safe — IDs are passed in via `WorkflowExecutionInput`; the runner never generates one. | +| `setTimeout` / `setInterval` | No | ✅ Safe — runner is pure scheduling logic; any delay primitive would have to go through Temporal's `sleep()` activity-equivalent. | +| `Promise.race` / `Promise.any` | No | ✅ Safe — only `Promise.all` is used, whose result array is positional (matches input order), so it is deterministic given a deterministic input. | +| `Map` iteration order | Yes | ✅ Safe — ES2015+ guarantees insertion order for `Map` and `Set`. Every `Map` in the runner (`adjacency`, `pendingPredecessors`, `liveIncoming`, `status`, `livePruneKind`, `nodeOutputs` object) is keyed by node id and populated in `definition.nodes` order, which the caller controls. | +| `Set` iteration order | No | ✅ Safe — runner uses no `Set`. (`Array.prototype.find` on a small `successors` list replaces what would otherwise be set membership.) | +| `JSON.stringify` key order | No | ✅ Safe — runner never serializes. Events/payloads handed to ports are plain objects; ports serialize outside the sandbox. | +| `process.env`, `fs`, `fetch` | No | ✅ Safe — runner has no I/O. All side effects flow through `EventEmitterPort` / `ActivityRunnerPort`, which are activity proxies in the Temporal adapter. | +| Object property iteration | Yes (`for…of`) | ✅ Safe — `for…of` on a `Map` follows insertion order. Object literal spread (`{ ...nodeOutputs }`) preserves own-property order per ECMA-262 §7.3.21 since 2015; both V8 and Temporal's sandboxed V8 respect this. | +| `Array.prototype.sort` | No | ✅ Safe — no sorting in the runner. If sorting becomes necessary, it must be stable AND use a deterministic comparator based on input data (not, e.g., insertion time). | +| `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 | 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. | +| `errors.ts` `NodeExecutionError` | Yes | ✅ Safe — constructor only calls `super(message, { cause })` and sets `this.name`. No `Date.now()` in the message, no UUID minting, no env reads. The `Permanent`/`Transient` subclasses add a literal field and their own name, nothing else. | +| `errors.ts` `classifyNodeError` | Yes | ✅ Safe — reads two fields off the error and returns a literal. Shape-based on purpose (`instanceof` cannot work across bundled copies of this module), so it never depends on which copy of the class the value came from. | +| `Error.cause` chain traversal | Yes | ✅ Safe — `extractDeepestError` walks the chain to surface the deepest non-empty message, and reads a `NodeErrorEnvelope` off `details[0]` where an adapter left one. Both are plain-data reads at a fixed check order. Walk is bounded by `MAX_CAUSE_DEPTH = 16`, so a cyclic chain produced by a buggy adapter (`a.cause = b; b.cause = a`) cannot spin the workflow. Bound is part of the contract — see rule 8. | ## How activities preserve determinism for the runner diff --git a/packages/execution-core/src/errors.test.ts b/packages/execution-core/src/errors.test.ts index 2032b0a2d..ca97dc5ce 100644 --- a/packages/execution-core/src/errors.test.ts +++ b/packages/execution-core/src/errors.test.ts @@ -29,7 +29,7 @@ describe('classifyNodeError', () => { }); it('leaves the base NodeExecutionError and a plain Error unclassified', () => { - expect(classifyNodeError(new NodeExecutionError('no_branch_matched', 'No branch'))).toBeUndefined(); + expect(classifyNodeError(new NodeExecutionError('test_unclassified', 'Unclassified failure'))).toBeUndefined(); expect(classifyNodeError(new Error('boom'))).toBeUndefined(); expect(classifyNodeError('not an error')).toBeUndefined(); }); @@ -118,6 +118,21 @@ describe('extractDeepestError — classification envelope', () => { expect(extractDeepestError(failure).attempt).toBeUndefined(); }); + it('falls back to the nearest non-empty message when the deepest cause has none', () => { + // The dangling colon is correct here, not a defect: an AggregateError crosses an + // adapter boundary without its entries, so a throw site that still holds them picks + // one itself (see connectionFailureCause in the execution worker). + // eslint-disable-next-line unicorn/error-message -- the empty message is the shape under test + const socket = new AggregateError([new Error('connect ECONNREFUSED ::1:11434')]); + const provider = new Error('Cannot connect to API: ', { cause: socket }); + const failure = new NodeExecutionError('provider_unreachable', 'Could not reach the provider', { cause: provider }); + + expect(extractDeepestError(failure)).toMatchObject({ + message: 'Cannot connect to API: ', + code: 'provider_unreachable', + }); + }); + it('reports no attempt when nothing in the chain carries an envelope', () => { expect(extractDeepestError(new Error('boom'))).toEqual({ message: 'boom', diff --git a/packages/execution-core/src/errors.ts b/packages/execution-core/src/errors.ts index 2fc7fd020..43ca74728 100644 --- a/packages/execution-core/src/errors.ts +++ b/packages/execution-core/src/errors.ts @@ -91,7 +91,7 @@ function readEnvelope(error: Error): NodeErrorEnvelope | undefined { } /** - * Walks the ES2022 `Error.cause` chain to the deepest cause and returns its + * Walks the ES2022 `Error.cause` chain and returns the deepest non-empty * message. Adapters that wrap activity throws (Temporal's `ActivityFailure` * is the canonical example) expose a generic top-level message * ("Activity task failed") while keeping the real reason one or two levels @@ -119,10 +119,13 @@ const MAX_CAUSE_DEPTH = 16; export function extractDeepestError(error: unknown): { message: string; code?: string; attempt?: number } { let current: unknown = error; + let message = ''; let code: string | undefined; let attempt: number | undefined; for (let depth = 0; depth < MAX_CAUSE_DEPTH && current instanceof Error; depth++) { + // The deepest non-empty message wins: a failed fetch ends in an AggregateError with none. + if (current.message !== '') message = current.message; if (code === undefined && current instanceof NodeExecutionError) { code = current.code; } @@ -137,9 +140,7 @@ export function extractDeepestError(error: unknown): { message: string; code?: s current = current.cause; } - return { - message: current instanceof Error ? current.message : String(current), - code, - attempt, - }; + if (!(current instanceof Error) && String(current) !== '') message = String(current); + + return { message, code, attempt }; } diff --git a/packages/execution-core/src/graph-runner.test.ts b/packages/execution-core/src/graph-runner.test.ts index 765cd46ea..23af952bc 100644 --- a/packages/execution-core/src/graph-runner.test.ts +++ b/packages/execution-core/src/graph-runner.test.ts @@ -7,7 +7,7 @@ import type { WorkflowEdgeDefinition, } from '@workflow-builder/types/workflow-execution/execution-model'; -import { NodeExecutionError } from './errors'; +import { NodeExecutionError, PermanentNodeExecutionError } from './errors'; import { runGraph } from './graph-runner'; import type { ActivityRunnerPort } from './ports/activity-runner.port'; import type { EventEmitterPort } from './ports/event-emitter.port'; @@ -416,14 +416,13 @@ describe('runGraph — topological scheduling', () => { }); it('NodeExecutionError thrown by an executor — code propagated into node_failed payload', async () => { - // Decision executor throws NodeExecutionError with a structured code when - // no branch matches. The runner's catch must forward that code into the + // The runner's catch must forward an executor's structured code into the // node_failed event's error payload (the existing ExecutionErrorPayload // already declares `code?: string` — this test pins down that wiring). const runner: ActivityRunnerPort = { async executeNode(node) { if (node.id === 'D') { - throw new NodeExecutionError('no_branch_matched', 'Decision node has no matching branch'); + throw new NodeExecutionError('test_unclassified', 'Unclassified failure'); } return { output: `out-${node.id}` }; }, @@ -434,13 +433,13 @@ describe('runGraph — topological scheduling', () => { const nodeFailed = events.events.find((event) => event.type === 'node_failed' && event.nodeId === 'D'); expect(nodeFailed?.payload).toEqual({ - error: { message: 'Decision node has no matching branch', code: 'no_branch_matched' }, + error: { message: 'Unclassified failure', code: 'test_unclassified' }, }); expect(events.events.some((event) => event.type === 'execution_failed')).toBe(true); expect(events.statuses.at(-1)).toEqual({ status: 'failed', - errorMessage: 'Decision node has no matching branch', + errorMessage: 'Unclassified failure', }); }); @@ -493,7 +492,7 @@ describe('runGraph — topological scheduling', () => { // ("Malformed template reference: …", LLM rate-limited, DB timeout) // behind the same opaque string. The runner must walk the chain. const wrapped = new Error('Activity task failed', { - cause: new Error('Malformed template reference: {{nodes.foo?bar}}'), + cause: new PermanentNodeExecutionError('template_malformed', 'Malformed template reference: {{nodes.foo?bar}}'), }); const runner: ActivityRunnerPort = { async executeNode(node) { @@ -507,7 +506,7 @@ describe('runGraph — topological scheduling', () => { const nodeFailed = events.events.find((event) => event.type === 'node_failed' && event.nodeId === 'B'); expect(nodeFailed?.payload).toEqual({ - error: { message: 'Malformed template reference: {{nodes.foo?bar}}' }, + error: { message: 'Malformed template reference: {{nodes.foo?bar}}', code: 'template_malformed' }, }); expect(events.statuses.at(-1)?.errorMessage).toBe('Malformed template reference: {{nodes.foo?bar}}'); }); diff --git a/packages/execution-core/src/templates/resolve-template.test.ts b/packages/execution-core/src/templates/resolve-template.test.ts index ba08e04ee..67c1ee104 100644 --- a/packages/execution-core/src/templates/resolve-template.test.ts +++ b/packages/execution-core/src/templates/resolve-template.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest'; +import { PermanentNodeExecutionError } from '../errors'; import type { ExecutionContext } from '../execution-context'; import { resolveTemplate } from './resolve-template'; @@ -238,3 +239,22 @@ describe('resolveTemplate — malformed templates throw loudly', () => { expect(() => resolveTemplate('{{nodes.foo*bar}}', makeContext())).toThrow(/Malformed template reference/); }); }); + +const resolving = (template: string) => () => resolveTemplate(template, makeContext()); + +describe('resolveTemplate — failures are permanent', () => { + it('a malformed reference is a permanent template_malformed failure', () => { + expect(resolving('{{nodes.foo?bar}}')).toThrow(PermanentNodeExecutionError); + expect(resolving('{{nodes.foo?bar}}')).toThrow(expect.objectContaining({ code: 'template_malformed' })); + }); + + it('a missing path is a permanent template_unresolved failure', () => { + expect(resolving('{{nodes.missing}}')).toThrow(PermanentNodeExecutionError); + expect(resolving('{{nodes.missing}}')).toThrow(expect.objectContaining({ code: 'template_unresolved' })); + }); + + it('an unknown namespace is a permanent template_unresolved failure', () => { + expect(resolving('{{unknown.x}}')).toThrow(PermanentNodeExecutionError); + expect(resolving('{{unknown.x}}')).toThrow(expect.objectContaining({ code: 'template_unresolved' })); + }); +}); diff --git a/packages/execution-core/src/templates/resolve-template.ts b/packages/execution-core/src/templates/resolve-template.ts index fad19ca83..fa56da289 100644 --- a/packages/execution-core/src/templates/resolve-template.ts +++ b/packages/execution-core/src/templates/resolve-template.ts @@ -11,6 +11,7 @@ // Keeping plain `{{x.y}}` strict is deliberate: a typo in a prompt template // should fail loudly during development. The safe forms are an opt-in for // authors who genuinely expect a value to be absent some of the time. +import { PermanentNodeExecutionError } from '../errors'; import type { ExecutionContext } from '../execution-context'; // Two-stage parse: the OUTER regex catches anything that *looks* like a @@ -54,7 +55,7 @@ export function resolveTemplate(template: string, context: ExecutionContext): st return template.replaceAll(OUTER_TEMPLATE_REGEX, (match) => { const groups = PARSE_REGEX.exec(match)?.groups; if (!groups) { - throw new Error(`Malformed template reference: ${match}`); + throw new PermanentNodeExecutionError('template_malformed', `Malformed template reference: ${match}`); } const { namespace, path, safe, default: defaultValue } = groups; @@ -64,7 +65,7 @@ export function resolveTemplate(template: string, context: ExecutionContext): st if (value === undefined) { if (safe === '?') return ''; if (defaultValue !== undefined) return defaultValue; - throw new Error(`Unresolved template reference: ${match}`); + throw new PermanentNodeExecutionError('template_unresolved', `Unresolved template reference: ${match}`); } return typeof value === 'string' ? value : JSON.stringify(value); @@ -86,7 +87,10 @@ function resolveNamespace(namespace: string, context: ExecutionContext, match: s return context.global; } default: { - throw new Error(`Unresolved template reference: ${match} (unknown namespace "${namespace}")`); + throw new PermanentNodeExecutionError( + 'template_unresolved', + `Unresolved template reference: ${match} (unknown namespace "${namespace}")`, + ); } } } diff --git a/packages/temporal/test/activities.test.ts b/packages/temporal/test/activities.test.ts index 01ded6d1d..0f78af9e1 100644 --- a/packages/temporal/test/activities.test.ts +++ b/packages/temporal/test/activities.test.ts @@ -126,7 +126,7 @@ const failing = { id: 'a', type: 'test/echo', config: {} } as const; describe('executeNode — error classification', () => { it.each([ ['a plain Error', new Error('boom')], - ['an unclassified NodeExecutionError', new NodeExecutionError('no_branch_matched', 'No branch')], + ['an unclassified NodeExecutionError', new NodeExecutionError('test_unclassified', 'Unclassified failure')], ])('rethrows %s as the very same object', async (_label, thrown) => { // Unclassified throws must reach the SDK's own conversion untouched. await expect(activitiesThrowing(thrown).executeNode(failing, context)).rejects.toBe(thrown); diff --git a/packages/temporal/test/error-boundary.test.ts b/packages/temporal/test/error-boundary.test.ts index ce08fea5e..f5fe58024 100644 --- a/packages/temporal/test/error-boundary.test.ts +++ b/packages/temporal/test/error-boundary.test.ts @@ -13,6 +13,7 @@ import { type NodeExecutorRegistry, PermanentNodeExecutionError, RUN_WORKFLOW_NAME, + TransientNodeExecutionError, WorkflowBuilderPlugin, type WorkflowDefinition, type WorkflowExecutionInput, @@ -117,13 +118,36 @@ describe('error classification across the activity boundary', () => { expect((failure as WorkflowFailedError).cause).toMatchObject({ type: 'ai_not_configured' }); }, 60_000); + it('a transient throw retries per the profile and reports the attempt it died on', async () => { + const { store, attempts, failure } = await run( + 'transient', + () => + new TransientNodeExecutionError('provider_unavailable', 'Provider failed to serve the request (HTTP 503)', { + cause: new Error('upstream overloaded'), + }), + ); + + expect(attempts).toBe(DEFAULT_NODE_ACTIVITY_PROFILE.retry.maximumAttempts); + // The deepest non-empty cause message is what reaches node_failed, so a wrapped provider + // error keeps showing the provider's own text next to the code. + expect(nodeFailedPayload(store)).toEqual({ + error: { + message: 'upstream overloaded', + code: 'provider_unavailable', + attempt: DEFAULT_NODE_ACTIVITY_PROFILE.retry.maximumAttempts, + }, + }); + expect(store.statuses.at(-1)).toMatchObject({ status: 'failed' }); + expect((failure as WorkflowFailedError).cause).toMatchObject({ type: 'provider_unavailable' }); + }, 60_000); + it('an unclassified throw retries per the profile and is reported exactly as before', async () => { const { store, attempts } = await run( 'unclassified', - () => new NodeExecutionError('no_branch_matched', 'No branch matched'), + () => new NodeExecutionError('test_unclassified', 'Unclassified failure'), ); expect(attempts).toBe(DEFAULT_NODE_ACTIVITY_PROFILE.retry.maximumAttempts); - expect(nodeFailedPayload(store)).toEqual({ error: { message: 'No branch matched' } }); + expect(nodeFailedPayload(store)).toEqual({ error: { message: 'Unclassified failure' } }); }, 60_000); });