diff --git a/.changeset/approvals-continue-restored-suspension.md b/.changeset/approvals-continue-restored-suspension.md new file mode 100644 index 0000000000..53b55438bf --- /dev/null +++ b/.changeset/approvals-continue-restored-suspension.md @@ -0,0 +1,16 @@ +--- +"@objectstack/plugin-approvals": minor +--- + +A restored approval suspension can now be decided again, not only cancelled. + +`AutomationEngine.restoreConsumedSuspension` re-arms the pause of a run that stranded mid-resume and tells the operator to *re-issue the continuation*. For an `approval` suspension nobody could: every approvals door that stamps the resume marker — `decide`, `recall`, `sendBack`, `resubmit` — guards on a `pending` request, and the row is terminal, written by the very call that stranded the run; and the generic engine door refuses an `approval` pause outright, because that node declares `resumeAuthority: 'service'`. The only remaining verb was `cancelRun`, which discards the branch's downstream work — so the advertised repair produced a run that looked resumable and was not decidable. + +Measured against the real engine and the real decision door: the restored suspension lacks nothing. A `resumeAuthority`-marked resume walks the restored pause to completion. What was missing was an **issuer** on the approvals side, and that is what this adds. + +- **`ApprovalService.continueRestoredRun(requestId, options?)`** re-issues the continuation the recorded outcome already produced once, against a pause an operator has re-armed. It reports which outcome it replayed, which edge it walked, and whether the signal was replayed exactly or rebuilt (`source: 'journal' | 'reconstructed'`). +- **The failing door now journals the signal it was carrying** on the repairable exit — the engine's own `status: 'stranded'` discriminator, the one exit that journals a repair snapshot — under `__strandedContinuation` in the request's `node_config_json`, beside the `__decisionOutputs` side-channel that was already there. Best-effort: it is awaited but can never replace the `RESUME_FAILED` throw the decision's caller is owed. +- **The continuation is tied to this request's own pause, by three guards.** A boolean "is this run suspended" is not enough: a run outlives any one request, so a terminal row's continuation could be issued against whatever pause the run happened to be sitting on. It now requires that the request is still the newest on its run, that a pause exists (strictly — an unreadable store throws rather than reading as "not suspended"), and that the pause is parked **where this request's recorded outcome was issued from**. That node is signal-aware, not simply the row's own: `approve`, `reject`, `revise` and `recall` are all issued at the request's own approval node, but a `resubmit` is only ever issued from the revise window the request's `revise` edge leads to, so its pause is re-armed there while the row still records the approval node. Comparing against the row's own node refused exactly that case, and told the operator the pause was not this request's when it was. The node check is fail-closed in every direction, including an engine that cannot report where a run is parked and a revise window this service cannot derive from the flow definition. This needs no new automation-engine surface: `listSuspendedRunsDurable` is already public, and the approvals-side resume interface simply declares it. +- **Runs stranded before this shipped are served too**, and where the signal cannot be proved the verb **refuses instead of guessing**. A status is not the same thing as a continuation, and three of the four terminal statuses have more than one writer or issuer: `approved` is unambiguous; `rejected` has two writers, discriminated by the `revise` action row that only ADR-0044's revision-limit auto-rejection leaves behind; `returned` has one writer but **two** issuers, discriminated by the `resubmit` action row whose sole writer is `resubmit` — without it a stranded resubmit was rebuilt as a send-back and walked the wrong edge, proceeding only through the engine's unmatched-label fallback with the wrong output; and `recalled` has two writers across **three** behaviours, two of which issue no continuation at all, so it is **refused on the rebuild path** with a message naming what an operator can do instead. Journal-recoverable is a **measured, named set** rather than a blanket claim: `approve`, `reject` and `resubmit` continuations replay end to end through the verb, and `reject` and `resubmit` do so on the rebuild path as well. Two shapes are refused by design and stay refused — a `rejected` row that also carries a `revise` action, and a `recalled` row with no journal. NOT covered by a pin, and so not claimed: the `approve` rebuild path and the `recall` journal path. + +⛔ What this deliberately does not do, each pinned: it does not re-open or rewrite the request row — all four `pending` guards are untouched and no status, mirror field or audit row is written, so a decided request still cannot be decided again through the front door; it does not relax `resumeAuthority: 'service'`, since the resume still goes through the one call site that stamps the marker; and it does not change `ApprovalDecisionResult`, whose shape is the subject of an open ruling. It also grants no capability in-process code did not already have — `RESUME_AUTHORITY_SERVICE` is importable by any host — what it adds is the guarded form, which checks that the pause is genuinely re-armed and that the signal matches what was recorded. Like the engine verb it completes, it is an in-process operator repair: no REST route, and no entry in the spec `ApprovalService` contract. diff --git a/content/docs/permissions/system-context.mdx b/content/docs/permissions/system-context.mdx index f2a9a94f18..20421d525b 100644 --- a/content/docs/permissions/system-context.mdx +++ b/content/docs/permissions/system-context.mdx @@ -145,7 +145,7 @@ The largest single consumer — **17 of the 105 sites**. |:--|:---|:---|:---|:---| | 40 | **Approval record lock released** — a locked record is writable | plugin-approvals | Get: engine self-writes (the status mirror) pass. Lose: the lock that stops edits while an approval is live. Note there is deliberately **no admin exemption** here — only `isSystem` | `lifecycle-hooks.ts:347` | | 41 | Delegation write guard bypassed | plugin-approvals | Get: service / seed / import may write delegation rows naming another delegator | `lifecycle-hooks.ts:570` | -| 42 | Approval actor / submitter / pending-approver checks bypassed (8 sites) | plugin-approvals | Get: approve, reject, recall, reassign without being a pending approver or the submitter | `plugin-approvals/src/approval-service.ts:963`, `:1072`, `:3305`, `:3453`, `:3621`, `:3692`, `:3881`, `:3921` | +| 42 | Approval actor / submitter / pending-approver checks bypassed (8 sites) | plugin-approvals | Get: approve, reject, recall, reassign without being a pending approver or the submitter | `plugin-approvals/src/approval-service.ts:1044`, `:1153`, `:3409`, `:3557`, `:3725`, `:3796`, `:3985`, `:4025` | | 43 | Saved-report ownership is **assignable**, and an update may reassign it | plugin-reports | Get: `ownerId` from input is honoured. A non-system caller always owns what it creates and can never reassign | `plugin-reports/src/report-service.ts:404`, `:425` | | 44 | Saved-report access / export / mutation gates bypassed | plugin-reports | Get: read, bulk-export and overwrite any report | `plugin-reports/src/report-service.ts:343`, `:372`, `:447`, `:684` | | 45 | Attachment access hooks return early (insert + update + delete, and the read AST) | service-storage | Lose: attachment visibility scoping | `attachment-access-hooks.ts:300`, `:349`, `:448`, `:524` | diff --git a/content/docs/permissions/tenant-audit-census.mdx b/content/docs/permissions/tenant-audit-census.mdx index 936abdd9de..6ce4bca15b 100644 --- a/content/docs/permissions/tenant-audit-census.mdx +++ b/content/docs/permissions/tenant-audit-census.mdx @@ -98,7 +98,7 @@ are reported as `undecidable` rather than assumed either way. The same holds twice over for the context. An options argument spelled as a literal can be read; one spelled `options`, `{ ...opts }`, or handed through a -forwarding shim cannot, and **67 of the 221 sites are spelled that way**. A +forwarding shim cannot, and **67 of the 222 sites are spelled that way**. A context resolved from an inline literal or a local `const` can be tested for `isSystem`; one arriving from a helper call cannot. @@ -147,10 +147,10 @@ reproduce them. Where it disagrees, it disagrees on the page: | carried figure | where it survives | this census | | :--- | :--- | ---: | -| 175 write call sites | quoted in the merged changeset | **221** | +| 175 write call sites | quoted in the merged changeset | **222** | | 24 carrying no tenant context | quoted in the merged changeset | **9** provable and tenancy-enabled; **32** more whose options argument is unreadable | -| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **147 of 221** decidable, **74** undecidable | -| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 103 decidably elevated, 0 decidably not, 101 undecidable | +| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **148 of 222** decidable, **74** undecidable | +| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 104 decidably elevated, 0 decidably not, 101 undecidable | | 141 and 132, two independent re-derivations | the card that filed this work | — | **The differences are not reconciled, and deliberately so.** The old census's @@ -167,11 +167,11 @@ would report a smaller number and would not say so. The fourth row is the one worth flagging to anyone citing it. **The 135 / 77% figure has no surviving corroboration anywhere in the tree.** This census reads -103 of 221 (47%) as decidably elevated, with 101 more whose elevation is a +104 of 222 (47%) as decidably elevated, with 101 more whose elevation is a run-time fact — so the claim is neither confirmed nor refuted, and the honest answer is that a static reading cannot settle it. -⇒ **Cite `9 / 221`, and say what it is**: the sites whose options argument was +⇒ **Cite `9 / 222`, and say what it is**: the sites whose options argument was READ and holds no tenant context, against a decidably tenancy-enabled object. That is the control's provable yield surface. ⛔ Do not cite it as "the sites without tenant context" — **32 further sites** have an options argument this @@ -183,28 +183,28 @@ cannot read, and they are neither in nor out. | what | count | | :--- | ---: | -| write call sites on the application surface | **221** | -| …whose object name is statically decidable | 147 | +| write call sites on the application surface | **222** | +| …whose object name is statically decidable | 148 | | …whose object name is chosen at run time | 74 | -| …against an object with tenancy ENABLED | 147 | +| …against an object with tenancy ENABLED | 148 | | …against an object that declares tenancy off | 0 | -| threading a tenant context | 137 | +| threading a tenant context | 138 | | PROVABLY carrying none (options read, no context key) | **17** | | …of those, against a decidably tenancy-enabled object | **9** | | options argument UNREADABLE — may or may not carry one | 67 | | …of those, against a decidably tenancy-enabled object | 32 | -| threading a decidably ELEVATED (`isSystem`) context | 103 | +| threading a decidably ELEVATED (`isSystem`) context | 104 | | threading a context that is decidably NOT elevated | 0 | | threading a context whose elevation is a run-time fact | 101 | | how the instrument reached the site | count | | :--- | ---: | -| receiver carried a readable engine type | 176 | +| receiver carried a readable engine type | 177 | | receiver erased, placed by the object NAME | 19 | | receiver erased, placed by an `object: string` PARAMETER | 15 | | receiver erased, placed by an `UNTYPED_RECEIVERS` row | 11 | -| object name spelled inline | 108 | +| object name spelled inline | 109 | | object name spelled through a `const` | 39 | | object name is an `object: string` parameter | 19 | | object name is some other run-time expression | 55 | @@ -224,11 +224,11 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-05 at `8a7446de8`. +Measured on 2026-09-05 at `63a1a410e`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 547 | +| tracked non-test sources scanned | 548 | | engine-shaped types recognised | 58 | | declared objects in the registry | 298 | | same-named calls subtracted as non-engine | 134 | diff --git a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md index 4b34f0ff2c..619328462f 100644 --- a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md +++ b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md @@ -29,17 +29,17 @@ silent, and `node scripts/tenant-audit-census.mjs --write` is the resolution. | Measure | Value | |---|---:| -| Write call sites | 221 | -| Object name statically decidable | 147 | +| Write call sites | 222 | +| Object name statically decidable | 148 | | Object name chosen at run time | 74 | -| Against a tenancy-enabled object | 147 | +| Against a tenancy-enabled object | 148 | | Against an object declaring tenancy off | 0 | -| Threading a tenant context | 137 | +| Threading a tenant context | 138 | | Provably carrying none | 17 | | …and decidably tenancy-enabled | 9 | | Options argument unreadable | 67 | | …and decidably tenancy-enabled | 32 | -| Threading a decidably elevated context | 103 | +| Threading a decidably elevated context | 104 | | Threading a decidably non-elevated context | 0 | | Threading a context of undecidable elevation | 101 | @@ -52,11 +52,11 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-05 at `8a7446de8`. +Measured on 2026-09-05 at `63a1a410e`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 547 | +| tracked non-test sources scanned | 548 | | engine-shaped types recognised | 58 | | declared objects in the registry | 298 | | same-named calls subtracted as non-engine | 134 | @@ -70,7 +70,7 @@ Measured on 2026-09-05 at `8a7446de8`. | `packages/plugins/plugin-approvals/src/approval-service.ts` | `delete` | `sys_approval_approver` | enabled | elevated | 2 | | `packages/plugins/plugin-approvals/src/approval-service.ts` | `insert` | `sys_approval_approver` | enabled | elevated | 2 | | `packages/plugins/plugin-approvals/src/approval-service.ts` | `insert` | `sys_approval_request` | enabled | elevated | 1 | -| `packages/plugins/plugin-approvals/src/approval-service.ts` | `update` | `sys_approval_request` | enabled | elevated | 9 | +| `packages/plugins/plugin-approvals/src/approval-service.ts` | `update` | `sys_approval_request` | enabled | elevated | 10 | | `packages/plugins/plugin-approvals/src/approval-service.ts` | `insert` | `sys_approval_token` | enabled | elevated | 1 | | `packages/plugins/plugin-approvals/src/approval-service.ts` | `update` | `sys_approval_token` | enabled | elevated | 1 | | `packages/plugins/plugin-approvals/src/backfill-platform-row-organizations.ts` | `update` | `objectPlan.object` | undecidable | context, elevation undecidable | 1 | diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index 73fe256215..27466859a3 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -160,6 +160,25 @@ export interface ApprovalResumeSurface { * engine that does not implement it simply gets no pre-flight. */ hasSuspendedRun?(runId: string): Promise; + /** + * [#15389] Where each suspended run is currently parked. Read by + * {@link ApprovalService.continueRestoredRun} to prove a re-armed pause is + * the pause THIS request's recorded outcome was refused on, and not merely + * some live pause on the same run — `hasSuspendedRun` answers a boolean and + * cannot tell the two apart. Which node that is depends on the signal, not + * only on the row: see {@link ApprovalService.expectedPauseNode}. + * + * ⚠️ Declares a method `AutomationEngine` ALREADY implements publicly + * (`listSuspendedRunsDurable`); it widens no engine surface. Durable-first, + * so it sees a pause parked by another replica or before a restart. + * + * Optional, and its absence is FAIL-CLOSED: a caller that cannot prove node + * identity refuses the continuation rather than proceeding on the weaker + * check. The engine degrades a store outage to its in-memory list rather + * than throwing, and that degradation is safe HERE for the same reason — a + * pause it cannot see is a pause this verb will not act on. + */ + listSuspendedRunsDurable?(): Promise>; } /** @@ -248,6 +267,17 @@ const TERMINAL_RUN_STATUSES: ReadonlySet = new Set([ */ const STRANDABLE_REQUEST_STATUSES = ['approved', 'rejected', 'returned'] as const; +/** + * Where {@link ApprovalService.journalStrandedContinuation} keeps the signal a + * stranded resume carried (#15389), inside `node_config_json`. + * + * A private side-channel on an existing JSON column, exactly like + * `__decisionOutputs` beside it — not a new column and not authored node + * config. `parseJson` is a plain parse with no schema, and every reader of that + * JSON reads named fields, so an extra underscore key is inert for all of them. + */ +const STRANDED_CONTINUATION_KEY = '__strandedContinuation'; + /** * The second oracle's verdict: which unrecoverable shape this run is in, or * `undefined` for every run that must NOT be reported (#13909). @@ -318,6 +348,59 @@ function classifyStrandedRunState(run: { status?: string } | null | undefined): */ export type StrandedRunState = 'missing' | 'failed'; +/** + * The continuation an approvals door already issued once, kept so it can be + * issued AGAIN after an operator re-arms the pause it was refused on (#15389). + * + * Stashed under `__strandedContinuation` in `sys_approval_request` + * `node_config_json` — the same private side-channel `__decisionOutputs` + * already uses, rather than a new column: this is recovery bookkeeping for a + * rare failure path, not part of the authored node config, and every reader of + * that JSON reads named fields. + */ +export interface StrandedContinuationSignal { + /** The `approve` / `reject` / `revise` / `resubmit` edge the outcome walks. */ + branchLabel?: string; + /** The exact flow-variable payload the original resume carried. */ + output?: Record; + /** Outcome label for the {@link strandedDecisionFailure} envelope. */ + decision: string; + /** How the original door described the outcome in prose, e.g. `the reject decision`. */ + what: string; +} + +/** + * Outcome of {@link ApprovalService.continueRestoredRun} (#15389). + * + * ⚠️ Deliberately its own shape rather than a reuse of `ApprovalDecisionResult`: + * that contract is the subject of an OPEN maintainer ruling on #15556, and this + * card must not pre-empt it. Nothing here changes what `decide` answers. + */ +export interface ApprovalContinuationResult { + /** True when the restored pause was consumed and the flow moved on. */ + resumed: boolean; + /** The run this continued — the one the request has always named. */ + runId: string; + /** The outcome that was replayed, exactly as it was first recorded. */ + decision: string; + /** The edge it walked. */ + branchLabel?: string; + /** + * Where the replayed signal came from, so a caller can tell an EXACT replay + * from an inferred one: + * + * - `journal` — the failing door stashed the literal signal it sent, so this + * is a byte-for-byte re-issue. + * - `reconstructed` — no stash (the strand predates that journalling, or its + * write failed), so the signal was rebuilt from the row's recorded outcome. + * Exact for every status this verb accepts; the one shape it CANNOT rebuild + * is refused rather than guessed (see the method's `AMBIGUOUS_RECORDED_OUTCOME`). + */ + source: 'journal' | 'reconstructed'; + /** Set only on the tolerated non-failure: a concurrent resume already had it. */ + resumeError?: string; +} + /** * One terminal request whose owning flow run is unrecoverable (#4469) — the * decision was recorded and the flow never moved. Reporting shape only: the @@ -2920,6 +3003,18 @@ export class ApprovalService implements IApprovalService { this.logger?.error?.('[approvals] resume failed — the run is stranded', { request: requestId, run: runId, outcome: what, error: reason, status, repairable, }); + // #15389: keep the continuation this resume was carrying, so + // `continueRestoredRun` can re-issue it EXACTLY once an operator re-arms + // the pause. Only on the repairable exit — the one whose suspension the + // engine journalled a snapshot for; on any other failure there is no + // pause to put back and a stash would be recovery bookkeeping for a + // repair that cannot happen. Awaited (the row must carry it before the + // caller can act on the throw) but never allowed to replace the throw. + if (repairable) { + await this.journalStrandedContinuation(requestId, { + branchLabel: signal.branchLabel, output: signal.output, decision, what, + }); + } throw strandedDecisionFailure( `RESUME_FAILED: ${what} was recorded on request ${requestId}, but its flow run '${runId}' ` + `could not be resumed and is now stranded: ${reason}`, @@ -3124,6 +3219,17 @@ export class ApprovalService implements IApprovalService { this.logger?.error?.('[approvals] resume after recall failed — the run may be stranded', { request: requestId, run: runId, error: resumeError, }); + // #15389: recall resumes directly rather than through + // `resumeRecordedOutcome`, so its stranded exit needs the same stash + // — otherwise a recalled run is the one outcome whose re-issue would + // have to be rebuilt from the row instead of replayed. + if (ApprovalService.resumeStatusOf(err) === 'stranded') { + await this.journalStrandedContinuation(requestId, { + branchLabel: APPROVAL_BRANCH_LABELS.reject, + output: { decision: 'recall', requestId }, + decision: 'recall', what: 'the recall', + }); + } } } } @@ -4219,6 +4325,482 @@ export class ApprovalService implements IApprovalService { return { scanned: rows.length, stranded, undetermined }; } + /** + * Stash the continuation a door just failed to deliver, so it can be issued + * again after the pause is re-armed (#15389). + * + * `AutomationEngine.restoreConsumedSuspension` puts a stranded approval run + * back on its pause and tells the operator to *re-issue the continuation* — + * but for an `approval` node the only issuers are this service's doors, and + * every one of them guards on a `pending` request that the stranding call + * itself just made terminal. Re-opening the row is excluded (it would let a + * decided request be decided again), so what is kept instead is the SIGNAL: + * the exact `branchLabel` + `output` the failed resume carried. + * + * ⚠️ Best-effort by construction, and it must stay that way: the decision is + * already durable and its caller is already owed a `RESUME_FAILED` throw. A + * failure to write recovery bookkeeping must not replace that throw with a + * storage error — {@link ApprovalService.continueRestoredRun} rebuilds the + * signal from the row when the stash is absent, so this failing costs + * fidelity on one shape, not the repair path. + */ + private async journalStrandedContinuation( + requestId: string, + signal: StrandedContinuationSignal, + ): Promise { + try { + const rows = await this.engine.find('sys_approval_request', { + where: { id: requestId }, limit: 1, context: SYSTEM_CTX, + }); + const raw: any = Array.isArray(rows) ? rows[0] : null; + if (!raw) return; + const config = parseJson>(raw.node_config_json, {}); + await this.engine.update('sys_approval_request', { + id: requestId, + node_config_json: JSON.stringify({ ...config, [STRANDED_CONTINUATION_KEY]: signal }), + }, { context: SYSTEM_CTX }); + } catch (err: any) { + this.logger?.warn?.( + '[approvals] could not journal the stranded continuation — the repair path falls back to rebuilding it from the row', + { request: requestId, error: err?.message ?? String(err) }, + ); + } + } + + /** + * The continuation to re-issue for a request whose recorded outcome stranded + * its run — the journalled one when there is one, otherwise rebuilt from the + * row (#15389). + * + * ## Why a rebuild path exists at all + * + * The journal only covers runs stranded by a build that HAS it. The card is + * explicitly about *"the runs already in this state"*, and one of those can + * still be restored whenever the durable run-history row carried its + * suspension snapshot — so a repair verb that only served future strands + * would miss the population the card was filed for. + * + * ## Which statuses it rebuilds, which it discriminates, and which it refuses + * + * ⛔ A status is NOT the same thing as a continuation. Three of the four + * terminal statuses have more than one writer or more than one issuer, so + * "one status, one signal" is false and is not what this relies on. Each row + * below states its own population and its own discriminator: + * + * | status | writers / issuers | rebuilt as | how it is decided | + * |---|---|---|---| + * | `approved` | 1 (`decide`; escalation auto-approve routes through it) | `approve` | unambiguous | + * | `rejected` | 2 (`decide`; ADR-0044 revision-limit auto-reject) | `reject`, or REFUSED | a `revise` action row means the auto-reject arm is possible | + * | `returned` | 1 writer, 2 issuers (`sendBack` → `revise`; a later `resubmit` → `resubmit`, writing no status) | `resubmit` or `revise` | a `resubmit` action row, whose sole writer is `resubmit` | + * | `recalled` | 2 writers, 3 behaviours, 2 issuing NO continuation | REFUSED | nothing on the row distinguishes them | + * + * ⚠️ **Both refusals are deliberate and neither is best-effort.** The failure + * mode of a wrong rebuild is a flow advanced down a branch nobody chose — + * strictly worse than the dead end this verb exists to open. Where the signal + * cannot be proved, this refuses and names what the operator can do instead; + * the journal is what makes both shapes recoverable going forward. + * + * ⛔ `pending` and `cancelled` are refused outright: neither names a recorded + * outcome to replay. A `pending` request's continuation is an ordinary + * decision through the front door, which is exactly the guard this verb + * exists to avoid weakening. + */ + private async resolveRecordedContinuation( + raw: any, + requestId: string, + ): Promise<{ signal: StrandedContinuationSignal; source: 'journal' | 'reconstructed' }> { + const config = parseJson>(raw.node_config_json, {}); + const stashed = config?.[STRANDED_CONTINUATION_KEY]; + if (stashed && typeof stashed === 'object' && typeof stashed.decision === 'string') { + return { signal: stashed as StrandedContinuationSignal, source: 'journal' }; + } + + const outputs: Record = { ...(config?.__decisionOutputs ?? {}) }; + const status = String(raw.status ?? ''); + + if (status === 'approved' || status === 'rejected') { + if (status === 'rejected') { + // The ADR-0044 auto-rejection is the second writer of this status, and + // its `revise` row is the only thing that tells them apart. + const priorRevise = await this.engine.find('sys_approval_action', { + where: { request_id: requestId, action: 'revise' }, limit: 1, context: SYSTEM_CTX, + }); + if (Array.isArray(priorRevise) && priorRevise.length) { + throw new Error( + `INVALID_STATE: request ${requestId} is 'rejected' and also carries a 'revise' action, so this ` + + `service cannot tell a decided rejection from an ADR-0044 revision-limit auto-rejection — and the ` + + `two resume the same edge with different flow output (\`autoRejected\`). Refusing to guess: replay ` + + `it by hand with the signal the flow expects, or cancel the run.`, + ); + } + } + const decision = status === 'approved' ? 'approve' : 'reject'; + return { + source: 'reconstructed', + signal: { + branchLabel: status === 'approved' + ? APPROVAL_BRANCH_LABELS.approve + : APPROVAL_BRANCH_LABELS.reject, + output: { ...outputs, decision, requestId }, + decision, + what: `the ${decision} decision`, + }, + }; + } + + if (status === 'returned') { + // ⚠️ ONE status writer, TWO continuation issuers. `sendBack` writes + // `returned` and resumes down `revise`; a later `resubmit` on that same + // row resumes down `resubmit` and writes NO status, so the row still + // reads `returned` when its resume is the one that stranded. Rebuilding + // both as `revise` sends a stranded resubmit down the wrong edge — and it + // does not even fail loudly, because the engine's unmatched-label + // fallback (#4414) warns and evaluates every out-edge, so the flow + // proceeds with `{decision:'revise'}` where `{resubmitted:true}` was owed. + // + // The discriminator is exact and structural: `action: 'resubmit'` has + // exactly ONE writer in this file (`resubmit`), it is inserted before + // that resume, and a resubmit opens the next round as a NEW row — so at + // most one such action row exists per request, and its presence means + // the last continuation this row issued was the resubmit. + const resubmitted = await this.engine.find('sys_approval_action', { + where: { request_id: requestId, action: 'resubmit' }, limit: 1, context: SYSTEM_CTX, + }); + if (Array.isArray(resubmitted) && resubmitted.length) { + return { + source: 'reconstructed', + signal: { + branchLabel: APPROVAL_BRANCH_LABELS.resubmit, + output: { resubmitted: true, requestId }, + decision: 'resubmit', + what: 'the resubmit', + }, + }; + } + return { + source: 'reconstructed', + signal: { + branchLabel: APPROVAL_BRANCH_LABELS.revise, + output: { decision: 'revise', requestId }, + decision: 'revise', + what: 'the send-back', + }, + }; + } + + // ⛔ `recalled` is REFUSED on the rebuild path, deliberately, and this is + // the one status where this verb declines rather than reconstructs. + // + // Two status writers and THREE continuation behaviours, two of which issue + // no continuation at all: `recall` on a `pending` request resumes down + // `reject`; `recall` inside a revision window calls `cancelRun` instead + // (there is no reject edge at a revise-window node); and the dead-run sweep + // marks a request `recalled` for a run that is already gone. Nothing on the + // row distinguishes which of the three wrote it, and the two silent ones + // have no branch that SHOULD be walked — so any rebuild here is a guess + // whose failure mode is opening a fresh pending round on a request somebody + // deliberately withdrew, which was measured on the pre-guard build. + // + // A recall stranded by a build that HAS the journal replays exactly, from + // the stash above; this refusal is the pre-journal population only, and it + // names what the operator can do instead. + if (status === 'recalled') { + throw new Error( + `INVALID_STATE: request ${requestId} is 'recalled' and carries no journalled continuation, so the ` + + `signal cannot be rebuilt: a recall reaches this state three ways (resumed down 'reject', ` + + `terminally cancelled inside a revision window, or swept as a dead run) and two of them issue no ` + + `continuation at all — replaying the wrong one would re-open a request that was deliberately ` + + `withdrawn. Refusing to guess: cancel the run with the engine's cancelRun('${raw.flow_run_id}') if ` + + `the withdrawal should stand, or resume it by hand with the signal the flow expects.`, + ); + } + + throw new Error( + `INVALID_STATE: request is ${status || 'unknown'} — only a request whose recorded outcome already ` + + `resumed its run can have that continuation re-issued (approved, rejected, returned, recalled)`, + ); + } + + /** + * WHERE the pause a recorded continuation was refused on actually sits + * (#15389) — the expected node guard 3 compares the run's parked node against. + * + * ⚠️ This is signal-aware, and that is the whole point of it. "This request's + * own node" is the right answer for three of the four signals and the WRONG + * answer for the fourth: + * + * | signal | issued from | why | + * |---|---|---| + * | `approve` / `reject` | the request's own approval node | the decision is taken at the pause it gates | + * | `revise` (send-back) | the request's own approval node | send-back resumes that same pause down the `revise` edge | + * | `recall` | the request's own approval node | recall-on-pending resumes that same pause down `reject` | + * | `resubmit` | the **revise window** the request's `revise` edge leads to | by construction: a resubmit is only reachable AFTER a send-back moved the run there, and it resumes THAT pause down the `resubmit` back-edge | + * + * Measured before this existed: a `returned` row whose resubmit stranded was + * refused by guard 3 on both the journal and the rebuild paths — the pause + * re-armed at the revise window while the row's `flow_node_id` still read the + * approval node — and the refusal told the operator the pause was not this + * request's when it was exactly this request's. A refusal may ship; a refusal + * that names a cause the code did not take may not. + * + * ⛔ It stays FAIL-CLOSED: the revise window is derived from the flow + * definition the same way {@link ApprovalService.assertReviseEdge} derives it + * — a `revise` out-edge of this request's node into a node the flow declares + * as `{@link APPROVAL_REVISE_NODE_TYPE}`, which is the pause only this service + * can continue. No engine, no flow, no such edge, or more than one candidate + * ⇒ refuse. It needs no automation surface `assertReviseEdge` did not already + * use (`getFlow`), and no engine change. + * + * ⚠️ It does not widen what guard 3 admits beyond that one signal: for every + * other decision the answer is byte-identical to the row's own node, so the + * recall-in-revise-window shape (row `recalled`, run at the revise window) + * stays refused — its journalled signal is `recall`, not `resubmit`. + */ + private async expectedPauseNode( + raw: any, + signal: StrandedContinuationSignal, + requestId: string, + runId: string, + ): Promise<{ nodeId: string; describe: string }> { + const ownNode: string | null = raw.flow_node_id ?? raw.current_step ?? null; + if (!ownNode) { + throw new Error( + `INVALID_STATE: request ${requestId} records no approval node, so the pause on run '${runId}' ` + + `cannot be proved to be the one ${signal.what} was refused on — refusing rather than resuming a ` + + `pause that may belong to another node`, + ); + } + if (signal.decision !== 'resubmit') { + return { nodeId: ownNode, describe: `its own approval node '${ownNode}'` }; + } + + const processName = String(raw.process_name ?? ''); + const flowName = processName.startsWith('flow:') ? processName.slice('flow:'.length) : ''; + if (!flowName || typeof this.automation?.getFlow !== 'function') { + throw new Error( + `INVALID_STATE: ${signal.what} on request ${requestId} was issued from the revise window that ` + + `approval node '${ownNode}' sends back to, and this service cannot read the owning flow ` + + `definition to say which node that is — refusing, because continuing a pause it cannot identify ` + + `advances a step nobody decided`, + ); + } + const flow: any = await this.automation.getFlow(flowName); + const nodeTypeById = new Map( + (Array.isArray(flow?.nodes) ? flow.nodes : []) + .filter((n: any) => typeof n?.id === 'string') + .map((n: any) => [n.id as string, typeof n.type === 'string' ? n.type : '']), + ); + const windows = Array.from(new Set( + (Array.isArray(flow?.edges) ? flow.edges : []) + .filter((e: any) => e?.source === ownNode && e?.label === APPROVAL_BRANCH_LABELS.revise) + .map((e: any) => (typeof e?.target === 'string' ? e.target : '')) + .filter((t: string) => t && nodeTypeById.get(t) === APPROVAL_REVISE_NODE_TYPE), + )) as string[]; + if (windows.length !== 1) { + throw new Error( + `INVALID_STATE: ${signal.what} on request ${requestId} was issued from the revise window that ` + + `approval node '${ownNode}' sends back to, and flow '${flowName}' declares ` + + `${windows.length === 0 ? 'no such window' : `${windows.length} of them (${windows.join(', ')})`} ` + + `— refusing, because a pause this service cannot identify must not be continued`, + ); + } + return { + nodeId: windows[0], + describe: `the revise window '${windows[0]}' that its approval node '${ownNode}' sends back to`, + }; + } + + /** + * Re-issue the continuation for a run an operator has re-armed with + * `AutomationEngine.restoreConsumedSuspension` — the missing half of that + * repair verb, for approvals (#15389). + * + * ## The dead end this exits + * + * A decision whose downstream node throws strands the run: the suspension is + * consumed, the decision is durable, and the caller gets `RESUME_FAILED` + * carrying `repairable: true`. `restoreConsumedSuspension` then genuinely + * re-arms the pause — measured `restored: true`, `hasSuspendedRun` back to + * `true` — and its own reason string tells the operator to *re-issue the + * continuation*. For an `approval` node there was then nobody who could: + * + * - `decide` / `recall` / `sendBack` / `resubmit` all guard on a `pending` + * request, and the row is terminal — written by the very call that + * stranded the run; + * - the generic `engine.resume` refuses, because the `approval` node + * declares `resumeAuthority: 'service'` and the #3801 gate turns away any + * resume that is not the tail of a decision this service authorized. + * + * So the only verb left was `cancelRun`, which discards the branch's + * downstream work. Measured on the real engine and the real door: the + * restored pause IS resumable, and a `resumeAuthority`-marked resume walks + * the reject branch to completion. Nothing was missing in the engine — what + * was missing was an ISSUER on this side. This is that issuer. + * + * ## What it deliberately does NOT do + * + * ⛔ It does not re-open, re-decide, or rewrite the request row: all four + * `pending` guards stay exactly as they are, and no status, mirror field or + * audit row is written. A person decided this once; this replays what they + * decided onto the pause that was put back, and replays nothing else. + * ⛔ It does not relax `resumeAuthority: 'service'` — the resume goes through + * {@link ApprovalService.serviceResume} like every other, so the marker is + * still stamped in exactly one place. + * ⛔ It grants no capability that in-process code did not already have: + * `RESUME_AUTHORITY_SERVICE` is importable by anything in the host, so the + * raw form of this call was always available. What this adds is the GUARDED + * form, and the guards are the substance of it — three, each with its own + * reverse-control pin, because the raw marker is not a guard and an + * unguarded repair verb advances flows nobody decided: + * + * 1. {@link assertLatestForRun} — this request is still the newest on its + * run, so a superseded row cannot drive a later round or a later node; + * 2. `hasSuspendedRun` — a pause exists at all (strict: an unreadable store + * throws rather than reading as "not suspended"); + * 3. node identity — that pause is parked where THIS request's recorded + * outcome was issued from: its own approval node for `approve`, + * `reject`, `revise` and `recall`, and — for a `resubmit`, which is only + * reachable from a revise window — the `approval_revise` node its own + * `revise` edge leads to. {@link ApprovalService.expectedPauseNode} + * derives it, fail-closed. + * + * Guard 3 is not redundant with guard 2: existence is not identity, and a + * boolean cannot tell this request's re-armed pause from any other live + * pause on the same run. + * + * ## Posture, and why it takes no `ExecutionContext` + * + * Deliberately shaped like the engine verb it completes: an in-process + * operator repair, reachable from a host or a console script, with no REST + * route and no entry in the spec `ApprovalService` contract — exactly as + * `restoreConsumedSuspension` is a class method on `AutomationEngine` and + * appears in no contract. It authorizes nothing new: the decision it replays + * was authorized and recorded when it was made, and re-authorizing it here + * against a present-day actor would be a different and wrong question (the + * original approver may be long gone). `requestedBy` / `reason` ride the log + * for the same reason they do on the restore. + * + * @returns what was replayed and whether the run moved — never a silent + * `false`. A resume that fails again throws the same `RESUME_FAILED` + * envelope the original decision did, `repairable` and all, so a second + * restore-and-continue is possible. + */ + async continueRestoredRun( + requestId: string, + options?: { requestedBy?: string; reason?: string }, + ): Promise { + if (!requestId) throw new Error('VALIDATION_FAILED: requestId is required'); + + const rows = await this.engine.find('sys_approval_request', { + where: { id: requestId }, limit: 1, context: SYSTEM_CTX, + }); + const raw: any = Array.isArray(rows) ? rows[0] : null; + if (!raw) throw new Error(`REQUEST_NOT_FOUND: ${requestId}`); + + const runId: string | null = raw.flow_run_id ?? null; + if (!runId) { + throw new Error( + `INVALID_STATE: request ${requestId} names no flow run — there is no continuation to re-issue`, + ); + } + + // ── GUARD 1. This request must still be the newest on its run. + // A run outlives any one request: later rounds and later approval nodes + // each mint their own. Without this, a superseded row's continuation walks + // whatever pause the run is sitting on NOW — measured advancing a later + // node's pending approval with no decision recorded on it, and orphaning a + // next round. Same guard `recall`-in-revise-window and `resubmit` already + // use, for the same reason. + await this.assertLatestForRun(raw); + + // The signal, before the store is asked anything — a request this verb + // cannot serve is refused without a suspension read. + const { signal, source } = await this.resolveRecordedContinuation(raw, requestId); + + // ── GUARD 2. A pause must exist at all. `hasSuspendedRun` is STRICT — it + // throws on an unreadable store — and that throw is deliberately not + // caught: a caller about to resume must not read an outage as "not + // suspended". + if (typeof this.automation?.hasSuspendedRun === 'function') { + const parked = await this.automation.hasSuspendedRun(runId); + if (!parked) { + throw new Error( + `INVALID_STATE: run '${runId}' behind request ${requestId} is not suspended, so there is no ` + + `re-armed pause to continue — restore it first with the automation engine's ` + + `restoreConsumedSuspension('${runId}'), which is what re-arms a consumed approval suspension`, + ); + } + } + + // ── GUARD 3. The pause must be the one THIS SIGNAL was refused on. + // Guards 1 and 2 together still admit a run parked somewhere this request's + // outcome never gated: a recall taken during a revision window leaves the + // row `recalled` while the run sits at the revise-window node, and issuing + // this request's `reject` there was measured opening a NEW pending round + // for a recalled request. Existence is not identity, so identity is checked. + // + // ⚠️ The expected node is SIGNAL-AWARE, not simply the row's own node — see + // {@link ApprovalService.expectedPauseNode}. A `resubmit` is issued from the + // revise window by construction, so comparing it against the row's approval + // node refused the one population the journal was built to serve, with a + // message that named a cause the code had not taken. + // + // FAIL-CLOSED in every direction: no reader, no listing, no matching entry, + // an entry at another node, or an expected node this service cannot derive + // — all refuse. The engine degrades a store outage to its in-memory list + // instead of throwing, so a pause this cannot see is a pause this verb + // declines to act on. A false refusal costs the operator a retry; a false + // admission advances a flow nobody decided. + const expected = await this.expectedPauseNode(raw, signal, requestId, runId); + if (typeof this.automation?.listSuspendedRunsDurable !== 'function') { + throw new Error( + `INVALID_STATE: this automation engine cannot report WHERE run '${runId}' is parked ` + + `(no listSuspendedRunsDurable), so the pause cannot be proved to be the one ${signal.what} on ` + + `request ${requestId} was refused on — refusing, because continuing the wrong pause advances a ` + + `flow with no decision behind it`, + ); + } + const parkedAt = (await this.automation.listSuspendedRunsDurable()) + .find(r => String(r.runId) === String(runId))?.nodeId; + if (parkedAt !== expected.nodeId) { + throw new Error( + `INVALID_STATE: run '${runId}' is parked at ` + + `${parkedAt ? `node '${parkedAt}'` : 'no node this engine can see'}, but ${signal.what} on ` + + `request ${requestId} was issued from ${expected.describe} — this re-armed pause is not the one ` + + `that outcome was refused on, and continuing it would advance a step nobody decided`, + ); + } + + this.logger?.warn?.( + '[approvals] re-issuing the continuation for a restored approval suspension', + { + request: requestId, run: runId, decision: signal.decision, + branchLabel: signal.branchLabel, source, + requestedBy: options?.requestedBy ?? 'not recorded', + reason: options?.reason ?? 'not recorded', + }, + ); + + // Reuses the door every other recorded outcome resumes through: the same + // `RESUME_IN_PROGRESS` tolerance, the same stranded envelope with the + // engine's own `repairable` discriminator, and the same re-journalling — + // so a continuation that strands AGAIN is repairable exactly like the first. + const outcome = await this.resumeRecordedOutcome( + runId, requestId, signal.what, + { branchLabel: signal.branchLabel, output: signal.output }, + signal.decision, + ); + + return { + resumed: outcome.resumed, + runId, + decision: signal.decision, + branchLabel: signal.branchLabel, + source, + ...(outcome.resumeError ? { resumeError: outcome.resumeError } : {}), + }; + } + async releaseDeadRunRequests(): Promise<{ scanned: number; released: number }> { // No liveness oracle → no basis to declare anything dead. if (typeof this.automation?.getRun !== 'function') return { scanned: 0, released: 0 }; diff --git a/packages/plugins/plugin-approvals/src/index.ts b/packages/plugins/plugin-approvals/src/index.ts index c440249004..bc175b67bf 100644 --- a/packages/plugins/plugin-approvals/src/index.ts +++ b/packages/plugins/plugin-approvals/src/index.ts @@ -27,6 +27,9 @@ export { type StrandedApprovalRequest, // #13909 — which unrecoverable shape a reported row is in. type StrandedRunState, + // #15389 — re-issuing a stranded continuation onto a restored suspension. + type StrandedContinuationSignal, + type ApprovalContinuationResult, } from './approval-service.js'; export { ApprovalsServicePlugin, diff --git a/packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts b/packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts new file mode 100644 index 0000000000..3e9ff71758 --- /dev/null +++ b/packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts @@ -0,0 +1,740 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * A restored approval suspension can be DECIDED again, not only cancelled + * (#15389). + * + * ## The reported dead end + * + * `AutomationEngine.restoreConsumedSuspension` is the platform's repair verb + * for a run stranded mid-resume: it puts the consumed pause back and answers + * *"the run is resumable again; re-issue the continuation"*. For an `approval` + * suspension nobody could re-issue it: + * + * - every approvals door that stamps the resume marker — `decide`, `recall`, + * `sendBack`, `resubmit` — guards on a `pending` request, and the row is + * terminal, written by the very call that stranded the run; + * - the generic engine door refuses, because the `approval` node declares + * `resumeAuthority: 'service'` (#3801). + * + * So the only remaining verb was `cancelRun`, which discards the branch's + * downstream work. A repair that advertises a follow-up the platform then + * refuses is worse than one that declines up front. + * + * ## The mechanism, named + * + * ⭐ The restored suspension lacks NOTHING. PIN 2 measures a `resumeAuthority` + * -marked resume walking the restored pause to completion — the engine side is + * whole. What was missing is an ISSUER on the approvals side: the authority to + * stamp the marker lives only behind those four `pending` guards. That is why + * the repair is a new approvals verb and not an engine change. + * + * ## What the fix must not do + * + * ⛔ Not re-open the request row (that would let a decided request be decided + * again — a permission-boundary change triage reserved for a ruling). ⛔ Not + * relax `resumeAuthority: 'service'`. ⛔ Not touch `ApprovalDecisionResult`, + * whose shape is under an open ruling on #15556. PIN 3 asserts all four + * `pending` guards still stand and the row is untouched. + */ + +import { describe, it, expect, beforeEach } from 'vitest'; +import { AutomationEngine, InMemorySuspendedRunStore } from '@objectstack/service-automation'; +// [#4550] The engine double below routes its write verbs through ObjectQL's OWN +// dispatch predicates rather than a hand-mirrored copy. +import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/objectql'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; +import { APPROVAL_REVISE_NODE_TYPE } from '@objectstack/spec/automation'; +import { strandedDecisionDetails } from '@objectstack/types'; +import { ApprovalService } from './approval-service.js'; +import { registerApprovalNode } from './approval-node.js'; + +const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as any; +const noopLogger = { info() {}, warn() {}, error() {}, debug() {} }; + +/** In-memory ObjectQL stand-in for the approvals tables. */ +function makeFakeEngine() { + const tables = new Map(); + const rows = (o: string) => (tables.get(o) ?? (tables.set(o, []), tables.get(o)!)); + // ⭐ The lever that makes a REAL strand reachable from a test: fail the very + // next insert into one table, once. The approval node's executor opens the + // next round by inserting a `sys_approval_request`, so failing that insert + // strands the resume the same way the card's own reject-branch throw does — + // `RESUME_FAILED` with `repairable: true`, the suspension already consumed. + // Set to a table name; the first insert into it throws and clears the lever. + let failNextInsertOn: string | undefined; + const matches = (row: any, where: any) => Object.entries(where ?? {}).every(([k, v]) => { + if (k.startsWith('$')) throw new Error(`fake engine: unsupported filter operator ${k}`); + if (v && typeof v === 'object' && '$in' in (v as any)) return (v as any).$in.includes(row[k]); + if (v && typeof v === 'object' && '$ne' in (v as any)) return row[k] !== (v as any).$ne; + return row[k] === v; + }); + return { + tables, + set failNextInsert(object: string | undefined) { failNextInsertOn = object; }, + get failNextInsert() { return failNextInsertOn; }, + async find(object: string, opts: any = {}) { + const where = opts.where ?? opts.filter ?? {}; + const out = rows(object).filter(r => matches(r, where)); + // ⚠️ `orderBy` is honoured, and that is load-bearing rather than polish: + // `assertLatestForRun` selects the newest request with + // `orderBy [{field:'created_at', order:'desc'}], limit 1`. A double that + // ignored it returned the OLDEST row, so the guard passed on every input + // and a pin naming it would have measured nothing — the phantom-check + // shape. SortNode's key is `order`, not `direction` (spec/data/query.zod.ts). + if (Array.isArray(opts.orderBy)) { + for (const sort of [...opts.orderBy].reverse()) { + const field = sort?.field; + if (!field) continue; + const dir = sort?.order === 'desc' ? -1 : 1; + out.sort((a, b) => (a[field] < b[field] ? -1 : a[field] > b[field] ? 1 : 0) * dir); + } + } + // The caller's bound is honoured by PRESENCE, never truthiness. + const start = opts.offset ?? 0; + const page = typeof opts.limit === 'number' ? out.slice(start, start + opts.limit) : out.slice(start); + return page.map(r => ({ ...r })); + }, + async insert(object: string, data: any) { + if (failNextInsertOn === object) { + failNextInsertOn = undefined; + throw new Error(`injected one-shot insert failure on ${object}`); + } + rows(object).push({ ...data }); return { ...data }; + }, + async update(object: string, data: any, options?: any) { + const dispatch = assertEngineUpdateDispatch(data, options); + const table = rows(object); + if (dispatch.kind === 'multi') { + let n = 0; + for (let i = 0; i < table.length; i++) { + if (matches(table[i], options?.where)) { table[i] = { ...table[i], ...data }; n++; } + } + return { updated: n }; + } + const i = table.findIndex(r => r.id === dispatch.id); + if (i >= 0) table[i] = { ...table[i], ...data }; + return i >= 0 ? { ...table[i] } : null; + }, + async delete(object: string, options?: any) { + const dispatch = assertEngineDeleteDispatch(options); + const table = rows(object); + if (dispatch.kind === 'multi') { + const survivors = table.filter(r => !matches(r, options?.where)); + const deleted = table.length - survivors.length; + table.splice(0, table.length, ...survivors); + return { deleted }; + } + const i = table.findIndex(r => r.id === dispatch.id); + if (i >= 0) table.splice(i, 1); + return { id: dispatch.id }; + }, + }; +} + +const DEAL_APPROVAL = { + name: 'deal_approval', label: 'Deal Approval', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'approve_step', type: 'approval', label: 'Manager Approval', + config: { approvers: [{ type: 'user', value: 'u1' }] } }, + { id: 'on_approved', type: 'mark', label: 'Approved' }, + { id: 'mark_rejected', type: 'mark', label: 'Rejected' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'approve_step' }, + { id: 'e2', source: 'approve_step', target: 'on_approved', label: 'approve' }, + { id: 'e3', source: 'approve_step', target: 'mark_rejected', label: 'reject' }, + { id: 'e4', source: 'on_approved', target: 'end' }, + { id: 'e5', source: 'mark_rejected', target: 'end' }, + ], +}; + +/** The card's own failing node text: the reject branch writes to a gone record. */ +const DOWNSTREAM_FAILURE = + 'update_record(crm_leave_request) failed: Record 9SEmlyRfw8D9-J7Z not found'; + +/** + * TWO approval nodes in sequence. The run parks at `a2` after `a1` is decided, + * so `a1`'s terminal request coexists with a live pause that is NOT its own — + * the shape that separates "a pause exists on this run" from "this request's + * pause is back". + */ +const TWO_STEP = { + name: 'two_step', label: 'Two Step', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'a1', type: 'approval', label: 'Manager', config: { approvers: [{ type: 'user', value: 'u1' }] } }, + { id: 'a2', type: 'approval', label: 'Finance', config: { approvers: [{ type: 'user', value: 'u2' }] } }, + { id: 'done', type: 'mark', label: 'Done' }, + { id: 'nope', type: 'mark', label: 'Nope' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'a1' }, + { id: 'e2', source: 'a1', target: 'a2', label: 'approve' }, + { id: 'e3', source: 'a1', target: 'nope', label: 'reject' }, + { id: 'e4', source: 'a2', target: 'done', label: 'approve' }, + { id: 'e5', source: 'a2', target: 'nope', label: 'reject' }, + { id: 'e6', source: 'done', target: 'end' }, + { id: 'e7', source: 'nope', target: 'end' }, + ], +}; + +/** ADR-0044 revise window: send-back parks the run at a node the approval request never gated. */ +const REVISE_FLOW = { + name: 'revise_flow', label: 'Revise Flow', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'review', type: 'approval', label: 'Review', config: { approvers: [{ type: 'user', value: 'u1' }] } }, + { id: 'wait_revision', type: APPROVAL_REVISE_NODE_TYPE, label: 'Awaiting Revision' }, + { id: 'on_approved', type: 'mark', label: 'Approved' }, + { id: 'on_rejected', type: 'mark', label: 'Rejected' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'review' }, + { id: 'e2', source: 'review', target: 'on_approved', label: 'approve' }, + { id: 'e3', source: 'review', target: 'on_rejected', label: 'reject' }, + { id: 'e4', source: 'review', target: 'wait_revision', label: 'revise' }, + { id: 'e5', source: 'wait_revision', target: 'review', label: 'resubmit', type: 'back' }, + { id: 'e6', source: 'on_approved', target: 'end' }, + { id: 'e7', source: 'on_rejected', target: 'end' }, + ], +}; + +describe('#15389 — a restored approval suspension has an issuer again', () => { + let data: ReturnType; + let service: ApprovalService; + let marks: string[]; + let rejectBranchThrows: string | undefined; + + /** One live process: real engine, real approval node, real approvals service. */ + function boot() { + const automation = new AutomationEngine(noopLogger as any, new InMemorySuspendedRunStore()); + registerApprovalNode(automation, service, noopLogger as any); + automation.registerNodeExecutor({ + type: 'mark', + async execute(node: any) { + if (node.id === 'mark_rejected' && rejectBranchThrows) throw new Error(rejectBranchThrows); + marks.push(node.id); + return { success: true }; + }, + } as never); + automation.registerFlow('deal_approval', DEAL_APPROVAL as never); + automation.registerFlow('two_step', TWO_STEP as never); + automation.registerFlow('revise_flow', REVISE_FLOW as never); + service.attachAutomation(automation); + return automation; + } + + beforeEach(() => { + marks = []; rejectBranchThrows = undefined; + data = makeFakeEngine(); + service = new ApprovalService({ engine: data as any, logger: noopLogger }); + }); + + async function park(automation: AutomationEngine) { + await automation.execute('deal_approval', { + object: 'crm_deal', record: { id: 'd1', amount: 100 }, userId: 'submitter', + } as never); + return (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + } + + const rowOf = async (id: string) => + (await data.find('sys_approval_request', { where: { id } }))[0]; + + /** Strand a reject decision, then re-arm its pause. Returns the request row. */ + async function strandThenRestore(automation: AutomationEngine) { + rejectBranchThrows = DOWNSTREAM_FAILURE; + const req = await park(automation); + const err = await service + .decide(req.id, { decision: 'reject', actorId: 'u1', comment: 'no' }, SYSTEM_CTX) + .then(() => null, (e: Error) => e); + // The strand the card starts from, and the repairable stamp #13807 added. + expect(err?.message).toMatch(/^RESUME_FAILED/); + expect(strandedDecisionDetails(err)?.repairable).toBe(true); + expect(await automation.hasSuspendedRun(req.flow_run_id)).toBe(false); + + const restored = await automation.restoreConsumedSuspension( + req.flow_run_id, { requestedBy: 'ops', reason: 'record recreated' }, + ); + expect(restored.restored, 'the re-arm itself works — this card is about what follows').toBe(true); + expect(restored.reason).toMatch(/re-issue the continuation/); + expect(await automation.hasSuspendedRun(req.flow_run_id), 'the pause really is back').toBe(true); + return req; + } + + it('PIN 1 — the reported dead end: restored, and every issuer but cancelRun refuses', async () => { + const automation = boot(); + const req = await strandThenRestore(automation); + const runId = req.flow_run_id; + + // ── The approvals doors: all guard on `pending`, and the row is terminal. + const decideAgain = await service + .decide(req.id, { decision: 'reject', actorId: 'u1' }, SYSTEM_CTX) + .then(() => null, (e: Error) => e); + expect(decideAgain?.message).toBe('INVALID_STATE: request is rejected'); + + const recallIt = await service + .recall(req.id, { actorId: 'submitter' }, SYSTEM_CTX) + .then(() => null, (e: Error) => e); + expect(recallIt, 'recall is refused too — it is an issuer of the same kind').toBeTruthy(); + expect(recallIt?.message).toMatch(/INVALID_STATE|FORBIDDEN/); + + // ── The generic engine door: refused by the #3801 `resumeAuthority` gate. + const generic: any = await automation.resume(runId, { branchLabel: 'reject' } as never); + expect(generic.success).toBe(false); + expect(generic.code).toBe('PERMISSION_DENIED'); + expect(generic.error).toMatch(/only its owning service may resume/); + + // ── The pause is still there through all of it. + expect(await automation.hasSuspendedRun(runId)).toBe(true); + + // ── The only verb the card found: cancel. It works, and it is a loss — + // `mark_rejected` never ran. + expect(await automation.cancelRun(runId)).toBe(true); + expect(marks, 'the reject branch never happened — what cancelling costs').toEqual([]); + + // ── ⭐ THE MECHANISM, measured without any of this card's own code. + // A SECOND stranded-and-restored run (so the cancel above is undisturbed), + // driven with the very resume the doors would have issued — marker stamped, + // straight at the engine. It COMPLETES. So the restored suspension lacks + // nothing and the `resumeAuthority` gate is not in the way: what is missing + // is an ISSUER permitted to stamp that marker, and the four `pending` + // guards are the only thing standing between an operator and this call. + // That is why the repair belongs on the approvals side, not in the engine. + const other = await strandThenRestore(automation); + rejectBranchThrows = undefined; + const marked: any = await automation.resume(other.flow_run_id, { + branchLabel: 'reject', + output: { decision: 'reject', requestId: other.id }, + [RESUME_AUTHORITY_SERVICE]: true, + } as never); + expect(marked.success, 'the re-armed pause IS resumable — the engine side is whole').toBe(true); + expect(marks, 'and the branch the cancel would have discarded runs').toEqual(['mark_rejected']); + }); + + it('PIN 2 — the fix: the recorded decision is re-issued and the flow completes', async () => { + const automation = boot(); + const req = await strandThenRestore(automation); + const runId = req.flow_run_id; + + // The operator fixed what the node choked on; the decision itself stands. + rejectBranchThrows = undefined; + + const out = await service.continueRestoredRun(req.id, { + requestedBy: 'ops', reason: 'record recreated, re-issuing the reject branch', + }); + + expect(out.resumed, 'the continuation the restore asked for, finally issuable').toBe(true); + expect(out.runId).toBe(runId); + expect(out.decision).toBe('reject'); + expect(out.branchLabel).toBe('reject'); + // The failing door stashed the literal signal it sent, so this is a + // byte-for-byte re-issue rather than an inference. + expect(out.source).toBe('journal'); + expect(out.resumeError).toBeUndefined(); + + // ⭐ The branch's downstream work — the thing cancelling discards — RAN. + expect(marks).toEqual(['mark_rejected']); + expect(await automation.hasSuspendedRun(runId), 'the pause was consumed by the continuation').toBe(false); + expect((await automation.getRun(runId))?.status).toBe('completed'); + + // And `restoreConsumedSuspension` now refuses, for the healthy reason. + expect((await automation.restoreConsumedSuspension(runId)).refusal).toBe('RUN_COMPLETED'); + }); + + it('PIN 3 — it replays the decision and rewrites NOTHING: the four `pending` guards stand', async () => { + const automation = boot(); + const req = await strandThenRestore(automation); + const before = await rowOf(req.id); + rejectBranchThrows = undefined; + + await service.continueRestoredRun(req.id); + + const after = await rowOf(req.id); + expect(after.status, 'still terminal — the request was NOT re-opened').toBe('rejected'); + expect(after.completed_at).toBe(before.completed_at); + expect(after.pending_approvers).toBe(before.pending_approvers); + + // No audit row was minted for the replay: `sys_approval_action.action` is a + // closed contract vocabulary (`APPROVAL_ACTION_KINDS`), and a repair is not + // a new approval action. The trail still reads exactly one `reject`. + const actions = (await data.find('sys_approval_action', { where: { request_id: req.id } })) + .map((a: any) => a.action); + expect(actions.filter((a: string) => a === 'reject')).toHaveLength(1); + + // ⛔ POPULATION: all FOUR `pending`-guarded doors, each asserted here by + // name. The earlier revision of this pin asserted `decide` alone while its + // prose claimed all four — a pin cited for a wider population than its + // cases, which is the shape that lets a real regression through. + const refusals = { + decide: await service.decide(req.id, { decision: 'approve', actorId: 'u1' }, SYSTEM_CTX) + .then(() => null, (e: Error) => e.message), + recall: await service.recall(req.id, { actorId: 'submitter' }, SYSTEM_CTX) + .then(() => null, (e: Error) => e.message), + sendBack: await service.sendBack(req.id, { actorId: 'u1', comment: 'redo' } as any, SYSTEM_CTX) + .then(() => null, (e: Error) => e.message), + resubmit: await service.resubmit(req.id, { actorId: 'submitter' } as any, SYSTEM_CTX) + .then(() => null, (e: Error) => e.message), + }; + expect(refusals.decide).toBe('INVALID_STATE: request is rejected'); + expect(refusals.recall, 'recall refuses a terminal row').toMatch(/INVALID_STATE|FORBIDDEN/); + expect(refusals.sendBack, 'send-back refuses a terminal row').toMatch(/INVALID_STATE|FORBIDDEN/); + expect(refusals.resubmit, 'resubmit refuses a non-returned row').toMatch(/INVALID_STATE|FORBIDDEN/); + }); + + it('PIN 6 — GUARD 1 (superseded): a terminal request cannot drive a pause its run moved on to', async () => { + // POPULATION: two approval nodes in sequence, `a1` decided and `a2` live. + // Before this guard, request 1's continuation walked a2's `approve` edge + // and completed the run while request 2 sat `pending` with no decision. + const automation = boot(); + await automation.execute('two_step', { + object: 'crm_deal', record: { id: 'd9', amount: 5 }, userId: 'submitter', + } as never); + const req1 = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + await service.decide(req1.id, { decision: 'approve', actorId: 'u1' }, SYSTEM_CTX); + + const req2 = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + expect(req2.id, 'the run moved on to a SECOND approval').not.toBe(req1.id); + expect(await automation.hasSuspendedRun(req1.flow_run_id), 'a pause exists on the run').toBe(true); + + const refused = await service.continueRestoredRun(req1.id).then(() => null, (e: Error) => e); + // Two identity guards independently reject this shape; the pin asserts the + // OUTCOME and names which one spoke, rather than pretending to isolate. + expect(refused?.message).toMatch(/^INVALID_STATE: (a newer approval request supersedes this one|run '.*' is parked at node 'a2')/); + + // ⭐ The point of the refusal: a2's pending approval was NOT advanced. + const after = (await data.find('sys_approval_request', { where: { id: req2.id } }))[0]; + expect(after.status, 'still awaiting a real decision').toBe('pending'); + expect(marks, 'and no downstream node ran').toEqual([]); + expect(await automation.hasSuspendedRun(req1.flow_run_id)).toBe(true); + + // GUARD 1 IN ISOLATION. The shape above is also caught by guard 3, so the + // supersede check is driven on its own here: a newer row on the SAME node, + // with `created_at` controlled so the ordering is the thing under test and + // not the clock's resolution. + await data.insert('sys_approval_request', { + id: 'areq_newer', flow_run_id: req1.flow_run_id, flow_node_id: 'a1', + status: 'pending', created_at: '2099-01-01T00:00:00.000Z', node_config_json: '{}', + }); + const superseded = await service.continueRestoredRun(req1.id).then(() => null, (e: Error) => e); + expect(superseded?.message, 'guard 1, driven and asserted alone') + .toBe('INVALID_STATE: a newer approval request supersedes this one'); + }); + + it('PIN 7 — GUARD 1 + GUARD 3: a pause this outcome was not refused on, and a plain retry', async () => { + // POPULATION (a) — GUARD 3, the send-back rebuild. A `returned` request + // (NOT `recalled`: a `recalled` row is refused by `resolveRecordedContinuation` + // before any guard runs, so it can never reach guard 3 on the rebuild path) + // whose run is parked at the revise window. Its recorded outcome is the + // send-back, which was issued at the approval node and LANDED — the pause + // now at the revise window is a fresh one, not the one that outcome was + // refused on. Before this guard the verb issued `revise` there. + const automation = boot(); + await automation.execute('revise_flow', { + object: 'crm_deal', record: { id: 'd7', amount: 3 }, userId: 'submitter', + } as never); + const rev = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + await service.sendBack(rev.id, { actorId: 'u1', comment: 'fix it' } as any, SYSTEM_CTX); + expect((await rowOf(rev.id)).status, 'the row is `returned`').toBe('returned'); + + const parked = await automation.listSuspendedRunsDurable(); + expect(parked.find(r => r.runId === rev.flow_run_id)?.nodeId, + 'the run is parked at the revise window, not at the approval node').toBe('wait_revision'); + + const refusedNode = await service.continueRestoredRun(rev.id).then(() => null, (e: Error) => e); + // ⚠️ The message must name the node the send-back was ISSUED FROM, which is + // the approval node — not merely "this request's own node", a phrase that + // was wrong for the one signal issued from somewhere else (population (c)). + expect(refusedNode?.message).toMatch( + /is parked at node 'wait_revision', but the send-back on request .* was issued from its own approval node 'review'/, + ); + const rounds = await data.find('sys_approval_request', { where: { flow_run_id: rev.flow_run_id } }); + expect(rounds, '⭐ no new pending round was opened').toHaveLength(1); + + // POPULATION (b) — GUARD 1, a plain RETRY of the verb, driven end to end + // through its own designed flow. This is the ordinary shape of using an + // operator tool, not a hostile construction: the first call consumes the + // re-armed pause and the run advances to the NEXT approval node; a second + // call must not walk that node's pending pause. + // + // ⚠️ This replaces a construction that never reached guards 1–3 at all: it + // set `rejectBranchThrows` for a flow with no `mark_rejected` node, never + // decided anything, restored a suspension that was never consumed, and then + // asserted a refusal that was only `request is pending`. It passed on every + // input — a pin whose condition is unreachable is worse than no pin. The + // strand below is real, and the lever that produces it is asserted consumed. + const second = boot(); + await second.execute('two_step', { + object: 'crm_deal', record: { id: 'd8', amount: 4 }, userId: 'submitter', + } as never); + const r1 = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + expect(r1.flow_node_id, 'parked at the FIRST approval node').toBe('a1'); + + // Strand a1's approve continuation: the resume walks the approve edge into + // `a2`, whose executor opens round 2 by inserting a request — and that + // insert fails, once. + data.failNextInsert = 'sys_approval_request'; + const strand = await service + .decide(r1.id, { decision: 'approve', actorId: 'u1' }, SYSTEM_CTX) + .then(() => null, (e: Error) => e); + expect(strand?.message, 'a REAL strand, the state this verb exists to repair').toMatch(/^RESUME_FAILED/); + expect(strandedDecisionDetails(strand)?.repairable).toBe(true); + expect(data.failNextInsert, 'the injected failure actually fired and was consumed').toBeUndefined(); + expect(await second.hasSuspendedRun(r1.flow_run_id), 'the suspension was consumed by the resume').toBe(false); + expect(await data.find('sys_approval_request', { where: { flow_run_id: r1.flow_run_id } }), + 'round 2 never opened — the insert is what failed').toHaveLength(1); + + const rearmed = await second.restoreConsumedSuspension(r1.flow_run_id, { requestedBy: 'ops' }); + expect(rearmed.restored, 'the pause is back at a1').toBe(true); + + // ── The FIRST call is ASSERTED, never discarded. Discarding it is what hid + // the previous version of this pin: a rejected promise assigned to a name + // nothing reads is indistinguishable from a resolved one. + const first = await service.continueRestoredRun(r1.id, { requestedBy: 'ops' }); + expect(first.resumed, '⭐ the first call is the one that must work').toBe(true); + expect(first.source, 'the failing door journalled the signal it was carrying').toBe('journal'); + expect(first.decision).toBe('approve'); + const round2 = (await data.find('sys_approval_request', { + where: { flow_run_id: r1.flow_run_id, status: 'pending' }, + }))[0]; + expect(round2?.flow_node_id, 'the run advanced to the SECOND approval node').toBe('a2'); + expect((await second.listSuspendedRunsDurable()).find(r => r.runId === r1.flow_run_id)?.nodeId, + 'and parked there — a live pause that is NOT r1\'s').toBe('a2'); + + // ── The RETRY. r1 is terminal and superseded; a2's pause is somebody + // else's. Guard 1 refuses before the pause is ever read. + const retry = await service.continueRestoredRun(r1.id).then(() => null, (e: Error) => e); + expect(retry?.message, 'guard 1, reached by an ordinary retry') + .toBe('INVALID_STATE: a newer approval request supersedes this one'); + expect((await rowOf(round2.id)).status, '⭐ a2 is STILL pending — nothing decided it').toBe('pending'); + expect(marks, 'and no downstream branch ran').toEqual([]); + expect(await second.hasSuspendedRun(r1.flow_run_id), 'the run is still parked, not completed').toBe(true); + }); + + it('PIN 8 — B2: a stranded resubmit is replayed as a resubmit, end to end, and `recalled` is refused', async () => { + // POPULATION (a) — the DISCRIMINATOR ALONE, at `resolveRecordedContinuation`. + // ⚠️ States its own population: this leg proves only that the rebuild picks + // `resubmit` over `revise`. It does NOT prove the verb reaches that code — + // populations (d) and (e) below are the legs through `continueRestoredRun` + // end to end, and they exist because this leg was green while guard 3 + // refused every real caller before the resolver's answer was ever acted on. + // + // A `returned` row whose LAST continuation was a resubmit, with the journal + // removed to force the rebuild path — the pre-ship population. One status + // writer, two issuers; rebuilding both as `revise` sent a stranded resubmit + // down the wrong edge, and the engine's unmatched-label fallback (#4414) + // let it proceed rather than fail loudly. + const automation = boot(); + await automation.execute('revise_flow', { + object: 'crm_deal', record: { id: 'd6', amount: 2 }, userId: 'submitter', + } as never); + const req = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + await service.sendBack(req.id, { actorId: 'u1', comment: 'redo' } as any, SYSTEM_CTX); + // The row is `returned`; record the resubmit action the real door writes. + await data.insert('sys_approval_action', { + id: 'aact_resub_1', request_id: req.id, action: 'resubmit', step_index: 0, + }); + const rebuilt = await (service as any).resolveRecordedContinuation( + (await data.find('sys_approval_request', { where: { id: req.id } }))[0], req.id, + ); + expect(rebuilt.source).toBe('reconstructed'); + expect(rebuilt.signal.decision, '⭐ resubmit, NOT revise').toBe('resubmit'); + expect(rebuilt.signal.branchLabel).toBe('resubmit'); + expect(rebuilt.signal.output).toEqual({ resubmitted: true, requestId: req.id }); + + // POPULATION (b): the same row with NO resubmit action row rebuilds as the + // send-back — the reverse control that keeps (a) from being a constant. + // Resolver-level like (a); end to end this shape is refused by guard 3, and + // that refusal is PIN 7(a). + const other = boot(); + await other.execute('revise_flow', { + object: 'crm_deal', record: { id: 'd5', amount: 1 }, userId: 'submitter', + } as never); + const req2 = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + await service.sendBack(req2.id, { actorId: 'u1', comment: 'redo' } as any, SYSTEM_CTX); + const rebuilt2 = await (service as any).resolveRecordedContinuation( + (await data.find('sys_approval_request', { where: { id: req2.id } }))[0], req2.id, + ); + expect(rebuilt2.signal.decision, 'no resubmit row ⇒ the send-back').toBe('revise'); + expect(rebuilt2.signal.branchLabel).toBe('revise'); + + // POPULATION (c): `recalled` with no journal is REFUSED, not guessed — + // two status writers, three behaviours, two issuing no continuation. + const recalledRow = { id: 'areq_r', status: 'recalled', flow_run_id: 'run_r', flow_node_id: 'review', node_config_json: '{}' }; + const refused = await (service as any).resolveRecordedContinuation(recalledRow, 'areq_r') + .then(() => null, (e: Error) => e); + expect(refused?.message).toMatch(/is 'recalled' and carries no journalled continuation/); + expect(refused?.message, 'and it names what the operator can do instead').toMatch(/cancelRun/); + + // ── POPULATION (d) — THE WHOLE PATH, journal leg. A resubmit that stranded, + // restored, and re-issued through `continueRestoredRun` itself. + // + // ⭐ This is the leg guard 3 used to refuse. A resubmit is issued from the + // revise window, so the pause it strands on — and the pause the restore + // re-arms — is at `wait_revision`, while the row's `flow_node_id` still + // reads the approval node `review`. Comparing the parked node against the + // row's own node refused this, and told the operator the pause was not this + // request's when it was exactly this request's. + const live = boot(); + await live.execute('revise_flow', { + object: 'crm_deal', record: { id: 'd9', amount: 9 }, userId: 'submitter', + } as never); + // ⚠️ Selected by record, not by "the first pending row": earlier populations + // in this pin leave their own pending rounds in the shared table, and the + // bare status filter picked one of THOSE — a run belonging to an engine no + // longer attached, which failed as `RESUME_TARGET_LOST` rather than as + // anything about this leg. + const rq = (await data.find('sys_approval_request', { where: { record_id: 'd9', status: 'pending' } }))[0]; + expect(rq?.flow_node_id, 'this leg owns its own run').toBe('review'); + await service.sendBack(rq.id, { actorId: 'u1', comment: 'redo' } as any, SYSTEM_CTX); + expect((await live.listSuspendedRunsDurable()).find(r => r.runId === rq.flow_run_id)?.nodeId, + 'the send-back parked the run at the revise window').toBe('wait_revision'); + + // Strand the resubmit: the back-edge re-enters `review`, whose executor + // opens round 2 by inserting a request — fail that insert, once. + data.failNextInsert = 'sys_approval_request'; + const strandedResubmit = await service + .resubmit(rq.id, { actorId: 'submitter' } as any, SYSTEM_CTX) + .then(() => null, (e: Error) => e); + expect(strandedResubmit?.message, 'a real stranded resubmit').toMatch(/^RESUME_FAILED/); + expect(strandedDecisionDetails(strandedResubmit)?.repairable).toBe(true); + expect(data.failNextInsert, 'the injected failure fired and was consumed').toBeUndefined(); + + const rearmed = await live.restoreConsumedSuspension(rq.flow_run_id, { requestedBy: 'ops' }); + expect(rearmed.restored).toBe(true); + expect((await live.listSuspendedRunsDurable()).find(r => r.runId === rq.flow_run_id)?.nodeId, + '⭐ re-armed at the revise window').toBe('wait_revision'); + expect((await rowOf(rq.id)).flow_node_id, + '⭐ while the row still records the approval node — the two differ, by construction').toBe('review'); + + const replayed = await service.continueRestoredRun(rq.id, { requestedBy: 'ops' }); + expect(replayed.resumed, '⭐ guard 3 admits it, because it asks where a RESUBMIT is issued from').toBe(true); + expect(replayed.source, 'the failing door journalled the literal signal').toBe('journal'); + expect(replayed.decision).toBe('resubmit'); + expect(replayed.branchLabel).toBe('resubmit'); + expect(await data.find('sys_approval_request', { where: { flow_run_id: rq.flow_run_id } }), + 'round 2 opened — the back-edge was walked as a resubmit').toHaveLength(2); + expect((await live.listSuspendedRunsDurable()).find(r => r.runId === rq.flow_run_id)?.nodeId, + 'and the run is parked back at the approval node').toBe('review'); + + // ── POPULATION (e) — THE WHOLE PATH, rebuild leg: the same shape with the + // journal stripped, which is what a run stranded BEFORE this shipped looks + // like. This is the population the card names, and it is the one (a) was + // cited for while nothing exercised it end to end. + const pre = boot(); + await pre.execute('revise_flow', { + object: 'crm_deal', record: { id: 'd10', amount: 10 }, userId: 'submitter', + } as never); + const rq2 = (await data.find('sys_approval_request', { where: { record_id: 'd10', status: 'pending' } }))[0]; + expect(rq2?.flow_node_id, 'this leg owns its own run').toBe('review'); + await service.sendBack(rq2.id, { actorId: 'u1', comment: 'redo' } as any, SYSTEM_CTX); + data.failNextInsert = 'sys_approval_request'; + await service.resubmit(rq2.id, { actorId: 'submitter' } as any, SYSTEM_CTX) + .then(() => null, () => null); + expect(data.failNextInsert, 'the injected failure fired').toBeUndefined(); + await pre.restoreConsumedSuspension(rq2.flow_run_id, { requestedBy: 'ops' }); + + const preConfig = JSON.parse((await rowOf(rq2.id)).node_config_json); + expect(preConfig.__strandedContinuation, 'the door DID journal it — this leg removes it on purpose').toBeTruthy(); + delete preConfig.__strandedContinuation; + await data.update('sys_approval_request', { + id: rq2.id, node_config_json: JSON.stringify(preConfig), + }, { context: SYSTEM_CTX }); + + const rebuiltRun = await service.continueRestoredRun(rq2.id, { requestedBy: 'ops' }); + expect(rebuiltRun.source, 'no journal — rebuilt from the resubmit action row').toBe('reconstructed'); + expect(rebuiltRun.resumed, '⭐ and the rebuild is ACTED ON, not merely computed').toBe(true); + expect(rebuiltRun.decision).toBe('resubmit'); + expect(await data.find('sys_approval_request', { where: { flow_run_id: rq2.flow_run_id } }), + 'round 2 opened on the rebuild path too').toHaveLength(2); + }); + + it('PIN 4 — REVERSE CONTROLS: it refuses when there is no re-armed pause to continue', async () => { + const automation = boot(); + + // (a) A healthy decision. The run completed; there is nothing to continue, + // and this must NOT re-run the branch a second time. + const healthy = await park(automation); + await service.decide(healthy.id, { decision: 'reject', actorId: 'u1' }, SYSTEM_CTX); + expect(marks).toEqual(['mark_rejected']); + const noPause = await service.continueRestoredRun(healthy.id).then(() => null, (e: Error) => e); + expect(noPause?.message).toMatch(/is not suspended/); + // ⭐ The point of the refusal: the branch did not run twice. + expect(marks, 'a completed run must not be walked again').toEqual(['mark_rejected']); + + // (b) Stranded but NOT restored — the operator skipped the restore, and the + // refusal has to name the step they missed rather than fail obscurely. + rejectBranchThrows = DOWNSTREAM_FAILURE; + const stranded = await park(automation); + await service.decide(stranded.id, { decision: 'reject', actorId: 'u1' }, SYSTEM_CTX) + .then(() => null, () => null); + const notRestored = await service.continueRestoredRun(stranded.id) + .then(() => null, (e: Error) => e); + expect(notRestored?.message).toMatch(/restoreConsumedSuspension/); + + // (c) A pending request has no recorded outcome to replay — `decide` is its + // continuation, and this verb must not become a second way in. + const pending = await park(automation); + const stillPending = await service.continueRestoredRun(pending.id) + .then(() => null, (e: Error) => e); + expect(stillPending?.message).toMatch(/^INVALID_STATE: request is pending/); + + // (d) Unknown request. + const missing = await service.continueRestoredRun('areq_nope').then(() => null, (e: Error) => e); + expect(missing?.message).toBe('REQUEST_NOT_FOUND: areq_nope'); + }); + + it('PIN 5 — with no journal it rebuilds the signal, and refuses the one shape it cannot', async () => { + const automation = boot(); + const req = await strandThenRestore(automation); + + // A run stranded by a build WITHOUT the journal — the population the card + // was filed for ("the runs already in this state"). Simulated by removing + // the stash the door just wrote, which is exactly what such a row looks like. + const stashed = await rowOf(req.id); + const config = JSON.parse(stashed.node_config_json); + expect(config.__strandedContinuation, 'the door DID journal it — this pin removes it on purpose') + .toBeTruthy(); + delete config.__strandedContinuation; + await data.update('sys_approval_request', { + id: req.id, node_config_json: JSON.stringify(config), + }, { context: SYSTEM_CTX }); + + rejectBranchThrows = undefined; + const out = await service.continueRestoredRun(req.id); + expect(out.source, 'rebuilt from the recorded outcome, and it says so').toBe('reconstructed'); + expect(out.resumed).toBe(true); + expect(out.decision).toBe('reject'); + expect(marks).toEqual(['mark_rejected']); + + // ── The shape the rebuild REFUSES rather than guesses. A `rejected` row + // that also carries a `revise` action may be ADR-0044's revision-limit + // auto-rejection, whose original resume carried `autoRejected: true`. + // Replaying it as a plain rejection would hand the flow a payload it never + // had — so with no journal to consult, this refuses. + const second = boot(); + rejectBranchThrows = DOWNSTREAM_FAILURE; + const amb = await park(second); + await service.decide(amb.id, { decision: 'reject', actorId: 'u1' }, SYSTEM_CTX) + .then(() => null, () => null); + await second.restoreConsumedSuspension(amb.flow_run_id); + const ambRow = await rowOf(amb.id); + const ambConfig = JSON.parse(ambRow.node_config_json); + delete ambConfig.__strandedContinuation; + await data.update('sys_approval_request', { + id: amb.id, node_config_json: JSON.stringify(ambConfig), + }, { context: SYSTEM_CTX }); + await data.insert('sys_approval_action', { + id: 'aact_revise_x', request_id: amb.id, action: 'revise', step_index: 0, + }); + + const ambiguous = await service.continueRestoredRun(amb.id).then(() => null, (e: Error) => e); + expect(ambiguous?.message).toMatch(/cannot tell a decided rejection from an ADR-0044/); + // ⛔ And it refused by NOT resuming — the run is still parked, still repairable. + expect(await second.hasSuspendedRun(amb.flow_run_id)).toBe(true); + }); +}); diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index f1ad403e2f..a78e84bfe5 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -2181,6 +2181,16 @@ "verb": "update", "pinned": 1 }, + { + "file": "packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts", + "verb": "delete", + "pinned": 1 + }, + { + "file": "packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts", + "verb": "update", + "pinned": 1 + }, { "file": "packages/plugins/plugin-approvals/src/subflow-hosted-approval-strand.test.ts", "verb": "delete",