From ff25ed89280d97a872430ad8f60f65689c617f59 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 12 Aug 2026 18:01:58 +0000 Subject: [PATCH] fix(service-automation): a durable PAUSED run is visible after a cold restart (#8050) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `sys_automation_run` holds two disjoint row families — terminal history rows (`run_`-prefixed, written on completion) and live suspension rows (keyed by the raw run id, status `paused`). `AutomationEngine.listRuns` merged the in-memory ring buffer with the first family only, and `getRun` fell back to the first family only, so after a process restart a parked run answered: GET /automation/:name/runs → 200, zero rows GET /automation/:name/runs?status=paused → 200, zero rows GET /automation/:name/runs/:runId → 404 RESOURCE_NOT_FOUND while the same run served `…/runs/:runId/screen` and resumed cleanly. Before a restart the gap is invisible because a paused run is still in the ring; after one, the ring is empty and the suspension rows had no reader. The sharp edge is `?status=paused` — #7359 had just made it a real filter, and with no post-restart producer of a `paused` entry it could never match a row, so the one query an operator reaches for was guaranteed to answer "nothing pending". Both reads now consult the suspension rows, through one rehydration (`suspendedRunToLogEntry`) that reproduces the entry the two `status: 'paused'` recordLog sites write — trigger attribution rebuilt via `buildRunTrigger` on the persisted context, and the #7639 variable snapshot carried through. Read-path only: no column, prefix or lifecycle changes, and paused rows are not reshaped into history rows. Merge precedence is stated and pinned — durable paused → durable history → in-memory ring, weakest first — because a paused row is the only source that can be stale (its delete on completion is best-effort), so a finished run is never reported as still waiting (#3456). The new read is best-effort like the history read beside it: a store outage degrades the listing and says so, rather than throwing. Tests: `paused-run-visibility.test.ts` — 15 cases, measured 8 red / 7 green against `origin/main` with only the engine change reverted, including a full-stack cold boot of a second kernel over the same sqlite FILE. Three cases drafted as "green" measured red and are relabelled with the reading rather than softened. `suspended-run-store.test.ts` has one contrast assertion inverted (`getRun` → null for a cross-restart pause was the defect, not the contract) and gains a case for the distinction that survives: `hasSuspendedRun` throws on an unreadable store where `getRun` degrades. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_018P4qoXGyfvwYDMS57NftKL --- .../paused-run-visibility-after-restart.md | 39 ++ content/docs/automation/flows.mdx | 8 + .../plugin-approvals/src/approval-service.ts | 23 +- .../services/service-automation/src/engine.ts | 164 +++++- .../src/paused-run-visibility.test.ts | 507 ++++++++++++++++++ .../src/suspended-run-store.test.ts | 19 +- 6 files changed, 746 insertions(+), 14 deletions(-) create mode 100644 .changeset/paused-run-visibility-after-restart.md create mode 100644 packages/services/service-automation/src/paused-run-visibility.test.ts diff --git a/.changeset/paused-run-visibility-after-restart.md b/.changeset/paused-run-visibility-after-restart.md new file mode 100644 index 0000000000..b0c8d8d776 --- /dev/null +++ b/.changeset/paused-run-visibility-after-restart.md @@ -0,0 +1,39 @@ +--- +"@objectstack/service-automation": patch +--- + +fix(service-automation): a durable PAUSED run is visible to `listRuns` and run-detail after a cold restart (#8050) + +After a process restart, a run parked at an `approval` / `screen` / `wait` node +disappeared from the automation API while remaining fully durable: + +| read | before | after | +| :--- | :--- | :--- | +| `GET /automation/:name/runs` | 200, **zero rows** | the parked run | +| `GET /automation/:name/runs?status=paused` | 200, **zero rows** | the parked run | +| `GET /automation/:name/runs/:runId` | **404** `RESOURCE_NOT_FOUND` | 200, `status: 'paused'` | + +`sys_automation_run` holds two disjoint row families — terminal history rows +(`run_`-prefixed, written on completion) and live suspension rows (keyed by the +raw run id, status `paused`). `AutomationEngine.listRuns` merged the in-memory +ring buffer with the first family only, and `getRun` fell back to the first +family only. Before a restart the gap is invisible because a paused run is still +in the ring; after one, the ring is empty and the suspension rows had no reader. + +The sharp edge was `?status=paused`. #7359 had just made that a real filter, and +with no post-restart producer of a `paused` entry it could never match a row — +so the one query an operator reaches for when asking "what is in flight?" was +structurally guaranteed to answer "nothing pending". + +This is a read-path change only. Nothing about persistence moves: suspension +rows keep their own id space, lifecycle and retention exemption, and are **not** +reshaped into history rows. Durability was never the defect — a parked run +already served `…/runs/:runId/screen` and resumed cleanly across a restart, and +still does. + +Merge precedence is now stated explicitly: durable paused → durable history → +in-memory ring, weakest first. A paused row is the only source that can be stale +(the delete on completion is best-effort), so a terminal row or ring entry for +the same run id is later evidence and wins — a finished run is never reported as +still waiting. The paused read is best-effort like the history read beside it: a +store outage degrades the listing and logs the shortfall rather than throwing. diff --git a/content/docs/automation/flows.mdx b/content/docs/automation/flows.mdx index b6fdef50f0..5a332f7a81 100644 --- a/content/docs/automation/flows.mdx +++ b/content/docs/automation/flows.mdx @@ -711,6 +711,14 @@ panel. Recent runs are held in an in-memory ring buffer; terminal runs history with a bounded step log, so `listRuns` / `getRun` still report a run's status, steps, and failure reason after a restart or ring-buffer eviction. +Runs still **in flight** survive the same way. A run parked at an `approval`, +`screen` or `wait` node is persisted as a live suspension row, and both +`listRuns` (including `?status=paused`) and `getRun` read those rows back — so +after a restart the Runs view shows what is *waiting*, not only what finished. +A paused run reports its trigger attribution and its variable snapshot exactly +as it did before the restart; it carries no `durationMs`, because a suspension +records when the run started, not when it parked. + ### Run summaries A run that reports `success: true` has not told you it did its job. A scheduled diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index a0fc0db91b..423c13622b 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -118,10 +118,13 @@ export interface ApprovalResumeSurface { * a decision from being recorded against a run that can never advance * (#4420). Read-only; it never consumes the suspension. * - * Distinct from {@link getRun}, which reports on the execution LOG: a run - * suspended by a PREVIOUS process resolves to `null` there even when its - * state is durable, so it cannot tell "waiting for a human" from "dead". - * This asks the suspension store itself. + * Still distinct from {@link getRun}, but no longer on the axis this comment + * used to name: since #8050 `getRun` also sees a run suspended by a PREVIOUS + * process (it resolved `null` there before, unable to tell "waiting for a + * human" from "dead"). The difference that remains is the one a pre-flight + * turns on — this asks the suspension store and REJECTS when it cannot be + * read, where `getRun` degrades an outage to `null`. A caller about to WRITE + * must not accept a degraded read. * * Rejects when the durable store cannot be read — existence is then * unknown, and callers must not read an outage as a dead run. Optional: an @@ -3023,10 +3026,14 @@ export class ApprovalService implements IApprovalService { * class of failure stayed silent. * * It also could not have answered the question even if it looked: its - * liveness oracle is `getRun`, which reads the execution LOG, and after a - * restart that returns `null` for a perfectly ALIVE suspended run. It treats - * `null` as alive (conservative, correct) — but that means it has no way to - * say "this run is really gone". + * liveness oracle is `getRun`, which treats both `null` and `paused` as alive + * (conservative, correct) — so it has no way to say "this run is really + * gone". (Until #8050 that oracle was weaker still: after a restart it + * returned `null` for a perfectly ALIVE suspended run, so "alive" and + * "unknown" were the same answer. It now reports such a run as `paused`, + * which does not change any branch here — both already meant "leave alone" — + * but it is why the sweep below needs `hasSuspendedRun` as a second oracle + * rather than a sharper reading of the first.) * * So this uses BOTH oracles, and a row must fail both to be reported: * diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 075fa5f185..cfa218bd80 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -2556,6 +2556,68 @@ export class AutomationEngine implements IAutomationService { const limit = options?.limit ?? 20; const inMem = this.executionLogs.filter(l => l.flowName === flowName); + // [#8050] Durable PAUSED rows — the arm this merge was missing. + // + // `sys_automation_run` holds TWO disjoint row families: the terminal + // history rows `recordTerminal` writes (id `run_` + runId, status + // completed/failed) and the LIVE suspension rows `save` writes (id = + // the raw runId, status `paused`). The history arm below reads the + // first family; nothing here read the second. Before a restart that is + // invisible, because a paused run is still in `executionLogs` — so + // every in-process test of this method passes. After a restart the ring + // is empty and the paused rows had no reader at all, which is the + // defect: an operator who restarts the process can enumerate what + // FINISHED and what is IN FLIGHT vanishes — the strictly more urgent + // half. It also structurally emptied #7359's just-enforced + // `?status=paused`: with no producer of a `paused` entry after a + // restart, that filter could never match a row, and the one query + // reached for here always answered "nothing pending". + // + // Read-side only, by construction: this rehydrates the row the suspend + // path already writes. No column, prefix or lifecycle changes — the + // paused row is NOT reshaped into a history row, because the two have + // different lifetimes (a paused row is live resumable state, deleted on + // completion and exempt from the age sweep; a history row is a + // tombstone). Unifying them would trade an observability gap for a + // persistence-semantics change. + // + // Skipped when the caller filters for a status a paused row can never + // have: `suspendedRunToLogEntry` always yields `paused`, and a run that + // has since finished is answered by the fresher history/ring entry that + // outranks it in the merge below — so the arm cannot change the result + // of `?status=failed`, only its cost. `store.list()` is a table scan of + // every paused row in the deployment (see its own contract), so not + // paying it on the monitoring queries is worth the one-line guard. + const wantsPaused = options?.status === undefined || options.status === 'paused'; + let durablePaused: ExecutionLogEntry[] = []; + if (this.store && wantsPaused) { + try { + const rows = await this.store.list(); + durablePaused = rows + .filter(r => r.flowName === flowName) + .map(r => this.suspendedRunToLogEntry(r)); + } catch (err) { + // #6499 — driver text to the structured slot; `warn(message, + // meta?)`, meta SECOND (no `Error` slot on `warn`). + // + // #4632 verdict: FUNCTIONAL — `warn`, for the same reason as + // the history arm below and `listSuspendedRunsDurable`. Nothing + // claimed-persisted failed to land: the paused rows are intact + // and still resumable by id (`resume` reads them through + // `loadSuspendedRun`, a different door that is unaffected by + // this failure). What degrades is this observability read, back + // to exactly the pre-#8050 answer. + this.logger.warn( + `[Automation] paused-run read failed for '${flowName}' — the Runs listing DEGRADES to the ` + + `in-memory ring buffer plus terminal history, so runs parked by a previous process are ` + + `missing and '?status=paused' can report an empty result for a flow that has runs ` + + `waiting. The rows themselves are untouched and still resumable by id. Fix the store ` + + `failure in this record's meta.`, + describeThrownForLog(err), + ); + } + } + // Merge durable run history so the "Runs" view survives a restart and // ring-buffer eviction. In-memory entries are the freshest (they carry // full step detail); durable rows backfill runs the process no longer @@ -2588,7 +2650,28 @@ export class AutomationEngine implements IAutomationService { ); } } + // Dedupe by run id, weakest source first — the same run legitimately + // appears in more than one of these (#8050): + // + // 1. durable PAUSED — the run parked, and the row is still there. + // 2. durable HISTORY — the run reached a terminal state. + // 3. in-memory ring — this process executed it. + // + // The order is a precedence claim, not an accident. Paused loses to + // both because it is the only one that can be STALE while the others + // cannot: `forgetSuspendedRun` deletes the paused row on completion, + // but that delete is best-effort (a store outage swallows it), so a + // finished run can leave its paused row behind. A terminal row or a + // terminal ring entry for the same id is therefore strictly later + // evidence, and letting the paused row win would report a completed run + // as still waiting — the exact defect `run-history.test.ts`'s "latest + // entry wins" block pins for `getRun`. There is no symmetric hazard: + // within a process the ring is written in the same breath as the paused + // row (`persistSuspendedRun` then `recordLog`, and again on re-suspend), + // so it is never the older of the two; across a restart the ring is + // empty and cannot mask anything. const byId = new Map(); + for (const e of durablePaused) byId.set(e.id, e); for (const e of durable) byId.set(e.id, e); for (const e of inMem) byId.set(e.id, e); // freshest wins @@ -2669,6 +2752,47 @@ export class AutomationEngine implements IAutomationService { }; } + /** + * Rehydrate a durably-stored {@link SuspendedRun} into the `paused` + * {@link ExecutionLogEntry} the Runs surfaces expect (#8050) — the + * suspension-row twin of {@link runRecordToLogEntry}. + * + * Deliberately reconstructs the SAME entry the two `status: 'paused'` + * `recordLog` sites write, from the same inputs, so that whether a paused + * run is read before or after a restart is invisible to the caller: + * + * - `trigger` goes through {@link buildRunTrigger} on the persisted + * `context_json`, not through the flattened `trigger_*` columns. Those + * columns exist for FILTERING (#7533) and drop `type` to `null` where + * the log entry says `'manual'`; the context is what the ring entry was + * built from, so reusing the chokepoint reproduces it exactly rather + * than approximating it. + * - `variables` is carried because #7639 made it part of what a PAUSED run + * discloses on run-detail, and the row has held the same snapshot all + * along (`variables_json` is written from the very object handed to the + * log entry). Dropping it here would have re-opened #7639 for exactly + * the runs an operator most needs it for — the ones that outlived the + * process. + * + * `durationMs` / `completedAt` are absent because a suspension row records + * no pause instant — only `started_at` / `start_time`. Absent reads as "not + * recorded", which is what the schema's `optional()` means; inventing an + * age-since-start here would publish a number that grows every time the row + * is read and is not the "time spent executing" the ring entry reports. + */ + private suspendedRunToLogEntry(run: SuspendedRun): ExecutionLogEntry { + return { + id: run.runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'paused', + startedAt: run.startedAt, + trigger: buildRunTrigger(run.context), + steps: run.steps ?? [], + variables: run.variables ?? {}, + }; + } + async getRun(runId: string): Promise { // LAST entry wins, not the first: a run that pauses and later finishes // records TWO entries under the same run id ('paused', then @@ -2712,6 +2836,38 @@ export class AutomationEngine implements IAutomationService { ); } } + // [#8050] …and the PAUSED fallback, so run-detail and `listRuns` answer + // out of one story. The card measured both surfaces failing together + // after a restart — list returning zero rows AND this method 404ing — + // and fixing only the list would have swapped a visible gap for an + // inconsistency between two reads of the same run. + // + // AFTER the terminal probe, matching the merge order in `listRuns`: a + // paused row can outlive the run it describes (the delete on completion + // is best-effort), so a terminal row for the same id is later evidence + // and must win. Trying this first would resurrect #3456's "paused + // forever" for any run whose cleanup delete was lost. + // + // This does NOT make a nonexistent run findable: `store.load` answers + // `null` for an unknown id exactly as `loadTerminal` does, so the route + // above still returns its 404 `RESOURCE_NOT_FOUND` envelope for one. + if (this.store) { + try { + const suspended = await this.store.load(runId); + if (suspended) return this.suspendedRunToLogEntry(suspended); + } catch (err) { + // #6499 / #4632: same verdict as the terminal probe above — + // FUNCTIONAL, so `warn`. The suspension row is intact and the + // run stays parked and resumable; what degrades is this read. + this.logger.warn( + `[Automation] durable paused-run lookup failed for '${runId}' — this read DEGRADES to null, ` + + `so a run that is parked and resumable reports as if it had never run, and the caller ` + + `cannot tell the two apart. The suspension row is untouched. Fix the store failure in ` + + `this record's meta.`, + describeThrownForLog(err), + ); + } + } return null; } @@ -3510,9 +3666,11 @@ export class AutomationEngine implements IAutomationService { * against a run that can no longer advance (#4420). * * THROWS when the durable store cannot be read — an outage means "unknown", - * and a caller must not act on it as if the run were gone. Contrast - * {@link getRun}, which reports on the execution LOG and returns null for a - * run suspended by a previous process even when its state is durable. + * and a caller must not act on it as if the run were gone. That is the one + * axis {@link getRun} still differs on: since #8050 it, too, sees a run + * suspended by a previous process, but as an OBSERVABILITY read it degrades + * a store failure to `null` with a warning rather than throwing. Use this + * one before writing anything of your own; use `getRun` to display. */ async hasSuspendedRun(runId: string): Promise { return (await this.loadSuspendedRunStrict(runId)) !== null; diff --git a/packages/services/service-automation/src/paused-run-visibility.test.ts b/packages/services/service-automation/src/paused-run-visibility.test.ts new file mode 100644 index 0000000000..bde2672ed9 --- /dev/null +++ b/packages/services/service-automation/src/paused-run-visibility.test.ts @@ -0,0 +1,507 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #8050 — a durably PAUSED run must be visible to the automation API after a + * cold restart, on BOTH read surfaces. + * + * `sys_automation_run` holds two disjoint row families: terminal history rows + * (`recordTerminal`, id `run_`+runId) and live suspension rows (`save`, id = + * the raw runId, status `paused`). `AutomationEngine.listRuns` merged the + * in-memory ring buffer with the FIRST family only, and `getRun` fell back to + * the first family only. So after a restart: + * + * - `GET /automation/:name/runs` → 200, zero rows + * - `GET /automation/:name/runs?status=paused` → 200, zero rows + * - `GET /automation/:name/runs/:runId` → 404 "Execution not found" + * + * while the same run answered `GET …/runs/:runId/screen` and resumed cleanly — + * the state was never lost, it simply had no reader. The second line is the + * sharp one: #7359 had just made `?status=paused` a real filter, and with no + * post-restart producer of a `paused` entry it could never match a row. The one + * query an operator reaches for when asking "what is in flight?" was + * structurally guaranteed to answer "nothing". + * + * ## Why the obvious test is worthless here + * + * Parking a run and enumerating it IN THE SAME PROCESS passes on `main` too — + * the run is still in the ring buffer, and the ring is what every existing + * `listRuns` test reads. Likewise the existing durability tests + * (`suspended-screen-durability.test.ts`, `run-history.test.ts`) assert + * RESUMABILITY across a restart, which is precisely the half that already + * worked. The gate has to be: park → cold restart onto the same storage → read. + * + * ## Reverse verification — measured, not asserted + * + * Every case below was run against `origin/main` with ONLY the engine change + * reverted (the test file unchanged). 8 red, 7 green: + * + * RED cold restart: ?status=paused returns the parked run → [] + * RED cold restart: bare enumeration returns the parked run → [] + * RED cold restart: run-detail answers → null + * RED cold restart: trigger attribution + #7639 variables → null + * RED cold restart: no cross-flow leak → [] + * RED cold restart: ?status= narrows without widening → [] (len 0, want 1) + * RED an unreadable paused store degrades rather than throws → no such warning + * RED full stack, cold boot over the same sqlite FILE → [] / null + * + * GREEN pre-restart: the run is in both sources and lists once + * GREEN pre-restart: another flow's paused run does not leak in + * GREEN a stale paused row cannot mask a finished run (ring side) + * GREEN a stale paused row cannot mask a finished run (durable side, cold) + * GREEN a parked run still resumes to completion after a restart + * GREEN an unknown run id is still not found + * GREEN a flow with no paused rows behaves exactly as before + * + * Three cases I had labelled GREEN when writing them are red above, and the + * labels — not the tests — were wrong: no-cross-flow-leak, ?status=-does-not- + * widen and the degradation pin all assert the row is THERE before they assert + * anything about its scoping, so none of them can pass on a tree where it never + * appears. They are recorded as red, with the extra invariant each carries + * named on the case itself. The one genuinely-green scoping guard is the + * PRE-restart half, which reads the ring and passes both sides. + * + * The greens are not decoration either. The fix adds a THIRD source to a merge + * that had two, and a third source is exactly how duplicate rows, a resurrected + * "paused forever" (#3456) and a cross-flow leak get introduced. + */ + +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { describe, it, expect, afterEach } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import { ObjectKernel } from '@objectstack/core'; +import { ObjectQLPlugin, type ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; + +import { AutomationEngine } from './engine.js'; +import type { RunRecord, SuspendedRun, SuspendedRunStore } from './engine.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; +import { AutomationServicePlugin } from './plugin.js'; + +const silent = { info() {}, warn() {}, error() {}, debug() {}, child() { return silent; } } as never; +const flush = () => new Promise((r) => setTimeout(r, 0)); + +/** + * `resumeAuthority: 'any'` is required of any pausing fixture since #5561 — a + * node type that declares no authority is refused at the public `resume` door. + * Nothing here is about that gate (`resume-authority-gate.test.ts` owns it), so + * the fixture states the posture it relies on, as the pausing built-ins do. + */ +const HOLD_DESCRIPTOR = defineActionDescriptor({ + type: 'hold', version: '1.0.0', name: 'Hold', + supportsPause: true, resumeAuthority: 'any', +}); + +const holdExecutor = { + type: 'hold', + descriptor: HOLD_DESCRIPTOR, + async execute() { return { success: true, suspend: true, correlation: 'held' }; }, +} as never; + +/** start → hold (parks) → tail → end. */ +function pausingFlow(name: string, tail = 'noop') { + return { + name, label: name, type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 's' }, + { id: 'hold', type: 'hold', label: 'h' }, + { id: 'tail', type: tail, label: 't' }, + { id: 'end', type: 'end', label: 'e' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'hold' }, + { id: 'e2', source: 'hold', target: 'tail' }, + { id: 'e3', source: 'tail', target: 'end' }, + ], + }; +} + +/** + * One simulated process lifetime over `store`. A shared + * {@link InMemorySuspendedRunStore} across two of these IS the cold restart at + * the engine seam: engine B's ring buffer is empty, and the store JSON + * round-trips on save/load so it exercises the same serialization boundary + * `sys_automation_run` imposes. (The full-stack version over a real sqlite FILE + * is the last describe block — this one isolates the engine.) + */ +function buildEngine(store: SuspendedRunStore, flows: string[] = ['approval_flow']) { + const engine = new AutomationEngine(silent, store); + engine.registerNodeExecutor(holdExecutor); + engine.registerNodeExecutor({ type: 'noop', async execute() { return { success: true }; } } as never); + for (const f of flows) engine.registerFlow(f, pausingFlow(f) as never); + return engine; +} + +const TRIGGER = { + event: 'record_change', + object: 'crm_order', + record: { id: 'ord_1' }, + userId: 'usr_ops', +} as unknown as AutomationContext; + +describe('#8050 durable paused runs are visible after a cold restart', () => { + it('cold restart: ?status=paused returns the parked run (the query an operator reaches for)', async () => { + const store = new InMemorySuspendedRunStore(); + const parked = await buildEngine(store).execute('approval_flow', TRIGGER); + expect(parked.status).toBe('paused'); + + // New process: empty ring, same durable rows. + const cold = buildEngine(store); + const paused = await cold.listRuns('approval_flow', { status: 'paused' }); + + // RED on main: `[]`. #7359's filter had no post-restart producer of a + // `paused` entry, so it could never match a row. + expect(paused.map(r => r.id)).toEqual([parked.runId]); + expect(paused[0].status).toBe('paused'); + }); + + it('cold restart: bare enumeration returns the parked run', async () => { + const store = new InMemorySuspendedRunStore(); + const parked = await buildEngine(store).execute('approval_flow', TRIGGER); + + const rows = await buildEngine(store).listRuns('approval_flow'); + + // RED on main: `[]` — the flow had exactly one run and the Runs view + // showed none of it. + expect(rows.map(r => r.id)).toEqual([parked.runId]); + }); + + it('cold restart: run-detail answers, with the same story the list tells', async () => { + const store = new InMemorySuspendedRunStore(); + const parked = await buildEngine(store).execute('approval_flow', TRIGGER); + + const cold = buildEngine(store); + const detail = await cold.getRun(parked.runId!); + + // RED on main: `null` → the route's 404 "Execution not found". Fixing + // only `listRuns` would have left this 404ing, trading a visible gap + // for an inconsistency between two reads of the same run. + expect(detail).not.toBeNull(); + expect(detail!.status).toBe('paused'); + + // ONE story: by-id and list agree field for field. + const [listed] = await cold.listRuns('approval_flow', { status: 'paused' }); + expect(detail).toEqual(listed); + }); + + it('cold restart: the rehydrated entry carries the trigger attribution and the #7639 variable snapshot', async () => { + const store = new InMemorySuspendedRunStore(); + const parked = await buildEngine(store).execute('approval_flow', TRIGGER); + + const before = await store.load(parked.runId!); + const cold = buildEngine(store); + const detail = (await cold.getRun(parked.runId!))!; + + // #7533 — rebuilt through `buildRunTrigger` on the persisted context, + // not from the flattened `trigger_*` filter columns (which spell an + // absent type `null` where the log entry says 'manual'). A run + // rehydrated after a restart still says what fired it and on which + // record. + expect(detail.trigger).toEqual({ + type: 'record_change', object: 'crm_order', recordId: 'ord_1', userId: 'usr_ops', + }); + // #7639 — `variables` is part of what a PAUSED run discloses on + // run-detail. Dropping it here would have re-opened that card for + // exactly the runs that need it most: the ones that outlived the + // process. It is the same snapshot the continuation will resume from. + expect(detail.variables).toEqual(before!.variables); + expect(detail.flowName).toBe('approval_flow'); + expect(detail.startedAt).toBe(before!.startedAt); + }); + + it('GREEN: durability is untouched — the run still resumes to completion after the restart', async () => { + const store = new InMemorySuspendedRunStore(); + const parked = await buildEngine(store).execute('approval_flow', TRIGGER); + + // The half that already worked, and the half a read-path change is most + // likely to break by accident: the filer measured resume 200 / acted:1 + // post-restart, so this must not regress. + const cold = buildEngine(store); + expect((await cold.resume(parked.runId!)).success).toBe(true); + await flush(); + + // …and the run now reports terminal on both surfaces, not "paused + // forever" (#3456) off a suspension row that was just consumed. + expect((await cold.getRun(parked.runId!))!.status).toBe('completed'); + expect(await cold.listRuns('approval_flow', { status: 'paused' })).toEqual([]); + expect((await cold.listRuns('approval_flow')).map(r => r.status)).toEqual(['completed']); + }); + + it('GREEN: a genuinely unknown run id is still not found (the 404 envelope is unchanged)', async () => { + const store = new InMemorySuspendedRunStore(); + await buildEngine(store).execute('approval_flow', TRIGGER); + + // The precondition of `deps.error('Execution not found', 404)` in + // `packages/runtime/src/domains/automation.ts`. The new fallback must + // not invent an entry for an id the store has never heard of — a store + // read that answers `null` is what keeps the refusal envelope intact. + expect(await buildEngine(store).getRun('run_does_not_exist')).toBeNull(); + }); +}); + +describe('#8050 merging a third source: no duplicates, and a stated precedence', () => { + it('GREEN: pre-restart the run is in BOTH the ring and the store, and enumerates once', async () => { + const store = new InMemorySuspendedRunStore(); + const engine = buildEngine(store); + const parked = await engine.execute('approval_flow', TRIGGER); + + // Both sources genuinely hold it — otherwise this pin proves nothing. + expect(await store.load(parked.runId!)).not.toBeNull(); + + const rows = await engine.listRuns('approval_flow'); + expect(rows).toHaveLength(1); + expect(rows[0].id).toBe(parked.runId); + // The RING copy wins, and it is the richer one: `durationMs` is the + // time the run spent executing before it parked, which a suspension row + // does not record (it stores only `started_at` / `start_time`). + expect(rows[0].durationMs).toEqual(expect.any(Number)); + expect(await engine.getRun(parked.runId!)).toEqual(rows[0]); + }); + + it('GREEN: a STALE paused row cannot mask a finished run (ring side)', async () => { + // The realistic shape: the run completes, but the best-effort delete of + // its suspension row is lost to a store outage. The ring says + // completed, the store still says paused. + const store = new InMemorySuspendedRunStore(); + const engine = buildEngine(store); + const parked = await engine.execute('approval_flow', TRIGGER); + const runId = parked.runId!; + const orphan = (await store.load(runId))!; + + expect((await engine.resume(runId)).success).toBe(true); + await flush(); + await store.save(orphan); // the delete that "failed" + + const rows = await engine.listRuns('approval_flow'); + expect(rows).toHaveLength(1); + expect(rows[0].status).toBe('completed'); + expect((await engine.getRun(runId))!.status).toBe('completed'); + // …and it is not double-counted into the paused view either. + expect(await engine.listRuns('approval_flow', { status: 'paused' })).toEqual([]); + }); + + it('GREEN: a STALE paused row cannot mask a finished run (durable side, cold)', async () => { + // Same orphan, but read by a process whose ring is empty — so the + // precedence is decided purely between the two DURABLE families. The + // terminal history row is later evidence and must win. + const store = new InMemorySuspendedRunStore(); + const engine = buildEngine(store); + const runId = (await engine.execute('approval_flow', TRIGGER)).runId!; + const orphan = (await store.load(runId))!; + expect((await engine.resume(runId)).success).toBe(true); + await flush(); + await store.save(orphan); + + const cold = buildEngine(store); + const rows = await cold.listRuns('approval_flow'); + expect(rows).toHaveLength(1); + expect(rows[0].status).toBe('completed'); + expect((await cold.getRun(runId))!.status).toBe('completed'); + expect(await cold.listRuns('approval_flow', { status: 'paused' })).toEqual([]); + }); + + it('RED: cold restart — paused rows do not leak across flows', async () => { + // Labelled red after measurement: on `main` this fails at `[]`, because + // it has to see the row before it can check whose row it is. The + // invariant it adds on top of the enumeration is the scoping one — + // `store.list()` has NO flow slot (it returns every paused row in the + // deployment), so narrowing is this method's job, and dropping the + // filter would show one flow's in-flight runs under another. The + // pre-restart twin below is the half that guards this on `main` too. + const store = new InMemorySuspendedRunStore(); + const engine = buildEngine(store, ['approval_flow', 'other_flow']); + const mine = await engine.execute('approval_flow', TRIGGER); + await engine.execute('other_flow', TRIGGER); + + const cold = buildEngine(store, ['approval_flow', 'other_flow']); + expect((await cold.listRuns('approval_flow')).map(r => r.id)).toEqual([mine.runId]); + }); + + it('GREEN: pre-restart — another flow\'s paused run does not leak in', async () => { + // The genuinely-green scoping guard: the ring is filtered by flow name + // on `main` too, so this passes on both trees and stays a real pin on + // the narrowing rather than a vacuous one. + const store = new InMemorySuspendedRunStore(); + const engine = buildEngine(store, ['approval_flow', 'other_flow']); + const mine = await engine.execute('approval_flow', TRIGGER); + await engine.execute('other_flow', TRIGGER); + + expect((await engine.listRuns('approval_flow')).map(r => r.id)).toEqual([mine.runId]); + expect(await engine.listRuns('approval_flow', { status: 'paused' })).toHaveLength(1); + }); + + it('RED: cold restart — ?status= still narrows, and the new arm does not widen it', async () => { + // Also red on `main`, for the same reason: the `paused` clause is the + // new behaviour. The two clauses above it are the anti-widening pin — + // a `paused` row must never be returned to a caller asking for a + // terminal status, which is how a third source silently defeats #7359 + // in the other direction. + const store = new InMemorySuspendedRunStore(); + const engine = buildEngine(store); + await engine.execute('approval_flow', TRIGGER); + + const cold = buildEngine(store); + expect(await cold.listRuns('approval_flow', { status: 'completed' })).toEqual([]); + expect(await cold.listRuns('approval_flow', { status: 'failed' })).toEqual([]); + expect(await cold.listRuns('approval_flow', { status: 'paused' })).toHaveLength(1); + }); + + it('RED: an unreadable paused store DEGRADES the listing — it never throws', async () => { + // Red on `main` only because the warning it looks for is part of the + // new arm — there is nothing on `main` to degrade. What it pins is that + // the new read is best-effort, exactly like the history arm beside it: + // a store outage must not turn the Runs view into a 500, and the + // shortfall must be said out loud rather than read as "nothing + // pending". The run itself stays parked and resumable through a + // different door (`loadSuspendedRun`), unaffected by this failure. + const store = new InMemorySuspendedRunStore(); + const engine = buildEngine(store); + await engine.execute('approval_flow', TRIGGER); + + const warnings: string[] = []; + const noisy = { + info() {}, error() {}, debug() {}, + warn(m: unknown) { warnings.push(String(m)); }, + child() { return noisy; }, + } as never; + // Delegating wrapper rather than a spread: `store` is a class instance, + // so its methods live on the prototype and a spread would copy none of + // them — the other arms of the merge would then fail for the wrong + // reason and the pin would pass without measuring anything. + const unreadable: SuspendedRunStore = { + save: (run) => store.save(run), + load: (id) => store.load(id), + delete: (id) => store.delete(id), + listHistory: (flow, n) => store.listHistory(flow, n), + loadTerminal: (id) => store.loadTerminal(id), + recordTerminal: (rec) => store.recordTerminal(rec), + list: async () => { throw new Error('sqlite: database is locked'); }, + }; + const cold = new AutomationEngine(noisy, unreadable); + cold.registerFlow('approval_flow', pausingFlow('approval_flow') as never); + + await expect(cold.listRuns('approval_flow', { status: 'paused' })).resolves.toEqual([]); + expect(warnings.join('\n')).toContain('paused-run read failed'); + }); + + it('GREEN: a store with no paused rows behaves exactly as before', async () => { + // The pre-#8050 path, unchanged: terminal history alone still backs the + // post-restart Runs view. + const store = new InMemorySuspendedRunStore(); + const engine = new AutomationEngine(silent, store); + engine.registerFlow('plain', { + name: 'plain', label: 'plain', type: 'autolaunched', + nodes: [{ id: 'start', type: 'start', label: 's' }, { id: 'end', type: 'end', label: 'e' }], + edges: [{ id: 'e1', source: 'start', target: 'end' }], + } as never); + expect((await engine.execute('plain', TRIGGER)).success).toBe(true); + await flush(); + + const cold = new AutomationEngine(silent, store); + const rows = await cold.listRuns('plain'); + expect(rows.map(r => r.status)).toEqual(['completed']); + expect(await cold.listRuns('plain', { status: 'paused' })).toEqual([]); + }); +}); + +/** + * The card's literal reproduction, minus the HTTP hop: a real `sys_automation_run` + * table in a real sqlite FILE, written by {@link ObjectStoreSuspendedRunStore} + * through ObjectQL, then read by a second kernel booted over the same file. + * + * The engine-level blocks above already pin the logic; this one pins that the + * fix survives the layer they stub — the serialize/deserialize boundary and the + * `where: { status: 'paused' }` scan the DB-backed store actually issues. That + * is where "works against a Map, empty against a table" would hide. + */ +describe('#8050 cold boot over the same sqlite file (full stack)', () => { + let dir: string | undefined; + const kernels: ObjectKernel[] = []; + + afterEach(async () => { + for (const k of kernels.splice(0)) { + try { await k.shutdown(); } catch { /* noop */ } + } + if (dir) { rmSync(dir, { recursive: true, force: true }); dir = undefined; } + }); + + /** One process lifetime: kernel + ObjectQL + automation over `file`. */ + async function boot(file: string) { + const kernel = new ObjectKernel({ logger: { level: 'fatal' } }); + kernels.push(kernel); + await kernel.use(new ObjectQLPlugin()); + await kernel.use(new AutomationServicePlugin()); + await kernel.bootstrap(); + + const ql = kernel.getService('objectql'); + const driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: file }, + useNullAsDefault: true, + }); + await driver.connect(); + ql.registerDriver(driver, true); + await ql.syncSchemas(); + + const automation = kernel.getService('automation'); + automation.registerNodeExecutor(holdExecutor); + automation.registerNodeExecutor({ type: 'noop', async execute() { return { success: true }; } } as never); + automation.registerFlow('showcase_budget_approval', pausingFlow('showcase_budget_approval') as never); + return { kernel, ql, automation }; + } + + it('a run parked before the restart is enumerable, readable and resumable after it', async () => { + dir = mkdtempSync(join(tmpdir(), 'os-8050-')); + const file = join(dir, 'data.db'); + + // ── process 1: park a run ──────────────────────────────────────────── + const first = await boot(file); + const parked = await first.automation.execute('showcase_budget_approval', TRIGGER); + expect(parked.status).toBe('paused'); + const runId = parked.runId!; + await first.kernel.shutdown(); + kernels.splice(kernels.indexOf(first.kernel), 1); + + // ── process 2: cold boot over the SAME file ───────────────────────── + const second = await boot(file); + + // The row survived — the card's control reading, over the data API. + // (If this ever fails, the defect is durability, not observability, and + // it is a heavier card than #8050.) + const row = await second.ql.findOne('sys_automation_run', { + where: { id: runId }, context: { isSystem: true } as never, + }); + expect(row, 'the suspension row must survive the restart').toBeTruthy(); + expect(row.status).toBe('paused'); + + // RED on main: `[]` for both listings, `null` for the detail. + const listed = await second.automation.listRuns('showcase_budget_approval', { status: 'paused' }); + expect(listed.map(r => r.id)).toEqual([runId]); + expect((await second.automation.listRuns('showcase_budget_approval')).map(r => r.id)).toEqual([runId]); + + const detail = await second.automation.getRun(runId); + expect(detail).not.toBeNull(); + expect(detail!.status).toBe('paused'); + expect(detail).toEqual(listed[0]); + + // GREEN: still not found for an id that never existed. + expect(await second.automation.getRun('run_never_existed')).toBeNull(); + + // GREEN: and the durability the filer measured is intact — the run + // resumes to completion, and both surfaces then say so. + expect((await second.automation.resume(runId)).success).toBe(true); + await flush(); + expect((await second.automation.getRun(runId))!.status).toBe('completed'); + expect(await second.automation.listRuns('showcase_budget_approval', { status: 'paused' })).toEqual([]); + }); +}); + +/** + * Type-level anchor: the two rehydration paths this card touches consume the + * store's own contract types, so a future change to either row family has to + * come past the compiler here rather than past a cast in the engine. + */ +export type _PausedRowSources = [SuspendedRun, RunRecord]; diff --git a/packages/services/service-automation/src/suspended-run-store.test.ts b/packages/services/service-automation/src/suspended-run-store.test.ts index 37d7a947f0..d56e50768f 100644 --- a/packages/services/service-automation/src/suspended-run-store.test.ts +++ b/packages/services/service-automation/src/suspended-run-store.test.ts @@ -311,11 +311,16 @@ describe('hasSuspendedRun', () => { const paused = await pausableEngine(new ObjectStoreSuspendedRunStore(table, createTestLogger())) .execute('approval_flow'); - // The case `getRun` cannot answer: no execution-log entry exists in - // this process, yet the run is alive and resumable. + // No execution-log entry exists in this process, yet the run is alive + // and resumable — and both reads now say so. `getRun` used to answer + // `null` here, and this line asserted that as the CONTRAST between the + // two methods; #8050 measured the same `null` as the defect (it is what + // made run-detail 404 for a healthy parked run) and gave `getRun` the + // durable paused fallback `hasSuspendedRun` always had. The assertion is + // inverted rather than deleted, so the pair stays pinned together. const cold = pausableEngine(new ObjectStoreSuspendedRunStore(table, createTestLogger())); expect(await cold.hasSuspendedRun(paused.runId!)).toBe(true); - expect(await cold.getRun(paused.runId!)).toBeNull(); + expect((await cold.getRun(paused.runId!))?.status).toBe('paused'); }); it('throws rather than answering false when the store is unreadable', async () => { @@ -326,6 +331,14 @@ describe('hasSuspendedRun', () => { // "Unknown" must not collapse into "gone" — a caller that treats an // outage as a dead run rejects every decision in the tenant. await expect(e.hasSuspendedRun('run_x')).rejects.toThrow(/connection refused/); + + // The distinction between the two reads that #8050 did NOT erase, and + // the one that carries the safety property: this method backs a WRITE + // decision, so an outage must read as "unknown"; `getRun` is an + // observability read and still degrades to null with a warning. Now + // that they agree on a healthy store, the place they must keep + // disagreeing is worth its own assertion. + await expect(e.getRun('run_x')).resolves.toBeNull(); }); it('answers false with no store and nothing in memory', async () => {