From bce1509f49809b559f52c771a0772b72982639b0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 8 Sep 2026 02:24:12 +0000 Subject: [PATCH] fix(approvals): keep an undifferentiable stranded row in the report, and stop a malformed host verdict aborting the scan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three residues of the #15358 contract review (#16709). Item 1 (test-only) — the restore verb's drop of a stale hot consumed-suspension copy was unpinned package-wide: the reviewer's E2 ablation deleted the behaviour and left the whole service-automation suite green. Pinned where the drop is distinguishable from a no-op — after the durable row that proves the copy stale is evicted by run-history retention, a kept copy would be the only witness left and would offer an operator a restore of a run that already COMPLETED on another replica. Item 2 (PM ruling, 2026-09-08) — a thrown third read counted `undetermined` and dropped the row. By the time that oracle is asked the first two have already answered (no live pause, terminal `failed`); it is asked only WHICH of the three shapes the row is, so a read that could not be made is exactly the "could not differentiate" case `'failed'` already means. The row now stays in the report; `undetermined` is kept as telemetry. Item 3 — `refineFailedRunState(verdict)` ran outside the `try`, so a host resolving `undefined` threw a `TypeError` out of `inspectStrandedRequests` and the scan enumerated nothing. The refinement now runs inside that `try`: a malformed verdict costs its own row the differentiation and no other row anything. No new `StrandedRunState` member and no widened export. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_012zTkyNHJ7TkuN2oXtP5x37 --- ...randed-inspection-undifferentiated-rows.md | 12 + .../plugin-approvals/src/approval-service.ts | 75 ++++-- .../src/stranded-request-inspection.test.ts | 169 ++++++++++++- .../src/stale-hot-consumed-suspension.test.ts | 222 ++++++++++++++++++ 4 files changed, 457 insertions(+), 21 deletions(-) create mode 100644 .changeset/stranded-inspection-undifferentiated-rows.md create mode 100644 packages/services/service-automation/src/stale-hot-consumed-suspension.test.ts diff --git a/.changeset/stranded-inspection-undifferentiated-rows.md b/.changeset/stranded-inspection-undifferentiated-rows.md new file mode 100644 index 0000000000..b0c87cd1e6 --- /dev/null +++ b/.changeset/stranded-inspection-undifferentiated-rows.md @@ -0,0 +1,12 @@ +--- +"@objectstack/plugin-approvals": patch +--- + +`inspectStrandedRequests` no longer drops a row it could not differentiate, and no longer lets one misbehaving host abort the whole scan (#16709, items 2 and 3). + +Both are the same mistake at two altitudes: the method exists to **enumerate** the terminal approval requests whose flow run cannot advance, so a failure to read the #15358 third oracle must never remove a row from the answer — and never remove the *other* rows either. + +- **A thrown third read now leaves its row in `stranded`, as `'failed'`.** It used to be counted `undetermined` and skipped, exactly as a thrown `hasSuspendedRun` or `getRun` is. Those two are not the same question: a throw from either leaves it unknown *whether* the row is stranded at all, and a storage outage must not be published as a lost run. By the time the third oracle is asked, both have answered — no live pause, terminal `failed` — and it is asked only *which* of the three shapes the row is. A read that could not be made is therefore the textbook "could not differentiate", which is what `'failed'` already means (`StrandedRunState`, #15358 ruling item 1). Dropping the row let `stranded: []` read as "nothing stranded" while a row was in fact stuck, with a log line as its only trace; for a report, fail-closed means showing the row. +- **A host that violates `ApprovalResumeSurface` no longer aborts the scan.** `refineFailedRunState(verdict)` ran outside the `try` that wrapped the read, so an implementation resolving `undefined` where a verdict is declared threw a `TypeError` out of `inspectStrandedRequests` itself and the scan enumerated **nothing**. The refinement now runs inside that `try`; a malformed verdict costs its own row the differentiation, is counted `undetermined`, and costs every other row nothing. + +⛔ No new `StrandedRunState` member and no widened export: both cases map onto the existing undifferentiated `'failed'`. The `undetermined` counter is kept as telemetry and now **overlaps** `stranded` by design — a row can be both reported and counted — so neither number alone sizes the scan's blind spot. diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index f65d0daa69..d6c8439960 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -210,8 +210,11 @@ export interface ApprovalResumeSurface { * 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}. + * counts such a row `undetermined` — but ⛔ unlike a thrown + * {@link hasSuspendedRun} it does NOT drop the row, because this oracle is + * asked only WHICH shape a row already known to be stranded is (#16709). + * A host that resolves a malformed verdict is treated the same way, and + * costs no OTHER row its answer. */ inspectConsumedSuspension?(runId: string): Promise< | { repairable: true } @@ -495,13 +498,15 @@ function refineFailedRunState(verdict: ConsumedSuspensionVerdict): StrandedRunSt * `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. + * - `failed` — the engine COULD NOT BE ASKED which of the three it is, or + * was asked and could not answer. Three ways in: the attached surface has + * no `inspectConsumedSuspension` (an engine build older than this plugin, + * or a test double); the read THREW (a store outage); or the host resolved + * a malformed verdict, violating its own declared surface (#16709). Today's + * undifferentiated label, kept on purpose as the fail-closed fallback + * (#15358 ruling, item 1): a failure to differentiate 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'` @@ -4394,7 +4399,10 @@ export class ApprovalService implements IApprovalService { * `'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. + * of anything. So does a read that THREW or answered a malformed verdict + * (#16709): by the time this oracle is asked the row is already known to be + * stranded, so a failure to differentiate it is not a reason to drop it from + * a report — it is counted `undetermined` as telemetry AND reported. * * ⚠️ **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 @@ -4412,7 +4420,16 @@ export class ApprovalService implements IApprovalService { async inspectStrandedRequests(options?: { limit?: number }): Promise<{ scanned: number; stranded: StrandedApprovalRequest[]; - /** Rows skipped because the suspension store could not be read — NOT healthy, just unknown. */ + /** + * Reads that could not be MADE — telemetry, ⛔ never a verdict and ⛔ never + * a "healthy" number. A thrown first or second oracle SKIPS its row + * (whether that row is stranded at all is then unknown, and a storage + * outage must not be published as a lost run); a thrown or malformed THIRD + * read leaves its row in `stranded` as the undifferentiated `'failed'` and + * is counted here as well — the row is known to be stranded, only its + * shape could not be told (#16709). So this counter and `stranded.length` + * overlap on purpose, and neither one alone sizes the scan's blind spot. + */ undetermined: number; }> { const empty = { scanned: 0, stranded: [] as StrandedApprovalRequest[], undetermined: 0 }; @@ -4480,19 +4497,43 @@ export class ApprovalService implements IApprovalService { // 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; + // ⚠️ [#16709 item 3] The REFINEMENT runs inside this `try`, with the + // read it refines. `refineFailedRunState` dereferences the verdict, so + // a host that violates the declared surface — resolving `undefined` + // where a verdict is declared — used to throw a `TypeError` out of + // `inspectStrandedRequests` itself, turning a PARTIAL answer into NO + // answer for every OTHER row in the scan. Enumerating the rows that + // cannot advance is this method's entire purpose, so a misbehaving + // implementation must cost at most the differentiation of its own row. + let refined: StrandedRunState | undefined; + let differentiated = true; try { - verdict = await this.automation.inspectConsumedSuspension(runId); + refined = refineFailedRunState(await this.automation.inspectConsumedSuspension(runId)); } catch (err: any) { + // [#16709 item 2 — PM ruling, 2026-09-08] The row STAYS in the + // report, as the undifferentiated `'failed'`. This oracle is not + // asked WHETHER the row is stranded: the first two already answered + // that (no live pause, terminal `failed`). It is asked only WHICH of + // the three shapes it is — so a read that could not be made is the + // textbook "could not differentiate" case, which is exactly what + // `'failed'` is kept for (#15358 ruling, item 1). + // + // ⛔ Never dropped from the list. This is a REPORT of rows that + // cannot advance, and a row whose state we failed to determine is + // precisely the row an operator has to see; skipping it would make + // "nothing stranded" read TRUE while a row is in fact stuck, with + // the only trace a log line nobody is paging on. `undetermined` + // still counts it, as telemetry — never as a verdict. + differentiated = false; 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; + if (differentiated) { + 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 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 d7780b46cf..18628cefe9 100644 --- a/packages/plugins/plugin-approvals/src/stranded-request-inspection.test.ts +++ b/packages/plugins/plugin-approvals/src/stranded-request-inspection.test.ts @@ -116,6 +116,15 @@ function automation(opts: { historyThrows?: boolean; repairability?: Record; repairabilityThrows?: boolean; + /** Runs whose third-oracle read THROWS — a store outage on those rows alone. */ + repairabilityThrowsFor?: string[]; + /** + * Runs whose host RESOLVES `undefined` — a contract-violating implementation + * of its own declared surface (#16709 item 3). ⛔ Deliberately outside + * `Verdict`: pinning what happens when a host lies is the whole point, and + * the cast that makes it expressible is confined to this double. + */ + repairabilityMalformedFor?: string[]; } = {}) { const inspectCalls: string[] = []; const surface: any = { @@ -130,10 +139,16 @@ function automation(opts: { return opts.history?.[runId] ?? null; }, }; - if (opts.repairability !== undefined || opts.repairabilityThrows) { + if ( + opts.repairability !== undefined || opts.repairabilityThrows + || opts.repairabilityThrowsFor || opts.repairabilityMalformedFor + ) { surface.inspectConsumedSuspension = async (runId: string): Promise => { inspectCalls.push(runId); - if (opts.repairabilityThrows) throw new Error('run history unreadable for the consumed suspension'); + if (opts.repairabilityThrows || opts.repairabilityThrowsFor?.includes(runId)) { + throw new Error('run history unreadable for the consumed suspension'); + } + if (opts.repairabilityMalformedFor?.includes(runId)) return undefined as unknown as Verdict; const v = opts.repairability?.[runId]; if (!v) throw new Error(`test surface: no verdict scripted for ${runId}`); return v; @@ -501,10 +516,26 @@ describe('#15358 — the third oracle splits `failed` three ways, and its ABSENC expect(out.undetermined).toBe(0); }); - it('a THROWN read is `undetermined`, exactly like the other two oracles — never a verdict', async () => { + it('⭐ [#16709 item 2] a THROWN read keeps the row REPORTED as `failed` — it never leaves the list', async () => { + // ⚠️ This assertion USED TO READ `expect(out.stranded).toEqual([])`: a + // thrown third read was counted `undetermined` and the row dropped, as for + // the other two oracles. Ruled the other way (PM seat, 2026-09-08). + // + // The two earlier oracles and this one are not asked the same question. A + // thrown `hasSuspendedRun` or `getRun` leaves it unknown WHETHER the row is + // stranded at all, and a storage outage must not be published as a lost + // run. By the time this oracle is asked, both have already answered: no + // live pause, terminal `failed`. It is asked only WHICH of the three + // shapes — so a read that could not be made is the textbook "could not + // differentiate", which is exactly what `'failed'` is kept for (#15358 + // ruling, item 1). Dropping the row would let "nothing stranded" read TRUE + // while a row is in fact stuck, with a log line as its only trace; for a + // REPORT, fail-closed means showing the row. svc.attachAutomation(automation({ ...failedRun, repairabilityThrows: true })); const out = await svc.inspectStrandedRequests(); - expect(out.stranded).toEqual([]); + expect(out.stranded.map(s => [s.requestId, s.runState])).toEqual([['areq_1', 'failed']]); + // The counter is KEPT, as telemetry — it and `stranded` now overlap by + // design, and neither alone sizes the scan's blind spot. expect(out.undetermined).toBe(1); }); @@ -589,3 +620,133 @@ describe('#15358 — the third oracle splits `failed` three ways, and its ABSENC expect(JSON.stringify(engine._tables)).toBe(before); }); }); + +// ── #16709: a failure to DIFFERENTIATE never costs a row its place, and never +// costs another row its answer ───────────────────────────────────────────── +// +// Two residues of the #15358 contract review, ruled together (PM seat, +// 2026-09-08): +// +// item 2 — a thrown third read counted `undetermined` and DROPPED the row. +// item 3 — `refineFailedRunState(verdict)` ran OUTSIDE the `try`, so a host +// that violates its own declared surface by resolving `undefined` +// threw a `TypeError` out of `inspectStrandedRequests` and the scan +// enumerated NOTHING. +// +// Both are the same mistake at two altitudes: this method exists to enumerate +// the rows that cannot advance, so a row it could not differentiate stays in +// the report as the undifferentiated `'failed'`, and a row it could not read +// at all costs no OTHER row its answer. ⛔ Neither is a new `StrandedRunState` +// member: `'failed'` already means "reported, could not differentiate". + +describe('#16709 — a failure to differentiate keeps the row, and stays local to it', () => { + 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('⭐ item 3 — a host resolving `undefined` is answered, not thrown out of the scan', async () => { + // The declared surface says this member resolves a verdict. A host that + // resolves `undefined` breaks that — and `refineFailedRunState` reads + // `verdict.repairable`, so the old code's `TypeError` escaped the method. + svc.attachAutomation(automation({ ...failedRun, repairabilityMalformedFor: ['run_1'] })); + await expect(svc.inspectStrandedRequests()).resolves.toMatchObject({ scanned: 1, undetermined: 1 }); + const out = await svc.inspectStrandedRequests(); + // Same disposition as a thrown read: reported, undifferentiated. + expect(out.stranded.map(s => [s.requestId, s.runState])).toEqual([['areq_1', 'failed']]); + }); + + it('⭐ items 2+3 — one bad row costs ITSELF a label and every other row nothing', async () => { + // The harm the two items share, measured on one population: before the + // fix the malformed row alone turned this whole call into a rejection, so + // `areq_ok` — a perfectly readable, perfectly repairable strand — was + // never enumerated either. A PARTIAL answer became NO answer. + engine._tables['sys_approval_request'] = [ + requestRow({ id: 'areq_throw', flow_run_id: 'run_throw' }), + requestRow({ id: 'areq_malformed', flow_run_id: 'run_malformed' }), + requestRow({ id: 'areq_ok', flow_run_id: 'run_ok' }), + requestRow({ id: 'areq_missing', flow_run_id: 'run_missing' }), + ]; + const auto = automation({ + history: { + run_throw: { status: 'failed' }, + run_malformed: { status: 'failed' }, + run_ok: { status: 'failed' }, + // `run_missing` absent on purpose — it never reaches the third oracle. + }, + repairability: { run_ok: { repairable: true } }, + repairabilityThrowsFor: ['run_throw'], + repairabilityMalformedFor: ['run_malformed'], + }); + svc.attachAutomation(auto); + + const out = await svc.inspectStrandedRequests(); + expect(out.scanned).toBe(4); + expect(out.stranded.map(s => [s.requestId, s.runState])).toEqual([ + ['areq_throw', 'failed'], + ['areq_malformed', 'failed'], + ['areq_ok', 'repairable'], + ['areq_missing', 'missing'], + ]); + // Both undifferentiated rows are counted, and only those two. + expect(out.undetermined).toBe(2); + // The third oracle really was reached for each `failed` row, and only + // those — so the labels above are its answers, not a skipped branch. + expect(auto.inspectCalls).toEqual(['run_throw', 'run_malformed', 'run_ok']); + }); + + it('⛔ item 2 does NOT widen to the two earlier oracles — those still SKIP their row', async () => { + // The control that makes the ruling legible. The distinction is not "a + // throw is fine now": it is WHICH question was being asked. A thrown first + // or second oracle leaves it unknown whether the row is stranded at all, + // and condemning on an outage is the harm those arms were written for. + engine._tables['sys_approval_request'] = [requestRow({ id: 'areq_h', flow_run_id: 'run_h' })]; + svc.attachAutomation(automation({ suspendedThrows: true })); + expect(await svc.inspectStrandedRequests()).toMatchObject({ scanned: 1, stranded: [], undetermined: 1 }); + + svc.attachAutomation(automation({ historyThrows: true })); + expect(await svc.inspectStrandedRequests()).toMatchObject({ scanned: 1, stranded: [], undetermined: 1 }); + + // Positive control on the same row: with both stores readable and only the + // THIRD read failing, the row IS reported — so the empty lists above are + // those two oracles' posture, not a row that was never strandable. + svc.attachAutomation(automation({ + history: { run_h: { status: 'failed' } }, repairabilityThrowsFor: ['run_h'], + })); + const out = await svc.inspectStrandedRequests(); + expect(out.stranded.map(s => s.runState)).toEqual(['failed']); + expect(out.undetermined).toBe(1); + }); + + it('⛔ still no sixth `StrandedRunState`: the undifferentiated rows are literally `failed`', async () => { + // Item 2's ruling is a re-use of an existing member, not a new one — the + // reason it touches no barrel-exported type. Every label this scan can + // emit is one of the five, and both undifferentiated shapes emit the same + // string an ABSENT member emits. + engine._tables['sys_approval_request'] = [ + requestRow({ id: 'areq_absent', flow_run_id: 'run_absent' }), + requestRow({ id: 'areq_throw', flow_run_id: 'run_throw' }), + requestRow({ id: 'areq_malformed', flow_run_id: 'run_malformed' }), + ]; + const history = { + run_absent: { status: 'failed' }, run_throw: { status: 'failed' }, run_malformed: { status: 'failed' }, + }; + // The member is absent for `run_absent`'s scan… + svc.attachAutomation(automation({ history })); + const blind = await svc.inspectStrandedRequests(); + // …and present-but-failing for the other two. + svc.attachAutomation(automation({ + history, repairabilityThrowsFor: ['run_throw', 'run_absent'], + repairabilityMalformedFor: ['run_malformed'], + })); + const failing = await svc.inspectStrandedRequests(); + + expect(new Set([...blind.stranded, ...failing.stranded].map(s => s.runState))).toEqual(new Set(['failed'])); + }); +}); diff --git a/packages/services/service-automation/src/stale-hot-consumed-suspension.test.ts b/packages/services/service-automation/src/stale-hot-consumed-suspension.test.ts new file mode 100644 index 0000000000..6f30c55b59 --- /dev/null +++ b/packages/services/service-automation/src/stale-hot-consumed-suspension.test.ts @@ -0,0 +1,222 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #16709 item 1 — the restore verb DROPS a hot copy the durable row proves + * stale, and this file is the only thing that observes it. + * + * ## The gap this closes, measured + * + * `resolveConsumedSuspensionWitnesses` (#15358) reports `staleHot` and + * `restoreConsumedSuspension` acts on it in one line: + * + * ```ts + * const { consumed, dropped, staleHot } = this.resolveConsumedSuspensionWitnesses(runId, terminal); + * if (staleHot) this.consumedSuspensions.delete(runId); + * ``` + * + * The #15358 contract review ablated exactly that: make the shared helper + * never report `staleHot`, so the restore verb never deletes the entry. The + * eight restore-verb test files stayed green and so did the full suite — a + * real behaviour could be deleted and nothing in either package noticed. The + * behaviour is not new (the inline deletes the refactor replaced had the same + * absent pins), but after the refactor it lives behind ONE helper flag, which + * makes the gap cheaper to fall into. + * + * ## Why the drop matters — the harm, not the flag + * + * The hot journal is a PER-PROCESS cache. The replica that stranded a run + * keeps its verbatim copy of the pause even after another replica restores, + * resumes and FINISHES that run. Re-arming it then re-runs every node after + * the pause: shape 2's silent double-run, through the restore verb's side + * door. The durable row is what tells the stranding replica the run moved on, + * and dropping the copy is how that knowledge outlives the row. + * + * ⭐ So the pin is written on the OUTLIVING, which is the only place the drop + * is distinguishable from a no-op: with the terminal row still present, both + * a dropped and a kept hot copy answer `NO_CONSUMED_SUSPENSION` (the row + * supersedes the copy on every read). Once the row is gone — the run-history + * retention cap evicts terminal rows per flow, `maxTerminalRunsPerFlow` / + * `DEFAULT_MAX_TERMINAL_RUNS_PER_FLOW`, #2585 — a KEPT copy becomes the only + * witness again and answers `repairable: true, witness: 'journal'` for a run + * that has already completed. That is the difference this file measures. + */ + +import { describe, it, expect } from 'vitest'; + +import { AutomationEngine, type RunRecord, type SuspendedRunStore } from './engine.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' }, + ], +}; + +const ctx = { event: 'test', record: { id: 'rec_1' }, params: { ticket: 'TKT-9' } } as unknown as AutomationContext; + +/** One replica: its own engine and journal, over a shared store. */ +function replica(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.registerFlow('strand_flow', STRAND_FLOW as never); + return { engine, state }; +} + +/** + * The shared store, with the one thing a real deployment does to a terminal + * row that an in-test cap cannot do deterministically: FORGET it. The store's + * own per-flow retention (`recordTerminal` prunes beyond + * `maxTerminalRunsPerFlow`) is the mechanism; `forget` is that eviction, + * addressed by run id so the fixture states which row went rather than racing + * two `startedAt` stamps for it. + */ +function prunableStore() { + const inner = new InMemorySuspendedRunStore(); + const forgotten = new Set(); + const store: SuspendedRunStore & { forget(runId: string): void } = { + 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): Promise { + if (forgotten.has(id)) return null; + return inner.loadTerminal(id); + }, + forget(runId: string) { forgotten.add(runId); }, + }; + return store; +} + +/** + * Drive `a` into the stranded state, then let `b` — another replica over the + * same store — restore it and run it to completion. Leaves: a hot journal + * entry on `a`, and a durable terminal row saying `completed`. + */ +async function strandOnAFinishOnB(store: SuspendedRunStore) { + const { engine: a } = replica(store); + const { engine: b, state: bState } = replica(store); + + const started = await a.execute('strand_flow', ctx); + expect(started.status).toBe('paused'); + const runId = started.runId as string; + const stranded = await a.resume(runId); + expect(stranded.status).toBe('stranded'); + // The fire-and-forget history write settles `persisted: 'pending'` → + // `'landed'` off the resume's own promise chain; flush it, because the + // "row supersedes a landed copy" arm is what this file drives. + await new Promise(resolve => setTimeout(resolve, 0)); + + // A really is the replica holding the copy — stated, not assumed. + await expect(a.inspectConsumedSuspension(runId)).resolves.toMatchObject({ + repairable: true, witness: 'journal', nodeId: 'pause', + }); + + // B repairs and finishes it. A is told nothing. + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + bState.throws = false; + expect((await b.resume(runId)).success).toBe(true); + + return { a, b, runId }; +} + +describe('#16709 item 1 — a stale hot copy is DROPPED by the restore verb, and does not outlive the row', () => { + it('⭐ after the row proving it stale is gone, the stranding replica answers NO_CONSUMED_SUSPENSION', async () => { + const store = prunableStore(); + const { a, runId } = await strandOnAFinishOnB(store); + + // A's own restore attempt is what reads the row and drops the copy. It + // refuses, naming the run's real end — and the refusal is NOT the pin: + // it reads identically whether or not the copy was dropped. + const refused = await a.restoreConsumedSuspension(runId); + expect(refused.restored).toBe(false); + expect((refused as { refusal?: string }).refusal).toBe('RUN_COMPLETED'); + + // Retention evicts the terminal row. A's journal is now the only + // witness left for this run — if it still holds the stale copy. + store.forget(runId); + expect(await store.loadTerminal!(runId)).toBeNull(); + + // ⛔ THE PIN. A kept copy answers `repairable: true, witness: 'journal'` + // here, offering an operator a restore of a run that COMPLETED. + await expect(a.inspectConsumedSuspension(runId)).resolves.toEqual({ + repairable: false, runId, reason: 'NO_CONSUMED_SUSPENSION', + }); + }); + + it('⭐ …and the restore verb itself refuses rather than re-arming a finished run', async () => { + // The same fact at the verb, where the cost is a silent double-run: + // re-arming the pause would replay every node after it on a run whose + // continuation already ran to `end` on another replica. + const store = prunableStore(); + const { a, runId } = await strandOnAFinishOnB(store); + expect((await a.restoreConsumedSuspension(runId)).restored).toBe(false); + + store.forget(runId); + const second = await a.restoreConsumedSuspension(runId); + expect(second.restored).toBe(false); + expect((second as { refusal?: string }).refusal).toBe('NO_CONSUMED_SUSPENSION'); + // Nothing was re-armed: no second pause exists for a finished run. + expect(await a.hasSuspendedRun(runId)).toBe(false); + }); + + it('CONTROL — the drop is the ROW\'s doing, not the read\'s: with no row, the copy survives', async () => { + // Without this, the two pins above could be passing because + // `inspectConsumedSuspension` or `restoreConsumedSuspension` consumes + // the journal entry on any refusal. Same replica, same verbs, same run + // — only the durable witness differs: a store that keeps NO history at + // all never supersedes the hot copy, and the copy is still there + // afterwards to be restored. + const inner = new InMemorySuspendedRunStore(); + const historyless: SuspendedRunStore = { + save: (r) => inner.save(r), load: (id) => inner.load(id), + delete: (id) => inner.delete(id), list: () => inner.list(), + }; + expect(historyless.loadTerminal).toBeUndefined(); + const { engine: a } = replica(historyless); + + const started = await a.execute('strand_flow', ctx); + const runId = started.runId as string; + expect((await a.resume(runId)).status).toBe('stranded'); + await new Promise(resolve => setTimeout(resolve, 0)); + + await expect(a.inspectConsumedSuspension(runId)).resolves.toMatchObject({ + repairable: true, witness: 'journal', + }); + expect((await a.restoreConsumedSuspension(runId)).restored).toBe(true); + }); +});