feat(automation): nested durable pause — subflow chains (linked runs)

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

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

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

@@ -3,6 +3,7 @@
 import { describe, it, expect, beforeEach } from 'vitest';
 import { AutomationEngine } from '../engine.js';
 import type { NodeExecutor } from '../engine.js';
+import { InMemorySuspendedRunStore } from '../suspended-run-store.js';
 import { registerSubflowNode } from './subflow-node.js';
 
 function silentLogger() {
@@ -43,11 +44,31 @@ describe('subflow node executor', () => {
  return { success: true };
  },
  } as NodeExecutor);
- // A node that suspends (to exercise the nested-pause guard).
+ // A node that suspends (to exercise nested durable pause).
  engine.registerNodeExecutor({
  type: 'pauser',
  async execute() { return { success: true, suspend: true }; },
  } as NodeExecutor);
+ // A screen-style pauser: suspends surfacing the screen from node config.
+ engine.registerNodeExecutor({
+ type: 'screenpauser',
+ async execute(node) {
+ return { success: true, suspend: true, screen: (node.config as any)?.screen };
+ },
+ } as NodeExecutor);
+ // Copies the screen-collected `new_val` (a bare resumed variable) to `result`.
+ engine.registerNodeExecutor({
+ type: 'copier',
+ async execute(_node, variables) {
+ variables.set('result', variables.get('new_val'));
+ return { success: true };
+ },
+ } as NodeExecutor);
+ // Fails terminally (post-pause failure propagation).
+ engine.registerNodeExecutor({
+ type: 'failer',
+ async execute() { return { success: false, error: 'boom' }; },
+ } as NodeExecutor);
 
  engine.registerFlow('child_flow', {
  name: 'child_flow',
@@ -118,25 +139,207 @@ describe('subflow node executor', () => {
  expect(captured).toEqual([]); // downstream did not run
  });
 
- it('fails with a clear error when the child suspends (nested pause unsupported)', async () => {
- engine.registerFlow('paused_child', {
- name: 'paused_child',
- label: 'Paused Child',
+ // ── Nested durable pause (linked-runs model) ─────────────────────────
+
+ /** Child that pauses, then sets its output var when resumed. */
+ const pausedChild = (pauseNodes: Array<{ id: string; type: string; config?: Record<string, unknown> }>) => ({
+ name: 'paused_child',
+ label: 'Paused Child',
+ type: 'autolaunched',
+ variables: [{ name: 'result', type: 'text', isOutput: true }],
+ nodes: [
+ { id: 's', type: 'start', label: 'Start' },
+ ...pauseNodes.map((n) => ({ label: n.id, ...n })),
+ { id: 'cm', type: 'childmark', label: 'Child Work' },
+ { id: 'e', type: 'end', label: 'End' },
+ ],
+ edges: [
+ { id: 'es', source: 's', target: pauseNodes[0].id },
+ ...pauseNodes.map((n, i) => ({
+ id: `ep${i}`,
+ source: n.id,
+ target: pauseNodes[i + 1]?.id ?? 'cm',
+ })),
+ { id: 'ee', source: 'cm', target: 'e' },
+ ],
+ });
+
+ const registerPausingPair = (pauseNodes: Array<{ id: string; type: string; config?: Record<string, unknown> }>) => {
+ engine.registerFlow('paused_child', pausedChild(pauseNodes) as never);
+ engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }));
+ };
+
+ const suspendedByFlow = (name: string) =>
+ engine.listSuspendedRuns().find((r) => r.flowName === name);
+
+ it('suspends the parent (not fails) when the child pauses, linking the runs', async () => {
+ registerPausingPair([{ id: 'p', type: 'pauser' }]);
+ const result = await engine.execute('parent_flow');
+
+ expect(result.success).toBe(true);
+ expect(result.status).toBe('paused');
+ const parent = suspendedByFlow('parent_flow');
+ const child = suspendedByFlow('paused_child');
+ expect(parent).toBeDefined();
+ expect(child).toBeDefined();
+ expect(result.runId).toBe(parent!.runId);
+ expect(parent!.nodeId).toBe('call');
+ expect(parent!.correlation).toBe(`subflow:${child!.runId}`);
+ });
+
+ it('bubbles a directly-resumed child completion up to the parent (approval/wait path)', async () => {
+ registerPausingPair([{ id: 'p', type: 'pauser' }]);
+ await engine.execute('parent_flow');
+ const child = suspendedByFlow('paused_child')!;
+
+ const childRes = await engine.resume(child.runId);
+
+ expect(childRes.success).toBe(true);
+ expect(childRes.status).toBeUndefined(); // child ran to completion
+ // Parent auto-continued: downstream captured the mapped output, both rows gone.
+ expect(captured).toEqual([{ result: 'CHILD_DONE' }]);
+ expect(engine.listSuspendedRuns()).toHaveLength(0);
+ });
+
+ it('delegates a parent resume down to the child (screen-flow path), surfacing the child screen', async () => {
+ const screen = { nodeId: 'p', title: 'Collect', fields: [{ name: 'new_val', type: 'text' }] };
+ registerPausingPair([{ id: 'p', type: 'screenpauser', config: { screen } }]);
+ // Replace cm: copy the collected input instead of the static marker.
+ const flow = pausedChild([{ id: 'p', type: 'screenpauser', config: { screen } }]);
+ flow.nodes = flow.nodes.map((n) => (n.id === 'cm' ? { ...n, type: 'copier' } : n));
+ engine.registerFlow('paused_child', flow as never);
+
+ const result = await engine.execute('parent_flow');
+ expect(result.status).toBe('paused');
+ expect(result.screen).toEqual(screen); // nested screen surfaces on the parent result
+
+ const parentRunId = result.runId!;
+ const final = await engine.resume(parentRunId, { variables: { new_val: 'typed-in' } });
+
+ expect(final.success).toBe(true);
+ expect(final.status).toBeUndefined();
+ expect(captured).toEqual([{ result: 'typed-in' }]);
+ expect(engine.listSuspendedRuns()).toHaveLength(0);
+ });
+
+ it('keeps the parent paused across a multi-screen child wizard', async () => {
+ const s1 = { nodeId: 'p1', title: 'Step 1', fields: [{ name: 'new_val', type: 'text' }] };
+ const s2 = { nodeId: 'p2', title: 'Step 2', fields: [{ name: 'other', type: 'text' }] };
+ const flow = pausedChild([
+ { id: 'p1', type: 'screenpauser', config: { screen: s1 } },
+ { id: 'p2', type: 'screenpauser', config: { screen: s2 } },
+ ]);
+ flow.nodes = flow.nodes.map((n) => (n.id === 'cm' ? { ...n, type: 'copier' } : n));
+ engine.registerFlow('paused_child', flow as never);
+ engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }));
+
+ const r1 = await engine.execute('parent_flow');
+ expect(r1.status).toBe('paused');
+ expect(r1.screen).toEqual(s1);
+ const parentRunId = r1.runId!;
+
+ const r2 = await engine.resume(parentRunId, { variables: { new_val: 'v1' } });
+ expect(r2.status).toBe('paused');
+ expect(r2.runId).toBe(parentRunId); // UI keeps one stable run id
+ expect(r2.screen).toEqual(s2); // next wizard screen
+ expect(engine.getSuspendedScreen(parentRunId)).toEqual(s2); // refresh-safe re-fetch
+
+ const r3 = await engine.resume(parentRunId, { variables: { other: 'x' } });
+ expect(r3.success).toBe(true);
+ expect(r3.status).toBeUndefined();
+ expect(captured).toEqual([{ result: 'v1' }]);
+ expect(engine.listSuspendedRuns()).toHaveLength(0);
+ });
+
+ it('bubbles through two levels of nesting', async () => {
+ registerPausingPair([{ id: 'p', type: 'pauser' }]);
+ // grandparent → parent_flow → paused_child
+ engine.registerNodeExecutor({
+ type: 'grandcheck',
+ async execute(_node, variables) {
+ captured.push(variables.get('grandResult'));
+ return { success: true };
+ },
+ } as NodeExecutor);
+ engine.registerFlow('grand_flow', {
+ name: 'grand_flow',
+ label: 'Grand Flow',
  type: 'autolaunched',
  nodes: [
- { id: 's', type: 'start', label: 'Start' },
- { id: 'p', type: 'pauser', label: 'Pause' },
- { id: 'e', type: 'end', label: 'End' },
+ { id: 'gs', type: 'start', label: 'Start' },
+ { id: 'gcall', type: 'subflow', label: 'Call Parent', config: { flowName: 'parent_flow', outputVariable: 'grandResult' } },
+ { id: 'gchk', type: 'grandcheck', label: 'Check' },
+ { id: 'ge', type: 'end', label: 'End' },
  ],
  edges: [
- { id: 'a', source: 's', target: 'p' },
- { id: 'b', source: 'p', target: 'e' },
+ { id: 'g1', source: 'gs', target: 'gcall' },
+ { id: 'g2', source: 'gcall', target: 'gchk' },
+ { id: 'g3', source: 'gchk', target: 'ge' },
  ],
- });
- engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child' }));
- const result = await engine.execute('parent_flow');
- expect(result.success).toBe(false);
- expect(result.error).toMatch(/suspended/i);
+ } as never);
+
+ const result = await engine.execute('grand_flow');
+ expect(result.status).toBe('paused');
+ expect(engine.listSuspendedRuns()).toHaveLength(3); // grand + parent + child
+
+ const child = suspendedByFlow('paused_child')!;
+ const childRes = await engine.resume(child.runId);
+ expect(childRes.success).toBe(true);
+ // parentcheck captured the child output; grandcheck captured the parent output (its output vars — none declared → {}).
+ expect(captured[0]).toEqual({ result: 'CHILD_DONE' });
+ expect(captured).toHaveLength(2);
+ expect(engine.listSuspendedRuns()).toHaveLength(0);
+ });
+
+ it('survives a process restart: chain persisted, resume on a fresh engine bubbles to the parent', async () => {
+ const store = new InMemorySuspendedRunStore();
+ engine.setSuspendedRunStore(store);
+ registerPausingPair([{ id: 'p', type: 'pauser' }]);
+ await engine.execute('parent_flow');
+ const child = suspendedByFlow('paused_child')!;
+ expect((await store.list()).length).toBe(2);
+
+ // "Restart": a fresh engine sharing only the durable store + flow registry.
+ const engineB = new AutomationEngine(silentLogger(), store);
+ registerSubflowNode(engineB, ctx());
+ const capturedB: unknown[] = [];
+ engineB.registerNodeExecutor({
+ type: 'childmark',
+ async execute(_node, variables) { variables.set('result', 'CHILD_DONE'); return { success: true }; },
+ } as NodeExecutor);
+ engineB.registerNodeExecutor({
+ type: 'parentcheck',
+ async execute(_node, variables) { capturedB.push(variables.get('subResult')); return { success: true }; },
+ } as NodeExecutor);
+ engineB.registerNodeExecutor({ type: 'pauser', async execute() { return { success: true, suspend: true }; } } as NodeExecutor);
+ engineB.registerFlow('paused_child', pausedChild([{ id: 'p', type: 'pauser' }]) as never);
+ engineB.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }) as never);
+
+ const res = await engineB.resume(child.runId);
+ expect(res.success).toBe(true);
+ expect(capturedB).toEqual([{ result: 'CHILD_DONE' }]);
+ expect(await store.list()).toHaveLength(0); // both rows consumed
+ });
+
+ it('fails the waiting parent when the resumed child fails terminally', async () => {
+ const flow = pausedChild([{ id: 'p', type: 'pauser' }]);
+ flow.nodes = flow.nodes.map((n) => (n.id === 'cm' ? { ...n, type: 'failer' } : n));
+ engine.registerFlow('paused_child', flow as never);
+ engine.registerFlow('parent_flow', parentFlow({ flowName: 'paused_child', outputVariable: 'subResult' }));
+
+ const r = await engine.execute('parent_flow');
+ const parentRunId = r.runId!;
+ const child = suspendedByFlow('paused_child')!;
+
+ const childRes = await engine.resume(child.runId);
+ expect(childRes.success).toBe(false);
+
+ // The parent is terminally failed, not left suspended forever.
+ expect(engine.listSuspendedRuns()).toHaveLength(0);
+ const again = await engine.resume(parentRunId);
+ expect(again.success).toBe(false);
+ expect(again.error).toMatch(/No suspended run/);
+ expect(captured).toEqual([]); // parent downstream never ran
  });
 
  it('guards against a recursive subflow cycle (clean error, no stack overflow)', async () => {

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

@@ -17,11 +17,23 @@ const MAX_SUBFLOW_DEPTH = 16;
  * the parent — under `${nodeId}.output`, and under `config.outputVariable` as a
  * bare variable when given.
  *
- * Scope (v1): **synchronous** subflows that run to completion. If the child
- * *suspends* (a nested `approval` / `screen` / `wait`), the node fails with a
- * clear message rather than silently dropping the run — nested durable pause is
- * a deliberate follow-up. A depth guard ({@link MAX_SUBFLOW_DEPTH}) turns an
- * accidental recursive cycle into a clean error instead of a stack overflow.
+ * **Nested durable pause (linked-runs model).** If the child *suspends* (a
+ * nested `approval` / `screen` / `wait`), the child's continuation is already
+ * persisted by the engine as its own run; this node then suspends the PARENT
+ * run at this node with `correlation: 'subflow:<childRunId>'`, so both rows
+ * survive a restart and stay linked. The engine's resume boundary completes
+ * the chain in both directions:
+ *
+ * - resuming the CHILD directly (approval service / wait timer hold the child
+ * `$runId`) bubbles UP on completion — the engine auto-resumes the parent
+ * with the child's output, mapped exactly like the synchronous path;
+ * - resuming the PARENT (a UI holds the parent run id from the original
+ * `execute()` response) delegates DOWN to the suspended child.
+ *
+ * The linkage rides on the child's context (`$parentRunId` / `$parentNodeId` /
+ * `$parentOutputVariable`), which the engine persists with the child run — no
+ * schema change. A depth guard ({@link MAX_SUBFLOW_DEPTH}) turns an accidental
+ * recursive cycle into a clean error instead of a stack overflow.
  */
 export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext): void {
  engine.registerNodeExecutor({
@@ -34,6 +46,9 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext
  icon: 'workflow',
  category: 'logic',
  source: 'builtin',
+ // A child that suspends (approval/screen/wait) suspends this node too —
+ // the parent run pauses here and resumes when the child completes.
+ supportsPause: true,
  }),
  async execute(node, variables, context) {
  const cfg = (node.config ?? {}) as Record<string, unknown>;
@@ -56,20 +71,42 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext
  const rawInput = (cfg.input && typeof cfg.input === 'object' ? cfg.input : {}) as Record<string, unknown>;
  const params = interpolate(rawInput, variables, context ?? ({} as AutomationContext)) as Record<string, unknown>;
 
+ const outVar = typeof cfg.outputVariable === 'string' && cfg.outputVariable ? cfg.outputVariable : undefined;
+
+ // Parent linkage for nested durable pause: should the child suspend, the
+ // engine persists these with the child run and uses them to bubble the
+ // child's eventual completion back into THIS run (resume at this node).
+ // `$runId` is injected by the engine at run start (ADR-0019).
+ const parentRunId = variables.get('$runId');
  const childContext = {
  ...(context ?? {}),
  $subflowDepth: depth + 1,
  params,
+ ...(parentRunId != null
+ ? {
+ $parentRunId: String(parentRunId),
+ $parentNodeId: node.id,
+ ...(outVar ? { $parentOutputVariable: outVar } : {}),
+ }
+ : {}),
  } as AutomationContext;
 
  const child = await engine.execute(flowName, childContext);
 
  if (child.status === 'paused') {
+ // Nested durable pause: the child's continuation is persisted under its
+ // own run id; suspend the parent here, linked via the correlation key.
+ // A nested screen surfaces on the parent's paused result so a UI runner
+ // can render it against the parent run id (the engine delegates the
+ // parent's resume down to the child).
+ if (!child.runId) {
+ return { success: false, error: `subflow '${flowName}' paused without a run id — cannot link the runs` };
+ }
  return {
- success: false,
-error:
-`subflow '${flowName}' suspended at a pausing node — a nested approval/screen/wait ` +
- `pause from a subflow is not yet supported`,
+ success: true,
+suspend: true,
+correlation: `subflow:${child.runId}`,
+screen: child.screen,
  };
  }
  if (!child.success) {
@@ -78,7 +115,6 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext
 
  // Bare output variable (like the assignment node, the executor may write
  // directly to the parent variable map).
- const outVar = typeof cfg.outputVariable === 'string' && cfg.outputVariable ? cfg.outputVariable : undefined;
  if (outVar) variables.set(outVar, child.output ?? null);
 
  return { success: true, output: { output: child.output ?? null } };