|
| 1 | +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. |
| 2 | + |
| 3 | +/** |
| 4 | + * #15944 — a run whose nodes ALL succeeded must never be journalled, reported |
| 5 | + * `stranded`, or re-armed because its terminal HISTORY write threw. |
| 6 | + * |
| 7 | + * ## The reported defect, in one sentence |
| 8 | + * |
| 9 | + * `resumeInternal`'s completion path called `this.recordLog({ status: |
| 10 | + * 'completed' })` from INSIDE the same `try` whose `catch` exists for node |
| 11 | + * failures, so a throw out of a history write on a run that finished |
| 12 | + * successfully was handled as though a node had thrown: the arm journalled a |
| 13 | + * repair snapshot, stamped `status: 'stranded'`, answered `success: false` — |
| 14 | + * and `restoreConsumedSuspension` then honoured that snapshot and re-armed the |
| 15 | + * pause, so the NEXT resume ran the downstream nodes a second time. |
| 16 | + * |
| 17 | + * ⚠️ The direction is the opposite of #15555's, which is the same window read |
| 18 | + * from the failure side. That card is a false `false` (an operator told not to |
| 19 | + * repair a run that is repairable). This is a false `true`: an operator told |
| 20 | + * to repair a run that already finished, where the repair RE-RUNS it. It |
| 21 | + * therefore crosses the #13937 shape-4 invariant outright — *"a re-armed run |
| 22 | + * must never become double-runnable"* — and the defence |
| 23 | + * `stranded-run-status.test.ts` pins for that invariant cannot fire here, |
| 24 | + * because it reads the durable terminal row first and the durable terminal row |
| 25 | + * is precisely what failed to land. |
| 26 | + * |
| 27 | + * ## What can throw out of `recordLog`, and why a `catch` there is needed |
| 28 | + * |
| 29 | + * `recordLog`'s own doc comment states the invariant this broke: *"Best-effort |
| 30 | + * + fire-and-forget: a history write must NEVER block or break the run that |
| 31 | + * produced it."* Two statements on its terminal path break it: |
| 32 | + * |
| 33 | + * 1. `this.store.recordTerminal(record)` — the `void write.catch(...)` |
| 34 | + * beneath that call only ever sees a RETURNED PROMISE's rejection, so a |
| 35 | + * store that throws SYNCHRONOUSLY, before returning a promise, escapes |
| 36 | + * `recordLog` entirely. (A store returning a non-thenable escapes the same |
| 37 | + * way: `write.catch` is then itself a synchronous `TypeError`.) |
| 38 | + * 2. The run-summary line `this.logger.info(line, meta)` — on by default |
| 39 | + * (`runSummaryLog: 'info'`) and calling a HOST-INJECTED `Logger`, so it |
| 40 | + * needs no store at all. |
| 41 | + * |
| 42 | + * Both are driven below; neither is modelled. |
| 43 | + * |
| 44 | + * ## What this file pins |
| 45 | + * |
| 46 | + * 1. **The completed run stays completed.** `resume` answers `success: true` |
| 47 | + * with NO `status` discriminator, no repair snapshot exists for it, and |
| 48 | + * `restoreConsumedSuspension` refuses with `RUN_COMPLETED` — the truthful |
| 49 | + * refusal, and the one that proves no journal was written (a journalled |
| 50 | + * run would have been re-armed instead of refused). |
| 51 | + * 2. **The sharp one: the downstream node runs exactly ONCE across two |
| 52 | + * resumes**, with an attempted repair in between. This is the assertion |
| 53 | + * the defect actually fails; every other pin here can be read as |
| 54 | + * bookkeeping. |
| 55 | + * 3. **The swallowed failure stays loud.** The guard must not trade a phantom |
| 56 | + * strand for silence: AGENTS.md "Degradation log levels" — the terminal |
| 57 | + * history row claims to persist and did not while the caller reads a clean |
| 58 | + * success — so `error`, with the consequence and the fix in the first line. |
| 59 | + * 4. **Controls, so the pins are readings and not constants** — a GENUINE |
| 60 | + * node failure on the very same throwing store still journals, still |
| 61 | + * reports `stranded`, and is still repairable; and a completed run on a |
| 62 | + * HEALTHY store logs no error at all. |
| 63 | + * |
| 64 | + * ## Deliberately NOT here |
| 65 | + * |
| 66 | + * ⛔ `restoreConsumedSuspension` is not weakened, and nothing here asserts a |
| 67 | + * change to it: the verb judges correctly on the evidence it is handed, and |
| 68 | + * the evidence is what was wrong. Its refusal is asserted, never its logic. |
| 69 | + * |
| 70 | + * ⛔ `inspectStrandedRequests` (#15358) is the adjacent over-reporting surface |
| 71 | + * one level up and is untouched by this file. |
| 72 | + */ |
| 73 | + |
| 74 | +import { describe, it, expect } from 'vitest'; |
| 75 | + |
| 76 | +import { AutomationEngine } from './engine.js'; |
| 77 | +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; |
| 78 | +import type { AutomationContext } from '@objectstack/spec/contracts'; |
| 79 | +import { defineActionDescriptor } from '@objectstack/spec/automation'; |
| 80 | + |
| 81 | +/** The history failure. Distinct text from the node's, so neither can stand in for the other. */ |
| 82 | +const TERMINAL_WRITE_FAILURE = 'run-history driver refused the terminal row'; |
| 83 | +const SUMMARY_LOG_FAILURE = 'log transport rejected the run-summary line'; |
| 84 | +/** The node failure used ONLY by the control, where a strand is correct. */ |
| 85 | +const NODE_FAILURE = 'tail blew up'; |
| 86 | + |
| 87 | +const holdDescriptor = defineActionDescriptor({ |
| 88 | + type: 'hold', version: '1.0.0', name: 'hold', |
| 89 | + supportsPause: true, resumeAuthority: 'any', |
| 90 | +}); |
| 91 | +const plain = (type: string) => defineActionDescriptor({ type, version: '1.0.0', name: type }); |
| 92 | + |
| 93 | +/** start → hold (pauses) → tail → end. `tail` SUCCEEDS unless a knob says otherwise. */ |
| 94 | +const RESUME_FLOW = { |
| 95 | + name: 'resume_flow', label: 'Resume', type: 'autolaunched', |
| 96 | + nodes: [ |
| 97 | + { id: 'start', type: 'start', label: 'Start' }, |
| 98 | + { id: 'hold', type: 'hold', label: 'Hold' }, |
| 99 | + { id: 'tail', type: 'tail', label: 'Tail' }, |
| 100 | + { id: 'end', type: 'end', label: 'End' }, |
| 101 | + ], |
| 102 | + edges: [ |
| 103 | + { id: 'e1', source: 'start', target: 'hold' }, |
| 104 | + { id: 'e2', source: 'hold', target: 'tail' }, |
| 105 | + { id: 'e3', source: 'tail', target: 'end' }, |
| 106 | + ], |
| 107 | +}; |
| 108 | + |
| 109 | +const ctx = { event: 'test', record: { id: 'rec_1' } } as unknown as AutomationContext; |
| 110 | + |
| 111 | +interface LoggedError { message: string; errorSlot: unknown; meta: unknown } |
| 112 | + |
| 113 | +/** |
| 114 | + * Records `error` calls positionally (the `Logger` contract is |
| 115 | + * `error(message, error?, meta?)`) and can be armed to throw from `info`. |
| 116 | + */ |
| 117 | +function recorder(opts: { infoThrows?: string } = {}) { |
| 118 | + const errors: LoggedError[] = []; |
| 119 | + return { |
| 120 | + errors, |
| 121 | + logger: { |
| 122 | + // Armed at exactly ONE call: the run-summary line `recordLog` |
| 123 | + // writes for the COMPLETED terminal run, which is the statement |
| 124 | + // inside the window. The engine also logs `info` while registering |
| 125 | + // executors and while running; throwing from those would measure a |
| 126 | + // different seam entirely. |
| 127 | + info(_msg: string, meta?: { status?: string }) { |
| 128 | + if (opts.infoThrows && meta?.status === 'completed') throw new Error(opts.infoThrows); |
| 129 | + }, |
| 130 | + warn() {}, |
| 131 | + debug() {}, |
| 132 | + error(message: string, errorSlot?: unknown, meta?: unknown) { |
| 133 | + errors.push({ message, errorSlot, meta }); |
| 134 | + }, |
| 135 | + } as never, |
| 136 | + }; |
| 137 | +} |
| 138 | + |
| 139 | +/** A durable store whose terminal write throws SYNCHRONOUSLY, before any promise exists. */ |
| 140 | +class SyncThrowTerminalStore extends InMemorySuspendedRunStore { |
| 141 | + override recordTerminal(): Promise<void> { |
| 142 | + throw new Error(TERMINAL_WRITE_FAILURE); |
| 143 | + } |
| 144 | +} |
| 145 | + |
| 146 | +function engineOver(store: InMemorySuspendedRunStore | undefined, logger: never) { |
| 147 | + const led = { tail: 0 }; |
| 148 | + const knobs = { throws: false }; |
| 149 | + const engine = new AutomationEngine(logger, store); |
| 150 | + engine.registerNodeExecutor({ |
| 151 | + type: 'hold', |
| 152 | + descriptor: holdDescriptor, |
| 153 | + async execute() { return { success: true, suspend: true, correlation: 'approval:req_1' }; }, |
| 154 | + } as never); |
| 155 | + engine.registerNodeExecutor({ |
| 156 | + type: 'tail', |
| 157 | + descriptor: plain('tail'), |
| 158 | + async execute() { |
| 159 | + led.tail++; |
| 160 | + if (knobs.throws) throw new Error(NODE_FAILURE); |
| 161 | + return { success: true, output: { done: true } }; |
| 162 | + }, |
| 163 | + } as never); |
| 164 | + engine.registerFlow('resume_flow', RESUME_FLOW as never); |
| 165 | + return { engine, led, knobs }; |
| 166 | +} |
| 167 | + |
| 168 | +/** |
| 169 | + * Resume, recording WHICH WAY the call ended. On the pre-guard tree the |
| 170 | + * completion path could also make `resume` THROW outright (both statements are |
| 171 | + * reachable and the catch arm's own `recordLog` throws again), so a plain |
| 172 | + * `await` would fail with a stack trace instead of producing a reading. |
| 173 | + */ |
| 174 | +async function resumeOutcome(engine: AutomationEngine, runId: string) { |
| 175 | + return engine.resume(runId).then( |
| 176 | + result => ({ kind: 'returned' as const, result, thrown: undefined }), |
| 177 | + (err: unknown) => ({ kind: 'threw' as const, result: undefined, thrown: err }), |
| 178 | + ); |
| 179 | +} |
| 180 | + |
| 181 | +async function park(engine: AutomationEngine) { |
| 182 | + const started = await engine.execute('resume_flow', ctx); |
| 183 | + expect(started.status).toBe('paused'); |
| 184 | + return started.runId as string; |
| 185 | +} |
| 186 | + |
| 187 | +describe('#15944 — a completed run must not be stranded, journalled or re-armed by its history write', () => { |
| 188 | + it('PIN 1 — the terminal write throws synchronously on a run whose nodes ALL succeeded: resume answers SUCCESS', async () => { |
| 189 | + const store = new SyncThrowTerminalStore(); |
| 190 | + const { logger } = recorder(); |
| 191 | + const { engine, led } = engineOver(store, logger); |
| 192 | + const runId = await park(engine); |
| 193 | + |
| 194 | + const outcome = await resumeOutcome(engine, runId); |
| 195 | + |
| 196 | + // ── The reproduction. Before the guard this returned |
| 197 | + // `{ success: false, status: 'stranded' }`, carrying the history |
| 198 | + // driver's text as the run's own error. |
| 199 | + expect(outcome.kind, 'a history write must never break the run that produced it').toBe('returned'); |
| 200 | + expect(outcome.result?.success, 'every node succeeded').toBe(true); |
| 201 | + expect(outcome.result?.status, 'a completed run is never stranded').toBeUndefined(); |
| 202 | + expect(outcome.result?.error, 'the run did not fail').toBeUndefined(); |
| 203 | + // The run's real answer survives the lost bookkeeping: the output the |
| 204 | + // flow produced, and the summary `recordLog` folds. The summary is |
| 205 | + // recomputed by the same pure function `recordLog` runs first, so the |
| 206 | + // two spellings cannot disagree. |
| 207 | + expect(outcome.result?.summary, 'the run summary survives the lost history row').toBeDefined(); |
| 208 | + expect(led.tail).toBe(1); |
| 209 | + |
| 210 | + // ── No repair snapshot exists for it. `RUN_COMPLETED` is what proves |
| 211 | + // that: a journalled run is RE-ARMED by this verb, not refused, so |
| 212 | + // this refusal and `restored: false` are one reading of "nothing was |
| 213 | + // journalled" — and it is the truthful refusal besides. |
| 214 | + const restored = await engine.restoreConsumedSuspension(runId, { requestedBy: 'ops' }); |
| 215 | + expect(restored.restored, 'a finished run has no unresumable state to exit').toBe(false); |
| 216 | + expect(restored.refusal).toBe('RUN_COMPLETED'); |
| 217 | + expect(await engine.hasSuspendedRun(runId), 'nothing was re-armed').toBe(false); |
| 218 | + |
| 219 | + // ── THE SHARP ONE. Across two resumes with an attempted repair |
| 220 | + // between them, the downstream node ran exactly ONCE. On the defective |
| 221 | + // tree the repair above answered `restored: true` and this second |
| 222 | + // resume drove `tail` a second time (#13937 shape 4: a re-armed run |
| 223 | + // must never become double-runnable). |
| 224 | + const second = await resumeOutcome(engine, runId); |
| 225 | + expect(second.kind).toBe('returned'); |
| 226 | + expect(second.result?.success).toBe(false); |
| 227 | + expect(second.result?.code, 'the run is terminal — there is nothing to resume').toBe('RUN_NOT_FOUND'); |
| 228 | + expect(led.tail, 'the downstream node ran EXACTLY once across two resumes').toBe(1); |
| 229 | + |
| 230 | + // ── The secondary failure was REAL, not simulated away: no durable |
| 231 | + // history row landed. That is the loss PIN 3 reports. |
| 232 | + expect(await store.loadTerminal(runId), 'the history row genuinely did not land').toBeFalsy(); |
| 233 | + }); |
| 234 | + |
| 235 | + it('PIN 2 — the run-summary log line throws, with NO store attached: same answer, same single run', async () => { |
| 236 | + // The second reachable statement in the same window, and one that |
| 237 | + // needs no store at all — the host-injected `Logger`'s `info`, on by |
| 238 | + // default. So this also shows the defect is not a store problem. |
| 239 | + const { errors, logger } = recorder({ infoThrows: SUMMARY_LOG_FAILURE }); |
| 240 | + const { engine, led } = engineOver(undefined, logger); |
| 241 | + const runId = await park(engine); |
| 242 | + |
| 243 | + const outcome = await resumeOutcome(engine, runId); |
| 244 | + |
| 245 | + expect(outcome.kind).toBe('returned'); |
| 246 | + expect(outcome.result?.success).toBe(true); |
| 247 | + expect(outcome.result?.status).toBeUndefined(); |
| 248 | + expect(led.tail).toBe(1); |
| 249 | + |
| 250 | + const restored = await engine.restoreConsumedSuspension(runId); |
| 251 | + expect(restored.restored).toBe(false); |
| 252 | + expect(restored.refusal).toBe('RUN_COMPLETED'); |
| 253 | + |
| 254 | + const second = await resumeOutcome(engine, runId); |
| 255 | + expect(second.result?.code).toBe('RUN_NOT_FOUND'); |
| 256 | + expect(led.tail, 'exactly once across two resumes').toBe(1); |
| 257 | + |
| 258 | + expect(errors.length, 'the swallowed throw is still reported').toBe(1); |
| 259 | + expect(errors[0]?.message).toContain(runId); |
| 260 | + }); |
| 261 | + |
| 262 | + it('PIN 3 — the swallowed history failure is loud: `error`, naming the run, the loss and the fix', async () => { |
| 263 | + // ⛔ The guard must not trade a phantom strand for a silent failure. |
| 264 | + // AGENTS.md "Degradation log levels": the terminal history row claims |
| 265 | + // to persist and did not, while every caller reads a healthy completed |
| 266 | + // run — the judgment question answers YES, so `error`, with the |
| 267 | + // consequence and the fix in the first line. |
| 268 | + const { errors, logger } = recorder(); |
| 269 | + const { engine } = engineOver(new SyncThrowTerminalStore(), logger); |
| 270 | + const runId = await park(engine); |
| 271 | + |
| 272 | + await resumeOutcome(engine, runId); |
| 273 | + |
| 274 | + expect(errors.length, 'said ONCE per run, not once per failed write').toBe(1); |
| 275 | + const line = errors[0]!; |
| 276 | + expect(line.message).toContain(runId); |
| 277 | + expect(line.message, 'the consequence: the run COMPLETED and its history row did not land') |
| 278 | + .toMatch(/completed/i); |
| 279 | + expect(line.message, 'the fix is the store failure in the meta').toMatch(/history/i); |
| 280 | + // THIRD argument per `error(message, error?, meta?)` — the driver text |
| 281 | + // goes to the structured slot, never into the message (#6499), and the |
| 282 | + // `Error` slot stays empty on purpose (#5575). |
| 283 | + expect(line.errorSlot).toBeUndefined(); |
| 284 | + expect(JSON.stringify(line.meta)).toContain(TERMINAL_WRITE_FAILURE); |
| 285 | + expect(line.message).not.toContain(TERMINAL_WRITE_FAILURE); |
| 286 | + }); |
| 287 | + |
| 288 | + it('CONTROL — a GENUINE node failure on the same throwing store still journals and still reports `stranded`', async () => { |
| 289 | + // ⛔ The guard must narrow nothing. This is the #15555 exit, driven on |
| 290 | + // the identical store: the node itself threw, so the run really did |
| 291 | + // strand and really is repairable, and both facts must survive. |
| 292 | + const { engine, led, knobs } = engineOver(new SyncThrowTerminalStore(), recorder().logger); |
| 293 | + knobs.throws = true; |
| 294 | + const runId = await park(engine); |
| 295 | + |
| 296 | + const outcome = await resumeOutcome(engine, runId); |
| 297 | + |
| 298 | + expect(outcome.kind).toBe('returned'); |
| 299 | + expect(outcome.result?.success).toBe(false); |
| 300 | + expect(outcome.result?.status, "the producer's discriminator, #13937 shape 4").toBe('stranded'); |
| 301 | + expect(outcome.result?.error).toContain(NODE_FAILURE); |
| 302 | + expect(led.tail).toBe(1); |
| 303 | + |
| 304 | + // The journal is there and the repair verb honours it — the exact |
| 305 | + // opposite of PIN 1, on the same store, distinguished only by whether |
| 306 | + // the NODE threw. |
| 307 | + const restored = await engine.restoreConsumedSuspension(runId, { requestedBy: 'ops' }); |
| 308 | + expect(restored.restored, 'a real strand is still repairable').toBe(true); |
| 309 | + expect(restored.refusal).toBeUndefined(); |
| 310 | + expect(await engine.hasSuspendedRun(runId)).toBe(true); |
| 311 | + }); |
| 312 | + |
| 313 | + it('CONTROL — a completed run on a HEALTHY store logs no error and lands its history row', async () => { |
| 314 | + // The reverse control for PIN 3. If this logged too, PIN 3 would be |
| 315 | + // measuring "the engine logs on every completed run", not "the guard |
| 316 | + // fired". |
| 317 | + const store = new InMemorySuspendedRunStore(); |
| 318 | + const { errors, logger } = recorder(); |
| 319 | + const { engine, led } = engineOver(store, logger); |
| 320 | + const runId = await park(engine); |
| 321 | + |
| 322 | + const outcome = await resumeOutcome(engine, runId); |
| 323 | + |
| 324 | + expect(outcome.result?.success).toBe(true); |
| 325 | + expect(outcome.result?.status).toBeUndefined(); |
| 326 | + expect(led.tail).toBe(1); |
| 327 | + expect(errors, 'no secondary failure ⇒ nothing to report').toEqual([]); |
| 328 | + |
| 329 | + // `recordTerminal` is fire-and-forget — let the microtask land. |
| 330 | + await new Promise((r) => setTimeout(r, 0)); |
| 331 | + expect((await store.loadTerminal(runId))?.status, 'the healthy path still persists').toBe('completed'); |
| 332 | + expect((await engine.restoreConsumedSuspension(runId)).refusal).toBe('RUN_COMPLETED'); |
| 333 | + }); |
| 334 | +}); |
0 commit comments