From bc779214421177c88aa3ba023a2b5d9b395e6781 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Jul 2026 07:34:38 +0000 Subject: [PATCH 1/2] fix(automation,approvals): gate the generic run-resume route on the suspended node (#3801) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `POST /api/v1/automation/:name/runs/:runId/resume` forwarded a caller-supplied `{ inputs, output, branchLabel }` straight into `AutomationEngine.resume`, and `resumeInternal` validated machine state only — the concurrent-resume latch, the run exists, the flow exists, the suspended node still exists. Nothing asked who. Approval nodes suspend and resume through exactly that mechanism, so a resume carrying `branchLabel: 'approve'` walked the approve edge with no approver check, no `sys_approval_action` row and no status mirror — leaving the request row and the run permanently disagreeing. The only thing between the route and the approvals rules was convention, spelled out in a showcase comment. Removing the route is not the fix: it is load-bearing for screen flows. So the gate keys on what the run is parked on: - `ActionDescriptor.resumeAuthority` ('any' | 'service', default 'any') — a pausing node declares who may continue it; `approval` declares 'service'. - A suspension records the node type that produced it (`SuspendedRun.nodeType` / `sys_automation_run.node_type`), captured at suspend time so a flow republished mid-pause cannot re-type the node out from under the gate; older rows fall back to the flow definition. A deprecated ADR-0018 alias resolves to its canonical type, so renaming is not an escape hatch. - The engine refuses a 'service' suspension unless the signal carries `RESUME_AUTHORITY_SERVICE` — a symbol, so a JSON body can never mint it. `ApprovalService` stamps it in one place, on the tail of a decision it has already authorized and recorded. - The gate follows a subflow pause down to the child the signal would actually reach, so resuming the parent is no way around it. - Refusal returns `{ success: false, code: 'forbidden' }` and the route answers 403. Nothing is consumed: the request stays pending and the run stays parked, so the real decision still lands. Engine-internal continuations (subflow delegation/up-bubble, `map` re-entry, wait-timer wake) go through the private `resumeInternal` and are not re-gated. `screen` and `wait` pauses are unchanged. Known adjacent gap, deliberately out of scope: ADR-0044's revise window parks the run on an ordinary author-placed `wait` node, which a type-keyed gate cannot tell from any other wait. Recorded in the ADR addendum and filed separately. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013gvN32u1EiuvY9uQEMJiMR --- .../automation-resume-authority-gate.md | 57 ++++ content/docs/automation/flows.mdx | 28 +- .../references/automation/node-executor.mdx | 1 + docs/adr/0019-approval-as-flow-node.md | 46 +++ .../src/automation/flows/index.ts | 6 +- packages/client/src/index.ts | 14 +- .../src/approval-node.test.ts | 40 +++ .../plugin-approvals/src/approval-node.ts | 9 + .../plugin-approvals/src/approval-service.ts | 46 ++- packages/runtime/src/domains/automation.ts | 12 + packages/runtime/src/http-dispatcher.test.ts | 30 ++ packages/runtime/src/route-ledger.ts | 3 +- .../services/service-automation/src/engine.ts | 163 ++++++++++- .../src/resume-authority-gate.test.ts | 272 ++++++++++++++++++ .../src/suspended-run-store.ts | 4 + .../src/sys-automation-run.object.ts | 8 + packages/spec/api-surface.json | 1 + .../spec/src/automation/node-executor.test.ts | 22 ++ .../spec/src/automation/node-executor.zod.ts | 25 ++ .../spec/src/contracts/automation-service.ts | 45 +++ skills/objectstack-automation/SKILL.md | 5 +- 21 files changed, 807 insertions(+), 30 deletions(-) create mode 100644 .changeset/automation-resume-authority-gate.md create mode 100644 packages/services/service-automation/src/resume-authority-gate.test.ts diff --git a/.changeset/automation-resume-authority-gate.md b/.changeset/automation-resume-authority-gate.md new file mode 100644 index 0000000000..6dfb2d2e5f --- /dev/null +++ b/.changeset/automation-resume-authority-gate.md @@ -0,0 +1,57 @@ +--- +"@objectstack/spec": minor +"@objectstack/service-automation": minor +"@objectstack/plugin-approvals": minor +"@objectstack/runtime": minor +"@objectstack/client": patch +--- + +fix(automation,approvals): the run-resume route is gated by the node the run is parked on (#3801) + +`POST /api/v1/automation/:name/runs/:runId/resume` forwarded a caller-supplied +`{ inputs, output, branchLabel }` straight into `AutomationEngine.resume`, and +`resumeInternal` validated **machine state only** — the concurrent-resume latch, +the run exists, the flow exists, the suspended node still exists. Nothing asked +*who was calling*. + +Approval nodes suspend and resume through exactly that mechanism. So a resume +carrying `branchLabel: 'approve'` walked the approve edge with **no approver +check, no `sys_approval_action` row and no status mirror** — the +`sys_approval_request` row and the run then disagreed permanently. The only +thing standing between the route and the approvals rules was convention; the +showcase spelled it out in a comment ("decide via the approvals API, never a raw +engine `resume`"), and a comment in an example is not an access control. + +Removing the route was not the fix: it is load-bearing for **screen flows** — +the UI flow-runner posts `{ inputs }` there to advance a paused `screen` node. +The gate therefore keys on **what the run is parked on**: + +- `ActionDescriptor.resumeAuthority` (`'any'` | `'service'`, default `'any'`) — + a pausing node declares who may continue it. `approval` declares `'service'`. +- The engine refuses a `'service'` suspension unless the signal carries + `RESUME_AUTHORITY_SERVICE` (`@objectstack/spec/contracts`), a **symbol** the + owning service stamps in-process — a JSON body can never produce one, so the + transport cannot forge it. `ApprovalService` stamps it on the tail of a + decision it has already authorized and recorded. +- The gate follows a **subflow** pause down to the child the signal would + actually reach, so resuming the parent is not a way around it. +- Refusal returns `{ success: false, code: 'forbidden' }` and the route answers + **403**. Nothing is consumed — the request stays pending and the run stays + parked, so the real decision still lands. + +`screen` and `wait` pauses are unchanged, as is every path that already went +through the approvals API. What changes for consumers: + +- **FROM:** finishing an approval with + `client.automation.resume(flow, runId, { branchLabel: 'approve' })` + **TO:** `client.approvals.approve(requestId, …)` (or `.reject` / `.recall`). + The old call now answers 403 and changes nothing. +- Registering your own pausing node whose continuation belongs to a service + rather than to whoever holds the run id? Declare `resumeAuthority: 'service'` + on its descriptor and stamp `RESUME_AUTHORITY_SERVICE` on the signal from that + service. + +A suspension now records the node type that produced it +(`SuspendedRun.nodeType` / `sys_automation_run.node_type`), captured at suspend +time so a flow republished mid-pause cannot re-type the node out from under the +gate; rows written before this fall back to the flow definition. diff --git a/content/docs/automation/flows.mdx b/content/docs/automation/flows.mdx index e91ce51449..8d5379f9b9 100644 --- a/content/docs/automation/flows.mdx +++ b/content/docs/automation/flows.mdx @@ -389,11 +389,34 @@ POST /api/v1/automation/{flow}/runs/{runId}/resume | Pausing node | Suspends until… | Resumed by | | :--- | :--- | :--- | -| `approval` | a human decision | the approvals service (`POST /api/v1/approvals/requests/:id/approve\|reject`) — resumes down the matching `approve` / `reject` edge. **Always decide through the approvals API**, never by calling `resume` directly: a direct resume strands the pending `sys_approval_request`, so the record stays locked and later approvals on the same record hit `DUPLICATE_REQUEST`. | +| `approval` | a human decision | the approvals service (`POST /api/v1/approvals/requests/:id/approve\|reject`) — resumes down the matching `approve` / `reject` edge. **Decide through the approvals API**; the resume route above **refuses** an approval pause outright (see below). | | `screen` | a user submits the form | the UI runner posting the collected `inputs`; a `paused` response carrying the next `screen` chains multi-step wizards under one stable `runId` | | `wait` (timer) | an ISO-8601 duration elapses | **automatically** — a one-shot job resumes the run; after a cold boot the engine re-arms pending timers from the durable store (overdue timers resume immediately) | | `wait` (signal) | a named external event | any caller invoking `resume(runId)` | +### Who may resume — the gate is the suspended node + +The resume route is generic, so **the node the run is parked on** decides +whether a raw resume is a legitimate continuation. Every action descriptor +carries a `resumeAuthority`: + +- **`'any'`** (the default — `screen`, `wait`, your own pausing nodes): the + caller supplies the continuation and the route is the intended door. +- **`'service'`** (`approval`): continuing is a *side effect* of a decision + that some service must authorize and record first, so only that service may + drive it. `ApprovalService.decide` enforces the approver slate, writes the + `sys_approval_action` row, mirrors the status field — **then** resumes. + +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. + ### Parallel approvals — one aggregating node, not two pauses "Finance **and** legal must both sign off, concurrently" is **one `approval` @@ -469,7 +492,8 @@ the chain resolves from either end: `${nodeId}.output` / `outputVariable` mapping as a synchronous subflow; - resuming the **parent** run id (what a UI holds from the original launch) **delegates** down to the suspended child — multi-screen child wizards keep - the parent run id stable. + the parent run id stable. The resume gate delegates with it: if the child is + parked on an `approval`, resuming the parent is refused too. A child that fails terminally after the pause fails every waiting ancestor, so no run is stranded as resumable-forever. See the worked example pair in the diff --git a/content/docs/references/automation/node-executor.mdx b/content/docs/references/automation/node-executor.mdx index a501e949d1..682a955249 100644 --- a/content/docs/references/automation/node-executor.mdx +++ b/content/docs/references/automation/node-executor.mdx @@ -76,6 +76,7 @@ Canonical cross-paradigm action/node descriptor (ADR-0018) | **supportsRetry** | `boolean` | ✅ | Supports retry on failure | | **needsOutbox** | `boolean` | ✅ | Dispatch via service-messaging outbox (retry/idempotency/dead-letter) | | **isAsync** | `boolean` | ✅ | Suspends the flow awaiting an external reply | +| **resumeAuthority** | `Enum<'any' \| 'service'>` | ✅ | Who may resume a run this node suspended: 'any' (the generic resume route) or 'service' (only the owning service, e.g. approvals) | | **maturity** | `Enum<'ga' \| 'beta' \| 'reserved'>` | ✅ | Runtime maturity: ga (shipped), beta, or reserved (contract only — designers grey this out) | | **source** | `Enum<'builtin' \| 'plugin'>` | ✅ | builtin = platform baseline; plugin = third-party contributed | | **deprecated** | `boolean` | ✅ | Deprecated alias kept for back-compat | diff --git a/docs/adr/0019-approval-as-flow-node.md b/docs/adr/0019-approval-as-flow-node.md index c854890ade..6a0981bdb7 100644 --- a/docs/adr/0019-approval-as-flow-node.md +++ b/docs/adr/0019-approval-as-flow-node.md @@ -195,3 +195,49 @@ not catch a *post-pause* child failure (the parent run fails terminally instead) does not count across a suspension; a crash exactly between child completion and the parent bubble leaves the parent paused — an operator can compensate with a manual `resume(parentRunId, { output })` (outbox-grade exactly-once chaining is future work). + +## Addendum (2026-07-28, #3801) — the resume seam is a trust boundary: `resumeAuthority` + +Collapsing approval onto the flow engine made `AutomationEngine.resume` the **one** way any pause +continues — including an approval decision. That is the right execution model, but it left the +authorization model behind: `POST /api/v1/automation/:name/runs/:runId/resume` forwards a +caller-supplied `{ inputs, output, branchLabel }` straight through, and `resumeInternal` validated +**machine state only** (the concurrent-resume latch, the run exists, the flow exists, the suspended +node still exists). Nothing asked *who*. A raw resume with `branchLabel: 'approve'` therefore walked +the approve edge with **no approver check, no `sys_approval_action` row and no status mirror** — +leaving the `sys_approval_request` row and the run permanently disagreeing. The only thing standing +between the route and the approvals rules was convention, spelled out in a showcase comment. A +comment in an example is not an access control. + +Deleting the route was never an option: it is load-bearing for **screen flows** — the UI flow-runner +posts `{ inputs }` to advance a paused `screen` node, which is its whole reason to exist. So the gate +discriminates by **what the run is parked on**, not by the route: + +- **`ActionDescriptor.resumeAuthority`** (ADR-0018 registry, `'any'` | `'service'`, default `'any'`). + A pausing node declares who may continue it. `approval` declares `'service'`; `screen` / `wait` + keep the default and behave exactly as before. +- **The suspension carries its own node type.** `SuspendedRun.nodeType` (column + `sys_automation_run.node_type`) is captured at suspend time rather than re-derived from the live + flow, so a flow republished mid-pause cannot re-type the node out from under the gate. Rows + written before this shipped fall back to the flow definition. +- **The marker is a symbol.** `RESUME_AUTHORITY_SERVICE` (`@objectstack/spec/contracts`) is stamped + on the `ResumeSignal` by the owning service — `ApprovalService` does it in one place + (`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. +- **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, + so the real decision still lands. + +Engine-internal continuations (subflow delegation and up-bubble, `map` re-entry, the wait-timer wake) +go through the private `resumeInternal` and are **not** re-gated — they continue work an +already-authorized call started. + +**Known gap, tracked separately:** ADR-0044's revise window parks the run on an ordinary `wait` node +(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. diff --git a/examples/app-showcase/src/automation/flows/index.ts b/examples/app-showcase/src/automation/flows/index.ts index 8a1fd07a38..7d5a65615d 100644 --- a/examples/app-showcase/src/automation/flows/index.ts +++ b/examples/app-showcase/src/automation/flows/index.ts @@ -987,8 +987,10 @@ export const ResilientSyncFlow = defineFlow({ * rejection resumes down `reject`. One node, one suspend/resume, no token tree — * the multi-instance pattern Camunda and Step Functions use for exactly this. * - * Decide via the approvals API (never a raw engine `resume`): - * POST /api/v1/automation/showcase_invoice_signoff/runs/{runId}/... ← no + * Decide via the approvals API — a raw engine `resume` is refused, not merely + * discouraged: the `approval` node declares `resumeAuthority: 'service'`, so + * the generic run-resume route answers 403 for a run parked on one (#3801). + * POST /api/v1/automation/showcase_invoice_signoff/runs/{runId}/resume ← 403 * POST /api/v1/approvals/requests/{id}/approve { actorId: 'position:finance' } * POST /api/v1/approvals/requests/{id}/approve { actorId: 'position:legal' } ← now it continues */ diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index a6f0e9c904..6d33585942 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -2720,7 +2720,7 @@ export class ObjectStackClient { return this.unwrapResponse(res) as Promise; }, /** - * Resume a run suspended at a `screen` (or `approval`) node — the + * Resume a run suspended at a `screen` (or `wait`) node — the * screen-flow runtime's second half (ADR-0019 durable pause). * * `execute()` returns `{ status: 'paused', runId, screen }` when a flow @@ -2729,6 +2729,13 @@ export class ObjectStackClient { * NEXT `{ status: 'paused', screen }` of a multi-step wizard or the * terminal `AutomationResult`. Without this method a paused run can only * be finished by hand-rolling the HTTP call (#3528). + * + * **Not the door for an approval (#3801).** A run parked on an + * `approval` node — directly, or as the child of a `subflow` pause — is + * resumable only through the approvals API + * ({@link ObjectStackClient.approvals}: `approve` / `reject` / `recall`), + * which authorizes the decision and records it first. This call answers + * **403** for one and changes nothing. */ resume: async ( flowName: string, @@ -4423,9 +4430,10 @@ export class ScopedProjectClient { return this.parent._unwrap(res); }, /** - * Resume a run suspended at a `screen` / `approval` node with the collected + * Resume a run suspended at a `screen` / `wait` node with the collected * input (ADR-0019 durable pause). Mirrors the unscoped - * `client.automation.resume`. + * `client.automation.resume` — including its refusal (403) to resume an + * `approval` pause, which belongs to the approvals API (#3801). */ resume: async ( flowName: string, diff --git a/packages/plugins/plugin-approvals/src/approval-node.test.ts b/packages/plugins/plugin-approvals/src/approval-node.test.ts index bf1d431e3e..3fdd68e792 100644 --- a/packages/plugins/plugin-approvals/src/approval-node.test.ts +++ b/packages/plugins/plugin-approvals/src/approval-node.test.ts @@ -154,6 +154,46 @@ describe('Approval node bridge (ADR-0019)', () => { expect(paused.runId).toBeDefined(); }); + // ── resume authorization gate (#3801) ─────────────────────────────── + // + // `POST /automation/:name/runs/:runId/resume` reaches + // `AutomationEngine.resume` with a caller-supplied signal. Before the gate, + // the only thing between that route and the approvals rules was convention — + // a comment in the showcase. These pin the enforcement. + + it('declares the approval node resumable only by its owning service', () => { + const approval = automation.getActionDescriptors().find(d => d.type === 'approval'); + expect(approval!.resumeAuthority).toBe('service'); + }); + + it('refuses a raw engine resume of an approval pause, leaving the request untouched', async () => { + registerDecisionFlow(automation, [{ type: 'user', value: 'u1' }]); + const paused = await automation.execute('deal_approval', { + object: 'crm_deal', record: { id: 'd1' }, userId: 'submitter', + }); + const request = (await fake.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + + // Exactly the signal the resume route builds from `{ branchLabel }`. + const refused = await automation.resume(paused.runId!, { + branchLabel: 'approve', output: { decision: 'approve' }, + }); + + expect(refused).toMatchObject({ success: false, code: 'forbidden' }); + // The approve branch did NOT run… + expect(marks).toHaveLength(0); + // …the request is still pending, with no decision recorded… + const stillPending = (await fake.find('sys_approval_request', { where: { id: request.id } }))[0]; + expect(stillPending.status).toBe('pending'); + const actions = await fake.find('sys_approval_action', { where: { request_id: request.id } }); + expect(actions.map((a: any) => a.action)).toEqual(['submit']); + // …and the run is still parked, so the real decision can still land. + expect(automation.listSuspendedRuns()).toHaveLength(1); + + const out = await service.decide(request.id, { decision: 'approve', actorId: 'u1' }, SYSTEM_CTX); + expect(out).toMatchObject({ finalized: true, resumed: true }); + expect(marks).toEqual(['on_approved']); + }); + it('resumes down the reject branch on rejection', async () => { registerDecisionFlow(automation, [{ type: 'user', value: 'u1' }]); await automation.execute('deal_approval', { object: 'crm_deal', record: { id: 'd1' } }); diff --git a/packages/plugins/plugin-approvals/src/approval-node.ts b/packages/plugins/plugin-approvals/src/approval-node.ts index bcfab04a27..728fed6c71 100644 --- a/packages/plugins/plugin-approvals/src/approval-node.ts +++ b/packages/plugins/plugin-approvals/src/approval-node.ts @@ -104,6 +104,15 @@ export function registerApprovalNode( // Human decision: the run suspends here awaiting an external reply. supportsPause: true, isAsync: true, + // #3801: this pause is NOT resumable through the generic run-resume + // route. Continuing an approval is a side effect of a DECISION, and the + // decision is the thing that must be authorized (the approver slate), + // recorded (`sys_approval_action`) and mirrored (the status field) — + // all of which lives in `ApprovalService.decide`. The engine now refuses + // any resume of an approval suspension that does not carry the service's + // in-process marker, so "decide via the approvals API, never a raw engine + // resume" is enforced rather than merely documented. + resumeAuthority: 'service', // Publish the node's config contract (ADR-0018 §configSchema) so the // Studio flow designer renders the Approval property form from the engine // rather than a hardcoded client form — the engine owns the shape. diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index 0aac090636..bbe2cb9df2 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -31,6 +31,7 @@ import type { ApprovalStatus, SharingExecutionContext, } from '@objectstack/spec/contracts'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; import { isFileIdToken } from '@objectstack/spec/data'; import { isGrantActive } from '@objectstack/core'; @@ -64,7 +65,18 @@ export interface ApprovalClock { now(): Date } * plugin when an automation engine is present (see `approval-node.ts`). */ export interface ApprovalResumeSurface { - resume?(runId: string, signal?: { output?: Record; branchLabel?: string }): Promise; + resume?(runId: string, signal?: { + output?: Record; + branchLabel?: string; + /** + * #3801: the engine refuses a resume of an `approval` suspension unless + * the signal carries this marker — the proof that the resume is the tail + * of a decision THIS service already authorized and recorded, not a raw + * `POST …/runs/:runId/resume` around it. Every resume below stamps it via + * {@link ApprovalService.serviceResume}. + */ + [RESUME_AUTHORITY_SERVICE]?: true; + }): Promise; /** Flow definition lookup, used to derive step-progress display data. */ getFlow?(name: string): Promise; /** @@ -1590,6 +1602,28 @@ export class ApprovalService implements IApprovalService { return { request: fresh!, runId, nodeId, finalized: true, decision: input.decision, outputs: mergedOutputs }; } + /** + * Continue the owning flow run after an outcome this service has already + * authorized and written down (#3801). + * + * The `approval` node declares `resumeAuthority: 'service'`, so the engine + * refuses any resume of an approval suspension that does not carry + * {@link RESUME_AUTHORITY_SERVICE}. Every approvals-side resume goes through + * here so the marker is stamped in ONE place — a new outcome path cannot + * quietly ship a resume that the gate then rejects at runtime, and nothing + * in this file hands the marker to a caller-supplied signal. + * + * Callers still guard on `typeof this.automation?.resume === 'function'` + * (approvals runs fine with no automation attached) and keep their own + * try/catch, because what a failed resume means differs per path. + */ + private async serviceResume( + runId: string, + signal: { output?: Record; branchLabel?: string }, + ): Promise { + await this.automation!.resume!(runId, { ...signal, [RESUME_AUTHORITY_SERVICE]: true }); + } + /** * Public contract entrypoint (ADR-0019). Records a decision on a node-driven * request via {@link ApprovalService.decideNode} and, when it finalizes, @@ -1608,7 +1642,7 @@ export class ApprovalService implements IApprovalService { ? APPROVAL_BRANCH_LABELS.approve : APPROVAL_BRANCH_LABELS.reject; try { - await this.automation.resume(result.runId, { + await this.serviceResume(result.runId, { branchLabel, // #3447 P2: accepted decision outputs ride the resume envelope and // land as `.` flow variables — a later approval node's @@ -1712,7 +1746,7 @@ export class ApprovalService implements IApprovalService { } } else if (runId && typeof this.automation?.resume === 'function') { try { - await this.automation.resume(runId, { + await this.serviceResume(runId, { branchLabel: APPROVAL_BRANCH_LABELS.reject, output: { decision: 'recall', requestId }, }); @@ -1802,7 +1836,7 @@ export class ApprovalService implements IApprovalService { let resumed = false; if (runId && typeof this.automation?.resume === 'function') { try { - await this.automation.resume(runId, { + await this.serviceResume(runId, { branchLabel: APPROVAL_BRANCH_LABELS.reject, output: { decision: 'reject', autoRejected: true, requestId }, }); @@ -1844,7 +1878,7 @@ export class ApprovalService implements IApprovalService { let resumed = false; if (runId && typeof this.automation?.resume === 'function') { try { - await this.automation.resume(runId, { + await this.serviceResume(runId, { branchLabel: APPROVAL_BRANCH_LABELS.revise, output: { decision: 'revise', requestId }, }); @@ -1931,7 +1965,7 @@ export class ApprovalService implements IApprovalService { let resumed = false; if (runId && typeof this.automation?.resume === 'function') { try { - await this.automation.resume(runId, { + await this.serviceResume(runId, { branchLabel: APPROVAL_BRANCH_LABELS.resubmit, output: { resubmitted: true, requestId }, }); diff --git a/packages/runtime/src/domains/automation.ts b/packages/runtime/src/domains/automation.ts index 9322530c07..3efef020da 100644 --- a/packages/runtime/src/domains/automation.ts +++ b/packages/runtime/src/domains/automation.ts @@ -209,6 +209,15 @@ export async function handleAutomationRequest(deps: DomainHandlerDeps, path: str // values, applied as bare flow variables; `output`/`branchLabel` also // forwarded for approval-style resumes. Returns the next paused // `{ screen }` (multi-screen) or the completed result. + // + // The signal is built key-by-key from the JSON body on purpose (#3801): + // the engine gates a suspension whose node declares + // `resumeAuthority: 'service'` — an `approval` pause, resumable only via + // `ApprovalService`, which records the decision and enforces the slate — + // on a SYMBOL-keyed marker. Assembling the signal field-wise (never + // spreading the body) keeps that unforgeable even if a caller invents + // extra keys; a refused resume comes back `code: 'forbidden'` and is + // answered 403 rather than a 200 carrying `success: false`. if (parts[1] === 'runs' && parts[2] && parts[3] === 'resume' && m === 'POST') { if (typeof automationService.resume === 'function') { const b = (body && typeof body === 'object') ? body : {}; @@ -218,6 +227,9 @@ export async function handleAutomationRequest(deps: DomainHandlerDeps, path: str 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); + if (result?.success === false && result.code === 'forbidden') { + return { handled: true, response: deps.error(result.error ?? 'Resume forbidden', 403) }; + } return { handled: true, response: deps.success(result) }; } return { handled: true, response: deps.error('Resume not supported', 501) }; diff --git a/packages/runtime/src/http-dispatcher.test.ts b/packages/runtime/src/http-dispatcher.test.ts index 3b5c5b9678..5b951ceefb 100644 --- a/packages/runtime/src/http-dispatcher.test.ts +++ b/packages/runtime/src/http-dispatcher.test.ts @@ -328,6 +328,36 @@ describe('HttpDispatcher', () => { expect(result.response?.body?.data?.screen?.nodeId).toBe('step2'); }); + // #3801: a run parked on a service-gated node (an `approval` pause, + // resumable only through ApprovalService) comes back `forbidden` from + // the engine. That is an AUTHORIZATION answer and must read as one — + // a 200 carrying `success: false` reads as "your resume ran and the + // flow failed", which is the opposite of what happened. + it('should answer 403 when the engine refuses the resume as service-gated', async () => { + mockAutomationService.resume.mockResolvedValue({ + success: false, code: 'forbidden', + error: "Run 'run_1' is paused at an 'approval' node, which only its owning service may resume", + }); + const result = await dispatcher.handleAutomation( + 'flow_a/runs/run_1/resume', 'POST', { branchLabel: 'approve' }, { request: {} }, + ); + expect(result.handled).toBe(true); + expect(result.response?.status).toBe(403); + expect(result.response?.body?.error?.message ?? result.response?.body?.message) + .toMatch(/only its owning service may resume/); + }); + + // A run that resumed and then FAILED is not an authorization answer — + // it keeps the ordinary success-envelope shape. + it('should not 403 an ordinary failed resume', async () => { + mockAutomationService.resume.mockResolvedValue({ success: false, error: 'node blew up' }); + const result = await dispatcher.handleAutomation( + 'flow_a/runs/run_1/resume', 'POST', { inputs: {} }, { request: {} }, + ); + expect(result.response?.status).not.toBe(403); + expect(result.response?.body?.data?.success).toBe(false); + }); + it('should return 501 when the automation service cannot resume', async () => { delete mockAutomationService.resume; const result = await dispatcher.handleAutomation( diff --git a/packages/runtime/src/route-ledger.ts b/packages/runtime/src/route-ledger.ts index 62b77b4f05..24958219a6 100644 --- a/packages/runtime/src/route-ledger.ts +++ b/packages/runtime/src/route-ledger.ts @@ -178,7 +178,8 @@ export const ROUTE_LEDGER: readonly RouteLedgerEntry[] = [ { route: 'GET /automation/_status', domain: '/automation', disposition: 'sdk', client: 'automation.getRuntimeStatus' }, { route: 'POST /automation/:name/trigger', domain: '/automation', disposition: 'sdk', client: 'automation.execute' }, { route: 'POST /automation/:name/toggle', domain: '/automation', disposition: 'sdk', client: 'automation.toggle' }, - { route: 'POST /automation/:name/runs/:runId/resume', domain: '/automation', disposition: 'sdk', client: 'automation.resume' }, + { route: 'POST /automation/:name/runs/:runId/resume', domain: '/automation', disposition: 'sdk', client: 'automation.resume', + note: "generic, so the SUSPENDED NODE gates it (#3801): a pause whose descriptor declares resumeAuthority:'service' — today `approval` — answers 403 here and continues only through its owning service (ApprovalService.decide), which authorizes and records the decision first. Screen/wait pauses are unaffected; this route is the screen-flow runner's door" }, { route: 'GET /automation/:name/runs/:runId/screen', domain: '/automation', disposition: 'sdk', client: 'automation.getScreen' }, { route: 'GET /automation/:name/runs/:runId', domain: '/automation', disposition: 'sdk', client: 'automation.getRun' }, { route: 'GET /automation/:name/runs', domain: '/automation', disposition: 'sdk', client: 'automation.listRuns' }, diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index e3ff5e8fc5..c063f54a65 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -3,6 +3,7 @@ import type { FlowParsed, FlowNodeParsed, FlowEdgeParsed } from '@objectstack/spec/automation'; import type { ExecutionLog, ActionDescriptor } from '@objectstack/spec/automation'; import type { AutomationContext, AutomationResult, ResumeSignal, IAutomationService, ScreenSpec } from '@objectstack/spec/contracts'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; import type { Logger } from '@objectstack/spec/contracts'; import { FlowSchema, FLOW_STRUCTURAL_NODE_TYPES, validateControlFlow, findRegionEntry, defineActionDescriptor } from '@objectstack/spec/automation'; import { applyConversionsToFlow } from '@objectstack/spec'; @@ -530,7 +531,13 @@ interface ExecutionLogEntry { */ class FlowSuspendSignal { readonly __flowSuspend = true as const; - constructor(readonly nodeId: string, readonly correlation?: string, readonly screen?: ScreenSpec) {} + constructor( + readonly nodeId: string, + readonly correlation?: string, + readonly screen?: ScreenSpec, + /** Registry type of the node that suspended — the resume gate's key (#3801). */ + readonly nodeType?: string, + ) {} } function isSuspendSignal(err: unknown): err is FlowSuspendSignal { @@ -552,6 +559,19 @@ export interface SuspendedRun { flowVersion?: number; /** The node the run paused at; resume continues from its out-edges. */ nodeId: string; + /** + * Registry type of the node that produced the pause (`approval`, `screen`, + * `wait`, …), captured at suspend time. Keys the resume gate (#3801): the + * descriptor's `resumeAuthority` decides whether a raw + * {@link AutomationEngine.resume} is a legitimate continuation. + * + * Recorded on the suspension rather than re-derived from the live flow so + * the gate reflects what actually paused the run — a flow republished + * mid-pause cannot re-type the node out from under it. Optional: rows + * persisted before this field existed have none, and the gate falls back + * to the flow definition for those. + */ + nodeType?: string; /** Snapshot of the flow variable map at suspend time. */ variables: Record; steps: StepLogEntry[]; @@ -1870,6 +1890,7 @@ export class AutomationEngine implements IAutomationService { flowName, flowVersion: flow.version, nodeId: err.nodeId, + nodeType: err.nodeType, variables: Object.fromEntries(variables), steps, context: runContext, @@ -1954,11 +1975,133 @@ export class AutomationEngine implements IAutomationService { * parent. Both directions compose recursively, so arbitrarily nested * subflow pauses resolve from either end (UI holds the parent run id; * approval/wait infrastructure holds the child's). + * + * **Authorization (#3801).** This is the public door — the generic REST + * resume route and the SDK land here — so it is gated on WHAT THE RUN IS + * PARKED ON before any state is touched: a suspension whose node declares + * `resumeAuthority: 'service'` is refused unless the signal carries + * {@link RESUME_AUTHORITY_SERVICE}. The engine's own continuations + * (subflow delegation / up-bubble, `map` re-entry, wait-timer wake) go + * through {@link resumeInternal} and are not re-gated — they continue work + * some already-authorized call started. */ async resume(runId: string, signal?: ResumeSignal): Promise { + const refusal = await this.refuseGatedResume(runId, signal); + if (refusal) return refusal; return this.resumeInternal(runId, signal, false); } + /** + * The resume gate (#3801): decide whether `signal` may continue the run + * `runId` is parked on, returning a refusal result or `null` to allow. + * + * Keyed on the SUSPENDED NODE, not the caller or the route — an `approval` + * pause continues only through `ApprovalService` (which records the + * decision and enforces the slate), while a `screen` pause stays open to + * 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. + */ + private async refuseGatedResume(runId: string, signal?: ResumeSignal): Promise { + const run = await this.resolveEffectiveSuspension(runId); + if (!run) return null; + + const nodeType = run.nodeType ?? this.flows.get(run.flowName)?.nodes.find(n => n.id === run.nodeId)?.type; + if (!nodeType) return null; + if (this.resolveResumeAuthority(nodeType) !== 'service') return null; + // The owning service stamped the signal — this resume IS the recorded + // 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}')`; + this.logger.warn( + `[automation] refused resume of run '${runId}': parked on ${nodeType} node ${at}, which is resumable ` + + `only through its owning service (resumeAuthority: 'service')`, + ); + 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`, + }; + } + + /** + * The `resumeAuthority` in force for a node type, following a deprecated + * ADR-0018 alias to its canonical type. + * + * An alias's descriptor is synthesized by {@link registerNodeAlias} and does + * NOT copy the canonical's capabilities, so reading it directly would hand + * anyone who authors the old type name an ungated pause — the gate is + * declared and not enforced, one rename away. Resolving live (rather than + * snapshotting at alias-registration time) also keeps it correct whichever + * order the two register in. No alias of a pausing type exists today; this + * keeps it from becoming a hole the day one does. + */ + private resolveResumeAuthority(nodeType: string): ActionDescriptor['resumeAuthority'] { + let descriptor = this.actionDescriptors.get(nodeType); + for (let hop = 0; descriptor?.aliasOf && hop < AutomationEngine.MAX_ALIAS_HOPS; hop++) { + const canonical = this.actionDescriptors.get(descriptor.aliasOf); + if (!canonical || canonical === descriptor) break; + descriptor = canonical; + } + return descriptor?.resumeAuthority ?? 'any'; + } + + /** Depth bound for the subflow chain walk — a corrupt correlation cycle + * must not spin the gate. Far above any real nesting. */ + private static readonly MAX_SUSPENSION_CHAIN_DEPTH = 32; + + /** Depth bound for the alias → canonical hop in {@link resolveResumeAuthority}. */ + 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). + */ + 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); + if (!childRunId || seen.has(childRunId)) return run; + seen.add(childRunId); + const child = await this.loadSuspendedRun(childRunId); + if (!child) return run; + run = child; + } + return run; + } + + /** Read a suspended run from the hot cache, falling back to the durable + * store. Read-only — never consumes the suspension. */ + private async loadSuspendedRun(runId: string): Promise { + const cached = this.suspendedRuns.get(runId); + if (cached) return cached; + if (!this.store) return null; + try { + return await this.store.load(runId); + } catch (err) { + this.logger.warn( + `[automation] failed to load suspended run '${runId}' from durable store: ${(err as Error).message}`, + ); + return null; + } + } + /** * @param skipBubble - Set when the caller is the subflow DELEGATION path, * which continues the parent itself after the child completes — the @@ -1976,16 +2119,7 @@ export class AutomationEngine implements IAutomationService { try { // Hot path: suspended in this process. Cold path: rehydrate from the // durable store (e.g. the process restarted since the pause, ADR-0019). - let run = this.suspendedRuns.get(runId) ?? null; - if (!run && this.store) { - try { - run = await this.store.load(runId); - } catch (err) { - this.logger.warn( - `[automation] failed to load suspended run '${runId}' from durable store: ${(err as Error).message}`, - ); - } - } + const run = await this.loadSuspendedRun(runId); if (!run) { return { success: false, error: `No suspended run '${runId}'` }; } @@ -2009,9 +2143,7 @@ export class AutomationEngine implements IAutomationService { const childRunId = run.correlation.slice('subflow:'.length); // Capture the child's row BEFORE resuming consumes it — the // output-variable mapping rides on the child's context. - const childRun = - this.suspendedRuns.get(childRunId) ?? - (this.store ? await this.store.load(childRunId).catch(() => null) : null); + const childRun = await this.loadSuspendedRun(childRunId); if (childRun) { const childRes = await this.resumeInternal(childRunId, signal, true); if (childRes.status === 'paused') { @@ -2128,6 +2260,7 @@ export class AutomationEngine implements IAutomationService { await this.persistSuspendedRun({ ...run, nodeId: err.nodeId, + nodeType: err.nodeType, variables: Object.fromEntries(variables), steps, correlation: err.correlation, @@ -2772,7 +2905,7 @@ export class AutomationEngine implements IAutomationService { // up to execute()/resume(), which persists a continuation. Traversal // of this node's out-edges happens on resume, not now. if (result.suspend) { - throw new FlowSuspendSignal(node.id, result.correlation, result.screen); + throw new FlowSuspendSignal(node.id, result.correlation, result.screen, node.type); } // #3447 P2: an executor may pick its own out-edge without suspending diff --git a/packages/services/service-automation/src/resume-authority-gate.test.ts b/packages/services/service-automation/src/resume-authority-gate.test.ts new file mode 100644 index 0000000000..9774e43074 --- /dev/null +++ b/packages/services/service-automation/src/resume-authority-gate.test.ts @@ -0,0 +1,272 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Resume authorization gate (#3801). + * + * `POST /automation/:name/runs/:runId/resume` forwards a caller-supplied + * signal straight into `AutomationEngine.resume`, which validated MACHINE + * state only — the run exists, the flow exists, the suspended node still + * exists — and nothing about who was asking. Approval nodes suspend and + * resume through exactly that mechanism, so a raw resume walked the `approve` + * edge with no `sys_approval_action` row and no status mirror, leaving the + * request row and the run permanently disagreeing. + * + * The gate keys on WHAT THE RUN IS PARKED ON: a suspension produced by a node + * whose descriptor declares `resumeAuthority: 'service'` is resumable only + * with the in-process {@link RESUME_AUTHORITY_SERVICE} marker — which no JSON + * body can carry. A `screen` pause (the reason the route exists) is untouched. + */ + +import { describe, it, expect, beforeEach } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +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'; + +function silentLogger(): any { + return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } }; +} +function pluginCtx(): any { + return { logger: silentLogger(), getService() { throw new Error('none'); } }; +} + +/** + * An engine wired with the node types every flow below is built from: a + * service-gated pauser, an ungated one, and a downstream marker that records + * into `downstream` so "did the resume actually continue the run?" is + * observable. `store` makes the pause durable (restart tests share one). + */ +function makeEngine(downstream: string[], store?: InMemorySuspendedRunStore): AutomationEngine { + const engine = new AutomationEngine(silentLogger(), store); + registerPausers(engine); + engine.registerNodeExecutor({ + type: 'after', + async execute(node) { downstream.push(node.id); return { success: true }; }, + }); + return engine; +} + +/** Registers the two pausing node types every flow below is built from. */ +function registerPausers(engine: AutomationEngine): void { + // Stands in for `approval`: the owning service authorizes + records the + // decision, then resumes. A raw resume must not. + engine.registerNodeExecutor({ + type: 'gated_pause', + descriptor: defineActionDescriptor({ + type: 'gated_pause', version: '1.0.0', name: 'Gated Pause', + supportsPause: true, isAsync: true, resumeAuthority: 'service', + }), + async execute() { return { success: true, suspend: true, correlation: 'req_1' }; }, + }); + // Stands in for `screen` / `wait`: the caller supplies the continuation, so + // the generic route is the intended door (descriptor defaults to 'any'). + engine.registerNodeExecutor({ + type: 'open_pause', + descriptor: defineActionDescriptor({ + type: 'open_pause', version: '1.0.0', name: 'Open Pause', supportsPause: true, + }), + async execute() { return { success: true, suspend: true }; }, + }); +} + +/** A linear flow: start → → after → end. */ +function pauseFlow(name: string, pauseType: string) { + return { + name, + label: name, + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'pause', type: pauseType, label: 'Pause' }, + { id: 'after', type: 'after', label: 'After' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'pause' }, + { id: 'e2', source: 'pause', target: 'after', label: 'approve' }, + { id: 'e3', source: 'after', target: 'end' }, + ], + }; +} + +describe('resume authorization gate (#3801)', () => { + let engine: AutomationEngine; + let downstream: string[]; + + beforeEach(() => { + downstream = []; + engine = makeEngine(downstream); + }); + + it('refuses a raw resume of a run parked on a service-gated node', async () => { + engine.registerFlow('gated_flow', pauseFlow('gated_flow', 'gated_pause') as never); + const paused = await engine.execute('gated_flow'); + expect(paused.status).toBe('paused'); + + // Exactly what the REST route builds out of a JSON body. + const refused = await engine.resume(paused.runId!, { branchLabel: 'approve', output: { decision: 'approve' } }); + + expect(refused.success).toBe(false); + expect(refused.code).toBe('forbidden'); + expect(refused.error).toMatch(/only its owning service may resume/i); + // Refused, not consumed: downstream never ran and the pause is still live, + // so the real decision can still land. + expect(downstream).toEqual([]); + expect(engine.listSuspendedRuns()).toHaveLength(1); + }); + + it('lets the owning service through with the in-process marker', async () => { + engine.registerFlow('gated_flow', pauseFlow('gated_flow', 'gated_pause') as never); + const paused = await engine.execute('gated_flow'); + + const resumed = await engine.resume(paused.runId!, { + branchLabel: 'approve', + output: { decision: 'approve' }, + [RESUME_AUTHORITY_SERVICE]: true, + }); + + expect(resumed.success).toBe(true); + expect(resumed.code).toBeUndefined(); + expect(downstream).toEqual(['after']); + expect(engine.listSuspendedRuns()).toHaveLength(0); + }); + + it('leaves an ungated pause (screen / wait) resumable through the route', async () => { + engine.registerFlow('open_flow', pauseFlow('open_flow', 'open_pause') as never); + const paused = await engine.execute('open_flow'); + + const resumed = await engine.resume(paused.runId!, { variables: { new_assignee: 'ada' } }); + + expect(resumed.success).toBe(true); + expect(downstream).toEqual(['after']); + }); + + it('a node type with no published descriptor stays ungated', async () => { + // `registerNodeExecutor` without a descriptor — the gate has nothing + // declaring itself service-owned, so behaviour is unchanged. + engine.registerNodeExecutor({ + type: 'bare_pause', + async execute() { return { success: true, suspend: true }; }, + }); + engine.registerFlow('bare_flow', pauseFlow('bare_flow', 'bare_pause') as never); + const paused = await engine.execute('bare_flow'); + + expect((await engine.resume(paused.runId!)).success).toBe(true); + expect(downstream).toEqual(['after']); + }); + + it('gates a flow authored against a deprecated ALIAS of the gated type', async () => { + // `registerNodeAlias` synthesizes the alias descriptor and does not copy + // the canonical's capabilities — reading it directly would hand anyone who + // authors the old type name an ungated pause. + engine.registerNodeAlias('legacy_pause', 'gated_pause', { name: 'Legacy Pause' }); + engine.registerFlow('alias_flow', pauseFlow('alias_flow', 'legacy_pause') as never); + const paused = await engine.execute('alias_flow'); + + const refused = await engine.resume(paused.runId!, { branchLabel: 'approve' }); + expect(refused.code).toBe('forbidden'); + expect(downstream).toEqual([]); + }); + + it('reports machine-state errors, not a refusal, for an unknown run', async () => { + const result = await engine.resume('run_does_not_exist'); + expect(result.success).toBe(false); + expect(result.code).toBeUndefined(); + expect(result.error).toContain('No suspended run'); + }); + + // ── the suspension carries its own node type ───────────────────────── + + it('tags the suspension with the node type that produced it', async () => { + engine.registerFlow('gated_flow', pauseFlow('gated_flow', 'gated_pause') as never); + const paused = await engine.execute('gated_flow'); + + const run = (engine as any).suspendedRuns.get(paused.runId!); + expect(run.nodeType).toBe('gated_pause'); + }); + + it('gates a run rehydrated from the durable store after a restart', async () => { + const store = new InMemorySuspendedRunStore(); + const engineA = makeEngine(downstream, store); + engineA.registerFlow('gated_flow', pauseFlow('gated_flow', 'gated_pause') as never); + const paused = await engineA.execute('gated_flow'); + + // Process lifetime #2: nothing in memory, the pause comes back off the row. + const engineB = makeEngine(downstream, store); + engineB.registerFlow('gated_flow', pauseFlow('gated_flow', 'gated_pause') as never); + + const refused = await engineB.resume(paused.runId!, { branchLabel: 'approve' }); + expect(refused.code).toBe('forbidden'); + // Still resumable by the service that owns it. + const ok = await engineB.resume(paused.runId!, { branchLabel: 'approve', [RESUME_AUTHORITY_SERVICE]: true }); + expect(ok.success).toBe(true); + }); + + it('falls back to the flow definition for a row stored before the tag existed', async () => { + const store = new InMemorySuspendedRunStore(); + const e = makeEngine(downstream, store); + e.registerFlow('gated_flow', pauseFlow('gated_flow', 'gated_pause') as never); + const paused = await e.execute('gated_flow'); + + // Strip the tag from both the hot cache and the durable row, exactly as a + // pause persisted by an older build would come back. + const cached = (e as any).suspendedRuns.get(paused.runId!); + delete cached.nodeType; + const stored = await store.load(paused.runId!); + delete (stored as any).nodeType; + await store.save(stored!); + + const refused = await e.resume(paused.runId!, { branchLabel: 'approve' }); + expect(refused.code).toBe('forbidden'); + }); + + // ── subflow chains: the signal lands on the CHILD ───────────────────── + + describe('through a subflow pause', () => { + /** Parent whose `subflow` node calls a child that parks on `pauseType`. */ + function registerChain(e: AutomationEngine, pauseType: string): void { + registerSubflowNode(e, pluginCtx()); + e.registerFlow('child_flow', pauseFlow('child_flow', pauseType) as never); + e.registerFlow('parent_flow', { + name: 'parent_flow', + label: 'parent_flow', + type: 'autolaunched', + nodes: [ + { id: 'ps', type: 'start', label: 'Start' }, + { id: 'call', type: 'subflow', label: 'Call Child', config: { flowName: 'child_flow' } }, + { id: 'pe', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'p1', source: 'ps', target: 'call' }, + { id: 'p2', source: 'call', target: 'pe' }, + ], + } as never); + } + + it('refuses a resume of the PARENT when the child is parked on a gated node', async () => { + registerChain(engine, 'gated_pause'); + const paused = await engine.execute('parent_flow'); + expect(paused.status).toBe('paused'); + + // The parent is parked on a `subflow` node — ungated on its own. But the + // signal is DELEGATED to the child, so the gate has to judge the child. + const refused = await engine.resume(paused.runId!, { branchLabel: 'approve' }); + + expect(refused.code).toBe('forbidden'); + expect(downstream).toEqual([]); + // Neither run was consumed — parent and child are both still suspended. + expect(engine.listSuspendedRuns()).toHaveLength(2); + }); + + it('still delegates a legitimate parent resume when the child pause is ungated', async () => { + registerChain(engine, 'open_pause'); + const paused = await engine.execute('parent_flow'); + + const resumed = await engine.resume(paused.runId!, { branchLabel: 'approve' }); + + expect(resumed.success).toBe(true); + expect(downstream).toEqual(['after']); + }); + }); +}); diff --git a/packages/services/service-automation/src/suspended-run-store.ts b/packages/services/service-automation/src/suspended-run-store.ts index ccf62274c4..03316dce22 100644 --- a/packages/services/service-automation/src/suspended-run-store.ts +++ b/packages/services/service-automation/src/suspended-run-store.ts @@ -358,6 +358,9 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { flow_name: run.flowName, flow_version: run.flowVersion ?? null, node_id: run.nodeId, + // Node TYPE, not just id — the resume gate (#3801) keys on what produced + // the pause, and it has to survive the restart the pause itself survives. + node_type: run.nodeType ?? null, status: 'paused', correlation: run.correlation ?? null, user_id: ctx.userId ?? null, @@ -378,6 +381,7 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { flowName: String(row.flow_name ?? ''), flowVersion: row.flow_version ?? undefined, nodeId: String(row.node_id ?? ''), + nodeType: row.node_type ?? undefined, variables: parseJson>(row.variables_json, {}), steps: parseJson(row.steps_json, []), context: parseJson(row.context_json, {}), diff --git a/packages/services/service-automation/src/sys-automation-run.object.ts b/packages/services/service-automation/src/sys-automation-run.object.ts index e6516bb4b6..1c9098c5a4 100644 --- a/packages/services/service-automation/src/sys-automation-run.object.ts +++ b/packages/services/service-automation/src/sys-automation-run.object.ts @@ -89,6 +89,14 @@ export const SysAutomationRun = ObjectSchema.create({ group: 'State', }), + node_type: Field.text({ + label: 'Node Type', + required: false, + maxLength: 255, + description: 'Registry type of the node a suspended run paused at (approval / screen / wait / …). Keys the resume authorization gate (#3801) — captured at suspend time rather than re-read from a flow that may have been republished since. Null on rows written before the gate shipped, and on terminal history rows.', + group: 'State', + }), + status: Field.select( ['running', 'paused', 'completed', 'failed'], { diff --git a/packages/spec/api-surface.json b/packages/spec/api-surface.json index 13f777e006..9994967818 100644 --- a/packages/spec/api-surface.json +++ b/packages/spec/api-surface.json @@ -3632,6 +3632,7 @@ "QueueMessageRecord (interface)", "QueuePublishOptions (interface)", "RESERVED_RLS_MEMBERSHIP_KEYS (const)", + "RESUME_AUTHORITY_SERVICE (const)", "RLS_MEMBERSHIP_RESOLVER_SERVICE (const)", "RealtimeEventHandler (type)", "RealtimeEventPayload (interface)", diff --git a/packages/spec/src/automation/node-executor.test.ts b/packages/spec/src/automation/node-executor.test.ts index fd9a4b33e4..4c32187503 100644 --- a/packages/spec/src/automation/node-executor.test.ts +++ b/packages/spec/src/automation/node-executor.test.ts @@ -5,6 +5,8 @@ import { WaitTimeoutBehaviorSchema, WaitExecutorConfigSchema, NodeExecutorDescriptorSchema, + ActionDescriptorSchema, + defineActionDescriptor, WAIT_EXECUTOR_DESCRIPTOR, type WaitResumePayload, type WaitExecutorConfig, @@ -252,3 +254,23 @@ describe('Wait Executor — pause/resume scenario', () => { expect(resume.variables?.approval_status).toBe('approved'); }); }); + +// --------------------------------------------------------------------------- +// ActionDescriptorSchema — resumeAuthority (#3801) +// --------------------------------------------------------------------------- +describe('ActionDescriptorSchema.resumeAuthority', () => { + const base = { type: 'demo', version: '1.0.0', name: 'Demo' }; + + it("defaults to 'any' — the generic resume route stays the door for screen / wait", () => { + expect(defineActionDescriptor(base).resumeAuthority).toBe('any'); + }); + + it("accepts 'service' for a node only its owning service may resume", () => { + const desc = defineActionDescriptor({ ...base, supportsPause: true, resumeAuthority: 'service' }); + expect(desc.resumeAuthority).toBe('service'); + }); + + it('rejects an unknown authority rather than silently defaulting it open', () => { + expect(() => ActionDescriptorSchema.parse({ ...base, resumeAuthority: 'admin' })).toThrow(); + }); +}); diff --git a/packages/spec/src/automation/node-executor.zod.ts b/packages/spec/src/automation/node-executor.zod.ts index 3bec43c1ff..dbb3f7e9c8 100644 --- a/packages/spec/src/automation/node-executor.zod.ts +++ b/packages/spec/src/automation/node-executor.zod.ts @@ -259,6 +259,31 @@ export const ActionDescriptorSchema = lazySchema(() => z.object({ isAsync: z.boolean().default(false) .describe('Suspends the flow awaiting an external reply'), + /** + * WHO may resume a run this node suspended (#3801). The generic resume + * route (`POST /automation/:name/runs/:runId/resume`) validates machine + * state only — the run exists, the flow exists, the suspended node still + * exists — so the node type that *produced* the pause is what decides + * whether a raw resume is a legitimate continuation or a bypass. + * + * - `'any'` (default) — the caller supplies the continuation and the route + * is the intended door: a `screen` node's collected inputs, a `wait` + * node's external signal. + * - `'service'` — resuming is a SIDE EFFECT of a decision some service must + * authorize and record first, so only that service may drive it. An + * `approval` node declares this: `ApprovalService.decide` enforces the + * approver slate, writes the `sys_approval_action` row and mirrors the + * status field, then resumes. A raw resume around it would walk the + * `approve` edge with no decision recorded, leaving the request row and + * the run permanently disagreeing. + * + * The engine enforces it: a resume of a `'service'` suspension is refused + * unless the signal carries the in-process `RESUME_AUTHORITY_SERVICE` + * marker — a symbol, so a JSON body can never carry it. + */ + resumeAuthority: z.enum(['any', 'service']).default('any') + .describe("Who may resume a run this node suspended: 'any' (the generic resume route) or 'service' (only the owning service, e.g. approvals)"), + /** * Runtime maturity of the capability behind this descriptor (ADR-0041 §4). * The platform routinely ships contracts ahead of runtimes; `reserved` diff --git a/packages/spec/src/contracts/automation-service.ts b/packages/spec/src/contracts/automation-service.ts index 1ce3134c99..1a14a8b153 100644 --- a/packages/spec/src/contracts/automation-service.ts +++ b/packages/spec/src/contracts/automation-service.ts @@ -164,6 +164,17 @@ export interface AutomationResult { error?: string; /** Execution duration in milliseconds */ durationMs?: number; + /** + * Machine-readable failure classification, set alongside `error` when the + * caller must distinguish *why* it failed rather than just report it. + * + * Today the one value is `'forbidden'` — {@link IAutomationService.resume} + * refused because the run is parked on a node whose descriptor declares + * `resumeAuthority: 'service'` (#3801). A transport maps it to 403; without + * it a resume denied on authorization grounds is indistinguishable from + * "no such run". + */ + code?: 'forbidden'; /** * Lifecycle status. `'paused'` means the run suspended at a node (e.g. * an Approval node awaiting a human decision, ADR-0019) and can be @@ -190,6 +201,24 @@ export interface AutomationResult { errorMessage?: string; } +/** + * Marker a SERVICE stamps on its {@link ResumeSignal} to prove the resume is + * the tail of a decision it already authorized and recorded (#3801). + * + * A run parked on a node whose descriptor declares `resumeAuthority: 'service'` + * (today: `approval`) is resumable ONLY with this marker present. It is a + * symbol on purpose: the generic resume route builds its signal out of a JSON + * body, and no JSON body can produce a symbol-keyed property — so the marker + * is unforgeable from outside the process, while an in-process owner + * (`ApprovalService.decide` and friends) just sets it. + * + * Registered via `Symbol.for` so duplicate copies of this package in one + * process still agree on identity. + */ +export const RESUME_AUTHORITY_SERVICE: unique symbol = Symbol.for( + 'objectstack.automation.resume.service', +); + /** Signal payload used to resume a paused run (ADR-0019). */ export interface ResumeSignal { /** @@ -211,6 +240,13 @@ export interface ResumeSignal { * `{var}` interpolation and conditions read them directly. */ variables?: Record; + /** + * Set by the service that OWNS the suspension to clear the resume gate on + * a `resumeAuthority: 'service'` node (#3801). See + * {@link RESUME_AUTHORITY_SERVICE} — unforgeable from an HTTP body, and + * ignored entirely on `'any'` nodes. + */ + [RESUME_AUTHORITY_SERVICE]?: true; } export interface IAutomationService { @@ -285,6 +321,15 @@ export interface IAutomationService { * have previously returned `{ status: 'paused', runId }` from * {@link execute} (or a prior `resume`). Continues traversal from the * suspended node's out-edges, applying `signal.output` / `signal.branchLabel`. + * + * **Gated by the suspended node (#3801).** When the run is parked on a node + * whose descriptor declares `resumeAuthority: 'service'`, only that node's + * owning service may continue it — the call is refused with + * `{ success: false, code: 'forbidden' }` unless `signal` carries + * {@link RESUME_AUTHORITY_SERVICE}. The gate follows a subflow pause down + * to the child the signal would actually land on, so a parent parked on a + * `subflow` node is no way around it. + * * @param runId - The paused run's id * @param signal - Optional output to merge and/or branch label to follow * @returns The result of continuing the run (may itself be `'paused'` again) diff --git a/skills/objectstack-automation/SKILL.md b/skills/objectstack-automation/SKILL.md index 4752ac1967..e8caf3a3e6 100644 --- a/skills/objectstack-automation/SKILL.md +++ b/skills/objectstack-automation/SKILL.md @@ -420,7 +420,10 @@ your normal `flows: [...]`. A decision is recorded through `ApprovalService.decide()` (or the REST routes `POST /api/v1/approvals/requests/:id/approve` | `/reject`). That finalizes the `sys_approval_request` and **resumes** the suspended run down the matching -branch — you never resume the flow by hand. +branch — you never resume the flow by hand, and since #3801 you *cannot*: the +`approval` node declares `resumeAuthority: 'service'`, so +`POST /api/v1/automation/:name/runs/:runId/resume` answers **403** for a run +parked on one (including via a `subflow` pause) and changes nothing. A decision may also carry **structured outputs** (`{ outputs: { … } }` in the decide body) when the node declares the keys in `decisionOutputs` — the author From 65202aaf0f0eb785f982e4df0c930b63b6a5ad84 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 28 Jul 2026 07:36:52 +0000 Subject: [PATCH 2/2] docs(approvals): the 'never resume directly' rule is now enforced, not advisory (#3801) Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013gvN32u1EiuvY9uQEMJiMR --- content/docs/automation/approvals.mdx | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/content/docs/automation/approvals.mdx b/content/docs/automation/approvals.mdx index 4d09b3aa09..5565f897fc 100644 --- a/content/docs/automation/approvals.mdx +++ b/content/docs/automation/approvals.mdx @@ -288,7 +288,13 @@ curl -b cookies.txt -X POST \ `actorId` defaults to the caller. The actor **must** be in `pending_approvers` or the call returns 403 (`FORBIDDEN: actor '…' is not a pending approver`); a request that isn't pending returns 409 (`INVALID_STATE`). Always go through -these endpoints — never resume the flow run directly. +these endpoints — never resume the flow run directly, and since #3801 you +**cannot**: `POST /api/v1/automation/{flow}/runs/{runId}/resume` answers 403 for +a run parked on an `approval` node (including via a `subflow` pause) and changes +nothing — the request stays pending and the run stays parked, so the real +decision still lands. The approver slate, the `sys_approval_action` row and the +status mirror all live on this path; it is the only one that produces a +consistent outcome. A decision may carry **file attachments** — `attachments: string[]` of `sys_file` ids — recorded on its audit row (e.g. a signed contract on the approve):