diff --git a/apps/backend/README.md b/apps/backend/README.md index 7e932a9d3..184a09468 100644 --- a/apps/backend/README.md +++ b/apps/backend/README.md @@ -39,7 +39,24 @@ The request is validated on `POST /:id/publish` and `POST /:id/execute`, never o One key is refused outright, wherever it sits. An own `__proto__` anywhere in the snapshot answers `invalid_snapshot` 400 naming its path: `JSON.parse` turns it into an ordinary key, and a loose object copies unknown keys by assignment, which for that one swaps the parsed output's prototype and hands the engine a request no schema ever saw. The check does not weigh position, so it also refuses a `__proto__` buried inside an opaque node property, where zod never copies keys one by one and the key is inert. A node type that keeps a raw JSON document in `data.properties` therefore cannot carry one. -A submitted decision is checked against the request by `validateSubmittedDecision` in `src/domain/decision/`; the decision endpoint that calls it is a separate change. Shape, rules and the reasoning are in [`decision-request.decision-log.md`](./decision-request.decision-log.md). +A submitted decision is checked against the request by `validateSubmittedDecision` in `src/domain/decision/` and delivered by the endpoint below. Shape, rules and the reasoning are in [`decision-request.decision-log.md`](./decision-request.decision-log.md). + +### Deciding: `POST /api/executions/:id/decision` + +Body: `{ nodeId, attempt, action, edits?, reason?, comment? }`. `action` is the `name` of one of the node's actions. `attempt` is how many times the node has parked in this run (its `node_waiting` count; today always 1). Checks run in this order, each answering before the next: row, authorization (`executions:decide` with the row's `{ workflowId, tenantId, status }`; a deny wins over 404), status, body, node, decision, `attempt`, effect, engine. The engine is asked once; nothing is retried. Success: `200 { executionId, nodeId, attempt, action, effect }`. Codes and messages live in `src/routes/decision-refusals.ts`. + +| Status | Code | When | +| ------ | --------------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | +| 400 | `validation_error` | Body shape | +| 400 | `invalid_decision` | Submission against the request; `details[0].code` is a `SUBMITTED_DECISION_ERRORS` key | +| 404 | `execution_not_found` | | +| 404 | `node_not_found` | Not in the run's snapshot | +| 409 | `execution_not_waiting` | Terminal or cancelling run, or the engine no longer has it | +| 409 | `node_not_waiting` | No request on the node, never parked, or not waiting now. Final | +| 409 | `decision_already_made` | The first decision won, whoever sent it | +| 409 | `decision_attempt_mismatch` | Body carries the current `attempt` | +| 501 | `effect_not_supported` | `rerun-source`, until the engine can re-run a source | +| 503 | `decision_delivery_timeout` | No worker accepted it in time. It may still land: resend (`Retry-After`); `decision_already_made` then names the wait, not the sender | ## Running individual processes diff --git a/apps/backend/auth-port.decision-log.md b/apps/backend/auth-port.decision-log.md index dd12a4546..492493910 100644 --- a/apps/backend/auth-port.decision-log.md +++ b/apps/backend/auth-port.decision-log.md @@ -21,19 +21,20 @@ This decision log records the structural piece (scope L from [`local-dev-binding ## Actions covered today -| Action | Resource | -| ------------------- | ------------------------------------ | -| `workflows:create` | `{ kind: 'workflows' }` | -| `workflows:list` | `{ kind: 'workflows' }` | -| `workflows:read` | `{ kind: 'workflow', workflowId }` | -| `workflows:update` | `{ kind: 'workflow', workflowId }` | -| `workflows:publish` | `{ kind: 'workflow', workflowId }` | -| `workflows:execute` | `{ kind: 'workflow', workflowId }` | -| `executions:read` | `{ kind: 'execution', executionId }` | -| `executions:stream` | `{ kind: 'execution', executionId }` | -| `executions:cancel` | `{ kind: 'execution', executionId }` | - -Per-row resource kinds (`workflow`, `execution`) also accept an optional `attributes: Record`. Routes that already loaded the row can pass it through so ABAC ports do not need to refetch. Pure RBAC ports ignore the field. Routes that load before authorize is wired (see follow-ups on data scoping) will start using it without a breaking change. +| Action | Resource | +| ------------------- | ------------------------------------------------- | +| `workflows:create` | `{ kind: 'workflows' }` | +| `workflows:list` | `{ kind: 'workflows' }` | +| `workflows:read` | `{ kind: 'workflow', workflowId }` | +| `workflows:update` | `{ kind: 'workflow', workflowId }` | +| `workflows:publish` | `{ kind: 'workflow', workflowId }` | +| `workflows:execute` | `{ kind: 'workflow', workflowId }` | +| `executions:read` | `{ kind: 'execution', executionId }` | +| `executions:stream` | `{ kind: 'execution', executionId }` | +| `executions:cancel` | `{ kind: 'execution', executionId }` | +| `executions:decide` | `{ kind: 'execution', executionId, attributes? }` | + +Per-row resource kinds (`workflow`, `execution`) also accept an optional `attributes: Record`. Routes that already loaded the row can pass it through so ABAC ports do not need to refetch. Pure RBAC ports ignore the field. The decision route is the first to load before it authorizes: `attributes` is `{ workflowId, tenantId, status }`, absent when the row does not exist, and a deny is answered before the 404. Hiding which ids exist depends on the port: it must deny when `attributes` is absent too, or a caller learns that 404 means unknown and 403 means someone else's. ## Alternative Options Considered diff --git a/apps/backend/decision-request.decision-log.md b/apps/backend/decision-request.decision-log.md index 63c1b6c80..b52111bc8 100644 --- a/apps/backend/decision-request.decision-log.md +++ b/apps/backend/decision-request.decision-log.md @@ -2,7 +2,7 @@ ### Proposed by: Piotr Błaszczyk -### Date: 07.09.2026 (shape), 08.09.2026 (names) +### Date: 07.09.2026 (shape), 08.09.2026 (names), 10.09.2026 (endpoint) ## Context @@ -25,6 +25,18 @@ The shape itself is documented on the type (`packages/types/src/workflow-executi 11. **An own `__proto__` key anywhere in a snapshot is refused before parsing.** `JSON.parse` makes it an ordinary key, and zod's loose objects copy unknown keys with a plain assignment, which for that key swaps the output's prototype: everything under it then reads back as validated, and the mapper would copy an inherited request into a real field on the way to the engine. Both parsers that preserve unknown keys are wrapped in a preprocess that rejects the key at its path: `workflowSnapshotSchema`, which answers the usual `invalid_snapshot` 400, and `decisionRequestSchema`, which guards itself so a caller parsing raw JSON with it cannot inherit a request no schema checked. `z.record` is immune by construction but cannot type known keys beside unknown ones, and it protects only its own level, so it is no substitute here. +## The endpoint (10.09.2026) + +What the endpoint does and answers is in the README. Only the reasons are here. + +12. **`nodeId` in the body, not the path**, so the pending-decision resource addresses the same node the same way. The route owns the body shape; `validateSubmittedDecision` judges only the rules, the workflow's validator only engine integrity. +13. **The row is read before authorization** so a port can scope by it; a deny therefore wins over 404 and reveals no id. +14. **Only terminal and `cancelling` runs are refused by status.** The status write is best-effort, so a parked run may read `pending`; the engine is the arbiter. +15. **`attempt` is the node's `node_waiting` count.** The engine has no attempt yet, so the check is not atomic with delivery; it holds only while a node parks at most once. The rerun loop has to carry the wait instance into the engine `(follow-up: decision-attempt-in-engine)`. +16. **A second submission is always 409 `decision_already_made`.** The backend stores nothing about a decision, so it cannot tell a repeat from a contradiction; a byte-identical replay needs a caller key `(follow-up: decision-idempotency-key)`. The Temporal update id stays random: a deterministic one would hand a second decider the first one's outcome. +17. **`rerun-source` is 501** until the engine can re-run a source; the LLM budget guard and rate limit move there with the verb `(follow-up: decision-rerun-source)`. +18. **`node_not_waiting` is final** because the runner registers the wait before announcing it. **`delivery_timeout` is 503 with a hedged message** because an update nobody accepted is not durable, yet the server may still hand it to the next worker. + ## Rejected - Detecting the node by its type string: the backend would have to learn every product's vocabulary. @@ -33,6 +45,9 @@ The shape itself is documented on the type (`packages/types/src/workflow-executi - Defaulting `deadline.policy` to `reject`: a timer that rejects is audit-relevant and must be written down, not implied. - Recording the whole action object on the decision: the port and label would then live twice, on the request and in every completion, with two sources of truth about where a verdict routes. - Exporting the duration pattern from the Temporal plugin: a published API widened for one regex; duplicated with a pointer instead `(follow-up: shared-duration-format)`. +- Persisting the first submission only to turn one 409 into a 200. +- Requiring `executions.status === 'waiting'`: it would lock the route to a best-effort write. +- Retrying `node_not_waiting`: the race was fixed at its root, in the runner's order of registering and announcing. ## Known gaps @@ -42,18 +57,17 @@ The shape itself is documented on the type (`packages/types/src/workflow-executi - It checks editability and presence at every level the form describes inline, following `properties` and `items`. A level reached only through `$ref` or a composition keyword describes nothing there, so an edit into it is refused as an unknown field rather than checked `(follow-up: decision-edit-schema-composition)`. - The snapshot schema does not check that edge endpoints exist, so an explicit source with a dangling edge passes. This predates the change. - Node ids are not checked for uniqueness either; with a duplicate, the graph rules see the first node of that id. Also pre-existing `(follow-up: snapshot-node-id-uniqueness)`. +- A reject whose port has no edge ends the run `incomplete`. The terminal-outcome work closes this; the seam is `toNodeResolution`. +- A decision records nothing about who decided. +- The route re-parses the stored snapshot with today's `workflowSnapshotSchema`, and a run can wait for days across deploys. A schema tightened in between makes every parked run whose snapshot no longer parses undecidable: the route answers 500 until the snapshot is migrated or the rule relaxed. -## Open points - -Taken conservatively; confirm or change when the decision endpoint lands. +## Open points, closed 10.09.2026 -- A whitespace-only `reason` counts as missing when `reasonRequired` is set, like a blank comment. -- "Emptied" for a required field means `undefined`, `null` or a whitespace-only string; empty arrays and objects are value validation. -- Edits on a non-`resume` submission are checked but do not change the effect; refusing them with a dedicated code is the recommended alternative. +A whitespace-only `reason` counts as missing, and "emptied" means `undefined`, `null` or whitespace: both kept. Edits on a non-`resume` action are now refused with `edits_not_allowed`, before the field rules; dropping them silently was the worse failure. ## Not in this change -Further request fields (condition, four-eyes, several decisions), identity and `x-pii` masking, the decision endpoint, the pending-decision resource, the rerun loop, the deadline timer, authoring the request in the editor `(follow-up: decision-request-properties-ui)`, and the node that actually parks. The runner learns no product's vocabulary by design, so a run stops where a node's executor returns a waiting result, never because a field is present. The node type whose executor does only that, and therefore waits without side effects of its own, is its own task `(follow-up: human-decision-node)`. +Further request fields (condition, four-eyes, several decisions), identity and `x-pii` masking, the pending-decision resource, the rerun loop, the deadline timer, authoring the request in the editor `(follow-up: decision-request-properties-ui)`, and the node that actually parks. The runner learns no product's vocabulary by design, so a run stops where a node's executor returns a waiting result, never because a field is present. The node type whose executor does only that, and therefore waits without side effects of its own, is its own task `(follow-up: human-decision-node)`. ## Status diff --git a/apps/backend/src/auth/auth-port.ts b/apps/backend/src/auth/auth-port.ts index e977341a1..77b58569a 100644 --- a/apps/backend/src/auth/auth-port.ts +++ b/apps/backend/src/auth/auth-port.ts @@ -36,7 +36,8 @@ export type AuthAction = | 'workflows:execute' | 'executions:read' | 'executions:stream' - | 'executions:cancel'; + | 'executions:cancel' + | 'executions:decide'; /** * Resources passed to `authorize`. The per-row kinds carry an optional diff --git a/apps/backend/src/domain/decision/decision-issues.ts b/apps/backend/src/domain/decision/decision-issues.ts index 78412216b..54fd80db9 100644 --- a/apps/backend/src/domain/decision/decision-issues.ts +++ b/apps/backend/src/domain/decision/decision-issues.ts @@ -29,6 +29,7 @@ export const SUBMITTED_DECISION_ERRORS = { unknown_action: "the decision request offers no action named '{value}'", reason_required: "action '{value}' requires a reason", comment_required: "action '{value}' requires a comment", + edits_not_allowed: "action '{value}' does not take edits", unknown_field: "field '{value}' is not in the decision schema", field_not_editable: "field '{value}' is read-only", required_field_missing: "required field '{value}' must not be emptied", @@ -36,7 +37,7 @@ export const SUBMITTED_DECISION_ERRORS = { export type SubmittedDecisionErrorCode = keyof typeof SUBMITTED_DECISION_ERRORS; -function fill(template: string, value: string | undefined): string { +export function fill(template: string, value: string | undefined): string { // A function replacer, so a value containing `$&` or `$1` lands verbatim. return template.replace('{value}', () => value ?? ''); } diff --git a/apps/backend/src/domain/decision/find-decision-request.test.ts b/apps/backend/src/domain/decision/find-decision-request.test.ts new file mode 100644 index 000000000..80f05fb74 --- /dev/null +++ b/apps/backend/src/domain/decision/find-decision-request.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it } from 'vitest'; + +import { workflowSnapshotSchema } from '../mapper/snapshot-schema'; +import { findDecisionRequest } from './find-decision-request'; + +const approve = { name: 'approve', label: 'Approve', effect: 'resume' }; +const reject = { name: 'reject', label: 'Reject', effect: 'reject' }; + +const snapshot = workflowSnapshotSchema.parse({ + nodes: [ + { id: 'source-1', data: { type: 'product/any', properties: {} } }, + { id: 'plain', data: { type: 'product/any' } }, + { + id: 'review-1', + data: { + type: 'product/any', + properties: { + decisionRequest: { version: 1, actions: [approve, reject], schema: { type: 'object', properties: {} } }, + }, + }, + }, + ], + edges: [{ id: 'e1', source: 'source-1', target: 'review-1' }], +}); + +describe('findDecisionRequest', () => { + it('finds the request of the node, with the parser defaults on its actions', () => { + const found = findDecisionRequest(snapshot, 'review-1'); + + expect(found.error).toBeUndefined(); + expect(found.request?.actions).toEqual([ + { ...approve, port: 'approved' }, + { ...reject, port: 'rejected', reasonRequired: false }, + ]); + }); + + it('tells a missing node from a node that carries no request', () => { + expect(findDecisionRequest(snapshot, 'ghost')).toEqual({ error: 'node_not_found' }); + expect(findDecisionRequest(snapshot, 'source-1')).toEqual({ error: 'node_without_decision_request' }); + expect(findDecisionRequest(snapshot, 'plain')).toEqual({ error: 'node_without_decision_request' }); + }); +}); diff --git a/apps/backend/src/domain/decision/find-decision-request.ts b/apps/backend/src/domain/decision/find-decision-request.ts new file mode 100644 index 000000000..417cce609 --- /dev/null +++ b/apps/backend/src/domain/decision/find-decision-request.ts @@ -0,0 +1,17 @@ +import type { DecisionRequest } from '@workflow-builder/types/workflow-execution/decision-request'; + +import type { WorkflowSnapshot } from '../mapper/snapshot-schema'; + +export type FindDecisionRequestError = 'node_not_found' | 'node_without_decision_request'; + +export type FindDecisionRequestResult = + | { request: DecisionRequest; error?: undefined } + | { request?: undefined; error: FindDecisionRequestError }; + +export function findDecisionRequest(snapshot: WorkflowSnapshot, nodeId: string): FindDecisionRequestResult { + const node = snapshot.nodes.find((candidate) => candidate.id === nodeId); + if (node === undefined) return { error: 'node_not_found' }; + const request = node.data.properties?.decisionRequest; + if (request === undefined) return { error: 'node_without_decision_request' }; + return { request }; +} diff --git a/apps/backend/src/domain/decision/node-resolution.test.ts b/apps/backend/src/domain/decision/node-resolution.test.ts new file mode 100644 index 000000000..aec0f38bb --- /dev/null +++ b/apps/backend/src/domain/decision/node-resolution.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, expectTypeOf, it } from 'vitest'; + +import type { Decision } from '@workflow-builder/types/workflow-execution/decision-request'; + +import { type RoutedDecision, type RoutedDecisionAction, hasNodeResolution, toNodeResolution } from './node-resolution'; + +const approve = { name: 'approve', label: 'Approve', effect: 'resume', port: 'approved' } as const; +const reject = { name: 'reject', label: 'Reject', effect: 'reject', port: 'rejected', reasonRequired: false } as const; + +describe('toNodeResolution', () => { + it.each<{ decision: RoutedDecision; action: RoutedDecisionAction; port: string }>([ + { decision: { action: 'approve', effect: 'resume', edits: {} }, action: approve, port: 'approved' }, + { + decision: { action: 'approve', effect: 'resume-with-edits', edits: { refundAmount: 120 } }, + action: approve, + port: 'approved', + }, + { decision: { action: 'reject', effect: 'reject', edits: {}, reason: 'late' }, action: reject, port: 'rejected' }, + ])('routes a $decision.effect decision on the action port', ({ decision, action, port }) => { + const completion = toNodeResolution(decision, action); + + expect(completion.nextPort).toBe(port); + expect(completion.output).toBe(decision); + expect(Object.keys(completion)).toEqual(['output', 'nextPort']); + }); + + it('hands the decision over as it is, with no key added or removed', () => { + const decision: RoutedDecision = { action: 'approve', effect: 'resume-with-edits', edits: { refundAmount: 120 } }; + const before = structuredClone(decision); + + const completion = toNodeResolution(decision, approve); + + expect(completion.output).toBe(decision); + expect(decision).toEqual(before); + }); + + it('routes a reject on its port even when no reason was given', () => { + const completion = toNodeResolution({ action: 'reject', effect: 'reject', edits: {} }, reject); + + expect(completion).toEqual({ output: { action: 'reject', effect: 'reject', edits: {} }, nextPort: 'rejected' }); + }); + + it('refuses the reserved errorRoute port, which the request parser already forbids', () => { + const decision: RoutedDecision = { action: 'approve', effect: 'resume', edits: {} }; + + expect(() => toNodeResolution(decision, { ...approve, port: 'errorRoute' })).toThrow("reserved 'errorRoute'"); + }); + + it('has no completion for a rerun-source decision', () => { + expect(hasNodeResolution({ action: 'ask-again', effect: 'rerun-source', edits: {}, comment: 'again' })).toBe(false); + expect(hasNodeResolution({ action: 'approve', effect: 'resume', edits: {} })).toBe(true); + expectTypeOf().toEqualTypeOf<'resume' | 'resume-with-edits' | 'reject'>(); + expectTypeOf>().toEqualTypeOf(); + expectTypeOf().toMatchTypeOf(); + }); +}); diff --git a/apps/backend/src/domain/decision/node-resolution.ts b/apps/backend/src/domain/decision/node-resolution.ts new file mode 100644 index 000000000..9c8a27f4e --- /dev/null +++ b/apps/backend/src/domain/decision/node-resolution.ts @@ -0,0 +1,21 @@ +import type { CompletedNodeExecution } from '@workflow-builder/execution-core/workflow'; +import type { + Decision, + DecisionAction, + DecisionEffect, +} from '@workflow-builder/types/workflow-execution/decision-request'; + +// No completion exists for `rerun-source` yet (follow-up: decision-rerun-source). +export type RoutedDecision = Decision & { effect: Exclude }; +export type RoutedDecisionAction = Exclude; + +export function hasNodeResolution(decision: Decision): decision is RoutedDecision { + return decision.effect !== 'rerun-source'; +} + +export function toNodeResolution(decision: RoutedDecision, action: RoutedDecisionAction): CompletedNodeExecution { + if (action.port === 'errorRoute') { + throw new Error(`action '${action.name}' routes to the reserved 'errorRoute' port`); + } + return { output: decision, nextPort: action.port }; +} diff --git a/apps/backend/src/domain/decision/validate-submitted-decision.test.ts b/apps/backend/src/domain/decision/validate-submitted-decision.test.ts index caa3fe817..08582a2a9 100644 --- a/apps/backend/src/domain/decision/validate-submitted-decision.test.ts +++ b/apps/backend/src/domain/decision/validate-submitted-decision.test.ts @@ -91,6 +91,7 @@ describe('validateSubmittedDecision', () => { effect: 'resume-with-edits', }, { name: 'a reject without a reason when none is required', call: { action: 'reject' }, effect: 'reject' }, + { name: 'a reject with empty edits', call: { action: 'reject', edits: {} }, effect: 'reject' }, { name: 'a reject with a reason when one is required', request: requestWith({ actions: [approve, { ...reject, reasonRequired: true }] }), @@ -102,6 +103,11 @@ describe('validateSubmittedDecision', () => { call: { action: 'ask-again', comment: 'Use the discounted price' }, effect: 'rerun-source', }, + { + name: 'a rerun with empty edits', + call: { action: 'ask-again', comment: 'again', edits: {} }, + effect: 'rerun-source', + }, ])('accepts $name', ({ request = requestWith(), call, effect }) => { const result = validateSubmittedDecision(request, call); @@ -111,18 +117,6 @@ describe('validateSubmittedDecision', () => { expect(result.action?.name).toBe(call.action); }); - // Open point in the decision log: until decided otherwise, edits on a non-resume action - // are checked field by field, kept on the decision, and leave the effect alone. - it.each<{ call: SubmittedDecision; effect: string }>([ - { call: { action: 'reject', reason: 'late', edits: { refundAmount: 0 } }, effect: 'reject' }, - { call: { action: 'ask-again', comment: 'again', edits: { note: 'x' } }, effect: 'rerun-source' }, - ])('keeps the declared effect $effect and carries the edits of a non-resume submission', ({ call, effect }) => { - const result = validateSubmittedDecision(requestWith(), call); - - expect(result.decision?.effect).toBe(effect); - expect(result.decision?.edits).toEqual(call.edits); - }); - it.each<{ name: string; request?: DecisionRequest; @@ -191,6 +185,27 @@ describe('validateSubmittedDecision', () => { value: 'ask-again', path: ['comment'], }, + { + name: 'a reject carrying edits', + call: { action: 'reject', reason: 'late', edits: { refundAmount: 0 } }, + code: 'edits_not_allowed', + value: 'reject', + path: ['edits'], + }, + { + name: 'a rerun carrying edits', + call: { action: 'ask-again', comment: 'again', edits: { note: 'x' } }, + code: 'edits_not_allowed', + value: 'ask-again', + path: ['edits'], + }, + { + name: 'a reject carrying an edit on a read-only field, named for the real problem', + call: { action: 'reject', edits: { orderDate: '2026-01-01' } }, + code: 'edits_not_allowed', + value: 'reject', + path: ['edits'], + }, { name: 'an edit on a read-only field', call: { action: 'approve', edits: { orderDate: '2026-01-01' } }, diff --git a/apps/backend/src/domain/decision/validate-submitted-decision.ts b/apps/backend/src/domain/decision/validate-submitted-decision.ts index 0000651f0..23e10b0cb 100644 --- a/apps/backend/src/domain/decision/validate-submitted-decision.ts +++ b/apps/backend/src/domain/decision/validate-submitted-decision.ts @@ -9,9 +9,8 @@ import type { import { type SubmittedDecisionErrorCode, submittedDecisionErrorMessage } from './decision-issues'; -// The shape of what the decider sent. The caller parses a body with this before calling -// `validateSubmittedDecision`, which assumes the shape and checks only the rules. -// Provisional: the decision endpoint owns the public request shape and may rename fields. +// Parsed at the endpoint, which extends it with `nodeId` and `attempt`, before +// `validateSubmittedDecision` checks the rules on the parsed shape. export const submittedDecisionSchema = z.object({ action: z.string(), edits: z.record(z.string(), z.unknown()).optional(), @@ -120,7 +119,12 @@ export function validateSubmittedDecision( return refuse('comment_required', action.name, ['comment']); } + // Before the walk, so the refusal names the edits and not one field. const edits = submitted.edits ?? {}; + if (action.effect !== 'resume' && Object.keys(edits).length > 0) { + return refuse('edits_not_allowed', action.name, ['edits']); + } + const refused = validateEdits(request.schema, edits, ['edits']); if (refused !== undefined) return refused; diff --git a/apps/backend/src/events/count-node-waits.ts b/apps/backend/src/events/count-node-waits.ts new file mode 100644 index 000000000..3a78b7dd0 --- /dev/null +++ b/apps/backend/src/events/count-node-waits.ts @@ -0,0 +1,21 @@ +import { and, count, eq } from 'drizzle-orm'; + +import type { ExecutionEventType } from '@workflow-builder/types/workflow-execution/execution-events'; + +import { database } from '../db/client'; +import { executionEvents } from '../db/schema'; + +// How many times the node has parked in this run: the wait instance a decision addresses. +export async function countNodeWaits(executionId: string, nodeId: string): Promise { + const [row] = await database + .select({ waits: count() }) + .from(executionEvents) + .where( + and( + eq(executionEvents.executionId, executionId), + eq(executionEvents.nodeId, nodeId), + eq(executionEvents.type, 'node_waiting' satisfies ExecutionEventType), + ), + ); + return row?.waits ?? 0; +} diff --git a/apps/backend/src/routes/backend-env.ts b/apps/backend/src/routes/backend-env.ts new file mode 100644 index 000000000..ebfde66e8 --- /dev/null +++ b/apps/backend/src/routes/backend-env.ts @@ -0,0 +1,4 @@ +import type { AuthVariables } from '../auth'; +import type { TenantVariables } from '../tenant'; + +export type BackendEnv = { Variables: AuthVariables & TenantVariables }; diff --git a/apps/backend/src/routes/decision-refusals.test.ts b/apps/backend/src/routes/decision-refusals.test.ts new file mode 100644 index 000000000..f63212117 --- /dev/null +++ b/apps/backend/src/routes/decision-refusals.test.ts @@ -0,0 +1,49 @@ +import { Hono } from 'hono'; +import { describe, expect, it } from 'vitest'; + +import { DECISION_REFUSALS, DECISION_REFUSAL_STATUS, type DecisionRefusal, refuse } from './decision-refusals'; + +async function answer(refusal: DecisionRefusal, value?: string, extra?: Record) { + const app = new Hono().get('/', (c) => refuse(c, refusal, { value, extra })); + const response = await app.request('/'); + return { + status: response.status, + retryAfter: response.headers.get('retry-after'), + body: (await response.json()) as Record, + }; +} + +describe('decision refusals', () => { + it('every code is answered in at least one situation', () => { + const answered = new Set(Object.values(DECISION_REFUSALS).map((refusal) => refusal.code)); + + expect([...answered].sort()).toEqual(Object.keys(DECISION_REFUSAL_STATUS).sort()); + }); + + it('answers with the code, its status, the filled message and whatever else the route adds', async () => { + expect(await answer('node_not_found', '$&-$1')).toEqual({ + status: 404, + retryAfter: null, + body: { code: 'node_not_found', message: "No node '$&-$1' in this execution" }, + }); + expect(await answer('attempt_mismatch', undefined, { attempt: 1 })).toEqual({ + status: 409, + retryAfter: null, + body: { + code: 'decision_attempt_mismatch', + message: 'The decision names a wait that is not the current one', + attempt: 1, + }, + }); + }); + + it('a delivery timeout asks for a retry in five seconds and says the decision may have landed', async () => { + const answered = await answer('delivery_timeout'); + + expect(answered.status).toBe(503); + expect(answered.retryAfter).toBe('5'); + expect(answered.body.code).toBe('decision_delivery_timeout'); + expect(answered.body.message).toContain('may or may not have landed'); + expect(answered.body.message).toContain('decision_already_made'); + }); +}); diff --git a/apps/backend/src/routes/decision-refusals.ts b/apps/backend/src/routes/decision-refusals.ts new file mode 100644 index 000000000..96effcd7b --- /dev/null +++ b/apps/backend/src/routes/decision-refusals.ts @@ -0,0 +1,87 @@ +import type { Context } from 'hono'; + +import type { ResolveNodeRejection } from '@workflow-builder/execution-core/workflow'; + +import { fill } from '../domain/decision/decision-issues'; +import type { FindDecisionRequestError } from '../domain/decision/find-decision-request'; + +// Every code the decision endpoint refuses with, and its status. The one place a code is spelled out. +export const DECISION_REFUSAL_STATUS = { + validation_error: 400, + invalid_decision: 400, + execution_not_found: 404, + node_not_found: 404, + execution_not_waiting: 409, + node_not_waiting: 409, + decision_already_made: 409, + decision_attempt_mismatch: 409, + effect_not_supported: 501, + decision_delivery_timeout: 503, +} as const; + +export type DecisionRefusalCode = keyof typeof DECISION_REFUSAL_STATUS; + +// One entry per situation; several situations may answer with the same code. +// `{value}` is the one interpolation slot. +export const DECISION_REFUSALS = { + body_invalid: { code: 'validation_error', message: 'Request body failed validation' }, + decision_invalid: { code: 'invalid_decision', message: 'Decision failed validation' }, + execution_not_found: { code: 'execution_not_found', message: 'Execution not found' }, + execution_not_waiting: { code: 'execution_not_waiting', message: 'Execution is not waiting for a decision' }, + run_gone: { code: 'execution_not_waiting', message: 'Execution is no longer running' }, + node_not_found: { code: 'node_not_found', message: "No node '{value}' in this execution" }, + node_without_request: { code: 'node_not_waiting', message: "Node '{value}' carries no decision request" }, + node_never_parked: { code: 'node_not_waiting', message: "Node '{value}' has not asked for a decision" }, + node_not_waiting: { code: 'node_not_waiting', message: "Node '{value}' is not waiting for a decision" }, + decision_already_made: { + code: 'decision_already_made', + message: "Node '{value}' already has a decision for this wait", + }, + attempt_mismatch: { + code: 'decision_attempt_mismatch', + message: 'The decision names a wait that is not the current one', + }, + effect_not_supported: { + code: 'effect_not_supported', + message: "Action '{value}' re-runs the proposal source, which is not supported yet", + }, + // The deadline says nothing about the decision's fate, and nothing here tells two + // senders apart, so the resend answer names the wait and not the sender. + delivery_timeout: { + code: 'decision_delivery_timeout', + message: + 'The decision was not confirmed within the deadline and may or may not have landed. Send it again: a decision_already_made answer means some decision won this wait, not necessarily yours.', + headers: { 'Retry-After': '5' }, + }, +} as const satisfies Record }>; + +export type DecisionRefusal = keyof typeof DECISION_REFUSALS; + +export const LOOKUP_REFUSALS = { + node_not_found: 'node_not_found', + node_without_decision_request: 'node_without_request', +} as const satisfies Record; + +// The route built the envelope, so a fault can only be its own bug: it surfaces as 500. +export const ENGINE_REFUSALS = { + node_not_waiting: 'node_not_waiting', + verdict_already_delivered: 'decision_already_made', + run_not_found: 'run_gone', + verdict_for_unknown_node: 'fault', + verdict_malformed: 'fault', + delivery_timeout: 'delivery_timeout', +} as const satisfies Record; + +export function refuse( + c: Context, + refusal: DecisionRefusal, + { value, extra }: { value?: string; extra?: Record } = {}, +) { + const entry = DECISION_REFUSALS[refusal]; + const headers = 'headers' in entry ? entry.headers : undefined; + return c.json( + { code: entry.code, message: fill(entry.message, value), ...extra }, + DECISION_REFUSAL_STATUS[entry.code], + headers, + ); +} diff --git a/apps/backend/src/routes/decision.test.ts b/apps/backend/src/routes/decision.test.ts new file mode 100644 index 000000000..b855e3dcb --- /dev/null +++ b/apps/backend/src/routes/decision.test.ts @@ -0,0 +1,578 @@ +import { Hono } from 'hono'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import type { ResolveNodeRejection } from '@workflow-builder/execution-core/workflow'; +import { TERMINAL_EXECUTION_STATUSES } from '@workflow-builder/types/workflow-execution/execution-events'; + +import { + AllowAllAuthPort, + AuthDeniedError, + type AuthPort, + type AuthVariables, + createAuthMiddleware, + makeAssertAuthorized, +} from '../auth'; +import { SUBMITTED_DECISION_ERRORS, type SubmittedDecisionErrorCode } from '../domain/decision/decision-issues'; +import { createDecisionRoutes } from './decision'; + +// ---- module mocks ----------------------------------------------------------- + +const { databaseMock, getEngineMock, engineMock } = vi.hoisted(() => ({ + databaseMock: { select: vi.fn() }, + engineMock: { submit: vi.fn(), cancel: vi.fn(), resolveNode: vi.fn() }, + getEngineMock: vi.fn(), +})); + +vi.mock('../db/client', () => ({ database: databaseMock })); +vi.mock('../engine', () => ({ getWorkflowEngine: getEngineMock })); + +// ---- helpers ---------------------------------------------------------------- + +function chainProxyHandler(value: T): ProxyHandler> { + return { + get(target, property, receiver) { + if (property === 'then' || property === 'catch' || property === 'finally') { + const v = Reflect.get(target, property, receiver); + return typeof v === 'function' ? v.bind(target) : v; + } + return () => chainResolving(value); + }, + }; +} + +function chainResolving(value: T): Promise { + return new Proxy(Promise.resolve(value), chainProxyHandler(value)); +} + +function buildApp(port: AuthPort) { + const app = new Hono<{ Variables: AuthVariables }>(); + app.use('*', createAuthMiddleware(port)); + app.onError((error, c) => { + if (error instanceof AuthDeniedError) { + if (!error.caller) { + return c.json({ code: 'unauthenticated', message: 'Authentication required' }, 401); + } + return c.json({ code: 'forbidden', message: error.message }, 403); + } + return c.json({ code: 'internal_error', message: 'Internal server error' }, 500); + }); + app.route('/api/executions', createDecisionRoutes(makeAssertAuthorized(port))); + return app; +} + +function allowAll(spy = vi.fn(async () => true)): AuthPort { + return { identify: vi.fn(async () => null), authorize: spy }; +} + +function denyAll(): AuthPort { + return { identify: vi.fn(async () => null), authorize: vi.fn(async () => false) }; +} + +async function decide(app: ReturnType, body: unknown): Promise { + return app.request('/api/executions/e-1/decision', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + }); +} + +type Body = { + code?: string; + message?: string; + details?: { code: string; path: (string | number)[] }[]; + attempt?: number; + effect?: string; +}; + +function bodyOf(response: Response): Promise { + return response.json() as Promise; +} + +async function codeOf(response: Response): Promise { + const body = await bodyOf(response); + return body.code; +} + +// 1st select: the execution row (none when `execution` is undefined). 2nd: the wait count. +function program(execution?: unknown, waits = 1) { + databaseMock.select.mockReturnValueOnce(chainResolving(execution === undefined ? [] : [execution])); + databaseMock.select.mockReturnValue(chainResolving([{ waits }])); +} + +// ---- fixtures --------------------------------------------------------------- + +const refundForm = { + type: 'object', + properties: { + orderDate: { type: 'string', readOnly: true }, + refundAmount: { type: 'number' }, + note: { type: 'string' }, + }, + required: ['refundAmount'], +}; +const approve = { name: 'approve', label: 'Approve', effect: 'resume' }; +const reject = { name: 'reject', label: 'Reject', effect: 'reject' }; +const askAgain = { name: 'ask-again', label: 'Ask again', effect: 'rerun-source' }; + +// source-1 feeds two deciding nodes. review-1 offers all three effects and takes a reject +// without a reason; review-2 requires one. after-1 hangs off review-1's approved port. +const snapshot = { + nodes: [ + { id: 'source-1', data: { type: 'product/any', properties: {} } }, + { + id: 'review-1', + data: { + type: 'product/any', + properties: { + decisionRequest: { + version: 1, + actions: [approve, reject, askAgain], + schema: refundForm, + proposalSourceNodeId: 'source-1', + }, + }, + }, + }, + { + id: 'review-2', + data: { + type: 'product/any', + properties: { + decisionRequest: { version: 1, actions: [approve, { ...reject, reasonRequired: true }], schema: refundForm }, + }, + }, + }, + { id: 'after-1', data: { type: 'product/any', properties: {} } }, + ], + edges: [ + { id: 'e1', source: 'source-1', target: 'review-1' }, + { id: 'e2', source: 'source-1', target: 'review-2' }, + { id: 'e3', source: 'review-1', target: 'after-1', sourceHandle: 'approved' }, + ], +}; + +const waitingExecution = { + id: 'e-1', + workflowId: 'w-1', + sourceVersion: 'published', + workflowSnapshotJson: snapshot, + status: 'waiting', + tenantId: 'acme', + triggerPayloadJson: null, + startedAt: new Date(0), + finishedAt: null, + errorMessage: null, + createdAt: new Date(0), + updatedAt: new Date(0), +}; + +const approveBody = { nodeId: 'review-1', attempt: 1, action: 'approve', edits: { refundAmount: 120 } }; +const approvedDecision = { action: 'approve', effect: 'resume-with-edits', edits: { refundAmount: 120 } }; + +beforeEach(() => { + vi.clearAllMocks(); + getEngineMock.mockReturnValue(engineMock); + engineMock.resolveNode.mockResolvedValue({}); +}); + +// ---- authorization ------------------------------------------------------------ + +describe('POST /api/executions/:id/decision - authorization', () => { + it('asserts executions:decide with the row attributes the port scopes by', async () => { + const authorizeSpy = vi.fn(async () => true); + program(waitingExecution); + + await decide(buildApp(allowAll(authorizeSpy)), approveBody); + + expect(authorizeSpy).toHaveBeenCalledWith(null, 'executions:decide', { + kind: 'execution', + executionId: 'e-1', + attributes: { workflowId: 'w-1', tenantId: 'acme', status: 'waiting' }, + }); + }); + + it('asserts without attributes when there is no row, and the deny still wins over the 404', async () => { + const authorizeSpy = vi.fn(async () => false); + program(); + + const response = await decide(buildApp(allowAll(authorizeSpy)), approveBody); + + expect(response.status).toBe(401); + expect(authorizeSpy.mock.calls[0]?.[2]).toEqual({ kind: 'execution', executionId: 'e-1' }); + expect(authorizeSpy.mock.calls[0]?.[2]).not.toHaveProperty('attributes'); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }); + + it('a deny answers 401 after the one row read, and the engine is never asked', async () => { + program(waitingExecution); + + const response = await decide(buildApp(denyAll()), approveBody); + + expect(response.status).toBe(401); + expect(databaseMock.select).toHaveBeenCalledTimes(1); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }); + + describe('under the reference AllowAllAuthPort', () => { + const originalAuthPort = process.env['WB_AUTH_PORT']; + + beforeEach(() => { + process.env['WB_AUTH_PORT'] = 'allow-all'; + vi.spyOn(console, 'warn').mockImplementation(() => {}); + }); + + afterEach(() => { + vi.restoreAllMocks(); + if (originalAuthPort === undefined) delete process.env['WB_AUTH_PORT']; + else process.env['WB_AUTH_PORT'] = originalAuthPort; + }); + + it('an anonymous caller decides', async () => { + program(waitingExecution); + + const response = await decide(buildApp(new AllowAllAuthPort()), approveBody); + + expect(response.status).toBe(200); + expect(engineMock.resolveNode).toHaveBeenCalledTimes(1); + }); + }); +}); + +// ---- the run ---------------------------------------------------------------------- + +describe('POST /api/executions/:id/decision - the run', () => { + it('404 execution_not_found when the row is missing', async () => { + program(); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(404); + expect(await response.json()).toEqual({ code: 'execution_not_found', message: 'Execution not found' }); + }); + + it('the row is checked before the body: a missing row with a bad body is a 404', async () => { + program(); + + const response = await decide(buildApp(allowAll()), { not: 'a decision' }); + + expect(response.status).toBe(404); + expect(await codeOf(response)).toBe('execution_not_found'); + }); + + it.each([...TERMINAL_EXECUTION_STATUSES, 'cancelling'])( + '409 execution_not_waiting on a %s run, before the body is even read', + async (status) => { + program({ ...waitingExecution, status }); + + const response = await decide(buildApp(allowAll()), { not: 'a decision' }); + + expect(response.status).toBe(409); + expect(await codeOf(response)).toBe('execution_not_waiting'); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }, + ); + + it.each(['pending', 'running', 'waiting'])('a %s run goes through to the engine, the arbiter', async (status) => { + program({ ...waitingExecution, status }); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(200); + expect(engineMock.resolveNode).toHaveBeenCalledTimes(1); + }); +}); + +// ---- the body --------------------------------------------------------------------- + +describe('POST /api/executions/:id/decision - the body', () => { + it.each<{ name: string; body: unknown; path: string }>([ + { name: 'a missing nodeId', body: { attempt: 1, action: 'approve' }, path: 'nodeId' }, + { name: 'an empty nodeId', body: { nodeId: '', attempt: 1, action: 'approve' }, path: 'nodeId' }, + { name: 'an attempt of 0', body: { nodeId: 'review-1', attempt: 0, action: 'approve' }, path: 'attempt' }, + { name: 'a non-integer attempt', body: { nodeId: 'review-1', attempt: 1.5, action: 'approve' }, path: 'attempt' }, + { name: 'a missing action', body: { nodeId: 'review-1', attempt: 1 }, path: 'action' }, + ])('400 validation_error for $name', async ({ body, path }) => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), body); + + expect(response.status).toBe(400); + const json = await bodyOf(response); + expect(json.code).toBe('validation_error'); + expect(json.details?.map((detail) => detail.path.join('.'))).toEqual([path]); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }); +}); + +// ---- the node ----------------------------------------------------------------------- + +describe('POST /api/executions/:id/decision - the node', () => { + it('404 node_not_found for a node the snapshot does not have', async () => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), { ...approveBody, nodeId: 'ghost' }); + + expect(response.status).toBe(404); + expect(await response.json()).toEqual({ code: 'node_not_found', message: "No node 'ghost' in this execution" }); + }); + + it('409 node_not_waiting for a node that carries no decision request', async () => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), { ...approveBody, nodeId: 'source-1' }); + + expect(response.status).toBe(409); + expect(await response.json()).toEqual({ + code: 'node_not_waiting', + message: "Node 'source-1' carries no decision request", + }); + }); + + it('409 node_not_waiting for a node that has never parked', async () => { + program(waitingExecution, 0); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(409); + expect(await response.json()).toEqual({ + code: 'node_not_waiting', + message: "Node 'review-1' has not asked for a decision", + }); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }); + + it('the node is looked up before the decision is judged: an unknown node with a bad decision is a 404', async () => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), { nodeId: 'ghost', attempt: 1, action: 'escalate' }); + + expect(response.status).toBe(404); + }); +}); + +// ---- the decision ------------------------------------------------------------------- + +const INVALID_DECISIONS = { + unknown_action: { nodeId: 'review-1', action: 'escalate' }, + reason_required: { nodeId: 'review-2', action: 'reject' }, + comment_required: { nodeId: 'review-1', action: 'ask-again' }, + edits_not_allowed: { nodeId: 'review-1', action: 'reject', edits: { refundAmount: 1 } }, + unknown_field: { nodeId: 'review-1', action: 'approve', edits: { discount: 10 } }, + field_not_editable: { nodeId: 'review-1', action: 'approve', edits: { orderDate: '2026-01-01' } }, + required_field_missing: { nodeId: 'review-1', action: 'approve', edits: { refundAmount: null } }, +} satisfies Record>; + +describe('POST /api/executions/:id/decision - the decision', () => { + it.each(Object.keys(SUBMITTED_DECISION_ERRORS) as SubmittedDecisionErrorCode[])( + '400 invalid_decision carrying %s', + async (code) => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), { attempt: 1, ...INVALID_DECISIONS[code] }); + + expect(response.status).toBe(400); + const json = await bodyOf(response); + expect(json.code).toBe('invalid_decision'); + expect(json.details).toHaveLength(1); + expect(json.details?.[0]).toMatchObject({ code, message: expect.any(String), path: expect.any(Array) }); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }, + ); + + it('the decision is judged before the attempt: a bad decision with a stale attempt is a 400', async () => { + program(waitingExecution, 1); + + const response = await decide(buildApp(allowAll()), { nodeId: 'review-1', attempt: 2, action: 'escalate' }); + + expect(response.status).toBe(400); + expect(await codeOf(response)).toBe('invalid_decision'); + }); +}); + +// ---- the wait instance ------------------------------------------------------------ + +describe('POST /api/executions/:id/decision - the wait instance', () => { + it('409 decision_attempt_mismatch carrying the current attempt', async () => { + program(waitingExecution, 1); + + const response = await decide(buildApp(allowAll()), { ...approveBody, attempt: 2 }); + + expect(response.status).toBe(409); + expect(await response.json()).toEqual({ + code: 'decision_attempt_mismatch', + message: 'The decision names a wait that is not the current one', + attempt: 1, + }); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }); + + it('the attempt is checked before the effect: a rerun-source naming a stale attempt is a 409', async () => { + program(waitingExecution, 1); + + const response = await decide(buildApp(allowAll()), { + nodeId: 'review-1', + attempt: 2, + action: 'ask-again', + comment: 'too generous', + }); + + expect(response.status).toBe(409); + expect(await codeOf(response)).toBe('decision_attempt_mismatch'); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }); + + it('a decision for the second of two waiting nodes is delivered while the first keeps waiting', async () => { + program(waitingExecution, 1); + + const response = await decide(buildApp(allowAll()), { nodeId: 'review-2', attempt: 1, action: 'approve' }); + + expect(response.status).toBe(200); + expect(engineMock.resolveNode).toHaveBeenCalledTimes(1); + expect(engineMock.resolveNode).toHaveBeenCalledWith('e-1', 'review-2', { + output: { action: 'approve', effect: 'resume', edits: {} }, + nextPort: 'approved', + }); + }); +}); + +// ---- the effect --------------------------------------------------------------------- + +describe('POST /api/executions/:id/decision - the effect', () => { + it('501 effect_not_supported for rerun-source, engine never asked', async () => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), { + nodeId: 'review-1', + attempt: 1, + action: 'ask-again', + comment: 'too generous', + }); + + expect(response.status).toBe(501); + expect(await response.json()).toEqual({ + code: 'effect_not_supported', + message: "Action 'ask-again' re-runs the proposal source, which is not supported yet", + }); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + }); +}); + +// ---- delivery ----------------------------------------------------------------------- + +describe('POST /api/executions/:id/decision - delivery', () => { + it('hands the engine the decision as the output and the action port as the route, and answers 200', async () => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ + executionId: 'e-1', + nodeId: 'review-1', + attempt: 1, + action: 'approve', + effect: 'resume-with-edits', + }); + expect(engineMock.resolveNode).toHaveBeenCalledWith('e-1', 'review-1', { + output: approvedDecision, + nextPort: 'approved', + }); + }); + + // Postgres accepts a non-canonical uuid and answers with the canonical row, so the url + // spelling and the row id can differ; the engine's workflow name is case-sensitive. + it('delivers to the id the row carries, not the spelling the url used', async () => { + const canonical = 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11'; + program({ ...waitingExecution, id: canonical }); + + const response = await buildApp(allowAll()).request(`/api/executions/${canonical.toUpperCase()}/decision`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(approveBody), + }); + + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ executionId: canonical }); + expect(engineMock.resolveNode).toHaveBeenCalledWith(canonical, 'review-1', { + output: approvedDecision, + nextPort: 'approved', + }); + }); + + it('a reject with no reason, none required, resumes on the reject port', async () => { + program(waitingExecution); + + const response = await decide(buildApp(allowAll()), { nodeId: 'review-1', attempt: 1, action: 'reject' }); + + expect(response.status).toBe(200); + const body = await bodyOf(response); + expect(body.effect).toBe('reject'); + expect(engineMock.resolveNode).toHaveBeenCalledWith('e-1', 'review-1', { + output: { action: 'reject', effect: 'reject', edits: {} }, + nextPort: 'rejected', + }); + }); + + it.each<{ code: ResolveNodeRejection; status: number; answer: string }>([ + { code: 'node_not_waiting', status: 409, answer: 'node_not_waiting' }, + { code: 'verdict_already_delivered', status: 409, answer: 'decision_already_made' }, + { code: 'run_not_found', status: 409, answer: 'execution_not_waiting' }, + ])('the engine answer $code becomes $status $answer on the first try, no retry', async ({ code, status, answer }) => { + program(waitingExecution); + engineMock.resolveNode.mockResolvedValue({ error: { code, message: 'engine said no' } }); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(status); + expect(await codeOf(response)).toBe(answer); + expect(engineMock.resolveNode).toHaveBeenCalledTimes(1); + }); + + it('the engine answer delivery_timeout becomes 503 with Retry-After, sent once, no retry', async () => { + program(waitingExecution); + engineMock.resolveNode.mockResolvedValue({ error: { code: 'delivery_timeout', message: 'Deadline exceeded' } }); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(503); + expect(response.headers.get('retry-after')).toBe('5'); + const body = await bodyOf(response); + expect(body.code).toBe('decision_delivery_timeout'); + expect(body.message).toContain('decision_already_made'); + expect(engineMock.resolveNode).toHaveBeenCalledTimes(1); + }); + + it.each(['verdict_malformed', 'verdict_for_unknown_node'])( + 'the engine answer %s is a backend fault: 500', + async (code) => { + program(waitingExecution); + engineMock.resolveNode.mockResolvedValue({ error: { code, message: 'engine said no' } }); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(500); + expect(await codeOf(response)).toBe('internal_error'); + }, + ); + + it('an error the engine throws instead of returning is a backend fault: 500', async () => { + program(waitingExecution); + engineMock.resolveNode.mockRejectedValue(new Error('connection lost')); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(500); + expect(await codeOf(response)).toBe('internal_error'); + }); + + it('a stored snapshot that no longer parses is a backend fault: 500', async () => { + program({ ...waitingExecution, workflowSnapshotJson: { nodes: 'broken' } }); + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); + + const response = await decide(buildApp(allowAll()), approveBody); + + expect(response.status).toBe(500); + expect(engineMock.resolveNode).not.toHaveBeenCalled(); + errorSpy.mockRestore(); + }); +}); diff --git a/apps/backend/src/routes/decision.ts b/apps/backend/src/routes/decision.ts new file mode 100644 index 000000000..6352579b5 --- /dev/null +++ b/apps/backend/src/routes/decision.ts @@ -0,0 +1,119 @@ +import { eq } from 'drizzle-orm'; +import { Hono } from 'hono'; +import { z } from 'zod'; + +import { + type ExecutionStatus, + TERMINAL_EXECUTION_STATUSES, +} from '@workflow-builder/types/workflow-execution/execution-events'; + +import type { AssertAuthorized, AuthResource } from '../auth'; +import { database } from '../db/client'; +import { executions } from '../db/schema'; +import { findDecisionRequest } from '../domain/decision/find-decision-request'; +import { hasNodeResolution, toNodeResolution } from '../domain/decision/node-resolution'; +import { submittedDecisionSchema, validateSubmittedDecision } from '../domain/decision/validate-submitted-decision'; +import { workflowSnapshotSchema } from '../domain/mapper/snapshot-schema'; +import { getWorkflowEngine } from '../engine'; +import { countNodeWaits } from '../events/count-node-waits'; +import { logger as backendLogger } from '../logger'; +import type { BackendEnv } from './backend-env'; +import { ENGINE_REFUSALS, LOOKUP_REFUSALS, refuse } from './decision-refusals'; +import { formatValidationDetails } from './snapshot-validation'; + +const logger = backendLogger.child({ component: 'decision-route' }); + +// Every other status goes to the engine: the advisory status write is best-effort, so a +// parked run can still read 'pending'. +const NOT_DECIDABLE_STATUSES = new Set([ + ...TERMINAL_EXECUTION_STATUSES, + 'cancelling' satisfies ExecutionStatus, +]); + +const decisionBodySchema = submittedDecisionSchema.extend({ + nodeId: z.string().min(1), + attempt: z.int().min(1), +}); + +export function createDecisionRoutes(assertAuthorized: AssertAuthorized): Hono { + const routes = new Hono(); + + routes.post('/:id/decision', async (c) => { + const executionId = c.req.param('id'); + + // Read before authorization so the port can scope by the row; a deny thus wins over 404. + const [execution] = await database.select().from(executions).where(eq(executions.id, executionId)); + const resource: AuthResource = execution + ? { + kind: 'execution', + executionId: execution.id, + attributes: { workflowId: execution.workflowId, tenantId: execution.tenantId, status: execution.status }, + } + : { kind: 'execution', executionId }; + await assertAuthorized(c, 'executions:decide', resource); + + if (!execution) return refuse(c, 'execution_not_found'); + if (NOT_DECIDABLE_STATUSES.has(execution.status)) return refuse(c, 'execution_not_waiting'); + + // Postgres answers a non-canonical uuid with the canonical row, so the two can differ. + // Every id below is the row's: the engine builds a case-sensitive workflow name from it. + const { id: resolvedId } = execution; + + const parsedBody = z.safeParse(decisionBodySchema, await c.req.json()); + if (!parsedBody.success) { + return refuse(c, 'body_invalid', { extra: { details: formatValidationDetails(parsedBody.error) } }); + } + const { nodeId, attempt, ...submitted } = parsedBody.data; + + const parsedSnapshot = z.safeParse(workflowSnapshotSchema, execution.workflowSnapshotJson); + if (!parsedSnapshot.success) { + logger.error('stored snapshot no longer parses', { + executionId: resolvedId, + error: { issues: formatValidationDetails(parsedSnapshot.error) }, + }); + throw new Error(`stored snapshot of execution ${resolvedId} no longer parses`); + } + const found = findDecisionRequest(parsedSnapshot.data, nodeId); + if (found.error !== undefined) return refuse(c, LOOKUP_REFUSALS[found.error], { value: nodeId }); + + const validated = validateSubmittedDecision(found.request, submitted); + if (validated.error !== undefined) { + return refuse(c, 'decision_invalid', { extra: { details: [validated.error] } }); + } + const { decision, action } = validated; + + // Not atomic with delivery, and safe only because a node parks at most once per run. + // A rerun re-parks it, and then the wait instance must reach the engine, which keys + // its waits by node id alone (follow-up: decision-attempt-in-engine). + const waits = await countNodeWaits(resolvedId, nodeId); + if (waits === 0) return refuse(c, 'node_never_parked', { value: nodeId }); + if (waits !== attempt) return refuse(c, 'attempt_mismatch', { extra: { attempt: waits } }); + + // Refused until the engine can re-run a source (follow-up: decision-rerun-source). + if (!hasNodeResolution(decision) || action.effect === 'rerun-source') { + return refuse(c, 'effect_not_supported', { value: action.name }); + } + + const result = await getWorkflowEngine().resolveNode(resolvedId, nodeId, toNodeResolution(decision, action)); + if (result.error !== undefined) { + const outcome = ENGINE_REFUSALS[result.error.code]; + if (outcome === 'fault') { + throw new Error( + `engine refused the completion for node '${nodeId}': ${result.error.code}: ${result.error.message}`, + ); + } + return refuse(c, outcome, { value: nodeId }); + } + + logger.info('decision delivered', { + executionId: resolvedId, + nodeId, + attempt, + action: action.name, + effect: decision.effect, + }); + return c.json({ executionId: resolvedId, nodeId, attempt, action: action.name, effect: decision.effect }); + }); + + return routes; +} diff --git a/apps/backend/src/routes/executions.test.ts b/apps/backend/src/routes/executions.test.ts index fbdffcaf8..97ce29116 100644 --- a/apps/backend/src/routes/executions.test.ts +++ b/apps/backend/src/routes/executions.test.ts @@ -10,7 +10,8 @@ import { createAuthMiddleware, makeAssertAuthorized, } from '../auth'; -import { type TenantContext, type TenantVariables, createTenantMiddleware } from '../tenant'; +import { type TenantContext, createTenantMiddleware } from '../tenant'; +import type { BackendEnv } from './backend-env'; import { createExecutionsRoutes } from './executions'; // ---- module mocks ----------------------------------------------------------- @@ -93,7 +94,7 @@ function denyAll(): AuthPort { // cross-check has a `c.var.tenant` to read. `tenant` is what the configured // TenantContextPort resolves to (null = single-tenant reference default). function buildAppWithTenant(port: AuthPort, tenant: TenantContext | null) { - const app = new Hono<{ Variables: AuthVariables & TenantVariables }>(); + const app = new Hono(); app.use('*', createAuthMiddleware(port)); app.use('*', createTenantMiddleware({ resolve: vi.fn(async () => tenant) })); app.route('/api/executions', createExecutionsRoutes(makeAssertAuthorized(port))); diff --git a/apps/backend/src/routes/executions.ts b/apps/backend/src/routes/executions.ts index d502c14a5..3a04b8d46 100644 --- a/apps/backend/src/routes/executions.ts +++ b/apps/backend/src/routes/executions.ts @@ -8,7 +8,7 @@ import { type TerminalExecutionEventType, } from '@workflow-builder/types/workflow-execution/execution-events'; -import type { AssertAuthorized, AuthVariables } from '../auth'; +import type { AssertAuthorized } from '../auth'; import { database } from '../db/client'; import { executions } from '../db/schema'; import { getWorkflowEngine } from '../engine'; @@ -17,16 +17,14 @@ import { subscribe } from '../events/execution-event-bus'; import { type ExecutionEventRow, fetchEventsAfter } from '../events/fetch-events-after'; import { createSerializedDrainer } from '../events/serialized-drainer'; import { logger as backendLogger } from '../logger'; -import type { TenantVariables } from '../tenant'; +import type { BackendEnv } from './backend-env'; const logger = backendLogger.child({ component: 'executions-route' }); const TERMINAL_STATUSES = new Set(TERMINAL_EXECUTION_STATUSES); -export function createExecutionsRoutes( - assertAuthorized: AssertAuthorized, -): Hono<{ Variables: AuthVariables & TenantVariables }> { - const routes = new Hono<{ Variables: AuthVariables & TenantVariables }>(); +export function createExecutionsRoutes(assertAuthorized: AssertAuthorized): Hono { + const routes = new Hono(); routes.get('/:id', async (c) => { const executionId = c.req.param('id'); diff --git a/apps/backend/src/routes/visualize.ts b/apps/backend/src/routes/visualize.ts index 8ebc5066c..69aa0eebb 100644 --- a/apps/backend/src/routes/visualize.ts +++ b/apps/backend/src/routes/visualize.ts @@ -3,11 +3,11 @@ import { generateText } from 'ai'; import { Hono } from 'hono'; import { z } from 'zod'; -import type { AssertAuthorized, AuthVariables } from '../auth'; +import type { AssertAuthorized } from '../auth'; import { env } from '../env'; import { logger as backendLogger } from '../logger'; import { guardExecution } from '../security/execution-guard'; -import type { TenantVariables } from '../tenant'; +import type { BackendEnv } from './backend-env'; const logger = backendLogger.child({ component: 'visualize-route' }); @@ -37,10 +37,8 @@ Rules: text: `Return the content as clean, readable plain text. Output ONLY the text.`, }; -export function createVisualizeRoutes( - assertAuthorized: AssertAuthorized, -): Hono<{ Variables: AuthVariables & TenantVariables }> { - const routes = new Hono<{ Variables: AuthVariables & TenantVariables }>(); +export function createVisualizeRoutes(assertAuthorized: AssertAuthorized): Hono { + const routes = new Hono(); routes.post('/adapt', async (c) => { await assertAuthorized(c, 'workflows:execute', { kind: 'workflows' }); diff --git a/apps/backend/src/routes/workflows.test.ts b/apps/backend/src/routes/workflows.test.ts index 1b7048953..f9f83d47f 100644 --- a/apps/backend/src/routes/workflows.test.ts +++ b/apps/backend/src/routes/workflows.test.ts @@ -10,7 +10,8 @@ import { createAuthMiddleware, makeAssertAuthorized, } from '../auth'; -import { type TenantContext, type TenantVariables, createTenantMiddleware } from '../tenant'; +import { type TenantContext, createTenantMiddleware } from '../tenant'; +import type { BackendEnv } from './backend-env'; import { createWorkflowsRoutes } from './workflows'; // ---- module mocks ----------------------------------------------------------- @@ -99,7 +100,7 @@ function denyAll(): AuthPort { // `c.var.tenant`. `tenant` is what the configured TenantContextPort resolves // to (null = single-tenant reference default). function buildAppWithTenant(port: AuthPort, tenant: TenantContext | null) { - const app = new Hono<{ Variables: AuthVariables & TenantVariables }>(); + const app = new Hono(); app.use('*', createAuthMiddleware(port)); app.use('*', createTenantMiddleware({ resolve: vi.fn(async () => tenant) })); app.route('/api/workflows', createWorkflowsRoutes(makeAssertAuthorized(port))); diff --git a/apps/backend/src/routes/workflows.ts b/apps/backend/src/routes/workflows.ts index 283424c80..b995cbc02 100644 --- a/apps/backend/src/routes/workflows.ts +++ b/apps/backend/src/routes/workflows.ts @@ -2,14 +2,14 @@ import { eq } from 'drizzle-orm'; import { Hono } from 'hono'; import { z } from 'zod'; -import type { AssertAuthorized, AuthVariables } from '../auth'; +import type { AssertAuthorized } from '../auth'; import { database } from '../db/client'; import { executions, workflows } from '../db/schema'; import { mapToExecutionModel } from '../domain/mapper/from-integration-data'; import { getWorkflowEngine } from '../engine'; import { logger as backendLogger } from '../logger'; import { guardExecution } from '../security/execution-guard'; -import type { TenantVariables } from '../tenant'; +import type { BackendEnv } from './backend-env'; import { formatValidationDetails, parseSnapshot } from './snapshot-validation'; const logger = backendLogger.child({ component: 'workflows-route' }); @@ -28,10 +28,8 @@ const executeSchema = z.object({ triggerPayload: z.record(z.string(), z.unknown()).optional(), }); -export function createWorkflowsRoutes( - assertAuthorized: AssertAuthorized, -): Hono<{ Variables: AuthVariables & TenantVariables }> { - const routes = new Hono<{ Variables: AuthVariables & TenantVariables }>(); +export function createWorkflowsRoutes(assertAuthorized: AssertAuthorized): Hono { + const routes = new Hono(); routes.post('/', async (c) => { await assertAuthorized(c, 'workflows:create', { kind: 'workflows' }); diff --git a/apps/backend/src/server.ts b/apps/backend/src/server.ts index 9a2d7773e..73a21632b 100644 --- a/apps/backend/src/server.ts +++ b/apps/backend/src/server.ts @@ -4,22 +4,17 @@ import { Hono } from 'hono'; import { bodyLimit } from 'hono/body-limit'; import { cors } from 'hono/cors'; -import { - AllowAllAuthPort, - AuthDeniedError, - type AuthPort, - type AuthVariables, - createAuthMiddleware, - makeAssertAuthorized, -} from './auth'; +import { AllowAllAuthPort, AuthDeniedError, type AuthPort, createAuthMiddleware, makeAssertAuthorized } from './auth'; import { runMigrations } from './db/migrate'; import { env } from './env'; import { logger } from './logger'; import { createRateLimitMiddleware } from './middleware/rate-limit'; +import type { BackendEnv } from './routes/backend-env'; +import { createDecisionRoutes } from './routes/decision'; import { createExecutionsRoutes } from './routes/executions'; import { createVisualizeRoutes } from './routes/visualize'; import { createWorkflowsRoutes } from './routes/workflows'; -import { NoopTenantContextPort, type TenantContextPort, type TenantVariables, createTenantMiddleware } from './tenant'; +import { NoopTenantContextPort, type TenantContextPort, createTenantMiddleware } from './tenant'; // Permissive default for local development. The constructor itself emits a // loud startup warning and refuses to boot unless `WB_AUTH_PORT=allow-all` is @@ -34,7 +29,7 @@ const assertAuthorized = makeAssertAuthorized(authPort); // claim, header, …) — see `apps/backend/tenant-context-port.decision-log.md`. const tenantPort: TenantContextPort = new NoopTenantContextPort(); -const app = new Hono<{ Variables: AuthVariables & TenantVariables }>(); +const app = new Hono(); app.use('/*', cors()); // Reject request bodies larger than 1 MB to prevent memory exhaustion @@ -81,6 +76,7 @@ if (env.RATE_LIMIT_EXECUTE_PER_MINUTE > 0 || env.RATE_LIMIT_EXECUTE_PER_DAY > 0) app.route('/api/workflows', createWorkflowsRoutes(assertAuthorized)); app.route('/api/executions', createExecutionsRoutes(assertAuthorized)); +app.route('/api/executions', createDecisionRoutes(assertAuthorized)); app.route('/api/visualize', createVisualizeRoutes(assertAuthorized)); // a failure (DB still starting) exits the process; the container restart policy retries diff --git a/packages/execution-core/src/index.ts b/packages/execution-core/src/index.ts index 517e47641..6ce1dfbaf 100644 --- a/packages/execution-core/src/index.ts +++ b/packages/execution-core/src/index.ts @@ -13,7 +13,14 @@ export type { NodeErrorClassification, NodeErrorEnvelope } from './errors'; export type { ExecutionContext } from './execution-context'; -export type { WorkflowEnginePort, WorkflowExecutionInput } from './ports/workflow-engine.port'; +export { RESOLVE_NODE_REJECTIONS } from './ports/workflow-engine.port'; +export type { + ResolveNodeRejection, + ResolveNodeResult, + VerdictRejection, + WorkflowEnginePort, + WorkflowExecutionInput, +} from './ports/workflow-engine.port'; export type { ActivityRunnerPort, CompletedNodeExecution, diff --git a/packages/execution-core/src/ports/workflow-engine.port.ts b/packages/execution-core/src/ports/workflow-engine.port.ts index daed6c6f4..5c0481b61 100644 --- a/packages/execution-core/src/ports/workflow-engine.port.ts +++ b/packages/execution-core/src/ports/workflow-engine.port.ts @@ -1,5 +1,7 @@ import type { BaseNode, WorkflowDefinition } from '@workflow-builder/types/workflow-execution/execution-model'; +import type { CompletedNodeExecution } from './activity-runner.port'; + export type WorkflowExecutionInput = { workflowId: string; executionId: string; @@ -9,9 +11,28 @@ export type WorkflowExecutionInput = { global: Record; }; +// The refusals a verdict can meet, declared once. The engine's own validator answers +// with the first group; the delivery itself answers with the second. +const VERDICT_REJECTIONS = [ + 'verdict_malformed', + 'verdict_for_unknown_node', + 'verdict_already_delivered', + 'node_not_waiting', +] as const; +const DELIVERY_REJECTIONS = ['run_not_found', 'delivery_timeout'] as const; +export const RESOLVE_NODE_REJECTIONS = [...VERDICT_REJECTIONS, ...DELIVERY_REJECTIONS] as const; + +export type VerdictRejection = (typeof VERDICT_REJECTIONS)[number]; +export type ResolveNodeRejection = (typeof RESOLVE_NODE_REJECTIONS)[number]; + +export type ResolveNodeResult = { error?: undefined } | { error: { code: ResolveNodeRejection; message: string } }; + // Backend calls this; concrete adapters (Temporal, in-memory, …) implement it. // TNode is opaque to the backend — only the worker narrows it to concrete types. export interface WorkflowEnginePort { submit(input: WorkflowExecutionInput): Promise; cancel(executionId: string): Promise; + // Delivers the completion a parked node waits for. Refusals are results, not throws; + // anything else that goes wrong is thrown. + resolveNode(executionId: string, nodeId: string, resolution: CompletedNodeExecution): Promise; } diff --git a/packages/execution-core/src/workflow.ts b/packages/execution-core/src/workflow.ts index 078c3c092..c9258ace3 100644 --- a/packages/execution-core/src/workflow.ts +++ b/packages/execution-core/src/workflow.ts @@ -13,7 +13,13 @@ export { NodeExecutionError } from './errors'; export type { ExecutionContext } from './execution-context'; -export type { WorkflowEnginePort, WorkflowExecutionInput } from './ports/workflow-engine.port'; +export type { + ResolveNodeRejection, + ResolveNodeResult, + VerdictRejection, + WorkflowEnginePort, + WorkflowExecutionInput, +} from './ports/workflow-engine.port'; export type { ActivityRunnerPort, CompletedNodeExecution, diff --git a/packages/temporal/README.md b/packages/temporal/README.md index 14c3e4ad1..d6ccde4f5 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -149,6 +149,7 @@ const engine = new TemporalWorkflowEngine({ await engine.submit({ workflowId, executionId, definition, triggerPayload: {}, variables: {}, global: {} }); await engine.cancel(executionId); +const { error } = await engine.resolveNode(executionId, 'approval-1', { output: 'approved', nextPort: 'approved' }); ``` The engine and the worker default to the same task queue (`workflow-execution`). Override it in both places together, or leave both alone. @@ -163,11 +164,11 @@ Three things are deliberately yours, and knowing which they are makes debugging ## Entry points -| Import | Use it for | -| ------------------------------------ | ------------------------------------------------------------------------------------------- | -| `@workflowbuilder/temporal` | Worker side: the plugin, `createActivities`, shared constants, types | -| `@workflowbuilder/temporal/client` | Starting and cancelling runs | -| `@workflowbuilder/temporal/workflow` | Sandbox-safe: `runWorkflow` to re-export, event emitter, profiles, the `resolveNode` update | +| Import | Use it for | +| ------------------------------------ | -------------------------------------------------------------------------------------------------------------------- | +| `@workflowbuilder/temporal` | Worker side: the plugin, `createActivities`, the `RUN_WORKFLOW_NAME` and `RESOLVE_NODE_UPDATE_NAME` constants, types | +| `@workflowbuilder/temporal/client` | Starting and cancelling runs, delivering verdicts | +| `@workflowbuilder/temporal/workflow` | Sandbox-safe: `runWorkflow` to re-export, event emitter, profiles, the `resolveNode` update | `/workflow` is the only entry point that is safe inside Temporal's V8 sandbox. The split also means a backend that only starts runs never pulls in the worker package and its native binary. @@ -219,7 +220,7 @@ await handle.executeUpdate(resolveNodeUpdate, { The `resolution` is the completion the node finishes with, exactly as if its executor had returned it: `output` becomes the node's output for everything downstream, and `nextPort` routes the graph. This package passes it through untouched. What a verdict contains, and who may deliver one, is your application's contract. -Because this is an Update and not a signal, the caller gets a synchronous answer, and the update is validated before it is accepted, so a rejected verdict leaves no trace in the run. The rejections, each an `ApplicationFailure` with a stable type: a malformed envelope is `verdict_malformed` (the envelope is an object carrying at most `output` and `nextPort`; `nextPort` must not be the reserved `errorRoute`, and a missing `output` is read as `undefined`, which is what the default JSON payload converter turns `output: undefined` into), a node id that is not in the graph is `verdict_for_unknown_node`, a node that is not currently waiting is `node_not_waiting` (final: it has not parked, or its wait was cancelled), and a second verdict for the same node is `verdict_already_delivered`: the first one wins. A verdict for a run that has already closed fails at the server. Cancelling a parked run closes it as `cancelled`, with `execution_cancelled` following the node's `node_waiting` and no `node_failed` recorded for the node that was waiting. +Because this is an Update and not a signal, the caller gets a synchronous answer, and the update is validated before it is accepted, so a rejected verdict leaves no trace in the run. The rejections, each an `ApplicationFailure` with a stable type: a malformed envelope is `verdict_malformed` (the envelope is an object carrying at most `output` and `nextPort`; `nextPort` must not be the reserved `errorRoute`, and a missing `output` is read as `undefined`, which is what the default JSON payload converter turns `output: undefined` into), a node id that is not in the graph is `verdict_for_unknown_node`, a node that is not currently waiting is `node_not_waiting` (final: it has not parked, or its wait was cancelled), and a second verdict for the same node is `verdict_already_delivered`: the first one wins. A verdict for a run that has already closed fails at the server. Through `engine.resolveNode` every one of these comes back as `{ error: { code, message } }` instead of a throw, plus `run_not_found` for a closed run and `delivery_timeout` when no worker accepted the update within `resolveTimeoutMs` (default 10 s). A timed-out update may still reach the next worker, so a resend can answer `verdict_already_delivered`; exactly one lands. Cancelling a parked run closes it as `cancelled`, with `execution_cancelled` following the node's `node_waiting` and no `node_failed` recorded for the node that was waiting. ### Wave-barrier limitations (deliberate) diff --git a/packages/temporal/src/client/index.ts b/packages/temporal/src/client/index.ts index 706c381f8..349d5718d 100644 --- a/packages/temporal/src/client/index.ts +++ b/packages/temporal/src/client/index.ts @@ -1,13 +1,26 @@ -// Client-side entry point: starting and cancelling runs. Split from the root entry -// so a backend-only consumer never pulls in @temporalio/worker and its native binary. +// Client-side entry point: starting and cancelling runs, delivering verdicts. Split from +// the root entry so a backend-only consumer never pulls in @temporalio/worker and its +// native binary. import { Client, WorkflowNotFoundError } from '@temporalio/client'; -import { DEFAULT_TASK_QUEUE, RUN_WORKFLOW_NAME, executionWorkflowId } from '../constants'; -import type { BaseNode, WorkflowEnginePort, WorkflowExecutionInput } from '../core-contract'; +import { DEFAULT_TASK_QUEUE, RESOLVE_NODE_UPDATE_NAME, RUN_WORKFLOW_NAME, executionWorkflowId } from '../constants'; +import type { + BaseNode, + CompletedNodeExecution, + ResolveNodeResult, + WorkflowEnginePort, + WorkflowExecutionInput, +} from '../core-contract'; // Type-only, so nothing from the workflow entry reaches this bundle at runtime — the // client must not load @temporalio/workflow. It is imported purely to type the start -// call below; see the note there. -import type { runWorkflow } from '../workflow/run-workflow'; +// and update calls below; see the note there. +import type { ResolveNodeUpdateInput, runWorkflow } from '../workflow/run-workflow'; +import { mapResolveNodeError } from './resolve-node-result'; + +// Without a worker nobody validates an update, and the RPC would wait for the server's +// own limit. The deadline settles nothing about the update: it may already be accepted, +// or reach the next worker later, so a resend may hear verdict_already_delivered. +const DEFAULT_RESOLVE_TIMEOUT_MS = 10_000; export type TemporalWorkflowEngineOptions = { // A ready client, or a factory awaited once on first use. The factory form keeps @@ -15,16 +28,20 @@ export type TemporalWorkflowEngineOptions = { client: Client | (() => Promise); // Must match the queue the worker serves — default on both sides is DEFAULT_TASK_QUEUE. taskQueue?: string; + // How long resolveNode waits for the update to be accepted before answering delivery_timeout. + resolveTimeoutMs?: number; }; export class TemporalWorkflowEngine implements WorkflowEnginePort { private readonly clientSource: Client | (() => Promise); private readonly taskQueue: string; + private readonly resolveTimeoutMs: number; private clientPromise: Promise | undefined; constructor(options: TemporalWorkflowEngineOptions) { this.clientSource = options.client; this.taskQueue = options.taskQueue ?? DEFAULT_TASK_QUEUE; + this.resolveTimeoutMs = options.resolveTimeoutMs ?? DEFAULT_RESOLVE_TIMEOUT_MS; } async submit(input: WorkflowExecutionInput): Promise { @@ -54,6 +71,28 @@ export class TemporalWorkflowEngine implement } } + async resolveNode( + executionId: string, + nodeId: string, + resolution: CompletedNodeExecution, + ): Promise { + const client = await this.client(); + const handle = client.workflow.getHandle(executionWorkflowId(executionId)); + try { + // Addressed by name and typed through the sandbox's input type, like `start` above. + // The update id stays the SDK's random one: the server deduplicates by id, so a + // deterministic id would hand a second decider the first decider's outcome as a success. + await client.withDeadline(Date.now() + this.resolveTimeoutMs, () => + handle.executeUpdate(RESOLVE_NODE_UPDATE_NAME, { + args: [{ nodeId, resolution }], + }), + ); + return {}; + } catch (error) { + return mapResolveNodeError(error); + } + } + private client(): Promise { if (!this.clientPromise) { this.clientPromise = diff --git a/packages/temporal/src/client/resolve-node-result.ts b/packages/temporal/src/client/resolve-node-result.ts new file mode 100644 index 000000000..56662d5da --- /dev/null +++ b/packages/temporal/src/client/resolve-node-result.ts @@ -0,0 +1,30 @@ +import { + ApplicationFailure, + WorkflowNotFoundError, + WorkflowUpdateFailedError, + WorkflowUpdateRPCTimeoutOrCancelledError, +} from '@temporalio/client'; + +import { RESOLVE_NODE_REJECTIONS, type ResolveNodeRejection, type ResolveNodeResult } from '../core-contract'; + +function isRejection(type: string): type is ResolveNodeRejection { + return (RESOLVE_NODE_REJECTIONS as readonly string[]).includes(type); +} + +// Turns the errors a verdict delivery can end in into the port's result. Anything the +// port has no code for is rethrown: a bug to surface, not a code to invent. +export function mapResolveNodeError(error: unknown): ResolveNodeResult { + if (error instanceof WorkflowUpdateFailedError && error.cause instanceof ApplicationFailure) { + const { type, message } = error.cause; + if (typeof type === 'string' && isRejection(type)) { + return { error: { code: type, message } }; + } + } + if (error instanceof WorkflowNotFoundError) { + return { error: { code: 'run_not_found', message: error.message } }; + } + if (error instanceof WorkflowUpdateRPCTimeoutOrCancelledError) { + return { error: { code: 'delivery_timeout', message: error.message } }; + } + throw error; +} diff --git a/packages/temporal/src/constants.ts b/packages/temporal/src/constants.ts index 4afa44121..50b539c71 100644 --- a/packages/temporal/src/constants.ts +++ b/packages/temporal/src/constants.ts @@ -7,6 +7,11 @@ export const DEFAULT_TASK_QUEUE = 'workflow-execution'; // because a rename here silently strands every already-started run. export const RUN_WORKFLOW_NAME = 'runWorkflow'; +// The update a verdict arrives through. The client addresses it by name for the same +// reason it starts the workflow by name: it must not load workflow code. Pinned by the +// same test. +export const RESOLVE_NODE_UPDATE_NAME = 'resolveNode'; + // One Temporal Workflow Execution per Workflow Builder execution row. Deterministic, // so cancel can address a run knowing only the execution id. export function executionWorkflowId(executionId: string): string { diff --git a/packages/temporal/src/core-contract.ts b/packages/temporal/src/core-contract.ts index a00a79e1b..e75ba6ac5 100644 --- a/packages/temporal/src/core-contract.ts +++ b/packages/temporal/src/core-contract.ts @@ -19,6 +19,7 @@ import type { BaseNode } from '../../types/src/workflow-execution/execution-mode export { NodeExecutionError, PermanentNodeExecutionError, + RESOLVE_NODE_REJECTIONS, TransientNodeExecutionError, classifyNodeError, resolveExecutor, @@ -45,6 +46,8 @@ export type { export type { ExecutionEventType, ExecutionStatus, + ResolveNodeRejection, + ResolveNodeResult, WorkflowEnginePort, WorkflowExecutionInput, } from './workflow/core-contract'; diff --git a/packages/temporal/src/index.ts b/packages/temporal/src/index.ts index 040502a2a..958b721b1 100644 --- a/packages/temporal/src/index.ts +++ b/packages/temporal/src/index.ts @@ -11,7 +11,7 @@ export type { CreateActivitiesOptions } from './activities'; export type { ExecutionStore } from './store'; -export { DEFAULT_TASK_QUEUE, RUN_WORKFLOW_NAME, executionWorkflowId } from './constants'; +export { DEFAULT_TASK_QUEUE, RESOLVE_NODE_UPDATE_NAME, RUN_WORKFLOW_NAME, executionWorkflowId } from './constants'; export { DEFAULT_DATABASE_ACTIVITY_PROFILE, DEFAULT_NODE_ACTIVITY_PROFILE } from './workflow/activity-profiles'; export type { ActivityProfile, NodeActivityProfiles } from './workflow/activity-profiles'; @@ -31,6 +31,8 @@ export type { NodeExecutionResult, NodeExecutor, NodeExecutorRegistry, + ResolveNodeRejection, + ResolveNodeResult, WaitingNodeExecution, WorkflowDefinition, WorkflowEdgeDefinition, diff --git a/packages/temporal/src/workflow/core-contract.ts b/packages/temporal/src/workflow/core-contract.ts index ba53b4d42..d2f20359f 100644 --- a/packages/temporal/src/workflow/core-contract.ts +++ b/packages/temporal/src/workflow/core-contract.ts @@ -1,3 +1,4 @@ +import type { CompletedNodeExecution, ResolveNodeResult } from '../../../execution-core/src/workflow'; import type { ExecutionEventType, ExecutionStatus } from '../../../types/src/workflow-execution/execution-events'; import type { BaseNode, WorkflowDefinition } from '../../../types/src/workflow-execution/execution-model'; @@ -14,7 +15,10 @@ export type { CompletedNodeExecution, ExecutionContext, NodeExecutionResult, + ResolveNodeRejection, + ResolveNodeResult, RunGraphOutcome, + VerdictRejection, } from '../../../execution-core/src/workflow'; export type { BaseNode } from '../../../types/src/workflow-execution/execution-model'; @@ -26,6 +30,8 @@ export type { ExecutionEventType, ExecutionStatus } from '../../../types/src/wor // emitted .d.ts and breaks types for consumers, since the package is not published. // Restating it in terms of the relatively-imported types keeps dist self-contained. // `test/core-contract.test.ts` fails to compile if this ever drifts from the core. +// The built d.ts already inlines those types through tsconfig `paths`, so the restatement +// can likely become a re-export (follow-up: temporal-core-contract-reexport). export type WorkflowExecutionInput = { workflowId: string; executionId: string; @@ -39,6 +45,7 @@ export type WorkflowExecutionInput = { export interface WorkflowEnginePort { submit(input: WorkflowExecutionInput): Promise; cancel(executionId: string): Promise; + resolveNode(executionId: string, nodeId: string, resolution: CompletedNodeExecution): Promise; } // Restated for the same reason as WorkflowExecutionInput: the core's port module diff --git a/packages/temporal/src/workflow/durable-pause.decision-log.md b/packages/temporal/src/workflow/durable-pause.decision-log.md index 347e0994d..2ed190b55 100644 --- a/packages/temporal/src/workflow/durable-pause.decision-log.md +++ b/packages/temporal/src/workflow/durable-pause.decision-log.md @@ -57,6 +57,9 @@ verdict carries. This file records the decisions behind the Temporal side of the for it would be accepted with no effect. A real outage fails the `node_failed` emit too and ends the run, so the gap needs the database to recover between two consecutive emits. +- **The client answers with results, not throws.** `resolveNode` addresses the update by + name, bounds the RPC (`resolveTimeoutMs`), and maps the SDK's errors to `{ error: { code } }` + with the codes declared once in the core's port; an unknown type is rethrown as a bug. - **Names.** Update `resolveNode`, input `{ nodeId, resolution }`. The verdict content is opaque here: `resolution` is a `CompletedNodeExecution` passed to the parked node untouched. Giving it a domain shape belongs to the decision-contract work. diff --git a/packages/temporal/src/workflow/run-workflow.ts b/packages/temporal/src/workflow/run-workflow.ts index 2bace054a..59eec3fa5 100644 --- a/packages/temporal/src/workflow/run-workflow.ts +++ b/packages/temporal/src/workflow/run-workflow.ts @@ -15,6 +15,7 @@ import { setHandler, } from '@temporalio/workflow'; +import { RESOLVE_NODE_UPDATE_NAME } from '../constants'; import type { Activities } from './activities-interface'; import { DEFAULT_DATABASE_ACTIVITY_PROFILE, type NodeActivityProfiles } from './activity-profiles'; import { @@ -41,7 +42,7 @@ export type ResolveNodeUpdateInput = { // Update-not-signal and the annotation shape: see durable-pause.decision-log.md. export const resolveNodeUpdate: ReturnType> = - defineUpdate('resolveNode'); + defineUpdate(RESOLVE_NODE_UPDATE_NAME); export type RunWorkflowOptions = { nodeActivityProfiles?: NodeActivityProfiles; @@ -55,6 +56,8 @@ export function createRunWorkflow(options: RunWorkflowOptions = {}) { return async function runWorkflow(input: WorkflowExecutionInput): Promise { // Per-instance: must stay inside the workflow function (durable-pause.decision-log.md). + // Keyed by node id: a node is scheduled once per run. A rerun that re-parks one needs + // the key to carry the attempt (follow-up: decision-attempt-in-engine). const waits = new Map(); const knownNodes = new Set(input.definition.nodes.map((node) => node.id)); diff --git a/packages/temporal/src/workflow/verdict-validation.test.ts b/packages/temporal/src/workflow/verdict-validation.test.ts index 207a443eb..61444e202 100644 --- a/packages/temporal/src/workflow/verdict-validation.test.ts +++ b/packages/temporal/src/workflow/verdict-validation.test.ts @@ -1,8 +1,9 @@ import { defaultPayloadConverter } from '@temporalio/common'; import { ApplicationFailure } from '@temporalio/workflow'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, expectTypeOf, it } from 'vitest'; -import { type NodeWaitState, validateVerdict } from './verdict-validation'; +import type { VerdictRejection } from './core-contract'; +import { type NodeWaitState, VERDICT_REJECTION_MESSAGES, validateVerdict } from './verdict-validation'; const KNOWN_NODES = new Set(['start', 'gate', 'after']); const GATE_WAITING = new Map([['gate', { status: 'waiting' }]]); @@ -18,6 +19,12 @@ function rejection(verdict?: unknown, waits: ReadonlyMap } describe('validateVerdict', () => { + it('throws exactly the codes the port declares for the validator', () => { + type Thrown = (typeof VERDICT_REJECTION_MESSAGES)[keyof typeof VERDICT_REJECTION_MESSAGES]['code']; + + expectTypeOf().toEqualTypeOf(); + }); + it('accepts a well-formed verdict for a waiting node', () => { expect(rejection({ nodeId: 'gate', resolution: { output: 'ok' } })).toBeUndefined(); expect(rejection({ nodeId: 'gate', resolution: { output: null, nextPort: 'approved' } })).toBeUndefined(); diff --git a/packages/temporal/src/workflow/verdict-validation.ts b/packages/temporal/src/workflow/verdict-validation.ts index 59129c4ff..915dc02e0 100644 --- a/packages/temporal/src/workflow/verdict-validation.ts +++ b/packages/temporal/src/workflow/verdict-validation.ts @@ -1,12 +1,33 @@ import { ApplicationFailure } from '@temporalio/workflow'; -import type { CompletedNodeExecution } from './core-contract'; +import type { CompletedNodeExecution, VerdictRejection } from './core-contract'; // A node's wait lifecycle; no entry means the node is not waiting. export type NodeWaitState = { status: 'waiting' } | { status: 'resolved'; resolution: CompletedNodeExecution }; -function malformed(message: string): ApplicationFailure { - return ApplicationFailure.nonRetryable(message, 'verdict_malformed'); +// Every way a verdict is refused before acceptance: the port's code and the one message +// for it. `{value}` is the single interpolation slot, as in the backend's dictionaries. +export const VERDICT_REJECTION_MESSAGES = { + not_an_object: { code: 'verdict_malformed', message: 'update input must be a { nodeId, resolution } object' }, + node_id_blank: { code: 'verdict_malformed', message: 'nodeId must be a non-empty string' }, + resolution_not_an_object: { code: 'verdict_malformed', message: 'resolution must be an object' }, + unknown_resolution_key: { code: 'verdict_malformed', message: "unknown resolution key '{value}'" }, + next_port_invalid: { + code: 'verdict_malformed', + message: "nextPort must be a non-empty string other than the reserved 'errorRoute'", + }, + unknown_node: { code: 'verdict_for_unknown_node', message: "no node '{value}' in this run" }, + already_delivered: { code: 'verdict_already_delivered', message: "node '{value}' already has a verdict" }, + not_waiting: { code: 'node_not_waiting', message: "node '{value}' is not waiting for a verdict" }, +} as const satisfies Record; + +function reject(key: keyof typeof VERDICT_REJECTION_MESSAGES, value?: string): ApplicationFailure { + const { code, message } = VERDICT_REJECTION_MESSAGES[key]; + // A function replacer, so a value containing `$&` or `$1` lands verbatim. + return ApplicationFailure.nonRetryable( + message.replace('{value}', () => value ?? ''), + code, + ); } // Runs before the update is accepted: a throw rejects it, writes nothing to history @@ -17,34 +38,34 @@ export function validateVerdict( waits: ReadonlyMap, ): void { if (typeof verdict !== 'object' || verdict === null) { - throw malformed('Update input must be a { nodeId, resolution } object'); + throw reject('not_an_object'); } const { nodeId, resolution } = verdict as { nodeId?: unknown; resolution?: unknown }; if (typeof nodeId !== 'string' || nodeId.length === 0) { - throw malformed('nodeId must be a non-empty string'); + throw reject('node_id_blank'); } // No `output` key is accepted as `output: undefined`: the default payload converter // is JSON and drops undefined fields before the update reaches the workflow. if (typeof resolution !== 'object' || resolution === null || Array.isArray(resolution)) { - throw malformed('resolution must be an object'); + throw reject('resolution_not_an_object'); } for (const key of Object.keys(resolution)) { if (key !== 'output' && key !== 'nextPort') { - throw malformed(`Unknown resolution key "${key}"`); + throw reject('unknown_resolution_key', key); } } const { nextPort } = resolution as { nextPort?: unknown }; if (nextPort !== undefined && (typeof nextPort !== 'string' || nextPort.length === 0 || nextPort === 'errorRoute')) { - throw malformed('nextPort must be a non-empty string other than the reserved errorRoute'); + throw reject('next_port_invalid'); } if (!knownNodes.has(nodeId)) { - throw ApplicationFailure.nonRetryable(`No node "${nodeId}" in this run`, 'verdict_for_unknown_node'); + throw reject('unknown_node', nodeId); } const state = waits.get(nodeId); if (state?.status === 'resolved') { - throw ApplicationFailure.nonRetryable(`Node "${nodeId}" already has a verdict`, 'verdict_already_delivered'); + throw reject('already_delivered', nodeId); } if (state === undefined) { - throw ApplicationFailure.nonRetryable(`Node "${nodeId}" is not waiting for a verdict`, 'node_not_waiting'); + throw reject('not_waiting', nodeId); } } diff --git a/packages/temporal/test/api-surface.test.ts b/packages/temporal/test/api-surface.test.ts index d5c0634d9..e5f6f0b7e 100644 --- a/packages/temporal/test/api-surface.test.ts +++ b/packages/temporal/test/api-surface.test.ts @@ -19,6 +19,7 @@ describe('public API surface', () => { 'DEFAULT_TASK_QUEUE', 'NodeExecutionError', 'PermanentNodeExecutionError', + 'RESOLVE_NODE_UPDATE_NAME', 'RUN_WORKFLOW_NAME', 'TransientNodeExecutionError', 'WorkflowBuilderPlugin', @@ -50,6 +51,13 @@ describe('public API surface', () => { expect(rootEntry.RUN_WORKFLOW_NAME in workflowEntry).toBe(true); expect(rootEntry.RUN_WORKFLOW_NAME).toBe('runWorkflow'); }); + + it('names the update the client sends after the one the sandbox handles', () => { + // Same failure mode as the workflow name: a verdict addressed to an update no + // workflow defines is rejected by the server, and every parked run stays parked. + expect(workflowEntry.resolveNodeUpdate.name).toBe(rootEntry.RESOLVE_NODE_UPDATE_NAME); + expect(rootEntry.RESOLVE_NODE_UPDATE_NAME).toBe('resolveNode'); + }); }); describe('default activity profiles', () => { diff --git a/packages/temporal/test/resolve-node-deadline.test.ts b/packages/temporal/test/resolve-node-deadline.test.ts new file mode 100644 index 000000000..5e52c4fcb --- /dev/null +++ b/packages/temporal/test/resolve-node-deadline.test.ts @@ -0,0 +1,45 @@ +import type { Client } from '@temporalio/client'; +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { TemporalWorkflowEngine } from '../src/client/index'; +import { RESOLVE_NODE_UPDATE_NAME } from '../src/constants'; + +const NOW = Date.parse('2026-09-10T12:00:00Z'); + +// Just enough of a Client for resolveNode: the deadline wrapper and the handle's update call. +function fakeClient() { + const executeUpdate = vi.fn(async () => {}); + const withDeadline = vi.fn(async (_deadline: number, run: () => Promise) => run()); + const client = { withDeadline, workflow: { getHandle: () => ({ executeUpdate }) } } as unknown as Client; + return { client, executeUpdate, withDeadline }; +} + +describe('TemporalWorkflowEngine.resolveNode deadline', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it('bounds the update RPC by 10 s when resolveTimeoutMs is not given', async () => { + vi.useFakeTimers(); + vi.setSystemTime(NOW); + const { client, withDeadline, executeUpdate } = fakeClient(); + + const result = await new TemporalWorkflowEngine({ client }).resolveNode('e-1', 'n-1', { output: 1 }); + + expect(result).toEqual({}); + expect(withDeadline).toHaveBeenCalledWith(NOW + 10_000, expect.any(Function)); + expect(executeUpdate).toHaveBeenCalledWith(RESOLVE_NODE_UPDATE_NAME, { + args: [{ nodeId: 'n-1', resolution: { output: 1 } }], + }); + }); + + it('bounds it by resolveTimeoutMs when given', async () => { + vi.useFakeTimers(); + vi.setSystemTime(NOW); + const { client, withDeadline } = fakeClient(); + + await new TemporalWorkflowEngine({ client, resolveTimeoutMs: 250 }).resolveNode('e-1', 'n-1', { output: 1 }); + + expect(withDeadline).toHaveBeenCalledWith(NOW + 250, expect.any(Function)); + }); +}); diff --git a/packages/temporal/test/resolve-node-result.test.ts b/packages/temporal/test/resolve-node-result.test.ts new file mode 100644 index 000000000..7eded2be4 --- /dev/null +++ b/packages/temporal/test/resolve-node-result.test.ts @@ -0,0 +1,75 @@ +import { + ApplicationFailure, + WorkflowNotFoundError, + WorkflowUpdateFailedError, + WorkflowUpdateRPCTimeoutOrCancelledError, +} from '@temporalio/client'; +import { describe, expect, it } from 'vitest'; + +import { mapResolveNodeError } from '../src/client/resolve-node-result'; + +function updateFailed(cause?: Error): WorkflowUpdateFailedError { + return new WorkflowUpdateFailedError('Workflow Update failed', cause); +} + +function rejection(type: string, message: string): ApplicationFailure { + return ApplicationFailure.create({ type, message, nonRetryable: true }); +} + +function thrownBy(run: () => unknown): unknown { + try { + run(); + } catch (error) { + return error; + } + return 'did not throw'; +} + +describe('mapResolveNodeError', () => { + it.each(['verdict_malformed', 'verdict_for_unknown_node', 'verdict_already_delivered', 'node_not_waiting'])( + 'answers the validator rejection %s as a result carrying its message', + (code) => { + const error = updateFailed(rejection(code, `because ${code}`)); + + expect(mapResolveNodeError(error)).toEqual({ error: { code, message: `because ${code}` } }); + }, + ); + + it('answers a run the server no longer has as run_not_found', () => { + const error = new WorkflowNotFoundError('workflow execution already completed', 'execution-1', undefined); + + expect(mapResolveNodeError(error)).toEqual({ + error: { code: 'run_not_found', message: 'workflow execution already completed' }, + }); + }); + + it('answers an abandoned RPC as delivery_timeout', () => { + const error = new WorkflowUpdateRPCTimeoutOrCancelledError('Deadline exceeded'); + + expect(mapResolveNodeError(error)).toEqual({ error: { code: 'delivery_timeout', message: 'Deadline exceeded' } }); + }); + + it('rethrows an update failure whose type the port does not declare', () => { + const error = updateFailed(rejection('handler_exploded', 'unexpected')); + + expect(thrownBy(() => mapResolveNodeError(error))).toBe(error); + }); + + it('rethrows an update failure with no ApplicationFailure behind it', () => { + const error = updateFailed(new Error('not a failure')); + + expect(thrownBy(() => mapResolveNodeError(error))).toBe(error); + }); + + it('rethrows an update failure without a cause', () => { + const error = updateFailed(); + + expect(thrownBy(() => mapResolveNodeError(error))).toBe(error); + }); + + it('rethrows anything that is not a delivery outcome', () => { + const error = new Error('connection refused'); + + expect(thrownBy(() => mapResolveNodeError(error))).toBe(error); + }); +}); diff --git a/packages/temporal/test/resolve-node.test.ts b/packages/temporal/test/resolve-node.test.ts new file mode 100644 index 000000000..02c31fad6 --- /dev/null +++ b/packages/temporal/test/resolve-node.test.ts @@ -0,0 +1,181 @@ +// The client side of the durable pause: a verdict travels through the engine port, and +// every refusal comes back as a result rather than a throw. +import { TestWorkflowEnvironment } from '@temporalio/testing'; +import { Worker, bundleWorkflowCode } from '@temporalio/worker'; +import { fileURLToPath } from 'node:url'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +import { TemporalWorkflowEngine, type TemporalWorkflowEngineOptions } from '../src/client/index'; +import { + type ResolveNodeResult, + WorkflowBuilderPlugin, + type WorkflowDefinition, + executionWorkflowId, +} from '../src/index'; +import { type RecordingStore, createRecordingStore } from './fixtures/graph'; +import { waitUntil } from './fixtures/helpers'; +import { + type PauseHarness, + type PauseTestNode, + SINGLE_GATE_GRAPH, + TWO_GATES_GRAPH, + createPauseExecutors, +} from './fixtures/pause-graph'; + +function whenAnnounced(store: RecordingStore, nodeId: string): Promise { + return waitUntil( + () => store.events.some((event) => event.nodeId === nodeId && event.type === 'node_waiting'), + `node_waiting for ${nodeId}`, + ); +} + +describe('resolveNode through the engine port', () => { + let env: TestWorkflowEnvironment; + let workflowBundle: { code: string }; + + beforeAll(async () => { + [workflowBundle, env] = await Promise.all([ + bundleWorkflowCode({ workflowsPath: fileURLToPath(new URL('fixtures/workflows.ts', import.meta.url)) }), + TestWorkflowEnvironment.createLocal(), + ]); + }, 300_000); + + afterAll(async () => { + await env?.teardown(); + }); + + function createWorker(taskQueue: string, store: RecordingStore, harness: PauseHarness): Promise { + const plugin = new WorkflowBuilderPlugin({ store, executors: harness.executors, taskQueue }); + return Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue: plugin.taskQueue, + workflowBundle, + plugins: [plugin], + }); + } + + function createEngine(taskQueue: string, options: Partial = {}) { + return new TemporalWorkflowEngine({ client: env.client, taskQueue, ...options }); + } + + async function submit( + engine: TemporalWorkflowEngine, + executionId: string, + definition: WorkflowDefinition, + ) { + await engine.submit({ + workflowId: definition.workflowId, + executionId, + definition, + triggerPayload: {}, + variables: {}, + global: {}, + }); + return env.client.workflow.getHandle(executionWorkflowId(executionId)); + } + + it('resumes the parked run; once the run has closed, a verdict answers run_not_found', async () => { + const taskQueue = 'resolve-node-resume'; + const executionId = 'resolve-node-resume-execution'; + const store = createRecordingStore(); + const harness = createPauseExecutors(); + const engine = createEngine(taskQueue); + const handle = await submit(engine, executionId, SINGLE_GATE_GRAPH); + + const worker = await createWorker(taskQueue, store, harness); + await worker.runUntil(async () => { + await whenAnnounced(store, 'gate'); + expect(await engine.resolveNode(executionId, 'gate', { output: 'approved' })).toEqual({}); + await handle.result(); + }); + + expect(harness.executed).toEqual(['start', 'gate', 'after']); + expect(harness.inputsSeen.after.gate).toBe('approved'); + expect(store.statuses.map((entry) => entry.status)).toEqual(['waiting', 'running', 'completed']); + + // The server answers for a closed run on its own; no worker is involved. + expect(await engine.resolveNode(executionId, 'gate', { output: 'late' })).toMatchObject({ + error: { code: 'run_not_found', message: expect.any(String) }, + }); + }, 120_000); + + it('answers each rejection as a result and leaves the parked run resolvable', async () => { + const taskQueue = 'resolve-node-rejections'; + const executionId = 'resolve-node-rejections-execution'; + const store = createRecordingStore(); + const harness = createPauseExecutors(); + const engine = createEngine(taskQueue); + const handle = await submit(engine, executionId, TWO_GATES_GRAPH); + + const worker = await createWorker(taskQueue, store, harness); + await worker.runUntil(async () => { + await Promise.all([whenAnnounced(store, 'gate-a'), whenAnnounced(store, 'gate-b')]); + + expect(await engine.resolveNode(executionId, 'ghost', { output: 1 })).toMatchObject({ + error: { code: 'verdict_for_unknown_node', message: expect.any(String) }, + }); + expect(await engine.resolveNode(executionId, 'join', { output: 1 })).toMatchObject({ + error: { code: 'node_not_waiting', message: expect.any(String) }, + }); + // Well-typed for the port, refused by the validator: the reserved port name. + expect(await engine.resolveNode(executionId, 'gate-a', { output: 1, nextPort: 'errorRoute' })).toMatchObject({ + error: { code: 'verdict_malformed', message: expect.any(String) }, + }); + + expect(await engine.resolveNode(executionId, 'gate-a', { output: 'first' })).toEqual({}); + expect(await engine.resolveNode(executionId, 'gate-a', { output: 'second' })).toMatchObject({ + error: { code: 'verdict_already_delivered', message: expect.any(String) }, + }); + expect(await engine.resolveNode(executionId, 'gate-b', { output: 'b-verdict' })).toEqual({}); + await handle.result(); + }); + + expect(harness.inputsSeen.join['gate-a']).toBe('first'); + expect(harness.inputsSeen.join['gate-b']).toBe('b-verdict'); + expect(harness.executed.filter((id) => id === 'join')).toHaveLength(1); + expect(store.statuses.map((entry) => entry.status)).toEqual(['waiting', 'running', 'completed']); + }, 120_000); + + it('with no worker the verdict answers delivery_timeout at the deadline; the retry lands, or hears that the first one did', async () => { + const taskQueue = 'resolve-node-timeout'; + const executionId = 'resolve-node-timeout-execution'; + const store = createRecordingStore(); + const harness = createPauseExecutors(); + const engine = createEngine(taskQueue); + // Only the call meant to time out gets the short deadline: the retry below races a + // worker's cold start, which the default deadline is sized for. + const impatient = createEngine(taskQueue, { resolveTimeoutMs: 1000 }); + const handle = await submit(engine, executionId, SINGLE_GATE_GRAPH); + + const worker1 = await createWorker(taskQueue, store, harness); + await worker1.runUntil( + waitUntil(() => store.statuses.some((entry) => entry.status === 'waiting'), 'the waiting status'), + ); + + // Parked, and nobody polls the queue: the update cannot reach a validator. + const startedAt = Date.now(); + const timedOut = await impatient.resolveNode(executionId, 'gate', { output: 'first attempt' }); + const elapsedMs = Date.now() - startedAt; + expect(timedOut).toMatchObject({ error: { code: 'delivery_timeout', message: expect.any(String) } }); + expect(elapsedMs).toBeGreaterThanOrEqual(900); + expect(elapsedMs).toBeLessThan(5000); + + let retry: ResolveNodeResult | undefined; + const worker2 = await createWorker(taskQueue, store, harness); + await worker2.runUntil(async () => { + retry = await engine.resolveNode(executionId, 'gate', { output: 'retry' }); + await handle.result(); + }); + + // The abandoned update is not durable, but the server may still hold it and hand it + // to the next worker. The retry's answer says which of the two landed; exactly one did. + expect(retry).toBeDefined(); + if (retry?.error !== undefined) { + expect(retry.error.code).toBe('verdict_already_delivered'); + } + expect(harness.inputsSeen.after.gate).toBe(retry?.error === undefined ? 'retry' : 'first attempt'); + expect(harness.executed).toEqual(['start', 'gate', 'after']); + expect(store.statuses.map((entry) => entry.status)).toEqual(['waiting', 'running', 'completed']); + }, 120_000); +});