From ea587adde6b1f0a9c6b9838645ca46f3633ecd55 Mon Sep 17 00:00:00 2001 From: os-zhuang Date: Wed, 10 Jun 2026 20:37:22 +0500 Subject: [PATCH] =?UTF-8?q?feat(automation):=20nested=20durable=20pause=20?= =?UTF-8?q?=E2=80=94=20subflow=20chains=20(linked=20runs)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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:', 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 --- docs/adr/0019-approval-as-flow-node.md | 25 ++ .../src/builtin/subflow-node.test.ts | 233 ++++++++++++++++-- .../src/builtin/subflow-node.ts | 56 ++++- .../services/service-automation/src/engine.ts | 168 +++++++++++++ 4 files changed, 457 insertions(+), 25 deletions(-) diff --git a/docs/adr/0019-approval-as-flow-node.md b/docs/adr/0019-approval-as-flow-node.md index 00628e10ed..d94ee07bd5 100644 --- a/docs/adr/0019-approval-as-flow-node.md +++ b/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:'`; 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). diff --git a/packages/services/service-automation/src/builtin/subflow-node.test.ts b/packages/services/service-automation/src/builtin/subflow-node.test.ts index 92995de8fc..d6ba5ef66b 100644 --- a/packages/services/service-automation/src/builtin/subflow-node.test.ts +++ b/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 }>) => ({ + 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 }>) => { + 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 () => { diff --git a/packages/services/service-automation/src/builtin/subflow-node.ts b/packages/services/service-automation/src/builtin/subflow-node.ts index 0fb3fae7c8..49efc99bf0 100644 --- a/packages/services/service-automation/src/builtin/subflow-node.ts +++ b/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:'`, 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; @@ -56,20 +71,42 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext const rawInput = (cfg.input && typeof cfg.input === 'object' ? cfg.input : {}) as Record; const params = interpolate(rawInput, variables, context ?? ({} as AutomationContext)) as Record; + 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 } }; diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 75c57e7df4..778a309497 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1015,8 +1015,25 @@ export class AutomationEngine implements IAutomationService { * restricted to the edge labelled `signal.branchLabel` (e.g. the approval * decision). The continuation may itself suspend again, in which case this * returns `{ status: 'paused', runId }` afresh. + * + * **Subflow chains (nested pause, linked-runs model).** A run paused at a + * `subflow` node (correlation `subflow:`) DELEGATES the signal + * down to the suspended child; a run that completes and carries + * `$parentRunId` in its context BUBBLES its output up by auto-resuming the + * parent. Both directions compose recursively, so arbitrarily nested + * subflow pauses resolve from either end (UI holds the parent run id; + * approval/wait infrastructure holds the child's). */ async resume(runId: string, signal?: ResumeSignal): Promise { + return this.resumeInternal(runId, signal, false); + } + + /** + * @param skipBubble - Set when the caller is the subflow DELEGATION path, + * which continues the parent itself after the child completes — the + * child's own up-bubble must stay off so the parent isn't resumed twice. + */ + private async resumeInternal(runId: string, signal: ResumeSignal | undefined, skipBubble: boolean): Promise { // Idempotency guard (set synchronously, before any await): reject a // concurrent duplicate resume of the same run so side effects can't run // twice. A duplicate that arrives *after* this one finishes finds no @@ -1049,6 +1066,55 @@ export class AutomationEngine implements IAutomationService { if (!node) { return { success: false, error: `Suspended node '${run.nodeId}' no longer exists in flow '${run.flowName}'` }; } + + // ── Subflow delegation (nested pause): this run is paused at a + // `subflow` node whose child run itself suspended. The caller's + // signal is meant for the node the CHILD paused on (its screen / + // approval / wait), so forward it down. The child resumes with + // bubbling off — when it completes, *this* invocation continues the + // parent from the subflow node with the child's output, using the + // same mapping as the synchronous path. + if (typeof run.correlation === 'string' && run.correlation.startsWith('subflow:')) { + const childRunId = run.correlation.slice('subflow:'.length); + // Capture the child's row BEFORE resuming consumes it — the + // output-variable mapping rides on the child's context. + const childRun = + this.suspendedRuns.get(childRunId) ?? + (this.store ? await this.store.load(childRunId).catch(() => null) : null); + if (childRun) { + const childRes = await this.resumeInternal(childRunId, signal, true); + if (childRes.status === 'paused') { + // Child paused again (e.g. the next screen of a wizard). + // This run stays suspended; refresh its surfaced screen + // so a re-fetch (getSuspendedScreen) shows the new one. + if (childRes.screen && childRes.screen !== run.screen) { + await this.persistSuspendedRun({ ...run, screen: childRes.screen }); + } + return { + success: true, + status: 'paused', + runId, + durationMs: Date.now() - run.startTime, + screen: childRes.screen, + }; + } + if (!childRes.success) { + const error = `subflow run '${childRunId}' (${childRun.flowName}) failed: ${childRes.error ?? 'unknown error'}`; + await this.failSuspendedRun(run, error); + return { success: false, error, durationMs: Date.now() - run.startTime }; + } + // Child completed — continue below with its output as the + // resume signal (replaces the caller's signal, which the + // child already consumed). + signal = this.buildSubflowResumeSignal(childRun.context, childRes.output); + } else { + this.logger.warn( + `[automation] run '${runId}' is paused at subflow node '${run.nodeId}' but child run '${childRunId}' ` + + `is gone — continuing without child output`, + ); + } + } + // Consume the suspension *before* running downstream work — a run // resumes exactly once per pause, and a duplicate resume after a // partial restart must not double-run side effects. @@ -1101,6 +1167,17 @@ export class AutomationEngine implements IAutomationService { steps, output, }); + + // ── Subflow up-bubble (nested pause): this run was a subflow + // child whose parent suspended awaiting it. Auto-resume the + // parent with our output, mapped like the synchronous path. + // Skipped when the DELEGATION path drives the chain (it + // continues the parent itself). Best-effort: the child's own + // completion stands even if the parent continuation fails. + if (!skipBubble) { + await this.bubbleToParent(run, output); + } + return { success: true, output, durationMs }; } catch (err: unknown) { // Re-suspended at a downstream node: persist a fresh continuation. @@ -1149,6 +1226,12 @@ export class AutomationEngine implements IAutomationService { steps, error: errorMessage, }); + // Subflow chain: a child failing terminally fails every + // ancestor awaiting it — they can never be resumed otherwise. + // The delegation path handles its own level (skipBubble). + if (!skipBubble) { + await this.failAncestors(run.context, errorMessage); + } return { success: false, error: errorMessage, durationMs }; } } finally { @@ -1156,6 +1239,91 @@ export class AutomationEngine implements IAutomationService { } } + /** + * Build the resume signal that maps a completed subflow child's output + * into its parent — mirroring the synchronous path exactly: the engine's + * standard `signal.output` merge lands it under `${subflowNodeId}.output`, + * and `signal.variables` writes the bare `config.outputVariable` when the + * child's context carries one (`$parentOutputVariable`). + */ + private buildSubflowResumeSignal(childContext: AutomationContext | undefined, childOutput: unknown): ResumeSignal { + const outVar = (childContext as Record | undefined)?.$parentOutputVariable; + return { + output: { output: childOutput ?? null }, + ...(typeof outVar === 'string' && outVar + ? { variables: { [outVar]: childOutput ?? null } } + : {}), + }; + } + + /** + * Up-bubble for the subflow chain: when a completed run carries + * `$parentRunId`, resume that parent with this run's output. Recursion via + * the parent's own completion bubbles multi-level chains. Best-effort — + * a failed parent continuation is logged, never thrown back at the + * caller who resumed the child. + */ + private async bubbleToParent(run: SuspendedRun, output: Record): Promise { + const parentRunId = (run.context as Record | undefined)?.$parentRunId; + if (typeof parentRunId !== 'string' || !parentRunId) return; + try { + const sig = this.buildSubflowResumeSignal(run.context, output); + const parentRes = await this.resumeInternal(parentRunId, sig, false); + if (!parentRes.success) { + this.logger.warn( + `[automation] subflow run '${run.runId}' completed but resuming parent '${parentRunId}' failed: ${parentRes.error}`, + ); + } + } catch (err) { + this.logger.warn( + `[automation] subflow run '${run.runId}' completed but resuming parent '${parentRunId}' threw: ${(err as Error).message}`, + ); + } + } + + /** + * Terminally fail a suspended run: consume its continuation and record a + * `failed` log so it stops surfacing as resumable. Used when a subflow + * descendant fails — the ancestor awaiting it can never be resumed. + */ + private async failSuspendedRun(run: SuspendedRun, error: string): Promise { + await this.forgetSuspendedRun(run.runId); + this.recordLog({ + id: run.runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'failed', + startedAt: run.startedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - run.startTime, + trigger: { + type: run.context?.event ?? 'manual', + userId: run.context?.userId, + object: run.context?.object, + }, + steps: run.steps, + error, + }); + } + + /** + * Walk a failed run's `$parentRunId` chain and fail each suspended + * ancestor (see {@link failSuspendedRun}). Bounded so a corrupt context + * can't loop forever. + */ + private async failAncestors(context: AutomationContext | undefined, error: string): Promise { + let parentId = (context as Record | undefined)?.$parentRunId; + let hops = 0; + while (typeof parentId === 'string' && parentId && hops++ < 32) { + const parent = + this.suspendedRuns.get(parentId) ?? + (this.store ? await this.store.load(parentId).catch(() => null) : null); + if (!parent) return; + await this.failSuspendedRun(parent, `subflow descendant failed: ${error}`); + parentId = (parent.context as Record | undefined)?.$parentRunId; + } + } + /** * List the runs currently suspended awaiting {@link resume} (ADR-0019). * Backs operability surfaces such as a "pending approvals" view.