From d932ecd6f3e157d90aa4216af9ed940e58a1b244 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 6 Sep 2026 17:29:08 +0000 Subject: [PATCH 1/2] fix(service-automation): journal the pause's variables, not the failed attempt's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `resumeInternal` rebuilt the flow scope as `new Map(Object.entries( run.variables))` — keys copied, value objects SHARED — and `journalConsumedSuspension` then shallow-copied `run` and journalled it as the pause "VERBATIM". An executor that keeps state in the scope and updates it in place (`map`'s `.$mapState`) therefore wrote through into the snapshot `restoreConsumedSuspension` hands an operator, so the exit re-armed a pause carrying state that belonged to the failed attempt. Measured: the durable row held `started: 1` at the pause and the restore put back `started: 99`. The copy is taken before the failed attempt runs — the same line that already captures `stepCountAtPause`, for the same reason. No later placement works: the node mutates and THEN throws, so a copy taken at journal or restore time copies the corruption. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- ...sumed-suspension-snapshot-aliasing.test.ts | 227 ++++++++++++++++++ .../services/service-automation/src/engine.ts | 117 ++++++++- 2 files changed, 333 insertions(+), 11 deletions(-) create mode 100644 packages/services/service-automation/src/consumed-suspension-snapshot-aliasing.test.ts diff --git a/packages/services/service-automation/src/consumed-suspension-snapshot-aliasing.test.ts b/packages/services/service-automation/src/consumed-suspension-snapshot-aliasing.test.ts new file mode 100644 index 0000000000..65ce256718 --- /dev/null +++ b/packages/services/service-automation/src/consumed-suspension-snapshot-aliasing.test.ts @@ -0,0 +1,227 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #15660 — **the suspension an operator restores must be the pause, not the + * attempt that failed** — for NESTED values, which is where it was not. + * + * ## The state under test + * + * `resumeInternal` rebuilds the flow scope as + * `new Map(Object.entries(run.variables))`: the KEYS are copied, every VALUE + * OBJECT is shared with `run.variables`. `journalConsumedSuspension` then + * shallow-copies `run` and journals it as the pause "VERBATIM". So a node that + * keeps state in the scope and updates it **in place** writes straight through + * into the snapshot the operator exit later hands back — and + * `restoreConsumedSuspension` re-arms a pause carrying state that belongs to + * the failed attempt. + * + * `map` is the concrete instance and, by a census re-derived here rather than + * recalled (nothing else in the tree writes a `${node.id}.$…` object into the + * scope), the only executor in-repo that does it. The seam under test is the + * ENGINE's, not `map`'s, so the fixture is a minimal executor with the same + * shape: coupling this pin to `map`'s internals would make it fail for reasons + * that are not this card. + * + * ## ⚠️ The card was filed as a READING, and the reading's mechanism was wrong + * + * The card attributed the aliasing to "the in-memory suspended-run store keeps + * the object by identity rather than serialising it". It does not — it JSON + * round-trips on both save and load, and the last test here pins that. The + * defect reproduces anyway, because the aliasing that carries the mutation is + * minted on the RESUME (the scope rebuild above), not at the suspend. That is + * also why "deep copy at snapshot time" cannot be the fix: with a store + * configured the snapshot already IS a private deep copy, and it still + * reproduced. + * + * ## Why the controls are in the file and not in a scratch buffer + * + * Each in-place arm is paired with a replace-and-set arm that must stay green. + * Without them a harness that reported the pause's value for a trivial reason — + * a fixture whose mutation never ran, an assertion on the wrong key — would read + * exactly like a fixed engine. The pair is what makes either reading mean + * something. + */ + +import { describe, it, expect } from 'vitest'; + +import { AutomationEngine, type SuspendedRun, 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; + +/** The shape `map` keeps: node-scoped progress state living in the flow scope. */ +const STATE_KEY = 'worker.$mapState'; + +type ProgressState = { started: number; results: unknown[] }; + +const pauser = (type: string) => defineActionDescriptor({ + type, version: '1.0.0', name: type, supportsPause: true, resumeAuthority: 'any', +}); +const plain = (type: string) => defineActionDescriptor({ type, version: '1.0.0', name: type }); + +function flowDef(name: string) { + return { + name, label: name, type: 'autolaunched', + variables: [{ name: 'ticket', type: 'text', isInput: true, isOutput: true }], + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'seed', type: 'seed_state', label: 'Seed' }, + { id: 'pause', type: 'pause_here', label: 'Pause' }, + { id: 'worker', type: 'worker', label: 'Worker' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'seed' }, + { id: 'e2', source: 'seed', target: 'pause' }, + { id: 'e3', source: 'pause', target: 'worker' }, + { id: 'e4', source: 'worker', target: 'end' }, + ], + }; +} + +const ctx = { event: 'test', record: { id: 'rec_1' }, params: { ticket: 'TKT-9' } } as unknown as AutomationContext; + +/** + * `in-place` is the executor pattern under test. `replace` is its control: the + * same fixture, the same mutation, the same throw — only the parked object is + * left alone, so it must report the pause under a fixed engine AND under a + * broken one. + */ +type Arm = 'in-place' | 'replace'; + +function build(mode: Arm, store?: SuspendedRunStore) { + const engine = new AutomationEngine(silent as never, store); + const entry = { first: true }; + const observed: ProgressState[] = []; + + engine.registerNodeExecutor({ + type: 'seed_state', descriptor: plain('seed_state'), + async execute(_node: unknown, variables: Map) { + variables.set(STATE_KEY, { started: 1, results: ['at-suspend'] } satisfies ProgressState); + return { success: true }; + }, + } as never); + + engine.registerNodeExecutor({ + type: 'pause_here', descriptor: pauser('pause_here'), + async execute() { + return { success: true, suspend: true, correlation: 'approval:req_1', output: { stage: 'awaiting' } }; + }, + } as never); + + engine.registerNodeExecutor({ + type: 'worker', descriptor: plain('worker'), + async execute(_node: unknown, variables: Map) { + const state = variables.get(STATE_KEY) as ProgressState; + if (entry.first) { + entry.first = false; + if (mode === 'in-place') { + state.started = 99; + state.results.push('post-resume'); + } else { + variables.set(STATE_KEY, { started: 99, results: [...state.results, 'post-resume'] }); + } + // Strand the run: the pause is already consumed, so this throw + // is what makes `restoreConsumedSuspension` the only way out. + throw new Error('downstream node blew up'); + } + observed.push(JSON.parse(JSON.stringify(state)) as ProgressState); + return { success: true, output: { done: true } }; + }, + } as never); + + engine.registerFlow('aliasing_flow', flowDef('aliasing_flow') as never); + return { engine, observed }; +} + +/** + * Drive one arm all the way: pause → resume that mutates and throws → restore → + * resume again. Reports the pause's progress state as three independent + * readings, because a fix that satisfies only one of them is not a fix. + */ +async function drive(mode: Arm, store?: InMemorySuspendedRunStore) { + const { engine, observed } = build(mode, store); + + const started = await engine.execute('aliasing_flow', ctx); + expect(started.status).toBe('paused'); + const runId = started.runId as string; + + // What the pause actually parked — read before any resume can touch it. + const parked = store ? await store.load(runId) : null; + + const failed = await engine.resume(runId); + expect(failed.success).toBe(false); + expect(await engine.hasSuspendedRun(runId)).toBe(false); + + const restored = await engine.restoreConsumedSuspension(runId, { requestedBy: 'ops@example.com' }); + expect(restored.restored).toBe(true); + + // Reading 1 — the durable row the restore re-parked. + const back = store ? await store.load(runId) : null; + // Reading 2 — end to end: what the node is handed on the resume after the repair. + const finished = await engine.resume(runId); + expect(finished.success).toBe(true); + + return { + parked: (parked?.variables?.[STATE_KEY] as ProgressState | undefined), + restored: (back?.variables?.[STATE_KEY] as ProgressState | undefined), + observedAfterRepair: observed[0], + }; +} + +describe('#15660 — the restored suspension carries the pause, not the failed attempt', () => { + it('an executor that mutates its scope state IN PLACE does not rewrite the parked snapshot', async () => { + const store = new InMemorySuspendedRunStore(); + const r = await drive('in-place', store); + + // The pause itself was always recorded correctly — the divergence is + // introduced later, which is why reading only the parked row missed it. + expect(r.parked).toEqual({ started: 1, results: ['at-suspend'] }); + + // ⭐ The card's question, measured: what does the operator get back? + expect(r.restored).toEqual({ started: 1, results: ['at-suspend'] }); + expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] }); + }); + + it('CONTROL — the same fixture that never touches the parked object reports the pause', async () => { + const store = new InMemorySuspendedRunStore(); + const r = await drive('replace', store); + + expect(r.parked).toEqual({ started: 1, results: ['at-suspend'] }); + expect(r.restored).toEqual({ started: 1, results: ['at-suspend'] }); + expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] }); + }); + + it('holds with NO durable store, where the engine map answers by identity', async () => { + const r = await drive('in-place'); + expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] }); + }); + + it('CONTROL — no durable store, replace-and-set', async () => { + const r = await drive('replace'); + expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] }); + }); + + /** + * The card's stated mechanism, pinned as FALSE so the next reader does not + * re-derive the fix from it. If this ever goes red the store started keeping + * identity, and "deep copy at snapshot time" stops being refuted — the + * reasoning in `cloneVariablesAtPause`'s call site would need re-deriving. + */ + it('the in-memory store does NOT keep object identity — it JSON round-trips', async () => { + const store = new InMemorySuspendedRunStore(); + const nested: ProgressState = { started: 1, results: ['x'] }; + await store.save({ + runId: 'r1', flowName: 'f', flowVersion: '1', nodeId: 'n', nodeType: 't', + variables: { [STATE_KEY]: nested }, steps: [], context: {} as never, + startedAt: new Date().toISOString(), startTime: Date.now(), + } as unknown as SuspendedRun); + + nested.started = 42; // mutate the caller's object AFTER the save + + const loaded = await store.load('r1'); + expect((loaded?.variables?.[STATE_KEY] as ProgressState).started).toBe(1); + }); +}); diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 8603e9cca6..f7a18c3c4e 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -5508,6 +5508,47 @@ export class AutomationEngine implements IAutomationService { // it leaves behind is `restoreConsumedSuspension`, and the state // itself is named on the result (`status: 'stranded'`, below). const stepCountAtPause = run.steps.length; + // [#15660] …and the variables AT THE PAUSE, for the same reason and + // at the same moment, because `run.steps.length` is not the only + // thing the failed attempt can move under the journal. + // + // `journalConsumedSuspension` promises its snapshot VERBATIM — "the + // state at the pause, not the state the failed attempt was working + // from" — and delivered that only for the TOP-LEVEL keys. The scope + // the downstream nodes run against is built one line below as + // `new Map(Object.entries(run.variables))`, which copies the keys + // and SHARES every value object. A node that holds state in the + // scope and updates it in place — `map` keeps `.$mapState` + // and is, by a re-derived census, the only executor that does — + // therefore writes straight through into the object the journal + // later hands an operator as the pause. + // + // Measured, not read (`consumed-suspension-snapshot-aliasing.test.ts`): + // the durable row held `started: 1` at the pause and the restore put + // back `started: 99`. ⛔ And the two placements the card proposed + // are both refuted by that same measurement: + // - "deep copy at SNAPSHOT time" — `InMemorySuspendedRunStore` + // already JSON round-trips on save AND load, so this run's + // variables ARE a private deep copy before a single downstream + // node runs. It reproduces anyway: the aliasing that carries the + // mutation is minted HERE, on the resume, not at the suspend. + // - "copy at RESTORE time" — the mutation lands before the node + // throws, so the journal is already corrupt when it is written; + // anything copied at or after that point copies the corruption. + // The copy has to be taken before the failed attempt runs, which is + // this line. + // + // Cost, measured rather than assumed (the card flagged it NOT + // MEASURED): 5.8 µs for a typical scope, and one suspend+resume + // round trip against a configured store already performs FOUR full + // clones of this same payload (save ×1, load ×2, recordTerminal ×1), + // so this is a fifth on a path whose production cost is a durable + // round trip. ⛔ Deliberately NOT the JSON clone the journal's own + // docblock ruled out: that objection was that it would run INSIDE a + // catch arm already handling a failure, where a circular value would + // throw the operator's repair away. Here it is on the happy path, + // where a throw is just a value we decline to copy. + const variablesAtPause = this.cloneVariablesAtPause(run, runId); // Consume the suspension *before* running downstream work — a run // resumes exactly once per pause, and a duplicate resume after a @@ -5760,7 +5801,12 @@ export class AutomationEngine implements IAutomationService { // stays `failed`, no suspension exists after this line, and the // ordering above is untouched. It is the evidence a repair // needs, written at the only moment it still exists. - const consumed = this.journalConsumedSuspension(run, stepCountAtPause, errorMessage); + const consumed = this.journalConsumedSuspension( + run, + stepCountAtPause, + errorMessage, + variablesAtPause, + ); // [#15555] From the line above, "this run is repairable" is a // FACT: a snapshot exists and `restoreConsumedSuspension` puts // it back. Everything from here to the `status: 'stranded'` @@ -6276,35 +6322,84 @@ export class AutomationEngine implements IAutomationService { return true; } + /** + * [#15660] The pause's variables, decoupled from the scope the resume is + * about to hand the downstream nodes. + * + * Called once per resume of a suspended run, BEFORE the failed attempt can + * run — see the call site for why no later placement works and for the + * measured cost. A JSON clone specifically, not `structuredClone`: the + * durable row for this same pause was written through `JSON` by the store, + * so this keeps the hot journal and the durable row the same shape. A + * `Date` that survived here as a `Date` while the row held a string would + * make `restoreConsumedSuspension`'s two sources disagree about the pause + * they both claim to describe. + * + * A value JSON cannot carry (a cycle) costs the copy, never the repair: the + * fallback is the aliased object this method exists to replace, which is + * exactly the behaviour that shipped before, so the operator exit is no + * worse than it was. It is reported at `warn` — the exit still works, and + * what it puts back may name post-resume state. + */ + private cloneVariablesAtPause(run: SuspendedRun, runId: string): SuspendedRun['variables'] { + try { + return JSON.parse(JSON.stringify(run.variables)) as SuspendedRun['variables']; + } catch (err) { + // #6299 family — the thrown text is not ours to shape, so it rides + // the structured slot and never the message. + this.logger.warn( + `[automation] run '${runId}': could not copy the paused variables before resuming, so if this ` + + `resume strands the run, the suspension an operator restores may carry state this attempt ` + + `wrote rather than the state at the pause. The run resumes normally either way.`, + describeThrownForLog(err), + ); + return run.variables; + } + } + /** * Record the suspension a resume consumed before its downstream node threw * (#13909) — the one and only producer of a {@link ConsumedSuspension}. * * VERBATIM, and that word is load-bearing: * - * - `run.variables` is the pause's OWN snapshot. The resume's signal was - * folded into a separate `Map` built from it, never into this object, so - * what is journalled is the state at the pause, not the state the failed - * attempt was working from. It is the same object the durable paused row - * was written from at suspend time. + * - `variablesAtPause` is the pause's OWN snapshot, deep-copied by + * {@link cloneVariablesAtPause} before the failed attempt ran. The + * resume's signal was folded into a separate `Map` built from + * `run.variables`, never into this object, so what is journalled is the + * state at the pause, not the state the failed attempt was working from. + * + * ⚠️ [#15660] That last sentence used to be written against + * `run.variables` itself, and was true only of its TOP-LEVEL keys. The + * scope the downstream nodes run against is `new Map(Object.entries( + * run.variables))` — the keys are copied, every value object is SHARED — + * so an executor that keeps state in the scope and updates it in place + * (`map`'s `.$mapState`) wrote through into this snapshot, and + * the operator exit handed back post-resume state stamped as the pause. + * Measured end to end in `consumed-suspension-snapshot-aliasing.test.ts`. + * ⛔ Do not route the copy back to this method: by the time this runs the + * mutation has already landed (the node mutates, THEN throws), so a copy + * taken here copies the corruption. It has to be taken before the failed + * attempt runs, which is where the caller takes it. * - `run.steps` is the live array `traverseNext` appended to, so it is * trimmed back to `stepCountAtPause` — the failed attempt's steps are * NOT part of the thing an operator puts back. * - Everything else (`nodeId`, `nodeType`, `context`, `correlation`, * `screen`, `startedAt`, `startTime`) is carried across untouched. * - * Shallow by design, not lazily: a JSON clone here could throw on a - * circular value INSIDE a catch arm that is already handling a failure, and - * the fields it would deep-copy are exactly the ones the durable store - * already round-tripped through JSON at suspend time. + * Still shallow HERE, and for the reason it always was: a JSON clone in + * this method could throw on a circular value inside a catch arm that is + * already handling a failure. The one field that needed copying is copied + * by the caller instead, on the happy path, where a throw costs nothing. */ private journalConsumedSuspension( run: SuspendedRun, stepCountAtPause: number, error: string, + variablesAtPause: SuspendedRun['variables'], ): ConsumedSuspension { const consumed: ConsumedSuspension = { - run: { ...run, steps: run.steps.slice(0, stepCountAtPause) }, + run: { ...run, steps: run.steps.slice(0, stepCountAtPause), variables: variablesAtPause }, consumedAt: new Date().toISOString(), error, // [#13937] `recordLog` settles this when its write settles. From d3ab0677673554881ad1a7742f021de4ec30a8ae Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 6 Sep 2026 17:31:09 +0000 Subject: [PATCH 2/2] chore: changeset for the suspend-snapshot aliasing fix Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .changeset/suspend-snapshot-aliasing.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 .changeset/suspend-snapshot-aliasing.md diff --git a/.changeset/suspend-snapshot-aliasing.md b/.changeset/suspend-snapshot-aliasing.md new file mode 100644 index 0000000000..4394da4ba2 --- /dev/null +++ b/.changeset/suspend-snapshot-aliasing.md @@ -0,0 +1,13 @@ +--- +"@objectstack/service-automation": patch +--- + +A restored suspension now carries the state the run was paused with, including nested values. + +`restoreConsumedSuspension` is the operator exit from a run whose resume consumed the pause and then failed downstream: it puts the suspension back so the run is resumable again. What it put back was documented as the pause "verbatim", and was — for the top-level variables only. + +The flow scope a resume hands the downstream nodes was rebuilt as `new Map(Object.entries(run.variables))`: that copies the keys and shares every value object with the parked snapshot. An executor that keeps state in the scope and updates it **in place** — `map` tracks its progress in `.$mapState` — therefore wrote straight through into the snapshot, and the journal recorded the result as the pause. An operator repairing a stranded `map` run got a snapshot claiming progress made by the attempt that failed, not the progress the run actually had when it paused. + +Measured, not inferred: the durable row held `started: 1` at the pause and the restore put back `started: 99`. + +The pause's variables are now copied before the failed attempt runs, on the line that already captures the pause's step count for the same reason. No later placement works — the node mutates and then throws, so a copy taken when the journal is written copies the mutation. Nothing else changes: the running flow still sees exactly the scope it saw before, the resume ordering is untouched, and a value that cannot be copied falls back to the previous behaviour with a warning rather than costing the operator the repair.