diff --git a/.changeset/resume-gate-map-chain-and-reserved-vars.md b/.changeset/resume-gate-map-chain-and-reserved-vars.md new file mode 100644 index 0000000000..29fa2a8aef --- /dev/null +++ b/.changeset/resume-gate-map-chain-and-reserved-vars.md @@ -0,0 +1,42 @@ +--- +"@objectstack/service-automation": patch +"@objectstack/runtime": patch +--- + +fix(automation): the resume gate follows `map:` too, and the route stops accepting engine-internal variables (#3853) + +Two holes in the #3801 resume gate, both demonstrated with a repro. + +**1. The chain walk missed `map:`.** `resumeInternal` handles the two linked-run +correlations oppositely — a `subflow:` pause *delegates* the signal to the child, +a `map:` pause *re-runs* the map node — and the gate followed only the first. So +a run parked on a `map` node was judged on `map` itself (`resumeAuthority: 'any'`) +and let through even while the item it was waiting on sat on an `approval`. + +`map` is the batch-approval shape, and the map parent's run id is the one a +launcher holds. Since `$mapState.started` is advanced past the in-flight item +before the suspend, an empty-body resume of the parent **skipped that item's +approval outright**, orphaning its still-pending request; a later real decision +then bubbled into a parent already waiting on the next item, cascading the +misalignment. + +The walk now follows both prefixes: a linked-run pause is waiting on a CHILD, so +the child's node carries the authority — the gate reads *the item, not the loop*. + +**2. Resume `inputs` could write the engine's `$` namespace.** They are applied +as bare flow variables, so a caller could set the exact handoff keys the engine's +map bubble uses (`.$mapItemDone` / `$mapItemOutput`) and have the map +record a per-item result for a decision nobody made — the node id is readable +from `GET /automation/:name`. The same reached `$runId`, which `approval` / +`wait` nodes use to correlate external state back to a run. + +`POST /automation/:name/runs/:runId/resume` now answers **400** when `inputs` +names anything in the engine namespace (`$…`, or a `.$` segment). Enforced at the +transport, not in the engine, so the in-process bubble keeps working — the same +trust split the gate itself uses. + +Nothing changes for author-declared variables: `{ new_assignee: 'ada' }` and +dotted names like `collect.note` are unaffected. If you were driving a batch- +approval `map` by resuming the map's own run id, resume the **item's** run +through its owning service instead (e.g. `client.approvals.approve`) — the map +advances itself when the item completes. diff --git a/content/docs/automation/flows.mdx b/content/docs/automation/flows.mdx index 8d5379f9b9..2d9f53de66 100644 --- a/content/docs/automation/flows.mdx +++ b/content/docs/automation/flows.mdx @@ -411,11 +411,28 @@ A resume of a `'service'` pause through the route answers **403** and changes nothing: the request stays pending and the run stays parked, so the real decision can still land. (Before this gate a raw resume walked the `approve` edge with no decision recorded, leaving the `sys_approval_request` row and the -run permanently disagreeing.) The gate follows a **subflow** pause down to the -child the signal would actually reach, so resuming the parent is no way around -it. Registering a pausing node of your own? Declare -`resumeAuthority: 'service'` on its descriptor when the decision to continue -belongs to your service rather than to whoever holds the run id. +run permanently disagreeing.) + +The gate follows a **linked-run** pause down to the child, so the run id a +launcher happens to hold is not a way around it — in both directions it reads +the item rather than the loop: + +- **`subflow`** — the signal is delegated to the suspended child, so the child's + node is where it lands; +- **`map`** — the signal is not delegated (the node re-runs to start the next + item), so continuing here would advance the batch **past** an item whose + decision is still open. Refused while that item is service-gated; the map + moves on when the item completes through its owning service. + +Two related rules on the same route: **resume `inputs` may not write the +engine's `$` namespace** (`$runId`, `$record`, `$flowName`, +`.$mapItemDone`, …) — those are the engine's own handoff variables, and +a caller who could set them could forge a map item's recorded result. Ordinary +author-declared variables are unaffected; a reserved name answers **400**. + +Registering a pausing node of your own? Declare `resumeAuthority: 'service'` on +its descriptor when the decision to continue belongs to your service rather +than to whoever holds the run id. ### Parallel approvals — one aggregating node, not two pauses @@ -475,8 +492,10 @@ the map waits for that item's decision, then moves to the next. The run holds a **single program counter** the whole time: only one item's approval is open at any moment; when it is decided the engine **re-enters** the -map node to start the next item. v1 is sequential and fail-fast (the first item -whose subflow fails fails the map). Concurrent fan-out (all items at once) is the +map node to start the next item. Resuming the map's own run id while an item is +still awaiting a service-gated decision is **refused** (403) — otherwise the +batch would step past that item as if it had been decided. v1 is sequential and +fail-fast (the first item whose subflow fails fails the map). Concurrent fan-out (all items at once) is the larger [ADR-0039](https://github.com/objectstack-ai/objectstack/blob/main/docs/adr/0039-token-scope-tree-execution.md) Track B work. Worked example: `showcase_release_signoff` → `showcase_one_task_signoff`. diff --git a/docs/adr/0019-approval-as-flow-node.md b/docs/adr/0019-approval-as-flow-node.md index 6a0981bdb7..c2a7d54a56 100644 --- a/docs/adr/0019-approval-as-flow-node.md +++ b/docs/adr/0019-approval-as-flow-node.md @@ -225,9 +225,11 @@ discriminates by **what the run is parked on**, not by the route: (`serviceResume`), on the tail of a decision it already authorized and recorded. The transport builds its signal out of a JSON body, and no JSON body can produce a symbol-keyed property, so the route cannot forge it; the gate does not need to trust the caller's claims about itself. -- **The gate follows the subflow chain.** A parent parked on a `subflow` node *delegates* the signal - to its suspended child (see the 2026-06-10 addendum), so the gate resolves the effective suspension - first and judges the node the signal actually lands on. Resuming the parent is not a way around it. +- **The gate follows the linked-run chain.** A parent parked on a `subflow` node *delegates* the + signal to its suspended child (see the 2026-06-10 addendum), so the gate resolves the effective + suspension first and judges the node the signal actually lands on. Resuming the parent is not a way + around it. *(As shipped this covered `subflow:` only; `map:` was added in #3853 — see the addendum + below.)* - **Refusal is an authorization answer.** `resume` returns `{ success: false, code: 'forbidden' }` and the route answers **403** — not a 200 carrying `success: false`, which reads as "your resume ran and the flow failed". Nothing is consumed: the request stays pending and the run stays parked, @@ -241,3 +243,37 @@ already-authorized call started. (signal flavor) placed by the flow author, which the type-keyed gate cannot distinguish from any other wait. A raw resume there still forces a resubmit without a `sys_approval_action` row. Filed as a follow-up — closing it needs a per-suspension owner claim rather than a node-type one. + +## Addendum (2026-07-28, #3853) — the chain walk covers `map:` too, and the transport owns the `$` namespace + +Two defects in the gate the addendum above describes, both demonstrated with a repro rather than +reasoned: + +**1. `map:` was not walked.** `resumeInternal` handles the two linked-run correlations *oppositely*: +a `subflow:` pause DELEGATES the signal down to the child, while a `map:` pause RE-RUNS the map node. +The gate's chain walk followed only the first, so a run parked on a `map` node was judged on `map` +itself — `resumeAuthority: 'any'` — and let through even while the item it was waiting on sat on an +`approval`. Since `$mapState.started` is advanced past the in-flight item *before* the suspend, and +the re-entry records a result only when `$mapItemDone` is set, an empty-body resume of the map parent +**skipped that item's approval outright**, orphaning its still-pending request; a later real decision +then bubbled into a parent already waiting on the *next* item, cascading the misalignment. The map +parent's run id is the one a launcher holds, which is what made it reachable. + +The walk now follows both prefixes. The unifying rule is that a linked-run pause is waiting on a +CHILD, so the child's node carries the authority — the gate reads *the item, not the loop*. The two +differ only in what continuing would mean: for `subflow` the signal lands on the child, for `map` it +advances past it. Both are refusals for the same reason. + +**2. The transport could write the engine's variable namespace.** `signal.variables` are applied as +BARE flow variables, so a caller could set the exact handoff keys `bubbleToParent` uses +(`.$mapItemDone` / `$mapItemOutput`) and have the map record a per-item result for a decision +nobody made — with the node id readable from `GET /automation/:name`. The same hole reaches `$runId`, +which `approval` / `wait` nodes use to correlate external state back to a run. `$` is the engine's +namespace (`$runId`, `$flowName`, `$flowLabel`, `$record`, `$error`, `$parentRunId`, `$parentMapNode`, +`$parentOutputVariable`, `.$mapState`); authors never write there. + +The **route** now refuses a resume whose `inputs` name anything in that namespace (`$…` or a `.$` +segment) with a 400. Deliberately at the transport and not in the engine: `bubbleToParent` legitimately +writes those keys in-process, and this is the same trust split the gate itself uses — strict at the +untrusted boundary, unrestricted for the code that already holds the authority. Refuse rather than +silently strip, so a mis-authored screen input fails at the door instead of many nodes downstream. diff --git a/packages/runtime/src/domains/automation.ts b/packages/runtime/src/domains/automation.ts index 3efef020da..a91a1f2a18 100644 --- a/packages/runtime/src/domains/automation.ts +++ b/packages/runtime/src/domains/automation.ts @@ -223,7 +223,25 @@ export async function handleAutomationRequest(deps: DomainHandlerDeps, path: str const b = (body && typeof body === 'object') ? body : {}; const inputs = (b.inputs ?? b.variables); const signal: any = {}; - if (inputs && typeof inputs === 'object') signal.variables = inputs; + if (inputs && typeof inputs === 'object') { + // #3853: `inputs` land as BARE flow variables, and `$` is the + // engine's own variable namespace (`$runId`, `$record`, + // `$flowName`, `.$mapItemDone`/`$mapItemOutput`/ + // `$mapState`, …). A caller who could write those could forge + // the map node's item handoff — recording a per-item result + // for an approval nobody made — or re-point `$runId`, which is + // how approval/wait nodes correlate external state back to a + // run. Author-declared variables never live in that namespace, + // so refuse rather than silently drop: a screen whose input is + // quietly discarded fails much further downstream. + const reserved = Object.keys(inputs).filter(k => k.startsWith('$') || k.includes('.$')); + if (reserved.length) { + return { handled: true, response: deps.error( + `Resume inputs may not set engine-internal variables (${reserved.join(', ')}) — ` + + `names starting with '$' (or containing '.$') are reserved by the flow engine`, 400) }; + } + signal.variables = inputs; + } if (b.output && typeof b.output === 'object') signal.output = b.output; if (typeof b.branchLabel === 'string') signal.branchLabel = b.branchLabel; const result = await automationService.resume(parts[2], signal); diff --git a/packages/runtime/src/http-dispatcher.test.ts b/packages/runtime/src/http-dispatcher.test.ts index 19076d5ceb..8d01e432e0 100644 --- a/packages/runtime/src/http-dispatcher.test.ts +++ b/packages/runtime/src/http-dispatcher.test.ts @@ -358,6 +358,36 @@ describe('HttpDispatcher', () => { expect(result.response?.body?.data?.success).toBe(false); }); + // #3853: `inputs` land as BARE flow variables, so a caller who could + // write the engine's `$` namespace could forge the `map` node's item + // handoff (recording a per-item result for an approval nobody made) or + // re-point `$runId`, which is how approval/wait nodes correlate. + it('should refuse resume inputs that write engine-internal variables', async () => { + for (const inputs of [ + { 'signoffs.$mapItemDone': true, 'signoffs.$mapItemOutput': { forged: true } }, + { $runId: 'someone_elses_run' }, + { $record: { id: 'other' } }, + ]) { + const result = await dispatcher.handleAutomation( + 'flow_a/runs/run_1/resume', 'POST', { inputs }, { request: {} }, + ); + expect(result.response?.status).toBe(400); + expect(result.response?.body?.error?.message).toMatch(/reserved by the flow engine/); + } + // Refused at the door — the engine is never asked. + expect(mockAutomationService.resume).not.toHaveBeenCalled(); + }); + + it('should still accept ordinary screen inputs alongside the reserved-name guard', async () => { + await dispatcher.handleAutomation( + 'flow_a/runs/run_1/resume', 'POST', + { inputs: { new_assignee: 'ada', 'collect.note': 'hi', price$: 3 } }, { request: {} }, + ); + expect(mockAutomationService.resume).toHaveBeenCalledWith('run_1', { + variables: { new_assignee: 'ada', 'collect.note': 'hi', price$: 3 }, + }); + }); + it('should return 501 when the automation service cannot resume', async () => { delete mockAutomationService.resume; const result = await dispatcher.handleAutomation( diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index c063f54a65..12f0c6ad1a 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -2001,12 +2001,13 @@ export class AutomationEngine implements IAutomationService { * the flow-runner that owns it. The service-side resume proves itself with * an in-process symbol the transport cannot mint from a JSON body. * - * Resolves the EFFECTIVE suspension first: a run parked on a `subflow` - * node delegates the signal to its suspended child, so the gate follows - * that chain and judges the node the signal actually lands on. Anything it - * cannot resolve (unknown run, missing flow, unregistered node type) is - * left to `resumeInternal`, which reports the machine-state error — the - * gate only ever speaks to authorization. + * Resolves the EFFECTIVE suspension first: a run parked on a `subflow` or + * `map` node is really waiting on a CHILD run, so the gate follows that + * chain and judges the node the signal lands on (subflow) or would advance + * past (map) — see {@link LINKED_RUN_PREFIXES}. Anything it cannot resolve + * (unknown run, missing flow, unregistered node type) is left to + * `resumeInternal`, which reports the machine-state error — the gate only + * ever speaks to authorization. */ private async refuseGatedResume(runId: string, signal?: ResumeSignal): Promise { const run = await this.resolveEffectiveSuspension(runId); @@ -2019,7 +2020,8 @@ export class AutomationEngine implements IAutomationService { // decision's tail, not a way around it. if (signal?.[RESUME_AUTHORITY_SERVICE]) return null; - const at = run.runId === runId ? `'${run.nodeId}'` : `'${run.nodeId}' (subflow run '${run.runId}')`; + const direct = run.runId === runId; + const at = direct ? `'${run.nodeId}'` : `'${run.nodeId}' (linked run '${run.runId}')`; this.logger.warn( `[automation] refused resume of run '${runId}': parked on ${nodeType} node ${at}, which is resumable ` + `only through its owning service (resumeAuthority: 'service')`, @@ -2027,9 +2029,12 @@ export class AutomationEngine implements IAutomationService { return { success: false, code: 'forbidden', - error: - `Run '${runId}' is paused at a '${nodeType}' node, which only its owning service may resume — ` + - `drive it through that service's API (e.g. an approval decision), not a raw resume`, + error: direct + ? `Run '${runId}' is paused at a '${nodeType}' node, which only its owning service may resume — ` + + `drive it through that service's API (e.g. an approval decision), not a raw resume` + : `Run '${runId}' is waiting on run '${run.runId}', which is paused at a '${nodeType}' node that ` + + `only its owning service may resume — resuming here would continue past a decision that has not ` + + `been made; drive it through that service's API instead`, }; } @@ -2063,20 +2068,40 @@ export class AutomationEngine implements IAutomationService { private static readonly MAX_ALIAS_HOPS = 4; /** - * Follow the subflow delegation chain from `runId` to the suspension a - * resume signal would actually land on. A run paused at a `subflow` node - * (correlation `subflow:`) forwards the signal down, so the - * deepest reachable suspension — not the id the caller holds — is what a - * resume really continues. Returns `null` when nothing is suspended under - * that id; stops at the last resolvable link when a child row is gone - * (that run is where `resumeInternal` will continue). + * Linked-run correlation prefixes the gate walks (#3853). Both park a + * parent on a child run, so in both the pending work — and therefore the + * authority that governs continuing it — belongs to the CHILD, even though + * `resumeInternal` handles the two oppositely: + * + * - `subflow:` — the signal is DELEGATED down to the child, so the child's + * node is literally where it lands. + * - `map:` — the signal is not delegated; the `map` node RE-RUNS, and since + * `$mapState.started` was advanced past the in-flight item before the + * suspend, continuing here advances the map *past* the item whose child + * is still parked. Judging the parent's own `map` node (always + * `resumeAuthority: 'any'`) let a raw resume skip a pending approval in a + * batch-approval flow — the gate has to read the item, not the loop. + */ + private static readonly LINKED_RUN_PREFIXES = ['subflow:', 'map:'] as const; + + /** + * Follow the linked-run chain from `runId` to the suspension whose node + * actually governs a resume — see {@link LINKED_RUN_PREFIXES}. The deepest + * reachable suspension, not the id the caller holds, is what a resume + * really continues (or skips past). Returns `null` when nothing is + * suspended under that id; stops at the last resolvable link when a child + * row is gone (that run is where `resumeInternal` will continue). */ private async resolveEffectiveSuspension(runId: string): Promise { const seen = new Set(); let run = await this.loadSuspendedRun(runId); for (let depth = 0; run && depth < AutomationEngine.MAX_SUSPENSION_CHAIN_DEPTH; depth++) { - if (typeof run.correlation !== 'string' || !run.correlation.startsWith('subflow:')) return run; - const childRunId = run.correlation.slice('subflow:'.length); + const correlation = run.correlation; + const prefix = typeof correlation === 'string' + ? AutomationEngine.LINKED_RUN_PREFIXES.find(p => correlation.startsWith(p)) + : undefined; + if (!prefix) return run; + const childRunId = correlation!.slice(prefix.length); if (!childRunId || seen.has(childRunId)) return run; seen.add(childRunId); const child = await this.loadSuspendedRun(childRunId); diff --git a/packages/services/service-automation/src/resume-authority-gate.test.ts b/packages/services/service-automation/src/resume-authority-gate.test.ts index 9774e43074..a2d35417db 100644 --- a/packages/services/service-automation/src/resume-authority-gate.test.ts +++ b/packages/services/service-automation/src/resume-authority-gate.test.ts @@ -23,6 +23,7 @@ import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; import { AutomationEngine } from './engine.js'; import { InMemorySuspendedRunStore } from './suspended-run-store.js'; import { registerSubflowNode } from './builtin/subflow-node.js'; +import { registerMapNode } from './builtin/map-node.js'; function silentLogger(): any { return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } }; @@ -269,4 +270,85 @@ describe('resume authorization gate (#3801)', () => { expect(downstream).toEqual(['after']); }); }); + + // ── map chains: the signal would ADVANCE PAST the child (#3853) ─────── + // + // A `map` pause is the batch-approval shape, and unlike `subflow` the signal + // is NOT delegated — the map node re-runs, and `$mapState.started` was + // already advanced past the in-flight item before the suspend. So resuming + // the parent skips the item whose decision is still open. The run id a + // launcher holds is the map parent's, which made it the reachable one. + + describe('through a map pause', () => { + /** Parent whose `map` node runs a per-item child that parks on `pauseType`. */ + function registerBatch(e: AutomationEngine, pauseType: string): void { + registerMapNode(e, pluginCtx()); + e.registerFlow('one_item', pauseFlow('one_item', pauseType) as never); + e.registerFlow('batch_flow', { + name: 'batch_flow', + label: 'batch_flow', + type: 'autolaunched', + variables: [{ name: 'items', type: 'list', isInput: true }], + nodes: [ + { id: 'bs', type: 'start', label: 'Start' }, + { + id: 'signoffs', type: 'map', label: 'Sign off each', + config: { collection: '{items}', iteratorVariable: 'task', flowName: 'one_item', outputVariable: 'results' }, + }, + { id: 'be', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'b1', source: 'bs', target: 'signoffs' }, + { id: 'b2', source: 'signoffs', target: 'be' }, + ], + } as never); + } + + const launch = (e: AutomationEngine) => + e.execute('batch_flow', { params: { items: [{ id: 't1' }, { id: 't2' }] } }); + const mapState = (e: AutomationEngine, runId: string) => + (e as any).suspendedRuns.get(runId)?.variables?.['signoffs.$mapState']; + + it('refuses a resume of the MAP PARENT while an item is parked on a gated node', async () => { + registerBatch(engine, 'gated_pause'); + const paused = await launch(engine); + expect(mapState(engine, paused.runId!)).toEqual({ started: 1, results: [] }); + + // Exactly what the route builds from an empty body. + const refused = await engine.resume(paused.runId!, {}); + + expect(refused.code).toBe('forbidden'); + expect(refused.error).toMatch(/waiting on run/); + // The map did NOT advance past item 1, and nothing was consumed. + expect(mapState(engine, paused.runId!)).toEqual({ started: 1, results: [] }); + expect(engine.listSuspendedRuns()).toHaveLength(2); // parent + item 1 + expect(downstream).toEqual([]); + }); + + it('lets the item complete through its owning service, which advances the map', async () => { + registerBatch(engine, 'gated_pause'); + await launch(engine); + const item1 = engine.listSuspendedRuns().find(r => r.flowName === 'one_item')!; + + // The service decides item 1 — the child bubbles up and the map moves on. + const ok = await engine.resume(item1.runId, { + branchLabel: 'approve', [RESUME_AUTHORITY_SERVICE]: true, + }); + + expect(ok.success).toBe(true); + const parent = engine.listSuspendedRuns().find(r => r.flowName === 'batch_flow')!; + expect(mapState(engine, parent.runId).started).toBe(2); // item 2 now in flight + expect(downstream).toEqual(['after']); // item 1 really ran through + }); + + it('leaves a map over ungated items resumable', async () => { + registerBatch(engine, 'open_pause'); + const paused = await launch(engine); + + const resumed = await engine.resume(paused.runId!, {}); + + expect(resumed.code).toBeUndefined(); + expect(resumed.success).toBe(true); + }); + }); });