diff --git a/docs/architecture/execution-model.md b/docs/architecture/execution-model.md index 9fa18054..4a0fd52f 100644 --- a/docs/architecture/execution-model.md +++ b/docs/architecture/execution-model.md @@ -119,15 +119,22 @@ that can reach the run: - VS Code: a sidebar / status-bar prompt and a WebviewPanel card. - CLI: a terminal prompt (`relavium gate`). -When a decision arrives the engine reloads state, emits `human_gate:resumed`, and -the run continues. Because the gate state is checkpointed, resolving it is -idempotent across a reconnect — re-delivering the same decision does not advance -the run twice. **Parallel branches may each reach a gate, so multiple gates can be pending at -once** — each resolves independently with its own timeout (a `run:paused` aggregate reflects that -≥1 gate is pending). A gate may carry a timeout with an `on_timeout` policy (`reject` / -`approve`; `escalate` is **reserved** in v1.0 — authored in YAML as `timeout_action`; see -[workflow-yaml-spec.md](../reference/contracts/workflow-yaml-spec.md#human_gate-node)); this -prevents a forgotten gate from blocking a run forever. +When a decision arrives — `approved`, `rejected`, or `input_provided` — the engine emits +`human_gate:resumed` and the run **continues**: the gate node completes with the decision as its +output, so the author routes on it with a downstream `condition` (a `rejected` decision does not +itself fail the run). Because the gate state is checkpointed, resolving it is idempotent across a +reconnect — re-delivering the same decision does not advance the run twice. **Parallel branches may +each reach a gate, so multiple gates can be pending at once** — each resolves independently with its +own timeout (a `run:paused` aggregate reflects that ≥1 gate is pending). A gate may carry a timeout +with an `on_timeout` policy (`reject` / `approve`; `escalate` is **reserved** in v1.0 — authored in +YAML as `timeout_action`; see +[workflow-yaml-spec.md](../reference/contracts/workflow-yaml-spec.md#human_gate-node)), armed as a +one-shot timer from the injected clock when the gate parks. The two timeout outcomes differ from a +human decision: `approve` **auto-resolves** the gate as approved (`decidedBy: 'timeout'`, the run +continues); `reject` **fails** the run with `run_timeout` (the `AwaitingGate → Failed` edge above) — +this is what stops a forgotten gate from blocking a run forever. A decision that arrives first +disarms the timer. + The gate event/decision shapes are part of the [SSE event schema](../reference/contracts/sse-event-schema.md) and the [IPC contract](../reference/contracts/ipc-contract.md). diff --git a/docs/architecture/shared-core-engine.md b/docs/architecture/shared-core-engine.md index ab994f75..0a529f9d 100644 --- a/docs/architecture/shared-core-engine.md +++ b/docs/architecture/shared-core-engine.md @@ -165,13 +165,47 @@ This is what enables: `runId + nodeId + retryCount`, so a retry never double-applies side effects. In Phase 1 there is **no separate checkpoint table**: the checkpoint is **reconstructed** by a -`Checkpointer` (`load(runId) → CheckpointState`) from the per-node `step_executions` rows -(`status` / `attempt_number` / `output_json` / `error_json`) and the ordered, replayable `run_events` -log, with the orchestrator's message history in `messages` (schema in -[../reference/desktop/database-schema.md](../reference/desktop/database-schema.md)). -`CheckpointState = { runStatus, nodeStates, completedNodeIds, pendingNodeIds, orchestratorMessages? }` -is **derived**, never a stored blob. The same derivation is what the Phase-2 cloud layer uses for -durable execution — see [cloud-phase-2.md](cloud-phase-2.md). +`Checkpointer` (`load(runId) → CheckpointState`) by folding the ordered, replayable `run_events` log +alone — each node's output/error rides its `node:completed` / `node:failed` event, so the stream is a +sufficient source. The persistence layer *also* denormalizes per-node state into `step_executions` and an +orchestrator's history into `messages` (schema in +[../reference/desktop/database-schema.md](../reference/desktop/database-schema.md)) for the run-trace UI +and fast querying — the same per-node truth, not an extra input the fold requires. +`CheckpointState` is **derived**, never a stored blob: a pure fold over the ordered event stream +(`reconstructCheckpointState(events)`) captures run status, the surrogate `workflowId`, per-node +settled/paused states (with a `condition`'s selected branch from `node:completed.selected` and dimmed +branches from `node:skipped`), pending and already-resolved gate ids, the last `sequenceNumber`, and the +running token/cost tallies. The exact field set is the `CheckpointState` interface in +[`packages/core/src/engine/checkpoint.ts`](../../packages/core/src/engine/checkpoint.ts) — the one +authoritative shape; this section does not restate it. The same derivation is what the Phase-2 cloud +layer uses for durable execution — see [cloud-phase-2.md](cloud-phase-2.md). + +**Reconstruction is total and deterministic** (same events → same state — the basis of idempotent +resume). A node that emitted `node:started` but no terminal event (it was running when the process +died) is simply **absent** from `nodeStates`, so the rehydrating engine seeds it `pending` and re-runs +it — bounded by the `runId + nodeId + retryCount` idempotency key, never by silently skipping it. What is +**not** in the checkpoint: the eager-once resolved `context` (`ctx.*`) is **re-resolved at run start**, +not reconstructed — and if a later change makes it part of a transported checkpoint it MUST cross that +boundary via `structuredClone`, never `JSON.stringify`→`parse` (which would re-materialise a `__proto__` +key as a real setter; the standing note lives at +[`interpolation/resolve.ts`](../../packages/core/src/interpolation/resolve.ts)). + +A run suspended at a gate resumes in **two ways**: in the same process, `engine.resume(runId, gateId, +decision)`; across a restart, `engine.resumeFromCheckpoint({ runId, workflow, gateId, decision })` +rehydrates a fresh `RunExecution` from the reconstructed state (seeding node states, pending gates, +tallies, and the `sequenceNumber` so post-resume events continue gap-free — no `run:started` is +re-emitted) and returns a `RunHandle` for the rest of the run. An **identity guard** refuses a resume +whose workflow is not the one the run started on: the Phase-1 in-memory reference compares the surrogate +`workflowId` reconstructed from `run:started` (a different workflow → a typed `workflow_mismatch`). The +stronger guard that also catches a *same-slug, edited-content* workflow rides on the frozen +`runs.workflow_definition_snapshot` column ([../reference/desktop/database-schema.md](../reference/desktop/database-schema.md)) +— a Phase-2 persistence concern wired with the real `RunStore`, not the event-derived in-memory state. **Idempotent re-delivery** never advances a run twice: re-delivering a decision to an +already-terminal run is a no-op (a closed handle, nothing re-emitted or re-persisted); re-delivering an +already-resolved gate on a still-running run drives the remaining work without re-applying the decision. +This holds within a process, and across processes once the prior process's `human_gate:resumed` is +persisted; the residual concurrent window (two processes loading the *same* still-pending gate before +either persists) is closed by a Phase-2 store-level uniqueness constraint on `human_gate:resumed` per +gate, not by the in-memory reference. ## Retry and fallback diff --git a/docs/reference/contracts/sse-event-schema.md b/docs/reference/contracts/sse-event-schema.md index a280aade..96831ed8 100644 --- a/docs/reference/contracts/sse-event-schema.md +++ b/docs/reference/contracts/sse-event-schema.md @@ -50,6 +50,7 @@ export type RunEvent = | CostUpdatedEvent | NodeCompletedEvent | NodeFailedEvent + | NodeSkippedEvent | HumanGatePausedEvent | HumanGateResumedEvent | RunCompletedEvent @@ -72,9 +73,10 @@ export type RunEvent = | `agent:tool_result` | A tool returned. | `nodeId`, `toolId`, `success`, `outputSummary` (truncated for UI), `attemptNumber?` | | `agent:file_patch_proposed` | An agent proposed a file change (**gated — no write until the user accepts**; e.g. the VS Code inline-diff review). | `nodeId`, `patches: [{ uri, unifiedDiff }]` (≥1 — an empty proposal is meaningless), `attemptNumber?` | | `cost:updated` | A node's token cost was tallied (drives the cost waterfall). | `nodeId`, `model`, `inputTokens`, `outputTokens`, `costMicrocents`, `cumulativeCostMicrocents` (integer micro-cents — canonical unit in [llm-provider-seam.md](../shared-core/llm-provider-seam.md#6-usage)), `attemptNumber?` (1-based retry attempt this cost belongs to, so per-attempt cost is reconstructable) | -| `node:completed` | A node finished successfully. | `nodeId`, `output`, `tokensUsed: {input, output, model?}` (`model` only for LLM nodes), `durationMs`, `attemptNumber?` | +| `node:completed` | A node finished successfully. | `nodeId`, `output`, `tokensUsed: {input, output, model?}` (`model` only for LLM nodes), `durationMs`, `selected?` (a `condition`'s chosen target ids — the authoritative branch record checkpoint/resume restores from, 1.R; **may be an empty array** when the condition routes to no branch, dimming all downstream), `attemptNumber?` | | `node:failed` | A node failed. | `nodeId`, `error: {code, message, retryable, correlationId?}` (`code` is an [`ErrorCode`](#error-code-taxonomy); `correlationId` is a secret-free id joined to the internal log — ADR-0036) | -| `human_gate:paused` | Execution suspended at a human gate. | `nodeId`, `gateId`, `gateType: 'approval' \| 'input' \| 'review'`, `message`, `assignee?`, `timeoutMs?`, `expiresAt?` | +| `node:skipped` | A node was skip-propagated (never ran). | `nodeId`, `reason: 'branch_not_taken' \| 'upstream_unreachable'` (`branch_not_taken` = a `condition` routed away from it; `upstream_unreachable` = every in-edge is dead because an upstream was skipped/failed). Emitted so the event log is a **complete, replayable** record — checkpoint/resume reconstructs a skipped vertex from it ([run-plan.md](../shared-core/run-plan.md)) and a surface can render the dimmed path instead of the node silently vanishing. | +| `human_gate:paused` | Execution suspended at a human gate. | `nodeId`, `gateId`, `gateType: 'approval' \| 'input' \| 'review'`, `message`, `assignee?`, `timeoutMs?`, `timeoutAction?: 'approve' \| 'reject'` (on-timeout policy, present only with `timeoutMs`), `expiresAt?` | | `human_gate:resumed` | A gate decision was applied; execution continues. | `nodeId`, `decision: 'approved' \| 'rejected' \| 'input_provided'`, `decidedBy`, `payload?` | | `run:paused` | The run is suspended with **≥1 gate pending** — the multi-gate aggregate that backs the pending-gate queue (parallel branches may each reach a gate). | `pendingGateCount`, `gateIds[]` | | `run:completed` | The run finished. | `outputs` (a record **keyed by each terminal `output` vertex's node id**, the value being that vertex's captured output — see [run-plan.md §output capture](../shared-core/run-plan.md)), `totalTokensUsed`, `totalCostMicrocents` (integer micro-cents closing total for the whole run), `durationMs` | @@ -117,9 +119,16 @@ export interface NodeCompletedEvent extends BaseEvent { // no model — so `model` is optional. tokensUsed: { input: number; output: number; model?: string }; durationMs: number; + selected?: string[]; // a `condition` node only: the immediate target ids it routed to (the live branches); MAY be empty when it routes to no branch (all downstream skip-propagated). The authoritative record checkpoint/resume restores `selectedTargets` from (1.R). attemptNumber?: number; // 1-based retry attempt this completion belongs to (matches cost:updated) } +export interface NodeSkippedEvent extends BaseEvent { + type: 'node:skipped'; + nodeId: string; + reason: 'branch_not_taken' | 'upstream_unreachable'; +} + export interface HumanGatePausedEvent extends BaseEvent { type: 'human_gate:paused'; nodeId: string; @@ -128,6 +137,7 @@ export interface HumanGatePausedEvent extends BaseEvent { message: string; assignee?: string; timeoutMs?: number; + timeoutAction?: 'approve' | 'reject'; // on-timeout policy (present only with timeoutMs); lets a surface show how the gate auto-resolves and a Phase-2 crash-resume re-arm the timer from the log expiresAt?: string; } ``` diff --git a/packages/core/src/engine/checkpoint.test.ts b/packages/core/src/engine/checkpoint.test.ts new file mode 100644 index 00000000..90072daf --- /dev/null +++ b/packages/core/src/engine/checkpoint.test.ts @@ -0,0 +1,240 @@ +import type { RunEvent } from '@relavium/shared'; +import { describe, expect, it } from 'vitest'; + +import { reconstructCheckpointState } from './checkpoint.js'; +import { InMemoryRunStore, createInMemoryCheckpointer } from './execution-host.js'; + +const TS = '2026-01-01T00:00:00.000Z'; +const base = (sequenceNumber: number) => ({ runId: 'r1', sequenceNumber, timestamp: TS }); + +const started: RunEvent = { + type: 'run:started', + ...base(0), + workflowId: '00000000-0000-4000-8000-000000000001', + inputs: {}, + executionMode: 'local', +}; +const completed = (seq: number, nodeId: string, output: unknown): RunEvent => ({ + type: 'node:completed', + ...base(seq), + nodeId, + output, + tokensUsed: { input: 0, output: 0 }, + durationMs: 1, +}); + +describe('reconstructCheckpointState', () => { + it('returns undefined for a run with no run:started', () => { + expect(reconstructCheckpointState([completed(1, 'a', 1)])).toBeUndefined(); + }); + + it('reconstructs a completed run (status + nodeStates + lastSequenceNumber)', () => { + const state = reconstructCheckpointState([ + started, + completed(1, 'a', { v: 1 }), + { + type: 'run:completed', + ...base(2), + outputs: {}, + totalTokensUsed: { input: 0, output: 0 }, + totalCostMicrocents: 0, + durationMs: 1, + }, + ]); + expect(state?.runStatus).toBe('completed'); + expect(state?.workflowId).toBe('00000000-0000-4000-8000-000000000001'); // captured from run:started + expect(state?.startedAtMs).toBe(Date.parse(TS)); // original start epoch, so resumed durationMs is total + expect(state?.nodeStates.get('a')).toEqual({ status: 'completed', output: { v: 1 } }); + expect(state?.completedNodeIds).toEqual(['a']); + expect(state?.lastSequenceNumber).toBe(2); + }); + + it('OMITS a node that started but never finished — so the rehydrating engine re-runs it (trap b)', () => { + const state = reconstructCheckpointState([ + started, + completed(1, 'a', 'A'), + { type: 'node:started', ...base(2), nodeId: 'b', nodeType: 'agent' }, // crashed mid-flight + ]); + expect(state?.runStatus).toBe('running'); + expect(state?.nodeStates.has('a')).toBe(true); + expect(state?.nodeStates.has('b')).toBe(false); // absent → engine seeds 'pending' → re-runs + }); + + it('restores a condition selectedTargets + the dimmed branch as skipped (resume routes correctly)', () => { + const state = reconstructCheckpointState([ + started, + { + type: 'node:completed', + ...base(1), + nodeId: 'gate', + output: { decision: true }, + tokensUsed: { input: 0, output: 0 }, + durationMs: 1, + selected: ['hi'], + }, + { type: 'node:skipped', ...base(2), nodeId: 'lo', reason: 'branch_not_taken' }, + ]); + expect(state?.nodeStates.get('gate')).toEqual({ + status: 'completed', + output: { decision: true }, + selectedTargets: ['hi'], + }); + expect(state?.nodeStates.get('lo')).toEqual({ status: 'skipped' }); + }); + + it('reconstructs a gate-parked run (paused status + pendingGates + paused node)', () => { + const state = reconstructCheckpointState([ + started, + { + type: 'human_gate:paused', + ...base(1), + nodeId: 'gate', + gateId: 'g1', + gateType: 'approval', + message: 'ok?', + }, + { type: 'run:paused', ...base(2), pendingGateCount: 1, gateIds: ['g1'] }, + ]); + expect(state?.runStatus).toBe('paused'); + expect(state?.nodeStates.get('gate')).toEqual({ status: 'paused' }); + expect(state?.pendingGates).toEqual([{ gateId: 'g1', nodeId: 'gate' }]); + }); + + it('a resumed gate clears the pending gate + records the decision as the node output', () => { + const state = reconstructCheckpointState([ + started, + { + type: 'human_gate:paused', + ...base(1), + nodeId: 'gate', + gateId: 'g1', + gateType: 'approval', + message: 'ok?', + }, + { + type: 'human_gate:resumed', + ...base(2), + nodeId: 'gate', + decision: 'approved', + decidedBy: 'u1', + }, + ]); + expect(state?.pendingGates).toEqual([]); + expect(state?.resolvedGateIds).toContain('g1'); // moved to resolved → idempotent re-delivery is a no-op + expect(state?.nodeStates.get('gate')).toEqual({ + status: 'completed', + output: { decision: 'approved' }, + }); + }); + + it('a resumed gate with a payload records the payload as the output', () => { + const state = reconstructCheckpointState([ + started, + { + type: 'human_gate:paused', + ...base(1), + nodeId: 'gate', + gateId: 'g1', + gateType: 'input', + message: 'value?', + }, + { + type: 'human_gate:resumed', + ...base(2), + nodeId: 'gate', + decision: 'input_provided', + decidedBy: 'u1', + payload: { x: 7 }, + }, + ]); + expect(state?.nodeStates.get('gate')).toEqual({ status: 'completed', output: { x: 7 } }); + }); + + it('restores running token + cost tallies so a resumed run keeps cumulative totals', () => { + const state = reconstructCheckpointState([ + started, + { + type: 'node:completed', + ...base(1), + nodeId: 'a', + output: 'A', + tokensUsed: { input: 10, output: 5 }, + durationMs: 1, + }, + { + type: 'cost:updated', + ...base(2), + nodeId: 'a', + model: 'm', + inputTokens: 10, + outputTokens: 5, + costMicrocents: 700, + cumulativeCostMicrocents: 700, + }, + { + type: 'node:completed', + ...base(3), + nodeId: 'b', + output: 'B', + tokensUsed: { input: 20, output: 8 }, + durationMs: 1, + }, + { + type: 'cost:updated', + ...base(4), + nodeId: 'b', + model: 'm', + inputTokens: 20, + outputTokens: 8, + costMicrocents: 900, + cumulativeCostMicrocents: 1600, + }, + ]); + expect(state?.totalInputTokens).toBe(30); + expect(state?.totalOutputTokens).toBe(13); + expect(state?.cumulativeCostMicrocents).toBe(1600); // the last running total, not a re-sum + }); + + it('records a failed node with its typed failure', () => { + const state = reconstructCheckpointState([ + started, + { + type: 'node:failed', + ...base(1), + nodeId: 'a', + error: { code: 'tool_failed', message: 'boom', retryable: false }, + }, + ]); + expect(state?.nodeStates.get('a')).toEqual({ + status: 'failed', + error: { code: 'tool_failed', message: 'boom', retryable: false }, + }); + }); +}); + +describe('createInMemoryCheckpointer', () => { + it('loads reconstructed state from an InMemoryRunStore event log', async () => { + const store = new InMemoryRunStore(); + await store.persistEvent(started); + await store.persistEvent(completed(1, 'a', 'A')); + const cp = createInMemoryCheckpointer(store); + const state = await cp.load('r1'); + expect(state?.runStatus).toBe('running'); + expect(state?.nodeStates.get('a')).toEqual({ status: 'completed', output: 'A' }); + }); + + it('returns undefined for an unknown run', async () => { + const cp = createInMemoryCheckpointer(new InMemoryRunStore()); + expect(await cp.load('nope')).toBeUndefined(); + }); + + it('returns undefined for an opaque (non-in-memory) store — a custom store supplies its own', async () => { + const opaque = { + resolveWorkflowId: () => Promise.resolve('x'), + persistEvent: () => Promise.resolve(), + listInterruptedRuns: () => Promise.resolve([]), + }; + const cp = createInMemoryCheckpointer(opaque); + expect(await cp.load('r1')).toBeUndefined(); + }); +}); diff --git a/packages/core/src/engine/checkpoint.ts b/packages/core/src/engine/checkpoint.ts new file mode 100644 index 00000000..a21b0df6 --- /dev/null +++ b/packages/core/src/engine/checkpoint.ts @@ -0,0 +1,223 @@ +/** + * Checkpoint/resume (1.R) — the read-side that reconstructs a run's state from its persisted event + * stream so a run interrupted (a crash, or suspended at a human gate) can resume without re-running the + * work already done. There is **no checkpoint table** — the {@link CheckpointState} is *derived* from the + * ordered `run_events` the {@link RunStore} already persists (ADR-0003; execution-model.md §5). The real + * SQLite/cloud-backed `Checkpointer` is Phase-2/CLI; 1.R ships the in-memory reference + * ({@link createInMemoryHost}). + * + * Reconstruction is a **pure replay**: walk the events in order and fold each into per-node state. + * Crucially, a node that emitted `node:started` but no terminal event (it was running when the process + * died) is simply ABSENT from {@link CheckpointState.nodeStates} — so the rehydrating engine seeds it + * `pending` and re-runs it (a half-run side effect is bounded by the `runId+nodeId+retryCount` + * idempotency key, not by skipping the node). A `condition`'s `selected` branch is restored from + * `node:completed.selected` so a selected branch mid-flight at the crash re-runs rather than being + * wrongly skip-propagated; the dimmed branches are restored from `node:skipped`. + */ + +import type { RunEvent, RunStatus } from '@relavium/shared'; + +import type { NodeFailure } from './node-executor.js'; + +/** The schema version of the *derivation* (not a stored blob) — lets a later engine refuse/migrate it. */ +export const CHECKPOINT_SCHEMA_VERSION = 1; + +/** The reconstructed terminal-or-paused state of one vertex (a still-running vertex is omitted — re-run). */ +export interface CheckpointNodeState { + readonly status: 'completed' | 'failed' | 'skipped' | 'paused'; + /** The node output, for a `completed` vertex (incl. a resumed gate's decision payload). */ + readonly output?: unknown; + /** The failure, for a `failed` vertex. */ + readonly error?: NodeFailure; + /** A `completed` `condition`'s selected immediate target ids — restores `selectedTargets` on resume. */ + readonly selectedTargets?: readonly string[]; +} + +/** A gate still awaiting a decision at the checkpoint — the run resumes by applying a `GateDecision`. */ +export interface CheckpointPendingGate { + readonly gateId: string; + readonly nodeId: string; +} + +/** The derived state a rehydrating run is rebuilt from — never a persisted blob (reconstructed from rows). */ +export interface CheckpointState { + readonly schemaVersion: number; + readonly runStatus: RunStatus; + /** The surrogate `workflows.id` UUID from `run:started` — resume refuses a different workflow (identity guard). */ + readonly workflowId: string; + /** `run:started.timestamp` as epoch ms — the resumed run keeps measuring `durationMs` from the ORIGINAL + * start, so a terminal event reports total wall-clock across the pre- and post-resume segments. */ + readonly startedAtMs: number; + /** Per-vertex settled/paused state; a vertex absent here is `pending` (never started, or running at crash). */ + readonly nodeStates: ReadonlyMap; + /** Convenience projection of the `completed` vertices (the engine derives `pending` from the plan). */ + readonly completedNodeIds: readonly string[]; + /** Gates still pending a decision — the run is resumable via `engine.resume(runId, gateId, decision)`. */ + readonly pendingGates: readonly CheckpointPendingGate[]; + /** Gate ids ALREADY resolved (a `human_gate:resumed` was persisted) — so re-delivering a decision after a + * reconnect is an idempotent no-op rather than advancing the run twice (execution-model.md §gate). */ + readonly resolvedGateIds: readonly string[]; + /** The highest persisted `sequenceNumber` — the resumed run seeds its counter to this + 1 (gap-free). */ + readonly lastSequenceNumber: number; + /** Running token totals (summed from `node:completed`), restored so a resumed run's `run:completed` totals stay correct. */ + readonly totalInputTokens: number; + readonly totalOutputTokens: number; + /** The last `cost:updated.cumulativeCostMicrocents` (a running total), restored so post-resume cost stays cumulative. */ + readonly cumulativeCostMicrocents: number; +} + +/** + * The read port that reconstructs a run's {@link CheckpointState} from persisted rows. Returns + * `undefined` for a run with no `run:started` (unknown / never-persisted). 1.N's {@link RunStore} is + * write+enumerate only; this is the 1.R read side, kept a separate port (single responsibility). + */ +export interface Checkpointer { + load: (runId: string) => Promise; +} + +/** The mutable fold accumulator, threaded through the per-category appliers below. */ +interface ReconAccumulator { + started: boolean; + workflowId: string; + startedAtMs: number; + runStatus: RunStatus; + lastSequenceNumber: number; + totalInputTokens: number; + totalOutputTokens: number; + cumulativeCostMicrocents: number; + readonly nodeStates: Map; + readonly pendingGates: Map; // gateId → nodeId + readonly resolvedGateIds: Set; +} + +const RUN_STATUS_BY_EVENT: Partial> = { + 'run:paused': 'paused', + 'run:completed': 'completed', + 'run:failed': 'failed', + 'run:cancelled': 'cancelled', +}; + +/** Run-level lifecycle: capture start identity/clock and fold the run status. */ +function applyRunEvent(acc: ReconAccumulator, event: RunEvent): void { + if (event.type === 'run:started') { + acc.started = true; + acc.workflowId = event.workflowId; + acc.startedAtMs = Date.parse(event.timestamp); + acc.runStatus = 'running'; + return; + } + const status = RUN_STATUS_BY_EVENT[event.type]; + if (status !== undefined) { + acc.runStatus = status; + } +} + +/** Node-level settlements: completed (+ branch selection, token tally), failed, skipped. */ +function applyNodeEvent(acc: ReconAccumulator, event: RunEvent): void { + switch (event.type) { + case 'node:completed': + acc.nodeStates.set(event.nodeId, { + status: 'completed', + output: event.output, + ...(event.selected === undefined ? {} : { selectedTargets: event.selected }), + }); + acc.totalInputTokens += event.tokensUsed.input; + acc.totalOutputTokens += event.tokensUsed.output; + break; + case 'node:failed': + acc.nodeStates.set(event.nodeId, { + status: 'failed', + error: { + code: event.error.code, + message: event.error.message, + retryable: event.error.retryable, + }, + }); + break; + case 'node:skipped': + acc.nodeStates.set(event.nodeId, { status: 'skipped' }); + break; + default: + break; // node:started has no terminal yet → omitted so the rehydrating engine re-runs it + } +} + +/** Human-gate lifecycle: park a pending gate, or resolve it (decision becomes the gate vertex output). */ +function applyGateEvent(acc: ReconAccumulator, event: RunEvent): void { + if (event.type === 'human_gate:paused') { + acc.nodeStates.set(event.nodeId, { status: 'paused' }); + acc.pendingGates.set(event.gateId, event.nodeId); + return; + } + if (event.type !== 'human_gate:resumed') { + return; + } + // The decision IS the gate vertex's output (engine resume: output = payload ?? { decision }). + acc.nodeStates.set(event.nodeId, { + status: 'completed', + output: event.payload === undefined ? { decision: event.decision } : event.payload, + }); + // Collect this gate's pending ids first, then mutate — never delete while iterating the Map. + const resolvedForNode = [...acc.pendingGates] + .filter(([, nodeId]) => nodeId === event.nodeId) + .map(([gateId]) => gateId); + for (const gateId of resolvedForNode) { + acc.pendingGates.delete(gateId); + acc.resolvedGateIds.add(gateId); + } +} + +/** + * Pure reconstruction: fold the ordered event stream into a {@link CheckpointState}. Total + deterministic + * (same events → same state — the basis of idempotent resume). The caller passes events in persisted + * (sequence) order; this does not re-sort (the store/bus already guarantee order). The per-category + * appliers ({@link applyRunEvent} / {@link applyNodeEvent} / {@link applyGateEvent}) keep this fold flat. + */ +export function reconstructCheckpointState( + events: readonly RunEvent[], +): CheckpointState | undefined { + const acc: ReconAccumulator = { + started: false, + workflowId: '', + startedAtMs: 0, + runStatus: 'running', + lastSequenceNumber: -1, + totalInputTokens: 0, + totalOutputTokens: 0, + cumulativeCostMicrocents: 0, + nodeStates: new Map(), + pendingGates: new Map(), + resolvedGateIds: new Set(), + }; + + for (const event of events) { + acc.lastSequenceNumber = Math.max(acc.lastSequenceNumber, event.sequenceNumber); + if (event.type === 'cost:updated') { + acc.cumulativeCostMicrocents = event.cumulativeCostMicrocents; // already a running total + } + applyRunEvent(acc, event); + applyNodeEvent(acc, event); + applyGateEvent(acc, event); + } + + if (!acc.started) { + return undefined; + } + const completedNodeIds = [...acc.nodeStates] + .filter(([, s]) => s.status === 'completed') + .map(([id]) => id); + return { + schemaVersion: CHECKPOINT_SCHEMA_VERSION, + runStatus: acc.runStatus, + workflowId: acc.workflowId, + startedAtMs: acc.startedAtMs, + nodeStates: acc.nodeStates, + completedNodeIds, + pendingGates: [...acc.pendingGates].map(([gateId, nodeId]) => ({ gateId, nodeId })), + resolvedGateIds: [...acc.resolvedGateIds], + lastSequenceNumber: acc.lastSequenceNumber, + totalInputTokens: acc.totalInputTokens, + totalOutputTokens: acc.totalOutputTokens, + cumulativeCostMicrocents: acc.cumulativeCostMicrocents, + }; +} diff --git a/packages/core/src/engine/engine.test.ts b/packages/core/src/engine/engine.test.ts index a54f8b56..493c472c 100644 --- a/packages/core/src/engine/engine.test.ts +++ b/packages/core/src/engine/engine.test.ts @@ -534,6 +534,538 @@ describe('WorkflowEngine — human gate suspend/resume', () => { expect(caught.code).toBe('unknown_gate'); } }); + + // --- gate timeouts (1.Q): one-shot timer → auto-resolve / run-fail ------------------------- + const gate = (over: Record): NodeOutcome => ({ + kind: 'paused', + gate: { gateType: 'approval', message: 'approve?', ...over }, + }); + + it('emits timeoutMs + expiresAt on human_gate:paused and auto-approves on timeout (decidedBy timeout)', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 1000, timeoutAction: 'approve' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused') { + host.fireTimers(); // the deadline elapsed with no human decision + } + } + const paused = events.find((e) => e.type === 'human_gate:paused'); + if (paused?.type !== 'human_gate:paused') { + throw new Error('expected human_gate:paused'); + } + expect(paused.timeoutMs).toBe(1000); + expect(typeof paused.expiresAt).toBe('string'); + const resumed = events.find((e) => e.type === 'human_gate:resumed'); + if (resumed?.type !== 'human_gate:resumed') { + throw new Error('expected human_gate:resumed'); + } + expect(resumed.decision).toBe('approved'); + expect(resumed.decidedBy).toBe('timeout'); + expect(terminalsIn(events)[0]?.type).toBe('run:completed'); + assertGapFreeSeq(events); + }); + + it('fails the run with run_timeout when a gate times out under timeout_action: reject', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 1000, timeoutAction: 'reject' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused') { + host.fireTimers(); + } + } + expect(events.some((e) => e.type === 'node:failed' && e.nodeId === 'g')).toBe(true); + const terminal = terminalsIn(events)[0]; + expect(terminal?.type).toBe('run:failed'); + if (terminal?.type === 'run:failed') { + expect(terminal.error.code).toBe('run_timeout'); + } + expect(events.some((e) => e.type === 'human_gate:resumed')).toBe(false); // reject-timeout never "resumes" + assertGapFreeSeq(events); + }); + + it('disarms the gate timer when a human decision arrives first (no timeout fires, single resolution)', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 1000, timeoutAction: 'reject' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused') { + const gateId = event.gateIds[0]; + if (gateId !== undefined) { + await engine.resume(handle.runId, gateId, { decision: 'approved', decidedBy: 'human' }); + } + expect(host.armedCount()).toBe(0); // resume disarmed the timer + host.fireTimers(); // a no-op now — the timer is gone + } + } + const resumes = events.filter((e) => e.type === 'human_gate:resumed'); + expect(resumes).toHaveLength(1); + if (resumes[0]?.type === 'human_gate:resumed') { + expect(resumes[0].decidedBy).toBe('human'); + } + expect(terminalsIn(events)[0]?.type).toBe('run:completed'); + }); + + it('arms no timer for a gate without timeout_ms', async () => { + const host = createInMemoryHost(); + const engine = engineWith({ g: () => gate({}) }, host); + const handle = engine.start({ workflow: workflow(GATED) }); + for await (const event of handle.events) { + if (event.type === 'run:paused') { + expect(host.armedCount()).toBe(0); + const gateId = event.gateIds[0]; + if (gateId !== undefined) { + await engine.resume(handle.runId, gateId, { decision: 'approved', decidedBy: 'h' }); + } + } + } + }); + + it('a human rejected decision completes the gate (carrying the decision) and continues the run', async () => { + const engine = engineWith({ + g: () => gate({}), + // Echo the gate's settled output so the test can observe the decision reached run.outputs (the real + // output handler captures its feeder verbatim; the stub otherwise returns its own id). + out: (ctx): NodeOutcome => ({ kind: 'completed', output: ctx.runOutputs.get('g') }), + }); + const handle = engine.start({ workflow: workflow(GATED) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused') { + const gateId = event.gateIds[0]; + if (gateId !== undefined) { + await engine.resume(handle.runId, gateId, { decision: 'rejected', decidedBy: 'human' }); + } + } + } + const resumed = events.find((e) => e.type === 'human_gate:resumed'); + expect(resumed?.type === 'human_gate:resumed' ? resumed.decision : undefined).toBe('rejected'); + // A rejected decision is NOT a run failure (execution-model.md §4): the gate vertex completes carrying + // {decision:'rejected'} as its output (signalled by human_gate:resumed, not a node:completed), the run + // continues, and the value flows downstream — `out` captures its single feeder (the gate) verbatim, so + // a downstream condition could route on it. + const outDone = events.find((e) => e.type === 'node:completed' && e.nodeId === 'out'); + expect(outDone?.type === 'node:completed' ? outDone.output : undefined).toEqual({ + decision: 'rejected', + }); + expect(terminalsIn(events)[0]?.type).toBe('run:completed'); + }); + + it('disarms an armed gate timer when the run terminates for an unrelated reason (cancel)', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 1000, timeoutAction: 'reject' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused') { + expect(host.armedCount()).toBe(1); // the gate timer is armed + engine.cancel(handle.runId); // cancel for an unrelated reason while the timer is still armed + } + } + expect(terminalsIn(events)[0]?.type).toBe('run:cancelled'); + expect(host.armedCount()).toBe(0); // #settle disarmed the armed timer on terminal close + host.fireTimers(); // a no-op now — nothing armed; must not emit anything after the terminal + expect(terminalsIn(events)).toHaveLength(1); + }); + + it('a reject-timeout marks the gate resolved, so a late re-delivery of its decision is a no-op (not a throw)', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 1000, timeoutAction: 'reject' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + let gateId = ''; + let lateResume: unknown = 'not-attempted'; + for await (const event of handle.events) { + if (event.type === 'run:paused') { + gateId = event.gateIds[0] ?? ''; + host.fireTimers(); // reject-timeout → run fails with run_timeout + } + if (event.type === 'run:failed') { + // A duplicate decision arriving after the timeout already failed the run is a silent no-op. + lateResume = await engine + .resume(handle.runId, gateId, { decision: 'rejected', decidedBy: 'late' }) + .then(() => 'no-op') + .catch((e: unknown) => e); + } + } + expect(lateResume).toBe('no-op'); // #resolvedGates was set on the reject-timeout path + }); + + it('emits node:skipped(out) before run:failed when a reject-timeout dims the downstream', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 1000, timeoutAction: 'reject' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused') { + host.fireTimers(); + } + } + const skipIdx = events.findIndex((e) => e.type === 'node:skipped' && e.nodeId === 'out'); + const failIdx = events.findIndex((e) => e.type === 'run:failed'); + expect(skipIdx).toBeGreaterThanOrEqual(0); // the downstream `out` is dimmed (upstream unreachable) + expect(skipIdx).toBeLessThan(failIdx); // …and recorded before the terminal, keeping the log complete + assertGapFreeSeq(events); + }); + + it('expiresAt equals the pause timestamp plus timeoutMs (a real ISO deadline, not just any string)', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 5000, timeoutAction: 'approve' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + let paused: Extract | undefined; + for await (const event of handle.events) { + if (event.type === 'human_gate:paused') { + paused = event; + } + if (event.type === 'run:paused') { + host.fireTimers(); + } + } + if (paused === undefined || paused.expiresAt === undefined) { + throw new Error('expected human_gate:paused with expiresAt'); + } + // expiresAt is a real ISO deadline ≈ the pause time + timeoutMs. The in-memory clock advances 1ms + // per read, so expiresAt (one clock read) and the event timestamp (a later read) differ by the small + // read skew, not exactly 0 — assert the gap is timeoutMs within that few-ms tolerance. + const deltaMs = Date.parse(paused.expiresAt) - Date.parse(paused.timestamp); + expect(deltaMs).toBeGreaterThan(4990); + expect(deltaMs).toBeLessThanOrEqual(5000); + }); + + it('a timer that fires after the run already terminated is an inert no-op (no second terminal)', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { g: () => gate({ timeoutMs: 1000, timeoutAction: 'approve' }) }, + host, + ); + const handle = engine.start({ workflow: workflow(GATED) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused') { + // Resolve by hand so the run completes; the armed timer is disarmed on resume + on settle. + const gateId = event.gateIds[0]; + if (gateId !== undefined) { + await engine.resume(handle.runId, gateId, { decision: 'approved', decidedBy: 'h' }); + } + } + } + host.fireTimers(); // post-terminal: nothing armed; must not emit a second terminal + expect(terminalsIn(events)).toHaveLength(1); + expect(terminalsIn(events)[0]?.type).toBe('run:completed'); + }); + + // Regression for the multi-gate stall race: two timeout-approve gates resolved back-to-back by one + // fireTimers() sweep. The second gate's resume schedules a #step while the first's durable persist is + // still in flight; only because each resume marks its vertex completed SYNCHRONOUSLY (before its await) + // does that step see both gates settled rather than mis-reading the run as stalled (a spurious + // run:failed{internal}). + const MULTIGATE = ` id: multigate + nodes: + - { id: start, type: input } + - { id: fan, type: parallel, parallel_of: [g1, g2] } + - { id: g1, type: human_gate, gate_type: approval } + - { id: g2, type: human_gate, gate_type: approval } + - { id: join, type: merge, merge_strategy: concat } + - { id: out, type: output } + edges: + - { from: start, to: fan } + - { from: g1, to: join } + - { from: g2, to: join } + - { from: join, to: out }`; + + it('resolves two concurrent gates settled in one timer sweep without a spurious stall', async () => { + const host = createInMemoryHost(); + const engine = engineWith( + { + g1: () => gate({ timeoutMs: 1000, timeoutAction: 'approve' }), + g2: () => gate({ timeoutMs: 1000, timeoutAction: 'approve' }), + }, + host, + ); + const handle = engine.start({ workflow: workflow(MULTIGATE) }); + const events: RunEvent[] = []; + for await (const event of handle.events) { + events.push(event); + if (event.type === 'run:paused' && event.pendingGateCount === 2) { + host.fireTimers(); // fire BOTH gate timers in one synchronous sweep + } + } + const resumes = events.filter((e) => e.type === 'human_gate:resumed'); + expect(resumes).toHaveLength(2); // both gates resolved, each exactly once + expect(terminalsIn(events)[0]?.type).toBe('run:completed'); // NOT a spurious run:failed{internal} + assertGapFreeSeq(events); + }); +}); + +// --- resumeFromCheckpoint: cross-process gate resume (1.R) ------------------------------------- + +describe('WorkflowEngine — resumeFromCheckpoint (cross-process resume, 1.R)', () => { + const GATED = ` id: gated + nodes: + - { id: start, type: input } + - { id: g, type: human_gate, gate_type: approval } + - { id: out, type: output } + edges: + - { from: start, to: g } + - { from: g, to: out }`; + const gateHandlers = { + g: (): NodeOutcome => ({ kind: 'paused', gate: { gateType: 'approval', message: 'approve?' } }), + }; + + /** Run a fresh gated run on `store` until it parks at the gate; return its runId, gateId, last seq. */ + async function runToGate( + store: RunStore, + ): Promise<{ runId: string; gateId: string; lastSeq: number }> { + const engine = engineWith(gateHandlers, createInMemoryHost({ store })); + const handle = engine.start({ workflow: workflow(GATED) }); + let gateId = ''; + let lastSeq = -1; + for await (const event of handle.events) { + lastSeq = Math.max(lastSeq, event.sequenceNumber); + if (event.type === 'run:paused') { + gateId = event.gateIds[0] ?? ''; + break; // the "process" dies here, parked at the gate — never resumed on this engine + } + } + return { runId: handle.runId, gateId, lastSeq }; + } + + it('rehydrates a gate-parked run in a fresh engine over the same store and drives it to completion', async () => { + const store = new InMemoryRunStore(); + const { runId, gateId, lastSeq } = await runToGate(store); + expect(gateId).not.toBe(''); + + // A brand-new engine (no in-memory state) resumes purely from the persisted event stream. + const engineB = engineWith({}, createInMemoryHost({ store })); + const handleB = await engineB.resumeFromCheckpoint({ + runId, + workflow: workflow(GATED), + gateId, + decision: { decision: 'approved', decidedBy: 'tester' }, + }); + const eventsB = await drain(handleB); + + expect(handleB.runId).toBe(runId); + expect(typesIn(eventsB)).toContain('human_gate:resumed'); + expect(eventsB.some((e) => e.type === 'node:started' && e.nodeId === 'out')).toBe(true); + expect(terminalsIn(eventsB)[0]?.type).toBe('run:completed'); + // The resumed stream continues gap-free from the last persisted sequence number (no reset, no gap). + eventsB.forEach((event, index) => expect(event.sequenceNumber).toBe(lastSeq + 1 + index)); + }); + + it('is a no-op (closed handle, nothing re-persisted) re-delivering to an already-terminal run', async () => { + const store = new InMemoryRunStore(); + const { runId, gateId } = await runToGate(store); + const decision = { decision: 'approved' as const, decidedBy: 't' }; + + const engineB = engineWith({}, createInMemoryHost({ store })); + await drain( + await engineB.resumeFromCheckpoint({ runId, workflow: workflow(GATED), gateId, decision }), + ); + const persistedAfterB = store.eventsFor(runId).length; + + // A second process re-delivers the same decision to the now-completed run — must not advance it. + const engineC = engineWith({}, createInMemoryHost({ store })); + const handleC = await engineC.resumeFromCheckpoint({ + runId, + workflow: workflow(GATED), + gateId, + decision, + }); + const eventsC = await drain(handleC); + expect(eventsC).toEqual([]); // closed handle: the iteration completes immediately + expect(store.eventsFor(runId).length).toBe(persistedAfterB); // nothing re-emitted / re-persisted + }); + + it('throws workflow_mismatch when handed a different workflow than the run started on', async () => { + const store = new InMemoryRunStore(); + const { runId, gateId } = await runToGate(store); + const OTHER = ` id: other + nodes: + - { id: start, type: input } + - { id: out, type: output } + edges: + - { from: start, to: out }`; + const engineB = engineWith({}, createInMemoryHost({ store })); + await expect( + engineB.resumeFromCheckpoint({ + runId, + workflow: workflow(OTHER), + gateId, + decision: { decision: 'approved', decidedBy: 't' }, + }), + ).rejects.toMatchObject({ code: 'workflow_mismatch' }); + }); + + it('throws unknown_run when no checkpoint exists for the runId', async () => { + const engine = engineWith({}, createInMemoryHost()); + await expect( + engine.resumeFromCheckpoint({ + runId: 'ghost', + workflow: workflow(GATED), + gateId: 'g', + decision: { decision: 'approved', decidedBy: 't' }, + }), + ).rejects.toMatchObject({ code: 'unknown_run' }); + }); + + it('throws run_already_active (use resume) when the run is already tracked in this engine', async () => { + const engine = engineWith(gateHandlers); + const handle = engine.start({ workflow: workflow(GATED) }); + let caught: unknown; + for await (const event of handle.events) { + if (event.type === 'run:paused') { + const gateId = event.gateIds[0] ?? ''; + try { + await engine.resumeFromCheckpoint({ + runId: handle.runId, + workflow: workflow(GATED), + gateId, + decision: { decision: 'approved', decidedBy: 't' }, + }); + } catch (error) { + caught = error; + } + await engine.resume(handle.runId, gateId, { decision: 'approved', decidedBy: 't' }); + } + } + expect(caught).toBeInstanceOf(EngineStateError); + expect(caught instanceof EngineStateError ? caught.code : '').toBe('run_already_active'); + }); + + it('throws invalid_decision for a malformed decision before touching the store', async () => { + const engine = engineWith({}, createInMemoryHost()); + await expect( + engine.resumeFromCheckpoint({ + runId: 'x', + workflow: workflow(GATED), + gateId: 'g', + // @ts-expect-error — an intentionally invalid decision value; safeParse must reject it + decision: { decision: 'maybe', decidedBy: 't' }, + }), + ).rejects.toMatchObject({ code: 'invalid_decision' }); + }); + + it('drives a run whose gate was already resolved in the prior process to completion WITHOUT re-applying the decision (kick path)', async () => { + const store = new InMemoryRunStore(); + // Process A: pause at the gate. + const { runId, gateId } = await runToGate(store); + // Process B: apply the decision, then "crash" mid-downstream — `out` hangs, so the run persists + // human_gate:resumed + node:started(out) but never run:completed. + const engineB = engineWith( + { out: () => new Promise(() => {}) }, + createInMemoryHost({ store }), + ); + const handleB = await engineB.resumeFromCheckpoint({ + runId, + workflow: workflow(GATED), + gateId, + decision: { decision: 'approved', decidedBy: 'human' }, + }); + for await (const event of handleB.events) { + if (event.type === 'node:started' && event.nodeId === 'out') { + break; // the process dies here, with `out` mid-flight + } + } + expect(store.eventsFor(runId).some((e) => e.type === 'human_gate:resumed')).toBe(true); + expect(store.eventsFor(runId).some((e) => e.type === 'run:completed')).toBe(false); + + // Process C: the gate is already resolved (resolvedGateIds), the run is non-terminal → kick(), which + // re-runs the unfinished `out` and completes WITHOUT a second human_gate:resumed. Snapshot the last + // persisted seq BEFORE the call — kick() emits synchronously, so a later read would include C's own + // first event. + const lastPersistedBeforeC = store + .eventsFor(runId) + .reduce((max, e) => Math.max(max, e.sequenceNumber), -1); + const engineC = engineWith({}, createInMemoryHost({ store })); + const handleC = await engineC.resumeFromCheckpoint({ + runId, + workflow: workflow(GATED), + gateId, + decision: { decision: 'approved', decidedBy: 'human' }, + }); + const eventsC = await drain(handleC); + expect(eventsC.some((e) => e.type === 'human_gate:resumed')).toBe(false); // never re-applied + expect(eventsC.some((e) => e.type === 'node:completed' && e.nodeId === 'out')).toBe(true); + expect(terminalsIn(eventsC)[0]?.type).toBe('run:completed'); + // The kick path shares #seedFromCheckpoint's seedSequence — its stream must also continue gap-free. + eventsC.forEach((event, index) => + expect(event.sequenceNumber).toBe(lastPersistedBeforeC + 1 + index), + ); + }); + + it('arms no gate timer on rehydration (re-arm is a Phase-2 reconciliation concern)', async () => { + const store = new InMemoryRunStore(); + // Process A: pause at a gate that carries a timeout. + const engineA = engineWith( + { + g: () => ({ + kind: 'paused', + gate: { gateType: 'approval', message: 'ok?', timeoutMs: 1000, timeoutAction: 'reject' }, + }), + }, + createInMemoryHost({ store }), + ); + const handleA = engineA.start({ workflow: workflow(GATED) }); + let gateId = ''; + for await (const event of handleA.events) { + if (event.type === 'run:paused') { + gateId = event.gateIds[0] ?? ''; + break; + } + } + // Process B rehydrates. Spy on setTimer to prove it is NEVER called during rehydration — distinguishing + // "never armed" from "armed then disarmed on resume" (which armedCount alone could not). + const baseHostB = createInMemoryHost({ store }); + let armCalls = 0; + const hostB: typeof baseHostB = { + ...baseHostB, + setTimer: (ms, onFire) => { + armCalls += 1; + return baseHostB.setTimer(ms, onFire); + }, + }; + const engineB = engineWith({}, hostB); + const handleB = await engineB.resumeFromCheckpoint({ + runId: handleA.runId, + workflow: workflow(GATED), + gateId, + decision: { decision: 'approved', decidedBy: 'h' }, + }); + await drain(handleB); + expect(armCalls).toBe(0); // rehydration armed no timer at all (re-arm is a Phase-2 concern) + }); }); // --- concurrency cap -------------------------------------------------------------------------- diff --git a/packages/core/src/engine/engine.ts b/packages/core/src/engine/engine.ts index 05dcb291..4c8c8eac 100644 --- a/packages/core/src/engine/engine.ts +++ b/packages/core/src/engine/engine.ts @@ -34,7 +34,9 @@ import { type ExecutionMode, type GateDecision, type MaskedSecret, + type NodeSkippedReason, type RunEvent, + type RunStatus, type TokensUsed, } from '@relavium/shared'; @@ -43,6 +45,7 @@ import type { PlanVertex, RunPlan } from '../run-plan.js'; import type { WorkflowDefinition } from '../parser.js'; import { EngineStateError } from './errors.js'; import { RunEventBus, type RunEventDraft } from './event-bus.js'; +import type { CheckpointState } from './checkpoint.js'; import type { AbortControllerLike, ExecutionHost } from './execution-host.js'; import type { GateRequest, @@ -52,7 +55,7 @@ import type { NodeOutcome, NodeStreamEvent, } from './node-executor.js'; -import { createRunHandle, type RunHandle } from './run-handle.js'; +import { createClosedRunHandle, createRunHandle, type RunHandle } from './run-handle.js'; /** A vertex's live status in one run. `paused` (at a gate) and `running` are not yet *settled*. */ type VertexStatus = 'pending' | 'running' | 'completed' | 'failed' | 'skipped' | 'paused'; @@ -78,6 +81,13 @@ const TERMINAL_TYPES: ReadonlySet = new Set( 'run:cancelled', ]); +/** The terminal `RunStatus` values — a checkpoint in one of these is a finished run (1.R resume no-op). */ +const TERMINAL_RUN_STATUSES: ReadonlySet = new Set([ + 'completed', + 'failed', + 'cancelled', +]); + /** The input to {@link WorkflowEngine.start} — a parsed workflow plus its run inputs and mode. */ export interface StartInput { /** The parsed, validated workflow (the host read the file and called `parseWorkflow`). */ @@ -90,6 +100,29 @@ export interface StartInput { readonly planOptions?: BuildRunPlanOptions; } +/** + * Inputs to {@link WorkflowEngine.resumeFromCheckpoint} — resume a run from a PRIOR process (1.R). + * + * **Invariant (caller's responsibility):** `workflow`, `inputs`, `executionMode`, and `planOptions` must + * be the SAME values the run started with. The checkpoint persists the workflow identity (verified — a + * mismatch throws `workflow_mismatch`) but does not yet persist `inputs` / `executionMode`, so passing + * different ones would silently diverge the rehydrated execution from its `run:started` state. A future + * revision will reconstruct these from the checkpoint and ignore the caller-supplied values. + */ +export interface ResumeFromCheckpointInput { + readonly runId: string; + /** The workflow to resume against — the engine refuses one whose identity differs (workflow_mismatch). */ + readonly workflow: WorkflowDefinition; + /** MUST match the run's original inputs (not yet checkpoint-derived — see the interface note). */ + readonly inputs?: Readonly>; + /** MUST match the run's original mode (not yet checkpoint-derived — see the interface note). */ + readonly executionMode?: ExecutionMode; + readonly planOptions?: BuildRunPlanOptions; + /** The gate to resolve + the decision to apply (the run was suspended at this gate). */ + readonly gateId: string; + readonly decision: GateDecision; +} + /** Construction dependencies for the engine — the injected host and node-executor seams. */ export interface WorkflowEngineDeps { readonly host: ExecutionHost; @@ -139,6 +172,10 @@ class RunExecution { readonly #abort: AbortControllerLike; readonly #states = new Map(); readonly #pendingGates = new Map(); + /** Gate ids whose decision was already applied — a re-delivery is an idempotent no-op (1.R). */ + readonly #resolvedGates = new Set(); + /** Disarm callbacks for armed gate-timeout timers, by gateId — disarmed on resume / settle (1.Q). */ + readonly #gateTimers = new Map void>(); #workflowId = ''; #settled = false; @@ -165,6 +202,8 @@ class RunExecution { bus: RunEventBus; capacity: number; onSettled: (runId: string) => void; + /** When present, the run is REHYDRATED from this checkpoint (resume) rather than started fresh (1.R). */ + checkpoint?: CheckpointState; }) { this.runId = params.runId; this.#plan = params.plan; @@ -185,8 +224,12 @@ class RunExecution { this.#secretInputNames = secretNames; this.#maskedInputs = maskInputs(params.inputs, secretNames); - for (const id of params.plan.vertices.keys()) { - this.#states.set(id, { status: 'pending' }); + if (params.checkpoint === undefined) { + for (const id of params.plan.vertices.keys()) { + this.#states.set(id, { status: 'pending' }); + } + } else { + this.#seedFromCheckpoint(params.plan, params.checkpoint, params.bus, params.runId); } this.handle = createRunHandle( params.bus, @@ -228,6 +271,56 @@ class RunExecution { } } + /** Seed `#states` / `#pendingGates` / tallies / the bus sequence from a checkpoint (rehydration, 1.R). */ + #seedFromCheckpoint(plan: RunPlan, cp: CheckpointState, bus: RunEventBus, runId: string): void { + for (const id of plan.vertices.keys()) { + const node = cp.nodeStates.get(id); + if (node === undefined) { + // Never started, OR running at the crash → re-run from `pending` (the idempotency key bounds a + // half-applied side effect; a settled node is never re-run). + this.#states.set(id, { status: 'pending' }); + continue; + } + this.#states.set(id, { + status: node.status, + ...(node.output === undefined ? {} : { output: node.output }), + ...(node.selectedTargets === undefined + ? {} + : { selectedTargets: new Set(node.selectedTargets) }), + }); + } + for (const gate of cp.pendingGates) { + // No gate-timeout timer is re-armed on rehydration: the gate this resume targets has its decision + // applied immediately. Re-arming a *remaining* gate's deadline is deferred to the Phase-2 + // crash-reconciliation that re-arms from persisted policy + a real clock (shared-core-engine.md) — + // the data it needs (timeoutAction + expiresAt) is now carried on `human_gate:paused`, so no backfill. + this.#pendingGates.set(gate.gateId, { vertexId: gate.nodeId }); + } + for (const gateId of cp.resolvedGateIds) { + this.#resolvedGates.add(gateId); + } + this.#totalInputTokens = cp.totalInputTokens; + this.#totalOutputTokens = cp.totalOutputTokens; + this.#cumulativeCostMicrocents = cp.cumulativeCostMicrocents; + // Post-resume events continue gap-free from the last persisted sequence number. + bus.seedSequence(runId, cp.lastSequenceNumber + 1); + // Keep measuring durationMs from the ORIGINAL start, so a resumed run's terminal reports total + // wall-clock (pre- + post-resume), not just the post-resume segment. NO `run:started` is re-emitted — + // it is already in the persisted log. + this.#startEpochMs = cp.startedAtMs; + } + + /** + * Drive a rehydrated run forward WITHOUT applying a gate decision — used by `resumeFromCheckpoint` + * when the target gate was already resolved in the prior process (a cross-process double-delivery): + * the decision must not be re-applied (no second `human_gate:resumed`), but the run still continues + * any unfinished downstream work, or re-pauses on a remaining gate. The terminal-checkpoint case never + * reaches here (it returns a closed handle); so this only ever finds work to do or another gate. + */ + kick(): void { + this.#schedule(); + } + requestCancel(): void { if (this.#settled) { throw new EngineStateError('run_already_terminal', 'the run has already terminated', { @@ -243,6 +336,12 @@ class RunExecution { } async resume(gateId: string, decision: GateDecision): Promise { + if (this.#resolvedGates.has(gateId)) { + // Idempotent: this gate's decision was already applied (a re-delivery / reconnect) — never advance + // the run twice (execution-model.md §gate). Checked BEFORE #settled so a re-delivery after the run + // completed is a no-op, not a `run_already_terminal` error. + return; + } if (this.#settled) { throw new EngineStateError('run_already_terminal', 'the run has already terminated', { runId: this.runId, @@ -262,8 +361,18 @@ class RunExecution { gateId, }); } + this.#resolvedGates.add(gateId); this.#pendingGates.delete(gateId); + this.#disarmTimer(gateId); // a decision arrived before the timeout — cancel the armed timer (1.Q) this.#pauseEpisode = false; // a later idle-with-gates re-emits run:paused for the remaining gates + // Mark the gate vertex completed SYNCHRONOUSLY before the await — mirroring #settleCompleted — so a + // concurrent #step (e.g. a sibling gate's timeout firing during this persist) never sees this gate as + // still `paused` while it is already out of #pendingGates, which would mis-read the run as stalled. + const state = this.#states.get(gate.vertexId); + if (state !== undefined) { + state.status = 'completed'; + state.output = decision.payload ?? { decision: decision.decision }; + } await this.#emitDurable({ type: 'human_gate:resumed', runId: this.runId, @@ -272,11 +381,6 @@ class RunExecution { decidedBy: decision.decidedBy, ...(decision.payload === undefined ? {} : { payload: decision.payload }), }); - const state = this.#states.get(gate.vertexId); - if (state !== undefined) { - state.status = 'completed'; - state.output = decision.payload ?? { decision: decision.decision }; - } this.#schedule(); } @@ -314,7 +418,11 @@ class RunExecution { if (this.#settled) { return; } - this.#propagateSkips(); + // Emit a durable `node:skipped` for each vertex the loop just dimmed — BEFORE any terminal settle — + // so the event log is a complete, replayable record (1.R reconstructs a skipped vertex from this). + for (const { id, reason } of this.#propagateSkips()) { + await this.#emitDurable({ type: 'node:skipped', runId: this.runId, nodeId: id, reason }); + } const running = this.#countRunning(); if (this.#cancelling) { @@ -477,6 +585,8 @@ class RunExecution { output: outcome.output, tokensUsed: tokens, durationMs: Math.max(0, this.#elapsedMs() - startedAtMs), + // A condition's branch selection — persisted so resume can restore `selectedTargets` (1.R). + ...(outcome.kind === 'branch' ? { selected: [...outcome.selected] } : {}), }); } @@ -501,7 +611,7 @@ class RunExecution { }); } - /** A `paused` outcome: park the gate and emit `human_gate:paused`. */ + /** A `paused` outcome: park the gate, arm its timeout timer (1.Q), and emit `human_gate:paused`. */ async #settlePaused(vertex: PlanVertex, gate: GateRequest): Promise { const gateId = gate.gateId ?? this.#host.ids.newId(); const state = this.#states.get(vertex.id); @@ -509,6 +619,26 @@ class RunExecution { state.status = 'paused'; } this.#pendingGates.set(gateId, { vertexId: vertex.id }); + // Compute the wall-clock deadline from the host clock (the handler has none) and arm a one-shot timer + // (1.Q). On fire, an `approve` action auto-resolves the gate; a `reject` (the safe default) fails the + // run with run_timeout. The timer is disarmed on resume / terminal settle so it never fires twice. + // The EFFECTIVE on-timeout policy (default the safe `reject`) — used for BOTH the armed timer and the + // emitted event, so the persisted `human_gate:paused` always carries the exact policy the engine acts + // on (even when a handler set timeoutMs but left timeoutAction implicit). A Phase-2 crash-resume reads + // it back to re-arm. `undefined` only when no timeout is configured. + const effectiveAction = + gate.timeoutMs === undefined ? undefined : (gate.timeoutAction ?? 'reject'); + const expiresAt = + gate.expiresAt ?? + (gate.timeoutMs === undefined + ? undefined + : new Date(Date.parse(this.#host.clock.now()) + gate.timeoutMs).toISOString()); + if (gate.timeoutMs !== undefined && effectiveAction !== undefined) { + const disarm = this.#host.setTimer(gate.timeoutMs, () => { + void this.#onGateTimeout(gateId, vertex.id, effectiveAction); + }); + this.#gateTimers.set(gateId, disarm); + } await this.#emitDurable({ type: 'human_gate:paused', runId: this.runId, @@ -518,10 +648,59 @@ class RunExecution { message: gate.message, ...(gate.assignee === undefined ? {} : { assignee: gate.assignee }), ...(gate.timeoutMs === undefined ? {} : { timeoutMs: gate.timeoutMs }), - ...(gate.expiresAt === undefined ? {} : { expiresAt: gate.expiresAt }), + ...(effectiveAction === undefined ? {} : { timeoutAction: effectiveAction }), + ...(expiresAt === undefined ? {} : { expiresAt }), }); } + /** Disarm and forget a gate's timeout timer (idempotent — safe if absent or already fired). */ + #disarmTimer(gateId: string): void { + const disarm = this.#gateTimers.get(gateId); + if (disarm !== undefined) { + this.#gateTimers.delete(gateId); + disarm(); + } + } + + /** + * A gate's timeout elapsed with no decision (1.Q). Idempotent: a no-op once the gate resolved (a human + * beat the timer — resume disarmed it, but a fired-and-queued callback still guards here) or the run + * settled. `approve` auto-resolves the gate as approved (`decidedBy: 'timeout'`); `reject` fails the run. + */ + async #onGateTimeout( + gateId: string, + vertexId: string, + action: 'approve' | 'reject', + ): Promise { + this.#disarmTimer(gateId); + if (this.#settled || !this.#pendingGates.has(gateId)) { + return; // already resolved or terminal + } + if (action === 'approve') { + await this.resume(gateId, { decision: 'approved', decidedBy: 'timeout' }); + return; + } + await this.#failGateOnTimeout(gateId, vertexId); + } + + /** Timeout with `timeout_action: reject` — fail the run with `run_timeout` (execution-model.md). */ + async #failGateOnTimeout(gateId: string, vertexId: string): Promise { + this.#pendingGates.delete(gateId); + // Mark the gate resolved (symmetry with resume / the approve path) so a late re-delivery of this + // gate's decision is an idempotent no-op rather than a `run_already_terminal` throw. + this.#resolvedGates.add(gateId); + const vertex = this.#plan.vertices.get(vertexId); + if (vertex === undefined) { + return; // unreachable: a pending gate always maps to a plan vertex + } + await this.#settleFailed(vertex, { + code: 'run_timeout', + message: 'the human gate timed out without a decision', + retryable: false, + }); + this.#schedule(); + } + /** Mark a vertex failed and fail the run (unless already cancelling/failing) — the internal backstop. */ #failNodeInternal(vertex: PlanVertex, message: string): void { const state = this.#states.get(vertex.id); @@ -554,6 +733,11 @@ class RunExecution { } this.#settled = true; this.#abort.abort(); // make sure any straggler executor sees cancellation + // The run is closing — no gate timer may fire afterwards (1.Q). Disarm each, then clear in one shot. + for (const disarm of this.#gateTimers.values()) { + disarm(); + } + this.#gateTimers.clear(); const durationMs = Math.max(0, this.#elapsedMs()); let draft: RunEventDraft; if (type === 'run:completed') { @@ -620,7 +804,9 @@ class RunExecution { return false; } - #propagateSkips(): void { + /** Skip-propagate to a fixpoint; return the vertices newly skipped this call (the caller emits them). */ + #propagateSkips(): Array<{ readonly id: string; readonly reason: NodeSkippedReason }> { + const skipped: Array<{ id: string; reason: NodeSkippedReason }> = []; let changed = true; while (changed) { changed = false; @@ -633,9 +819,30 @@ class RunExecution { continue; } state.status = 'skipped'; // all deps settled and every in-edge is dead → unreachable + skipped.push({ id, reason: this.#skipReason(vertex) }); changed = true; } } + return skipped; + } + + /** Why a vertex was skipped: a completed `condition` dependency routed away from it, else an upstream + * dependency was itself skipped/failed (so this vertex is unreachable). */ + /** + * Precedence (deliberate): a vertex is `branch_not_taken` if **any** dependency is a *completed* + * `condition` (one that ran and routed away from it) — that is the most specific, actionable cause. + * Only when no such dependency exists is the skip attributed to `upstream_unreachable` (a dead in-edge + * from a skipped/failed upstream). So a node downstream of both a taken-away condition and an + * unreachable upstream reports `branch_not_taken`. + */ + #skipReason(vertex: PlanVertex): NodeSkippedReason { + for (const dep of vertex.dependencies) { + const depVertex = this.#plan.vertices.get(dep); + if (depVertex?.type === 'condition' && this.#states.get(dep)?.status === 'completed') { + return 'branch_not_taken'; + } + } + return 'upstream_unreachable'; } /** How many vertices are currently executing — derived from status, the single source of truth. */ @@ -829,6 +1036,102 @@ export class WorkflowEngine { await execution.resume(gateId, parsed.data); } + /** + * Resume a run suspended at a gate in a PRIOR process (1.R): reconstruct its {@link CheckpointState} + * from the persisted event stream, rehydrate a {@link RunExecution} (seed node states / pending gates / + * tallies / the sequence counter — no `run:started` is re-emitted), apply the gate decision, and return + * the {@link RunHandle} so the caller observes the rest of the run. + * + * Idempotent re-delivery is a no-op (never advances the run twice; never re-emits a terminal event): + * - if the checkpoint is already **terminal** (the run finished in the prior process), a closed handle + * is returned and nothing is re-emitted or re-persisted; + * - if the target gate was already **resolved** but the run has not finished (a remaining gate, or + * downstream work the prior process did not reach), the decision is NOT re-applied — the run is just + * driven forward. + * + * Throws `unknown_run` when no checkpoint exists, or `run_already_active` when the run is already in + * memory (use {@link resume}). Within a single process the same guarantee holds via {@link resume}; the cross-process + * guarantee is bounded by the store's durable single-writer of `human_gate:resumed` per gate — a true + * concurrent double-resolve (two processes loading the same pending gate before either persists) is + * closed by a Phase-2 store-level uniqueness constraint, not the in-memory reference (checkpoint.ts). + */ + async resumeFromCheckpoint(input: ResumeFromCheckpointInput): Promise { + const parsed = GateDecisionSchema.safeParse(input.decision); + if (!parsed.success) { + throw new EngineStateError('invalid_decision', 'the gate decision failed validation', { + runId: input.runId, + gateId: input.gateId, + }); + } + if (this.#runs.has(input.runId)) { + throw new EngineStateError( + 'run_already_active', + 'the run is already in memory — use resume() rather than resumeFromCheckpoint()', + { runId: input.runId }, + ); + } + const checkpoint = await this.#host.checkpointer.load(input.runId); + if (checkpoint === undefined) { + throw new EngineStateError('unknown_run', 'no checkpoint exists for the supplied runId', { + runId: input.runId, + }); + } + // Only CHECKPOINT_SCHEMA_VERSION (v1) exists today, so no migration/guard runs here yet. When the + // derivation shape changes, this is the single point a future engine must refuse or migrate an older + // `checkpoint.schemaVersion` before consuming the state (the field exists precisely for that, 1.R). + // Identity guard: the workflow handed in must be the one the run started on. Comparing the surrogate + // `workflows.id` UUID catches resuming the wrong workflow entirely (a different slug). A subtler + // same-slug-edited-content drift needs a content hash on `run:started` — deferred (a canonical event + // contract change; checkpoint.ts), so resuming an edited-but-same-slug workflow is the caller's risk. + const expectedWorkflowId = await this.#host.store.resolveWorkflowId(input.workflow.workflow.id); + if (expectedWorkflowId !== checkpoint.workflowId) { + throw new EngineStateError( + 'workflow_mismatch', + 'the supplied workflow is not the one this run started on', + { runId: input.runId }, + ); + } + if (TERMINAL_RUN_STATUSES.has(checkpoint.runStatus)) { + // The run already settled in the prior process — re-delivery is a safe no-op (the terminal event + // is in the persisted log). Returning a closed handle avoids re-emitting/re-persisting a terminal. + return createClosedRunHandle(input.runId); + } + const plan = buildRunPlan(input.workflow, input.planOptions); + const bus = new RunEventBus({ now: this.#host.clock.now, validate: this.#validateEvents }); + const execution = new RunExecution({ + runId: input.runId, + plan, + workflow: input.workflow, + inputs: input.inputs ?? {}, + executionMode: input.executionMode ?? 'local', + host: this.#host, + executor: this.#executor, + bus, + capacity: this.#capacity, + onSettled: () => { + /* retained like a started run (see start) */ + }, + checkpoint, + }); + this.#runs.set(input.runId, execution); + try { + if (checkpoint.resolvedGateIds.includes(input.gateId)) { + // The gate was already resolved in the prior process (double-delivery); do not re-apply the + // decision — just drive any unfinished downstream work (or re-pause on a remaining gate). + execution.kick(); + } else { + // Apply the decision + drive the loop (events buffer on the handle for the returned consumer). + await execution.resume(input.gateId, parsed.data); + } + } catch (error) { + // resume() validates the gate AFTER rehydration; an unknown_gate / run_not_paused throw must not + // strand the half-initialized execution in #runs (a retry would then wrongly hit run_already_active). + this.#runs.delete(input.runId); + throw error; + } + return execution.handle; + } + /** Request cooperative cancellation. Throws {@link EngineStateError} for an unknown/terminal run. */ cancel(runId: string): void { const execution = this.#runs.get(runId); diff --git a/packages/core/src/engine/errors.ts b/packages/core/src/engine/errors.ts index 3fd8912f..97f04c2a 100644 --- a/packages/core/src/engine/errors.ts +++ b/packages/core/src/engine/errors.ts @@ -15,11 +15,13 @@ /** Stable discriminant for an engine-API-boundary fault — narrow on this, never on `message`. */ export type EngineStateErrorCode = - | 'unknown_run' // `resume` / `cancel` named a `runId` this engine instance is not tracking + | 'unknown_run' // `resume`/`cancel` named a `runId` this engine is not tracking, or `resumeFromCheckpoint` found no checkpoint for it + | 'run_already_active' // `resumeFromCheckpoint` named a run THIS engine already holds in memory — use `resume` instead | 'run_already_terminal' // the run already settled (completed / failed / cancelled) — no resume/cancel | 'run_not_paused' // `resume` was called while the run has no pending gate to resolve | 'unknown_gate' // the `gateId` does not match any gate currently pending on the run - | 'invalid_decision'; // the supplied `GateDecision` failed schema validation at the boundary + | 'invalid_decision' // the supplied `GateDecision` failed schema validation at the boundary + | 'workflow_mismatch'; // `resumeFromCheckpoint` was handed a workflow that is not the one the run started on /** * A `WorkflowEngine` API call could not be honoured. Thrown synchronously from `start` / `resume` / diff --git a/packages/core/src/engine/event-bus.ts b/packages/core/src/engine/event-bus.ts index e2c7d84d..48cad3e7 100644 --- a/packages/core/src/engine/event-bus.ts +++ b/packages/core/src/engine/event-bus.ts @@ -101,6 +101,18 @@ export class RunEventBus { return event; } + /** + * Seed the next `sequenceNumber` for a correlation key — used ONLY when rehydrating a run from a + * checkpoint (1.R), so events emitted after resume continue gap-free from the last persisted seq. + * Idempotent before any `next(key)`; never lower an already-advanced counter (a no-op guard). + */ + seedSequence(key: string, next: number): void { + const current = this.#sequence.get(key) ?? 0; + if (next > current) { + this.#sequence.set(key, next); + } + } + /** Fan a fully-stamped event out to every subscriber, isolating a throwing subscriber. */ deliver(event: RunEvent): void { for (const listener of this.#listeners) { diff --git a/packages/core/src/engine/execution-host.test.ts b/packages/core/src/engine/execution-host.test.ts index d9398aa9..46ae91b9 100644 --- a/packages/core/src/engine/execution-host.test.ts +++ b/packages/core/src/engine/execution-host.test.ts @@ -2,7 +2,12 @@ import { describe, expect, it, vi } from 'vitest'; import type { RunEvent } from '@relavium/shared'; -import { createAbortController, createInMemoryHost, InMemoryRunStore } from './execution-host.js'; +import { + createAbortController, + createInMemoryHost, + createManualTimerController, + InMemoryRunStore, +} from './execution-host.js'; describe('createAbortController — platform-free abort', () => { it('reports aborted, fires listeners once, and is idempotent', () => { @@ -173,3 +178,54 @@ describe('createInMemoryHost', () => { expect(host.ids.newId()).not.toBe(host.ids.newId()); }); }); + +describe('createManualTimerController — deterministic one-shot timer', () => { + it('fires an armed timer exactly once on fireTimers, then drops it', () => { + const timers = createManualTimerController(); + const fired = vi.fn(); + timers.setTimer(1000, fired); + expect(timers.armedCount()).toBe(1); + timers.fireTimers(); + expect(fired).toHaveBeenCalledTimes(1); + expect(timers.armedCount()).toBe(0); // dropped after firing + }); + + it('does not fire a disarmed timer', () => { + const timers = createManualTimerController(); + const fired = vi.fn(); + const disarm = timers.setTimer(1000, fired); + disarm(); + expect(timers.armedCount()).toBe(0); + timers.fireTimers(); + expect(fired).not.toHaveBeenCalled(); + }); + + it('is idempotent across consecutive fireTimers calls (no double-fire)', () => { + const timers = createManualTimerController(); + const fired = vi.fn(); + timers.setTimer(1000, fired); + timers.fireTimers(); + timers.fireTimers(); // a second sweep has nothing armed + expect(fired).toHaveBeenCalledTimes(1); + }); + + it('a callback that disarms a sibling timer mid-sweep is honored (snapshot is re-checked)', () => { + const timers = createManualTimerController(); + const second = vi.fn(); + let disarmSecond = (): void => undefined; + timers.setTimer(1000, () => { + disarmSecond(); // the first timer disarms the second before the sweep reaches it + }); + disarmSecond = timers.setTimer(1000, second); + timers.fireTimers(); + expect(second).not.toHaveBeenCalled(); // the armed re-check inside the sweep skipped it + }); + + it('disarm is safe to call after the timer already fired (idempotent)', () => { + const timers = createManualTimerController(); + const disarm = timers.setTimer(1000, () => undefined); + timers.fireTimers(); + expect(() => disarm()).not.toThrow(); + expect(timers.armedCount()).toBe(0); + }); +}); diff --git a/packages/core/src/engine/execution-host.ts b/packages/core/src/engine/execution-host.ts index 6f014676..464f1512 100644 --- a/packages/core/src/engine/execution-host.ts +++ b/packages/core/src/engine/execution-host.ts @@ -17,6 +17,8 @@ import type { AbortSignalLike, RunEvent } from '@relavium/shared'; +import { type Checkpointer, reconstructCheckpointState } from './checkpoint.js'; + /** A platform-free ISO-8601 timestamp source — injected so the engine never reads an ambient clock. */ export interface Clock { /** An ISO-8601 timestamp with offset (`…Z` or `±HH:MM`), matching the run-event envelope. */ @@ -111,17 +113,33 @@ export interface RunStore { } /** - * The injected execution-mode seam: clock + id source + persistence + abort, nothing platform-specific. - * The Phase-1 slice ships `clock.now()`; the one-shot **timer** port (for gate / run `timeout_ms` - * deadlines — ADR-0036 Decision 5) is added when the human gate (1.Q) and budget governor (1.AC) wire - * timeouts, since 1.N arms no timers. + * Arm a one-shot timer: invoke `onFire` **once** after `ms`, unless the returned disarm is called first. + * Injected so core never names the ambient `setTimeout`/`clearTimeout` (absent from the strict + * `lib: ["ES2023"]` purity build; CLAUDE.md rule 5). A real surface injects a `setTimeout`-backed timer; + * {@link createManualTimerController} provides a deterministic manual timer the engine tests fire by hand. + * Used by the human gate (1.Q) and budget governor (1.AC) for `timeout_ms` deadlines (ADR-0036 Decision 5) + * — never a sleep/poll loop, so the completion-driven scheduler stays event-driven. + */ +export type SetTimer = (ms: number, onFire: () => void) => () => void; + +/** + * The injected execution-mode seam: clock + id source + persistence + checkpointer + abort + timer, + * nothing platform-specific. The loop never branches on the execution mode — it calls the host. */ export interface ExecutionHost { readonly clock: Clock; readonly ids: IdSource; readonly store: RunStore; + /** + * The read side that reconstructs a run's {@link CheckpointState} from persisted rows so an interrupted + * run (crash, or suspended at a gate) can resume (1.R). Kept separate from the write `store` port. The + * real SQLite/cloud one is Phase-2/CLI; the in-memory reference is {@link createInMemoryCheckpointer}. + */ + readonly checkpointer: Checkpointer; /** Create a fresh abort controller for a run — injected so core never names the ambient global. */ readonly newAbortController: () => AbortControllerLike; + /** Arm a one-shot timer (gate / run `timeout_ms`); see {@link SetTimer}. */ + readonly setTimer: SetTimer; } // --- In-memory reference implementation (engine tests + the local reference) ------------------- @@ -204,21 +222,91 @@ export class InMemoryRunStore implements RunStore { } } +/** + * A deterministic, manual {@link SetTimer}: arming registers a timer but never fires it on a wall clock; + * a test fires every still-armed timer by calling {@link ManualTimerController.fireTimers}. This keeps + * gate/run-timeout tests reproducible and platform-free (no ambient `setTimeout`). Firing snapshots the + * armed set first, so a callback that arms or disarms timers cannot perturb the in-progress sweep. + */ +export interface ManualTimerController { + readonly setTimer: SetTimer; + /** Fire every currently-armed timer once (in arm order), then drop it. A disarmed timer never fires. */ + readonly fireTimers: () => void; + /** The count of still-armed timers — for a test asserting a gate's timer was disarmed on resume. */ + readonly armedCount: () => number; +} + +export function createManualTimerController(): ManualTimerController { + interface ManualTimer { + armed: boolean; + readonly onFire: () => void; + } + const timers = new Set(); + return { + setTimer: (_ms, onFire) => { + const timer: ManualTimer = { armed: true, onFire }; + timers.add(timer); + return () => { + timer.armed = false; + timers.delete(timer); + }; + }, + fireTimers: () => { + // Snapshot the armed set BEFORE firing: a callback may arm a new timer (which must NOT fire in this + // same sweep) or disarm a sibling — iterating the live Set would do both. The snapshot is required, + // not a convenience. + const due = Array.from(timers); + for (const timer of due) { + if (timer.armed) { + timer.armed = false; + timers.delete(timer); + timer.onFire(); + } + } + }, + armedCount: () => timers.size, + }; +} + /** * A deterministic in-memory {@link ExecutionHost} for the engine tests and the local reference: a clock - * that advances 1ms per read from a fixed base (valid ISO-8601, reproducible), a counter id source, and - * an {@link InMemoryRunStore}. A real surface injects wall-clock/UUID sources instead. + * that advances 1ms per read from a fixed base (valid ISO-8601, reproducible), a counter id source, an + * {@link InMemoryRunStore}, and a manual timer fired by hand (exposed as {@link fireTimers}/ + * {@link armedCount}). A real surface injects wall-clock/UUID/`setTimeout` sources instead. */ export function createInMemoryHost(options?: { store?: RunStore; + checkpointer?: Checkpointer; baseEpochMs?: number; -}): ExecutionHost & { store: RunStore } { +}): ExecutionHost & { store: RunStore } & Pick { let tick = options?.baseEpochMs ?? Date.parse('2026-01-01T00:00:00.000Z'); let idCounter = 0; + const store = options?.store ?? new InMemoryRunStore(); + const timers = createManualTimerController(); return { clock: { now: () => new Date(tick++).toISOString() }, ids: { newId: () => `id-${++idCounter}` }, - store: options?.store ?? new InMemoryRunStore(), + store, + checkpointer: options?.checkpointer ?? createInMemoryCheckpointer(store), newAbortController: createAbortController, + setTimer: timers.setTimer, + fireTimers: timers.fireTimers, + armedCount: timers.armedCount, + }; +} + +/** + * The in-memory reference {@link Checkpointer}: reconstructs from an {@link InMemoryRunStore}'s event log + * ({@link reconstructCheckpointState}). For any other (opaque) store it returns `undefined` — a custom + * store must supply its own checkpointer. Deterministic + dependency-free, for the engine tests. + */ +export function createInMemoryCheckpointer(store: RunStore): Checkpointer { + return { + load: (runId) => + Promise.resolve( + store instanceof InMemoryRunStore + ? reconstructCheckpointState(store.eventsFor(runId)) + : undefined, + ), }; } diff --git a/packages/core/src/engine/node-executor.ts b/packages/core/src/engine/node-executor.ts index 65a68829..ac002526 100644 --- a/packages/core/src/engine/node-executor.ts +++ b/packages/core/src/engine/node-executor.ts @@ -69,6 +69,14 @@ export interface GateRequest { readonly message: string; readonly assignee?: string; readonly timeoutMs?: number; + /** + * What the engine does if the gate's `timeoutMs` elapses with no decision (1.Q): `approve` auto-resolves + * the gate as approved (`decidedBy: 'timeout'`, the run continues); `reject` fails the run with + * `run_timeout` (execution-model.md `AwaitingGate → Failed`). The handler supplies it from the node's + * `timeout_action` (defaulting to the safe `reject`); it is only acted on when `timeoutMs` is set. + */ + readonly timeoutAction?: 'approve' | 'reject'; + /** The wall-clock deadline; the engine computes it from `timeoutMs` against its clock when omitted. */ readonly expiresAt?: string; } diff --git a/packages/core/src/engine/node-handlers/dispatcher.ts b/packages/core/src/engine/node-handlers/dispatcher.ts index 6615af51..76cb499b 100644 --- a/packages/core/src/engine/node-handlers/dispatcher.ts +++ b/packages/core/src/engine/node-handlers/dispatcher.ts @@ -15,6 +15,7 @@ import type { NodeExecutor } from '../node-executor.js'; import { createConditionNodeExecutor } from './condition.js'; import { createFanInNodeExecutor } from './fan-in.js'; import { createFanOutNodeExecutor } from './fan-out.js'; +import { createHumanGateNodeExecutor, type HumanGateNodeExecutorDeps } from './human-gate.js'; import { createInputNodeExecutor, createOutputNodeExecutor } from './io.js'; import { failed } from './scope.js'; import { createTransformNodeExecutor } from './transform.js'; @@ -42,11 +43,13 @@ export interface StandardNodeExecutorDeps { readonly sandbox: ExpressionSandbox; /** Agent-node wiring (provider resolution + tools). Omit to leave `agent` vertices unhandled. */ readonly agent?: AgentRunnerDeps; + /** Human-gate wiring (1.Q) — resolver capabilities for the gate's text templates. Defaults to none. */ + readonly humanGate?: HumanGateNodeExecutorDeps; } /** - * Wire the standard executor: the six 1.P handlers plus, when `agent` deps are supplied, the 1.O agent - * arm. `human_in_the_loop` (1.Q) and the reserved `loop`/`subworkflow`/`tool` types are intentionally + * Wire the standard executor: the six 1.P handlers, the 1.Q `human_in_the_loop` gate, plus — when `agent` + * deps are supplied — the 1.O agent arm. The reserved `loop`/`subworkflow`/`tool` types are intentionally * absent — they fail loud until their workstream lands. */ export function createStandardNodeExecutor(deps: StandardNodeExecutorDeps): NodeExecutor { @@ -56,6 +59,7 @@ export function createStandardNodeExecutor(deps: StandardNodeExecutorDeps): Node transform: createTransformNodeExecutor({ sandbox: deps.sandbox }), fan_in: createFanInNodeExecutor({ sandbox: deps.sandbox }), fan_out: createFanOutNodeExecutor(), + human_in_the_loop: createHumanGateNodeExecutor(deps.humanGate ?? {}), input: createInputNodeExecutor(), output: createOutputNodeExecutor(), }); diff --git a/packages/core/src/engine/node-handlers/human-gate.test.ts b/packages/core/src/engine/node-handlers/human-gate.test.ts new file mode 100644 index 00000000..25eea51e --- /dev/null +++ b/packages/core/src/engine/node-handlers/human-gate.test.ts @@ -0,0 +1,169 @@ +import type { AbortSignalLike } from '@relavium/shared'; +import { describe, expect, it } from 'vitest'; + +import type { NodeExecContext, NodeOutcome } from '../node-executor.js'; +import type { HumanGatePlanConfig, PlanVertex } from '../../run-plan.js'; +import { createHumanGateNodeExecutor } from './human-gate.js'; + +const LIVE: AbortSignalLike = { + aborted: false, + addEventListener: () => undefined, + removeEventListener: () => undefined, +}; +const ABORTED: AbortSignalLike = { ...LIVE, aborted: true }; + +/** Reads NOT-aborted once (passing the handler's entry guard), then aborted — so the abort surfaces from + * inside resolveTemplate and is caught, pinning the cancel-during-resolution window. */ +function abortAfterFirstRead(): AbortSignalLike { + let reads = 0; + return { + get aborted() { + return reads++ > 0; + }, + addEventListener: () => undefined, + removeEventListener: () => undefined, + }; +} + +type GateNode = HumanGatePlanConfig['node']; + +function gateVertex(node: Partial & Pick): PlanVertex { + const full: GateNode = { id: 'g', type: 'human_gate', ...node }; + return { + id: 'g', + type: 'human_in_the_loop', + dependencies: [], + dependents: [], + inputSites: [], + config: { kind: 'human_in_the_loop', node: full }, + }; +} + +function ctxFor( + vertex: PlanVertex, + opts: { + inputs?: Record; + runOutputs?: ReadonlyMap; + signal?: AbortSignalLike; + } = {}, +): NodeExecContext { + return { + vertex, + runOutputs: opts.runOutputs ?? new Map(), + inputs: opts.inputs ?? {}, + secretInputNames: new Set(), + toolPolicy: {}, + emit: () => undefined, + signal: opts.signal ?? LIVE, + attemptNumber: 1, + }; +} + +/** Narrow a NodeOutcome to its `paused` arm (no `as`), surfacing the gate request. */ +function gateOf(out: NodeOutcome): Extract['gate'] { + if (out.kind !== 'paused') { + throw new Error(`expected a paused outcome, got '${out.kind}'`); + } + return out.gate; +} + +const handler = createHumanGateNodeExecutor(); + +describe('createHumanGateNodeExecutor', () => { + it('resolves message_template + assignee against inputs / run.outputs', async () => { + const vertex = gateVertex({ + gate_type: 'approval', + assignee: '{{inputs.reviewer}}', + message_template: 'Approve {{inputs.file}} (score {{run.outputs["scan"].score}})?', + }); + const out = await handler.execute( + ctxFor(vertex, { + inputs: { reviewer: 'cem@example.com', file: 'auth.ts' }, + runOutputs: new Map([['scan', { score: 4 }]]), + }), + ); + const gate = gateOf(out); + expect(gate.gateType).toBe('approval'); + expect(gate.message).toBe('Approve auth.ts (score 4)?'); + expect(gate.assignee).toBe('cem@example.com'); + expect(gate.timeoutMs).toBeUndefined(); + expect(gate.timeoutAction).toBeUndefined(); + }); + + it('defaults timeout_action to the safe reject when timeout_ms is set without an action', async () => { + const gate = gateOf( + await handler.execute(ctxFor(gateVertex({ gate_type: 'review', timeout_ms: 60000 }))), + ); + expect(gate.timeoutMs).toBe(60000); + expect(gate.timeoutAction).toBe('reject'); + }); + + it('passes through an explicit timeout_action: approve', async () => { + const gate = gateOf( + await handler.execute( + ctxFor(gateVertex({ gate_type: 'approval', timeout_ms: 1000, timeout_action: 'approve' })), + ), + ); + expect(gate.timeoutAction).toBe('approve'); + }); + + it('omits the message when no template is authored (an empty, schema-valid string)', async () => { + const gate = gateOf(await handler.execute(ctxFor(gateVertex({ gate_type: 'input' })))); + expect(gate.message).toBe(''); + expect(gate.assignee).toBeUndefined(); + }); + + it('returns cancelled when the signal is already aborted', async () => { + const out = await handler.execute( + ctxFor(gateVertex({ gate_type: 'approval' }), { signal: ABORTED }), + ); + expect(out.kind).toBe('failed'); + if (out.kind === 'failed') { + expect(out.error.code).toBe('cancelled'); + expect(out.error.retryable).toBe(false); + } + }); + + it('returns cancelled (not validation) when the signal aborts DURING template resolution', async () => { + const out = await handler.execute( + ctxFor(gateVertex({ gate_type: 'approval', message_template: '{{inputs.x}}' }), { + inputs: { x: 'v' }, + signal: abortAfterFirstRead(), // passes the entry guard, then aborts inside resolveTemplate + }), + ); + expect(out.kind).toBe('failed'); + if (out.kind === 'failed') { + expect(out.error.code).toBe('cancelled'); // an abort is a deliberate cancel, not a data fault + } + }); + + it('maps a template interpolation failure to a fatal validation outcome', async () => { + // read_file with no injected capability throws InterpolationError → the handler returns `validation`. + const out = await handler.execute( + ctxFor(gateVertex({ gate_type: 'approval', message_template: '{{inputs.p | read_file}}' }), { + inputs: { p: 'secret.txt' }, + }), + ); + expect(out.kind).toBe('failed'); + if (out.kind === 'failed') { + expect(out.error.code).toBe('validation'); + expect(out.error.retryable).toBe(false); + } + }); + + it('fails loud (internal) if handed a non-gate node', async () => { + const wrong: PlanVertex = { + id: 'x', + type: 'output', + dependencies: [], + dependents: [], + inputSites: [], + config: { kind: 'output', node: { id: 'x', type: 'output' } }, + }; + const out = await handler.execute(ctxFor(wrong)); + expect(out.kind).toBe('failed'); + if (out.kind === 'failed') { + expect(out.error.code).toBe('internal'); + } + }); +}); diff --git a/packages/core/src/engine/node-handlers/human-gate.ts b/packages/core/src/engine/node-handlers/human-gate.ts new file mode 100644 index 00000000..4166a890 --- /dev/null +++ b/packages/core/src/engine/node-handlers/human-gate.ts @@ -0,0 +1,89 @@ +/** + * The `human_in_the_loop` node handler (1.Q) — the one node that suspends the run for an external + * decision. It fills the `paused`/`GateRequest` arm of {@link NodeOutcome} the seam reserved: it resolves + * the gate's human-facing `message_template` / `assignee` and returns `{ kind: 'paused', gate }`. The + * engine owns everything after that — generating the gate id, emitting `human_gate:paused`, arming the + * `timeout_ms` timer, parking the run, and resuming on a `GateDecision` (engine.ts `#settlePaused` / + * `resume`). The handler is intentionally thin and clock-free: deadlines (`expiresAt`) are the engine's + * job (only it holds the host clock). + * + * **Secrets — two layers.** A `secret`-typed `inputs.*` / `ctx.*` reference in `message_template` / + * `assignee` is rejected at PARSE time by the secret-taint analyzer (`node-text` category; analyze.ts). + * A `{{ run.outputs[…] }}` reference is *not* parse-gated (its content is runtime data), but is protected + * at RUNTIME: the `input` node masks every `secret`-typed input before it enters `run.outputs` (io.ts + * `maskSecretInputs`) and an agent prompt can't interpolate a secret (the same parse gate), so a raw + * secret never reaches `ctx.runOutputs` for this handler to surface. Together that lets the gate text + * resolve against raw inputs without a secret reaching the `human_gate:paused` payload — mirroring the + * agent's `prompt_template` (agent-runner.ts). + */ + +import { resolveTemplate } from '../../interpolation/resolve.js'; +import type { ResolverCapabilities, RunScope } from '../../interpolation/scope.js'; +import type { GateRequest, NodeExecContext, NodeExecutor, NodeOutcome } from '../node-executor.js'; +import { cancelled, failed } from './scope.js'; + +export interface HumanGateNodeExecutorDeps { + /** Resolver capabilities for `{{ … }}` in the gate's `message_template` / `assignee` (e.g. `read_file`). */ + readonly resolverCapabilities?: ResolverCapabilities; +} + +async function runHumanGate( + ctx: NodeExecContext, + deps: HumanGateNodeExecutorDeps, +): Promise { + const { config } = ctx.vertex; + if (config.kind !== 'human_in_the_loop') { + return failed('internal', `the human-gate handler received a '${config.kind}' node`, false); + } + if (ctx.signal.aborted) { + return cancelled(); + } + const { node } = config; + // Resolve the human-facing text against inputs + run.outputs (secrets are parse-gated; see file header). + const scope: RunScope = { + inputs: ctx.inputs, + ctx: {}, + outputs: Object.fromEntries(ctx.runOutputs), + }; + const caps = deps.resolverCapabilities ?? {}; + let message: string; + let assignee: string | undefined; + try { + message = + node.message_template === undefined + ? '' + : await resolveTemplate(node.message_template, scope, caps, ctx.signal); + assignee = + node.assignee === undefined + ? undefined + : await resolveTemplate(node.assignee, scope, caps, ctx.signal); + } catch (err) { + // A run cancelled mid-resolution surfaces as the throw from resolveTemplate's abort check — classify + // it as a deliberate `cancelled` (a distinct fatal reason node retry never re-runs), not a data fault. + if (ctx.signal.aborted) { + return cancelled(); + } + // Otherwise an interpolation failure is an authoring/data fault — fatal `validation`, matching the + // agent handler's prompt-resolution failure mapping (agent-runner.ts). + return failed( + 'validation', + err instanceof Error ? err.message : 'gate template interpolation failed', + false, + ); + } + const gate: GateRequest = { + gateType: node.gate_type, + message, + ...(assignee === undefined ? {} : { assignee }), + // A timeout is acted on by the engine only when timeout_ms is set; the action defaults to the safe + // `reject` (auto-approve is opt-in — dangerous; workflow-yaml-spec.md). expiresAt is the engine's job. + ...(node.timeout_ms === undefined + ? {} + : { timeoutMs: node.timeout_ms, timeoutAction: node.timeout_action ?? 'reject' }), + }; + return { kind: 'paused', gate }; +} + +export function createHumanGateNodeExecutor(deps: HumanGateNodeExecutorDeps = {}): NodeExecutor { + return { execute: (ctx) => runHumanGate(ctx, deps) }; +} diff --git a/packages/core/src/engine/node-handlers/node-handlers.e2e.test.ts b/packages/core/src/engine/node-handlers/node-handlers.e2e.test.ts index a1972e38..e11f3c21 100644 --- a/packages/core/src/engine/node-handlers/node-handlers.e2e.test.ts +++ b/packages/core/src/engine/node-handlers/node-handlers.e2e.test.ts @@ -94,6 +94,10 @@ describe('node-type handlers end-to-end through the WorkflowEngine (1.P)', () => const completedIds = events.filter((e) => e.type === 'node:completed').map((e) => e.nodeId); expect(completedIds).toContain('hi'); expect(completedIds).not.toContain('lo'); + // The dimmed branch emits a durable node:skipped (a complete, replayable log for 1.R + observability). + const skipped = events.find((e) => e.type === 'node:skipped'); + expect(skipped?.type === 'node:skipped' && skipped.nodeId).toBe('lo'); + expect(skipped?.type === 'node:skipped' && skipped.reason).toBe('branch_not_taken'); // The terminal output captured `hi`'s value (its single feeder). const completed = events.find((e) => e.type === 'run:completed'); diff --git a/packages/core/src/engine/node-handlers/node-handlers.test.ts b/packages/core/src/engine/node-handlers/node-handlers.test.ts index e631eac2..d1a9a62a 100644 --- a/packages/core/src/engine/node-handlers/node-handlers.test.ts +++ b/packages/core/src/engine/node-handlers/node-handlers.test.ts @@ -688,13 +688,22 @@ describe('dispatching executor (1.P)', () => { expect(out).toMatchObject({ kind: 'failed', error: { code: 'internal', retryable: false } }); }); - it('createStandardNodeExecutor wires the six non-agent handlers; an agent vertex stays unhandled without agent deps', async () => { + it('createStandardNodeExecutor wires the non-agent handlers incl. the 1.Q gate; an agent vertex stays unhandled without agent deps', async () => { const exec = createStandardNodeExecutor({ sandbox }); const transformV = makeVertex({ kind: 'transform', node: { id: 't', type: 'transform', transform: '7' }, }); expect(await exec.execute(makeCtx(transformV))).toEqual({ kind: 'completed', output: 7 }); + // The human_in_the_loop gate (1.Q) is now wired -> a gate vertex suspends, never fails loud. + const gateV = makeVertex({ + kind: 'human_in_the_loop', + node: { id: 'g', type: 'human_gate', gate_type: 'approval', message_template: 'ok?' }, + }); + expect(await exec.execute(makeCtx(gateV))).toMatchObject({ + kind: 'paused', + gate: { gateType: 'approval', message: 'ok?' }, + }); // No agent deps supplied -> the agent arm is absent -> loud internal failure, never a silent skip. const agentV = makeVertex({ kind: 'agent', node: { id: 'a', type: 'agent', agent_ref: 'x' } }); expect(await exec.execute(makeCtx(agentV))).toMatchObject({ diff --git a/packages/core/src/engine/run-handle.ts b/packages/core/src/engine/run-handle.ts index 149ad71a..29e8f24f 100644 --- a/packages/core/src/engine/run-handle.ts +++ b/packages/core/src/engine/run-handle.ts @@ -179,3 +179,22 @@ export function createRunHandle( whenConsumersReady: () => primary.whenDrained(), }; } + +/** + * A handle whose stream is already closed — for {@link RunHandle} consumers of a run that **already + * terminated in a prior process** (1.R `resumeFromCheckpoint` re-delivering a gate decision to a run + * whose checkpoint is already `completed`/`failed`/`cancelled`). It is a safe idempotent no-op: no event + * is re-emitted or re-persisted; the `events` iteration completes immediately (the actual terminal + * outcome is in the persisted `run_events`). `cancel`/`subscribe` are inert (the run is done). + */ +export function createClosedRunHandle(runId: string): RunHandle { + const primary = new RunEventStream(DEFAULT_CAPACITY); + primary.close(); + return { + runId, + events: primary, + subscribe: () => () => undefined, + cancel: () => undefined, + whenConsumersReady: () => Promise.resolve(), + }; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 6cdcdc16..d5500698 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -90,15 +90,26 @@ export type { // exactly-one-terminal-event guarantee (ADR-0036; sse-event-schema.md). Platform-free: host concerns // (clock / ids / persistence / abort) are injected via ExecutionHost. export { WorkflowEngine } from './engine/engine.js'; -export type { StartInput, WorkflowEngineDeps } from './engine/engine.js'; +export type { StartInput, ResumeFromCheckpointInput, WorkflowEngineDeps } from './engine/engine.js'; export { RunEventBus } from './engine/event-bus.js'; export type { RunEventBusOptions, RunEventListener, RunEventDraft } from './engine/event-bus.js'; export type { RunHandle } from './engine/run-handle.js'; export { InMemoryRunStore, createInMemoryHost, + createInMemoryCheckpointer, createAbortController, + createManualTimerController, } from './engine/execution-host.js'; +// Checkpointer + resume (1.R) — reconstruct a run's state from its persisted event stream (no checkpoint +// table; ADR-0003). The in-memory reference ships here; the SQLite/cloud one is Phase-2/CLI. +export { reconstructCheckpointState, CHECKPOINT_SCHEMA_VERSION } from './engine/checkpoint.js'; +export type { + Checkpointer, + CheckpointState, + CheckpointNodeState, + CheckpointPendingGate, +} from './engine/checkpoint.js'; export type { ExecutionHost, RunStore, @@ -106,6 +117,8 @@ export type { IdSource, AbortControllerLike, InterruptedRun, + SetTimer, + ManualTimerController, } from './engine/execution-host.js'; export type { NodeExecutor, @@ -149,6 +162,8 @@ export type { TransformNodeExecutorDeps } from './engine/node-handlers/transform export { createFanInNodeExecutor } from './engine/node-handlers/fan-in.js'; export type { FanInNodeExecutorDeps } from './engine/node-handlers/fan-in.js'; export { createFanOutNodeExecutor } from './engine/node-handlers/fan-out.js'; +export { createHumanGateNodeExecutor } from './engine/node-handlers/human-gate.js'; +export type { HumanGateNodeExecutorDeps } from './engine/node-handlers/human-gate.js'; export { createInputNodeExecutor, createOutputNodeExecutor } from './engine/node-handlers/io.js'; // Built-in ToolRegistry + dispatch (1.T) — the engine-side registry the AgentRunner (1.O) and diff --git a/packages/shared/src/constants.ts b/packages/shared/src/constants.ts index 41747214..22cce702 100644 --- a/packages/shared/src/constants.ts +++ b/packages/shared/src/constants.ts @@ -30,6 +30,7 @@ export const RUN_EVENT_TYPES = [ 'cost:updated', 'node:completed', 'node:failed', + 'node:skipped', 'human_gate:paused', 'human_gate:resumed', 'run:completed', diff --git a/packages/shared/src/run-event.test.ts b/packages/shared/src/run-event.test.ts index f4732aad..68dfd7e6 100644 --- a/packages/shared/src/run-event.test.ts +++ b/packages/shared/src/run-event.test.ts @@ -74,6 +74,12 @@ const valid: Record> = { nodeId: 'n', error: { code: 'tool_failed', message: 'boom', retryable: false }, }, + 'node:skipped': { + type: 'node:skipped', + ...env, + nodeId: 'n', + reason: 'branch_not_taken', + }, 'human_gate:paused': { type: 'human_gate:paused', ...env, @@ -81,6 +87,9 @@ const valid: Record> = { gateId: 'g1', gateType: 'approval', message: 'approve?', + timeoutMs: 1000, + timeoutAction: 'reject', + expiresAt: '2026-06-14T00:00:00.000Z', }, 'human_gate:resumed': { type: 'human_gate:resumed', @@ -191,6 +200,7 @@ const reject: Record> = { durationMs: 100, }, 'node:failed (missing error)': { type: 'node:failed', ...env, nodeId: 'n' }, + 'node:skipped (bad reason)': { type: 'node:skipped', ...env, nodeId: 'n', reason: 'because' }, 'human_gate:paused (bad gateType)': { type: 'human_gate:paused', ...env, @@ -242,7 +252,7 @@ describe('RunEvent union — every variant', () => { expect(RunEventSchema.safeParse(reject[name]).success).toBe(false); }); - it('covers exactly the 18 canonical colon-namespaced names, pinned to a literal list', () => { + it('covers exactly the 19 canonical colon-namespaced names, pinned to a literal list', () => { // A hardcoded contract list — independent of RUN_EVENT_TYPES — so the union and the // constant cannot silently drift together. const CONTRACT_NAMES = [ @@ -255,6 +265,7 @@ describe('RunEvent union — every variant', () => { 'cost:updated', 'node:completed', 'node:failed', + 'node:skipped', 'human_gate:paused', 'human_gate:resumed', 'run:completed', @@ -271,7 +282,7 @@ describe('RunEvent union — every variant', () => { // RunEventSchema wraps the union in the correlation-key refinement; reach the raw union. expect(RunEventSchema.innerType().options).toHaveLength(CONTRACT_NAMES.length); expect(new Set(RUN_EVENT_TYPES)).toEqual(new Set(CONTRACT_NAMES)); - expect(Object.keys(valid)).toEqual(CONTRACT_NAMES); // the matrix covers all 18 + expect(Object.keys(valid)).toEqual(CONTRACT_NAMES); // the matrix covers all 19 }); it('pins the RunEvent discriminant to RunEventType (type-level)', () => { diff --git a/packages/shared/src/run-event.ts b/packages/shared/src/run-event.ts index a85973b7..86548122 100644 --- a/packages/shared/src/run-event.ts +++ b/packages/shared/src/run-event.ts @@ -8,7 +8,7 @@ import { FS_SCOPE_TIERS, STOP_REASONS, } from './constants.js'; -import { GateTypeSchema } from './node.js'; +import { GateTypeSchema, TimeoutActionSchema } from './node.js'; /** * The run-event stream contract (sse-event-schema.md). A workflow run produces one ordered @@ -202,6 +202,13 @@ export const NodeCompletedEventSchema = z.object({ tokensUsed: TokensUsedSchema, durationMs: nonNegativeInt, attemptNumber: positiveInt.optional(), // 1-based retry attempt (matches cost:updated) + // The immediate downstream ids a `condition` kept live (its branch selection). Present ONLY for a + // condition's branch outcome — it is the authoritative record checkpoint/resume (1.R) reconstructs + // `selectedTargets` from, so a selected branch that was mid-flight at a crash re-runs (not skipped). + // NOT `.min(1)`: an EMPTY `selected` is a valid outcome — a condition that routes to no branch, which + // the engine skip-propagates across all downstream (engine.ts `#hasLiveEdge`); only the standard + // condition handler never emits it (it fails without a default), but the engine contract allows it. + selected: z.array(nonEmptyString).optional(), }); export const NodeFailedEventSchema = z.object({ @@ -211,6 +218,23 @@ export const NodeFailedEventSchema = z.object({ error: z.object(eventErrorFields), }); +/** Why a node was skipped — `branch_not_taken` (a `condition` routed away) or `upstream_unreachable`. */ +export const NodeSkippedReasonSchema = z.enum(['branch_not_taken', 'upstream_unreachable']); +export type NodeSkippedReason = z.infer; + +/** + * A vertex the run loop skip-propagated (a `condition` routed away from it, or every in-edge is dead + * because an upstream was skipped/failed). Emitted so the event log is a **complete, replayable** record + * — checkpoint/resume (1.R) reconstructs a skipped vertex from this event, and a surface can render the + * dimmed path instead of seeing the node silently vanish. + */ +export const NodeSkippedEventSchema = z.object({ + type: z.literal('node:skipped'), + ...runBase, + nodeId: nonEmptyString, + reason: NodeSkippedReasonSchema, +}); + export const HumanGatePausedEventSchema = z.object({ type: z.literal('human_gate:paused'), ...runBase, @@ -220,6 +244,10 @@ export const HumanGatePausedEventSchema = z.object({ message: z.string(), assignee: z.string().optional(), timeoutMs: nonNegativeInt.optional(), + // The on-timeout policy (present only with timeoutMs). Carried on the event so a surface can show how a + // gate auto-resolves AND so a Phase-2 crash-resume can re-arm the timer from the persisted log (the + // engine derives no separate gate record — execution-model.md). Absent ⇒ no timeout configured. + timeoutAction: TimeoutActionSchema.optional(), expiresAt: z.string().datetime({ offset: true }).optional(), }); export type HumanGatePausedEvent = z.infer; @@ -299,6 +327,7 @@ const RunEventUnionSchema = z.discriminatedUnion('type', [ CostUpdatedEventSchema, NodeCompletedEventSchema, NodeFailedEventSchema, + NodeSkippedEventSchema, HumanGatePausedEventSchema, HumanGateResumedEventSchema, RunCompletedEventSchema, @@ -328,6 +357,19 @@ export const RunEventSchema = RunEventUnionSchema.superRefine((event, ctx) => { path: [hasRunId ? 'sessionId' : 'runId'], }); } + // A gate's on-timeout policy only has meaning when a timeout is configured — refused at the union level + // because a discriminatedUnion member can't carry its own cross-field refinement (see note above). + if ( + event.type === 'human_gate:paused' && + event.timeoutAction !== undefined && + event.timeoutMs === undefined + ) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + message: 'timeoutAction is only valid when timeoutMs is also present', + path: ['timeoutAction'], + }); + } }); export type RunEvent = z.infer; @@ -392,6 +434,7 @@ export type AgentToolResultEvent = z.infer; export type AgentFilePatchProposedEvent = z.infer; export type NodeCompletedEvent = z.infer; export type NodeFailedEvent = z.infer; +export type NodeSkippedEvent = z.infer; export type RunCompletedEvent = z.infer; export type RunFailedEvent = z.infer; export type RunCancelledEvent = z.infer;