From 65a9454675b4b557c5ca827a63b9748c1ba1ac26 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 06:06:00 +0000 Subject: [PATCH] feat(service-automation)!: decline to arm a region-nested call to a child that can suspend, and cancel the child run the region refusal leaves behind A region body cannot carry a durable pause, so a `subflow` / `map` node inside a `loop` body, a `parallel` branch or a `try_catch` slot whose child suspends fails the run at the region boundary. The arming gate now declines such a caller with its own unbound cause, `suspending_subflow_in_region`, re-judged whenever a flow its calls reach is registered. The region refusal itself is unchanged, but before it is thrown the engine cancels the child run the call parked, and any run that child is parked on. Co-Authored-By: Claude --- .../28897-region-call-suspending-subflow.md | 20 ++ .../service-automation/src/dag-traversal.ts | 39 ++- .../services/service-automation/src/engine.ts | 82 +++++ .../src/flow-definitions.ts | 13 +- .../src/region-suspension-refusal.ts | 18 +- .../src/subflow-guard-region-calls.test.ts | 27 +- .../service-automation/src/subflow-guards.ts | 36 +-- .../src/subflow-notice-guard.ts | 22 +- .../src/subflow-region-guard.ts | 133 ++++++++ .../src/subflow-region-suspension.test.ts | 285 ++++++++++++++++++ .../src/contracts/automation-service.test.ts | 2 +- .../spec/src/contracts/automation-service.ts | 8 + 12 files changed, 650 insertions(+), 35 deletions(-) create mode 100644 .changeset/28897-region-call-suspending-subflow.md create mode 100644 packages/services/service-automation/src/subflow-region-guard.ts create mode 100644 packages/services/service-automation/src/subflow-region-suspension.test.ts 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;