From 4807b7f65bedff068e4553a0f9a55cad81babcc5 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 08:40:57 +0200 Subject: [PATCH 01/17] feat(temporal): resolveNode on the engine port delivers a verdict to a parked run WB-501 step 1. WorkflowEnginePort gains a required resolveNode(executionId, nodeId, resolution) that answers every expected refusal as a result, never a throw: the validator's four codes plus run_not_found and delivery_timeout. The codes live once, as `as const` arrays in the execution-core port module; the validator types its throws with them and the Temporal adapter derives its runtime check from them. The adapter addresses the update by name (RESOLVE_NODE_UPDATE_NAME, a deliberate root export pinned like RUN_WORKFLOW_NAME) and bounds the RPC with client.withDeadline (resolveTimeoutMs, default 10 s). Unknown failure types are rethrown. The harness showed that an update abandoned at the client deadline is not dropped: the server still hands it to the next worker, so a retry may hear verdict_already_delivered. The test pins "exactly one lands". The validator's messages moved into one dictionary keyed by rejection, in the backend's style. --- packages/execution-core/src/index.ts | 9 +- .../src/ports/workflow-engine.port.ts | 21 ++ packages/execution-core/src/workflow.ts | 8 +- packages/temporal/src/client/index.ts | 51 ++++- .../src/client/resolve-node-result.ts | 30 +++ packages/temporal/src/constants.ts | 5 + packages/temporal/src/core-contract.ts | 3 + packages/temporal/src/index.ts | 4 +- .../temporal/src/workflow/core-contract.ts | 7 + .../temporal/src/workflow/run-workflow.ts | 3 +- .../src/workflow/verdict-validation.ts | 43 +++-- packages/temporal/test/api-surface.test.ts | 8 + .../temporal/test/resolve-node-result.test.ts | 75 ++++++++ packages/temporal/test/resolve-node.test.ts | 181 ++++++++++++++++++ 14 files changed, 427 insertions(+), 21 deletions(-) create mode 100644 packages/temporal/src/client/resolve-node-result.ts create mode 100644 packages/temporal/test/resolve-node-result.test.ts create mode 100644 packages/temporal/test/resolve-node.test.ts 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/src/client/index.ts b/packages/temporal/src/client/index.ts index 706c381f8..b90e7fd59 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. A timed-out update is not durable, yet the server may still hand it to the +// next worker: the caller resends and 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/run-workflow.ts b/packages/temporal/src/workflow/run-workflow.ts index 2bace054a..4cdafea33 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; diff --git a/packages/temporal/src/workflow/verdict-validation.ts b/packages/temporal/src/workflow/verdict-validation.ts index 59129c4ff..dee68c934 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. +const VERDICT_REJECTIONS = { + 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_REJECTIONS, value?: string): ApplicationFailure { + const { code, message } = VERDICT_REJECTIONS[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-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); +}); From fc1f8090f4ef48a8643e316966b39409f7170ea8 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 09:11:59 +0200 Subject: [PATCH 02/17] feat(backend): domain pieces the decision endpoint builds on WB-501 step 2. toNodeResolution turns an accepted Decision and its matched action into the completion the engine delivers: output is the decision itself, nextPort the action's port, rerun-source excluded at the type level and the reserved errorRoute port refused. findDecisionRequest reads a node's request out of the parsed snapshot. A non-resume submission carrying edits is now refused with edits_not_allowed, checked before the field rules so the refusal names the edits. --- .../src/domain/decision/decision-issues.ts | 1 + .../decision/find-decision-request.test.ts | 42 +++++++++++++++ .../domain/decision/find-decision-request.ts | 15 ++++++ .../domain/decision/node-resolution.test.ts | 54 +++++++++++++++++++ .../src/domain/decision/node-resolution.ts | 17 ++++++ .../validate-submitted-decision.test.ts | 39 +++++++++----- .../decision/validate-submitted-decision.ts | 10 ++-- 7 files changed, 163 insertions(+), 15 deletions(-) create mode 100644 apps/backend/src/domain/decision/find-decision-request.test.ts create mode 100644 apps/backend/src/domain/decision/find-decision-request.ts create mode 100644 apps/backend/src/domain/decision/node-resolution.test.ts create mode 100644 apps/backend/src/domain/decision/node-resolution.ts diff --git a/apps/backend/src/domain/decision/decision-issues.ts b/apps/backend/src/domain/decision/decision-issues.ts index 78412216b..62e357dd2 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", 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..67e961962 --- /dev/null +++ b/apps/backend/src/domain/decision/find-decision-request.ts @@ -0,0 +1,15 @@ +import type { DecisionRequest } from '@workflow-builder/types/workflow-execution/decision-request'; + +import type { WorkflowSnapshot } from '../mapper/snapshot-schema'; + +export type FindDecisionRequestResult = + | { request: DecisionRequest; error?: undefined } + | { request?: undefined; error: 'node_not_found' | 'node_without_decision_request' }; + +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..82a2b2dd7 --- /dev/null +++ b/apps/backend/src/domain/decision/node-resolution.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, expectTypeOf, it } from 'vitest'; + +import type { Decision } from '@workflow-builder/types/workflow-execution/decision-request'; + +import { type RoutedDecision, type RoutedDecisionAction, 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', () => { + 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..504c02822 --- /dev/null +++ b/apps/backend/src/domain/decision/node-resolution.ts @@ -0,0 +1,17 @@ +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 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; From ef04b0586352e60738e78cf8f186e4e9a3724191 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 10:44:21 +0200 Subject: [PATCH 03/17] feat(backend): count a node's waits, the attempt a decision addresses WB-501 step 3. countNodeWaits(executionId, nodeId) counts the node_waiting events of one node in one run. The number is the wait instance a decision must name, and zero says the node never parked. Shared with the coming pending-decision resource, so it lives beside the event query, not in a route. --- apps/backend/src/events/count-node-waits.ts | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) create mode 100644 apps/backend/src/events/count-node-waits.ts 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; +} From 527445f7b584df47cd9a3f5b313a63073e17fbfc Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 11:07:25 +0200 Subject: [PATCH 04/17] feat(backend): POST /api/executions/:id/decision delivers a human's decision to a parked run WB-501 step 4. One door for every future channel. The route loads the row, authorizes executions:decide with the row's attributes (a deny wins over 404), refuses terminal and cancelling runs, parses the body, reads the node's request out of the parsed snapshot, judges the submission, checks the wait instance (attempt = the node's node_waiting count), refuses rerun-source with 501 until the engine can re-run a source, and delivers the completion through engine.resolveNode. Every engine refusal is answered on the first try; no retry. Codes, statuses and messages live once in decision-refusals.ts: the status map spells each code, situations are typed against it, and total maps over the engine's and the lookup's codes make a new code a compile error here. --- apps/backend/src/auth/auth-port.ts | 3 +- .../domain/decision/find-decision-request.ts | 4 +- .../domain/decision/node-resolution.test.ts | 4 +- .../src/domain/decision/node-resolution.ts | 4 + .../src/routes/decision-refusals.test.ts | 33 ++ apps/backend/src/routes/decision-refusals.ts | 72 +++ apps/backend/src/routes/decision.test.ts | 510 ++++++++++++++++++ apps/backend/src/routes/decision.ts | 108 ++++ apps/backend/src/server.ts | 2 + 9 files changed, 737 insertions(+), 3 deletions(-) create mode 100644 apps/backend/src/routes/decision-refusals.test.ts create mode 100644 apps/backend/src/routes/decision-refusals.ts create mode 100644 apps/backend/src/routes/decision.test.ts create mode 100644 apps/backend/src/routes/decision.ts 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/find-decision-request.ts b/apps/backend/src/domain/decision/find-decision-request.ts index 67e961962..417cce609 100644 --- a/apps/backend/src/domain/decision/find-decision-request.ts +++ b/apps/backend/src/domain/decision/find-decision-request.ts @@ -2,9 +2,11 @@ import type { DecisionRequest } from '@workflow-builder/types/workflow-execution 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: 'node_not_found' | 'node_without_decision_request' }; + | { request?: undefined; error: FindDecisionRequestError }; export function findDecisionRequest(snapshot: WorkflowSnapshot, nodeId: string): FindDecisionRequestResult { const node = snapshot.nodes.find((candidate) => candidate.id === nodeId); diff --git a/apps/backend/src/domain/decision/node-resolution.test.ts b/apps/backend/src/domain/decision/node-resolution.test.ts index 82a2b2dd7..aec0f38bb 100644 --- a/apps/backend/src/domain/decision/node-resolution.test.ts +++ b/apps/backend/src/domain/decision/node-resolution.test.ts @@ -2,7 +2,7 @@ import { describe, expect, expectTypeOf, it } from 'vitest'; import type { Decision } from '@workflow-builder/types/workflow-execution/decision-request'; -import { type RoutedDecision, type RoutedDecisionAction, toNodeResolution } from './node-resolution'; +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; @@ -47,6 +47,8 @@ describe('toNodeResolution', () => { }); 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 index 504c02822..9c8a27f4e 100644 --- a/apps/backend/src/domain/decision/node-resolution.ts +++ b/apps/backend/src/domain/decision/node-resolution.ts @@ -9,6 +9,10 @@ import type { 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`); 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..d583422f6 --- /dev/null +++ b/apps/backend/src/routes/decision-refusals.test.ts @@ -0,0 +1,33 @@ +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, 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, + body: { code: 'node_not_found', message: "No node '$&-$1' in this execution" }, + }); + expect(await answer('attempt_mismatch', undefined, { attempt: 1 })).toEqual({ + status: 409, + body: { + code: 'decision_attempt_mismatch', + message: 'The decision names a wait that is not the current one', + attempt: 1, + }, + }); + }); +}); diff --git a/apps/backend/src/routes/decision-refusals.ts b/apps/backend/src/routes/decision-refusals.ts new file mode 100644 index 000000000..931195456 --- /dev/null +++ b/apps/backend/src/routes/decision-refusals.ts @@ -0,0 +1,72 @@ +import type { Context } from 'hono'; + +import type { ResolveNodeRejection } from '@workflow-builder/execution-core/workflow'; + +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, +} 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", + }, +} 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: 'fault', +} as const satisfies Record; + +export function refuse(c: Context, refusal: DecisionRefusal, value?: string, extra: Record = {}) { + const { code, message } = DECISION_REFUSALS[refusal]; + // A function replacer, so a value containing `$&` or `$1` lands verbatim. + return c.json( + { code, message: message.replace('{value}', () => value ?? ''), ...extra }, + DECISION_REFUSAL_STATUS[code], + ); +} diff --git a/apps/backend/src/routes/decision.test.ts b/apps/backend/src/routes/decision.test.ts new file mode 100644 index 000000000..0495d926c --- /dev/null +++ b/apps/backend/src/routes/decision.test.ts @@ -0,0 +1,510 @@ +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.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('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', + }); + }); + + 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.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('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..e3cfe8cea --- /dev/null +++ b/apps/backend/src/routes/decision.ts @@ -0,0 +1,108 @@ +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, AuthVariables } 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 { TenantVariables } from '../tenant'; +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<{ Variables: AuthVariables & TenantVariables }> { + const routes = new Hono<{ Variables: AuthVariables & TenantVariables }>(); + + 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, + 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'); + + const parsedBody = z.safeParse(decisionBodySchema, await c.req.json()); + if (!parsedBody.success) { + return refuse(c, 'body_invalid', undefined, { 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, + error: { issues: formatValidationDetails(parsedSnapshot.error) }, + }); + throw new Error(`stored snapshot of execution ${executionId} no longer parses`); + } + const found = findDecisionRequest(parsedSnapshot.data, nodeId); + if (found.error !== undefined) return refuse(c, LOOKUP_REFUSALS[found.error], nodeId); + + const validated = validateSubmittedDecision(found.request, submitted); + if (validated.error !== undefined) { + return refuse(c, 'decision_invalid', undefined, { details: [validated.error] }); + } + const { decision, action } = validated; + + const waits = await countNodeWaits(executionId, nodeId); + if (waits === 0) return refuse(c, 'node_never_parked', nodeId); + if (waits !== attempt) return refuse(c, 'attempt_mismatch', undefined, { attempt: waits }); + + // (follow-up: decision-rerun-source) + if (!hasNodeResolution(decision) || action.effect === 'rerun-source') { + return refuse(c, 'effect_not_supported', action.name); + } + + const result = await getWorkflowEngine().resolveNode(executionId, 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, nodeId); + } + + logger.info('decision delivered', { executionId, nodeId, attempt, action: action.name, effect: decision.effect }); + return c.json({ executionId, nodeId, attempt, action: action.name, effect: decision.effect }); + }); + + return routes; +} diff --git a/apps/backend/src/server.ts b/apps/backend/src/server.ts index 9a2d7773e..0ad6afc86 100644 --- a/apps/backend/src/server.ts +++ b/apps/backend/src/server.ts @@ -16,6 +16,7 @@ import { runMigrations } from './db/migrate'; import { env } from './env'; import { logger } from './logger'; import { createRateLimitMiddleware } from './middleware/rate-limit'; +import { createDecisionRoutes } from './routes/decision'; import { createExecutionsRoutes } from './routes/executions'; import { createVisualizeRoutes } from './routes/visualize'; import { createWorkflowsRoutes } from './routes/workflows'; @@ -81,6 +82,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 From 81e26c3549c105f163efc1388805d5d7083f4229 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 11:08:17 +0200 Subject: [PATCH 05/17] test(temporal): pin that the validator throws every code the port declares A code added to the port's validator group without a throw site compiled fine and stayed dead. The dictionary is exported and a type-level pin equates its codes with VerdictRejection in both directions. --- .../temporal/src/workflow/verdict-validation.test.ts | 11 +++++++++-- packages/temporal/src/workflow/verdict-validation.ts | 2 +- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/temporal/src/workflow/verdict-validation.test.ts b/packages/temporal/src/workflow/verdict-validation.test.ts index 207a443eb..ccbc41583 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_REJECTIONS, 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_REJECTIONS)[keyof typeof VERDICT_REJECTIONS]['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 dee68c934..4cf5b2aee 100644 --- a/packages/temporal/src/workflow/verdict-validation.ts +++ b/packages/temporal/src/workflow/verdict-validation.ts @@ -7,7 +7,7 @@ export type NodeWaitState = { status: 'waiting' } | { status: 'resolved'; resolu // 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. -const VERDICT_REJECTIONS = { +export const VERDICT_REJECTIONS = { 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' }, From d489776dd6f5b29bb6e5721c429930916fff824c Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 11:14:00 +0200 Subject: [PATCH 06/17] feat(backend): a decision the engine could not confirm in time answers 503 with Retry-After WB-501 step 5. The adapter's delivery_timeout becomes decision_delivery_timeout. The update is not durable until a worker accepts it, yet the server may still hand it to the next worker, so the message says the decision may or may not have landed and that a retry answering decision_already_made means it did. No retry anywhere on the server side. --- .../src/routes/decision-refusals.test.ts | 18 ++++++++++++++++- apps/backend/src/routes/decision-refusals.ts | 20 ++++++++++++++----- apps/backend/src/routes/decision.test.ts | 14 +++++++++++++ 3 files changed, 46 insertions(+), 6 deletions(-) diff --git a/apps/backend/src/routes/decision-refusals.test.ts b/apps/backend/src/routes/decision-refusals.test.ts index d583422f6..6622004a8 100644 --- a/apps/backend/src/routes/decision-refusals.test.ts +++ b/apps/backend/src/routes/decision-refusals.test.ts @@ -6,7 +6,11 @@ import { DECISION_REFUSALS, DECISION_REFUSAL_STATUS, type DecisionRefusal, refus 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, body: (await response.json()) as Record }; + return { + status: response.status, + retryAfter: response.headers.get('retry-after'), + body: (await response.json()) as Record, + }; } describe('decision refusals', () => { @@ -19,10 +23,12 @@ describe('decision refusals', () => { 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', @@ -30,4 +36,14 @@ describe('decision refusals', () => { }, }); }); + + 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 index 931195456..39b2b72dc 100644 --- a/apps/backend/src/routes/decision-refusals.ts +++ b/apps/backend/src/routes/decision-refusals.ts @@ -15,6 +15,7 @@ export const DECISION_REFUSAL_STATUS = { 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; @@ -43,7 +44,14 @@ export const DECISION_REFUSALS = { code: 'effect_not_supported', message: "Action '{value}' re-runs the proposal source, which is not supported yet", }, -} as const satisfies Record; + // Not durable until accepted, yet the server may still hand it to the next worker. + 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 it did.', + headers: { 'Retry-After': '5' }, + }, +} as const satisfies Record }>; export type DecisionRefusal = keyof typeof DECISION_REFUSALS; @@ -59,14 +67,16 @@ export const ENGINE_REFUSALS = { run_not_found: 'run_gone', verdict_for_unknown_node: 'fault', verdict_malformed: 'fault', - delivery_timeout: 'fault', + delivery_timeout: 'delivery_timeout', } as const satisfies Record; export function refuse(c: Context, refusal: DecisionRefusal, value?: string, extra: Record = {}) { - const { code, message } = DECISION_REFUSALS[refusal]; + const entry = DECISION_REFUSALS[refusal]; + const headers = 'headers' in entry ? entry.headers : undefined; // A function replacer, so a value containing `$&` or `$1` lands verbatim. return c.json( - { code, message: message.replace('{value}', () => value ?? ''), ...extra }, - DECISION_REFUSAL_STATUS[code], + { code: entry.code, message: entry.message.replace('{value}', () => 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 index 0495d926c..7a45ba5ce 100644 --- a/apps/backend/src/routes/decision.test.ts +++ b/apps/backend/src/routes/decision.test.ts @@ -484,6 +484,20 @@ describe('POST /api/executions/:id/decision - delivery', () => { 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) => { From 982a210665690829389bbd269d2ed473c1df6d6f Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 11:23:57 +0200 Subject: [PATCH 07/17] docs: the decision endpoint, its codes and the reasons behind them WB-501 step 6. The backend README is the one place for the endpoint and its answers; the decision log keeps only the reasons and closes its open points; the Temporal README and decision log say what resolveNode answers with. --- apps/backend/README.md | 19 +++++++++++- apps/backend/auth-port.decision-log.md | 27 ++++++++--------- apps/backend/decision-request.decision-log.md | 29 ++++++++++++++----- packages/temporal/README.md | 5 ++-- .../workflow/durable-pause.decision-log.md | 3 ++ 5 files changed, 59 insertions(+), 24 deletions(-) diff --git a/apps/backend/README.md b/apps/backend/README.md index 7e932a9d3..1770d1e69 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 means it did | ## Running individual processes diff --git a/apps/backend/auth-port.decision-log.md b/apps/backend/auth-port.decision-log.md index dd12a4546..e1e4e2bca 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. ## Alternative Options Considered diff --git a/apps/backend/decision-request.decision-log.md b/apps/backend/decision-request.decision-log.md index 63c1b6c80..f73659069 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; when the rerun loop re-parks a node, the count follows with no change here. +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,16 @@ 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. -## 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/packages/temporal/README.md b/packages/temporal/README.md index 14c3e4ad1..cd61d3f32 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. @@ -166,7 +167,7 @@ Three things are deliberately yours, and knowing which they are makes debugging | Import | Use it for | | ------------------------------------ | ------------------------------------------------------------------------------------------- | | `@workflowbuilder/temporal` | Worker side: the plugin, `createActivities`, shared constants, types | -| `@workflowbuilder/temporal/client` | Starting and cancelling runs | +| `@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/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. From e74e8fd9aba670d14b5e1b54ad0669f63a207656 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 12:29:42 +0200 Subject: [PATCH 08/17] test(backend): pin two more links of the check order and a thrown engine error The row is checked before the body, the attempt before the effect. An error the engine throws instead of returning surfaces as 500. --- apps/backend/src/routes/decision.test.ts | 34 ++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/apps/backend/src/routes/decision.test.ts b/apps/backend/src/routes/decision.test.ts index 7a45ba5ce..510c7d8c6 100644 --- a/apps/backend/src/routes/decision.test.ts +++ b/apps/backend/src/routes/decision.test.ts @@ -250,6 +250,15 @@ describe('POST /api/executions/:id/decision - the run', () => { 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) => { @@ -397,6 +406,21 @@ describe('POST /api/executions/:id/decision - the wait instance', () => { 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); @@ -511,6 +535,16 @@ describe('POST /api/executions/:id/decision - delivery', () => { }, ); + 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(() => {}); From a025854a76f48e4e4da7db2aa1030d1e701511f8 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 12:29:48 +0200 Subject: [PATCH 09/17] refactor(backend): one fill for both decision dictionaries The refusal table reuses the domain's {value} filler instead of carrying its own copy. The rerun-source marker in the route states the limitation in words. --- apps/backend/src/domain/decision/decision-issues.ts | 2 +- apps/backend/src/routes/decision-refusals.ts | 4 ++-- apps/backend/src/routes/decision.ts | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/apps/backend/src/domain/decision/decision-issues.ts b/apps/backend/src/domain/decision/decision-issues.ts index 62e357dd2..54fd80db9 100644 --- a/apps/backend/src/domain/decision/decision-issues.ts +++ b/apps/backend/src/domain/decision/decision-issues.ts @@ -37,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/routes/decision-refusals.ts b/apps/backend/src/routes/decision-refusals.ts index 39b2b72dc..4b9066890 100644 --- a/apps/backend/src/routes/decision-refusals.ts +++ b/apps/backend/src/routes/decision-refusals.ts @@ -2,6 +2,7 @@ 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. @@ -73,9 +74,8 @@ export const ENGINE_REFUSALS = { export function refuse(c: Context, refusal: DecisionRefusal, value?: string, extra: Record = {}) { const entry = DECISION_REFUSALS[refusal]; const headers = 'headers' in entry ? entry.headers : undefined; - // A function replacer, so a value containing `$&` or `$1` lands verbatim. return c.json( - { code: entry.code, message: entry.message.replace('{value}', () => value ?? ''), ...extra }, + { code: entry.code, message: fill(entry.message, value), ...extra }, DECISION_REFUSAL_STATUS[entry.code], headers, ); diff --git a/apps/backend/src/routes/decision.ts b/apps/backend/src/routes/decision.ts index e3cfe8cea..ad64de9fb 100644 --- a/apps/backend/src/routes/decision.ts +++ b/apps/backend/src/routes/decision.ts @@ -84,7 +84,7 @@ export function createDecisionRoutes( if (waits === 0) return refuse(c, 'node_never_parked', nodeId); if (waits !== attempt) return refuse(c, 'attempt_mismatch', undefined, { attempt: waits }); - // (follow-up: decision-rerun-source) + // 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', action.name); } From 7d49fb0f18735cfcaa356adca46384e395c30fa9 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 12:29:53 +0200 Subject: [PATCH 10/17] refactor(temporal): name the validator's dictionary apart from the port's code array VERDICT_REJECTIONS meant a private array of codes in execution-core and an exported map of messages in the validator. The validator's is now VERDICT_REJECTION_MESSAGES. --- packages/temporal/src/workflow/verdict-validation.test.ts | 4 ++-- packages/temporal/src/workflow/verdict-validation.ts | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/temporal/src/workflow/verdict-validation.test.ts b/packages/temporal/src/workflow/verdict-validation.test.ts index ccbc41583..61444e202 100644 --- a/packages/temporal/src/workflow/verdict-validation.test.ts +++ b/packages/temporal/src/workflow/verdict-validation.test.ts @@ -3,7 +3,7 @@ import { ApplicationFailure } from '@temporalio/workflow'; import { describe, expect, expectTypeOf, it } from 'vitest'; import type { VerdictRejection } from './core-contract'; -import { type NodeWaitState, VERDICT_REJECTIONS, validateVerdict } from './verdict-validation'; +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' }]]); @@ -20,7 +20,7 @@ function rejection(verdict?: unknown, waits: ReadonlyMap describe('validateVerdict', () => { it('throws exactly the codes the port declares for the validator', () => { - type Thrown = (typeof VERDICT_REJECTIONS)[keyof typeof VERDICT_REJECTIONS]['code']; + type Thrown = (typeof VERDICT_REJECTION_MESSAGES)[keyof typeof VERDICT_REJECTION_MESSAGES]['code']; expectTypeOf().toEqualTypeOf(); }); diff --git a/packages/temporal/src/workflow/verdict-validation.ts b/packages/temporal/src/workflow/verdict-validation.ts index 4cf5b2aee..915dc02e0 100644 --- a/packages/temporal/src/workflow/verdict-validation.ts +++ b/packages/temporal/src/workflow/verdict-validation.ts @@ -7,7 +7,7 @@ export type NodeWaitState = { status: 'waiting' } | { status: 'resolved'; resolu // 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_REJECTIONS = { +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' }, @@ -21,8 +21,8 @@ export const VERDICT_REJECTIONS = { 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_REJECTIONS, value?: string): ApplicationFailure { - const { code, message } = VERDICT_REJECTIONS[key]; +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 ?? ''), From 2bc51bebf4372627ccd000452753bcc4cfda3583 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 12:29:57 +0200 Subject: [PATCH 11/17] test(temporal): pin the default resolveNode deadline the README quotes A fake client with a frozen clock checks that withDeadline receives now + 10 s by default and now + resolveTimeoutMs when set. The entry-points table names RUN_WORKFLOW_NAME and RESOLVE_NODE_UPDATE_NAME. --- packages/temporal/README.md | 10 ++--- .../test/resolve-node-deadline.test.ts | 45 +++++++++++++++++++ 2 files changed, 50 insertions(+), 5 deletions(-) create mode 100644 packages/temporal/test/resolve-node-deadline.test.ts diff --git a/packages/temporal/README.md b/packages/temporal/README.md index cd61d3f32..d6ccde4f5 100644 --- a/packages/temporal/README.md +++ b/packages/temporal/README.md @@ -164,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, delivering verdicts | -| `@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. 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)); + }); +}); From 0c6e34e88b7596ad6ff8d7f67ab34893647cf61f Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Thu, 10 Sep 2026 12:30:01 +0200 Subject: [PATCH 12/17] docs(backend): when a deny hides ids, and the snapshot re-parse gap Deny-before-404 hides which ids exist only if the port denies on absent attributes too. Re-parsing a stored snapshot with today's schema can leave a parked run undecidable after a deploy. --- apps/backend/auth-port.decision-log.md | 2 +- apps/backend/decision-request.decision-log.md | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/backend/auth-port.decision-log.md b/apps/backend/auth-port.decision-log.md index e1e4e2bca..492493910 100644 --- a/apps/backend/auth-port.decision-log.md +++ b/apps/backend/auth-port.decision-log.md @@ -34,7 +34,7 @@ This decision log records the structural piece (scope L from [`local-dev-binding | `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. +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 f73659069..1a18dc4b8 100644 --- a/apps/backend/decision-request.decision-log.md +++ b/apps/backend/decision-request.decision-log.md @@ -59,6 +59,7 @@ What the endpoint does and answers is in the README. Only the reasons are here. - 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, closed 10.09.2026 From a68011c8ecc0246e0c0fc1e2690410bff719025c Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Mon, 14 Sep 2026 12:13:35 +0200 Subject: [PATCH 13/17] fix(backend): deliver a decision to the id the row carries, not the url spelling Postgres accepts a non-canonical uuid and answers with the canonical row, so the spelling in the url and the row's id can differ. The route passed the url string on to the engine, which builds a case-sensitive workflow name from it: a request that found the right row could address a workflow that does not exist and come back as a 409 for a run that is still parked. Submit and cancel already take the row's id; the decision route was the one call site that did not. Every id downstream of the row read is now the row's. --- apps/backend/src/routes/decision.test.ts | 20 ++++++++++++++++++++ apps/backend/src/routes/decision.ts | 24 +++++++++++++++++------- 2 files changed, 37 insertions(+), 7 deletions(-) diff --git a/apps/backend/src/routes/decision.test.ts b/apps/backend/src/routes/decision.test.ts index 510c7d8c6..b855e3dcb 100644 --- a/apps/backend/src/routes/decision.test.ts +++ b/apps/backend/src/routes/decision.test.ts @@ -479,6 +479,26 @@ describe('POST /api/executions/:id/decision - delivery', () => { }); }); + // 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); diff --git a/apps/backend/src/routes/decision.ts b/apps/backend/src/routes/decision.ts index ad64de9fb..60a5267c8 100644 --- a/apps/backend/src/routes/decision.ts +++ b/apps/backend/src/routes/decision.ts @@ -48,7 +48,7 @@ export function createDecisionRoutes( const resource: AuthResource = execution ? { kind: 'execution', - executionId, + executionId: execution.id, attributes: { workflowId: execution.workflowId, tenantId: execution.tenantId, status: execution.status }, } : { kind: 'execution', executionId }; @@ -57,6 +57,10 @@ export function createDecisionRoutes( 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', undefined, { details: formatValidationDetails(parsedBody.error) }); @@ -66,10 +70,10 @@ export function createDecisionRoutes( const parsedSnapshot = z.safeParse(workflowSnapshotSchema, execution.workflowSnapshotJson); if (!parsedSnapshot.success) { logger.error('stored snapshot no longer parses', { - executionId, + executionId: resolvedId, error: { issues: formatValidationDetails(parsedSnapshot.error) }, }); - throw new Error(`stored snapshot of execution ${executionId} no longer parses`); + 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], nodeId); @@ -80,7 +84,7 @@ export function createDecisionRoutes( } const { decision, action } = validated; - const waits = await countNodeWaits(executionId, nodeId); + const waits = await countNodeWaits(resolvedId, nodeId); if (waits === 0) return refuse(c, 'node_never_parked', nodeId); if (waits !== attempt) return refuse(c, 'attempt_mismatch', undefined, { attempt: waits }); @@ -89,7 +93,7 @@ export function createDecisionRoutes( return refuse(c, 'effect_not_supported', action.name); } - const result = await getWorkflowEngine().resolveNode(executionId, nodeId, toNodeResolution(decision, action)); + const result = await getWorkflowEngine().resolveNode(resolvedId, nodeId, toNodeResolution(decision, action)); if (result.error !== undefined) { const outcome = ENGINE_REFUSALS[result.error.code]; if (outcome === 'fault') { @@ -100,8 +104,14 @@ export function createDecisionRoutes( return refuse(c, outcome, nodeId); } - logger.info('decision delivered', { executionId, nodeId, attempt, action: action.name, effect: decision.effect }); - return c.json({ executionId, nodeId, attempt, action: action.name, effect: decision.effect }); + 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; From 57a0e38e038b50f2fdfb2089c7cff5a17aed42a9 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Mon, 14 Sep 2026 12:13:49 +0200 Subject: [PATCH 14/17] fix(backend): a timed-out decision resend names the wait, not the sender The 503 told a caller that a decision_already_made answer to a resend proves their own decision landed. Nothing in the endpoint tells two senders apart, so with two deciders racing one wait the answer can be about the other one's verdict while another parked node keeps the run open. The wording now claims only what first-write-wins can prove. The comment above the RPC deadline said a timed-out update is not durable, which conflates the deadline with acceptance: the update may already have been accepted when the deadline hits. --- apps/backend/README.md | 24 ++++++++++---------- apps/backend/src/routes/decision-refusals.ts | 5 ++-- packages/temporal/src/client/index.ts | 4 ++-- 3 files changed, 17 insertions(+), 16 deletions(-) diff --git a/apps/backend/README.md b/apps/backend/README.md index 1770d1e69..184a09468 100644 --- a/apps/backend/README.md +++ b/apps/backend/README.md @@ -45,18 +45,18 @@ A submitted decision is checked against the request by `validateSubmittedDecisio 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 means it did | +| 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/src/routes/decision-refusals.ts b/apps/backend/src/routes/decision-refusals.ts index 4b9066890..98dd2cb40 100644 --- a/apps/backend/src/routes/decision-refusals.ts +++ b/apps/backend/src/routes/decision-refusals.ts @@ -45,11 +45,12 @@ export const DECISION_REFUSALS = { code: 'effect_not_supported', message: "Action '{value}' re-runs the proposal source, which is not supported yet", }, - // Not durable until accepted, yet the server may still hand it to the next worker. + // 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 it did.', + '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 }>; diff --git a/packages/temporal/src/client/index.ts b/packages/temporal/src/client/index.ts index b90e7fd59..349d5718d 100644 --- a/packages/temporal/src/client/index.ts +++ b/packages/temporal/src/client/index.ts @@ -18,8 +18,8 @@ import type { ResolveNodeUpdateInput, runWorkflow } from '../workflow/run-workfl import { mapResolveNodeError } from './resolve-node-result'; // Without a worker nobody validates an update, and the RPC would wait for the server's -// own limit. A timed-out update is not durable, yet the server may still hand it to the -// next worker: the caller resends and may hear verdict_already_delivered. +// 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 = { From abc8a0fb4f2f58b58342c9fcd8b3f70d3a31be02 Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Mon, 14 Sep 2026 14:22:12 +0200 Subject: [PATCH 15/17] docs: mark the seam where a decision's attempt has to reach the engine The attempt check reads the node's node_waiting count from Postgres and then calls the engine, which is handed only a node id: the wait map is keyed by node id alone, so the check is not atomic with delivery. It holds today only because a node parks at most once per run, an invariant the rerun loop breaks. Nothing in code said so. Two comments now name the hazard at both ends and the decision log carries the slug, so the rerun work starts from a grep rather than from rediscovering the gap. --- apps/backend/decision-request.decision-log.md | 2 +- apps/backend/src/routes/decision.ts | 3 +++ packages/temporal/src/workflow/run-workflow.ts | 2 ++ 3 files changed, 6 insertions(+), 1 deletion(-) diff --git a/apps/backend/decision-request.decision-log.md b/apps/backend/decision-request.decision-log.md index 1a18dc4b8..b52111bc8 100644 --- a/apps/backend/decision-request.decision-log.md +++ b/apps/backend/decision-request.decision-log.md @@ -32,7 +32,7 @@ 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; when the rerun loop re-parks a node, the count follows with no change here. +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. diff --git a/apps/backend/src/routes/decision.ts b/apps/backend/src/routes/decision.ts index 60a5267c8..928b7876d 100644 --- a/apps/backend/src/routes/decision.ts +++ b/apps/backend/src/routes/decision.ts @@ -84,6 +84,9 @@ export function createDecisionRoutes( } 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', nodeId); if (waits !== attempt) return refuse(c, 'attempt_mismatch', undefined, { attempt: waits }); diff --git a/packages/temporal/src/workflow/run-workflow.ts b/packages/temporal/src/workflow/run-workflow.ts index 4cdafea33..59eec3fa5 100644 --- a/packages/temporal/src/workflow/run-workflow.ts +++ b/packages/temporal/src/workflow/run-workflow.ts @@ -56,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)); From caf66e0053fb7c9d41ecaba7e6dff42295206e3d Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Tue, 15 Sep 2026 08:45:04 +0200 Subject: [PATCH 16/17] refactor(backend): refuse takes named options, so no call site reads as a riddle Three of the nine call sites passed a bare `undefined` to skip the message's interpolation value and reach the extra response fields behind it, which told a reader nothing about what either argument was for. Both are now one options object: `{ value }` fills the message's slot, `{ extra }` adds fields beside `code` and `message`. The refusals with neither pass nothing at all, as before. --- apps/backend/src/routes/decision-refusals.test.ts | 2 +- apps/backend/src/routes/decision-refusals.ts | 6 +++++- apps/backend/src/routes/decision.ts | 14 +++++++------- 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/apps/backend/src/routes/decision-refusals.test.ts b/apps/backend/src/routes/decision-refusals.test.ts index 6622004a8..f63212117 100644 --- a/apps/backend/src/routes/decision-refusals.test.ts +++ b/apps/backend/src/routes/decision-refusals.test.ts @@ -4,7 +4,7 @@ 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 app = new Hono().get('/', (c) => refuse(c, refusal, { value, extra })); const response = await app.request('/'); return { status: response.status, diff --git a/apps/backend/src/routes/decision-refusals.ts b/apps/backend/src/routes/decision-refusals.ts index 98dd2cb40..96effcd7b 100644 --- a/apps/backend/src/routes/decision-refusals.ts +++ b/apps/backend/src/routes/decision-refusals.ts @@ -72,7 +72,11 @@ export const ENGINE_REFUSALS = { delivery_timeout: 'delivery_timeout', } as const satisfies Record; -export function refuse(c: Context, refusal: DecisionRefusal, value?: string, extra: 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( diff --git a/apps/backend/src/routes/decision.ts b/apps/backend/src/routes/decision.ts index 928b7876d..893cca46b 100644 --- a/apps/backend/src/routes/decision.ts +++ b/apps/backend/src/routes/decision.ts @@ -63,7 +63,7 @@ export function createDecisionRoutes( const parsedBody = z.safeParse(decisionBodySchema, await c.req.json()); if (!parsedBody.success) { - return refuse(c, 'body_invalid', undefined, { details: formatValidationDetails(parsedBody.error) }); + return refuse(c, 'body_invalid', { extra: { details: formatValidationDetails(parsedBody.error) } }); } const { nodeId, attempt, ...submitted } = parsedBody.data; @@ -76,11 +76,11 @@ export function createDecisionRoutes( 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], 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', undefined, { details: [validated.error] }); + return refuse(c, 'decision_invalid', { extra: { details: [validated.error] } }); } const { decision, action } = validated; @@ -88,12 +88,12 @@ export function createDecisionRoutes( // 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', nodeId); - if (waits !== attempt) return refuse(c, 'attempt_mismatch', undefined, { attempt: waits }); + 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', action.name); + return refuse(c, 'effect_not_supported', { value: action.name }); } const result = await getWorkflowEngine().resolveNode(resolvedId, nodeId, toNodeResolution(decision, action)); @@ -104,7 +104,7 @@ export function createDecisionRoutes( `engine refused the completion for node '${nodeId}': ${result.error.code}: ${result.error.message}`, ); } - return refuse(c, outcome, nodeId); + return refuse(c, outcome, { value: nodeId }); } logger.info('decision delivered', { From c752f77d6e2b4e8ce19245da91da8bafef1d058d Mon Sep 17 00:00:00 2001 From: Piotr Blaszczyk Date: Tue, 15 Sep 2026 08:45:24 +0200 Subject: [PATCH 17/17] refactor(backend): one name for the env every route mounts on `Hono<{ Variables: AuthVariables & TenantVariables }>` was spelled out nine times: twice in each of the four route factories, once in the server, and once in two route test harnesses. The alias names the Hono env rather than the whole app type, because the constructor needs the generic either way. --- apps/backend/src/routes/backend-env.ts | 4 ++++ apps/backend/src/routes/decision.ts | 10 ++++------ apps/backend/src/routes/executions.test.ts | 5 +++-- apps/backend/src/routes/executions.ts | 10 ++++------ apps/backend/src/routes/visualize.ts | 10 ++++------ apps/backend/src/routes/workflows.test.ts | 5 +++-- apps/backend/src/routes/workflows.ts | 10 ++++------ apps/backend/src/server.ts | 14 ++++---------- 8 files changed, 30 insertions(+), 38 deletions(-) create mode 100644 apps/backend/src/routes/backend-env.ts 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.ts b/apps/backend/src/routes/decision.ts index 893cca46b..6352579b5 100644 --- a/apps/backend/src/routes/decision.ts +++ b/apps/backend/src/routes/decision.ts @@ -7,7 +7,7 @@ import { TERMINAL_EXECUTION_STATUSES, } from '@workflow-builder/types/workflow-execution/execution-events'; -import type { AssertAuthorized, AuthResource, AuthVariables } from '../auth'; +import type { AssertAuthorized, AuthResource } from '../auth'; import { database } from '../db/client'; import { executions } from '../db/schema'; import { findDecisionRequest } from '../domain/decision/find-decision-request'; @@ -17,7 +17,7 @@ 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 { TenantVariables } from '../tenant'; +import type { BackendEnv } from './backend-env'; import { ENGINE_REFUSALS, LOOKUP_REFUSALS, refuse } from './decision-refusals'; import { formatValidationDetails } from './snapshot-validation'; @@ -35,10 +35,8 @@ const decisionBodySchema = submittedDecisionSchema.extend({ attempt: z.int().min(1), }); -export function createDecisionRoutes( - assertAuthorized: AssertAuthorized, -): Hono<{ Variables: AuthVariables & TenantVariables }> { - const routes = new Hono<{ Variables: AuthVariables & TenantVariables }>(); +export function createDecisionRoutes(assertAuthorized: AssertAuthorized): Hono { + const routes = new Hono(); routes.post('/:id/decision', async (c) => { const executionId = c.req.param('id'); 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 0ad6afc86..73a21632b 100644 --- a/apps/backend/src/server.ts +++ b/apps/backend/src/server.ts @@ -4,23 +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 @@ -35,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