diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index fb98f82701a..f9bc84e40a4 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -153,6 +153,7 @@ import { loadSuspendedRun, loadSuspendedRunStrict, creditChildRun, + resumeInternal, failSuspendedRun, settleUnresumableSuspension, cancelRun, @@ -1399,7 +1400,7 @@ export function negotiatedServiceLocale(service: RefusalI18nService, requested: * reason both legs converge on that one signal rather than growing a second * terminal exit apiece. */ -interface ChildRunRefusal { +export interface ChildRunRefusal { /** * The child run's own `refusalMessage`, passed through verbatim. * `undefined` only when the child carried none — recorded honestly rather @@ -1453,7 +1454,7 @@ const RETRYABLE_RESUME_REFUSAL_CODES: ReadonlySet = new Set([ ] satisfies ReadonlyArray>); /** Whether an {@link AutomationResult} code names a resume that never ran. */ -function isRetryableResumeRefusal(code: AutomationResult['code']): boolean { +export function isRetryableResumeRefusal(code: AutomationResult['code']): boolean { return code !== undefined && RETRYABLE_RESUME_REFUSAL_CODES.has(code); } @@ -1494,7 +1495,7 @@ function engineBuilt(signal: ResumeSignal): ResumeSignal { * DOWN to the child, so shape alone would let a caller's own bag silence the * genuine degraded warning. Only the engine can mint the symbol. */ -function carriesSubflowChildOutput(signal: ResumeSignal): boolean { +export function carriesSubflowChildOutput(signal: ResumeSignal): boolean { if ((signal as Record)[ENGINE_BUILT_SIGNAL] !== true) return false; const output = signal.output; return typeof output === 'object' && output !== null && 'output' in output; @@ -1529,7 +1530,7 @@ function isEngineVariable(name: string): boolean { * signal is never absent here — `resume` normalises a missing one to `{}` * (commit 7307191db), which folds nothing and rejects nothing. */ -function applyResumeSignal( +export function applyResumeSignal( variables: Map, signal: ResumeSignal, nodeId: string, @@ -6482,7 +6483,7 @@ export class AutomationEngine implements IAutomationService { * variable map should ever see — the reserved-name guard * (`applyResumeSignal`) polices that map, and a refusal is not a variable. */ - private async resumeInternal( + private resumeInternal( runId: string, // Never `undefined` past the public door: `resume` normalises an // absent caller signal to `{}` (commit 7307191db), and the engine's own @@ -6494,912 +6495,35 @@ export class AutomationEngine implements IAutomationService { childSummary?: FlowRunSummary, childRefusal?: ChildRunRefusal, ): Promise { - // Idempotency guard (set synchronously, before any await): reject a - // concurrent duplicate resume of the same run so side effects can't run - // twice. A duplicate that arrives *after* this one finishes finds no - // suspended run and returns the "no suspended run" error below. - if (this.resuming.has(runId)) { - return { success: false, code: 'RESUME_IN_PROGRESS', error: `Run '${runId}' is already being resumed` }; - } - this.resuming.add(runId); - try { - // Hot path: suspended in this process. Cold path: rehydrate from the - // durable store (e.g. the process restarted since the pause, ADR-0019). - // - // Strict load: a store that cannot be READ must not report as - // "no such run" (#4420). A caller that already persisted a decision - // needs "retry when the store is back" to be distinguishable from - // "this run is gone for good" — same failure, opposite remedy. - let run: SuspendedRun | null; - try { - run = await this.loadSuspendedRunStrict(runId); - } catch (err) { - const message = (err as Error).message; - // #5912 — the LOG record's cause goes to `meta`, never into the - // message. `message` is the datasource DRIVER's own failure text - // and we do not control how many lines it has; - // `ObjectLogger.write()` adds one ` ` head per call, so - // a newline in it turns this ONE record into several physical - // lines of which only the first is greppable — the family of - // #5048 / #5575 / #5636 / #5661 / #5737, and cloud#971's shape. - // - // Third argument, per the `Logger` contract - // (`packages/spec/src/contracts/logger.ts`) - // `error(message, error?, meta?)`. NOT the second: that is the - // `Error` slot and a raw error there ships its whole stack on - // every record (#5575). `runId` stays in the message — it is the - // caller's own handle, the same call the family's other seams - // make for their ids, not the foreign text this fix is about. - // - // Level stays `error`: the run is on disk and the resume did not - // land, which is #4632's durability degradation exactly. - this.logger.error( - `[automation] durable suspended-run store unreachable while resuming '${runId}' — the suspension was ` + - `NOT consumed, so the run stays parked and this resume can be retried verbatim once the store is ` + - `reachable; the caller was told the same via the STORE_UNAVAILABLE result. The store's own failure ` + - `is in this record's meta.`, - undefined, - describeThrownForLog(err), - ); - // The RESULT envelope keeps the cause spliced in, verbatim and - // deliberately (#5636 made the same call for `degradedReason`): - // this is a structured return value a caller reads as a whole — - // over REST it is a JSON string field, never split on newlines — - // and PR #5911 already made wait-node put it into `meta` intact. - return { - success: false, - code: 'STORE_UNAVAILABLE', - error: `Durable suspended-run store unreachable for run '${runId}' — retry once the store is available: ${message}`, - }; - } - if (!run) { - return { success: false, code: 'RUN_NOT_FOUND', error: `No suspended run '${runId}'` }; - } - // [#8684] The two NEVER-DISPATCHED exits, classified HERE rather than - // guessed at the transport. Both mean the suspension is real but can - // never continue — the flow it belongs to is no longer registered, or - // the node it parked on was edited away underneath it. Nothing ran, - // so neither is a business rejection: they are the same terminal - // "this pause is gone for good" class as a missing suspension, and a - // transport answers all three **404**. - // - // Code-less until now, which is what forced the route to either - // answer them 200 (the defect this card is about) or sniff the result - // for `summary`/`durationMs` to tell them from a run that ran and - // failed — the tolerant-consumer shape PD #12 forbids. The producer - // is the only component that knows, so it says so. - // - // Deliberately the EXISTING `RUN_NOT_FOUND` rather than a new code: - // the remedy is identical (this run is finished; start a new one), - // the consumer branches on the 404 STATUS and not on the code - // (objectui's `interpretFlowResponse`), and `error` below still names - // which of the three causes it was, verbatim on the wire. A distinct - // code would be vocabulary nothing reads — add one the day a caller - // needs to branch on the difference. - // - // [#23963] And the verdict is carried onto the RUN, not only onto - // the answer: both exits go through - // {@link settleUnresumableSuspension}, which moves the parked row to - // `failed` with this same text in `error`. Answering "gone for - // good" while the row kept `status: 'paused'` left it resumable on - // every surface that reads the row, and outside retention, whose - // scope is terminal statuses only. - const flow = this.flows.get(run.flowName); - if (!flow) { - return this.settleUnresumableSuspension(run, `Flow '${run.flowName}' not found for run '${runId}'`); - } - const node = flow.nodes.find(n => n.id === run.nodeId); - if (!node) { - return this.settleUnresumableSuspension( - run, - `Suspended node '${run.nodeId}' no longer exists in flow '${run.flowName}'`, - ); - } - - // #4354 — up-bubble: the child that just finished did work this run - // is accountable for. Credit it to the step that suspended awaiting - // it, before traversal appends anything further. - this.creditChildRun(run.steps, run.nodeId, childSummary); - - // [#18714] The refusal this run must end on instead of continuing - // past its `subflow` / `map` node, if any. Seeded from the UP-BUBBLE - // leg (the child's own frame drove this resume and already knows), - // and set below by the DELEGATED leg (this frame drives the child - // and reads its answer). Both legs hand it to the ONE throw site - // inside the traversal `try` further down, so a refusal leaves this - // method through the same `finishRefusedRun` chokepoint every other - // producer uses — see `ChildRunRefusal`. - let pendingChildRefusal: ChildRunRefusal | undefined = childRefusal; - - // ── Subflow delegation (nested pause): this run is paused at a - // `subflow` node whose child run itself suspended. The caller's - // signal is meant for the node the CHILD paused on (its screen / - // approval / wait), so forward it down. The child resumes with - // bubbling off — when it completes, *this* invocation continues the - // parent from the subflow node with the child's output, using the - // same mapping as the synchronous path. - if (typeof run.correlation === 'string' && run.correlation.startsWith('subflow:')) { - 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 = await this.loadSuspendedRun(childRunId); - if (childRun) { - const childRes = await this.resumeInternal(childRunId, signal, true); - if (childRes.status === 'paused') { - // Child paused again (e.g. the next screen of a wizard). - // This run stays suspended; refresh its surfaced screen - // so a re-fetch (getSuspendedScreen) shows the new one. - if (childRes.screen && childRes.screen !== run.screen) { - await this.persistSuspendedRun({ ...run, screen: childRes.screen }); - } - return { - success: true, - status: 'paused', - runId, - durationMs: Date.now() - run.startTime, - screen: childRes.screen, - // THIS run's flow, not the child's: the caller - // addressed this run id and its runner names the - // flow it launched. The child only lends a screen. - flowLabel: flow.label, - }; - } - // [#14379] A child REFUSAL is not a child failure. The - // codes in {@link RETRYABLE_RESUME_REFUSAL_CODES} are this - // method's own answers for a resume that never ran: the - // child's screen contract is checked BEFORE - // `forgetSuspendedRun` precisely so "a rejected bag leaves - // the pause live and the legitimate submission still lands" - // (#4477), so the child is parked exactly where it was. - // - // Reading those as a terminal failure consumed the PARENT's - // pause over a mistyped form field — and the screen-flow - // path is where a caller holds ONE stable run id, the - // parent's, and posts every wizard step to it. The run was - // gone, the still-paused child orphaned with nothing left - // to bubble into, the caller told `400 FLOW_FAILED` ("it - // ran and was rejected") for something that never ran, and - // their corrected retry on that same id answered - // `RUN_NOT_FOUND`. - // - // The child's envelope is answered VERBATIM but for the - // parent's `durationMs`. Propagating the `code` is half the - // fix: a code-less envelope is exactly what forced the - // transport onto `400 FLOW_FAILED`, and leaving both pauses - // alive while still answering one repairs the state and - // leaves the caller equally misled. The child's `error` is - // the actionable half — `Screen field "kind" is required` — - // where the failure text below names neither the problem - // nor anything a caller can act on. Nothing else moves: - // neither pause is consumed, and the parent's surfaced - // screen needs no refresh because the child did not - // advance. - if (!childRes.success && isRetryableResumeRefusal(childRes.code)) { - return { ...childRes, durationMs: Date.now() - run.startTime }; - } - if (!childRes.success) { - const error = `subflow run '${childRunId}' (${childRun.flowName}) failed: ${childRes.error ?? 'unknown error'}`; - // [#15222] This frame consumes its OWN pause here, and - // before this card journalled nothing for it — so a - // child that stranded left a repairable leaf under an - // ancestor no verb could reach. Journalled exactly when - // the child is repairable, asked of the journal rather - // than of the child's `status`: a child that itself - // cascade-failed from a deeper strand carries no status - // word at all (see the return below) and is repairable - // all the same, which is what makes a chain deeper than - // two levels re-armable by induction. - // - // ⛔ And the envelope below is UNCHANGED — no status is - // stamped on it. `'stranded'` belongs to the run that - // consumed its own pause and then threw downstream; - // stamping it here would tell an operator to retry a - // recovery that cannot succeed, because nothing re-arms - // this run by resuming it. What reaches this run is - // {@link restoreConsumedSuspension}, through the - // journal one line up. - await this.failSuspendedRun( - run, - error, - this.consumedSuspensions.has(childRunId) ? childRunId : undefined, - ); - return { success: false, error, durationMs: Date.now() - run.startTime, flowLabel: flow.label }; - } - // [#18714] DELEGATED-LEG REFUSAL. The child ran to a - // refusing terminal — an `end` declaring - // `outcome: 'refused'`, or a node of its own whose child - // refused — and answered `{ success: true, status: - // 'refused' }`, because *a refusal is a successful - // evaluation that says no*. The branch set above tests only - // `paused` and `!success`, so until this arm existed a - // refused child was neither and fell straight through the - // ordinary success exit below: measured, the parent - // returned `{ success: true, successMessage: … }`, its run - // row recorded `completed`, and the node downstream of the - // `subflow` RAN. That is the identical fail-open shape - // #18110 / #18555 closed on the synchronous leg, still open - // on the leg 「the one a screen flow actually takes」. - // - // ⛔ NOT folded into the `!childRes.success` arm above. That - // arm calls `failSuspendedRun` and records the parent - // `failed`; a refusal is not a failure — it must not be - // routable by a `fault` edge, must not consume retry budget - // and must not be counted in `nodes[].failures`, all of - // which the failure arm would confer. The parent's terminal - // row must read `refused`, the same word the child's does. - // - // Recorded, ⛔ not thrown here. The pause is not consumed - // yet — `claimAdvance` / `forgetSuspendedRun` are still - // below — and a refusal that unwound from this position - // would leave the parent's own suspension live while its - // run row said terminal. The throw site inside the - // traversal `try` is past the consumption, which is why - // both legs converge there. - // - // The mapping below still runs, deliberately: the child's - // declared outputs really were produced and the nodes - // before its refusal really ran, so the parent's answer - // must not depend on HOW the child ended — the same call - // the synchronous `subflow` arm makes when it returns the - // success envelope PLUS `refuse`. - if (childRes.status === 'refused') { - pendingChildRefusal = { message: childRes.refusalMessage }; - } - // Child completed — continue below with its output as the - // resume signal (replaces the caller's signal, which the - // child already consumed). - signal = this.buildSubflowResumeSignal(childRun.context, childRes.output); - // #4354 — down-delegation is the other way a child's work - // lands under a parent step written at suspend time. - this.creditChildRun(run.steps, run.nodeId, childRes.summary); - } else if (carriesSubflowChildOutput(signal)) { - // The engine-built UP-BUBBLE — the HEALTHY path, and the - // one this `else` used to describe as a degradation. The - // child COMPLETED and `bubbleToParent` resumed this run - // with the child's mapped output, so the miss above is the - // SUSPENSION lookup missing (a completed run has no - // suspension), not the child having disappeared, and the - // parent continues *with* the child's output — which is - // what the old sentence denied outright. - // - // `debug`, for the same reason as the lost-advance-claim - // record further down: an ordinary outcome logged at `warn` - // on every healthy subflow completion is the cry-wolf - // shape — the operator who reads "child run is gone" on - // each one stops reading the line, and the one time it - // means what it says is the time it is ignored. Nothing - // else moves: this branch continues the parent exactly as - // before, and the degraded branch below keeps its text and - // its level. - this.logger.debug( - `[automation] run '${runId}' continues from subflow node '${run.nodeId}' with the output of ` + - `completed child run '${childRunId}' — carried on the engine-built up-bubble signal ` + - `(the child's suspension is consumed, which is why it is not loadable here).`, - ); - } else { - // The genuinely degraded case, and the only one the - // sentence below was ever true of: nothing suspended under - // the child id AND no carried output, so this run really - // does continue without the child's output. #4632 verdict: - // FUNCTIONAL — stays `warn`. - this.logger.warn( - `[automation] run '${runId}' is paused at subflow node '${run.nodeId}' but child run '${childRunId}' ` + - `is gone — continuing without child output`, - ); - } - } - - // The SCREEN contract (#4477). A run parked on a `screen` node - // declared exactly which keys it collects and which are required; - // until this ran, `resume` folded any bag at all straight into the - // variables, so a caller that skipped the dialog bypassed every - // `required` the author wrote. Checked here — beside the engine's - // other resume refusals and BEFORE the suspension is consumed — so - // a rejected bag leaves the pause live and the legitimate - // submission still lands. - const screenRefusal = this.refuseInvalidScreenInput(run, runId, signal); - if (screenRefusal) return screenRefusal; - - // Restore the variable context and fold the signal in — the ONE - // place a resume signal reaches the variable map. Runs BEFORE the - // suspension is consumed, so a rejected signal changes nothing: - // the pause stays live and the legitimate continuation still lands. - const variables = new Map(Object.entries(run.variables)); - const rejected = applyResumeSignal(variables, signal, run.nodeId); - if (rejected.length) { - // #6654 — the rejected names are the CALLER's resume-signal - // keys (nothing constrains them against newlines), so they - // ride the structured slot, never the message; see - // `forgetSuspendedRun`'s catch for the full mechanism (#6299). - // The returned INVALID_SIGNAL envelope below is caller-facing - // refusal text, not a log record — it keeps naming the - // variables (the envelope class is ruled elsewhere). - // #4632: FUNCTIONAL — stays `warn`. - this.logger.warn( - `[automation] refused resume of run '${runId}': signal writes engine-internal ` + - `variable(s) — the rejected names are in this record's meta.`, - { rejected }, - ); - return { - success: false, - code: 'INVALID_SIGNAL', - error: - `Resume signal may not set engine-internal variables (${rejected.join(', ')}) — ` + - `names starting with '$' (or containing '.$') are reserved by the flow engine`, - }; - } - - // [#13909] How long the step log was AT THE PAUSE, read before - // anything downstream runs. `steps` below is the SAME array - // `traverseNext` appends to, so by the time the catch arm builds a - // restore snapshot the pause's own step log is no longer - // recoverable from it — this integer is what trims it back. One - // `int` on every resume and nothing else: the snapshot itself is - // built only on the failure path. - // - // ⛔ Deliberately NOT a change to the ordering. The consumption - // below still precedes the traversal, `forgetSuspendedRun` is - // untouched, and `hasSuspendedRun` still answers false for the - // whole traversal window (pinned in - // `consumed-suspension-restore.test.ts`). That order is RULED, - // not inherited: #13937 shape 4 (maintainer 2026-09-01) keeps it - // for exactly-once across a crash, excludes the flip (shape 2) - // unless a durable claim/lease lands with it, and excludes - // consume-then-re-arm (shape 3) outright. The exit from the state - // it leaves behind is `restoreConsumedSuspension`, and the state - // itself is named on the result (`status: 'stranded'`, below). - const stepCountAtPause = run.steps.length; - // [#15660] …and the variables AT THE PAUSE, for the same reason and - // at the same moment, because `run.steps.length` is not the only - // thing the failed attempt can move under the journal. - // - // `journalConsumedSuspension` promises its snapshot VERBATIM — "the - // state at the pause, not the state the failed attempt was working - // from" — and delivered that only for the TOP-LEVEL keys. The scope - // the downstream nodes run against is built one line below as - // `new Map(Object.entries(run.variables))`, which copies the keys - // and SHARES every value object. A node that holds state in the - // scope and updates it in place — `map` keeps `.$mapState` - // and is, by a re-derived census, the only executor that does — - // therefore writes straight through into the object the journal - // later hands an operator as the pause. - // - // Measured, not read (`consumed-suspension-snapshot-aliasing.test.ts`): - // the durable row held `started: 1` at the pause and the restore put - // back `started: 99`. ⛔ And the two placements the card proposed - // are both refuted by that same measurement: - // - "deep copy at SNAPSHOT time" — `InMemorySuspendedRunStore` - // already JSON round-trips on save AND load, so this run's - // variables ARE a private deep copy before a single downstream - // node runs. It reproduces anyway: the aliasing that carries the - // mutation is minted HERE, on the resume, not at the suspend. - // - "copy at RESTORE time" — the mutation lands before the node - // throws, so the journal is already corrupt when it is written; - // anything copied at or after that point copies the corruption. - // The copy has to be taken before the failed attempt runs, which is - // this line. - // - // Cost, measured rather than assumed (the card flagged it NOT - // MEASURED): 5.8 µs for a typical scope, and one suspend+resume - // round trip against a configured store already performs FOUR full - // clones of this same payload (save ×1, load ×2, recordTerminal ×1), - // so this is a fifth on a path whose production cost is a durable - // round trip. ⛔ Deliberately NOT the JSON clone the journal's own - // docblock ruled out: that objection was that it would run INSIDE a - // catch arm already handling a failure, where a circular value would - // throw the operator's repair away. Here it is on the happy path, - // where a throw is just a value we decline to copy. - const variablesAtPause = this.cloneVariablesAtPause(run, runId); - - // Ruling A on `FlowSchema.runAs`: who runs the nodes after this - // pause — the decider, when the owning service's signal names one - // (with their own grants on a `runAs:'user'` run, as the attribution - // user of a still-elevated `runAs:'system'` run); otherwise the - // identity the run paused with. - // Resolved BEFORE the claim, so nothing here can run after the - // suspension is consumed and leave it consumed with no continuation; - // the persisted `run.context` is left as it was, and the repair - // journal therefore still snapshots the pause exactly as it parked. - const segmentContext = await this.resolveDecisionSegmentContext(run, signal); - - // Consume the suspension *before* running downstream work — a run - // resumes exactly once per pause, and a duplicate resume after a - // partial restart must not double-run side effects. (Folding the - // signal above is pure in-memory work, not downstream work.) - // This is also where the paused node learns its pause is over and - // disarms what it armed on entry (#5512) — see forgetSuspendedRun. - // - // [#14333] And the consumption is now a CLAIM, not an assertion. - // `resuming` above answered "is this process already resuming it"; - // this answers "did any process get here first", against the shared - // store, atomically. Placed exactly where the unconditional consume - // was: every refusal above it still refuses without consuming, and - // the ordering #13937 ruled to keep is untouched. - const claim = await this.claimAdvance(run); - if (claim.kind === 'lost') { - // An ORDINARY outcome, not a degradation: this is the guard - // doing its job, and the caller is told in the result. `debug` - // rather than `warn` for exactly that reason — a losing racer - // per approval on a busy any-of level would otherwise be a - // steady `warn` stream describing correct behaviour. - this.logger.debug( - `[automation] resume of run '${runId}' lost the advance claim at node '${run.nodeId}' — ` + - `another replica consumed this suspension first; nothing was dispatched here.`, - ); - // The SAME code the in-process guard answers, deliberately: the - // observation differs (another process, not this one) but the - // remedy is identical — a concurrent resume already has this - // run, the caller's own recorded outcome stands, and nothing - // here should be retried. `plugin-approvals` already branches - // on it that way (`resumeRecordedOutcome`), and the transport - // already maps it to 409. A distinct code would be vocabulary - // nothing reads — add one the day a caller needs the - // difference. - // [#15832] The store's compare-and-set ANSWERED: no row is parked - // where this replica read it. Whatever snapshot this process - // still holds for the run is stale by construction — it names a - // node the run has left — so it stops being published by the two - // listings. ⛔ NOT `forgetSuspendedRun`: nothing was consumed - // here, and firing that choke point would tear the pause down a - // second time in this process on top of the winner's own. - this.evictConsumedSuspension(runId); - return { - success: false, - code: 'RESUME_IN_PROGRESS', - error: - `Run '${runId}' is already being resumed — another replica consumed its suspension ` + - `at node '${run.nodeId}' first`, - }; - } - if (claim.kind === 'unavailable') { - // Same envelope, same reason as the strict load's failure - // above: existence is UNKNOWN, not "gone for good", and the - // identical call is expected to work once the store recovers - // (#4420). - // - // The text states only what is KNOWN. The strict load's failure - // can say the suspension was not consumed, because it fails - // before anything is consumed; this one cannot — a claim throws - // just as readily after a committed delete as before one, so - // "NOT consumed" would be a claim about a fact this seam does - // not have. What it does know is that this resume did not - // continue the run, and that a retry settles the rest. - return { - success: false, - code: 'STORE_UNAVAILABLE', - error: - `Durable suspended-run store unreachable while claiming the advance for run '${runId}' — ` + - `this resume did NOT continue the run, and whether the store consumed the suspension is ` + - `UNKNOWN (a failure can arrive after a committed delete). Retry once the store is ` + - `available: a retry that finds the run still parked continues it, and one that finds the ` + - `suspension gone answers RUN_NOT_FOUND: ${claim.message}`, - }; - } - await this.forgetSuspendedRun(run, 'resumed', claim.kind === 'claimed'); - - const steps = run.steps; - const context = segmentContext; - - try { - // [#18714] The child this run is parked on REFUSED — on either - // resumed leg (see `ChildRunRefusal`). Thrown HERE, and the - // position is the point: - // - // - Past the consumption. `claimAdvance` and - // `forgetSuspendedRun` have run, so the parent's own pause is - // gone exactly as it is for every other way this resume can - // end. A refusal raised before them would record a terminal - // run while leaving its suspension live in - // `listSuspendedRuns()` — the very leak the up-bubble leg is - // filed for, moved one frame up. - // - Before the traversal. Nothing downstream of the awaiting - // node runs, which is the whole content of "a refusal stops - // the run". - // - As `FlowRefusalSignal`, so the `catch` below converts it - // through `finishRefusedRun` — ONE terminal shape, whichever - // producer raised it. ⛔ Deliberately not a second terminal - // exit of its own: this file's own history is a list of - // outcomes that became a function of WHICH ROUTE a run took. - // - // `run.nodeId` is the node that was awaiting the child — the - // `subflow` / `map` this frame is parked on, the same node the - // synchronous leg names on its `FlowRefusalSignal`. - if (pendingChildRefusal) { - throw new FlowRefusalSignal(run.nodeId, pendingChildRefusal.message); - } - // ── Map re-entry (sequential multi-instance, ADR-0037 A2). - // A run paused at a `map` node (correlation `map:`) - // does NOT continue past the node on resume — it RE-RUNS the - // node so the executor can record the just-completed unit and - // start the next item. The default path continues past the node. - if (typeof run.correlation === 'string' && run.correlation.startsWith('map:')) { - await this.executeNode(node, flow, variables, context, steps); - } else { - await this.traverseNext(node, flow, variables, context, steps, signal.branchLabel); - } - - // Collect output variables - const output: Record = {}; - if (flow.variables) { - for (const v of flow.variables) { - if (v.isOutput) output[v.name] = variables.get(v.name); - } - } - const durationMs = Date.now() - run.startTime; - // [#15944] THE RUN IS OVER AND IT SUCCEEDED. Everything from - // here to the return is BOOKKEEPING ABOUT that fact, and the - // `catch` below this `try` exists for NODE failures — so a - // throw out of the history write was handled as though a node - // had thrown: the arm journalled a repair snapshot, stamped - // `status: 'stranded'`, answered `success: false`, and - // {@link restoreConsumedSuspension} then honoured that - // snapshot and re-armed the pause, so the NEXT resume RE-RAN - // every node after it. Measured: `tail` ran twice. - // - // The guard restores the invariant `recordLog`'s own doc - // states two screens down — "a history write must NEVER block - // or break the run that produced it" — which that call was - // relied upon to keep and did not. - // - // Two statements inside `recordLog` reach here on the terminal - // path, and neither is hypothetical: `store.recordTerminal`, - // whose SYNCHRONOUS throw escapes because the - // `void write.catch(...)` beneath it only ever sees a returned - // promise's rejection (both shipped stores are `async` and - // cannot; the interface is exported, optional, and - // host-implementable, and a store returning a non-thenable - // makes `write.catch` itself a synchronous TypeError), and the - // run-summary line `this.logger.info(line, meta)`, on by - // default and calling a HOST-INJECTED logger. - // - // ⛔ NOT a widening of anything: no exit gains a status it - // did not have, the node-failure arm below is untouched, and - // the failure is REPORTED rather than swallowed — see the - // catch. ⛔ And ⛔ deliberately not fixed at - // `restoreConsumedSuspension`: that verb judged correctly on - // the evidence it was handed; the evidence is what was wrong, - // and no journal is written for a completed run at all now. - let logged: ExecutionLogEntry | undefined; - try { - logged = this.recordLog({ - id: runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - status: 'completed', - startedAt: run.startedAt, - completedAt: new Date().toISOString(), - durationMs, - trigger: buildRunTrigger(context), - steps, - output, - }, context); - } catch (bookkeeping) { - // #4632 verdict: DURABILITY, so `error` — the caller is - // told the truthful thing (the run completed), which is - // exactly what makes the rest invisible from the outside: - // the terminal history row never landed, nothing retries - // it, and no envelope carries a word about it. After the - // next restart the run is invisible to the Runs surfaces - // and the approvals sweeps read the hole — - // `inspectStrandedRequests` reads "no suspension + no - // terminal row" as a STRANDED request, and - // `releasePendingForTerminalRuns` reads "no terminal row" - // as still-alive. Consequence and fix in the first line, - // per AGENTS.md. Said ONCE per run, not once per failed - // write. - // - // THIRD argument per `error(message, error?, meta?)`; the - // `Error` slot stays empty on purpose (#5575), and the - // thrown text goes to the structured slot rather than into - // the message (#6499). - this.logger.error( - `[Automation] run '${runId}' of flow '${run.flowName}' COMPLETED successfully but its ` + - `run-history bookkeeping threw, so its terminal history row never landed — nothing ` + - `retries it, the caller is told the run succeeded, and after the next restart this ` + - `run is invisible to the Runs surfaces while the approvals sweeps read it as ` + - `stranded and never-finished. The run itself is COMPLETE and must NOT be repaired ` + - `or re-run. Fix the history failure in this record's meta.`, - undefined, - describeThrownForLog(bookkeeping), - ); - } - // [#15944] Recomputed when the guard above had to abandon - // `recordLog`: the same pure function of the same steps that - // `recordLog`'s own first statement runs, so the two spellings - // cannot disagree. Same shape as the strand arm's below. - const summary = logged?.summary ?? summarizeRun(steps); - - // ── Subflow up-bubble (nested pause): this run was a subflow - // child whose parent suspended awaiting it. Auto-resume the - // parent with our output, mapped like the synchronous path. - // Skipped when the DELEGATION path drives the chain (it - // continues the parent itself). Best-effort: the child's own - // completion stands even if the parent continuation fails. - if (!skipBubble) { - await this.bubbleToParent(run, output, summary); - } - - // Surface the flow's friendly completion message so a screen-flow - // runner shows it instead of a generic "Done". `summary` (#4354) - // covers the WHOLE run — the steps before the pause and after it - // are one log, so a resumed approval reports what it did in total. - return { - success: true, - output, - durationMs, - successMessage: flow.successMessage, - flowLabel: flow.label, - summary, - }; - } catch (err: unknown) { - // [#15788] A resumed run reached a refusing `end` — the SECOND - // producer, and the one a screen flow actually takes: a wizard - // that collects an answer and then refuses on it leaves through - // here, never through `execute()`'s exit. Tested first, beside - // the re-suspend, for the same reason it is tested first there. - if (isRefusalSignal(err)) { - const refused = this.finishRefusedRun({ - runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - startedAt: run.startedAt, - durationMs: Date.now() - run.startTime, - steps, flow, variables, - refusalMessage: err.message, context, - }); - // [#18714] UP-BUBBLE LEG. This run is terminal and the - // caller holds ITS id — but if it was a subflow CHILD, some - // ancestor is still parked at the `subflow` / `map` node - // that started it, and nothing else in the engine will ever - // move that ancestor: `bubbleToParent` was called on the - // completion path alone, so a child resumed to a refusal - // returned from here having resolved exactly one of the two - // runs it is responsible for. Measured: the child row read - // `refused` while its parent stayed `paused` and stayed in - // `listSuspendedRuns()` indefinitely — a leaked run, not a - // wrong answer, which is why it fails DIFFERENTLY from the - // delegated leg above and needs its own arm. - // - // The same call the completion path makes, with the refusal - // attached: the parent is genuinely resumed — it consumes - // its pause, records its own terminal row and bubbles to - // ITS parent in turn, so a chain of any depth resolves by - // the same induction completions already rely on. ⛔ Not a - // direct walk like `failAncestors`: that verb exists for a - // cascade in which no ancestor can be resumed at all, and - // it records them `failed` — the wrong word here. - // - // `output` and `summary` are the refused run's own, for the - // reason `finishRefusedRun` collects them: the nodes before - // the refusal really ran, and a parent that refuses must - // still be able to report what its child did. - // - // `skipBubble` is honoured exactly as on the completion - // path — under the DELEGATED leg the parent's frame is the - // caller, and it raises its own refusal at the throw site - // above. Bubbling here as well would resume the parent - // twice. - if (!skipBubble) { - await this.bubbleToParent( - run, - (refused.output ?? {}) as Record, - refused.summary, - { message: err.message }, - ); - } - return refused; - } - // Re-suspended at a downstream node: persist a fresh continuation. - if (isSuspendSignal(err)) { - const durationMs = Date.now() - run.startTime; - // #7639 — the re-suspend half of the same rule as the - // initial-execution site above: one snapshot, both consumers. - // A multi-stage approval re-pauses HERE on every stage but the - // first, so covering only the other site would leave every - // stage after stage 1 — the ones an operator actually needs to - // inspect — unreadable. - const variablesSnapshot = Object.fromEntries(variables); - await this.persistSuspendedRun({ - ...run, - // The segment's identity, not the one the run parked - // with last time: after a decision switch (ruling A on - // `FlowSchema.runAs`) the next pause must continue as - // the decider, whatever resumes it — never fall back to - // the trigger user. Identical to `run.context` when no - // switch happened. - context, - nodeId: err.nodeId, - nodeType: err.nodeType, - variables: variablesSnapshot, - steps, - correlation: err.correlation, - screen: err.screen, - }); - this.recordLog({ - id: runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - status: 'paused', - startedAt: run.startedAt, - durationMs, - trigger: buildRunTrigger(context), - steps, - variables: variablesSnapshot, - }, context); - return { success: true, status: 'paused', runId, durationMs, screen: err.screen, flowLabel: flow.label }; - } - - const errorMessage = err instanceof Error ? err.message : String(err); - const durationMs = Date.now() - run.startTime; - // [#13909] THE seam this card is about. The pause was consumed - // above, this node merely threw, and the record below makes the - // run terminal — so from here on nothing in the engine can move - // it: `resume` answers RUN_NOT_FOUND, `cancelRun` is a no-op on - // it, and no other public method takes it anywhere EXCEPT the - // deliberate operator exit, {@link restoreConsumedSuspension}. - // Journal the suspension that was consumed so that exit has - // something to put back. - // - // ⛔ Not a repair, and nothing here re-arms anything: the run - // stays `failed`, no suspension exists after this line, and the - // ordering above is untouched. It is the evidence a repair - // needs, written at the only moment it still exists. - const consumed = this.journalConsumedSuspension( - run, - stepCountAtPause, - errorMessage, - variablesAtPause, - ); - // [#15555] From the line above, "this run is repairable" is a - // FACT: a snapshot exists and `restoreConsumedSuspension` puts - // it back. Everything from here to the `status: 'stranded'` - // stamp below is BOOKKEEPING ABOUT that fact — and none of it - // may be allowed to delete the fact by throwing. - // - // It could, and the direction is the dangerous one. The stamp - // is the ONLY thing that tells a consumer the run is - // repairable (`plugin-approvals` derives its operator-facing - // `repairable` from it verbatim: `status === 'stranded'`), and - // two statements in here throw out of the whole arm. Both live - // in `recordLog`'s terminal path: the run-summary line - // (`logger.info`, on by default) and `store.recordTerminal`, - // whose SYNCHRONOUS throw escapes — the `void write.catch(...)` - // beneath that call only ever sees a returned promise's - // rejection. Then `failAncestors` awaits a walk that can throw. - // A throw anywhere in that window replaced a truthful - // `repairable: true` with `false`, which does not merely lose - // information: it tells an operator NOT to attempt a repair - // that succeeds. A false negative on a repair instruction is - // worse than silence, and it is the opposite of the direction - // everybody checks for. - // - // [#15944] THE SAME TWO STATEMENTS HAVE A SECOND CONSEQUENCE, - // and it is the direction everybody DOES check for. On the - // COMPLETION path above, a throw out of the same `recordLog` - // fell into this arm on a run whose nodes ALL succeeded — a - // false `true`: a journal and a `stranded` stamp for a run - // that finished, whose "repair" re-armed it and re-ran every - // node after the pause. That half is guarded at its own site - // (see the completion path), NOT here: this arm stays the - // node-failure arm, and reaching it at all was the defect. - // - // ⛔ This is NOT "assume repairable when the failure is - // unknown" — that would invert the honest default and promise - // a repair for a lost run. The guard opens AFTER the journal, - // so only a run that demonstrably HAS a snapshot can reach the - // stamp: a throw from `journalConsumedSuspension` itself still - // propagates, and every exit above the consumption point is - // untouched and still carries no status at all. - // - // ⛔ And the journal is NOT moved down to sit beside the stamp - // instead. `recordLog` is what carries the snapshot into the - // durable row, so journalling after it would leave a - // `recordLog` failure with NO snapshot anywhere — converting - // this false negative into a TRUE one by destroying the repair - // rather than by reporting it. - let logged: ExecutionLogEntry | undefined; - try { - logged = this.recordLog({ - id: runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - status: 'failed', - startedAt: run.startedAt, - completedAt: new Date().toISOString(), - durationMs, - trigger: buildRunTrigger(context), - steps, - error: errorMessage, - }, context, consumed.run); - // Subflow chain: a child failing terminally fails every - // ancestor awaiting it — they can never be resumed otherwise. - // The delegation path handles its own level (skipBubble). - if (!skipBubble) { - // [#15222] `runId` names the run that stranded two - // statements above, so the walk journals every ancestor - // whose pause this cascade consumes and the chain is - // re-armable to its full depth, leaf-first. - await this.failAncestors(run.context, errorMessage, runId); - } - } catch (bookkeeping) { - // #4632 verdict: DURABILITY, so `error` — the caller is - // told a truthful, actionable thing (the run stranded, and - // it is repairable), which is exactly what makes the rest - // invisible from the outside: the terminal history row - // never landed and/or the ancestor cascade stopped - // part-way, nothing retries either, and no envelope - // carries a word about it. Consequence and fix in the - // first line, per AGENTS.md. Said ONCE per stranded run, - // not once per failed write. - // - // THIRD argument per `error(message, error?, meta?)`; the - // `Error` slot stays empty on purpose (#5575), and the - // thrown text goes to the structured slot rather than into - // the message (#6499). - this.logger.error( - `[Automation] run '${runId}' of flow '${run.flowName}' is STRANDED and its ` + - `post-strand bookkeeping threw, so its terminal history row never landed ` + - `and/or its subflow ancestors were not failed — nothing retries either, and ` + - `the run reads healthy to the Runs surfaces and the approvals sweeps. The ` + - `strand itself IS reported and repairable right now: restore it with ` + - `restoreConsumedSuspension('${runId}') before this process restarts, which ` + - `drops the in-memory journal this repair rides on. Fix the failure in this ` + - `record's meta.`, - undefined, - describeThrownForLog(bookkeeping), - ); - } - // Surface the flow's friendly error message (the raw error stays - // in `error` for logs/diagnostics). - return { - success: false, - error: errorMessage, - durationMs, - // [#13937] STRANDED, said by the producer that knows — the - // services half of the shape-4 ruling (maintainer - // 2026-09-01): the ordering stays, and the condition it - // leaves behind carries the platform-level name #14384 put - // on the contract. Stamped HERE and on no other exit, - // because this is the one exit that journalled a snapshot - // two statements up: the pause a durable decision was - // waiting on is gone, the run is recorded `failed`, and it - // can be re-armed only by {@link restoreConsumedSuspension} - // — never by `resume` (which now answers RUN_NOT_FOUND), - // never automatically. Terminal exactly like `'failed'` - // and distinct from it on purpose: `'failed'` says the run - // ran and was rejected; this says a recorded continuation - // stopped mid-flight and an operator has something to - // repair. The verdict and the journal are one fact stated - // twice — `stranded-run-status.test.ts` holds them equal. - // - // ⚠️ The run's RECORDED status — the log entry above and - // the durable history row — stays `'failed'`: that is - // `ExecutionStatus` (`@objectstack/spec`, - // `automation/execution.zod.ts`), a vocabulary the ruling - // did not widen; the durable discriminator for this - // condition is the snapshot the terminal row carries. - // - // ⛔ The cascade-failed ancestors above are NOT stranded: - // `failSuspendedRun` consumes their pauses and journals - // nothing, so no verb can re-arm them. A different (and - // worse) condition, which this stamp must not claim. - status: 'stranded', - errorMessage: flow.errorMessage, - flowLabel: flow.label, - // [#15555] Recomputed when the guard above had to abandon - // `recordLog`: the same pure function of the same steps - // that `recordLog`'s own first statement runs, so the two - // spellings cannot disagree. - summary: logged?.summary ?? summarizeRun(steps), - // [#24914, ruling B′] As `execute()`'s failure exit: the refusal - // the node after the consumed pause met, off the throw's brand. The - // screen-flow leg (a write behind a screen) reaches its caller HERE. - ...readCarriedRefusal(err), - }; - } - } finally { - this.resuming.delete(runId); - } + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + resuming: this.resuming, + loadSuspendedRunStrict: (...a) => this.loadSuspendedRunStrict(...a), + logger: this.logger, + flows: this.flows, + settleUnresumableSuspension: (...a) => this.settleUnresumableSuspension(...a), + creditChildRun: (...a) => this.creditChildRun(...a), + loadSuspendedRun: (...a) => this.loadSuspendedRun(...a), + resumeInternal: (...a) => this.resumeInternal(...a), + persistSuspendedRun: (...a) => this.persistSuspendedRun(...a), + failSuspendedRun: (...a) => this.failSuspendedRun(...a), + consumedSuspensions: this.consumedSuspensions, + buildSubflowResumeSignal: (...a) => this.buildSubflowResumeSignal(...a), + refuseInvalidScreenInput: (...a) => this.refuseInvalidScreenInput(...a), + cloneVariablesAtPause: (...a) => this.cloneVariablesAtPause(...a), + resolveDecisionSegmentContext: (...a) => this.resolveDecisionSegmentContext(...a), + claimAdvance: (...a) => this.claimAdvance(...a), + evictConsumedSuspension: (...a) => this.evictConsumedSuspension(...a), + forgetSuspendedRun: (...a) => this.forgetSuspendedRun(...a), + executeNode: (...a) => this.executeNode(...a), + traverseNext: (...a) => this.traverseNext(...a), + recordLog: (...a) => this.recordLog(...a), + bubbleToParent: (...a) => this.bubbleToParent(...a), + finishRefusedRun: (...a) => this.finishRefusedRun(...a), + journalConsumedSuspension: (...a) => this.journalConsumedSuspension(...a), + failAncestors: (...a) => this.failAncestors(...a), + }; + return resumeInternal.call(host, runId, signal, skipBubble, childSummary, childRefusal); } /** diff --git a/packages/services/service-automation/src/run-resume.ts b/packages/services/service-automation/src/run-resume.ts index fae17cc5006..c0065223582 100644 --- a/packages/services/service-automation/src/run-resume.ts +++ b/packages/services/service-automation/src/run-resume.ts @@ -21,13 +21,25 @@ import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; import { runIsUnscopedUserMode, flowTouchesData, runOrgUserIds, runTriggerUserId, type SeededRunContext } from './runtime-identity.js'; import { refuseNode } from './guard-refusal.js'; import { describeThrownForLog } from './thrown-cause-diagnostics.js'; -// The value below (`buildRunTrigger`) is read only inside the function bodies, -// at call time, never while a module is evaluating, so the import cycle with +import { readCarriedRefusal } from './inner-refusal.js'; +import { summarizeRun } from './run-summary.js'; +// The values below (`buildRunTrigger`, `FlowRefusalSignal`, `isRefusalSignal`, +// `isSuspendSignal`, `isRetryableResumeRefusal`, `carriesSubflowChildOutput` +// and `applyResumeSignal`) are read only inside the function bodies, at call +// time, never while a module is evaluating, so the import cycle with // ./engine.js is safe under ESM live bindings; the tsup bundle is one file and // has no cycle. import { buildRunTrigger, + FlowRefusalSignal, + isRefusalSignal, + isSuspendSignal, + isRetryableResumeRefusal, + carriesSubflowChildOutput, + applyResumeSignal, type AutomationEngine, + type ChildRunRefusal, + type ExecutionLogEntry, type NodeExecutionResult, type StepLogEntry, type SubflowParentStrand, @@ -77,6 +89,20 @@ export interface RunResumeHost { readonly consumedSuspensions: AutomationEngine['consumedSuspensions']; readonly releaseSuspension: AutomationEngine['releaseSuspension']; readonly subflowParentStrands: AutomationEngine['subflowParentStrands']; + readonly resuming: AutomationEngine['resuming']; + readonly flows: AutomationEngine['flows']; + readonly settleUnresumableSuspension: AutomationEngine['settleUnresumableSuspension']; + readonly creditChildRun: AutomationEngine['creditChildRun']; + readonly persistSuspendedRun: AutomationEngine['persistSuspendedRun']; + readonly buildSubflowResumeSignal: AutomationEngine['buildSubflowResumeSignal']; + readonly refuseInvalidScreenInput: AutomationEngine['refuseInvalidScreenInput']; + readonly cloneVariablesAtPause: AutomationEngine['cloneVariablesAtPause']; + readonly resolveDecisionSegmentContext: AutomationEngine['resolveDecisionSegmentContext']; + readonly executeNode: AutomationEngine['executeNode']; + readonly traverseNext: AutomationEngine['traverseNext']; + readonly bubbleToParent: AutomationEngine['bubbleToParent']; + readonly finishRefusedRun: AutomationEngine['finishRefusedRun']; + readonly failAncestors: AutomationEngine['failAncestors']; } /** The body of {@link AutomationEngine.resolveRunContext}, moved unedited; that method's docblock is its contract. */ @@ -875,3 +901,925 @@ export function takeSubflowParentStrand(this: Pick, + runId: string, + // Never `undefined` past the public door: `resume` normalises an + // absent caller signal to `{}` (commit 7307191db), and the engine's own + // continuations (subflow delegation / up-bubble, `map` re-entry) + // always hand over a built signal. Typed so, the chokepoints below + // cannot grow a falsy-signal branch again. + signal: ResumeSignal, + skipBubble: boolean, + childSummary?: FlowRunSummary, + childRefusal?: ChildRunRefusal, +): Promise { + // Idempotency guard (set synchronously, before any await): reject a + // concurrent duplicate resume of the same run so side effects can't run + // twice. A duplicate that arrives *after* this one finishes finds no + // suspended run and returns the "no suspended run" error below. + if (this.resuming.has(runId)) { + return { success: false, code: 'RESUME_IN_PROGRESS', error: `Run '${runId}' is already being resumed` }; + } + this.resuming.add(runId); + try { + // Hot path: suspended in this process. Cold path: rehydrate from the + // durable store (e.g. the process restarted since the pause, ADR-0019). + // + // Strict load: a store that cannot be READ must not report as + // "no such run" (#4420). A caller that already persisted a decision + // needs "retry when the store is back" to be distinguishable from + // "this run is gone for good" — same failure, opposite remedy. + let run: SuspendedRun | null; + try { + run = await this.loadSuspendedRunStrict(runId); + } catch (err) { + const message = (err as Error).message; + // #5912 — the LOG record's cause goes to `meta`, never into the + // message. `message` is the datasource DRIVER's own failure text + // and we do not control how many lines it has; + // `ObjectLogger.write()` adds one ` ` head per call, so + // a newline in it turns this ONE record into several physical + // lines of which only the first is greppable — the family of + // #5048 / #5575 / #5636 / #5661 / #5737, and cloud#971's shape. + // + // Third argument, per the `Logger` contract + // (`packages/spec/src/contracts/logger.ts`) + // `error(message, error?, meta?)`. NOT the second: that is the + // `Error` slot and a raw error there ships its whole stack on + // every record (#5575). `runId` stays in the message — it is the + // caller's own handle, the same call the family's other seams + // make for their ids, not the foreign text this fix is about. + // + // Level stays `error`: the run is on disk and the resume did not + // land, which is #4632's durability degradation exactly. + this.logger.error( + `[automation] durable suspended-run store unreachable while resuming '${runId}' — the suspension was ` + + `NOT consumed, so the run stays parked and this resume can be retried verbatim once the store is ` + + `reachable; the caller was told the same via the STORE_UNAVAILABLE result. The store's own failure ` + + `is in this record's meta.`, + undefined, + describeThrownForLog(err), + ); + // The RESULT envelope keeps the cause spliced in, verbatim and + // deliberately (#5636 made the same call for `degradedReason`): + // this is a structured return value a caller reads as a whole — + // over REST it is a JSON string field, never split on newlines — + // and PR #5911 already made wait-node put it into `meta` intact. + return { + success: false, + code: 'STORE_UNAVAILABLE', + error: `Durable suspended-run store unreachable for run '${runId}' — retry once the store is available: ${message}`, + }; + } + if (!run) { + return { success: false, code: 'RUN_NOT_FOUND', error: `No suspended run '${runId}'` }; + } + // [#8684] The two NEVER-DISPATCHED exits, classified HERE rather than + // guessed at the transport. Both mean the suspension is real but can + // never continue — the flow it belongs to is no longer registered, or + // the node it parked on was edited away underneath it. Nothing ran, + // so neither is a business rejection: they are the same terminal + // "this pause is gone for good" class as a missing suspension, and a + // transport answers all three **404**. + // + // Code-less until now, which is what forced the route to either + // answer them 200 (the defect this card is about) or sniff the result + // for `summary`/`durationMs` to tell them from a run that ran and + // failed — the tolerant-consumer shape PD #12 forbids. The producer + // is the only component that knows, so it says so. + // + // Deliberately the EXISTING `RUN_NOT_FOUND` rather than a new code: + // the remedy is identical (this run is finished; start a new one), + // the consumer branches on the 404 STATUS and not on the code + // (objectui's `interpretFlowResponse`), and `error` below still names + // which of the three causes it was, verbatim on the wire. A distinct + // code would be vocabulary nothing reads — add one the day a caller + // needs to branch on the difference. + // + // [#23963] And the verdict is carried onto the RUN, not only onto + // the answer: both exits go through + // {@link settleUnresumableSuspension}, which moves the parked row to + // `failed` with this same text in `error`. Answering "gone for + // good" while the row kept `status: 'paused'` left it resumable on + // every surface that reads the row, and outside retention, whose + // scope is terminal statuses only. + const flow = this.flows.get(run.flowName); + if (!flow) { + return this.settleUnresumableSuspension(run, `Flow '${run.flowName}' not found for run '${runId}'`); + } + const node = flow.nodes.find(n => n.id === run.nodeId); + if (!node) { + return this.settleUnresumableSuspension( + run, + `Suspended node '${run.nodeId}' no longer exists in flow '${run.flowName}'`, + ); + } + + // #4354 — up-bubble: the child that just finished did work this run + // is accountable for. Credit it to the step that suspended awaiting + // it, before traversal appends anything further. + this.creditChildRun(run.steps, run.nodeId, childSummary); + + // [#18714] The refusal this run must end on instead of continuing + // past its `subflow` / `map` node, if any. Seeded from the UP-BUBBLE + // leg (the child's own frame drove this resume and already knows), + // and set below by the DELEGATED leg (this frame drives the child + // and reads its answer). Both legs hand it to the ONE throw site + // inside the traversal `try` further down, so a refusal leaves this + // method through the same `finishRefusedRun` chokepoint every other + // producer uses — see `ChildRunRefusal`. + let pendingChildRefusal: ChildRunRefusal | undefined = childRefusal; + + // ── Subflow delegation (nested pause): this run is paused at a + // `subflow` node whose child run itself suspended. The caller's + // signal is meant for the node the CHILD paused on (its screen / + // approval / wait), so forward it down. The child resumes with + // bubbling off — when it completes, *this* invocation continues the + // parent from the subflow node with the child's output, using the + // same mapping as the synchronous path. + if (typeof run.correlation === 'string' && run.correlation.startsWith('subflow:')) { + 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 = await this.loadSuspendedRun(childRunId); + if (childRun) { + const childRes = await this.resumeInternal(childRunId, signal, true); + if (childRes.status === 'paused') { + // Child paused again (e.g. the next screen of a wizard). + // This run stays suspended; refresh its surfaced screen + // so a re-fetch (getSuspendedScreen) shows the new one. + if (childRes.screen && childRes.screen !== run.screen) { + await this.persistSuspendedRun({ ...run, screen: childRes.screen }); + } + return { + success: true, + status: 'paused', + runId, + durationMs: Date.now() - run.startTime, + screen: childRes.screen, + // THIS run's flow, not the child's: the caller + // addressed this run id and its runner names the + // flow it launched. The child only lends a screen. + flowLabel: flow.label, + }; + } + // [#14379] A child REFUSAL is not a child failure. The + // codes in {@link RETRYABLE_RESUME_REFUSAL_CODES} are this + // method's own answers for a resume that never ran: the + // child's screen contract is checked BEFORE + // `forgetSuspendedRun` precisely so "a rejected bag leaves + // the pause live and the legitimate submission still lands" + // (#4477), so the child is parked exactly where it was. + // + // Reading those as a terminal failure consumed the PARENT's + // pause over a mistyped form field — and the screen-flow + // path is where a caller holds ONE stable run id, the + // parent's, and posts every wizard step to it. The run was + // gone, the still-paused child orphaned with nothing left + // to bubble into, the caller told `400 FLOW_FAILED` ("it + // ran and was rejected") for something that never ran, and + // their corrected retry on that same id answered + // `RUN_NOT_FOUND`. + // + // The child's envelope is answered VERBATIM but for the + // parent's `durationMs`. Propagating the `code` is half the + // fix: a code-less envelope is exactly what forced the + // transport onto `400 FLOW_FAILED`, and leaving both pauses + // alive while still answering one repairs the state and + // leaves the caller equally misled. The child's `error` is + // the actionable half — `Screen field "kind" is required` — + // where the failure text below names neither the problem + // nor anything a caller can act on. Nothing else moves: + // neither pause is consumed, and the parent's surfaced + // screen needs no refresh because the child did not + // advance. + if (!childRes.success && isRetryableResumeRefusal(childRes.code)) { + return { ...childRes, durationMs: Date.now() - run.startTime }; + } + if (!childRes.success) { + const error = `subflow run '${childRunId}' (${childRun.flowName}) failed: ${childRes.error ?? 'unknown error'}`; + // [#15222] This frame consumes its OWN pause here, and + // before this card journalled nothing for it — so a + // child that stranded left a repairable leaf under an + // ancestor no verb could reach. Journalled exactly when + // the child is repairable, asked of the journal rather + // than of the child's `status`: a child that itself + // cascade-failed from a deeper strand carries no status + // word at all (see the return below) and is repairable + // all the same, which is what makes a chain deeper than + // two levels re-armable by induction. + // + // ⛔ And the envelope below is UNCHANGED — no status is + // stamped on it. `'stranded'` belongs to the run that + // consumed its own pause and then threw downstream; + // stamping it here would tell an operator to retry a + // recovery that cannot succeed, because nothing re-arms + // this run by resuming it. What reaches this run is + // {@link restoreConsumedSuspension}, through the + // journal one line up. + await this.failSuspendedRun( + run, + error, + this.consumedSuspensions.has(childRunId) ? childRunId : undefined, + ); + return { success: false, error, durationMs: Date.now() - run.startTime, flowLabel: flow.label }; + } + // [#18714] DELEGATED-LEG REFUSAL. The child ran to a + // refusing terminal — an `end` declaring + // `outcome: 'refused'`, or a node of its own whose child + // refused — and answered `{ success: true, status: + // 'refused' }`, because *a refusal is a successful + // evaluation that says no*. The branch set above tests only + // `paused` and `!success`, so until this arm existed a + // refused child was neither and fell straight through the + // ordinary success exit below: measured, the parent + // returned `{ success: true, successMessage: … }`, its run + // row recorded `completed`, and the node downstream of the + // `subflow` RAN. That is the identical fail-open shape + // #18110 / #18555 closed on the synchronous leg, still open + // on the leg 「the one a screen flow actually takes」. + // + // ⛔ NOT folded into the `!childRes.success` arm above. That + // arm calls `failSuspendedRun` and records the parent + // `failed`; a refusal is not a failure — it must not be + // routable by a `fault` edge, must not consume retry budget + // and must not be counted in `nodes[].failures`, all of + // which the failure arm would confer. The parent's terminal + // row must read `refused`, the same word the child's does. + // + // Recorded, ⛔ not thrown here. The pause is not consumed + // yet — `claimAdvance` / `forgetSuspendedRun` are still + // below — and a refusal that unwound from this position + // would leave the parent's own suspension live while its + // run row said terminal. The throw site inside the + // traversal `try` is past the consumption, which is why + // both legs converge there. + // + // The mapping below still runs, deliberately: the child's + // declared outputs really were produced and the nodes + // before its refusal really ran, so the parent's answer + // must not depend on HOW the child ended — the same call + // the synchronous `subflow` arm makes when it returns the + // success envelope PLUS `refuse`. + if (childRes.status === 'refused') { + pendingChildRefusal = { message: childRes.refusalMessage }; + } + // Child completed — continue below with its output as the + // resume signal (replaces the caller's signal, which the + // child already consumed). + signal = this.buildSubflowResumeSignal(childRun.context, childRes.output); + // #4354 — down-delegation is the other way a child's work + // lands under a parent step written at suspend time. + this.creditChildRun(run.steps, run.nodeId, childRes.summary); + } else if (carriesSubflowChildOutput(signal)) { + // The engine-built UP-BUBBLE — the HEALTHY path, and the + // one this `else` used to describe as a degradation. The + // child COMPLETED and `bubbleToParent` resumed this run + // with the child's mapped output, so the miss above is the + // SUSPENSION lookup missing (a completed run has no + // suspension), not the child having disappeared, and the + // parent continues *with* the child's output — which is + // what the old sentence denied outright. + // + // `debug`, for the same reason as the lost-advance-claim + // record further down: an ordinary outcome logged at `warn` + // on every healthy subflow completion is the cry-wolf + // shape — the operator who reads "child run is gone" on + // each one stops reading the line, and the one time it + // means what it says is the time it is ignored. Nothing + // else moves: this branch continues the parent exactly as + // before, and the degraded branch below keeps its text and + // its level. + this.logger.debug( + `[automation] run '${runId}' continues from subflow node '${run.nodeId}' with the output of ` + + `completed child run '${childRunId}' — carried on the engine-built up-bubble signal ` + + `(the child's suspension is consumed, which is why it is not loadable here).`, + ); + } else { + // The genuinely degraded case, and the only one the + // sentence below was ever true of: nothing suspended under + // the child id AND no carried output, so this run really + // does continue without the child's output. #4632 verdict: + // FUNCTIONAL — stays `warn`. + this.logger.warn( + `[automation] run '${runId}' is paused at subflow node '${run.nodeId}' but child run '${childRunId}' ` + + `is gone — continuing without child output`, + ); + } + } + + // The SCREEN contract (#4477). A run parked on a `screen` node + // declared exactly which keys it collects and which are required; + // until this ran, `resume` folded any bag at all straight into the + // variables, so a caller that skipped the dialog bypassed every + // `required` the author wrote. Checked here — beside the engine's + // other resume refusals and BEFORE the suspension is consumed — so + // a rejected bag leaves the pause live and the legitimate + // submission still lands. + const screenRefusal = this.refuseInvalidScreenInput(run, runId, signal); + if (screenRefusal) return screenRefusal; + + // Restore the variable context and fold the signal in — the ONE + // place a resume signal reaches the variable map. Runs BEFORE the + // suspension is consumed, so a rejected signal changes nothing: + // the pause stays live and the legitimate continuation still lands. + const variables = new Map(Object.entries(run.variables)); + const rejected = applyResumeSignal(variables, signal, run.nodeId); + if (rejected.length) { + // #6654 — the rejected names are the CALLER's resume-signal + // keys (nothing constrains them against newlines), so they + // ride the structured slot, never the message; see + // `forgetSuspendedRun`'s catch for the full mechanism (#6299). + // The returned INVALID_SIGNAL envelope below is caller-facing + // refusal text, not a log record — it keeps naming the + // variables (the envelope class is ruled elsewhere). + // #4632: FUNCTIONAL — stays `warn`. + this.logger.warn( + `[automation] refused resume of run '${runId}': signal writes engine-internal ` + + `variable(s) — the rejected names are in this record's meta.`, + { rejected }, + ); + return { + success: false, + code: 'INVALID_SIGNAL', + error: + `Resume signal may not set engine-internal variables (${rejected.join(', ')}) — ` + + `names starting with '$' (or containing '.$') are reserved by the flow engine`, + }; + } + + // [#13909] How long the step log was AT THE PAUSE, read before + // anything downstream runs. `steps` below is the SAME array + // `traverseNext` appends to, so by the time the catch arm builds a + // restore snapshot the pause's own step log is no longer + // recoverable from it — this integer is what trims it back. One + // `int` on every resume and nothing else: the snapshot itself is + // built only on the failure path. + // + // ⛔ Deliberately NOT a change to the ordering. The consumption + // below still precedes the traversal, `forgetSuspendedRun` is + // untouched, and `hasSuspendedRun` still answers false for the + // whole traversal window (pinned in + // `consumed-suspension-restore.test.ts`). That order is RULED, + // not inherited: #13937 shape 4 (maintainer 2026-09-01) keeps it + // for exactly-once across a crash, excludes the flip (shape 2) + // unless a durable claim/lease lands with it, and excludes + // consume-then-re-arm (shape 3) outright. The exit from the state + // it leaves behind is `restoreConsumedSuspension`, and the state + // itself is named on the result (`status: 'stranded'`, below). + const stepCountAtPause = run.steps.length; + // [#15660] …and the variables AT THE PAUSE, for the same reason and + // at the same moment, because `run.steps.length` is not the only + // thing the failed attempt can move under the journal. + // + // `journalConsumedSuspension` promises its snapshot VERBATIM — "the + // state at the pause, not the state the failed attempt was working + // from" — and delivered that only for the TOP-LEVEL keys. The scope + // the downstream nodes run against is built one line below as + // `new Map(Object.entries(run.variables))`, which copies the keys + // and SHARES every value object. A node that holds state in the + // scope and updates it in place — `map` keeps `.$mapState` + // and is, by a re-derived census, the only executor that does — + // therefore writes straight through into the object the journal + // later hands an operator as the pause. + // + // Measured, not read (`consumed-suspension-snapshot-aliasing.test.ts`): + // the durable row held `started: 1` at the pause and the restore put + // back `started: 99`. ⛔ And the two placements the card proposed + // are both refuted by that same measurement: + // - "deep copy at SNAPSHOT time" — `InMemorySuspendedRunStore` + // already JSON round-trips on save AND load, so this run's + // variables ARE a private deep copy before a single downstream + // node runs. It reproduces anyway: the aliasing that carries the + // mutation is minted HERE, on the resume, not at the suspend. + // - "copy at RESTORE time" — the mutation lands before the node + // throws, so the journal is already corrupt when it is written; + // anything copied at or after that point copies the corruption. + // The copy has to be taken before the failed attempt runs, which is + // this line. + // + // Cost, measured rather than assumed (the card flagged it NOT + // MEASURED): 5.8 µs for a typical scope, and one suspend+resume + // round trip against a configured store already performs FOUR full + // clones of this same payload (save ×1, load ×2, recordTerminal ×1), + // so this is a fifth on a path whose production cost is a durable + // round trip. ⛔ Deliberately NOT the JSON clone the journal's own + // docblock ruled out: that objection was that it would run INSIDE a + // catch arm already handling a failure, where a circular value would + // throw the operator's repair away. Here it is on the happy path, + // where a throw is just a value we decline to copy. + const variablesAtPause = this.cloneVariablesAtPause(run, runId); + + // Ruling A on `FlowSchema.runAs`: who runs the nodes after this + // pause — the decider, when the owning service's signal names one + // (with their own grants on a `runAs:'user'` run, as the attribution + // user of a still-elevated `runAs:'system'` run); otherwise the + // identity the run paused with. + // Resolved BEFORE the claim, so nothing here can run after the + // suspension is consumed and leave it consumed with no continuation; + // the persisted `run.context` is left as it was, and the repair + // journal therefore still snapshots the pause exactly as it parked. + const segmentContext = await this.resolveDecisionSegmentContext(run, signal); + + // Consume the suspension *before* running downstream work — a run + // resumes exactly once per pause, and a duplicate resume after a + // partial restart must not double-run side effects. (Folding the + // signal above is pure in-memory work, not downstream work.) + // This is also where the paused node learns its pause is over and + // disarms what it armed on entry (#5512) — see forgetSuspendedRun. + // + // [#14333] And the consumption is now a CLAIM, not an assertion. + // `resuming` above answered "is this process already resuming it"; + // this answers "did any process get here first", against the shared + // store, atomically. Placed exactly where the unconditional consume + // was: every refusal above it still refuses without consuming, and + // the ordering #13937 ruled to keep is untouched. + const claim = await this.claimAdvance(run); + if (claim.kind === 'lost') { + // An ORDINARY outcome, not a degradation: this is the guard + // doing its job, and the caller is told in the result. `debug` + // rather than `warn` for exactly that reason — a losing racer + // per approval on a busy any-of level would otherwise be a + // steady `warn` stream describing correct behaviour. + this.logger.debug( + `[automation] resume of run '${runId}' lost the advance claim at node '${run.nodeId}' — ` + + `another replica consumed this suspension first; nothing was dispatched here.`, + ); + // The SAME code the in-process guard answers, deliberately: the + // observation differs (another process, not this one) but the + // remedy is identical — a concurrent resume already has this + // run, the caller's own recorded outcome stands, and nothing + // here should be retried. `plugin-approvals` already branches + // on it that way (`resumeRecordedOutcome`), and the transport + // already maps it to 409. A distinct code would be vocabulary + // nothing reads — add one the day a caller needs the + // difference. + // [#15832] The store's compare-and-set ANSWERED: no row is parked + // where this replica read it. Whatever snapshot this process + // still holds for the run is stale by construction — it names a + // node the run has left — so it stops being published by the two + // listings. ⛔ NOT `forgetSuspendedRun`: nothing was consumed + // here, and firing that choke point would tear the pause down a + // second time in this process on top of the winner's own. + this.evictConsumedSuspension(runId); + return { + success: false, + code: 'RESUME_IN_PROGRESS', + error: + `Run '${runId}' is already being resumed — another replica consumed its suspension ` + + `at node '${run.nodeId}' first`, + }; + } + if (claim.kind === 'unavailable') { + // Same envelope, same reason as the strict load's failure + // above: existence is UNKNOWN, not "gone for good", and the + // identical call is expected to work once the store recovers + // (#4420). + // + // The text states only what is KNOWN. The strict load's failure + // can say the suspension was not consumed, because it fails + // before anything is consumed; this one cannot — a claim throws + // just as readily after a committed delete as before one, so + // "NOT consumed" would be a claim about a fact this seam does + // not have. What it does know is that this resume did not + // continue the run, and that a retry settles the rest. + return { + success: false, + code: 'STORE_UNAVAILABLE', + error: + `Durable suspended-run store unreachable while claiming the advance for run '${runId}' — ` + + `this resume did NOT continue the run, and whether the store consumed the suspension is ` + + `UNKNOWN (a failure can arrive after a committed delete). Retry once the store is ` + + `available: a retry that finds the run still parked continues it, and one that finds the ` + + `suspension gone answers RUN_NOT_FOUND: ${claim.message}`, + }; + } + await this.forgetSuspendedRun(run, 'resumed', claim.kind === 'claimed'); + + const steps = run.steps; + const context = segmentContext; + + try { + // [#18714] The child this run is parked on REFUSED — on either + // resumed leg (see `ChildRunRefusal`). Thrown HERE, and the + // position is the point: + // + // - Past the consumption. `claimAdvance` and + // `forgetSuspendedRun` have run, so the parent's own pause is + // gone exactly as it is for every other way this resume can + // end. A refusal raised before them would record a terminal + // run while leaving its suspension live in + // `listSuspendedRuns()` — the very leak the up-bubble leg is + // filed for, moved one frame up. + // - Before the traversal. Nothing downstream of the awaiting + // node runs, which is the whole content of "a refusal stops + // the run". + // - As `FlowRefusalSignal`, so the `catch` below converts it + // through `finishRefusedRun` — ONE terminal shape, whichever + // producer raised it. ⛔ Deliberately not a second terminal + // exit of its own: this file's own history is a list of + // outcomes that became a function of WHICH ROUTE a run took. + // + // `run.nodeId` is the node that was awaiting the child — the + // `subflow` / `map` this frame is parked on, the same node the + // synchronous leg names on its `FlowRefusalSignal`. + if (pendingChildRefusal) { + throw new FlowRefusalSignal(run.nodeId, pendingChildRefusal.message); + } + // ── Map re-entry (sequential multi-instance, ADR-0037 A2). + // A run paused at a `map` node (correlation `map:`) + // does NOT continue past the node on resume — it RE-RUNS the + // node so the executor can record the just-completed unit and + // start the next item. The default path continues past the node. + if (typeof run.correlation === 'string' && run.correlation.startsWith('map:')) { + await this.executeNode(node, flow, variables, context, steps); + } else { + await this.traverseNext(node, flow, variables, context, steps, signal.branchLabel); + } + + // Collect output variables + const output: Record = {}; + if (flow.variables) { + for (const v of flow.variables) { + if (v.isOutput) output[v.name] = variables.get(v.name); + } + } + const durationMs = Date.now() - run.startTime; + // [#15944] THE RUN IS OVER AND IT SUCCEEDED. Everything from + // here to the return is BOOKKEEPING ABOUT that fact, and the + // `catch` below this `try` exists for NODE failures — so a + // throw out of the history write was handled as though a node + // had thrown: the arm journalled a repair snapshot, stamped + // `status: 'stranded'`, answered `success: false`, and + // {@link restoreConsumedSuspension} then honoured that + // snapshot and re-armed the pause, so the NEXT resume RE-RAN + // every node after it. Measured: `tail` ran twice. + // + // The guard restores the invariant `recordLog`'s own doc + // states two screens down — "a history write must NEVER block + // or break the run that produced it" — which that call was + // relied upon to keep and did not. + // + // Two statements inside `recordLog` reach here on the terminal + // path, and neither is hypothetical: `store.recordTerminal`, + // whose SYNCHRONOUS throw escapes because the + // `void write.catch(...)` beneath it only ever sees a returned + // promise's rejection (both shipped stores are `async` and + // cannot; the interface is exported, optional, and + // host-implementable, and a store returning a non-thenable + // makes `write.catch` itself a synchronous TypeError), and the + // run-summary line `this.logger.info(line, meta)`, on by + // default and calling a HOST-INJECTED logger. + // + // ⛔ NOT a widening of anything: no exit gains a status it + // did not have, the node-failure arm below is untouched, and + // the failure is REPORTED rather than swallowed — see the + // catch. ⛔ And ⛔ deliberately not fixed at + // `restoreConsumedSuspension`: that verb judged correctly on + // the evidence it was handed; the evidence is what was wrong, + // and no journal is written for a completed run at all now. + let logged: ExecutionLogEntry | undefined; + try { + logged = this.recordLog({ + id: runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'completed', + startedAt: run.startedAt, + completedAt: new Date().toISOString(), + durationMs, + trigger: buildRunTrigger(context), + steps, + output, + }, context); + } catch (bookkeeping) { + // #4632 verdict: DURABILITY, so `error` — the caller is + // told the truthful thing (the run completed), which is + // exactly what makes the rest invisible from the outside: + // the terminal history row never landed, nothing retries + // it, and no envelope carries a word about it. After the + // next restart the run is invisible to the Runs surfaces + // and the approvals sweeps read the hole — + // `inspectStrandedRequests` reads "no suspension + no + // terminal row" as a STRANDED request, and + // `releasePendingForTerminalRuns` reads "no terminal row" + // as still-alive. Consequence and fix in the first line, + // per AGENTS.md. Said ONCE per run, not once per failed + // write. + // + // THIRD argument per `error(message, error?, meta?)`; the + // `Error` slot stays empty on purpose (#5575), and the + // thrown text goes to the structured slot rather than into + // the message (#6499). + this.logger.error( + `[Automation] run '${runId}' of flow '${run.flowName}' COMPLETED successfully but its ` + + `run-history bookkeeping threw, so its terminal history row never landed — nothing ` + + `retries it, the caller is told the run succeeded, and after the next restart this ` + + `run is invisible to the Runs surfaces while the approvals sweeps read it as ` + + `stranded and never-finished. The run itself is COMPLETE and must NOT be repaired ` + + `or re-run. Fix the history failure in this record's meta.`, + undefined, + describeThrownForLog(bookkeeping), + ); + } + // [#15944] Recomputed when the guard above had to abandon + // `recordLog`: the same pure function of the same steps that + // `recordLog`'s own first statement runs, so the two spellings + // cannot disagree. Same shape as the strand arm's below. + const summary = logged?.summary ?? summarizeRun(steps); + + // ── Subflow up-bubble (nested pause): this run was a subflow + // child whose parent suspended awaiting it. Auto-resume the + // parent with our output, mapped like the synchronous path. + // Skipped when the DELEGATION path drives the chain (it + // continues the parent itself). Best-effort: the child's own + // completion stands even if the parent continuation fails. + if (!skipBubble) { + await this.bubbleToParent(run, output, summary); + } + + // Surface the flow's friendly completion message so a screen-flow + // runner shows it instead of a generic "Done". `summary` (#4354) + // covers the WHOLE run — the steps before the pause and after it + // are one log, so a resumed approval reports what it did in total. + return { + success: true, + output, + durationMs, + successMessage: flow.successMessage, + flowLabel: flow.label, + summary, + }; + } catch (err: unknown) { + // [#15788] A resumed run reached a refusing `end` — the SECOND + // producer, and the one a screen flow actually takes: a wizard + // that collects an answer and then refuses on it leaves through + // here, never through `execute()`'s exit. Tested first, beside + // the re-suspend, for the same reason it is tested first there. + if (isRefusalSignal(err)) { + const refused = this.finishRefusedRun({ + runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + startedAt: run.startedAt, + durationMs: Date.now() - run.startTime, + steps, flow, variables, + refusalMessage: err.message, context, + }); + // [#18714] UP-BUBBLE LEG. This run is terminal and the + // caller holds ITS id — but if it was a subflow CHILD, some + // ancestor is still parked at the `subflow` / `map` node + // that started it, and nothing else in the engine will ever + // move that ancestor: `bubbleToParent` was called on the + // completion path alone, so a child resumed to a refusal + // returned from here having resolved exactly one of the two + // runs it is responsible for. Measured: the child row read + // `refused` while its parent stayed `paused` and stayed in + // `listSuspendedRuns()` indefinitely — a leaked run, not a + // wrong answer, which is why it fails DIFFERENTLY from the + // delegated leg above and needs its own arm. + // + // The same call the completion path makes, with the refusal + // attached: the parent is genuinely resumed — it consumes + // its pause, records its own terminal row and bubbles to + // ITS parent in turn, so a chain of any depth resolves by + // the same induction completions already rely on. ⛔ Not a + // direct walk like `failAncestors`: that verb exists for a + // cascade in which no ancestor can be resumed at all, and + // it records them `failed` — the wrong word here. + // + // `output` and `summary` are the refused run's own, for the + // reason `finishRefusedRun` collects them: the nodes before + // the refusal really ran, and a parent that refuses must + // still be able to report what its child did. + // + // `skipBubble` is honoured exactly as on the completion + // path — under the DELEGATED leg the parent's frame is the + // caller, and it raises its own refusal at the throw site + // above. Bubbling here as well would resume the parent + // twice. + if (!skipBubble) { + await this.bubbleToParent( + run, + (refused.output ?? {}) as Record, + refused.summary, + { message: err.message }, + ); + } + return refused; + } + // Re-suspended at a downstream node: persist a fresh continuation. + if (isSuspendSignal(err)) { + const durationMs = Date.now() - run.startTime; + // #7639 — the re-suspend half of the same rule as the + // initial-execution site above: one snapshot, both consumers. + // A multi-stage approval re-pauses HERE on every stage but the + // first, so covering only the other site would leave every + // stage after stage 1 — the ones an operator actually needs to + // inspect — unreadable. + const variablesSnapshot = Object.fromEntries(variables); + await this.persistSuspendedRun({ + ...run, + // The segment's identity, not the one the run parked + // with last time: after a decision switch (ruling A on + // `FlowSchema.runAs`) the next pause must continue as + // the decider, whatever resumes it — never fall back to + // the trigger user. Identical to `run.context` when no + // switch happened. + context, + nodeId: err.nodeId, + nodeType: err.nodeType, + variables: variablesSnapshot, + steps, + correlation: err.correlation, + screen: err.screen, + }); + this.recordLog({ + id: runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'paused', + startedAt: run.startedAt, + durationMs, + trigger: buildRunTrigger(context), + steps, + variables: variablesSnapshot, + }, context); + return { success: true, status: 'paused', runId, durationMs, screen: err.screen, flowLabel: flow.label }; + } + + const errorMessage = err instanceof Error ? err.message : String(err); + const durationMs = Date.now() - run.startTime; + // [#13909] THE seam this card is about. The pause was consumed + // above, this node merely threw, and the record below makes the + // run terminal — so from here on nothing in the engine can move + // it: `resume` answers RUN_NOT_FOUND, `cancelRun` is a no-op on + // it, and no other public method takes it anywhere EXCEPT the + // deliberate operator exit, {@link restoreConsumedSuspension}. + // Journal the suspension that was consumed so that exit has + // something to put back. + // + // ⛔ Not a repair, and nothing here re-arms anything: the run + // stays `failed`, no suspension exists after this line, and the + // ordering above is untouched. It is the evidence a repair + // needs, written at the only moment it still exists. + const consumed = this.journalConsumedSuspension( + run, + stepCountAtPause, + errorMessage, + variablesAtPause, + ); + // [#15555] From the line above, "this run is repairable" is a + // FACT: a snapshot exists and `restoreConsumedSuspension` puts + // it back. Everything from here to the `status: 'stranded'` + // stamp below is BOOKKEEPING ABOUT that fact — and none of it + // may be allowed to delete the fact by throwing. + // + // It could, and the direction is the dangerous one. The stamp + // is the ONLY thing that tells a consumer the run is + // repairable (`plugin-approvals` derives its operator-facing + // `repairable` from it verbatim: `status === 'stranded'`), and + // two statements in here throw out of the whole arm. Both live + // in `recordLog`'s terminal path: the run-summary line + // (`logger.info`, on by default) and `store.recordTerminal`, + // whose SYNCHRONOUS throw escapes — the `void write.catch(...)` + // beneath that call only ever sees a returned promise's + // rejection. Then `failAncestors` awaits a walk that can throw. + // A throw anywhere in that window replaced a truthful + // `repairable: true` with `false`, which does not merely lose + // information: it tells an operator NOT to attempt a repair + // that succeeds. A false negative on a repair instruction is + // worse than silence, and it is the opposite of the direction + // everybody checks for. + // + // [#15944] THE SAME TWO STATEMENTS HAVE A SECOND CONSEQUENCE, + // and it is the direction everybody DOES check for. On the + // COMPLETION path above, a throw out of the same `recordLog` + // fell into this arm on a run whose nodes ALL succeeded — a + // false `true`: a journal and a `stranded` stamp for a run + // that finished, whose "repair" re-armed it and re-ran every + // node after the pause. That half is guarded at its own site + // (see the completion path), NOT here: this arm stays the + // node-failure arm, and reaching it at all was the defect. + // + // ⛔ This is NOT "assume repairable when the failure is + // unknown" — that would invert the honest default and promise + // a repair for a lost run. The guard opens AFTER the journal, + // so only a run that demonstrably HAS a snapshot can reach the + // stamp: a throw from `journalConsumedSuspension` itself still + // propagates, and every exit above the consumption point is + // untouched and still carries no status at all. + // + // ⛔ And the journal is NOT moved down to sit beside the stamp + // instead. `recordLog` is what carries the snapshot into the + // durable row, so journalling after it would leave a + // `recordLog` failure with NO snapshot anywhere — converting + // this false negative into a TRUE one by destroying the repair + // rather than by reporting it. + let logged: ExecutionLogEntry | undefined; + try { + logged = this.recordLog({ + id: runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'failed', + startedAt: run.startedAt, + completedAt: new Date().toISOString(), + durationMs, + trigger: buildRunTrigger(context), + steps, + error: errorMessage, + }, context, consumed.run); + // Subflow chain: a child failing terminally fails every + // ancestor awaiting it — they can never be resumed otherwise. + // The delegation path handles its own level (skipBubble). + if (!skipBubble) { + // [#15222] `runId` names the run that stranded two + // statements above, so the walk journals every ancestor + // whose pause this cascade consumes and the chain is + // re-armable to its full depth, leaf-first. + await this.failAncestors(run.context, errorMessage, runId); + } + } catch (bookkeeping) { + // #4632 verdict: DURABILITY, so `error` — the caller is + // told a truthful, actionable thing (the run stranded, and + // it is repairable), which is exactly what makes the rest + // invisible from the outside: the terminal history row + // never landed and/or the ancestor cascade stopped + // part-way, nothing retries either, and no envelope + // carries a word about it. Consequence and fix in the + // first line, per AGENTS.md. Said ONCE per stranded run, + // not once per failed write. + // + // THIRD argument per `error(message, error?, meta?)`; the + // `Error` slot stays empty on purpose (#5575), and the + // thrown text goes to the structured slot rather than into + // the message (#6499). + this.logger.error( + `[Automation] run '${runId}' of flow '${run.flowName}' is STRANDED and its ` + + `post-strand bookkeeping threw, so its terminal history row never landed ` + + `and/or its subflow ancestors were not failed — nothing retries either, and ` + + `the run reads healthy to the Runs surfaces and the approvals sweeps. The ` + + `strand itself IS reported and repairable right now: restore it with ` + + `restoreConsumedSuspension('${runId}') before this process restarts, which ` + + `drops the in-memory journal this repair rides on. Fix the failure in this ` + + `record's meta.`, + undefined, + describeThrownForLog(bookkeeping), + ); + } + // Surface the flow's friendly error message (the raw error stays + // in `error` for logs/diagnostics). + return { + success: false, + error: errorMessage, + durationMs, + // [#13937] STRANDED, said by the producer that knows — the + // services half of the shape-4 ruling (maintainer + // 2026-09-01): the ordering stays, and the condition it + // leaves behind carries the platform-level name #14384 put + // on the contract. Stamped HERE and on no other exit, + // because this is the one exit that journalled a snapshot + // two statements up: the pause a durable decision was + // waiting on is gone, the run is recorded `failed`, and it + // can be re-armed only by {@link restoreConsumedSuspension} + // — never by `resume` (which now answers RUN_NOT_FOUND), + // never automatically. Terminal exactly like `'failed'` + // and distinct from it on purpose: `'failed'` says the run + // ran and was rejected; this says a recorded continuation + // stopped mid-flight and an operator has something to + // repair. The verdict and the journal are one fact stated + // twice — `stranded-run-status.test.ts` holds them equal. + // + // ⚠️ The run's RECORDED status — the log entry above and + // the durable history row — stays `'failed'`: that is + // `ExecutionStatus` (`@objectstack/spec`, + // `automation/execution.zod.ts`), a vocabulary the ruling + // did not widen; the durable discriminator for this + // condition is the snapshot the terminal row carries. + // + // ⛔ The cascade-failed ancestors above are NOT stranded: + // `failSuspendedRun` consumes their pauses and journals + // nothing, so no verb can re-arm them. A different (and + // worse) condition, which this stamp must not claim. + status: 'stranded', + errorMessage: flow.errorMessage, + flowLabel: flow.label, + // [#15555] Recomputed when the guard above had to abandon + // `recordLog`: the same pure function of the same steps + // that `recordLog`'s own first statement runs, so the two + // spellings cannot disagree. + summary: logged?.summary ?? summarizeRun(steps), + // [#24914, ruling B′] As `execute()`'s failure exit: the refusal + // the node after the consumed pause met, off the throw's brand. The + // screen-flow leg (a write behind a screen) reaches its caller HERE. + ...readCarriedRefusal(err), + }; + } + } finally { + this.resuming.delete(runId); + } +}