Skip to content

Commit 3890244

Browse files
os-warrenclaude
andauthored
fix(service-automation): a restored suspension carries the pause, not the failed attempt (#16396)
* fix(service-automation): journal the pause's variables, not the failed attempt's `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 `<nodeId>.$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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y * chore: changeset for the suspend-snapshot aliasing fix Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 4177ed3 commit 3890244

3 files changed

Lines changed: 346 additions & 11 deletions

File tree

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
---
2+
"@objectstack/service-automation": patch
3+
---
4+
5+
A restored suspension now carries the state the run was paused with, including nested values.
6+
7+
`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.
8+
9+
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 `<nodeId>.$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.
10+
11+
Measured, not inferred: the durable row held `started: 1` at the pause and the restore put back `started: 99`.
12+
13+
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.
Lines changed: 227 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,227 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
/**
4+
* #15660 — **the suspension an operator restores must be the pause, not the
5+
* attempt that failed** — for NESTED values, which is where it was not.
6+
*
7+
* ## The state under test
8+
*
9+
* `resumeInternal` rebuilds the flow scope as
10+
* `new Map(Object.entries(run.variables))`: the KEYS are copied, every VALUE
11+
* OBJECT is shared with `run.variables`. `journalConsumedSuspension` then
12+
* shallow-copies `run` and journals it as the pause "VERBATIM". So a node that
13+
* keeps state in the scope and updates it **in place** writes straight through
14+
* into the snapshot the operator exit later hands back — and
15+
* `restoreConsumedSuspension` re-arms a pause carrying state that belongs to
16+
* the failed attempt.
17+
*
18+
* `map` is the concrete instance and, by a census re-derived here rather than
19+
* recalled (nothing else in the tree writes a `${node.id}.$…` object into the
20+
* scope), the only executor in-repo that does it. The seam under test is the
21+
* ENGINE's, not `map`'s, so the fixture is a minimal executor with the same
22+
* shape: coupling this pin to `map`'s internals would make it fail for reasons
23+
* that are not this card.
24+
*
25+
* ## ⚠️ The card was filed as a READING, and the reading's mechanism was wrong
26+
*
27+
* The card attributed the aliasing to "the in-memory suspended-run store keeps
28+
* the object by identity rather than serialising it". It does not — it JSON
29+
* round-trips on both save and load, and the last test here pins that. The
30+
* defect reproduces anyway, because the aliasing that carries the mutation is
31+
* minted on the RESUME (the scope rebuild above), not at the suspend. That is
32+
* also why "deep copy at snapshot time" cannot be the fix: with a store
33+
* configured the snapshot already IS a private deep copy, and it still
34+
* reproduced.
35+
*
36+
* ## Why the controls are in the file and not in a scratch buffer
37+
*
38+
* Each in-place arm is paired with a replace-and-set arm that must stay green.
39+
* Without them a harness that reported the pause's value for a trivial reason —
40+
* a fixture whose mutation never ran, an assertion on the wrong key — would read
41+
* exactly like a fixed engine. The pair is what makes either reading mean
42+
* something.
43+
*/
44+
45+
import { describe, it, expect } from 'vitest';
46+
47+
import { AutomationEngine, type SuspendedRun, type SuspendedRunStore } from './engine.js';
48+
import { InMemorySuspendedRunStore } from './suspended-run-store.js';
49+
import type { AutomationContext } from '@objectstack/spec/contracts';
50+
import { defineActionDescriptor } from '@objectstack/spec/automation';
51+
52+
const silent = { info() {}, warn() {}, error() {}, debug() {} } as never;
53+
54+
/** The shape `map` keeps: node-scoped progress state living in the flow scope. */
55+
const STATE_KEY = 'worker.$mapState';
56+
57+
type ProgressState = { started: number; results: unknown[] };
58+
59+
const pauser = (type: string) => defineActionDescriptor({
60+
type, version: '1.0.0', name: type, supportsPause: true, resumeAuthority: 'any',
61+
});
62+
const plain = (type: string) => defineActionDescriptor({ type, version: '1.0.0', name: type });
63+
64+
function flowDef(name: string) {
65+
return {
66+
name, label: name, type: 'autolaunched',
67+
variables: [{ name: 'ticket', type: 'text', isInput: true, isOutput: true }],
68+
nodes: [
69+
{ id: 'start', type: 'start', label: 'Start' },
70+
{ id: 'seed', type: 'seed_state', label: 'Seed' },
71+
{ id: 'pause', type: 'pause_here', label: 'Pause' },
72+
{ id: 'worker', type: 'worker', label: 'Worker' },
73+
{ id: 'end', type: 'end', label: 'End' },
74+
],
75+
edges: [
76+
{ id: 'e1', source: 'start', target: 'seed' },
77+
{ id: 'e2', source: 'seed', target: 'pause' },
78+
{ id: 'e3', source: 'pause', target: 'worker' },
79+
{ id: 'e4', source: 'worker', target: 'end' },
80+
],
81+
};
82+
}
83+
84+
const ctx = { event: 'test', record: { id: 'rec_1' }, params: { ticket: 'TKT-9' } } as unknown as AutomationContext;
85+
86+
/**
87+
* `in-place` is the executor pattern under test. `replace` is its control: the
88+
* same fixture, the same mutation, the same throw — only the parked object is
89+
* left alone, so it must report the pause under a fixed engine AND under a
90+
* broken one.
91+
*/
92+
type Arm = 'in-place' | 'replace';
93+
94+
function build(mode: Arm, store?: SuspendedRunStore) {
95+
const engine = new AutomationEngine(silent as never, store);
96+
const entry = { first: true };
97+
const observed: ProgressState[] = [];
98+
99+
engine.registerNodeExecutor({
100+
type: 'seed_state', descriptor: plain('seed_state'),
101+
async execute(_node: unknown, variables: Map<string, unknown>) {
102+
variables.set(STATE_KEY, { started: 1, results: ['at-suspend'] } satisfies ProgressState);
103+
return { success: true };
104+
},
105+
} as never);
106+
107+
engine.registerNodeExecutor({
108+
type: 'pause_here', descriptor: pauser('pause_here'),
109+
async execute() {
110+
return { success: true, suspend: true, correlation: 'approval:req_1', output: { stage: 'awaiting' } };
111+
},
112+
} as never);
113+
114+
engine.registerNodeExecutor({
115+
type: 'worker', descriptor: plain('worker'),
116+
async execute(_node: unknown, variables: Map<string, unknown>) {
117+
const state = variables.get(STATE_KEY) as ProgressState;
118+
if (entry.first) {
119+
entry.first = false;
120+
if (mode === 'in-place') {
121+
state.started = 99;
122+
state.results.push('post-resume');
123+
} else {
124+
variables.set(STATE_KEY, { started: 99, results: [...state.results, 'post-resume'] });
125+
}
126+
// Strand the run: the pause is already consumed, so this throw
127+
// is what makes `restoreConsumedSuspension` the only way out.
128+
throw new Error('downstream node blew up');
129+
}
130+
observed.push(JSON.parse(JSON.stringify(state)) as ProgressState);
131+
return { success: true, output: { done: true } };
132+
},
133+
} as never);
134+
135+
engine.registerFlow('aliasing_flow', flowDef('aliasing_flow') as never);
136+
return { engine, observed };
137+
}
138+
139+
/**
140+
* Drive one arm all the way: pause → resume that mutates and throws → restore →
141+
* resume again. Reports the pause's progress state as three independent
142+
* readings, because a fix that satisfies only one of them is not a fix.
143+
*/
144+
async function drive(mode: Arm, store?: InMemorySuspendedRunStore) {
145+
const { engine, observed } = build(mode, store);
146+
147+
const started = await engine.execute('aliasing_flow', ctx);
148+
expect(started.status).toBe('paused');
149+
const runId = started.runId as string;
150+
151+
// What the pause actually parked — read before any resume can touch it.
152+
const parked = store ? await store.load(runId) : null;
153+
154+
const failed = await engine.resume(runId);
155+
expect(failed.success).toBe(false);
156+
expect(await engine.hasSuspendedRun(runId)).toBe(false);
157+
158+
const restored = await engine.restoreConsumedSuspension(runId, { requestedBy: 'ops@example.com' });
159+
expect(restored.restored).toBe(true);
160+
161+
// Reading 1 — the durable row the restore re-parked.
162+
const back = store ? await store.load(runId) : null;
163+
// Reading 2 — end to end: what the node is handed on the resume after the repair.
164+
const finished = await engine.resume(runId);
165+
expect(finished.success).toBe(true);
166+
167+
return {
168+
parked: (parked?.variables?.[STATE_KEY] as ProgressState | undefined),
169+
restored: (back?.variables?.[STATE_KEY] as ProgressState | undefined),
170+
observedAfterRepair: observed[0],
171+
};
172+
}
173+
174+
describe('#15660 — the restored suspension carries the pause, not the failed attempt', () => {
175+
it('an executor that mutates its scope state IN PLACE does not rewrite the parked snapshot', async () => {
176+
const store = new InMemorySuspendedRunStore();
177+
const r = await drive('in-place', store);
178+
179+
// The pause itself was always recorded correctly — the divergence is
180+
// introduced later, which is why reading only the parked row missed it.
181+
expect(r.parked).toEqual({ started: 1, results: ['at-suspend'] });
182+
183+
// ⭐ The card's question, measured: what does the operator get back?
184+
expect(r.restored).toEqual({ started: 1, results: ['at-suspend'] });
185+
expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] });
186+
});
187+
188+
it('CONTROL — the same fixture that never touches the parked object reports the pause', async () => {
189+
const store = new InMemorySuspendedRunStore();
190+
const r = await drive('replace', store);
191+
192+
expect(r.parked).toEqual({ started: 1, results: ['at-suspend'] });
193+
expect(r.restored).toEqual({ started: 1, results: ['at-suspend'] });
194+
expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] });
195+
});
196+
197+
it('holds with NO durable store, where the engine map answers by identity', async () => {
198+
const r = await drive('in-place');
199+
expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] });
200+
});
201+
202+
it('CONTROL — no durable store, replace-and-set', async () => {
203+
const r = await drive('replace');
204+
expect(r.observedAfterRepair).toEqual({ started: 1, results: ['at-suspend'] });
205+
});
206+
207+
/**
208+
* The card's stated mechanism, pinned as FALSE so the next reader does not
209+
* re-derive the fix from it. If this ever goes red the store started keeping
210+
* identity, and "deep copy at snapshot time" stops being refuted — the
211+
* reasoning in `cloneVariablesAtPause`'s call site would need re-deriving.
212+
*/
213+
it('the in-memory store does NOT keep object identity — it JSON round-trips', async () => {
214+
const store = new InMemorySuspendedRunStore();
215+
const nested: ProgressState = { started: 1, results: ['x'] };
216+
await store.save({
217+
runId: 'r1', flowName: 'f', flowVersion: '1', nodeId: 'n', nodeType: 't',
218+
variables: { [STATE_KEY]: nested }, steps: [], context: {} as never,
219+
startedAt: new Date().toISOString(), startTime: Date.now(),
220+
} as unknown as SuspendedRun);
221+
222+
nested.started = 42; // mutate the caller's object AFTER the save
223+
224+
const loaded = await store.load('r1');
225+
expect((loaded?.variables?.[STATE_KEY] as ProgressState).started).toBe(1);
226+
});
227+
});

0 commit comments

Comments
 (0)