Skip to content

Commit a7d5ac2

Browse files
claudeclaude
committed
fix(service-automation): a failing parallel keeps every branch's steps
A parallel whose branch fails now returns its failure with childSteps: the steps of the branches that completed and the failing branch's own steps up to and including the one that failed, each tagged with the block, regionKind 'parallel-branch' and its branch index. Each branch runs with its own runRegion partialSteps sink, and the block reports its failure once every branch has settled, so a sibling that finishes after the failure keeps its step too. The failure is still RETURNED, so a fault edge on the block routes exactly as before, with the same step code, error text and $error; the engine already folds a returned failure's childSteps behind the block's own step. A region-suspension refusal is still re-thrown unchanged, now carrying the branch steps through attachPartialSteps, as loop does. Adds a real-engine pin: the real notify node on an inline messaging service and the real http node failing with fetch failed, showing the run log keeps both branch steps and the summary counts the delivery. Claude-Session: https://claude.ai/code/session_018VqJPGA1NDNSCZsnecsChP Co-authored-by: Claude <noreply@anthropic.com>
1 parent 836fb42 commit a7d5ac2

2 files changed

Lines changed: 245 additions & 31 deletions

File tree

‎packages/services/service-automation/src/builtin/parallel-failed-branch-steps.integration.test.ts‎

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
import { describe, it, expect, afterEach, vi } from 'vitest';
4+
import { MessagingService } from '@objectstack/service-messaging';
5+
import type { MessagingChannel } from '@objectstack/service-messaging';
6+
import { AutomationEngine } from '../engine.js';
7+
import { registerParallelNode } from './parallel-node.js';
8+
import { registerNotifyNode } from './notify-node.js';
9+
import { registerHttpNodes } from './http-nodes.js';
10+
11+
/**
12+
* A failed `parallel` run on the REAL engine and the REAL built-in nodes: the
13+
* run log and the run summary an operator reads after a fan-out whose one
14+
* branch delivered a notification and whose other branch's HTTP call failed.
15+
*
16+
* The graph is the showcase's `showcase_fan_out_notify` fan-out
17+
* (`examples/app-showcase/src/automation/flows/index.ts`): a `parallel` named
18+
* `fan_out` whose branch 0 is an inbox `notify` (`notify_owner`) and whose
19+
* branch 1 is an `http` POST (`slack_post`). On a runner with no Slack egress
20+
* the POST fails with undici's `fetch failed`, which is what the global
21+
* `fetch` double below answers — the only stand-in here. The notify node runs
22+
* against a real inline `MessagingService` with a recording `inbox` channel,
23+
* so "the notification was delivered" is read off the channel, not assumed.
24+
*
25+
* Before the fix this run's log was `[start:success, fan_out:failure]` and its
26+
* summary `{ selected: 0, acted: 0, failed: 1 }`, while the inbox held the
27+
* notification: an operator reading `acted: 0` re-runs the flow and the owner
28+
* is notified twice.
29+
*/
30+
31+
function silentLogger(): any {
32+
const l: any = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} };
33+
l.child = () => l;
34+
return l;
35+
}
36+
37+
/** A channel that records what it was handed, so a real send is observable. */
38+
function recordingChannel(id: string): { channel: MessagingChannel; sent: unknown[] } {
39+
const sent: unknown[] = [];
40+
return {
41+
sent,
42+
channel: {
43+
id,
44+
async send(_ctx, delivery) {
45+
sent.push(delivery);
46+
return { ok: true };
47+
},
48+
},
49+
};
50+
}
51+
52+
function boot() {
53+
const inbox = recordingChannel('inbox');
54+
const messaging = new MessagingService({ logger: silentLogger() });
55+
messaging.registerChannel(inbox.channel);
56+
const pluginCtx = {
57+
logger: silentLogger(),
58+
getService: (name: string) => (name === 'messaging' ? messaging : undefined),
59+
} as any;
60+
const engine = new AutomationEngine(silentLogger());
61+
registerParallelNode(engine, pluginCtx);
62+
registerNotifyNode(engine, pluginCtx);
63+
registerHttpNodes(engine, pluginCtx);
64+
return { engine, inbox };
65+
}
66+
67+
const FAN_OUT_FLOW = {
68+
name: 'fan_out_notify',
69+
label: 'Fan-out Notify (Parallel)',
70+
type: 'autolaunched' as const,
71+
nodes: [
72+
{ id: 'start', type: 'start' as const, label: 'Start' },
73+
{
74+
id: 'fan_out',
75+
type: 'parallel' as const,
76+
label: 'Notify in parallel',
77+
config: {
78+
branches: [
79+
{
80+
name: 'Notify the owner',
81+
nodes: [
82+
{
83+
id: 'notify_owner',
84+
type: 'notify',
85+
label: 'Notify Owner',
86+
config: { recipients: ['user_1'], title: 'Done: task', channels: ['inbox'] },
87+
},
88+
],
89+
edges: [],
90+
},
91+
{
92+
name: 'Post to Slack',
93+
nodes: [
94+
{
95+
id: 'slack_post',
96+
type: 'http',
97+
label: 'Slack Notify',
98+
config: {
99+
url: 'https://hooks.slack.com/services/T000/B000/XXXX',
100+
method: 'POST',
101+
},
102+
},
103+
],
104+
edges: [],
105+
},
106+
],
107+
},
108+
},
109+
{ id: 'end', type: 'end' as const, label: 'End' },
110+
],
111+
edges: [
112+
{ id: 'e1', source: 'start', target: 'fan_out' },
113+
{ id: 'e2', source: 'fan_out', target: 'end' },
114+
],
115+
};
116+
117+
describe('a failed parallel run keeps its branch steps on the real engine', () => {
118+
afterEach(() => {
119+
vi.unstubAllGlobals();
120+
});
121+
122+
const runFanOut = async () => {
123+
vi.stubGlobal('fetch', vi.fn(async () => {
124+
throw new TypeError('fetch failed');
125+
}));
126+
const { engine, inbox } = boot();
127+
engine.registerFlow('fan_out_notify', FAN_OUT_FLOW);
128+
const result = await engine.execute('fan_out_notify');
129+
const record = (await engine.listRuns('fan_out_notify'))[0];
130+
return { result, record, inbox };
131+
};
132+
133+
it('the step log carries the delivered notify and the failed http, each tagged with its branch', async () => {
134+
const { result, record, inbox } = await runFanOut();
135+
136+
// The premise: the run failed, and branch 0 really delivered.
137+
expect(result.success).toBe(false);
138+
expect(record?.status).toBe('failed');
139+
expect(inbox.sent).toHaveLength(1);
140+
141+
const steps = record?.steps ?? [];
142+
const branchSteps = steps.filter(s => s.parentNodeId === 'fan_out');
143+
expect(branchSteps.map(s => ({
144+
nodeId: s.nodeId,
145+
status: s.status,
146+
regionKind: s.regionKind,
147+
branch: s.branch,
148+
}))).toEqual([
149+
{ nodeId: 'notify_owner', status: 'success', regionKind: 'parallel-branch', branch: 0 },
150+
{ nodeId: 'slack_post', status: 'failure', regionKind: 'parallel-branch', branch: 1 },
151+
]);
152+
// The failed branch step carries its own error.
153+
const post = branchSteps.find(s => s.nodeId === 'slack_post');
154+
expect(post?.error?.code).toBe('NODE_FAILURE');
155+
expect(post?.error?.message).toContain('fetch failed');
156+
});
157+
158+
it('the run summary counts the delivered notification instead of reporting acted 0', async () => {
159+
const { result, record, inbox } = await runFanOut();
160+
161+
expect(inbox.sent).toHaveLength(1);
162+
// Never lower than the deliveries that happened.
163+
expect(result.summary?.acted).toBeGreaterThanOrEqual(inbox.sent.length);
164+
// The persisted run row reads the same as the live result.
165+
expect(record?.summary).toMatchObject({ selected: 1, acted: 1 });
166+
expect(result.summary).toMatchObject({ selected: 1, acted: 1 });
167+
// The failed POST is an effect the run cannot count, and says so.
168+
expect(result.summary?.unmeasured).toBe(1);
169+
// Two failures, each on its own node: the branch's POST and the block.
170+
const byId = new Map((result.summary?.nodes ?? []).map(n => [n.nodeId, n]));
171+
expect(byId.get('notify_owner')).toMatchObject({ runs: 1, failures: 0, acted: 1 });
172+
expect(byId.get('slack_post')).toMatchObject({ runs: 1, failures: 1 });
173+
expect(byId.get('fan_out')).toMatchObject({ runs: 1, failures: 1, status: 'failure' });
174+
expect(result.summary?.failed).toBe(2);
175+
});
176+
});

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

Lines changed: 69 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import type { AutomationContext } from '@objectstack/spec/contracts';
77
import type { AutomationEngine, StepLogEntry } from '../engine.js';
88
import { parseNodeConfig } from './parse-config.js';
99
import { isRegionSuspensionRefusal } from '../region-suspension-refusal.js';
10+
import { attachPartialSteps } from '../partial-steps.js';
1011

1112
/**
1213
* `parallel` built-in node — a **structured parallel block** with an
@@ -27,8 +28,10 @@ import { isRegionSuspensionRefusal } from '../region-suspension-refusal.js';
2728
*
2829
* If any branch fails (a node returns `success: false` or throws), the block
2930
* fails — surfaced as a node failure so the flow's fault edge / error handling
30-
* applies. Durable pause inside a branch is unsupported (a clear error), mirror-
31-
* ing the loop container.
31+
* applies. The failure is reported once every branch has settled, and it
32+
* carries every branch's steps: the siblings that completed, and the failing
33+
* branch's own steps up to and including the one that failed. Durable pause
34+
* inside a branch is unsupported (a clear error), mirroring the loop container.
3235
*/
3336
export function registerParallelNode(engine: AutomationEngine, ctx: PluginContext): void {
3437
engine.registerNodeExecutor({
@@ -69,44 +72,79 @@ export function registerParallelNode(engine: AutomationEngine, ctx: PluginContex
6972
if (!parsed.ok) return parsed.refusal;
7073
const branches = parsed.config.branches;
7174

72-
let branchSteps: StepLogEntry[][];
73-
try {
74-
// Implicit join: continue once when ALL branches have completed.
75-
// #1479: each branch returns its body steps, tagged with the branch index.
76-
//
77-
// #15230: the branch index goes on `branch`, NOT on `iteration`. It
78-
// used to share `iteration` with the loop's row index — one field, two
79-
// meanings, told apart only by reading `regionKind` first — and because
80-
// the tagger let the innermost region win outright, a branch step
81-
// inside a loop body recorded the branch and lost the row entirely.
82-
// `iteration` is now single-valued (the enclosing loop's, carried
83-
// through nesting by `runRegion`'s tagger), so nothing here writes it:
84-
// a branch step inside a loop body gets the row from the loop's own
85-
// tagging pass, and a bare `parallel` leaves it absent because there is
86-
// no enclosing loop to report. Maintainer ruling 2026-09-03.
87-
branchSteps = await Promise.all(
88-
branches.map((branchRegion, i) =>
89-
engine.runRegion(branchRegion, variables, context ?? ({} as AutomationContext), {
90-
parentNodeId: node.id,
91-
branch: i,
92-
regionKind: 'parallel-branch',
93-
}),
94-
),
95-
);
96-
} catch (err) {
75+
// A failing block keeps the record of what its branches did. A branch
76+
// that completed may already have delivered a notification or called
77+
// out, and a run log without its step makes the #4354 summary report
78+
// `acted: 0` for that run — which reads as "nothing happened, safe to
79+
// re-run", the double-execution misread a dying `loop` was fixed for
80+
// (`partial-steps.ts`).
81+
//
82+
// One #7546 `partialSteps` sink PER BRANCH: `runRegion` pushes a failing
83+
// branch's tagged steps (its failing step included) into that branch's
84+
// sink before it rethrows, and a completed branch returns them instead,
85+
// so each step is collected exactly once and the log keeps declaration
86+
// order, as on the success path.
87+
//
88+
// The block waits for EVERY branch to settle before it reports a
89+
// failure: a sibling still running when another branch fails would
90+
// otherwise finish its effect after the block had already failed, with
91+
// no step on the record. `rejections` keeps the settle order, so the
92+
// failure the block reports is the one `Promise.all` rejected with.
93+
const sinks: StepLogEntry[][] = branches.map(() => []);
94+
const rejections: { err: unknown }[] = [];
95+
// Implicit join: continue once when ALL branches have completed.
96+
// #1479: each branch returns its body steps, tagged with the branch index.
97+
//
98+
// #15230: the branch index goes on `branch`, NOT on `iteration`. It
99+
// used to share `iteration` with the loop's row index — one field, two
100+
// meanings, told apart only by reading `regionKind` first — and because
101+
// the tagger let the innermost region win outright, a branch step
102+
// inside a loop body recorded the branch and lost the row entirely.
103+
// `iteration` is now single-valued (the enclosing loop's, carried
104+
// through nesting by `runRegion`'s tagger), so nothing here writes it:
105+
// a branch step inside a loop body gets the row from the loop's own
106+
// tagging pass, and a bare `parallel` leaves it absent because there is
107+
// no enclosing loop to report. Maintainer ruling 2026-09-03.
108+
const branchSteps = await Promise.all(
109+
branches.map(async (branchRegion, i) => {
110+
try {
111+
return await engine.runRegion(
112+
branchRegion,
113+
variables,
114+
context ?? ({} as AutomationContext),
115+
{ parentNodeId: node.id, branch: i, regionKind: 'parallel-branch' },
116+
sinks[i],
117+
);
118+
} catch (err) {
119+
rejections.push({ err });
120+
return sinks[i];
121+
}
122+
}),
123+
);
124+
const childSteps = branchSteps.flat();
125+
126+
const first = rejections[0];
127+
if (first) {
128+
const { err } = first;
97129
// [#18881] A durable pause raised inside a BRANCH is refused at the
98130
// region boundary and must reach the run, ⛔ not be folded into this
99131
// node's returned failure. A returned failure is routable by a `fault`
100132
// edge on the `parallel` node, and routing this one would re-open the
101133
// silence the card closes: the run would continue past a region that
102134
// parked nothing and report success. Re-thrown, so the engine's own
103-
// catch path records it once and fails the run.
104-
if (isRegionSuspensionRefusal(err)) throw err;
135+
// catch path records it once and fails the run — carrying the branch
136+
// steps on the error the way a dying `loop` does, which leaves the
137+
// error itself (identity, message, refusal marking) untouched.
138+
if (isRegionSuspensionRefusal(err)) throw attachPartialSteps(err, childSteps);
105139
const message = err instanceof Error ? err.message : String(err);
106-
return { success: false, error: `parallel '${node.id}': branch failed — ${message}` };
140+
// Returned, not thrown, exactly as before: the block's failure stays
141+
// routable by a `fault` edge on the `parallel`, with the same step
142+
// code, error text and `$error`. The engine folds a returned failure's
143+
// `childSteps` into the run log right behind the block's own step.
144+
return { success: false, error: `parallel '${node.id}': branch failed — ${message}`, childSteps };
107145
}
108146

109-
return { success: true, output: { branches: branches.length }, childSteps: branchSteps.flat() };
147+
return { success: true, output: { branches: branches.length }, childSteps };
110148
},
111149
});
112150

0 commit comments

Comments
 (0)