diff --git a/.changeset/28897-region-call-suspending-subflow.md b/.changeset/28897-region-call-suspending-subflow.md new file mode 100644 index 00000000000..213d4c35060 --- /dev/null +++ b/.changeset/28897-region-call-suspending-subflow.md @@ -0,0 +1,20 @@ +--- +'@objectstack/service-automation': patch +'@objectstack/spec': patch +--- + +feat(service-automation)!: a flow whose region body calls a child that can suspend is registered but not armed, and the region refusal cancels the child run it leaves behind (#28897) + +Clause-②: yes (narrowing) +Metadata-protocol: none + +**BREAKING** for a record-triggered flow with a `subflow` or `map` node inside a `loop` body, a `parallel` branch or a `try_catch` slot, whose child flow can suspend the run. + +- **Before.** A region body cannot carry a durable pause (ruling D on #15646). The flow parse refuses the pausing node types inside a region, but it cannot see into a `subflow` / `map` child. Such a flow armed, and the first time its child suspended, the run failed at the region boundary with the named refusal (`durable pause inside a structured region …`). The child run that node had started was left parked in the suspended-run store: a later resume ran the rest of the child for a caller that had already failed, and then failed to resume that caller. +- **Now, the first door.** The arming gate in `activateFlowTrigger` declines such a flow, after its disabled-subflow and save-notice halves. "Can suspend" is the save-notice half's reading: the child holds a node of an always-suspending type reachable from its start, or calls, from its top-level graph, a child that can suspend. The flow still registers, is not armed, and warns once, naming the region node (the innermost container), the `subflow` / `map` node and the child's suspending node. `GET /automation/_status` reports it `bound: false` with that reason and the new `unboundCause` value `suspending_subflow_in_region` (added to `FLOW_UNBOUND_CAUSES` in `@objectstack/spec/contracts`). Whenever a flow is registered, every flow whose calls reach it is judged again, so the decline does not depend on load order, and a child republished without its suspension arms the caller again. +- **Now, the second door.** A run that still meets the region refusal (a direct execute, or a flow with no trigger) fails exactly as before, with the same refusal. Before throwing it, the engine cancels the child run the `subflow` / `map` node parked, and every run that child is parked on, through `cancelRun`. Each cancelled run records `cancelled`, with the refusal as its reason. +- **Not declined.** A region-nested call whose child cannot suspend, and a top-level call to a child that can (the top-level graph carries the pause). Registration, the save door and `objectstack validate` accept the same flows as before. +- **The fix.** Move the `subflow` / `map` node the reason names onto the top-level graph and route the region's exit to it, or republish the child without the node that suspends. The flow arms on its own once the shape is gone. +- **Who moves.** No shipped flow has the shape. HotCRM's `campaign_enrollment` calls two children from loop bodies, and neither child suspends. The showcase and dogfood calls are all top-level. + + diff --git a/packages/services/service-automation/src/dag-traversal.ts b/packages/services/service-automation/src/dag-traversal.ts index 14c4714ae09..373f5d850f8 100644 --- a/packages/services/service-automation/src/dag-traversal.ts +++ b/packages/services/service-automation/src/dag-traversal.ts @@ -34,6 +34,7 @@ import { FlowSuspendSignal, isRefusalSignal, isSuspendSignal, + nestedChildRunId, type ExecutionLogEntry, type NodeExecutionResult, type StepLogEntry, @@ -141,6 +142,8 @@ export interface DagTraversalHost { readonly persistSuspendedRun: AutomationEngine['persistSuspendedRun']; readonly flowChecksumOf: AutomationEngine['flowChecksumOf']; readonly renderFlowTerminalMessage: AutomationEngine['renderFlowTerminalMessage']; + readonly resolveSuspensionChain: AutomationEngine['resolveSuspensionChain']; + readonly cancelRun: AutomationEngine['cancelRun']; } /** The body of {@link AutomationEngine.executeNode}, moved unedited; that method's docblock is its contract. */ @@ -746,9 +749,9 @@ export async function traverseNext( } } -/** The body of {@link AutomationEngine.runRegion}, moved unedited; that method's docblock is its contract. */ +/** The body of {@link AutomationEngine.runRegion}; that method's docblock is its contract. */ export async function runRegion( - this: Pick, + this: Pick, region: FlowRegionParsed, variables: Map, context: AutomationContext, @@ -835,12 +838,42 @@ export async function runRegion( if (isSuspendSignal(err)) { const suspended = region.nodes.find(n => n.id === err.nodeId); const flowName = (suspended?.config as { flowName?: unknown } | undefined)?.flowName; - throw refuseRegionSuspension({ + const refusal = refuseRegionSuspension({ regionNodeId: grouping?.parentNodeId ?? entryId, regionKind: grouping?.regionKind ?? 'region', suspendedNodeId: err.nodeId, ...(typeof flowName === 'string' && flowName ? { subFlowName: flowName } : {}), }); + // [#28897] A `subflow` / `map` node suspends by parking the CHILD + // run it started and linking it through its correlation + // (`subflow:` / `map:`). This run fails + // here, so nothing will ever resume through that link: measured, + // the child stayed parked in the durable store, and a later + // resume ran the rest of the child for a caller that had already + // failed, then failed to resume that caller. So the child run is + // cancelled, with every run it is itself parked on, deepest + // first, through the engine's own `cancelRun`. ⛔ The refusal is + // thrown exactly as before: a failed cancellation is logged, never + // put in its place. + const childRunId = nestedChildRunId(err.correlation); + if (childRunId !== undefined) { + try { + const chain = await this.resolveSuspensionChain(childRunId); + const reason = `the run that started it failed at a region boundary: ${refusal.message}`; + for (const run of [...chain].reverse()) await this.cancelRun(run.runId, reason); + } catch (cancelErr) { + // #4632 verdict: DURABILITY, `error`, as `cancelRun`'s own + // store failure: the child may still be parked. + this.logger.error( + `[automation] node '${err.nodeId}' suspended inside region '${refusal.regionNodeId}', and ` + + `cancelling the child run '${childRunId}' it parked failed, so that run may still be ` + + `parked. Cancel it with cancelRun('${childRunId}'). The failure is in this record's meta.`, + undefined, + describeThrownForLog(cancelErr), + ); + } + } + throw refusal; } // [#15788] The refusing `end` node's signal, converted at exactly // the same boundary and for the same reason: a control signal must diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 6322db7a15d..d49846ed3c3 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -222,6 +222,7 @@ import { refuseCustomerAuthoredToggle, } from './subflow-guards.js'; import { describeSubflowNoticeTrap, subflowNoticeTrap } from './subflow-notice-guard.js'; +import { describeRegionSuspendingCall, regionSuspendingCall } from './subflow-region-guard.js'; // [#25222] The consumed-suspension restore's bodies: journaling a consumed // suspension, resolving its witnesses, and restoring, peeking at or inspecting // one. Each `AutomationEngine` method of the same name calls its body. @@ -2915,6 +2916,16 @@ export class AutomationEngine implements IAutomationService { * registered, and dropped with the flow by {@link withdrawFlow}. */ private readonly subflowNoticeDeclinedFlows = new Map(); + /** + * [#28897] Flows {@link activateFlowTrigger} declined to arm because a + * `subflow` / `map` node inside one of their region bodies calls a child + * flow that can suspend, each with the reason `/_status` and the binding + * audit read. A record for {@link subflowNoticeDeclinedFlows}'s reasons, + * kept the same way: deleted the moment the flow gets past the gate, + * re-asked by {@link rejudgeSubflowCallers}, dropped by + * {@link withdrawFlow}. + */ + private readonly subflowRegionDeclinedFlows = new Map(); /** Connectors registered by integration plugins, keyed by connector name (ADR-0018 §Addendum). */ private connectors = new Map(); /** Connector provider factories keyed by provider name (ADR-0097 §2 — `openapi`/`mcp`/`rest`/…). */ @@ -3899,6 +3910,15 @@ export class AutomationEngine implements IAutomationService { // so the gate asks here and {@link rejudgeSubflowCallers} asks again // when the child arrives or changes. if (this.declineOntoSuspendingSubflowNotice(flowName, resolved.triggerType)) return; + // [#28897] The region half, a decline for the same reasons: a region + // body cannot carry a durable pause, so a region-nested `subflow` / + // `map` whose child suspends fails the run at the region boundary + // (`region-suspension-refusal.ts`), the first time that child + // suspends. That refusal stays as the second door; this is the first. + // Last of the halves, and {@link describeUnboundReason} reads the + // records in the same order, so a record an earlier half's decline + // left unasked is never the one reported. + if (this.declineOntoRegionSuspendingCall(flowName, resolved.triggerType)) return; const trigger = this.triggers.get(resolved.triggerType); if (!trigger) return; try { @@ -4649,6 +4669,7 @@ export class AutomationEngine implements IAutomationService { policyDisabledFlows: this.policyDisabledFlows, subflowDeclinedFlows: this.subflowDeclinedFlows, subflowNoticeDeclinedFlows: this.subflowNoticeDeclinedFlows, + subflowRegionDeclinedFlows: this.subflowRegionDeclinedFlows, flowRefusals: this.flowRefusals, logger: this.logger, }; @@ -4780,6 +4801,7 @@ export class AutomationEngine implements IAutomationService { policyDisabledFlows: this.policyDisabledFlows, subflowDeclinedFlows: this.subflowDeclinedFlows, subflowNoticeDeclinedFlows: this.subflowNoticeDeclinedFlows, + subflowRegionDeclinedFlows: this.subflowRegionDeclinedFlows, triggers: this.triggers, }; return describeUnboundReason.call(host, name, resolved); @@ -5222,6 +5244,11 @@ export class AutomationEngine implements IAutomationService { * child be switched off under it. The regions are read through * `collectFlowGraphs`, the spec's one region walk, never a second walker. * + * [#28897] That reading is `subflowCalls` (`subflow-region-guard.ts`), + * which also keeps each call's node and the container holding it: the + * arming gate's region half ({@link declineOntoRegionSuspendingCall}) + * reads the same calls, never a walk of its own. + * * ⛔ The historical `flow` alias for `config.flowName` is NOT read here. * ADR-0087 D2's conversion `flow-node-subflow-flow-alias` canonicalizes it * at load, so only the canonical key reaches a registered definition; @@ -5559,6 +5586,48 @@ export class AutomationEngine implements IAutomationService { return true; } + /** + * [#28897] The arming gate's region half, asked by + * {@link activateFlowTrigger} of an enabled flow it is about to arm: does a + * `subflow` / `map` node inside one of its region bodies (a `loop` body, a + * `parallel` branch, a `try_catch` slot, at any depth) call a child flow + * that can suspend ({@link regionSuspendingCall}, which reads the calls + * through {@link subflowTargets}' reading and where the child suspends + * through the save-notice half's)? A region body cannot carry a durable + * pause, so that run would fail at the region boundary the first time the + * child suspends (`region-suspension-refusal.ts`). Then it is declined: + * recorded on {@link subflowRegionDeclinedFlows} for `/_status` and the + * binding audit, and said once as a `warn` naming the region node, the call + * node and where the child suspends. + * + * Any flow, packaged or authored here, as the save-notice half: the failed + * run is the same. ⛔ A decline, never a refusal, for that half's reasons. + * + * #4632 verdict: FUNCTIONAL, `warn`, as the other halves. Said once per + * decline reason. + * + * @returns `true` when the flow is declined and must not be armed. + */ + private declineOntoRegionSuspendingCall(flowName: string, triggerType: string): boolean { + const trap = regionSuspendingCall(this.flows, flowName); + if (!trap) { + this.subflowRegionDeclinedFlows.delete(flowName); + return false; + } + const why = describeRegionSuspendingCall(trap); + const reason = + `not armed — ${why}, and it is armed; it is also armed once '${trap.child}' is republished without a ` + + `node that suspends`; + const previous = this.subflowRegionDeclinedFlows.get(flowName); + this.subflowRegionDeclinedFlows.set(flowName, reason); + if (previous === reason) return true; + this.logger.warn( + `Flow '${flowName}' is registered but NOT armed on trigger '${triggerType}' — ${why}. It is armed the ` + + `moment '${trap.callNode.id}' moves, or '${trap.child}' is republished without a node that suspends.`, + ); + return true; + } + /** * [#20725, ADR-0126 §7.3] `subflow` just changed state — (re)registered, * switched on or off, or read off the ledger at boot — so ask the arming @@ -5582,6 +5651,9 @@ export class AutomationEngine implements IAutomationService { * change two calls away moves the answer. A caller that half declined is * re-offered, and an armed caller it would now decline is unbound and * offered again, both through the one gate. + * + * [#28897] The region half reads the same children through the same + * calls, so the same callers are re-asked for it, the same way. */ private rejudgeSubflowCallers(subflow: string): void { // The body lives in ./subflow-guards.ts (#25222). @@ -5589,6 +5661,7 @@ export class AutomationEngine implements IAutomationService { boundFlowTriggers: this.boundFlowTriggers, subflowDeclinedFlows: this.subflowDeclinedFlows, subflowNoticeDeclinedFlows: this.subflowNoticeDeclinedFlows, + subflowRegionDeclinedFlows: this.subflowRegionDeclinedFlows, flows: this.flows, subflowTargets: (...a) => this.subflowTargets(...a), packagedSubflowCallers: (...a) => this.packagedSubflowCallers(...a), @@ -9267,6 +9340,12 @@ export class AutomationEngine implements IAutomationService { * failure, ran its catch handler, and the run reported success over a sweep * that had processed nothing. The container executors test for the named * type and re-throw, so no region can contain it. + * + * [#28897] When the suspending node is a `subflow` / `map`, the child run + * it parked (and any run that child is parked on) is cancelled before the + * refusal is thrown, so no run is left parked under a caller that failed. + * The arming gate declines such a caller first + * ({@link declineOntoRegionSuspendingCall}); this is the second door. */ runRegion( region: FlowRegionParsed, @@ -9278,6 +9357,9 @@ export class AutomationEngine implements IAutomationService { // The body lives in ./dag-traversal.ts (#25222). const host: ThisParameterType = { executeNode: (...a) => this.executeNode(...a), + resolveSuspensionChain: (...a) => this.resolveSuspensionChain(...a), + cancelRun: (...a) => this.cancelRun(...a), + logger: this.logger, }; return runRegion.call(host, region, variables, context, grouping, partialSteps); } diff --git a/packages/services/service-automation/src/flow-definitions.ts b/packages/services/service-automation/src/flow-definitions.ts index b6bcf48b2c4..30d2f0f2755 100644 --- a/packages/services/service-automation/src/flow-definitions.ts +++ b/packages/services/service-automation/src/flow-definitions.ts @@ -75,6 +75,7 @@ export interface FlowDefinitionsHost { readonly policyDisabledFlows: AutomationEngine['policyDisabledFlows']; readonly subflowDeclinedFlows: AutomationEngine['subflowDeclinedFlows']; readonly subflowNoticeDeclinedFlows: AutomationEngine['subflowNoticeDeclinedFlows']; + readonly subflowRegionDeclinedFlows: AutomationEngine['subflowRegionDeclinedFlows']; readonly triggers: AutomationEngine['triggers']; readonly flowCredentialSource: AutomationEngine['flowCredentialSource']; readonly flowVersionHistory: AutomationEngine['flowVersionHistory']; @@ -290,7 +291,7 @@ export function getFlowRuntimeStates(this: Pick, + this: Pick, name: string, resolved: { triggerType: string } | undefined, ): { reason: string; cause: FlowUnboundCause } | undefined { @@ -311,6 +312,13 @@ export function describeUnboundReason( // stops suspending). const noticeDeclined = this.subflowNoticeDeclinedFlows.get(name); if (noticeDeclined !== undefined) return { reason: noticeDeclined, cause: 'notice_after_suspending_subflow' }; + // [#28897] The gate's region half declined it: a `subflow` / `map` node + // inside a region body calls a child that can suspend, and the region + // cannot carry the pause. Its own cause, for its own remedy (the author + // moves the call onto the top-level graph, or the child stops suspending). + // Read last, the gate's own order. + const regionDeclined = this.subflowRegionDeclinedFlows.get(name); + if (regionDeclined !== undefined) return { reason: regionDeclined, cause: 'suspending_subflow_in_region' }; return this.triggers.has(resolved.triggerType) ? { reason: `trigger '${resolved.triggerType}' is registered but binding failed — see earlier warnings`, @@ -423,7 +431,7 @@ export function unregisterConnectorProvider(this: Pick, name: string): void { +export function withdrawFlow(this: Pick, name: string): void { this.deactivateFlowTrigger(name); this.flows.delete(name); // [ADR-0126 §7.2] `flowLedgerDisabled` is deliberately NOT cleared. It @@ -441,6 +449,7 @@ export function withdrawFlow(this: Pick { expect(thrown.message).toContain("Flow 'shared_step' cannot be disabled"); }); + it('enable: re-enabling a switched-off caller whose loop body calls a switched-off packaged child is refused with RESOURCE_CONFLICT / 409, naming the child (#28897)', async () => { + const { engine, isBound } = engineWithLedger(); + engine.registerFlow('shared_step', child('shared_step')); + await engine.toggleFlow('shared_step', false); + engine.registerFlow('vendor_process', regionCaller('vendor_process', 'shared_step', 'loop', 'map')); + await engine.toggleFlow('vendor_process', false); + + const thrown: any = await engine.toggleFlow('vendor_process', true).then(() => undefined, (e: unknown) => e); + expect(thrown, 're-enabling the caller was accepted').toBeDefined(); + expect(thrown.code).toBe('RESOURCE_CONFLICT'); + expect(thrown.status).toBe(409); + expect(thrown.message).toContain("Flow 'vendor_process' cannot be enabled"); + expect(thrown.message).toContain("'shared_step'"); + expect(isBound('vendor_process')).toBe(false); + }); + it('CONTROL: a caller whose loop body calls an ACTIVE packaged child is armed, with no warning', () => { const { engine, logger, isBound } = engineWithLedger(); engine.registerFlow('shared_step', child('shared_step')); @@ -220,12 +237,14 @@ describe('the save-notice half of the arming gate keeps reading top-level calls const WAIT: Node = { id: 'hold', type: 'wait', label: 'Hold', waitEventConfig: { eventType: 'timer', timerDuration: 'PT1H' } }; const NOTICE: Node = { id: 'tell', type: 'save_notice', label: 'Tell the saver', config: { message: 'Saved' } }; - it('a save_notice after a region whose body calls a suspending child is not declined on that call: the region cannot carry the pause, so the run fails at the region, before the notice', () => { + it('a save_notice after a region whose body calls a suspending child is not declined on the notice: the region cannot carry the pause, so the run fails at the region, before the notice, and the gate declines it on the region call instead (#28897)', () => { const { engine, isBound } = engineWithLedger(); engine.registerFlow('deal_review', chain('deal_review', [start(false), WAIT, END])); engine.registerFlow('deal_submit', chain('deal_submit', [start(true), region('loop', call('deal_review')), NOTICE, END])); - expect(isBound('deal_submit')).toBe(true); - expect(stateOf(engine, 'deal_submit')?.unboundCause).toBeUndefined(); + expect(isBound('deal_submit')).toBe(false); + const row = stateOf(engine, 'deal_submit'); + expect(row?.unboundCause).toBe('suspending_subflow_in_region'); + expect(row?.reason).not.toContain('save_notice'); }); it('a child whose only call to a suspending flow sits in a region body does not count as suspending', () => { diff --git a/packages/services/service-automation/src/subflow-guards.ts b/packages/services/service-automation/src/subflow-guards.ts index 64ce62c1df8..d62794a92e6 100644 --- a/packages/services/service-automation/src/subflow-guards.ts +++ b/packages/services/service-automation/src/subflow-guards.ts @@ -5,19 +5,21 @@ * subflow and its callers are found and described, and the refusals that keep * an enable, a toggle or a caller from reaching a disabled subflow. * - * [#25222] Slice 7 of the `engine.ts` split, move-only. Each function here is - * the body of the `AutomationEngine` method of the same name, moved unedited. - * The class keeps that method (its name, docblock and position are in the + * [#25222] Slice 7 of the `engine.ts` split. Each function here is the body of + * the `AutomationEngine` method of the same name, moved there unedited; a body + * edited since says so in its own docblock, which then drops "moved unedited" + * (`describeDisabledFlow`, `subflowTargets`, `rejudgeSubflowCallers`). The + * class keeps that method (its name, docblock and position are in the * published `.d.ts`) and calls the body with a {@link SubflowGuardsHost} * literal holding exactly the members the body reads. An async method's stub is * no longer `async`: it returns the promise the moved function makes. */ -import { collectFlowGraphs } from '@objectstack/spec/automation'; import type { FlowParsed } from '@objectstack/spec/automation'; import { describeTurnedOffPackagedItem } from '@objectstack/types'; import type { AutomationEngine } from './engine.js'; import { subflowNoticeTrap, transitiveSubflowCallers } from './subflow-notice-guard.js'; +import { regionSuspendingCall, subflowCalls } from './subflow-region-guard.js'; /** * The `AutomationEngine` members the moved bodies read, each typed as the class @@ -42,6 +44,7 @@ export interface SubflowGuardsHost { readonly boundFlowTriggers: AutomationEngine['boundFlowTriggers']; readonly subflowDeclinedFlows: AutomationEngine['subflowDeclinedFlows']; readonly subflowNoticeDeclinedFlows: AutomationEngine['subflowNoticeDeclinedFlows']; + readonly subflowRegionDeclinedFlows: AutomationEngine['subflowRegionDeclinedFlows']; readonly packagedSubflowCallers: AutomationEngine['packagedSubflowCallers']; readonly deactivateFlowTrigger: AutomationEngine['deactivateFlowTrigger']; readonly activateFlowTrigger: AutomationEngine['activateFlowTrigger']; @@ -60,16 +63,9 @@ export function describeDisabledFlow(this: Pick call.flowName); } /** The body of {@link AutomationEngine.packagedSubflowCallers}, moved unedited; that method's docblock is its contract. */ @@ -250,8 +246,13 @@ export function describeDisabledSubflows( } /** The body of {@link AutomationEngine.rejudgeSubflowCallers}; that method's docblock is its contract. */ -export function rejudgeSubflowCallers(this: Pick, subflow: string): void { - if (this.boundFlowTriggers.size === 0 && this.subflowDeclinedFlows.size === 0 && this.subflowNoticeDeclinedFlows.size === 0) return; +export function rejudgeSubflowCallers(this: Pick, subflow: string): void { + if ( + this.boundFlowTriggers.size === 0 && + this.subflowDeclinedFlows.size === 0 && + this.subflowNoticeDeclinedFlows.size === 0 && + this.subflowRegionDeclinedFlows.size === 0 + ) return; const reoffered = new Set(); for (const caller of this.packagedSubflowCallers(subflow)) { const settled = this.boundFlowTriggers.has(caller) @@ -264,11 +265,12 @@ export function rejudgeSubflowCallers(this: Pick 2 ? `, which it reaches through ${suspends.through.slice(1, -1).map((f) => `'${f}'`).join(', ')}` : ''}`; + return `\`${suspends.type}\` node '${suspends.id}'${where}`; +} + +/** + * The sentence the warn and the `/_status` reason share, in the #28806 shape: + * the notice, the call node, where the child suspends, and the two moves that + * make the flow armable. + */ +export function describeSubflowNoticeTrap(trap: SubflowNoticeTrap): string { + const { noticeId, callNode, child, suspends } = trap; return ( `its \`save_notice\` node '${noticeId}' is reached after \`${callNode.type}\` node '${callNode.id}', whose ` + - `child flow '${child}' can suspend the run at \`${suspends.type}\` node '${suspends.id}'${where}. A suspended ` + + `child flow '${child}' can suspend the run at ${describeSuspendingNode(child, suspends)}. A suspended ` + `run resumes after the write has already answered, so on that path the node is refused by name at run time, ` + `and after an approval the decision cannot finish its run. Move the \`save_notice\` ahead of ` + `'${callNode.id}', or tell the person after it with a \`notify\` node` diff --git a/packages/services/service-automation/src/subflow-region-guard.ts b/packages/services/service-automation/src/subflow-region-guard.ts new file mode 100644 index 00000000000..92ed9f80535 --- /dev/null +++ b/packages/services/service-automation/src/subflow-region-guard.ts @@ -0,0 +1,133 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#28897] The reading behind the arming gate's region half: does a `subflow` + * / `map` node inside one of a flow's region bodies (a `loop` body, a + * `parallel` branch, a `try_catch` slot, at any depth the spec walks) call a + * child flow that can suspend? + * + * A region body runs synchronously inside the enclosing run and cannot carry a + * durable pause (#3267, ruling D on #15646). The flow parse refuses the + * pausing node TYPES inside a region (#18688), but it cannot see past a + * `subflow` / `map` node: whether that node suspends is a fact of ANOTHER flow, + * which may register after the caller. So the run met it, at the region + * boundary, as the named refusal in `region-suspension-refusal.ts` (#18881), + * the first time the child suspended. The engine now reads it here, over the + * flows it holds, and {@link AutomationEngine.activateFlowTrigger} declines to + * arm such a caller. The run-time refusal stays as the second door. + * + * Where the child suspends is the #28840 reading ({@link suspendingNode}): a + * pausing node reachable from the child's start, or a top-level call of the + * child's whose own child can suspend, transitively. A child's region-nested + * call is not read as a way for the CHILD to suspend: it fails the run at the + * child's own region instead, so it never parks the caller. + */ + +import { collectFlowGraphs } from '@objectstack/spec/automation'; +import type { FlowParsed } from '@objectstack/spec/automation'; +import { describeSuspendingNode, suspendingNode } from './subflow-notice-guard.js'; +import type { SuspendingNode } from './subflow-notice-guard.js'; + +/** A `subflow` / `map` node of a flow and the flow it names. */ +export interface SubflowCall { + readonly id: string; + readonly type: 'subflow' | 'map'; + /** The flow its `config.flowName` names. */ + readonly flowName: string; + /** + * The container node whose region body holds it, the innermost one when + * regions nest. Absent for a node on the top-level graph. + */ + readonly region?: { readonly id: string; readonly type: string }; +} + +/** + * The container node a region graph's `path` names: the object at the path up + * to and including the last `nodes` index (`['nodes', 1, 'config', 'body']` → + * the flow's `nodes[1]`). + */ +function containerAt(flow: FlowParsed, path: readonly (string | number)[]): { id: string; type: string } | undefined { + const last = path.lastIndexOf('nodes'); + if (last < 0) return undefined; + let at: unknown = flow; + for (const key of path.slice(0, last + 2)) { + if (!at || typeof at !== 'object') return undefined; + at = (at as Record)[key]; + } + const node = at as { id?: unknown; type?: unknown } | null | undefined; + return node && typeof node.id === 'string' && typeof node.type === 'string' + ? { id: node.id, type: node.type } + : undefined; +} + +/** + * Every `subflow` / `map` node of `flow` that names a flow, on the top-level + * graph and in every region body `collectFlowGraphs` (`@objectstack/spec`) + * reaches, in that walk's order. The one reading + * {@link AutomationEngine.subflowTargets} and the region half share (#28846, + * #28897), so no second region walk exists. + */ +export function subflowCalls(flow: FlowParsed): SubflowCall[] { + const calls: SubflowCall[] = []; + for (const graph of collectFlowGraphs(flow)) { + const region = graph.path.length === 0 ? undefined : containerAt(flow, graph.path); + for (const node of graph.nodes) { + const n = node as { id?: unknown; type?: unknown; config?: { flowName?: unknown } } | null; + if (!n || (n.type !== 'subflow' && n.type !== 'map')) continue; + if (typeof n.config?.flowName !== 'string') continue; + calls.push({ + id: typeof n.id === 'string' ? n.id : '', + type: n.type, + flowName: n.config.flowName, + ...(region ? { region } : {}), + }); + } + } + return calls; +} + +/** A region-nested `subflow` / `map` node whose child flow can suspend. */ +export interface RegionSuspendingCall { + /** The container node whose region body holds the call (the innermost). */ + readonly region: { readonly id: string; readonly type: string }; + /** The `subflow` / `map` node. */ + readonly callNode: { readonly id: string; readonly type: string }; + /** The flow that node calls. */ + readonly child: string; + /** Where that child can suspend. */ + readonly suspends: SuspendingNode; +} + +/** The first region-nested call of `name` whose child can suspend, or `undefined`. */ +export function regionSuspendingCall( + flows: ReadonlyMap, + name: string, +): RegionSuspendingCall | undefined { + const flow = flows.get(name); + if (!flow) return undefined; + for (const call of subflowCalls(flow)) { + if (!call.region) continue; + const suspends = suspendingNode(flows, call.flowName); + if (suspends) { + return { region: call.region, callNode: { id: call.id, type: call.type }, child: call.flowName, suspends }; + } + } + return undefined; +} + +/** + * The sentence the warn and the `/_status` reason share: the call, the region + * that holds it, where the child suspends, why the run would fail, and the + * move that makes the flow armable, the prescription the run-time refusal + * gives. + */ +export function describeRegionSuspendingCall(trap: RegionSuspendingCall): string { + const { region, callNode, child, suspends } = trap; + return ( + `its \`${callNode.type}\` node '${callNode.id}' sits inside \`${region.type}\` node '${region.id}', and its ` + + `child flow '${child}' can suspend the run at ${describeSuspendingNode(child, suspends)}. A region body ` + + `runs synchronously inside the run and cannot carry a durable pause, so the run would fail at ` + + `'${region.id}' the first time that child suspends, and nothing after it would run. Move '${callNode.id}' ` + + `onto the top-level graph and route the region's exit to it` + ); +} diff --git a/packages/services/service-automation/src/subflow-region-suspension.test.ts b/packages/services/service-automation/src/subflow-region-suspension.test.ts new file mode 100644 index 00000000000..352487c9fc3 --- /dev/null +++ b/packages/services/service-automation/src/subflow-region-suspension.test.ts @@ -0,0 +1,285 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// [#28897] A `subflow` / `map` call inside a region body (a `loop` body, a +// `parallel` branch, a `try_catch` slot) whose child flow can suspend. +// +// A region body cannot carry a durable pause (#3267, ruling D on #15646). When +// such a child suspends, the run fails at the region boundary with the named +// refusal (`region-suspension-refusal.ts`, #18881), and nothing after the +// container runs. Two halves are pinned here: +// +// 1. The arming gate meets it first. A caller whose region-nested call names +// a child that can suspend is registered but not armed, with its own +// `unboundCause`, `suspending_subflow_in_region`, naming the region node, +// the call node and where the child suspends. It is asked again whenever a +// flow its calls reach is registered, as the #28840 save-notice half is. +// 2. A run that meets the refusal anyway (a direct execute, or a caller +// armed under a reading the gate could not see) no longer leaves the child +// run it started parked. Measured before this change: the child stayed +// `paused` in the durable store, and a later resume ran the rest of the +// child for a caller that had already failed, then failed to resume that +// caller. The refusal now cancels the child run, and any run that child is +// parked on, through the engine's own `cancelRun`. +// +// Neither half changes the refusal itself (code, text, un-routability) or +// `FLOW_PAUSE_CAPABLE_NODE_TYPES`. + +import { describe, it, expect, vi } from 'vitest'; +import { AutomationEngine } from './engine.js'; +import type { FlowTrigger, FlowTriggerBinding } from './engine.js'; +import { InMemoryFlowActivationStore } from './flow-activation-store.js'; +import { + registerSubflowNode, + registerMapNode, + registerLoopNode, + registerParallelNode, + registerTryCatchNode, + registerWaitNode, + registerLogicNodes, +} from './builtin/index.js'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import { withLoaderSetFromPull } from './loader-set.test-support.js'; + +function createTestLogger(): any { + const l: any = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }; + l.child = () => l; + return l; +} + +const declineWarnings = (logger: any, caller: string): string[] => + logger.warn.mock.calls + .map((c: unknown[]) => String(c[0])) + .filter((m: string) => m.includes(`Flow '${caller}'`) && m.includes('NOT armed')); + +function recordingTrigger() { + const bound = new Set(); + const trigger: FlowTrigger = { + type: 'record_change', + start(binding: FlowTriggerBinding, _cb: (ctx: AutomationContext) => Promise) { + bound.add(binding.flowName); + }, + stop(flowName: string) { + bound.delete(flowName); + }, + }; + return { trigger, isBound: (n: string) => bound.has(n) }; +} + +function rig() { + const logger = createTestLogger(); + const engine = withLoaderSetFromPull(new AutomationEngine(logger)); + engine.setFlowActivationStore(new InMemoryFlowActivationStore()); + const ctx = { logger: createTestLogger(), getService: () => undefined } as any; + for (const register of [ + registerSubflowNode, registerMapNode, registerLoopNode, registerParallelNode, + registerTryCatchNode, registerWaitNode, registerLogicNodes, + ]) { + (register as (e: AutomationEngine, c: unknown) => void)(engine, ctx); + } + const trigger = recordingTrigger(); + engine.registerTrigger(trigger.trigger); + return { engine, logger, isBound: trigger.isBound }; +} + +const stateOf = (engine: AutomationEngine, name: string) => + engine.getFlowRuntimeStates().find((s) => s.name === name); + +type Node = Record; +type Region = 'loop' | 'parallel' | 'try_catch'; +type CallType = 'subflow' | 'map'; + +const END: Node = { id: 'end', type: 'end', label: 'End' }; +const start = (trigger: boolean): Node => ({ + id: 'start', type: 'start', label: 'Start', + config: trigger ? { objectName: 'lead', triggerType: 'record-after-create' } : {}, +}); +const call = (flowName: string, type: CallType = 'subflow', id = 'step'): Node => ({ + id, type, label: `Call ${flowName}`, + config: { flowName, ...(type === 'map' ? { collection: [1] } : {}) }, +}); +const NOOP: Node = { id: 'noop', type: 'assignment', label: 'Nothing', config: {} }; +const WAIT: Node = { id: 'hold', type: 'wait', label: 'Hold', waitEventConfig: { eventType: 'timer', timerDuration: 'PT1H' } }; + +/** `inner` as the one node of a region body of `kind`, in a container node `id`. */ +function region(kind: Region, inner: Node, id = 'box'): Node { + const body = { nodes: [inner], edges: [] }; + if (kind === 'loop') { + return { id, type: 'loop', label: 'Box', config: { collection: [1], iteratorVariable: 'item', body } }; + } + if (kind === 'parallel') { + return { + id, type: 'parallel', label: 'Box', + config: { branches: [{ name: 'a', ...body }, { name: 'b', nodes: [{ ...NOOP, id: `${id}_noop` }], edges: [] }] }, + }; + } + return { + id, type: 'try_catch', label: 'Box', + config: { try: body, catch: { nodes: [{ ...NOOP, id: `${id}_noop` }], edges: [] } }, + }; +} + +function chain(name: string, nodes: Node[], extra: Record = {}) { + return { + name, + label: name, + type: 'autolaunched', + nodes, + edges: nodes.slice(1).map((n, i) => ({ id: `e${i}`, source: nodes[i].id as string, target: n.id as string })), + ...extra, + }; +} + +/** A child flow that suspends at its `wait` node `hold`. */ +const pausingChild = (name: string) => chain(name, [start(false), WAIT, END]); +/** A child flow that runs straight through. */ +const plainChild = (name: string, extra: Record = {}) => chain(name, [start(false), END], extra); + +/** A record-triggered caller whose only call to `target` sits inside a region body. */ +const regionCaller = (name: string, target: string, kind: Region, type: CallType = 'subflow', trigger = true) => + chain(name, [start(trigger), region(kind, call(target, type)), END]); + +const CASES: Array<[Region, CallType]> = [ + ['loop', 'subflow'], ['loop', 'map'], + ['parallel', 'subflow'], ['parallel', 'map'], + ['try_catch', 'subflow'], ['try_catch', 'map'], +]; + +describe('the arming gate declines a region-nested call to a child that can suspend (#28897)', () => { + it.each(CASES)('a caller whose %s body holds a `%s` call to a suspending child is registered but not armed, naming the region, the call and the pause', (kind, type) => { + const { engine, logger, isBound } = rig(); + engine.registerFlow('deal_review', pausingChild('deal_review')); + engine.registerFlow('deal_submit', regionCaller('deal_submit', 'deal_review', kind, type)); + + expect(engine.getFlow('deal_submit'), 'the caller is still registered').toBeDefined(); + expect(isBound('deal_submit')).toBe(false); + const row = stateOf(engine, 'deal_submit'); + expect(row).toMatchObject({ enabled: true, bound: false, unboundCause: 'suspending_subflow_in_region' }); + expect(row?.reason).toContain(`\`${type}\` node 'step' sits inside \`${kind}\` node 'box'`); + expect(row?.reason).toContain("child flow 'deal_review' can suspend the run at `wait` node 'hold'"); + expect(declineWarnings(logger, 'deal_submit')).toHaveLength(1); + }); + + it('names the innermost container when the call is two regions deep', () => { + const { engine, isBound } = rig(); + engine.registerFlow('deal_review', pausingChild('deal_review')); + engine.registerFlow( + 'deal_submit', + chain('deal_submit', [start(true), region('loop', region('parallel', call('deal_review'), 'fan')), END]), + ); + expect(isBound('deal_submit')).toBe(false); + const row = stateOf(engine, 'deal_submit'); + expect(row?.unboundCause).toBe('suspending_subflow_in_region'); + expect(row?.reason).toContain("`subflow` node 'step' sits inside `parallel` node 'fan'"); + }); + + it('reads where the child suspends through its own top-level calls', () => { + const { engine, isBound } = rig(); + engine.registerFlow('deal_ask', pausingChild('deal_ask')); + engine.registerFlow('deal_review', chain('deal_review', [start(false), call('deal_ask'), END])); + engine.registerFlow('deal_submit', regionCaller('deal_submit', 'deal_review', 'loop')); + expect(isBound('deal_submit')).toBe(false); + expect(stateOf(engine, 'deal_submit')?.reason).toContain("at `wait` node 'hold' of 'deal_ask'"); + }); + + it('is asked again on every child registration: a caller armed before its child turns suspending is disarmed, and armed again when the child stops', () => { + const { engine, isBound } = rig(); + engine.registerFlow('deal_submit', regionCaller('deal_submit', 'deal_review', 'loop', 'map')); + expect(isBound('deal_submit'), 'no child yet: nothing can suspend').toBe(true); + + engine.registerFlow('deal_review', pausingChild('deal_review')); + expect(isBound('deal_submit')).toBe(false); + expect(stateOf(engine, 'deal_submit')?.unboundCause).toBe('suspending_subflow_in_region'); + + engine.registerFlow('deal_review', plainChild('deal_review')); + expect(isBound('deal_submit')).toBe(true); + expect(stateOf(engine, 'deal_submit')?.unboundCause).toBeUndefined(); + }); + + it('is asked again when a flow two calls away changes', () => { + const { engine, isBound } = rig(); + engine.registerFlow('deal_ask', plainChild('deal_ask')); + engine.registerFlow('deal_review', chain('deal_review', [start(false), call('deal_ask'), END])); + engine.registerFlow('deal_submit', regionCaller('deal_submit', 'deal_review', 'try_catch')); + expect(isBound('deal_submit')).toBe(true); + + engine.registerFlow('deal_ask', pausingChild('deal_ask')); + expect(isBound('deal_submit')).toBe(false); + expect(stateOf(engine, 'deal_submit')?.unboundCause).toBe('suspending_subflow_in_region'); + }); + + it('CONTROL: a top-level call to a suspending child arms, since the top-level graph carries the pause', () => { + const { engine, logger, isBound } = rig(); + engine.registerFlow('deal_review', pausingChild('deal_review')); + engine.registerFlow('deal_submit', chain('deal_submit', [start(true), call('deal_review'), END])); + expect(isBound('deal_submit')).toBe(true); + expect(stateOf(engine, 'deal_submit')?.unboundCause).toBeUndefined(); + expect(declineWarnings(logger, 'deal_submit')).toEqual([]); + }); + + it('CONTROL (the HotCRM campaign_enrollment shape): two loop bodies calling packaged children that do not suspend still arm, and the run completes', async () => { + const { engine, logger, isBound } = rig(); + const PACKAGED = { _packageId: 'crm' }; + engine.registerFlow('lead_member_enroll', plainChild('lead_member_enroll', PACKAGED)); + engine.registerFlow('contact_member_enroll', plainChild('contact_member_enroll', PACKAGED)); + engine.registerFlow('enrollment', chain('enrollment', [ + start(true), + region('loop', call('lead_member_enroll', 'subflow', 'enroll_lead'), 'loop_leads'), + region('loop', call('contact_member_enroll', 'subflow', 'enroll_contact'), 'loop_contacts'), + END, + ], PACKAGED)); + + expect(isBound('enrollment')).toBe(true); + expect(stateOf(engine, 'enrollment')?.unboundCause).toBeUndefined(); + expect(declineWarnings(logger, 'enrollment')).toEqual([]); + const result = await engine.execute('enrollment', {} as any); + expect(result.success).toBe(true); + }); +}); + +describe('the region refusal cancels the child run it leaves behind (#28897)', () => { + it.each(CASES)('%s / `%s`: the run fails at the region as before, and the suspended child run is cancelled, not left parked', async (kind, type) => { + const { engine } = rig(); + engine.registerFlow('deal_review', pausingChild('deal_review')); + engine.registerFlow('deal_submit', regionCaller('deal_submit', 'deal_review', kind, type, false)); + + const result: any = await engine.execute('deal_submit', {} as any); + // The refusal itself is unchanged. + expect(result.success).toBe(false); + expect(String(result.error)).toContain( + `durable pause inside a structured region: node 'step' (sub-flow 'deal_review') suspended`, + ); + + expect(await engine.listSuspendedRunsDurable(), 'no run is left parked').toEqual([]); + expect(engine.listSuspendedRuns()).toEqual([]); + const runs = await engine.listRuns('deal_review'); + expect(runs).toHaveLength(1); + expect(runs[0].status).toBe('cancelled'); + expect(String(runs[0].error)).toContain('durable pause inside a structured region'); + }); + + it('a child parked on a grandchild is cancelled with it', async () => { + const { engine } = rig(); + engine.registerFlow('deal_ask', pausingChild('deal_ask')); + engine.registerFlow('deal_review', chain('deal_review', [start(false), call('deal_ask', 'subflow', 'ask'), END])); + engine.registerFlow('deal_submit', regionCaller('deal_submit', 'deal_review', 'loop', 'subflow', false)); + + const result: any = await engine.execute('deal_submit', {} as any); + expect(result.success).toBe(false); + expect(await engine.listSuspendedRunsDurable()).toEqual([]); + for (const name of ['deal_review', 'deal_ask']) { + const runs = await engine.listRuns(name); + expect(runs.map((r: any) => r.status), name).toEqual(['cancelled']); + } + }); + + it('CONTROL: a top-level call to a suspending child still parks both runs', async () => { + const { engine } = rig(); + engine.registerFlow('deal_review', pausingChild('deal_review')); + engine.registerFlow('deal_submit', chain('deal_submit', [start(false), call('deal_review'), END])); + + const result: any = await engine.execute('deal_submit', {} as any); + expect(result.status).toBe('paused'); + const parked = (await engine.listSuspendedRunsDurable()).map((r) => r.flowName).sort(); + expect(parked).toEqual(['deal_review', 'deal_submit']); + }); +}); diff --git a/packages/spec/src/contracts/automation-service.test.ts b/packages/spec/src/contracts/automation-service.test.ts index 7d2e8ea9c5c..07eb919ddc9 100644 --- a/packages/spec/src/contracts/automation-service.test.ts +++ b/packages/spec/src/contracts/automation-service.test.ts @@ -60,7 +60,7 @@ export type FlowRuntimeStateUnboundCauseIsTheClosedTuple = Assert< Eq >; export type FlowUnboundCauseIsTheTupleUnion = Assert< - Eq + Eq >; describe('Automation Service Contract', () => { diff --git a/packages/spec/src/contracts/automation-service.ts b/packages/spec/src/contracts/automation-service.ts index 131c90caffd..6ca57f728e3 100644 --- a/packages/spec/src/contracts/automation-service.ts +++ b/packages/spec/src/contracts/automation-service.ts @@ -949,6 +949,13 @@ export interface FlowRuntimeState { * notice would be refused by name and strand the run. Author remedy: move the * `save_notice` ahead of that node or use a `notify` node, or republish the * child without a node that suspends, and the flow arms. + * - `suspending_subflow_in_region` — the arming gate declined the flow because + * a `subflow` / `map` node inside one of its region bodies (a `loop` body, a + * `parallel` branch, a `try_catch` slot) calls a child flow that can suspend + * (#28897): a region body cannot carry a durable pause, so the run would fail + * at the region boundary the first time that child suspends. Author remedy: + * move the call onto the top-level graph, or republish the child without a + * node that suspends, and the flow arms. * - `missing_trigger` — no trigger of the declared type is registered on this * kernel. Author remedy: depend on the trigger package * (`requires: ['triggers']`). @@ -959,6 +966,7 @@ export const FLOW_UNBOUND_CAUSES = [ 'deployment_policy', 'disabled_subflow', 'notice_after_suspending_subflow', + 'suspending_subflow_in_region', 'missing_trigger', 'binding_failure', ] as const;