diff --git a/.changeset/stranded-run-state-discriminator.md b/.changeset/stranded-run-state-discriminator.md new file mode 100644 index 0000000000..4dbfb8ed1f --- /dev/null +++ b/.changeset/stranded-run-state-discriminator.md @@ -0,0 +1,17 @@ +--- +"@objectstack/service-automation": minor +"@objectstack/plugin-approvals": minor +--- + +The stranded-request inspection tells a repairable strand from a cascade-failed run — through a dedicated read-only engine member, not through the wire (#15358, ruling B′). + +`ApprovalService.inspectStrandedRequests` keyed on `run.status === 'failed'`, which over-reports in one direction: a **cascade-failed** run — an ancestor `failAncestors` failed while it was parked at its `subflow` node, whose pause `failSuspendedRun` consumed and journalled nothing — has the same terminal `failed` row as the #13909 strand, so both came back `runState: 'failed'`, and `restoreConsumedSuspension` re-arms one and refuses the other (`NO_CONSUMED_SUSPENSION`). The engine's discriminator (the consumed-suspension snapshot on the durable `RunRecord`) is deliberately NOT on the `ExecutionLogEntry` that `getRun` answers, because `GET /automation/:name/runs/:runId` serves that object verbatim — so the plugin could not read it, and reading its absence as "not a strand" would have called the repairable row dead. + +**`@objectstack/service-automation` — additive, `minor`.** `AutomationEngine.inspectConsumedSuspension(runId)` answers whether `restoreConsumedSuspension` would have a consumed suspension to put back, from the SAME two witnesses that verb reads (this process's hot journal and the durable row, reconciled by the same `rowSupersedesJournal` / `persisted` / drop-notice rules — the read is now one private method both call), and re-arms nothing. Four answers, none folded: `repairable: true` (with the pause it would re-arm and which witness answered); `SNAPSHOT_DROPPED` (the strand happened, the store could not persist the snapshot, and this process holds no hot copy — repairable only by the replica that stranded it, while it lives); `NO_CONSUMED_SUSPENSION` (cascade-failed or never paused); `RUN_SUSPENDED` (already resumable). It REJECTS when a store cannot be read — an outage is unknown, not "nothing to restore". The result type is exported as `ConsumedSuspensionInspection`. `restoreConsumedSuspension` behaves exactly as before; nothing on `ExecutionLogEntry`, the run-detail route, or `@objectstack/spec` changes. + +**`@objectstack/plugin-approvals` — additive on two published types, `minor`.** + +- `ApprovalResumeSurface` gains the optional `inspectConsumedSuspension?(runId)`, declared the way `listSuspendedRunsDurable` is: a method `AutomationEngine` already implements, widening no engine surface. +- `StrandedRunState` splits `'failed'` three ways and keeps `'missing'` untouched: `'repairable'` (the #13909 strand — restore, then `continueRestoredRun`), `'snapshot_dropped'` (its own class: as `repairable` it over-reports, as `unrepairable` it is a false negative), and `'unrepairable'` (the cascade-failed / never-paused run, #15222's shape — nothing re-arms it). **`'failed'` stays a member, on purpose**: it is what a `failed` row is reported as when the attached surface has no `inspectConsumedSuspension` (an engine build older than this plugin, or a host double). Absence of the discriminator is fail-closed for a report — the row is reported, undifferentiated, never labelled `unrepairable` and never dropped. A thrown read counts `undetermined`, as the other two oracles' do. + +A consumer switching exhaustively over `StrandedRunState` gains three arms; nothing it matched before stops arriving. The inspection's summary log adds `runRepairable` / `runSnapshotDropped` / `runUnrepairable` beside the existing counts. diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index 141b630052..f65d0daa69 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -182,8 +182,47 @@ export interface ApprovalResumeSurface { * pause it cannot see is a pause this verb will not act on. */ listSuspendedRunsDurable?(): Promise>; + /** + * [#15358] Whether the consumed suspension behind a `failed` run SURVIVES — + * i.e. whether the engine's `restoreConsumedSuspension` would re-arm it. + * Read by {@link ApprovalService.inspectStrandedRequests}, for `failed` rows + * only, to tell the #13909 strand (repairable) from a cascade-failed + * ancestor — the `failAncestors` shape the engine itself calls NOT stranded, + * because `failSuspendedRun` consumed the ancestor's pause and journalled + * nothing, so no verb re-arms it (#15222's shape). {@link getRun} answers + * `status: 'failed'` for both: the discriminator is deliberately NOT on + * `ExecutionLogEntry`, which `GET /automation/:name/runs/:runId` serves + * verbatim. Ruled B′ on #15358 (2026-09-07): it is published as a dedicated + * read-only engine member, never on the wire. + * + * ⚠️ Declares a method `AutomationEngine` ALREADY implements publicly + * (`inspectConsumedSuspension`); it widens no engine surface. It answers + * from the same two witnesses the restore verb reads — this process's hot + * journal and the durable row — so what it calls repairable IS what that + * verb restores. The three negatives are distinct on purpose (see + * {@link StrandedRunState}); the middle one, `'SNAPSHOT_DROPPED'`, is a + * strand the store could not persist, repairable only by the process that + * stranded it while that process lives. + * + * Optional, and its absence is FAIL-CLOSED for a report: a `failed` row the + * engine cannot be asked about is reported as today's undifferentiated + * `'failed'` — ⛔ never as `'unrepairable'` (that calls the repairable row + * dead: the #15555 false-negative harm, one surface over) and ⛔ never + * skipped (that hides the row). Absence of the discriminator is not + * evidence of anything. Rejects when a store cannot be read; the inspection + * counts such a row `undetermined`, exactly as it does a thrown + * {@link hasSuspendedRun}. + */ + inspectConsumedSuspension?(runId: string): Promise< + | { repairable: true } + | { repairable: false; reason: 'RUN_SUSPENDED' | 'SNAPSHOT_DROPPED' | 'NO_CONSUMED_SUSPENSION' } + >; } +/** What {@link ApprovalResumeSurface.inspectConsumedSuspension} answers. */ +type ConsumedSuspensionVerdict = + Awaited>>; + /** * Optional messaging surface (ADR-0012 `messaging` service). When attached, * thread interactions (reassign / remind / request-info / comment) notify the @@ -349,7 +388,12 @@ function classifyStrandedRunState(run: { status?: string } | null | undefined): // No history row at all — the #4469 shape this inspection was built for. if (!run) return 'missing'; switch (run.status) { - // The resume consumed the pause and a downstream node threw. Reported. + // A terminal `failed` row. Reported — and, when the engine can be asked, + // refined by the third oracle ({@link refineFailedRunState}) into which of + // the three `failed` shapes it is. This arm alone cannot tell them apart: + // the row reads identically for a resume that consumed the pause and + // threw downstream (repairable) and for an ancestor `failAncestors` + // cascade-failed (not). `'failed'` here means "reported, undifferentiated". case 'failed': return 'failed'; @@ -377,26 +421,94 @@ function classifyStrandedRunState(run: { status?: string } | null | undefined): } /** - * WHY a terminal request's run is unrecoverable — the two shapes the inspection - * reports, which have different causes and different remedies (#13909). + * [#15358] The third oracle's verdict over a `failed` row — which of the three + * differentiated shapes it is, read from the engine's own consumed-suspension + * witnesses — or `undefined` for the one answer that means the row is not + * stranded after all. + * + * ⛔ Absence of the discriminator is NOT an input here, on purpose: this runs + * only when the engine answered. A surface without the member never reaches + * it and the row stays `'failed'` (see {@link StrandedRunState}) — reading + * "the engine could not be asked" as "not a strand" would call the repairable + * row dead, which is #15555's false negative one surface over. + */ +function refineFailedRunState(verdict: ConsumedSuspensionVerdict): StrandedRunState | undefined { + if (verdict.repairable) return 'repairable'; + switch (verdict.reason) { + // The strand happened; the store could not keep the snapshot, and the + // engine asked holds no hot copy. Its own class — see the type below. + case 'SNAPSHOT_DROPPED': + return 'snapshot_dropped'; + // Neither witness holds anything: cascade-failed, or never paused. + case 'NO_CONSUMED_SUSPENSION': + return 'unrepairable'; + // Re-armed between the two reads (an operator's restore landed while this + // scan was running): the run is alive and resumable, which is what the + // first oracle would have said a moment later. Not stranded. + case 'RUN_SUSPENDED': + return undefined; + // An answer this build does not know (an engine ahead of this plugin). + // Fail-closed exactly as an absent member: reported, undifferentiated — + // never condemned on a word this code cannot read. + default: + return 'failed'; + } +} + +/** + * WHY a terminal request's run is unrecoverable — the shapes the inspection + * reports, which have different causes and different remedies (#13909; + * split three ways by the #15358 B′ ruling, 2026-09-07). * * - `missing` — `getRun` finds no history row at all (#4469's original shape): * the run was lost before it could record anything, typically a pause that * never reached a durable store and did not survive a restart. - * - `failed` — the run DID record a terminal `failed` row. The engine consumes - * a suspension *before* running the downstream nodes - * (`AutomationEngine.resumeInternal`: `forgetSuspendedRun(run, 'resumed')` - * precedes `traverseNext`), so a downstream node that merely THREW threw with - * the pause already gone — the catch arm recorded `failed` and there is no - * suspension left to resume. The decision is durable, the flow stopped - * mid-continuation, and no verb moves the run out of that state. + * + * The other four all describe a run that DID record a terminal `failed` row. + * The engine consumes a suspension *before* running the downstream nodes + * (`AutomationEngine.resumeInternal`: `forgetSuspendedRun(run, 'resumed')` + * precedes `traverseNext`), so a downstream node that merely THREW threw with + * the pause already gone — the catch arm recorded `failed` and there is no + * suspension left to resume. The decision is durable and the flow stopped + * mid-continuation. What differs is whether anything can put the pause back, + * and `status` alone cannot say — the three differentiated members come from + * the engine's own consumed-suspension witnesses + * ({@link ApprovalResumeSurface.inspectConsumedSuspension}): + * + * - `repairable` — the engine holds the consumed suspension (in its journal + * or on the durable row): the #13909 strand. `restoreConsumedSuspension` + * re-arms it and {@link ApprovalService.continueRestoredRun} re-issues the + * decision. The remedy is a repair. + * - `snapshot_dropped` — the run DID strand, but the store could not persist + * the snapshot (over its row budget) and the engine asked holds no hot + * copy. Repairable only by the process that stranded it, while that process + * lives; the restore verb refuses it elsewhere naming the budget. ⛔ Not + * folded into either neighbour: as `repairable` it over-reports, as + * `unrepairable` it is #15555's false negative. Reported as what it is. + * - `unrepairable` — the engine holds NO consumed suspension: the run was + * cascade-failed (`failAncestors` → `failSuspendedRun`, which consumes the + * ancestor's pause and journals nothing — #15222's shape, the one the + * engine's own words call not a strand), never paused at all, or did + * strand and its snapshot is no longer held (the journal evicted a copy + * whose write never landed; the run was restored and then finished; a + * store class without `loadTerminal`, after a restart) — the engine's + * `NO_CONSUMED_SUSPENSION` covers all three and does not say which. The + * label is faithful to the verb: nothing re-arms it; the remedy is a new + * run, not a restore. + * - `failed` — the engine COULD NOT BE ASKED which of the three it is: the + * attached surface has no `inspectConsumedSuspension` (an engine build + * older than this plugin, or a test double). Today's undifferentiated + * label, kept on purpose as the fail-closed fallback (#15358 ruling, item + * 1): absence of the discriminator is not evidence, so the row is reported + * and its repairability left unstated — ⛔ never `unrepairable`, ⛔ never + * dropped from the report. * * ⚠️ This names the shapes for the REPORT only. It is not a run state: the * engine's own vocabulary is still `'completed' | 'paused' | 'failed'` * (`AutomationResult.status`) and nothing persists or queries "stranded". * Giving the condition a platform-level name is #13909's own deliverable. */ -export type StrandedRunState = 'missing' | 'failed'; +export type StrandedRunState = 'missing' | 'failed' | 'repairable' | 'snapshot_dropped' | 'unrepairable'; /** * The continuation an approvals door already issued once, kept so it can be @@ -464,14 +576,18 @@ export interface StrandedApprovalRequest { /** * The `flow_run_id` that resolves to no live suspension and no recoverable * run — see `runState`: no history row at all (`missing`), or a terminal - * `failed` row (`failed`). + * `failed` row (`repairable` / `snapshot_dropped` / `unrepairable`, or + * `failed` when the engine could not say which). */ runId: string; /** * Which unrecoverable shape this is — see {@link StrandedRunState}. Carried - * because the two need different remedies: a `missing` run has no history to - * read, while a `failed` one has a step log and an error message naming the - * node that threw. + * because the shapes need different remedies: a `missing` run has no history + * to read; a `repairable` one is put back by `restoreConsumedSuspension` and + * continued by {@link ApprovalService.continueRestoredRun}; an + * `unrepairable` one (#15222's cascade-failed ancestor) is refused by that + * verb and needs a new run. An operator reading `'failed'` has to ask the + * engine directly. */ runState: StrandedRunState; flowName?: string; @@ -4263,8 +4379,22 @@ export class ApprovalService implements IApprovalService { * ⚠️ The widening does NOT reverse the conservatism: `completed`, `cancelled` * and `paused` are each still skipped, for reasons named one at a time in * `classifyStrandedRunState`, and an unrecognised status is skipped too. - * What the widening buys is that a `failed` run is now reported with - * `runState: 'failed'` instead of counted as healthy. + * What the widening buys is that a `failed` run is now reported instead of + * counted as healthy. + * + * **A THIRD oracle tells the `failed` rows apart (#15358).** `status === + * 'failed'` over-reports in one specific direction: a cascade-failed run + * — an ancestor `failAncestors` failed while it was parked at its `subflow` + * node, whose pause `failSuspendedRun` consumed and journalled nothing — has + * the same terminal row as the #13909 strand, and `restoreConsumedSuspension` + * refuses it. The engine's discriminator (the consumed-suspension snapshot) + * is deliberately NOT on the object `getRun` answers, so it is asked through + * a dedicated read-only member, `inspectConsumedSuspension`, and only for + * `failed` rows: the answer splits `'failed'` into `'repairable'`, + * `'snapshot_dropped'` and `'unrepairable'` (see {@link StrandedRunState}). + * A surface without that member leaves the row `'failed'` — reported, + * undifferentiated — because absence of the discriminator is not evidence + * of anything; a thrown read counts `undetermined`, like the other two. * * ⚠️ **What this can and cannot size.** It makes the condition *visible* in a * deployment; it is not itself a census, and it says nothing about this @@ -4339,12 +4469,36 @@ export class ApprovalService implements IApprovalService { // #13909 — the widened verdict. `undefined` means "not a shape this // reports": healthy, deliberate, or unresolvable. See // `classifyStrandedRunState` for which, and why each one. - const runState = classifyStrandedRunState(terminal); + let runState = classifyStrandedRunState(terminal); if (!runState) continue; + // #15358 — the third oracle, for `failed` rows only: does the engine hold + // the consumed suspension a restore would put back? Asked through the + // dedicated read-only member, never inferred from the row. Absence of + // the member is FAIL-CLOSED for a report — the row stays `'failed'`, + // reported and undifferentiated; a thrown read is `undetermined`, as for + // the other two oracles. See `refineFailedRunState` and + // `StrandedRunState` for the three answers and why none is folded. + if (runState === 'failed' && typeof this.automation.inspectConsumedSuspension === 'function') { + let verdict: ConsumedSuspensionVerdict; + try { + verdict = await this.automation.inspectConsumedSuspension(runId); + } catch (err: any) { + undetermined++; + this.logger?.warn?.('[approvals] stranded-request scan could not read the consumed-suspension state', { + request: raw?.id, run: runId, error: err?.message ?? String(err), + }); + continue; + } + const refined = refineFailedRunState(verdict); + if (!refined) continue; // re-armed between the two reads — alive after all + runState = refined; + } + // Neither suspended nor recoverable: the run this decision was supposed to // advance is gone (`missing`) or terminally failed mid-continuation with - // its pause already consumed (`failed`). + // its pause already consumed (`repairable` / `snapshot_dropped` / + // `unrepairable` — or `failed`, when the engine could not say which). const config = parseJson( raw.node_config_json, { approvers: [], behavior: 'first_response' } as any, ); @@ -4383,6 +4537,9 @@ export class ApprovalService implements IApprovalService { scanned: rows.length, stranded: stranded.length, undetermined, runMissing: stranded.filter(s => s.runState === 'missing').length, runFailed: stranded.filter(s => s.runState === 'failed').length, + runRepairable: stranded.filter(s => s.runState === 'repairable').length, + runSnapshotDropped: stranded.filter(s => s.runState === 'snapshot_dropped').length, + runUnrepairable: stranded.filter(s => s.runState === 'unrepairable').length, requests: stranded.map(s => `${s.requestId}@${s.nodeId ?? '?'} → run ${s.runId} (${s.runState})`), }); } diff --git a/packages/plugins/plugin-approvals/src/stranded-request-inspection.test.ts b/packages/plugins/plugin-approvals/src/stranded-request-inspection.test.ts index a975886e0f..d7780b46cf 100644 --- a/packages/plugins/plugin-approvals/src/stranded-request-inspection.test.ts +++ b/packages/plugins/plugin-approvals/src/stranded-request-inspection.test.ts @@ -97,14 +97,29 @@ function requestRow(over: Record = {}): FakeRow { }; } -/** An automation surface with both oracles, each independently steerable. */ +/** What the #15358 third oracle answers, per run — see `ApprovalResumeSurface`. */ +type Verdict = + | { repairable: true } + | { repairable: false; reason: 'RUN_SUSPENDED' | 'SNAPSHOT_DROPPED' | 'NO_CONSUMED_SUSPENSION' }; + +/** + * An automation surface with both oracles, each independently steerable — and, + * ONLY when `repairability` / `repairabilityThrows` is given, the #15358 third + * oracle (`inspectConsumedSuspension`). Its absence by default is deliberate: + * every test above the #15358 block drives a surface that cannot be asked, + * which is exactly the population the undifferentiated `'failed'` is kept for. + */ function automation(opts: { suspended?: Record; suspendedThrows?: boolean; history?: Record; historyThrows?: boolean; + repairability?: Record; + repairabilityThrows?: boolean; } = {}) { - return { + const inspectCalls: string[] = []; + const surface: any = { + inspectCalls, async resume() { return { success: true }; }, async hasSuspendedRun(runId: string) { if (opts.suspendedThrows) throw new Error('suspended-run store unreadable'); @@ -114,7 +129,17 @@ function automation(opts: { if (opts.historyThrows) throw new Error('run history unreadable'); return opts.history?.[runId] ?? null; }, - } as any; + }; + if (opts.repairability !== undefined || opts.repairabilityThrows) { + surface.inspectConsumedSuspension = async (runId: string): Promise => { + inspectCalls.push(runId); + if (opts.repairabilityThrows) throw new Error('run history unreadable for the consumed suspension'); + const v = opts.repairability?.[runId]; + if (!v) throw new Error(`test surface: no verdict scripted for ${runId}`); + return v; + }; + } + return surface; } describe('stranded terminal request inspection (#4469)', () => { @@ -405,3 +430,162 @@ describe('stranded inspection sees a run that FAILED mid-resume (#13909)', () => expect(engine._tables['sys_approval_action'] ?? []).toHaveLength(0); }); }); + +// ── #15358: the third oracle tells the `failed` rows apart ───────────────── +// +// `status === 'failed'` over-reports in one direction: a cascade-failed run +// (an ancestor `failAncestors` failed while parked at its `subflow` node — +// `failSuspendedRun` consumed its pause and journalled nothing, so nothing +// re-arms it; #15222) has the same terminal row as the #13909 strand that +// `restoreConsumedSuspension` repairs. The engine publishes the difference as +// a dedicated read-only member (ruling B′, 2026-09-07), never on the object +// `getRun` answers. This block pins the plugin's side of that contract on a +// scripted surface; `stranded-run-repairability.test.ts` drives the real +// engine through both shapes. + +describe('#15358 — the third oracle splits `failed` three ways, and its ABSENCE is fail-closed', () => { + let engine: ReturnType; + let svc: ApprovalService; + + beforeEach(() => { + engine = makeFakeEngine(); + svc = new ApprovalService({ engine: engine as any }); + engine._tables['sys_approval_request'] = [requestRow()]; + }); + + const failedRun = { history: { run_1: { status: 'failed' as const } } }; + + it('⭐ a surface WITHOUT the member reports the row `failed` — never `unrepairable`, never skipped', async () => { + // Absence of the discriminator is not evidence of anything. On a real + // engine it is absent from `getRun` for BOTH the repairable strand and the + // cascade-failed ancestor, so reading absence as "not a strand" would call + // the repairable row dead — #15555's false negative, one surface over. + const auto = automation(failedRun); + expect(typeof auto.inspectConsumedSuspension).toBe('undefined'); + svc.attachAutomation(auto); + const out = await svc.inspectStrandedRequests(); + expect(out.undetermined).toBe(0); + expect(out.stranded.map(s => s.runState)).toEqual(['failed']); + }); + + it('`repairable: true` → `repairable`', async () => { + svc.attachAutomation(automation({ ...failedRun, repairability: { run_1: { repairable: true } } })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded.map(s => s.runState)).toEqual(['repairable']); + expect(out.undetermined).toBe(0); + }); + + it('`SNAPSHOT_DROPPED` → `snapshot_dropped` — its own class, folded into neither neighbour', async () => { + svc.attachAutomation(automation({ + ...failedRun, repairability: { run_1: { repairable: false, reason: 'SNAPSHOT_DROPPED' } }, + })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded.map(s => s.runState)).toEqual(['snapshot_dropped']); + }); + + it('`NO_CONSUMED_SUSPENSION` → `unrepairable` — the cascade-failed / never-paused shape', async () => { + svc.attachAutomation(automation({ + ...failedRun, repairability: { run_1: { repairable: false, reason: 'NO_CONSUMED_SUSPENSION' } }, + })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded.map(s => s.runState)).toEqual(['unrepairable']); + }); + + it('`RUN_SUSPENDED` → not stranded: re-armed between the two reads, the run is alive', async () => { + svc.attachAutomation(automation({ + ...failedRun, repairability: { run_1: { repairable: false, reason: 'RUN_SUSPENDED' } }, + })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded).toEqual([]); + // Skipped as alive — NOT counted as unknown. + expect(out.undetermined).toBe(0); + }); + + it('a THROWN read is `undetermined`, exactly like the other two oracles — never a verdict', async () => { + svc.attachAutomation(automation({ ...failedRun, repairabilityThrows: true })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded).toEqual([]); + expect(out.undetermined).toBe(1); + }); + + it('an answer this build does not know stays `failed` — reported, undifferentiated', async () => { + // An engine ahead of this plugin. Fail-closed exactly as an absent member: + // a word this code cannot read condemns nothing. + svc.attachAutomation(automation({ + ...failedRun, + repairability: { run_1: { repairable: false, reason: 'SOMETHING_NEWER' as any } }, + })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded.map(s => s.runState)).toEqual(['failed']); + expect(out.undetermined).toBe(0); + }); + + it('is asked for `failed` rows ONLY — with the failed row as the positive control', async () => { + engine._tables['sys_approval_request'] = [ + requestRow({ id: 'areq_missing', flow_run_id: 'run_missing' }), + requestRow({ id: 'areq_failed', flow_run_id: 'run_failed' }), + requestRow({ id: 'areq_done', flow_run_id: 'run_done' }), + requestRow({ id: 'areq_cancelled', flow_run_id: 'run_cancelled' }), + requestRow({ id: 'areq_parked', flow_run_id: 'run_parked' }), + ]; + const auto = automation({ + suspended: { run_parked: true }, + history: { + run_failed: { status: 'failed' }, + run_done: { status: 'completed' }, + run_cancelled: { status: 'cancelled' }, + run_parked: { status: 'failed' }, + }, + repairability: { run_failed: { repairable: false, reason: 'NO_CONSUMED_SUSPENSION' } }, + }); + svc.attachAutomation(auto); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded.map(s => [s.requestId, s.runState])).toEqual([ + ['areq_missing', 'missing'], + ['areq_failed', 'unrepairable'], + ]); + // Exactly one read, for exactly the failed-and-not-suspended row: a + // `missing` run has nothing to ask about, a finished or cancelled run is + // not reported at all, and a parked run never reaches the second oracle. + expect(auto.inspectCalls).toEqual(['run_failed']); + }); + + it('one mixed population, every label distinct — nothing folded', async () => { + engine._tables['sys_approval_request'] = [ + requestRow({ id: 'areq_missing', flow_run_id: 'run_missing' }), + requestRow({ id: 'areq_repairable', flow_run_id: 'run_repairable' }), + requestRow({ id: 'areq_dropped', flow_run_id: 'run_dropped' }), + requestRow({ id: 'areq_cascade', flow_run_id: 'run_cascade' }), + ]; + svc.attachAutomation(automation({ + history: { + run_repairable: { status: 'failed' }, + run_dropped: { status: 'failed' }, + run_cascade: { status: 'failed' }, + }, + repairability: { + run_repairable: { repairable: true }, + run_dropped: { repairable: false, reason: 'SNAPSHOT_DROPPED' }, + run_cascade: { repairable: false, reason: 'NO_CONSUMED_SUSPENSION' }, + }, + })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded.map(s => [s.requestId, s.runState])).toEqual([ + ['areq_missing', 'missing'], + ['areq_repairable', 'repairable'], + ['areq_dropped', 'snapshot_dropped'], + ['areq_cascade', 'unrepairable'], + ]); + expect(out.undetermined).toBe(0); + }); + + it('still NEVER rewrites anything — the third oracle is a read like the other two', async () => { + engine._tables['opportunity'] = [{ id: 'opp1', approval_status: 'pending' }]; + svc.attachAutomation(automation({ + ...failedRun, repairability: { run_1: { repairable: false, reason: 'NO_CONSUMED_SUSPENSION' } }, + })); + const before = JSON.stringify(engine._tables); + await svc.inspectStrandedRequests(); + expect(JSON.stringify(engine._tables)).toBe(before); + }); +}); diff --git a/packages/plugins/plugin-approvals/src/stranded-run-repairability.test.ts b/packages/plugins/plugin-approvals/src/stranded-run-repairability.test.ts index dc30845b1d..613c8f6b29 100644 --- a/packages/plugins/plugin-approvals/src/stranded-run-repairability.test.ts +++ b/packages/plugins/plugin-approvals/src/stranded-run-repairability.test.ts @@ -1,14 +1,18 @@ // Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. /** - * #15358 — THE REPRODUCTION: `inspectStrandedRequests` labels a repairable - * strand and an UNREPAIRABLE cascade-failed ancestor identically. + * #15358 — THE REPRODUCTION, and its resolution: `inspectStrandedRequests` + * labelled a repairable strand and an UNREPAIRABLE cascade-failed ancestor + * identically; under ruling B′ (2026-09-07) it tells them apart through the + * engine's dedicated read-only member, `inspectConsumedSuspension`. * * The card was filed from a reading of the sources. This file is the drive, - * against a real `AutomationEngine` and a real `ApprovalService`, and it - * reproduces: one fixture produces both rows, the inspection returns both as - * `runState: 'failed'`, and the repair verb an operator would reach for next - * answers `restored: true` for one and refuses the other. + * against a real `AutomationEngine` and a real `ApprovalService`: one fixture + * produces both rows, the repair verb an operator would reach for answers + * `restored: true` for one and refuses the other — and the inspection now + * reports them as `'repairable'` and `'unrepairable'` (PIN 1), from the SAME + * engine reading the verb uses (PIN 2), while `getRun` still carries no + * discriminator (PIN 3 — the wire surface is untouched, by ruling). * * ## The two rows, from one fixture * @@ -29,30 +33,29 @@ * cancelled, cascade-failed)". UNREPAIRABLE — and this row's decision did * advance its flow, which is the half the shared label denies. * - * ## What must turn these assertions RED + * ## What PIN 1 pins now, and the assertion it replaced * - * ⛔ The `runState` assertions below record what the inspection reports - * TODAY. The maintainer ruling on this card (2026-09-04, decision batch #36, - * option B) splits `StrandedRunState` into a repairable and an unrepairable - * member and reports both, labelled apart — so whatever lands for B MUST turn - * the "both come back identical" assertion red on purpose. It is written as a - * single `toEqual` over both rows for exactly that reason: a split that - * relabels only one of them still fails it. + * PIN 1 was written as a single `toEqual` over both rows recording what the + * inspection reported before B′ — `['failed', 'failed']` — so that a split + * relabelling only one of them would still fail it. B′ landed; the same + * single `toEqual` now records the split, so a regression that folds either + * row back into the other's label (or into bare `'failed'`) fails it. * - * ## The measurement that sent the card back to the decision box + * ## The measurement B′ was ruled on — kept as PIN 3 * - * `PIN 3` is the load-bearing one. B's first clause is "`getRun` widens to - * carry the discriminator the engine already records". The engine records it - * on the DURABLE `RunRecord`, and `AutomationEngine.getRun` answers an - * `ExecutionLogEntry`, which carries neither field — deliberately: `recordLog` - * says the snapshot is "a parameter rather than a field of - * `ExecutionLogEntry`" because that interface "is served verbatim by - * `GET /automation/:name/runs/:runId`". So widening the plugin-side - * declaration alone cannot separate these two rows: on a real engine the - * discriminator is absent for BOTH, and a classifier reading absence as "not a - * strand" would answer UNREPAIRABLE for the repairable row — the #15555 - * false-negative harm, one surface over. Where the discriminator gets - * published is a producer-side contract decision, and it is open. + * The first ruling (2026-09-04, batch #36, option B) said "`getRun` widens to + * carry the discriminator the engine already records". PIN 3 measured that it + * cannot: the engine records the discriminator on the DURABLE `RunRecord`, + * and `AutomationEngine.getRun` answers an `ExecutionLogEntry`, which carries + * neither field — deliberately, because that interface "is served verbatim by + * `GET /automation/:name/runs/:runId`". On a real engine the discriminator is + * absent for BOTH rows, so a classifier reading absence as "not a strand" + * would answer UNREPAIRABLE for the repairable row — the #15555 + * false-negative harm, one surface over. B′ (batch #76) therefore publishes it + * as a dedicated read-only engine member and leaves the wire untouched; PIN 3 + * now pins that the wire IS untouched, and PIN 5 pins the fail-closed half: a + * surface that lacks the member reports both rows `'failed'` — never + * `'unrepairable'`, never dropped. * * ## The control that makes the readings trustworthy * @@ -71,7 +74,7 @@ import { AutomationEngine, InMemorySuspendedRunStore, installBuiltinNodes } from // it stands in for is how #4434 shipped a dead REST route with its suite green. import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/objectql'; import { strandedDecisionDetails } from '@objectstack/types'; -import { ApprovalService } from './approval-service.js'; +import { ApprovalService, type ApprovalResumeSurface } from './approval-service.js'; import { registerApprovalNode } from './approval-node.js'; const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as any; @@ -284,7 +287,7 @@ describe('#15358 — a cascade-failed ancestor is reported as the repairable str return { parentReq, parentRunId, childReq, childRunId, childError: err }; } - it('PIN 1 — the two rows come back with the SAME `runState`, and one of them cannot be repaired', async () => { + it('PIN 1 — the two rows come back labelled APART: the strand `repairable`, the ancestor `unrepairable`', async () => { const automation = boot(); const { parentReq, parentRunId, childReq, childRunId } = await driveBothShapes(automation); @@ -297,11 +300,13 @@ describe('#15358 — a cascade-failed ancestor is reported as the repairable str const labelled = out.stranded .map(s => [s.requestId === parentReq.id ? 'cascade-failed ancestor' : 'genuine strand', s.runState]) .sort((a, b) => a[0].localeCompare(b[0])); - // ⛔ THE DEFECT, as one assertion: the label does not distinguish them. - // Option B splits `StrandedRunState`, so this MUST go red when B lands. + // The resolution, as one assertion over both rows: before B′ this read + // `['failed', 'failed']` (the defect — one label for two remedies). A fold + // of either row back into the other's label, or into bare `'failed'` on + // an engine that CAN be asked, fails it. expect(labelled).toEqual([ - ['cascade-failed ancestor', 'failed'], - ['genuine strand', 'failed'], + ['cascade-failed ancestor', 'unrepairable'], + ['genuine strand', 'repairable'], ]); // …and both are reported with their decision durable, which is the true @@ -383,4 +388,39 @@ describe('#15358 — a cascade-failed ancestor is reported as the repairable str // …while the strand behind it is every bit as repairable as door A's. expect((await automation.restoreConsumedSuspension(recalledRunId)).restored).toBe(true); }); + + it('PIN 5 — FAIL-CLOSED: a surface WITHOUT the member reports both rows `failed`, never `unrepairable`', async () => { + // The same real engine, the same two rows — seen through a surface that + // carries the two older oracles and not the third (an engine build older + // than this plugin, or a host double). Absence of the discriminator is not + // evidence: the repairable row must not be called dead (#15555's false + // negative, one surface over) and neither row may vanish from the report. + const automation = boot(); + const { parentReq, parentRunId, childRunId } = await driveBothShapes(automation); + const blind: ApprovalResumeSurface = { + hasSuspendedRun: (runId) => automation.hasSuspendedRun(runId), + getRun: (runId) => automation.getRun(runId), + }; + expect(typeof (blind as { inspectConsumedSuspension?: unknown }).inspectConsumedSuspension).toBe('undefined'); + service.attachAutomation(blind); + + const out = await service.inspectStrandedRequests(); + expect(out.scanned).toBe(2); + expect(out.undetermined).toBe(0); + const labelled = out.stranded + .map(s => [s.requestId === parentReq.id ? 'cascade-failed ancestor' : 'genuine strand', s.runState]) + .sort((a, b) => a[0].localeCompare(b[0])); + expect(labelled).toEqual([ + ['cascade-failed ancestor', 'failed'], + ['genuine strand', 'failed'], + ]); + expect(new Set(out.stranded.map(s => s.runId))).toEqual(new Set([parentRunId, childRunId])); + + // Positive control, same engine, same rows: re-attach the full surface and + // the split comes back — so the `'failed'` above was the member's absence, + // not the rows. + service.attachAutomation(automation); + const again = await service.inspectStrandedRequests(); + expect(new Set(again.stranded.map(s => s.runState))).toEqual(new Set(['repairable', 'unrepairable'])); + }); }); diff --git a/packages/services/service-automation/src/consumed-suspension-inspection.test.ts b/packages/services/service-automation/src/consumed-suspension-inspection.test.ts new file mode 100644 index 0000000000..2de047d7d9 --- /dev/null +++ b/packages/services/service-automation/src/consumed-suspension-inspection.test.ts @@ -0,0 +1,340 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #15358 — `inspectConsumedSuspension`: the READ-ONLY half of the operator + * exit, published as a dedicated engine member (ruling B′, 2026-09-07). + * + * ## Why this member exists + * + * `AutomationEngine.getRun` answers an `ExecutionLogEntry`, which carries + * neither `consumedSuspension` nor `consumedSuspensionDropped` — on purpose, + * because `GET /automation/:name/runs/:runId` serves that object verbatim. So + * a consumer reading `getRun` sees `status: 'failed'` for BOTH the #13909 + * strand (the resume consumed the pause and a downstream node threw — + * `restoreConsumedSuspension` re-arms it) and a cascade-failed ancestor + * (`failAncestors` → `failSuspendedRun`, which consumes the pause and journals + * nothing — nothing re-arms it; #15222). plugin-approvals' stranded-request + * inspection reported both as one label. The ruling: publish the answer as a + * read-only engine member, not on the wire. + * + * ## What is pinned + * + * 1. **Same reading as the restore verb, re-arming nothing.** The member + * answers from the two witnesses `restoreConsumedSuspension` reads (this + * process's hot journal, the durable row), and a `repairable: true` from + * it leaves the run exactly as it found it: still not suspended, and the + * restore verb still able to restore. A read that consumed the copy it + * read would be a second side door into shape 2. + * 2. **Every negative separately, each with its own reason** — and the + * middle one, `SNAPSHOT_DROPPED`, distinct from both neighbours. Folding + * it into "unrepairable" is #15555's false negative; folding it into + * "repairable" over-reports. ⛔ Not a single `loadTerminal` read: the drop + * notice is repairable from the hot copy on the very replica that + * stranded the run, and this file drives that replica AND a fresh one over + * the same row. + * 3. **Agreement, pinned as one fact stated twice**: for every shape driven + * here, `inspect(...).repairable === restore(...).restored`. The member is + * a prediction of the verb; a prediction the verb contradicts is worse + * than none. + * 4. **An unreadable store REJECTS** — never `NO_CONSUMED_SUSPENSION`. That + * answer is what a sweep would act on by giving up on a repairable run. + * 5. **Nameable from the barrel**, method and result type both. + */ + +import { describe, it, expect } from 'vitest'; + +import { + AutomationEngine, + type ConsumedSuspensionDropNotice, + type RunRecord, + type SuspendedRunStore, +} from './engine.js'; +// Barrel imports on purpose — the #13951 witness, for this member: the type +// half breaks at `tsc --noEmit`, the runtime half right here in vitest. +import { + AutomationEngine as BarrelAutomationEngine, + type ConsumedSuspensionInspection, +} from './index.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; + +const silent = { info() {}, warn() {}, error() {}, debug() {} } as never; + +const pauser = defineActionDescriptor({ + type: 'pause_here', version: '1.0.0', name: 'pause_here', + supportsPause: true, resumeAuthority: 'any', +}); +const plain = (type: string) => defineActionDescriptor({ type, version: '1.0.0', name: type }); + +/** start → pause (suspends) → after (the node that throws) → end. */ +const STRAND_FLOW = { + name: 'strand_flow', label: 'strand_flow', type: 'autolaunched', + variables: [{ name: 'ticket', type: 'text', isInput: true, isOutput: true }], + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'pause', type: 'pause_here', label: 'Pause' }, + { id: 'after', type: 'after_pause', label: 'After' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'pause' }, + { id: 'e2', source: 'pause', target: 'after' }, + { id: 'e3', source: 'after', target: 'end' }, + ], +}; + +/** A flow with no pause at all — the never-suspended shape. */ +const NO_PAUSE_FLOW = { + name: 'no_pause_flow', label: 'no_pause_flow', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'boom', type: 'always_throws', label: 'Boom' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'boom' }, + { id: 'e2', source: 'boom', target: 'end' }, + ], +}; + +const ctx = { event: 'test', record: { id: 'rec_1' }, params: { ticket: 'TKT-9' } } as unknown as AutomationContext; + +function newEngine(store?: SuspendedRunStore) { + const engine = new AutomationEngine(silent, store); + const state = { throws: true }; + engine.registerNodeExecutor({ + type: 'pause_here', descriptor: pauser, + async execute() { + return { success: true, suspend: true, correlation: 'approval:req_1', output: { stage: 'awaiting' } }; + }, + } as never); + engine.registerNodeExecutor({ + type: 'after_pause', descriptor: plain('after_pause'), + async execute() { + if (state.throws) throw new Error('downstream node blew up'); + return { success: true, output: { done: true } }; + }, + } as never); + engine.registerNodeExecutor({ + type: 'always_throws', descriptor: plain('always_throws'), + async execute() { throw new Error('never paused, just failed'); }, + } as never); + engine.registerFlow('strand_flow', STRAND_FLOW as never); + engine.registerFlow('no_pause_flow', NO_PAUSE_FLOW as never); + return { engine, state }; +} + +/** Drive a run into the stranded state. Returns its id. */ +async function strandRun(engine: AutomationEngine): Promise { + const started = await engine.execute('strand_flow', ctx); + expect(started.status).toBe('paused'); + const runId = started.runId as string; + const failed = await engine.resume(runId); + expect(failed.success).toBe(false); + expect(failed.status).toBe('stranded'); + expect(await engine.hasSuspendedRun(runId)).toBe(false); + return runId; +} + +/** + * A store that keeps every row but serves the terminal one the way the object + * store does when a snapshot is over its byte budget: no `consumedSuspension`, + * and a drop notice in its place (`stranded-run-object-store.test.ts` drives + * the real store into this; here the shape is enough). + */ +function droppingStore(inner: InMemorySuspendedRunStore, notice: Omit): SuspendedRunStore { + return { + save: (r) => inner.save(r), + load: (id) => inner.load(id), + delete: (id) => inner.delete(id), + list: () => inner.list(), + recordTerminal: (r) => inner.recordTerminal(r), + async loadTerminal(id) { + const row = await inner.loadTerminal(id); + if (!row?.consumedSuspension) return row; + const { consumedSuspension, ...rest } = row; + const dropped: RunRecord = { + ...rest, + consumedSuspensionDropped: { + ...notice, + nodeId: consumedSuspension.nodeId, + correlation: consumedSuspension.correlation, + }, + }; + return dropped; + }, + }; +} + +describe('#15358 — inspectConsumedSuspension: the read-only half of the exit', () => { + it('answers `repairable: true` for a strand, from this process\'s journal, and RE-ARMS NOTHING', async () => { + const { engine } = newEngine(undefined); + const runId = await strandRun(engine); + + const verdict = await engine.inspectConsumedSuspension(runId); + expect(verdict).toMatchObject({ + repairable: true, runId, flowName: 'strand_flow', nodeId: 'pause', + correlation: 'approval:req_1', witness: 'journal', + }); + expect(typeof (verdict as { consumedAt?: string }).consumedAt).toBe('string'); + + // Read-only, in both directions that matter: the run is no more + // resumable than before, and the copy the answer came from was not + // consumed by answering — the restore verb still finds it. + expect(await engine.hasSuspendedRun(runId)).toBe(false); + expect((await engine.resume(runId)).code).toBe('RUN_NOT_FOUND'); + const restored = await engine.restoreConsumedSuspension(runId); + expect(restored.restored).toBe(true); + expect(restored.nodeId).toBe('pause'); + }); + + it('answers `RUN_SUSPENDED` once the pause is back — the run is already resumable', async () => { + const { engine } = newEngine(new InMemorySuspendedRunStore()); + const runId = await strandRun(engine); + expect((await engine.restoreConsumedSuspension(runId)).restored).toBe(true); + + const verdict = await engine.inspectConsumedSuspension(runId); + expect(verdict).toEqual({ repairable: false, runId, reason: 'RUN_SUSPENDED', nodeId: 'pause' }); + }); + + it('answers `NO_CONSUMED_SUSPENSION` for a run that never paused — the "neither witness" shape', async () => { + const { engine } = newEngine(new InMemorySuspendedRunStore()); + const started = await engine.execute('no_pause_flow', ctx); + expect(started.success).toBe(false); + expect(started.status).toBe('failed'); + // A failed `execute` does not carry its run id on the result; the run + // log does (same derivation as the restore verb's own never-suspended pin). + const runId = (started.runId ?? (await engine.listRuns('no_pause_flow'))[0]?.id) as string; + expect(runId).toBeTruthy(); + + const verdict = await engine.inspectConsumedSuspension(runId); + expect(verdict).toEqual({ repairable: false, runId, reason: 'NO_CONSUMED_SUSPENSION' }); + // …and the restore verb agrees, for the same reason. + const res = await engine.restoreConsumedSuspension(runId); + expect(res.restored).toBe(false); + expect(res.refusal).toBe('NO_CONSUMED_SUSPENSION'); + }); + + it('answers from the DURABLE row after a restart, where no journal exists', async () => { + const store = new InMemorySuspendedRunStore(); + const { engine: a } = newEngine(store); + const runId = await strandRun(a); + + const { engine: b } = newEngine(store); + const verdict = await b.inspectConsumedSuspension(runId); + expect(verdict).toMatchObject({ repairable: true, runId, nodeId: 'pause', witness: 'durable' }); + // Still read-only across the restart: B has not re-armed it either. + expect(await b.hasSuspendedRun(runId)).toBe(false); + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + }); + + it('a run that was restored and then FINISHED is no longer repairable — the snapshot cleared with it', async () => { + const store = new InMemorySuspendedRunStore(); + const { engine, state } = newEngine(store); + const runId = await strandRun(engine); + expect((await engine.restoreConsumedSuspension(runId)).restored).toBe(true); + state.throws = false; + expect((await engine.resume(runId)).success).toBe(true); + + const fresh = new AutomationEngine(silent, store); + expect(await fresh.inspectConsumedSuspension(runId)).toEqual({ + repairable: false, runId, reason: 'NO_CONSUMED_SUSPENSION', + }); + }); +}); + +describe('#15358 — the dropped snapshot is its OWN answer, and which replica asks decides it', () => { + const notice = { bytes: 300 * 1024, budget: 256 * 1024 }; + + it('⭐ the replica that stranded the run still answers `repairable: true` from its hot copy', async () => { + const inner = new InMemorySuspendedRunStore(); + const store = droppingStore(inner, notice); + const { engine } = newEngine(store); + const runId = await strandRun(engine); + // The row really is snapshot-less and carries the notice. + const row = await store.loadTerminal!(runId); + expect(row?.consumedSuspension).toBeUndefined(); + expect(row?.consumedSuspensionDropped).toMatchObject({ ...notice, nodeId: 'pause' }); + + const verdict = await engine.inspectConsumedSuspension(runId); + expect(verdict).toMatchObject({ repairable: true, runId, nodeId: 'pause', witness: 'journal' }); + // ⛔ A naive single `loadTerminal` read would have answered + // SNAPSHOT_DROPPED here — on the one replica able to restore it. + expect((await engine.restoreConsumedSuspension(runId)).restored).toBe(true); + }); + + it('a fresh replica answers `SNAPSHOT_DROPPED` carrying the notice — neither "repairable" nor "never a strand"', async () => { + const inner = new InMemorySuspendedRunStore(); + const store = droppingStore(inner, notice); + const { engine: a } = newEngine(store); + const runId = await strandRun(a); + + const { engine: b } = newEngine(store); + const verdict = await b.inspectConsumedSuspension(runId); + expect(verdict).toEqual({ + repairable: false, runId, reason: 'SNAPSHOT_DROPPED', + dropped: { ...notice, nodeId: 'pause', correlation: 'approval:req_1' }, + }); + // The restore verb's refusal names the same budget: one fact, twice. + const res = await b.restoreConsumedSuspension(runId); + expect(res.restored).toBe(false); + expect(res.refusal).toBe('NO_CONSUMED_SUSPENSION'); + expect(res.reason).toContain(`${notice.budget}-byte row budget`); + }); +}); + +describe('#15358 — an unreadable store REJECTS; it never answers "nothing to restore"', () => { + it('when the suspended-run store cannot be read', async () => { + const inner = new InMemorySuspendedRunStore(); + const store: SuspendedRunStore = { + save: (r) => inner.save(r), + load: async () => { throw new Error('connection reset'); }, + delete: (id) => inner.delete(id), + list: () => inner.list(), + recordTerminal: (r) => inner.recordTerminal(r), + loadTerminal: (id) => inner.loadTerminal(id), + }; + const { engine } = newEngine(store); + await expect(engine.inspectConsumedSuspension('run_whatever')).rejects.toThrow('connection reset'); + }); + + it('when the run history cannot be read', async () => { + const inner = new InMemorySuspendedRunStore(); + const store: SuspendedRunStore = { + save: (r) => inner.save(r), + load: (id) => inner.load(id), + delete: (id) => inner.delete(id), + list: () => inner.list(), + recordTerminal: (r) => inner.recordTerminal(r), + loadTerminal: async () => { throw new Error('history table unreachable'); }, + }; + const { engine } = newEngine(store); + await expect(engine.inspectConsumedSuspension('run_elsewhere')).rejects.toThrow('history table unreachable'); + // Positive control: the same engine answers for a store that works. + const { engine: healthy } = newEngine(new InMemorySuspendedRunStore()); + await expect(healthy.inspectConsumedSuspension('run_elsewhere')).resolves.toMatchObject({ repairable: false }); + }); +}); + +describe('#15358 — nameable from the barrel', () => { + it('publishes the member on the same class the barrel exports, and its result type', async () => { + expect(BarrelAutomationEngine).toBe(AutomationEngine); + expect(typeof BarrelAutomationEngine.prototype.inspectConsumedSuspension).toBe('function'); + const { engine } = newEngine(new InMemorySuspendedRunStore()); + // The annotation is the point — the line a missing export makes unwritable. + const verdict: ConsumedSuspensionInspection = await engine.inspectConsumedSuspension('no-such-run'); + expect(verdict.repairable).toBe(false); + expect(remedyFor(verdict)).toBe('start a new run'); + }); +}); + +/** A consumer switching exhaustively over the three negatives. */ +function remedyFor(v: ConsumedSuspensionInspection): string { + if (v.repairable) return 'restoreConsumedSuspension, then re-issue the continuation'; + switch (v.reason) { + case 'RUN_SUSPENDED': return 'resume it'; + case 'SNAPSHOT_DROPPED': return 'restore from the replica that stranded it'; + case 'NO_CONSUMED_SUSPENSION': return 'start a new run'; + } +} diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 10ad655e47..45dbeba189 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1463,6 +1463,57 @@ export interface SuspensionRestoreResult { consumedAt?: string; } +/** + * [#15358] What {@link AutomationEngine.inspectConsumedSuspension} answers: + * would {@link AutomationEngine.restoreConsumedSuspension} have a consumed + * suspension to put back for this run? READ from the same two witnesses that + * verb reads, and re-arming nothing. + * + * Every arm is earned by its own observation, and the three negative ones are + * deliberately not folded into one boolean — a consumer that branches on + * `repairable` alone reads the middle one wrong in both directions: + * + * - `repairable: true` — a consumed-suspension snapshot is held for the run, + * in this process's hot journal or on the durable terminal row (`witness` + * says which). The restore verb re-arms it. + * - `'SNAPSHOT_DROPPED'` — the run DID strand, but the store could not + * persist the snapshot (over its row budget — the row says so, + * {@link RunRecord.consumedSuspensionDropped}) and THIS process holds no + * hot copy of it. Repairable only by the process that stranded it, while + * that process is still running; the restore verb refuses it here naming + * the budget. ⛔ Neither "unrepairable" nor "never a strand". + * - `'NO_CONSUMED_SUSPENSION'` — neither witness holds anything: the run + * reached a terminal state that was NOT a strand (completed, cancelled, + * cascade-failed — `failSuspendedRun` consumes an ancestor's pause and + * journals nothing), never paused at all, or DID strand and its snapshot + * is no longer held (the journal evicted a copy whose write never landed; + * the run was restored and then finished; a store class without + * `loadTerminal`, after a restart). Deliberately does not claim which: + * nothing in the engine can tell those apart, and the restore verb's own + * refusal text names the same three. + * - `'RUN_SUSPENDED'` — a live suspension exists, so the run is resumable + * already (typically: it was restored). Nothing to repair. + */ +export type ConsumedSuspensionInspection = + | { + repairable: true; + runId: string; + flowName: string; + /** + * The pause a restore would re-arm — its identity, the pair + * {@link SuspendedRunStore.claimSuspension} compares. + */ + nodeId: string; + correlation?: string; + /** Which of the two witnesses answered (see the restore verb's read). */ + witness: 'journal' | 'durable'; + /** When the resume that consumed the suspension failed. */ + consumedAt: string; + } + | { repairable: false; runId: string; reason: 'RUN_SUSPENDED'; nodeId: string } + | { repairable: false; runId: string; reason: 'SNAPSHOT_DROPPED'; dropped: ConsumedSuspensionDropNotice } + | { repairable: false; runId: string; reason: 'NO_CONSUMED_SUSPENSION' }; + /** * [#14333] Where a suspension was parked when the caller READ it — the * condition a {@link SuspendedRunStore.claimSuspension} compare-and-set is @@ -6587,6 +6638,102 @@ export class AutomationEngine implements IAutomationService { return hot.persisted === 'landed'; } + /** + * [#15358] The two-witness READ of a run's consumed suspension — this + * process's hot journal against the durable terminal row — factored out of + * {@link restoreConsumedSuspension} so that {@link inspectConsumedSuspension} + * answers from the same reading and the two cannot disagree. PURE: it + * mutates nothing. A hot copy the row proves stale is REPORTED + * (`staleHot`) for the restore verb to drop; the read-only verb leaves it. + * + * [#13937] Two witnesses of one strand, and neither is trusted alone — + * the contract review of that ruling's services half measured both + * single-witness readings wrong, one store class apart: + * + * - This process's HOT copy is the verbatim object the failure was + * journalled from: the pause's own node, variables, step log as of the + * pause. It is a per-process cache. The replica that stranded a run + * keeps it after another replica restored, resumed and FINISHED the + * run, and re-arming it then re-runs every node after the pause — + * shape 2's silent double-run, through the restore verb's side door + * (pinned in `stranded-run-status.test.ts`, red on the hot-only tree). + * - The DURABLE row is the record every replica can read, and a + * flattened, column-bounded copy of the same snapshot: the object + * store rebuilds it from columns, DROPS it over a byte budget — and + * says so in the row, `consumedSuspensionDropped` — and receives it + * fire-and-forget. Read alone, a snapshot-less row sent a run the + * store could not persist into NO_CONSUMED_SUSPENSION on the very + * replica holding its copy (pinned in `stranded-run-object-store.test.ts`, + * red on the durable-first tree). + * + * So: the hot copy is preferred whenever both describe the SAME pause + * ({@link rowSupersedesJournal}). When they describe different pauses, + * the newest strand wins — a hot copy whose own write never landed + * (`persisted` is not `'landed'`: the #13617 exception, a row the store + * was never handed says nothing) beats the older row, and a landed hot + * copy yields to the later strand another replica recorded. A row with + * neither a snapshot nor a drop notice is "the run moved on" only for a + * hot copy whose write did land — that copy is then stale rather than + * honoured. A hot copy answers alone where there is no row to ask: no + * store, no run history, a write that never landed or is still in flight. + * + * ⚠️ Sampling instant (#15358 contract review): the hot copy is read HERE, + * i.e. AFTER the caller's `await loadTerminal(...)`, where the inline read + * this replaced took it BEFORE that await. The decision table is the same; + * the sample time is later. With {@link MAX_CONSUMED_SUSPENSIONS} bounding + * the journal, an eviction that lands during that await now answers a + * refusal (`NO_CONSUMED_SUSPENSION`) where the old read would have restored + * from the copy it had already captured. The direction is refusal, never a + * double-run — the copy is gone either way; only which of the two verbs + * notices moved. + * + * @param terminal - The durable terminal row, already loaded by the + * caller (each caller owns its own outage posture for that read). + */ + private resolveConsumedSuspensionWitnesses( + runId: string, + terminal: RunRecord | null, + ): { + consumed?: ConsumedSuspension; + witness?: 'journal' | 'durable'; + dropped?: ConsumedSuspensionDropNotice; + /** The hot copy describes a pause the run has since LEFT. */ + staleHot: boolean; + } { + const hot = this.consumedSuspensions.get(runId); + if (!terminal) { + return hot ? { consumed: hot, witness: 'journal', staleHot: false } : { staleHot: false }; + } + if (terminal.consumedSuspension) { + const durable: ConsumedSuspension = { + run: terminal.consumedSuspension, + consumedAt: terminal.finishedAt ?? terminal.startedAt, + error: terminal.error ?? '', + }; + if (hot && !this.rowSupersedesJournal(hot, durable.run)) { + return { consumed: hot, witness: 'journal', staleHot: false }; + } + return { consumed: durable, witness: 'durable', staleHot: hot !== undefined }; + } + if (terminal.consumedSuspensionDropped) { + const dropped = terminal.consumedSuspensionDropped; + if (hot && !this.rowSupersedesJournal(hot, dropped)) { + return { consumed: hot, witness: 'journal', dropped, staleHot: false }; + } + return { dropped, staleHot: hot !== undefined }; + } + if (hot && hot.persisted !== 'landed') { + // The row predates this strand — this process's own write for it + // never reached the store (in flight, or failed and reported at + // `error`). The store's silence says nothing. + return { consumed: hot, witness: 'journal', staleHot: false }; + } + // A later terminal record with no snapshot and no drop notice: + // completed, cancelled or cascade-failed after any hot copy was + // taken. Stale by definition. + return { staleHot: hot !== undefined }; + } + /** Build a refusal from {@link restoreConsumedSuspension}. */ private refuseRestore( runId: string, @@ -6756,40 +6903,11 @@ export class AutomationEngine implements IAutomationService { } // [#13937] Two witnesses of one strand, and neither is trusted - // alone — the contract review of this ruling's services half - // measured both single-witness readings wrong, one store class - // apart: - // - // - This process's HOT copy is the verbatim object the failure - // was journalled from: the pause's own node, variables, step - // log as of the pause. It is a per-process cache. The replica - // that stranded a run keeps it after another replica restored, - // resumed and FINISHED the run, and re-arming it then re-runs - // every node after the pause — shape 2's silent double-run, - // through this verb's side door (pinned in - // `stranded-run-status.test.ts`, red on the hot-only tree). - // - The DURABLE row is the record every replica can read, and a - // flattened, column-bounded copy of the same snapshot: the - // object store rebuilds it from columns, DROPS it over a byte - // budget — and says so in the row, `consumedSuspensionDropped` - // — and receives it fire-and-forget. Read alone, a snapshot-less - // row sent a run the store could not persist into - // NO_CONSUMED_SUSPENSION on the very replica holding its copy - // (pinned in `stranded-run-object-store.test.ts`, red on the - // durable-first tree). - // - // So: the hot copy is preferred whenever both describe the SAME - // pause (`rowSupersedesJournal`). When they describe different - // pauses, the newest strand wins — a hot copy whose own write never - // landed (`persisted` is not `'landed'`: the #13617 exception, a - // row the store was never handed says nothing) beats the older - // row, and a landed hot copy yields to the later strand another - // replica recorded. A row with neither a snapshot nor a drop notice - // is "the run moved on" only for a hot copy whose write did land — - // that copy is then DROPPED rather than honoured. A hot copy - // answers alone where there is no row to ask: no store, no run - // history, a write that never landed or is still in flight. - const hot = this.consumedSuspensions.get(runId); + // alone — the read is {@link resolveConsumedSuspensionWitnesses}, + // shared with the read-only {@link inspectConsumedSuspension} so + // what that verb calls repairable is what this one restores. The + // durable row is loaded HERE because this verb's posture on an + // unreadable history is its own: a refusal, never a guess. let terminal: RunRecord | null = null; if (this.store?.loadTerminal) { try { @@ -6811,38 +6929,10 @@ export class AutomationEngine implements IAutomationService { } } - let consumed: ConsumedSuspension | undefined; - let dropped: ConsumedSuspensionDropNotice | undefined; - if (!terminal) { - consumed = hot; - } else if (terminal.consumedSuspension) { - const durable: ConsumedSuspension = { - run: terminal.consumedSuspension, - consumedAt: terminal.finishedAt ?? terminal.startedAt, - error: terminal.error ?? '', - }; - consumed = hot && !this.rowSupersedesJournal(hot, durable.run) ? hot : durable; - // A hot copy of a pause the run has since LEFT — re-arming it - // would send the run back through work it already did. - if (hot && consumed !== hot) this.consumedSuspensions.delete(runId); - } else if (terminal.consumedSuspensionDropped) { - dropped = terminal.consumedSuspensionDropped; - if (hot && !this.rowSupersedesJournal(hot, dropped)) { - consumed = hot; - } else if (hot) { - this.consumedSuspensions.delete(runId); - } - } else if (hot && hot.persisted !== 'landed') { - // The row predates this strand — this process's own write for - // it never reached the store (in flight, or failed and - // reported at `error`). The store's silence says nothing. - consumed = hot; - } else if (hot) { - // A later terminal record with no snapshot and no drop notice: - // completed, cancelled or cascade-failed after this copy was - // taken. Stale by definition. - this.consumedSuspensions.delete(runId); - } + const { consumed, dropped, staleHot } = this.resolveConsumedSuspensionWitnesses(runId, terminal); + // A hot copy of a pause the run has since LEFT — re-arming it + // would send the run back through work it already did. + if (staleHot) this.consumedSuspensions.delete(runId); if (!consumed) { // Nothing to restore — say WHICH nothing. The remedy differs for @@ -6975,6 +7065,80 @@ export class AutomationEngine implements IAutomationService { } } + /** + * [#15358] **Read-only**: would {@link restoreConsumedSuspension} have a + * consumed suspension to put back for `runId`? Answers from the SAME two + * witnesses that verb reads — this process's hot journal and the durable + * terminal row, reconciled by {@link resolveConsumedSuspensionWitnesses} + * — and re-arms nothing and writes nothing. The one incidental mutation is + * the strict suspension read's own: {@link loadSuspendedRunStrict} may + * evict a phantom `suspendedRuns` entry the store has already answered + * "no row" for (#15832; `evictConsumedSuspension` touches `suspendedRuns` + * only, never the consumed-suspension journal) — identical to + * {@link hasSuspendedRun} today. A stale hot copy the row supersedes is + * REPORTED by the shared read and left in place; only the restore verb + * drops it. + * + * ## Why a dedicated member, and not a field on the run + * + * The discriminator lives on the durable {@link RunRecord} + * (`consumedSuspension` / `consumedSuspensionDropped`) and in this + * process's journal; {@link getRun} answers an {@link ExecutionLogEntry}, + * which carries neither ON PURPOSE — `recordLog` keeps the snapshot off + * that interface because `GET /automation/:name/runs/:runId` serves it + * verbatim. So a consumer reading `getRun` sees `status: 'failed'` and + * cannot tell the #13909 strand (repairable by the restore verb) from a + * cascade-failed ancestor (`failAncestors` → `failSuspendedRun`, which + * consumes the ancestor's pause and journals nothing — repairable by + * nothing; #15222's shape). Ruled on #15358 (B′, 2026-09-07): the answer is + * published as a dedicated read-only engine member, never on the wire. + * plugin-approvals' stranded-request inspection is the first consumer. + * + * ⛔ Not a single `loadTerminal` read. A snapshot-less row is "the run + * moved on" only when this process holds no hot copy whose write never + * landed, and a drop notice is repairable from the hot copy on the very + * replica that stranded the run — both measured wrong under a + * single-witness reading (#13937's contract review). Sharing the read + * with the restore verb is what keeps that from regressing in one of the + * two alone. + * + * ## What it does NOT judge + * + * Whether re-arming would be SAFE at this instant — the `restoring` / + * `resuming` guards — is the restore verb's own pre-flight. A verdict + * here is about what survives, not about the moment. + * + * @throws when a store read fails — the suspended-run store or the run + * history. An unreadable store is UNKNOWN, and a read-only verdict that + * turned an outage into `NO_CONSUMED_SUSPENSION` would send an operator + * (or a sweep) to give up on a run that is repairable. Same posture as + * {@link hasSuspendedRun}; the restore verb answers `STORE_UNAVAILABLE` + * for the same observation. + */ + async inspectConsumedSuspension(runId: string): Promise { + // Already resumable? STRICT read — a store outage throws through. + const live = await this.loadSuspendedRunStrict(runId); + if (live) return { repairable: false, runId, reason: 'RUN_SUSPENDED', nodeId: live.nodeId }; + + let terminal: RunRecord | null = null; + if (this.store?.loadTerminal) terminal = await this.store.loadTerminal(runId); + + const { consumed, witness, dropped } = this.resolveConsumedSuspensionWitnesses(runId, terminal); + if (consumed && witness) { + return { + repairable: true, + runId, + flowName: consumed.run.flowName, + nodeId: consumed.run.nodeId, + ...(consumed.run.correlation !== undefined ? { correlation: consumed.run.correlation } : {}), + witness, + consumedAt: consumed.consumedAt, + }; + } + if (dropped) return { repairable: false, runId, reason: 'SNAPSHOT_DROPPED', dropped }; + return { repairable: false, runId, reason: 'NO_CONSUMED_SUSPENSION' }; + } + /** * Walk a failed run's `$parentRunId` chain and fail each suspended * ancestor (see {@link failSuspendedRun}). Bounded so a corrupt context diff --git a/packages/services/service-automation/src/index.ts b/packages/services/service-automation/src/index.ts index 2b4bf8299e..7ebeb4d8e6 100644 --- a/packages/services/service-automation/src/index.ts +++ b/packages/services/service-automation/src/index.ts @@ -58,6 +58,12 @@ export type { // host store implementing `recordTerminal` / `loadTerminal` writes and // reads; unnameable, the field would be writable only by structural luck. ConsumedSuspensionDropNotice, + // [#15358] The read-only repairability verdict + // (`AutomationEngine.inspectConsumedSuspension`), for the same reason as + // `SuspensionRestoreResult` above: the method is barrel-reachable, so a + // consumer needs the name to annotate a result or switch exhaustively over + // `reason` — the three negatives exist precisely to be branched on. + ConsumedSuspensionInspection, } from './engine.js'; // [#11997] ADR-0005 overlay precedence for same-named flow definitions. The boot