Skip to content

Commit ea587ad

Browse files
os-zhuangclaude
andcommitted
feat(automation): nested durable pause — subflow chains (linked runs)
A pausing node (approval/screen/wait) inside a subflow previously failed the parent run and stranded the child's already-persisted continuation as an orphan. Subflow pauses now suspend the whole chain as linked runs: - subflow node: child pauses → suspend the PARENT at the node with correlation 'subflow:<childRunId>', surfacing the child's screen; parent linkage ($parentRunId/$parentNodeId/$parentOutputVariable) rides on the child's persisted context — no sys_automation_run schema change - resume() boundary (no traverseNext/executeNode changes): - down-delegation: resuming a run paused at a subflow forwards the signal to the suspended child; multi-screen children keep the parent run id stable and refresh its surfaced screen - up-bubble: a completed run carrying $parentRunId auto-resumes its parent with the child output mapped exactly like the synchronous path - failure propagation: a child failing terminally after the pause fails every waiting ancestor (bounded walk) instead of stranding them - both directions compose recursively (multi-level nesting), survive process restarts via the durable store, and reuse the existing resume idempotency guards Tests: +7 covering parent-suspend linking, direct-child bubble, parent delegation with screens, multi-screen wizard, two-level nesting, cold-boot restart, and post-pause failure propagation (172 total green). ADR-0019 addendum documents the model and v1 boundaries. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent 038ecc7 commit ea587ad

4 files changed

Lines changed: 457 additions & 25 deletions

File tree

docs/adr/0019-approval-as-flow-node.md

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -170,3 +170,28 @@ removal (A4/A5) can be reviewed and sequenced on its own once consumers move ove
170170

171171
The open-source / enterprise split is **not** an architectural concern and is **out of scope for this ADR** — the open registry (ADR-0018) plus the node-config shape make the tier line a *packaging* decision (which approver types / orchestration features ship in which package), not an engine boundary. The split is maintained privately in `cloud/docs/design/approval-tiering.md`. This ADR keeps the engine and the node contract tier-neutral.
172172

173+
174+
## Addendum (2026-06-10) — Nested durable pause: subflow chains (linked-runs model)
175+
176+
A pausing node inside a **subflow** now suspends the whole chain instead of failing the parent.
177+
Model: **linked runs** (the inter-flow half of the long-term execution-state architecture —
178+
cf. Step Functions nested executions / Temporal child workflows; the intra-flow half, a
179+
token/scope tree replacing the single-program-counter continuation, is a separate future ADR).
180+
181+
- The child's continuation persists under its **own run id** (run identity keeps per-flow version
182+
pinning, run logs, and `$runId`-based approval/wait correlation intact). The parent suspends at
183+
the `subflow` node with `correlation: 'subflow:<childRunId>'`; linkage metadata
184+
(`$parentRunId` / `$parentNodeId` / `$parentOutputVariable`) rides on the child's persisted
185+
`context`**no schema change** to `sys_automation_run`.
186+
- `resume()` completes the chain in both directions, recursively: resuming the **child** directly
187+
(approval service, wait timer) **bubbles up** — the parent auto-resumes with the child's output,
188+
mapped exactly like the synchronous path (`${nodeId}.output` + bare `outputVariable`); resuming
189+
the **parent** (a UI holding the original run id, incl. multi-screen wizards) **delegates down**
190+
to the suspended child. A child failing terminally after the pause **fails every waiting
191+
ancestor** (bounded walk), so no run is stranded as resumable-forever.
192+
193+
**v1 boundaries (deliberate):** the subflow node's `fault` out-edges / enclosing `try_catch` do
194+
not catch a *post-pause* child failure (the parent run fails terminally instead); `timeoutMs`
195+
does not count across a suspension; a crash exactly between child completion and the parent
196+
bubble leaves the parent paused — an operator can compensate with a manual
197+
`resume(parentRunId, { output })` (outbox-grade exactly-once chaining is future work).

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

Lines changed: 218 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import { describe, it, expect, beforeEach } from 'vitest';
44
import { AutomationEngine } from '../engine.js';
55
import type { NodeExecutor } from '../engine.js';
6+
import { InMemorySuspendedRunStore } from '../suspended-run-store.js';
67
import { registerSubflowNode } from './subflow-node.js';
78

89
function silentLogger() {
@@ -43,11 +44,31 @@ describe('subflow node executor', () => {
4344
return { success: true };
4445
},
4546
} as NodeExecutor);
46-
// A node that suspends (to exercise the nested-pause guard).
47+
// A node that suspends (to exercise nested durable pause).
4748
engine.registerNodeExecutor({
4849
type: 'pauser',
4950
async execute() { return { success: true, suspend: true }; },
5051
} as NodeExecutor);
52+
// A screen-style pauser: suspends surfacing the screen from node config.
53+
engine.registerNodeExecutor({
54+
type: 'screenpauser',
55+
async execute(node) {
56+
return { success: true, suspend: true, screen: (node.config as any)?.screen };
57+
},
58+
} as NodeExecutor);
59+
// Copies the screen-collected `new_val` (a bare resumed variable) to `result`.
60+
engine.registerNodeExecutor({
61+
type: 'copier',
62+
async execute(_node, variables) {
63+
variables.set('result', variables.get('new_val'));
64+
return { success: true };
65+
},
66+
} as NodeExecutor);
67+
// Fails terminally (post-pause failure propagation).
68+
engine.registerNodeExecutor({
69+
type: 'failer',
70+
async execute() { return { success: false, error: 'boom' }; },
71+
} as NodeExecutor);
5172

5273
engine.registerFlow('child_flow', {
5374
name: 'child_flow',
@@ -118,25 +139,207 @@ describe('subflow node executor', () => {
118139
expect(captured).toEqual([]); // downstream did not run
119140
});
120141

121-
it('fails with a clear error when the child suspends (nested pause unsupported)', async () => {
122-
engine.registerFlow('paused_child', {
123-
name: 'paused_child',
124-
label: 'Paused Child',
142+
// ── Nested durable pause (linked-runs model) ─────────────────────────
143+
144+
/** Child that pauses, then sets its output var when resumed. */
145+
const pausedChild = (pauseNodes: Array<{ id: string; type: string; config?: Record<string, unknown> }>) => ({
146+
name: 'paused_child',
147+
label: 'Paused Child',
148+
type: 'autolaunched',
149+
variables: [{ name: 'result', type: 'text', isOutput: true }],
150+
nodes: [
151+
{ id: 's', type: 'start', label: 'Start' },
152+
...pauseNodes.map((n) => ({ label: n.id, ...n })),
153+
{ id: 'cm', type: 'childmark', label: 'Child Work' },
154+
{ id: 'e', type: 'end', label: 'End' },
155+
],
156+
edges: [
157+
{ id: 'es', source: 's', target: pauseNodes[0].id },
158+
...pauseNodes.map((n, i) => ({
159+
id: `ep${i}`,
160+
source: n.id,
161+
target: pauseNodes[i + 1]?.id ?? 'cm',
162+
})),
163+
{ id: 'ee', source: 'cm', target: 'e' },
164+
],
165+
});
166+
167+
const registerPausingPair = (pauseNodes: Array<{ id: string; type: string; config?: Record<string, unknown> }>) => {
168+
engine.registerFlow('paused_child', pausedChild(pauseNodes) as never);
169+
engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }));
170+
};
171+
172+
const suspendedByFlow = (name: string) =>
173+
engine.listSuspendedRuns().find((r) => r.flowName === name);
174+
175+
it('suspends the parent (not fails) when the child pauses, linking the runs', async () => {
176+
registerPausingPair([{ id: 'p', type: 'pauser' }]);
177+
const result = await engine.execute('parent_flow');
178+
179+
expect(result.success).toBe(true);
180+
expect(result.status).toBe('paused');
181+
const parent = suspendedByFlow('parent_flow');
182+
const child = suspendedByFlow('paused_child');
183+
expect(parent).toBeDefined();
184+
expect(child).toBeDefined();
185+
expect(result.runId).toBe(parent!.runId);
186+
expect(parent!.nodeId).toBe('call');
187+
expect(parent!.correlation).toBe(`subflow:${child!.runId}`);
188+
});
189+
190+
it('bubbles a directly-resumed child completion up to the parent (approval/wait path)', async () => {
191+
registerPausingPair([{ id: 'p', type: 'pauser' }]);
192+
await engine.execute('parent_flow');
193+
const child = suspendedByFlow('paused_child')!;
194+
195+
const childRes = await engine.resume(child.runId);
196+
197+
expect(childRes.success).toBe(true);
198+
expect(childRes.status).toBeUndefined(); // child ran to completion
199+
// Parent auto-continued: downstream captured the mapped output, both rows gone.
200+
expect(captured).toEqual([{ result: 'CHILD_DONE' }]);
201+
expect(engine.listSuspendedRuns()).toHaveLength(0);
202+
});
203+
204+
it('delegates a parent resume down to the child (screen-flow path), surfacing the child screen', async () => {
205+
const screen = { nodeId: 'p', title: 'Collect', fields: [{ name: 'new_val', type: 'text' }] };
206+
registerPausingPair([{ id: 'p', type: 'screenpauser', config: { screen } }]);
207+
// Replace cm: copy the collected input instead of the static marker.
208+
const flow = pausedChild([{ id: 'p', type: 'screenpauser', config: { screen } }]);
209+
flow.nodes = flow.nodes.map((n) => (n.id === 'cm' ? { ...n, type: 'copier' } : n));
210+
engine.registerFlow('paused_child', flow as never);
211+
212+
const result = await engine.execute('parent_flow');
213+
expect(result.status).toBe('paused');
214+
expect(result.screen).toEqual(screen); // nested screen surfaces on the parent result
215+
216+
const parentRunId = result.runId!;
217+
const final = await engine.resume(parentRunId, { variables: { new_val: 'typed-in' } });
218+
219+
expect(final.success).toBe(true);
220+
expect(final.status).toBeUndefined();
221+
expect(captured).toEqual([{ result: 'typed-in' }]);
222+
expect(engine.listSuspendedRuns()).toHaveLength(0);
223+
});
224+
225+
it('keeps the parent paused across a multi-screen child wizard', async () => {
226+
const s1 = { nodeId: 'p1', title: 'Step 1', fields: [{ name: 'new_val', type: 'text' }] };
227+
const s2 = { nodeId: 'p2', title: 'Step 2', fields: [{ name: 'other', type: 'text' }] };
228+
const flow = pausedChild([
229+
{ id: 'p1', type: 'screenpauser', config: { screen: s1 } },
230+
{ id: 'p2', type: 'screenpauser', config: { screen: s2 } },
231+
]);
232+
flow.nodes = flow.nodes.map((n) => (n.id === 'cm' ? { ...n, type: 'copier' } : n));
233+
engine.registerFlow('paused_child', flow as never);
234+
engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }));
235+
236+
const r1 = await engine.execute('parent_flow');
237+
expect(r1.status).toBe('paused');
238+
expect(r1.screen).toEqual(s1);
239+
const parentRunId = r1.runId!;
240+
241+
const r2 = await engine.resume(parentRunId, { variables: { new_val: 'v1' } });
242+
expect(r2.status).toBe('paused');
243+
expect(r2.runId).toBe(parentRunId); // UI keeps one stable run id
244+
expect(r2.screen).toEqual(s2); // next wizard screen
245+
expect(engine.getSuspendedScreen(parentRunId)).toEqual(s2); // refresh-safe re-fetch
246+
247+
const r3 = await engine.resume(parentRunId, { variables: { other: 'x' } });
248+
expect(r3.success).toBe(true);
249+
expect(r3.status).toBeUndefined();
250+
expect(captured).toEqual([{ result: 'v1' }]);
251+
expect(engine.listSuspendedRuns()).toHaveLength(0);
252+
});
253+
254+
it('bubbles through two levels of nesting', async () => {
255+
registerPausingPair([{ id: 'p', type: 'pauser' }]);
256+
// grandparent → parent_flow → paused_child
257+
engine.registerNodeExecutor({
258+
type: 'grandcheck',
259+
async execute(_node, variables) {
260+
captured.push(variables.get('grandResult'));
261+
return { success: true };
262+
},
263+
} as NodeExecutor);
264+
engine.registerFlow('grand_flow', {
265+
name: 'grand_flow',
266+
label: 'Grand Flow',
125267
type: 'autolaunched',
126268
nodes: [
127-
{ id: 's', type: 'start', label: 'Start' },
128-
{ id: 'p', type: 'pauser', label: 'Pause' },
129-
{ id: 'e', type: 'end', label: 'End' },
269+
{ id: 'gs', type: 'start', label: 'Start' },
270+
{ id: 'gcall', type: 'subflow', label: 'Call Parent', config: { flowName: 'parent_flow', outputVariable: 'grandResult' } },
271+
{ id: 'gchk', type: 'grandcheck', label: 'Check' },
272+
{ id: 'ge', type: 'end', label: 'End' },
130273
],
131274
edges: [
132-
{ id: 'a', source: 's', target: 'p' },
133-
{ id: 'b', source: 'p', target: 'e' },
275+
{ id: 'g1', source: 'gs', target: 'gcall' },
276+
{ id: 'g2', source: 'gcall', target: 'gchk' },
277+
{ id: 'g3', source: 'gchk', target: 'ge' },
134278
],
135-
});
136-
engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child' }));
137-
const result = await engine.execute('parent_flow');
138-
expect(result.success).toBe(false);
139-
expect(result.error).toMatch(/suspended/i);
279+
} as never);
280+
281+
const result = await engine.execute('grand_flow');
282+
expect(result.status).toBe('paused');
283+
expect(engine.listSuspendedRuns()).toHaveLength(3); // grand + parent + child
284+
285+
const child = suspendedByFlow('paused_child')!;
286+
const childRes = await engine.resume(child.runId);
287+
expect(childRes.success).toBe(true);
288+
// parentcheck captured the child output; grandcheck captured the parent output (its output vars — none declared → {}).
289+
expect(captured[0]).toEqual({ result: 'CHILD_DONE' });
290+
expect(captured).toHaveLength(2);
291+
expect(engine.listSuspendedRuns()).toHaveLength(0);
292+
});
293+
294+
it('survives a process restart: chain persisted, resume on a fresh engine bubbles to the parent', async () => {
295+
const store = new InMemorySuspendedRunStore();
296+
engine.setSuspendedRunStore(store);
297+
registerPausingPair([{ id: 'p', type: 'pauser' }]);
298+
await engine.execute('parent_flow');
299+
const child = suspendedByFlow('paused_child')!;
300+
expect((await store.list()).length).toBe(2);
301+
302+
// "Restart": a fresh engine sharing only the durable store + flow registry.
303+
const engineB = new AutomationEngine(silentLogger(), store);
304+
registerSubflowNode(engineB, ctx());
305+
const capturedB: unknown[] = [];
306+
engineB.registerNodeExecutor({
307+
type: 'childmark',
308+
async execute(_node, variables) { variables.set('result', 'CHILD_DONE'); return { success: true }; },
309+
} as NodeExecutor);
310+
engineB.registerNodeExecutor({
311+
type: 'parentcheck',
312+
async execute(_node, variables) { capturedB.push(variables.get('subResult')); return { success: true }; },
313+
} as NodeExecutor);
314+
engineB.registerNodeExecutor({ type: 'pauser', async execute() { return { success: true, suspend: true }; } } as NodeExecutor);
315+
engineB.registerFlow('paused_child', pausedChild([{ id: 'p', type: 'pauser' }]) as never);
316+
engineB.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }) as never);
317+
318+
const res = await engineB.resume(child.runId);
319+
expect(res.success).toBe(true);
320+
expect(capturedB).toEqual([{ result: 'CHILD_DONE' }]);
321+
expect(await store.list()).toHaveLength(0); // both rows consumed
322+
});
323+
324+
it('fails the waiting parent when the resumed child fails terminally', async () => {
325+
const flow = pausedChild([{ id: 'p', type: 'pauser' }]);
326+
flow.nodes = flow.nodes.map((n) => (n.id === 'cm' ? { ...n, type: 'failer' } : n));
327+
engine.registerFlow('paused_child', flow as never);
328+
engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }));
329+
330+
const r = await engine.execute('parent_flow');
331+
const parentRunId = r.runId!;
332+
const child = suspendedByFlow('paused_child')!;
333+
334+
const childRes = await engine.resume(child.runId);
335+
expect(childRes.success).toBe(false);
336+
337+
// The parent is terminally failed, not left suspended forever.
338+
expect(engine.listSuspendedRuns()).toHaveLength(0);
339+
const again = await engine.resume(parentRunId);
340+
expect(again.success).toBe(false);
341+
expect(again.error).toMatch(/No suspended run/);
342+
expect(captured).toEqual([]); // parent downstream never ran
140343
});
141344

142345
it('guards against a recursive subflow cycle (clean error, no stack overflow)', async () => {

packages/services/service-automation/src/builtin/subflow-node.ts

Lines changed: 46 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,23 @@ const MAX_SUBFLOW_DEPTH = 16;
1717
* the parent — under `${nodeId}.output`, and under `config.outputVariable` as a
1818
* bare variable when given.
1919
*
20-
* Scope (v1): **synchronous** subflows that run to completion. If the child
21-
* *suspends* (a nested `approval` / `screen` / `wait`), the node fails with a
22-
* clear message rather than silently dropping the run — nested durable pause is
23-
* a deliberate follow-up. A depth guard ({@link MAX_SUBFLOW_DEPTH}) turns an
24-
* accidental recursive cycle into a clean error instead of a stack overflow.
20+
* **Nested durable pause (linked-runs model).** If the child *suspends* (a
21+
* nested `approval` / `screen` / `wait`), the child's continuation is already
22+
* persisted by the engine as its own run; this node then suspends the PARENT
23+
* run at this node with `correlation: 'subflow:<childRunId>'`, so both rows
24+
* survive a restart and stay linked. The engine's resume boundary completes
25+
* the chain in both directions:
26+
*
27+
* - resuming the CHILD directly (approval service / wait timer hold the child
28+
* `$runId`) bubbles UP on completion — the engine auto-resumes the parent
29+
* with the child's output, mapped exactly like the synchronous path;
30+
* - resuming the PARENT (a UI holds the parent run id from the original
31+
* `execute()` response) delegates DOWN to the suspended child.
32+
*
33+
* The linkage rides on the child's context (`$parentRunId` / `$parentNodeId` /
34+
* `$parentOutputVariable`), which the engine persists with the child run — no
35+
* schema change. A depth guard ({@link MAX_SUBFLOW_DEPTH}) turns an accidental
36+
* recursive cycle into a clean error instead of a stack overflow.
2537
*/
2638
export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext): void {
2739
engine.registerNodeExecutor({
@@ -34,6 +46,9 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext
3446
icon: 'workflow',
3547
category: 'logic',
3648
source: 'builtin',
49+
// A child that suspends (approval/screen/wait) suspends this node too —
50+
// the parent run pauses here and resumes when the child completes.
51+
supportsPause: true,
3752
}),
3853
async execute(node, variables, context) {
3954
const cfg = (node.config ?? {}) as Record<string, unknown>;
@@ -56,20 +71,42 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext
5671
const rawInput = (cfg.input && typeof cfg.input === 'object' ? cfg.input : {}) as Record<string, unknown>;
5772
const params = interpolate(rawInput, variables, context ?? ({} as AutomationContext)) as Record<string, unknown>;
5873

74+
const outVar = typeof cfg.outputVariable === 'string' && cfg.outputVariable ? cfg.outputVariable : undefined;
75+
76+
// Parent linkage for nested durable pause: should the child suspend, the
77+
// engine persists these with the child run and uses them to bubble the
78+
// child's eventual completion back into THIS run (resume at this node).
79+
// `$runId` is injected by the engine at run start (ADR-0019).
80+
const parentRunId = variables.get('$runId');
5981
const childContext = {
6082
...(context ?? {}),
6183
$subflowDepth: depth + 1,
6284
params,
85+
...(parentRunId != null
86+
? {
87+
$parentRunId: String(parentRunId),
88+
$parentNodeId: node.id,
89+
...(outVar ? { $parentOutputVariable: outVar } : {}),
90+
}
91+
: {}),
6392
} as AutomationContext;
6493

6594
const child = await engine.execute(flowName, childContext);
6695

6796
if (child.status === 'paused') {
97+
// Nested durable pause: the child's continuation is persisted under its
98+
// own run id; suspend the parent here, linked via the correlation key.
99+
// A nested screen surfaces on the parent's paused result so a UI runner
100+
// can render it against the parent run id (the engine delegates the
101+
// parent's resume down to the child).
102+
if (!child.runId) {
103+
return { success: false, error: `subflow '${flowName}' paused without a run id — cannot link the runs` };
104+
}
68105
return {
69-
success: false,
70-
error:
71-
`subflow '${flowName}' suspended at a pausing node — a nested approval/screen/wait ` +
72-
`pause from a subflow is not yet supported`,
106+
success: true,
107+
suspend: true,
108+
correlation: `subflow:${child.runId}`,
109+
screen: child.screen,
73110
};
74111
}
75112
if (!child.success) {
@@ -78,7 +115,6 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext
78115

79116
// Bare output variable (like the assignment node, the executor may write
80117
// directly to the parent variable map).
81-
const outVar = typeof cfg.outputVariable === 'string' && cfg.outputVariable ? cfg.outputVariable : undefined;
82118
if (outVar) variables.set(outVar, child.output ?? null);
83119

84120
return { success: true, output: { output: child.output ?? null } };

0 commit comments

Comments
 (0)