Skip to content

Commit 836fb42

Browse files
committed
test(service-automation): pin that a failing parallel keeps every branch's steps
Red at base: a parallel whose branch fails returns no childSteps, so the run log drops the delivered sibling's step and the failing branch's own step, and the run summary reports acted 0. Claude-Session: https://claude.ai/code/session_018VqJPGA1NDNSCZsnecsChP Co-authored-by: Claude <noreply@anthropic.com>
1 parent 7277c07 commit 836fb42

1 file changed

Lines changed: 292 additions & 1 deletion

File tree

‎packages/services/service-automation/src/builtin/parallel-node.test.ts‎

Lines changed: 292 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,9 @@
22

33
import { describe, it, expect, beforeEach } from 'vitest';
44
import { AutomationEngine } from '../engine.js';
5-
import type { NodeExecutor } from '../engine.js';
5+
import type { NodeExecutor, StepLogEntry } from '../engine.js';
66
import { registerParallelNode } from './parallel-node.js';
7+
import { registerLoopNode } from './loop-node.js';
78

89
function silentLogger() {
910
return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } } as any;
@@ -172,4 +173,294 @@ describe('parallel block executor (ADR-0031)', () => {
172173
])),
173174
).toThrow(/parallel 'par' branch 1/);
174175
});
176+
177+
/**
178+
* A failing `parallel` used to drop every branch's steps from the run log.
179+
*
180+
* The block returned `{ success: false }` with no `childSteps`, so the run
181+
* log kept the container's own `failure` step and nothing beneath it: not
182+
* the failing branch's step, and not the steps of a SIBLING branch that had
183+
* already completed — and had side effects. The run summary folded over that
184+
* log then reported `acted: 0` for a run that had delivered a notification,
185+
* and `acted: 0` on a failed run reads as "nothing happened, safe to re-run".
186+
* The same harm the dying-`loop` fix closed (`loop-dying-body-steps.test.ts`).
187+
*
188+
* The pins are on the run log and the summary an operator reads, and on the
189+
* DIRECTION of `acted`: never lower than the effects that really happened.
190+
*/
191+
describe('a failing parallel keeps every branch\'s steps in the run log', () => {
192+
/** Stands in for the store: every push is an effect that really happened. */
193+
let delivered: string[];
194+
/** What `$error` named when the `fault` handler ran. */
195+
let handledError: unknown;
196+
197+
/** `nodeId@parentNodeId/branch:status` — one line per branch step, in log order. */
198+
const branchLog = (steps: readonly StepLogEntry[]) =>
199+
steps
200+
.filter(s => s.regionKind === 'parallel-branch')
201+
.map(s => `${s.nodeId}@${s.parentNodeId}/${s.branch}:${s.status}`);
202+
203+
const runOnce = async (name = 'par_flow') => {
204+
const result = await engine.execute(name);
205+
const record = (await engine.listRuns(name))[0];
206+
return { result, steps: record?.steps ?? [] };
207+
};
208+
209+
beforeEach(() => {
210+
delivered = [];
211+
handledError = undefined;
212+
213+
// Stands in for `notify`: a non-idempotent effect that reports its own
214+
// count (#4354). `delayMs` makes it settle on a LATER macrotask than a
215+
// sibling that fails at once, so "completed after the failure" is real.
216+
engine.registerNodeExecutor({
217+
type: 'deliver',
218+
async execute(node) {
219+
const cfg = (node.config ?? {}) as Record<string, unknown>;
220+
const delayMs = Number(cfg.delayMs ?? 0);
221+
if (delayMs > 0) await new Promise(r => setTimeout(r, delayMs));
222+
delivered.push(node.id);
223+
order.push(node.id);
224+
return { success: true, metrics: { selected: 1, acted: 1 } };
225+
},
226+
} as NodeExecutor);
227+
228+
// A `fault` handler that records what `$error` named when it ran.
229+
engine.registerNodeExecutor({
230+
type: 'handler',
231+
async execute(_node, variables) {
232+
order.push('handler');
233+
handledError = variables.get('$error');
234+
return { success: true };
235+
},
236+
} as NodeExecutor);
237+
238+
// A branch node that asks to pause. It publishes no descriptor, so the
239+
// engine's `supportsPause` gate has nothing to enforce and the REGION
240+
// boundary is what refuses the pause — the named region refusal the
241+
// `parallel` re-throws rather than returns.
242+
engine.registerNodeExecutor({
243+
type: 'pause',
244+
async execute() { return { success: true, suspend: true }; },
245+
} as NodeExecutor);
246+
});
247+
248+
it('keeps the delivered sibling AND the failing branch\'s own step, each tagged with its branch', async () => {
249+
engine.registerFlow('par_flow', parallelFlow([
250+
{ name: 'notify', nodes: [{ id: 'notify_owner', type: 'deliver', label: 'Notify owner' }], edges: [] },
251+
{ name: 'post', nodes: [{ id: 'x', type: 'boom', label: 'Post' }], edges: [] },
252+
]));
253+
254+
const { result, steps } = await runOnce();
255+
256+
// The premise: the block really failed, and the sibling really delivered.
257+
expect(result.success).toBe(false);
258+
expect(delivered).toEqual(['notify_owner']);
259+
260+
expect(branchLog(steps)).toEqual([
261+
'notify_owner@par/0:success',
262+
'x@par/1:failure',
263+
]);
264+
// The failing step carries its own error — "what failed" is answerable.
265+
const failing = steps.find(s => s.nodeId === 'x');
266+
expect(failing?.error?.code).toBe('NODE_FAILURE');
267+
expect(failing?.error?.message).toContain('branch kaboom');
268+
});
269+
270+
it('the run summary counts the delivered branch — `acted` never lower than the effects that happened', async () => {
271+
engine.registerFlow('par_flow', parallelFlow([
272+
{ name: 'notify', nodes: [{ id: 'notify_owner', type: 'deliver', label: 'Notify owner' }], edges: [] },
273+
{ name: 'post', nodes: [{ id: 'x', type: 'boom', label: 'Post' }], edges: [] },
274+
]));
275+
276+
const { result } = await runOnce();
277+
278+
expect(result.success).toBe(false);
279+
expect(result.summary?.acted).toBeGreaterThanOrEqual(delivered.length);
280+
// Exactly reconciled: nothing invented, nothing folded twice.
281+
expect(result.summary).toMatchObject({ selected: 1, acted: 1 });
282+
const byId = new Map((result.summary?.nodes ?? []).map(n => [n.nodeId, n]));
283+
expect(byId.get('notify_owner')).toMatchObject({ runs: 1, failures: 0, acted: 1 });
284+
expect(byId.get('x')).toMatchObject({ runs: 1, failures: 1 });
285+
// The container's own record is unchanged: one execution, one failure.
286+
expect(byId.get('par')).toMatchObject({ status: 'failure', runs: 1, failures: 1 });
287+
});
288+
289+
it('keeps the failing branch\'s steps up to AND including its failing step, and nothing after it', async () => {
290+
engine.registerFlow('par_flow', parallelFlow([
291+
{ name: 'solo', nodes: [{ id: 'd', type: 'deliver', label: 'D' }], edges: [] },
292+
{
293+
name: 'chain',
294+
nodes: [
295+
{ id: 'c1', type: 'deliver', label: 'C1' },
296+
{ id: 'x', type: 'boom', label: 'X' },
297+
{ id: 'c3', type: 'deliver', label: 'C3' },
298+
],
299+
edges: [
300+
{ id: 'ce1', source: 'c1', target: 'x' },
301+
{ id: 'ce2', source: 'x', target: 'c3' },
302+
],
303+
},
304+
]));
305+
306+
const { result, steps } = await runOnce();
307+
308+
expect(delivered.sort()).toEqual(['c1', 'd']);
309+
// Branch order is the declaration order, as on the success path.
310+
expect(branchLog(steps)).toEqual([
311+
'd@par/0:success',
312+
'c1@par/1:success',
313+
'x@par/1:failure',
314+
]);
315+
expect(result.summary?.acted).toBe(delivered.length);
316+
});
317+
318+
it('keeps a sibling that completes AFTER the failing branch has already failed', async () => {
319+
engine.registerFlow('par_flow', parallelFlow([
320+
{ name: 'slow', nodes: [{ id: 'slow', type: 'deliver', label: 'Slow', config: { delayMs: 15 } }], edges: [] },
321+
{ name: 'bad', nodes: [{ id: 'x', type: 'boom', label: 'X' }], edges: [] },
322+
]));
323+
324+
const { result, steps } = await runOnce();
325+
326+
expect(result.success).toBe(false);
327+
expect(delivered).toEqual(['slow']);
328+
expect(branchLog(steps)).toEqual([
329+
'slow@par/0:success',
330+
'x@par/1:failure',
331+
]);
332+
expect(result.summary?.acted).toBe(1);
333+
});
334+
335+
it('a `fault` edge on the block routes exactly as before; the branch steps sit between the block and its handler', async () => {
336+
engine.registerFlow('par_flow', {
337+
name: 'par_flow',
338+
label: 'Parallel Flow',
339+
type: 'autolaunched' as const,
340+
nodes: [
341+
{ id: 'start', type: 'start', label: 'Start' },
342+
{
343+
id: 'par', type: 'parallel', label: 'Fan out',
344+
config: {
345+
branches: [
346+
{ name: 'notify', nodes: [{ id: 'notify_owner', type: 'deliver', label: 'Notify owner' }], edges: [] },
347+
{ name: 'post', nodes: [{ id: 'x', type: 'boom', label: 'Post' }], edges: [] },
348+
],
349+
},
350+
},
351+
{ id: 'join', type: 'after', label: 'After' },
352+
{ id: 'recover', type: 'handler', label: 'Recover' },
353+
{ id: 'end', type: 'end', label: 'End' },
354+
],
355+
edges: [
356+
{ id: 'e1', source: 'start', target: 'par' },
357+
{ id: 'e2', source: 'par', target: 'join' },
358+
{ id: 'e3', source: 'join', target: 'end' },
359+
{ id: 'ef', source: 'par', target: 'recover', type: 'fault' },
360+
],
361+
});
362+
363+
const { result, steps } = await runOnce();
364+
365+
// Routing is unchanged: the returned failure is routable, the handler
366+
// runs, the join does not, and the handled run completes.
367+
expect(result.success).toBe(true);
368+
expect(order).toContain('handler');
369+
expect(order).not.toContain('after');
370+
// `$error` names the block, as a returned failure always has.
371+
expect((handledError as { nodeId?: string } | undefined)?.nodeId).toBe('par');
372+
373+
// The block's own step keeps the returned-failure shape.
374+
const parStep = steps.find(s => s.nodeId === 'par' && s.parentNodeId === undefined);
375+
expect(parStep?.status).toBe('failure');
376+
expect(parStep?.error?.code).toBe('NODE_FAILURE');
377+
378+
// Position: right behind the block's own step, ahead of the handler's.
379+
const ids = steps.map(s => s.nodeId);
380+
expect(ids.slice(ids.indexOf('par'), ids.indexOf('recover') + 1)).toEqual([
381+
'par', 'notify_owner', 'x', 'recover',
382+
]);
383+
});
384+
385+
it('reverse control: branches that fail before any effect still report acted: 0', async () => {
386+
engine.registerFlow('par_flow', parallelFlow([
387+
{ name: 'bad_a', nodes: [{ id: 'xa', type: 'boom', label: 'XA' }], edges: [] },
388+
{ name: 'bad_b', nodes: [{ id: 'xb', type: 'boom', label: 'XB' }], edges: [] },
389+
]));
390+
391+
const { result, steps } = await runOnce();
392+
393+
expect(result.success).toBe(false);
394+
expect(delivered).toEqual([]);
395+
// Both failures are on the record; the count stays the honest zero.
396+
expect(branchLog(steps)).toEqual(['xa@par/0:failure', 'xb@par/1:failure']);
397+
expect(result.summary?.acted).toBe(0);
398+
});
399+
400+
it('a pause refused at a branch boundary still fails the run, and keeps the sibling\'s steps', async () => {
401+
engine.registerFlow('par_flow', parallelFlow([
402+
{ name: 'notify', nodes: [{ id: 'notify_owner', type: 'deliver', label: 'Notify owner' }], edges: [] },
403+
{ name: 'wait', nodes: [{ id: 'hold', type: 'pause', label: 'Hold' }], edges: [] },
404+
]));
405+
406+
const { result, steps } = await runOnce();
407+
408+
// The region refusal reaches the run as before: failed, not paused, and
409+
// the join never ran.
410+
expect(result.success).toBe(false);
411+
expect(result.status).toBe('failed');
412+
expect(order).not.toContain('after');
413+
expect(branchLog(steps)).toContain('notify_owner@par/0:success');
414+
expect(result.summary?.acted).toBe(delivered.length);
415+
// Still ONE failure: the refusal is counted on the block, once.
416+
expect(result.summary?.failed).toBe(1);
417+
});
418+
419+
it('nested in a loop, every branch step reaches the log exactly once', async () => {
420+
registerLoopNode(engine, ctx());
421+
engine.registerFlow('nest_flow', {
422+
name: 'nest_flow',
423+
label: 'Nested',
424+
type: 'autolaunched' as const,
425+
nodes: [
426+
{ id: 'start', type: 'start', label: 'Start' },
427+
{
428+
id: 'each', type: 'loop', label: 'Each',
429+
config: {
430+
collection: [1, 2],
431+
iteratorVariable: 'row',
432+
body: {
433+
nodes: [{
434+
id: 'par', type: 'parallel', label: 'Fan out',
435+
config: {
436+
branches: [
437+
{ name: 'notify', nodes: [{ id: 'notify_owner', type: 'deliver', label: 'Notify owner' }], edges: [] },
438+
{ name: 'post', nodes: [{ id: 'x', type: 'boom', label: 'Post' }], edges: [] },
439+
],
440+
},
441+
}],
442+
edges: [],
443+
},
444+
},
445+
},
446+
{ id: 'end', type: 'end', label: 'End' },
447+
],
448+
edges: [
449+
{ id: 'e1', source: 'start', target: 'each' },
450+
{ id: 'e2', source: 'each', target: 'end' },
451+
],
452+
});
453+
454+
const { result, steps } = await runOnce('nest_flow');
455+
456+
// The first row's block fails, so the loop dies on iteration 0.
457+
expect(result.success).toBe(false);
458+
expect(delivered).toEqual(['notify_owner']);
459+
expect(branchLog(steps)).toEqual(['notify_owner@par/0:success', 'x@par/1:failure']);
460+
expect(steps.filter(s => s.nodeId === 'par')).toHaveLength(1);
461+
expect(new Set(steps).size).toBe(steps.length);
462+
// A step folded twice would push `acted` above the real count.
463+
expect(result.summary?.acted).toBe(delivered.length);
464+
});
465+
});
175466
});

0 commit comments

Comments
 (0)