Merge pull request #1693 from objectstack-ai/feat-subflow-nested-pause

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

Author: os-zhuang

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 } };