From ad9caf7866d7a4f403b38a890b6a1f21d4334651 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 24 Oct 2026 00:38:35 +0000 Subject: [PATCH] refactor(service-automation): move the run-context and resume bodies into run-resume.ts Move-only. The bodies of fifteen AutomationEngine methods move unedited into a new run-resume.ts: resolveRunContext, seedRunGrants, resolveDecisionSegmentContext and expandDeclaredLookups, which resolve a run's context; resume, refuseGatedResume, refuseUndeclaredSuspension, loadSuspendedRun, loadSuspendedRunStrict and creditChildRun, which load and gate a suspended run; and failSuspendedRun, settleUnresumableSuspension, cancelRun, cloneVariablesAtPause and failAncestors, which fail, settle or cancel one. Each method keeps its name, docblock, signature and position in the class and calls its body with a RunResumeHost literal holding exactly the members the body reads; creditChildRun reads none and is called directly. Each async stub drops async and returns the promise the moved function makes. Fields assigned only at their declaration or in the constructor go by value. store, userGrantsResolver, recordExpander and nodeTypeVocabularySealed go through getters, because a method reassigns each. refuseInvalidScreenInput and buildSubflowResumeSignal stay in the class: they read the module-private ENGINE_BUILT_SIGNAL. The imports only moved bodies read move with them. The ENGINE_SOURCES grep pins read the new module too. Claude-Session: https://claude.ai/code/session_01Mnj4nkWuhJ18xEdZvb2cjE Co-Authored-By: Claude --- .../services/service-automation/src/engine.ts | 678 ++++------------- .../src/flow-activation-ledger.test.ts | 2 +- .../service-automation/src/run-resume.ts | 689 ++++++++++++++++++ 3 files changed, 822 insertions(+), 547 deletions(-) create mode 100644 packages/services/service-automation/src/run-resume.ts diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index e870a3bec66..3e937a874ab 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -63,8 +63,7 @@ import { ConnectorSchema } from '@objectstack/spec/integration'; */ export const DEFAULT_BRANCH_LABEL = 'default'; -import { runIsUnscopedUserMode, flowTouchesData, runOrgUserIds, runTriggerUserId, type SeededRunContext } from './runtime-identity.js'; -import { refuseNode } from './guard-refusal.js'; +import { runTriggerUserId, type SeededRunContext } from './runtime-identity.js'; import { carriedRefusalOfResult, readCarriedRefusal, type CarriedRefusal } from './inner-refusal.js'; import { summarizeRun } from './run-summary.js'; // #5660 — the degrade registration reports a FOREIGN failure (a third-party @@ -141,6 +140,27 @@ import { recordLog, finishRefusedRun, } from './run-history.js'; +// [#25222] The run context and resume bodies: resolving a run's context, +// grants and declared lookups; loading, gating and resuming a suspended run; +// and failing, settling or cancelling one, along with its ancestors. Each +// `AutomationEngine` method of the same name calls its body. +import { + resolveRunContext, + seedRunGrants, + resolveDecisionSegmentContext, + expandDeclaredLookups, + resume, + refuseGatedResume, + refuseUndeclaredSuspension, + loadSuspendedRun, + loadSuspendedRunStrict, + creditChildRun, + failSuspendedRun, + settleUnresumableSuspension, + cancelRun, + cloneVariablesAtPause, + failAncestors, +} from './run-resume.js'; // ─── Node Executor Interface (Plugin Extension Point) ─────────────── @@ -5749,74 +5769,14 @@ export class AutomationEngine implements IAutomationService { * statically-decidable shapes earlier — but NOT the record-change shape, which * is only knowable at run time). */ - private async resolveRunContext(flow: FlowParsed, context?: AutomationContext, runId?: string): Promise { - // `flowRunId` is stamped alongside `runAs` because it shares that field's - // lifetime and its single construction point: set once here, copied into - // every data node's ObjectQL context by `resolveRunDataContext`, and - // persisted with a suspended run so it survives pause/resume — including a - // cold resume after a restart (#3456). Provenance, not authorization. - const runContext: AutomationContext = { - ...(context ?? {}), - runAs: flow.runAs ?? 'user', - // `flowName` shares the same lifetime and construction point: it feeds - // audit attribution (`svc:flow:` on ExecutionContext.actor) for - // runs that resolve no user (#4366). - flowName: flow.name, - ...(runId ? { flowRunId: runId } : {}), + private resolveRunContext(flow: FlowParsed, context?: AutomationContext, runId?: string): Promise { + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + seedRunGrants: (...a) => this.seedRunGrants(...a), + logger: this.logger, + expandDeclaredLookups: (...a) => this.expandDeclaredLookups(...a), }; - - // #3356 (follow-up to #1888) — a `runAs:'user'` run must enforce its data - // ops as the TRIGGERING user's real authorization. Most trigger surfaces - // (REST action / trigger endpoint) already resolve the full envelope and - // forward `permissions`; the ObjectQL record-change hook does NOT — its - // session carries only a `userId`, so the run used to fall back to a bare - // member/everyone principal (403 on private objects; silent field strips - // on public ones) even when the triggering user was fully authorized. - // When a grants resolver is wired and the trigger left the authz envelope - // unresolved (no `permissions`), resolve the user's real positions + - // permission sets here — the single point where every trigger type's run - // identity is established. Contexts that ALREADY carry `permissions` are - // left untouched: that includes an ADR-0090 agent principal acting - // on-behalf-of a user (its scope-derived ceiling is always non-empty), so - // this never re-broadens a deliberately narrowed identity. - // - // The run's peer list (`org_user_ids`) is seeded here too, through the - // same resolver, whenever the context does not carry one yet — so also - // for a context whose positions/permissions the door already resolved. - // No trigger surface forwards it (the REST, action and record-change - // doors all hand over positions/permissions/tenant at most), so without - // it every `runAs:'user'` run's `sys_user_org_members` policy compiled - // `unresolved-variable` and dropped out: the run saw only the acting - // user's own `sys_user` row while the same user's request saw their - // organization's members too. Taking ONLY the peer list from a - // re-resolution leaves the carried positions/permissions verbatim, so an - // agent ceiling stays narrowed; the list itself is the delegating - // human's peers in the run's organization, which is what the request - // path's agent envelope carries as well. A context that already carries - // one — a `subflow` / `map` child, a resumed leg — keeps it. - // - // The user's email is seeded by the same rule. No trigger surface - // forwards it either, and without it every row policy that reads - // `current_user.email` dropped out: the run was refused on rows the - // same user's request reads and writes. A context that already carries - // an email keeps it verbatim, as the request path keeps a session's. - await this.seedRunGrants(flow.name, runContext, 'triggering'); - - if (runIsUnscopedUserMode(runContext) && flowTouchesData(flow)) { - this.logger.warn( - `[runAs] flow '${flow.name}' executes with runAs:'user' but its trigger resolved no user ` + - `— its data operations will be REFUSED. Without a user they would carry no principal: ` + - `refused by the security plugin where one is composed, unscoped where none is (the fail-open ADR-0049 forbids). ` + - `Declare runAs:'system' to make the elevation explicit and intended, or arrange ` + - `for the trigger to supply a user. Note a user-less trigger is NOT only a schedule: a ` + - `record-change flow fired by a system write carries no user either (ADR-0049).`, - ); - } - - // #3475 — opt-in single-hop lookup expansion, AFTER identity resolution so - // the re-read runs as this run's own principal (see helper). - await this.expandDeclaredLookups(flow, runContext); - return runContext; + return resolveRunContext.call(host, flow, context, runId); } /** @@ -5841,72 +5801,18 @@ export class AutomationEngine implements IAutomationService { * A `runAs:'system'` run, a user-less run, a context carrying every fact, * and an engine with no resolver wired resolve nothing. */ - private async seedRunGrants( + private seedRunGrants( flowName: string, runContext: AutomationContext, whose: 'triggering' | 'deciding', ): Promise { - const needsGrants = !Array.isArray(runContext.permissions); - const needsPeers = runOrgUserIds(runContext) === undefined; - const needsEmail = !runContext.email; - if ( - runContext.runAs !== 'system' && - runContext.userId && - (needsGrants || needsPeers || needsEmail) && - this.userGrantsResolver - ) { - try { - const grants = await this.userGrantsResolver(runContext.userId, runContext.tenantId); - if (grants) { - if (needsGrants) { - runContext.positions = Array.isArray(grants.positions) ? grants.positions : []; - runContext.permissions = Array.isArray(grants.permissions) ? grants.permissions : []; - if (grants.tenantId && !runContext.tenantId) runContext.tenantId = grants.tenantId; - } - if (needsPeers && Array.isArray(grants.org_user_ids)) { - (runContext as SeededRunContext).org_user_ids = grants.org_user_ids; - } - if (needsEmail && grants.email) runContext.email = grants.email; - } - } catch (err) { - // Fail-safe, never fail-open: on a resolution error the run keeps - // the user's (unresolved) identity — the data middleware applies - // its baseline member fallback, NOT elevation — and we warn loudly - // so the degraded authorization is audible rather than silent. - // - // #6499 — the resolver's thrown text (transitively the - // datasource's) goes to the structured slot; see - // `forgetSuspendedRun`'s catch above for the full mechanism - // (#6299). #4632 verdict: FUNCTIONAL — stays `warn`: the - // degradation is fail-safe as argued above, its symptom is the - // run's own data-op refusals/strips, and nothing - // claimed-persisted failed to land. - // - // Unresolved peers and email degrade the same way, narrower and - // never wider: a policy that reads `current_user.org_user_ids` - // or `current_user.email` drops out, so the run sees its own - // `sys_user` row and not its organization's members, and no row - // an email-keyed policy would admit. The sentence names which - // half applies, and which facts it lost. - const lost = [ - ...(needsPeers ? ['current_user.org_user_ids'] : []), - ...(needsEmail ? ['current_user.email'] : []), - ].join(' or '); - // Without grants the lost list may be empty (the context carried - // its peers and email); with grants carried it never is, since - // one of the two is what sent the run here. - const consequence = needsGrants - ? `its data ops fall back to baseline member permissions (not elevated)` + - (lost ? `, and every row policy reading ${lost} drops out` : '') - : `its data ops keep the trigger's positions and permission sets, but every row ` + - `policy reading ${lost} drops out (narrower, not elevated)`; - this.logger.warn( - `[runAs] flow '${flowName}' could not resolve grants for ${whose} user ` + - `'${runContext.userId}' — ${consequence}. The resolver's failure is in this record's meta.`, - describeThrownForLog(err), - ); - } - } + // The body lives in ./run-resume.ts (#25222). + const readUserGrantsResolver = () => this.userGrantsResolver; + const host: ThisParameterType = { + get userGrantsResolver() { return readUserGrantsResolver(); }, + logger: this.logger, + }; + return seedRunGrants.call(host, flowName, runContext, whose); } /** @@ -5962,31 +5868,15 @@ export class AutomationEngine implements IAutomationService { * set — narrower, never the trigger user's grants (see the warning in * {@link seedRunGrants}). */ - private async resolveDecisionSegmentContext( + private resolveDecisionSegmentContext( run: SuspendedRun, signal: ResumeSignal, ): Promise { - const context = run.context; - const decider = signal[RESUME_AUTHORITY_SERVICE] === true ? signal.resumedBy : undefined; - if (typeof decider !== 'string' || decider === '') return context; - if (context?.userId === decider) return context; - const switched: SeededRunContext = { - ...(context ?? {}), - userId: decider, - triggerUserId: runTriggerUserId(context) ?? null, + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + seedRunGrants: (...a) => this.seedRunGrants(...a), }; - delete switched.positions; - delete switched.permissions; - delete switched.org_user_ids; - // The trigger user's email is theirs too: kept under the decider's id, - // it would admit the decider to every row an email-keyed policy grants - // the submitter. Removed on both postures, and seeded for the decider - // below like their grants. - delete switched.email; - // A system run keeps its elevation: attribution moves, no grant is resolved. - if (switched.runAs === 'system') return switched; - await this.seedRunGrants(run.flowName, switched, 'deciding'); - return switched; + return resolveDecisionSegmentContext.call(host, run, signal); } /** @@ -6004,53 +5894,14 @@ export class AutomationEngine implements IAutomationService { * (the template then renders the scalar id) — expansion must never break the * flow it feeds. */ - private async expandDeclaredLookups(flow: FlowParsed, runContext: AutomationContext): Promise { - if (!this.recordExpander) return; - const record = runContext.record as Record | undefined; - const id = record?.id; - const object = runContext.object; - if (!record || id == null || id === '' || !object) return; - - const startNode = flow.nodes.find((n) => n.type === 'start'); - const raw = (startNode?.config as { expand?: unknown } | undefined)?.expand; - const expandFields = - typeof raw === 'string' - ? raw - ? [raw] - : [] - : Array.isArray(raw) - ? raw.filter((f): f is string => typeof f === 'string' && f.length > 0) - : []; - if (expandFields.length === 0) return; - - try { - const full = await this.recordExpander(object, id, expandFields, runContext); - if (full && typeof full === 'object') { - for (const field of expandFields) { - const expanded = (full as Record)[field]; - // Graft only a genuinely expanded relation (object/array); a - // scalar means the field is not a resolvable lookup — leave the - // raw id in place rather than overwrite it. - if (expanded !== null && typeof expanded === 'object') { - record[field] = expanded; - } - } - } - } catch (err) { - // #6499 — the expander's thrown text (transitively the - // datasource's) goes to the structured slot; see - // `forgetSuspendedRun`'s catch above for the full mechanism - // (#6299). #4632 verdict: FUNCTIONAL — stays `warn`, per the - // docblock: expansion is best-effort enrichment that must never - // break the flow it feeds, and the visible symptom is templates - // rendering the scalar id. - this.logger.warn( - `[expand] flow '${flow.name}' could not expand lookups [${expandFields.join(', ')}] on ` + - `'${object}#${String(id)}' — templates referencing these relations resolve to the scalar ` + - `id. The expander's failure is in this record's meta.`, - describeThrownForLog(err), - ); - } + private expandDeclaredLookups(flow: FlowParsed, runContext: AutomationContext): Promise { + // The body lives in ./run-resume.ts (#25222). + const readRecordExpander = () => this.recordExpander; + const host: ThisParameterType = { + get recordExpander() { return readRecordExpander(); }, + logger: this.logger, + }; + return expandDeclaredLookups.call(host, flow, runContext); } async execute(flowName: string, context?: AutomationContext): Promise { @@ -6698,24 +6549,13 @@ export class AutomationEngine implements IAutomationService { * through {@link resumeInternal} and are not re-gated — they continue work * some already-authorized call started. */ - async resume(runId: string, signal?: ResumeSignal): Promise { - const refusal = await this.refuseGatedResume(runId, signal); - if (refusal) return refusal; - // An ABSENT signal is an EMPTY caller submission, never an exemption - // (commit 7307191db). This is the in-process door, and `resume(runId)` used to - // skip the screen contract that `resume(runId, {})` is held to: - // `refuseInvalidScreenInput` short-circuited on a falsy signal — a - // second, unnamed spelling of the exemption the engine already states - // through `ENGINE_BUILT_SIGNAL`, so a run parked on a screen with an - // unconditional `required` field proceeded with that variable unbound. - // The HTTP door has always assembled `{}` for an empty body; this makes - // the two doors agree, and the only exemption left is the engine's own - // continuation, which proves itself by BUILDING an engine-built signal. - // A pause with no screen contract — `wait`, `approval`, a message-only - // or object-form screen — is untouched: an empty submission against no - // declared fields is conformant, so the wait node's timer wake - // (`engine.resume(runId)`) continues exactly as before. - return this.resumeInternal(runId, signal ?? {}, false); + resume(runId: string, signal?: ResumeSignal): Promise { + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + refuseGatedResume: (...a) => this.refuseGatedResume(...a), + resumeInternal: (...a) => this.resumeInternal(...a), + }; + return resume.call(host, runId, signal); } /** @@ -6748,53 +6588,16 @@ export class AutomationEngine implements IAutomationService { * never declared — the failure mode this whole issue is about, restated as a * log line. */ - private async refuseGatedResume(runId: string, signal?: ResumeSignal): Promise { - const run = await this.resolveEffectiveSuspension(runId); - if (!run) return null; - - const nodeType = this.resolveSuspendedNodeType(run); - if (!nodeType) return null; - if (this.resolveResumeAuthority(nodeType) !== 'service') return null; - // The owning service stamped the signal — this resume IS the recorded - // decision's tail, not a way around it. - if (signal?.[RESUME_AUTHORITY_SERVICE]) return null; - - // Refusing. Which of the two reasons? A second registry walk, on the - // refusal path only, so the message can tell a node that deliberately - // declared `'service'` from one that declared nothing at all. - const declared = this.resolveDeclaredResumeAuthority(nodeType); - const direct = run.runId === runId; - const at = direct ? `'${run.nodeId}'` : `'${run.nodeId}' (linked run '${run.runId}')`; - const why = declared === 'service' - ? `which is resumable only through its owning service (resumeAuthority: 'service')` - : `whose type never declares resumeAuthority, so it is closed to the generic route until it does ` + - `(an undeclared resumeAuthority resolves to 'service', fail-closed) — declare ` + - `resumeAuthority: 'any' on its descriptor if this route IS the intended door`; - this.logger.warn(`[automation] refused resume of run '${runId}': parked on ${nodeType} node ${at}, ${why}`); - - // The fix, identical in both the direct and the linked-run phrasing — - // what has to change is a descriptor, not the call that just failed. - const undeclaredFix = - `and that node type never declares resumeAuthority, so the generic resume route is closed to the ` + - `pauses it creates: an undeclared resumeAuthority resolves to 'service', fail-closed. If that ` + - `route IS the intended door — a screen's collected inputs, a ` + - `signal wait's external producer — declare resumeAuthority: 'any' on its action descriptor; declare ` + - `'service' if resuming is the tail of a decision some service must authorize and record first`; - return { - success: false, - code: 'PERMISSION_DENIED', - error: declared === 'service' - ? direct - ? `Run '${runId}' is paused at a '${nodeType}' node, which only its owning service may resume — ` + - `drive it through that service's API (e.g. an approval decision), not a raw resume` - : `Run '${runId}' is waiting on run '${run.runId}', which is paused at a '${nodeType}' node that ` + - `only its owning service may resume — resuming here would continue past a decision that has not ` + - `been made; drive it through that service's API instead` - : direct - ? `Run '${runId}' is paused at a '${nodeType}' node, ${undeclaredFix}` - : `Run '${runId}' is waiting on run '${run.runId}', which is paused at a '${nodeType}' node, ` + - `${undeclaredFix}`, + private refuseGatedResume(runId: string, signal?: ResumeSignal): Promise { + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + resolveEffectiveSuspension: (...a) => this.resolveEffectiveSuspension(...a), + resolveSuspendedNodeType: (...a) => this.resolveSuspendedNodeType(...a), + resolveResumeAuthority: (...a) => this.resolveResumeAuthority(...a), + resolveDeclaredResumeAuthority: (...a) => this.resolveDeclaredResumeAuthority(...a), + logger: this.logger, }; + return refuseGatedResume.call(host, runId, signal); } /** @@ -6905,22 +6708,11 @@ export class AutomationEngine implements IAutomationService { * `resume-authority-gate.test.ts`'s `bare_pause` case pins on purpose. */ private refuseUndeclaredSuspension(nodeType: string): NodeExecutionResult | null { - const descriptor = this.resolveCanonicalDescriptor(nodeType); - // No descriptor ⇒ no declaration ⇒ nothing to enforce (see above). - if (!descriptor) return null; - if (descriptor.supportsPause === true) return null; - return refuseNode( - `node type '${nodeType}' suspended the run but its action descriptor declares ` + - `supportsPause: false, so the pause is refused — a run that paused here could not be ` + - `continued on the generic resume route anyway: a type that declares no pause declares no ` + - `resumeAuthority either, and an undeclared resumeAuthority resolves to 'service', which the ` + - `generic route refuses. Declare ` + - `supportsPause: true on the descriptor together with the resumeAuthority the pauses need ` + - `('any' if POST /automation/:name/runs/:runId/resume is the intended door, 'service' if ` + - `resuming is the tail of a decision some service must authorize and record first) — or stop ` + - `returning suspend: true from execute(). This is a metadata defect, not a runtime one, so a ` + - `fault edge does not route it.`, - ); + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + resolveCanonicalDescriptor: (...a) => this.resolveCanonicalDescriptor(...a), + }; + return refuseUndeclaredSuspension.call(host, nodeType); } /** @@ -7019,50 +6811,13 @@ export class AutomationEngine implements IAutomationService { * only need a best-effort answer. NOT correct for {@link resumeInternal}, * which must tell a dead run from an unreachable store — it uses * {@link loadSuspendedRunStrict}. */ - private async loadSuspendedRun(runId: string): Promise { - try { - return await this.loadSuspendedRunStrict(runId); - } catch (err) { - // #6230 — the cause goes to `meta`, never into the message. It 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 / - // #5912, and cloud#971's shape. - // - // This seam is the `warn` half of that family and carries the extra - // harm the module docblock of ./thrown-cause-diagnostics.ts calls - // out: `ObjectLogger` routes `warn` to **stdout**, and `serve`'s - // boot-quiet window wraps `process.stdout.write`, where - // `BootLogCapture.offer()` retains a physical line only when - // `classifyBootLogLine` finds a level head on it. Continuation lines - // are therefore DROPPED outright, not merely misread — and this is - // live during boot: `plugin.ts` `start()` → `rearmSuspendedWaitTimers` - // → `engine.resume()` for an overdue run → the gate - // (`refuseGatedResume` → `resolveEffectiveSuspension`) → here. - // - // SECOND argument, per the `Logger` contract - // (`packages/spec/src/contracts/logger.ts`): `warn(message, meta?)`. - // Unlike `error(message, error?, meta?)` (#5912's seam, PR #6228) - // `warn` has no `Error` slot, so `meta` is the second parameter. - // - // Level stays `warn` on purpose. This is a deliberate FUNCTIONAL - // degradation — the loader's whole contract is "a best-effort answer - // for incidental readers" and `resumeInternal` uses the strict form - // when the difference matters — so #4632's durability rule does not - // apply and raising it to `error` would be that rule's mirror-image - // misuse, an alarm on every gate lookup during an outage. - this.logger.warn( - `[automation] durable suspended-run store unreadable for run '${runId}' — this read is best-effort ` + - `and DEGRADES to null, so its caller (the resume gate, a screen fetch) sees exactly what it ` + - `would see if no suspension existed under that id; the run itself is untouched and stays ` + - `parked. Fix the store failure in this record's meta — a strict read reports it as ` + - `STORE_UNAVAILABLE instead of degrading.`, - describeThrownForLog(err), - ); - return null; - } + private loadSuspendedRun(runId: string): Promise { + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + loadSuspendedRunStrict: (...a) => this.loadSuspendedRunStrict(...a), + logger: this.logger, + }; + return loadSuspendedRun.call(host, runId); } /** {@link loadSuspendedRun} without the degradation: a store read failure @@ -7100,26 +6855,16 @@ export class AutomationEngine implements IAutomationService { * the #13937 shape-4 ruling (maintainer 2026-09-01) keeps that order, for * exactly-once across a crash. This changes only WHICH suspension is * read, never when it is consumed. */ - private async loadSuspendedRunStrict(runId: string): Promise { - if (!this.store) return this.suspendedRuns.get(runId) ?? null; - const stored = await this.store.load(runId); - if (stored) return stored; - // The store has no row. For every run it ever accepted that IS the - // answer — including the runs this process advanced past, whose stale - // map entries are the whole defect above. The lone exception is a run - // whose durable save failed here: the store was never handed that row, - // so its silence says nothing about it, and `persistSuspendedRun` - // deliberately keeps such a run resumable in-process (it reports the - // lost durability at `error`). - if (this.cacheOnlySuspensions.has(runId)) return this.suspendedRuns.get(runId) ?? null; - // [#15832] The store ANSWERED, and the answer is "no row". Any entry - // this process still holds for that run is a run it parked and another - // replica consumed — the phantom the two listings hand back. This is - // the definitive per-id evidence the paragraph above already rests on, - // so the same reading that refuses to serve it here stops publishing it - // there. A store read that threw never reaches this line. - this.evictConsumedSuspension(runId); - return null; + private loadSuspendedRunStrict(runId: string): Promise { + // The body lives in ./run-resume.ts (#25222). + const readStore = () => this.store; + const host: ThisParameterType = { + get store() { return readStore(); }, + suspendedRuns: this.suspendedRuns, + cacheOnlySuspensions: this.cacheOnlySuspensions, + evictConsumedSuspension: (...a) => this.evictConsumedSuspension(...a), + }; + return loadSuspendedRunStrict.call(host, runId); } /** @@ -7163,34 +6908,8 @@ export class AutomationEngine implements IAutomationService { * failures of a child that COMPLETED, never a failed child's own `failed`. */ private creditChildRun(steps: StepLogEntry[], nodeId: string, child: FlowRunSummary | undefined): void { - if (!child) return; - for (let i = steps.length - 1; i >= 0; i--) { - if (steps[i].nodeId !== nodeId) continue; - const prior = steps[i].metrics ?? {}; - steps[i] = { - ...steps[i], - metrics: { - selected: (prior.selected ?? 0) + child.selected, - acted: (prior.acted ?? 0) + child.acted, - // N uncountable effects in the child collapse to ONE flag on - // the parent's step: this execution dispatched something the - // platform cannot count. The child keeps the real count in - // its own run row, and the question this feeds — "is the - // parent's `acted` complete?" — is boolean either way. - ...(prior.unmeasuredEffect || child.unmeasured ? { unmeasuredEffect: true } : {}), - // #15617 — the pausing item's own contained failures, which - // the synchronous path reports through the executor's - // `metrics`. Written only when one of the two sides actually - // tracked a count: an absent `failures` means "not tracked", - // and a `0` written here would claim a measurement of a - // child recorded before the count existed. - ...(prior.failures !== undefined || child.failed !== undefined - ? { failures: (prior.failures ?? 0) + (child.failed ?? 0) } - : {}), - }, - }; - return; - } + // The body lives in ./run-resume.ts (#25222). + creditChildRun(steps, nodeId, child); } /** @@ -8481,27 +8200,18 @@ export class AutomationEngine implements IAutomationService { * aliasing hazard is a FAILED ATTEMPT writing through a shared scope * object, and no attempt has run against this run. */ - private async failSuspendedRun( + private failSuspendedRun( run: SuspendedRun, error: string, repairableDescendant?: string, ): Promise { - const consumed = repairableDescendant !== undefined - ? this.journalConsumedSuspension(run, run.steps.length, error, run.variables) - : undefined; - await this.forgetSuspendedRun(run, 'failed'); - this.recordLog({ - id: run.runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - status: 'failed', - startedAt: run.startedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - run.startTime, - trigger: buildRunTrigger(run.context), - steps: run.steps, - error, - }, run.context, consumed?.run); + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + journalConsumedSuspension: (...a) => this.journalConsumedSuspension(...a), + forgetSuspendedRun: (...a) => this.forgetSuspendedRun(...a), + recordLog: (...a) => this.recordLog(...a), + }; + return failSuspendedRun.call(host, run, error, repairableDescendant); } /** @@ -8550,60 +8260,18 @@ export class AutomationEngine implements IAutomationService { * (#16274): the run is already consumed when it runs, so a throw out of * it must not turn the refusal into a rejected promise. */ - private async settleUnresumableSuspension(run: SuspendedRun, cause: string): Promise { - const refusal: AutomationResult = { success: false, code: 'RUN_NOT_FOUND', error: cause }; - if (!this.nodeTypeVocabularySealed) return refusal; - const claim = await this.claimAdvance(run); - if (claim.kind === 'lost') { - // [#15832] The store answered and holds no row where this replica - // read one, so whatever snapshot this process still holds is stale. - this.evictConsumedSuspension(run.runId); - return refusal; - } - if (claim.kind === 'unavailable') { - return { - success: false, - code: 'STORE_UNAVAILABLE', - error: - `Run '${run.runId}' can never continue (${cause}), but the durable suspended-run store was ` + - `unreachable while settling it, and whether the store consumed the suspension is UNKNOWN (a ` + - `failure can arrive after a committed delete), so its record may still read paused. Retry once ` + - `the store is available: a retry that finds the run still parked settles it, and one that finds ` + - `the suspension gone answers RUN_NOT_FOUND: ${claim.message}`, - }; - } - await this.forgetSuspendedRun(run, 'failed', claim.kind === 'claimed'); - try { - this.recordLog({ - id: run.runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - status: 'failed', - startedAt: run.startedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - run.startTime, - trigger: buildRunTrigger(run.context), - steps: run.steps, - error: cause, - }, run.context); - } catch (bookkeeping) { - // #4632 verdict: DURABILITY, so `error`. The suspension is consumed - // and the caller is answered the truthful refusal, so nothing reads - // as wrong from the outside. But the `failed` row that says why - // never landed, and nothing retries it. Same sink and same level as - // the #16274 completion guard. THIRD argument per - // `error(message, error?, meta?)`; the `Error` slot stays empty - // (#5575) and the thrown text rides the structured slot (#6499). - this.logger.error( - `[Automation] run '${run.runId}' of flow '${run.flowName}' can never continue and its suspension ` + - `was consumed, but recording its terminal 'failed' history threw, so no history row says why: ` + - `the caller was answered RUN_NOT_FOUND, the run no longer reads as paused, and nothing retries ` + - `the write. Fix the history failure in this record's meta.`, - undefined, - describeThrownForLog(bookkeeping), - ); - } - return refusal; + private settleUnresumableSuspension(run: SuspendedRun, cause: string): Promise { + // The body lives in ./run-resume.ts (#25222). + const readNodeTypeVocabularySealed = () => this.nodeTypeVocabularySealed; + const host: ThisParameterType = { + get nodeTypeVocabularySealed() { return readNodeTypeVocabularySealed(); }, + claimAdvance: (...a) => this.claimAdvance(...a), + evictConsumedSuspension: (...a) => this.evictConsumedSuspension(...a), + forgetSuspendedRun: (...a) => this.forgetSuspendedRun(...a), + recordLog: (...a) => this.recordLog(...a), + logger: this.logger, + }; + return settleUnresumableSuspension.call(host, run, cause); } /** @@ -8641,67 +8309,15 @@ export class AutomationEngine implements IAutomationService { * the SNAPSHOT, so a stale replica tore down the pause of a node the run had * already left and left the live one's armed. */ - async cancelRun(runId: string, reason?: string): Promise { - let run: SuspendedRun | null = null; - try { - run = await this.loadSuspendedRunStrict(runId); - } catch (err) { - // #6299 — same family, same mechanism as `forgetSuspendedRun` - // above: the driver's uncontrolled text goes to the structured - // slot so the record stays one physical line. - // - // #4632 verdict: DURABILITY — raised from `warn` to `error`. The - // failed read is silently turned into "no such suspended run" - // and this method returns `false`, which its own contract - // documents as idempotent success (already terminal / unknown), - // so the cancellation is SKIPPED while the call reads clean. The - // only in-repo caller measures the cost: plugin-approvals' - // revise-window recall - // (`packages/plugins/plugin-approvals/src/approval-service.ts`) - // never reads the boolean at all — it only catches a THROW, and - // grades that throw `error` with "the run may be stranded" - // (#4420). A store-read failure produces precisely that stranded - // run WITHOUT firing that alarm: the request is marked - // `recalled`, the record lock is released, `resumeError` stays - // undefined — and the run stays parked in the store, to be - // re-armed and resumed by the next restart, inside a flow whose - // approval has already been withdrawn. - // - // This is why #6230's verdict must not be copied here. - // `loadSuspendedRun` is a DECLARED best-effort reader for - // incidental callers (a gate lookup, a screen fetch), and - // `resumeInternal` takes the strict form exactly where the - // difference matters. `cancelRun` has no strict alternative, and - // its degradation decides a WRITE. - // - // THIRD argument (`error(message, error?, meta?)`), `Error` slot - // deliberately empty (#5575). - this.logger.error( - `[automation] cancelRun('${runId}') could not read the durable suspended-run store, so the ` + - `cancellation was SKIPPED and reported as idempotent success — this call returns false, which ` + - `its callers read as "no such suspended run". The run is NOT cancelled: if it is parked in the ` + - `store it stays parked, and the next restart re-arms and resumes it while the caller has ` + - `already recorded the cancellation. Fix the store failure in this record's meta, then re-issue ` + - `cancelRun('${runId}').`, - undefined, - describeThrownForLog(err), - ); - } - if (!run) return false; - await this.forgetSuspendedRun(run, 'cancelled'); - this.recordLog({ - id: run.runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - status: 'cancelled', - startedAt: run.startedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - run.startTime, - trigger: buildRunTrigger(run.context), - steps: run.steps, - error: reason, - }, run.context); - return true; + cancelRun(runId: string, reason?: string): Promise { + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + loadSuspendedRunStrict: (...a) => this.loadSuspendedRunStrict(...a), + logger: this.logger, + forgetSuspendedRun: (...a) => this.forgetSuspendedRun(...a), + recordLog: (...a) => this.recordLog(...a), + }; + return cancelRun.call(host, runId, reason); } /** @@ -8724,19 +8340,11 @@ export class AutomationEngine implements IAutomationService { * what it puts back may name post-resume state. */ private cloneVariablesAtPause(run: SuspendedRun, runId: string): SuspendedRun['variables'] { - try { - return JSON.parse(JSON.stringify(run.variables)) as SuspendedRun['variables']; - } catch (err) { - // #6299 family — the thrown text is not ours to shape, so it rides - // the structured slot and never the message. - this.logger.warn( - `[automation] run '${runId}': could not copy the paused variables before resuming, so if this ` + - `resume strands the run, the suspension an operator restores may carry state this attempt ` + - `wrote rather than the state at the pause. The run resumes normally either way.`, - describeThrownForLog(err), - ); - return run.variables; - } + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + logger: this.logger, + }; + return cloneVariablesAtPause.call(host, run, runId); } /** @@ -9499,40 +9107,18 @@ export class AutomationEngine implements IAutomationService { * @param strandedRunId - The run that stranded and started this cascade. * Omitted by a caller that has no repairable run below the walk. */ - private async failAncestors( + private failAncestors( context: AutomationContext | undefined, error: string, strandedRunId?: string, ): Promise { - let parentId = (context as Record | undefined)?.$parentRunId; - let below = strandedRunId; - let hops = 0; - while (typeof parentId === 'string' && parentId && hops++ < 32) { - // [#14332] The DEGRADING loader, by deliberate choice: this walk runs - // inside the catch arm that is already handling a run's failure, so it - // must not throw, and its recorded posture is exactly - // `loadSuspendedRun`'s — a store failure reads as "no ancestor here" - // and stops the walk. What changes is only WHICH suspension is read: - // the shared store's, not this replica's memory of where the parent - // was last parked. The old `??` chain had #13617's own harm shape — - // a stale parent failed at a node it had already left, so - // `forgetSuspendedRun` released the wrong node's pause. The one thing - // gained beyond that: the bare `.catch(() => null)` swallowed a store - // failure in total silence, and the loader records it (at `warn`, - // its declared best-effort level — no new `error` seam here). - const parent = await this.loadSuspendedRun(parentId); - if (!parent) return; - await this.failSuspendedRun( - parent, - `subflow descendant failed: ${error}`, - // [#15222] Earned, not assumed: only a descendant this process - // actually holds a snapshot for makes the ancestor's own - // snapshot a promise the restore verb can keep. - below !== undefined && this.consumedSuspensions.has(below) ? below : undefined, - ); - below = parent.runId; - parentId = (parent.context as Record | undefined)?.$parentRunId; - } + // The body lives in ./run-resume.ts (#25222). + const host: ThisParameterType = { + loadSuspendedRun: (...a) => this.loadSuspendedRun(...a), + failSuspendedRun: (...a) => this.failSuspendedRun(...a), + consumedSuspensions: this.consumedSuspensions, + }; + return failAncestors.call(host, context, error, strandedRunId); } /** diff --git a/packages/services/service-automation/src/flow-activation-ledger.test.ts b/packages/services/service-automation/src/flow-activation-ledger.test.ts index 0411afd7f48..f51619ac115 100644 --- a/packages/services/service-automation/src/flow-activation-ledger.test.ts +++ b/packages/services/service-automation/src/flow-activation-ledger.test.ts @@ -1029,7 +1029,7 @@ describe('#10243 — the process-local `flowEnabled` map is retired', () => { // [#25222] The engine's source is `engine.ts` plus every module a method's // body moved to: a grep-level pin reads them all, or a moved body is a place // the retired mechanism could come back unseen. - const ENGINE_SOURCES = ['./engine.ts', './flow-registration-checks.ts', './plugin-extension-api.ts', './dag-traversal.ts', './run-history.ts']; + const ENGINE_SOURCES = ['./engine.ts', './flow-registration-checks.ts', './plugin-extension-api.ts', './dag-traversal.ts', './run-history.ts', './run-resume.ts']; const engineSource = ENGINE_SOURCES.map((path) => readFileSync(fileURLToPath(new URL(path, import.meta.url)), 'utf8'), ).join('\n'); diff --git a/packages/services/service-automation/src/run-resume.ts b/packages/services/service-automation/src/run-resume.ts new file mode 100644 index 00000000000..0b8574ec0f5 --- /dev/null +++ b/packages/services/service-automation/src/run-resume.ts @@ -0,0 +1,689 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The run context and resume of {@link AutomationEngine}: how a run's context, + * grants and declared lookups are resolved; how a suspended run is loaded, + * gated and resumed; and how a suspended run is failed, settled or cancelled, + * along with its ancestors. + * + * [#25222] Slice 6 of the `engine.ts` split, move-only. Each function here is + * the body of the `AutomationEngine` method of the same name, moved unedited. + * The class keeps that method (its name, docblock and position are in the + * published `.d.ts`) and calls the body with a {@link RunResumeHost} literal + * holding exactly the members the body reads. An async method's stub is no + * longer `async`: it returns the promise the moved function makes. + */ + +import type { FlowParsed, FlowRunSummary } from '@objectstack/spec/automation'; +import type { AutomationContext, AutomationResult, ResumeSignal } from '@objectstack/spec/contracts'; +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 +// ./engine.js is safe under ESM live bindings; the tsup bundle is one file and +// has no cycle. +import { + buildRunTrigger, + type AutomationEngine, + type NodeExecutionResult, + type StepLogEntry, + type SuspendedRun, +} from './engine.js'; + +/** + * The `AutomationEngine` members the moved bodies read, each typed as the class + * member itself, so a renamed or respelled member is a compile error. Each + * function takes the `Pick` of the members its own body reads, and the class + * passes exactly those. Methods go as late-bound closures, so a spy or test + * double set on the instance is still the one called. A field the class assigns + * only at its declaration or in its constructor goes by value, so the value + * passed is the field, including where an async body reads it after an + * `await`. Four fields go through a getter, because a method reassigns each: + * `store` (`setSuspendedRunStore`), `userGrantsResolver` + * (`setUserGrantsResolver`), `recordExpander` (`setRecordExpander`) and + * `nodeTypeVocabularySealed` (`sealNodeTypeVocabulary`). Each read in the body + * reads the engine's current value, as it did through `this`. + */ +export interface RunResumeHost { + readonly seedRunGrants: AutomationEngine['seedRunGrants']; + readonly logger: AutomationEngine['logger']; + readonly expandDeclaredLookups: AutomationEngine['expandDeclaredLookups']; + readonly userGrantsResolver: AutomationEngine['userGrantsResolver']; + readonly recordExpander: AutomationEngine['recordExpander']; + readonly refuseGatedResume: AutomationEngine['refuseGatedResume']; + readonly resumeInternal: AutomationEngine['resumeInternal']; + readonly resolveEffectiveSuspension: AutomationEngine['resolveEffectiveSuspension']; + readonly resolveSuspendedNodeType: AutomationEngine['resolveSuspendedNodeType']; + readonly resolveResumeAuthority: AutomationEngine['resolveResumeAuthority']; + readonly resolveDeclaredResumeAuthority: AutomationEngine['resolveDeclaredResumeAuthority']; + readonly resolveCanonicalDescriptor: AutomationEngine['resolveCanonicalDescriptor']; + readonly loadSuspendedRunStrict: AutomationEngine['loadSuspendedRunStrict']; + readonly store: AutomationEngine['store']; + readonly suspendedRuns: AutomationEngine['suspendedRuns']; + readonly cacheOnlySuspensions: AutomationEngine['cacheOnlySuspensions']; + readonly evictConsumedSuspension: AutomationEngine['evictConsumedSuspension']; + readonly journalConsumedSuspension: AutomationEngine['journalConsumedSuspension']; + readonly forgetSuspendedRun: AutomationEngine['forgetSuspendedRun']; + readonly recordLog: AutomationEngine['recordLog']; + readonly nodeTypeVocabularySealed: AutomationEngine['nodeTypeVocabularySealed']; + readonly claimAdvance: AutomationEngine['claimAdvance']; + readonly loadSuspendedRun: AutomationEngine['loadSuspendedRun']; + readonly failSuspendedRun: AutomationEngine['failSuspendedRun']; + readonly consumedSuspensions: AutomationEngine['consumedSuspensions']; +} + +/** The body of {@link AutomationEngine.resolveRunContext}, moved unedited; that method's docblock is its contract. */ +export async function resolveRunContext(this: Pick, flow: FlowParsed, context?: AutomationContext, runId?: string): Promise { + // `flowRunId` is stamped alongside `runAs` because it shares that field's + // lifetime and its single construction point: set once here, copied into + // every data node's ObjectQL context by `resolveRunDataContext`, and + // persisted with a suspended run so it survives pause/resume — including a + // cold resume after a restart (#3456). Provenance, not authorization. + const runContext: AutomationContext = { + ...(context ?? {}), + runAs: flow.runAs ?? 'user', + // `flowName` shares the same lifetime and construction point: it feeds + // audit attribution (`svc:flow:` on ExecutionContext.actor) for + // runs that resolve no user (#4366). + flowName: flow.name, + ...(runId ? { flowRunId: runId } : {}), + }; + + // #3356 (follow-up to #1888) — a `runAs:'user'` run must enforce its data + // ops as the TRIGGERING user's real authorization. Most trigger surfaces + // (REST action / trigger endpoint) already resolve the full envelope and + // forward `permissions`; the ObjectQL record-change hook does NOT — its + // session carries only a `userId`, so the run used to fall back to a bare + // member/everyone principal (403 on private objects; silent field strips + // on public ones) even when the triggering user was fully authorized. + // When a grants resolver is wired and the trigger left the authz envelope + // unresolved (no `permissions`), resolve the user's real positions + + // permission sets here — the single point where every trigger type's run + // identity is established. Contexts that ALREADY carry `permissions` are + // left untouched: that includes an ADR-0090 agent principal acting + // on-behalf-of a user (its scope-derived ceiling is always non-empty), so + // this never re-broadens a deliberately narrowed identity. + // + // The run's peer list (`org_user_ids`) is seeded here too, through the + // same resolver, whenever the context does not carry one yet — so also + // for a context whose positions/permissions the door already resolved. + // No trigger surface forwards it (the REST, action and record-change + // doors all hand over positions/permissions/tenant at most), so without + // it every `runAs:'user'` run's `sys_user_org_members` policy compiled + // `unresolved-variable` and dropped out: the run saw only the acting + // user's own `sys_user` row while the same user's request saw their + // organization's members too. Taking ONLY the peer list from a + // re-resolution leaves the carried positions/permissions verbatim, so an + // agent ceiling stays narrowed; the list itself is the delegating + // human's peers in the run's organization, which is what the request + // path's agent envelope carries as well. A context that already carries + // one — a `subflow` / `map` child, a resumed leg — keeps it. + // + // The user's email is seeded by the same rule. No trigger surface + // forwards it either, and without it every row policy that reads + // `current_user.email` dropped out: the run was refused on rows the + // same user's request reads and writes. A context that already carries + // an email keeps it verbatim, as the request path keeps a session's. + await this.seedRunGrants(flow.name, runContext, 'triggering'); + + if (runIsUnscopedUserMode(runContext) && flowTouchesData(flow)) { + this.logger.warn( + `[runAs] flow '${flow.name}' executes with runAs:'user' but its trigger resolved no user ` + + `— its data operations will be REFUSED. Without a user they would carry no principal: ` + + `refused by the security plugin where one is composed, unscoped where none is (the fail-open ADR-0049 forbids). ` + + `Declare runAs:'system' to make the elevation explicit and intended, or arrange ` + + `for the trigger to supply a user. Note a user-less trigger is NOT only a schedule: a ` + + `record-change flow fired by a system write carries no user either (ADR-0049).`, + ); + } + + // #3475 — opt-in single-hop lookup expansion, AFTER identity resolution so + // the re-read runs as this run's own principal (see helper). + await this.expandDeclaredLookups(flow, runContext); + return runContext; +} + +/** The body of {@link AutomationEngine.seedRunGrants}, moved unedited; that method's docblock is its contract. */ +export async function seedRunGrants( + this: Pick, + flowName: string, + runContext: AutomationContext, + whose: 'triggering' | 'deciding', +): Promise { + const needsGrants = !Array.isArray(runContext.permissions); + const needsPeers = runOrgUserIds(runContext) === undefined; + const needsEmail = !runContext.email; + if ( + runContext.runAs !== 'system' && + runContext.userId && + (needsGrants || needsPeers || needsEmail) && + this.userGrantsResolver + ) { + try { + const grants = await this.userGrantsResolver(runContext.userId, runContext.tenantId); + if (grants) { + if (needsGrants) { + runContext.positions = Array.isArray(grants.positions) ? grants.positions : []; + runContext.permissions = Array.isArray(grants.permissions) ? grants.permissions : []; + if (grants.tenantId && !runContext.tenantId) runContext.tenantId = grants.tenantId; + } + if (needsPeers && Array.isArray(grants.org_user_ids)) { + (runContext as SeededRunContext).org_user_ids = grants.org_user_ids; + } + if (needsEmail && grants.email) runContext.email = grants.email; + } + } catch (err) { + // Fail-safe, never fail-open: on a resolution error the run keeps + // the user's (unresolved) identity — the data middleware applies + // its baseline member fallback, NOT elevation — and we warn loudly + // so the degraded authorization is audible rather than silent. + // + // #6499 — the resolver's thrown text (transitively the + // datasource's) goes to the structured slot; see + // `forgetSuspendedRun`'s catch above for the full mechanism + // (#6299). #4632 verdict: FUNCTIONAL — stays `warn`: the + // degradation is fail-safe as argued above, its symptom is the + // run's own data-op refusals/strips, and nothing + // claimed-persisted failed to land. + // + // Unresolved peers and email degrade the same way, narrower and + // never wider: a policy that reads `current_user.org_user_ids` + // or `current_user.email` drops out, so the run sees its own + // `sys_user` row and not its organization's members, and no row + // an email-keyed policy would admit. The sentence names which + // half applies, and which facts it lost. + const lost = [ + ...(needsPeers ? ['current_user.org_user_ids'] : []), + ...(needsEmail ? ['current_user.email'] : []), + ].join(' or '); + // Without grants the lost list may be empty (the context carried + // its peers and email); with grants carried it never is, since + // one of the two is what sent the run here. + const consequence = needsGrants + ? `its data ops fall back to baseline member permissions (not elevated)` + + (lost ? `, and every row policy reading ${lost} drops out` : '') + : `its data ops keep the trigger's positions and permission sets, but every row ` + + `policy reading ${lost} drops out (narrower, not elevated)`; + this.logger.warn( + `[runAs] flow '${flowName}' could not resolve grants for ${whose} user ` + + `'${runContext.userId}' — ${consequence}. The resolver's failure is in this record's meta.`, + describeThrownForLog(err), + ); + } + } +} + +/** The body of {@link AutomationEngine.resolveDecisionSegmentContext}, moved unedited; that method's docblock is its contract. */ +export async function resolveDecisionSegmentContext( + this: Pick, + run: SuspendedRun, + signal: ResumeSignal, +): Promise { + const context = run.context; + const decider = signal[RESUME_AUTHORITY_SERVICE] === true ? signal.resumedBy : undefined; + if (typeof decider !== 'string' || decider === '') return context; + if (context?.userId === decider) return context; + const switched: SeededRunContext = { + ...(context ?? {}), + userId: decider, + triggerUserId: runTriggerUserId(context) ?? null, + }; + delete switched.positions; + delete switched.permissions; + delete switched.org_user_ids; + // The trigger user's email is theirs too: kept under the decider's id, + // it would admit the decider to every row an email-keyed policy grants + // the submitter. Removed on both postures, and seeded for the decider + // below like their grants. + delete switched.email; + // A system run keeps its elevation: attribution moves, no grant is resolved. + if (switched.runAs === 'system') return switched; + await this.seedRunGrants(run.flowName, switched, 'deciding'); + return switched; +} + +/** The body of {@link AutomationEngine.expandDeclaredLookups}, moved unedited; that method's docblock is its contract. */ +export async function expandDeclaredLookups(this: Pick, flow: FlowParsed, runContext: AutomationContext): Promise { + if (!this.recordExpander) return; + const record = runContext.record as Record | undefined; + const id = record?.id; + const object = runContext.object; + if (!record || id == null || id === '' || !object) return; + + const startNode = flow.nodes.find((n) => n.type === 'start'); + const raw = (startNode?.config as { expand?: unknown } | undefined)?.expand; + const expandFields = + typeof raw === 'string' + ? raw + ? [raw] + : [] + : Array.isArray(raw) + ? raw.filter((f): f is string => typeof f === 'string' && f.length > 0) + : []; + if (expandFields.length === 0) return; + + try { + const full = await this.recordExpander(object, id, expandFields, runContext); + if (full && typeof full === 'object') { + for (const field of expandFields) { + const expanded = (full as Record)[field]; + // Graft only a genuinely expanded relation (object/array); a + // scalar means the field is not a resolvable lookup — leave the + // raw id in place rather than overwrite it. + if (expanded !== null && typeof expanded === 'object') { + record[field] = expanded; + } + } + } + } catch (err) { + // #6499 — the expander's thrown text (transitively the + // datasource's) goes to the structured slot; see + // `forgetSuspendedRun`'s catch above for the full mechanism + // (#6299). #4632 verdict: FUNCTIONAL — stays `warn`, per the + // docblock: expansion is best-effort enrichment that must never + // break the flow it feeds, and the visible symptom is templates + // rendering the scalar id. + this.logger.warn( + `[expand] flow '${flow.name}' could not expand lookups [${expandFields.join(', ')}] on ` + + `'${object}#${String(id)}' — templates referencing these relations resolve to the scalar ` + + `id. The expander's failure is in this record's meta.`, + describeThrownForLog(err), + ); + } +} + +/** The body of {@link AutomationEngine.resume}, moved unedited; that method's docblock is its contract. */ +export async function resume(this: Pick, runId: string, signal?: ResumeSignal): Promise { + const refusal = await this.refuseGatedResume(runId, signal); + if (refusal) return refusal; + // An ABSENT signal is an EMPTY caller submission, never an exemption + // (commit 7307191db). This is the in-process door, and `resume(runId)` used to + // skip the screen contract that `resume(runId, {})` is held to: + // `refuseInvalidScreenInput` short-circuited on a falsy signal — a + // second, unnamed spelling of the exemption the engine already states + // through `ENGINE_BUILT_SIGNAL`, so a run parked on a screen with an + // unconditional `required` field proceeded with that variable unbound. + // The HTTP door has always assembled `{}` for an empty body; this makes + // the two doors agree, and the only exemption left is the engine's own + // continuation, which proves itself by BUILDING an engine-built signal. + // A pause with no screen contract — `wait`, `approval`, a message-only + // or object-form screen — is untouched: an empty submission against no + // declared fields is conformant, so the wait node's timer wake + // (`engine.resume(runId)`) continues exactly as before. + return this.resumeInternal(runId, signal ?? {}, false); +} + +/** The body of {@link AutomationEngine.refuseGatedResume}, moved unedited; that method's docblock is its contract. */ +export async function refuseGatedResume(this: Pick, runId: string, signal?: ResumeSignal): Promise { + const run = await this.resolveEffectiveSuspension(runId); + if (!run) return null; + + const nodeType = this.resolveSuspendedNodeType(run); + if (!nodeType) return null; + if (this.resolveResumeAuthority(nodeType) !== 'service') return null; + // The owning service stamped the signal — this resume IS the recorded + // decision's tail, not a way around it. + if (signal?.[RESUME_AUTHORITY_SERVICE]) return null; + + // Refusing. Which of the two reasons? A second registry walk, on the + // refusal path only, so the message can tell a node that deliberately + // declared `'service'` from one that declared nothing at all. + const declared = this.resolveDeclaredResumeAuthority(nodeType); + const direct = run.runId === runId; + const at = direct ? `'${run.nodeId}'` : `'${run.nodeId}' (linked run '${run.runId}')`; + const why = declared === 'service' + ? `which is resumable only through its owning service (resumeAuthority: 'service')` + : `whose type never declares resumeAuthority, so it is closed to the generic route until it does ` + + `(an undeclared resumeAuthority resolves to 'service', fail-closed) — declare ` + + `resumeAuthority: 'any' on its descriptor if this route IS the intended door`; + this.logger.warn(`[automation] refused resume of run '${runId}': parked on ${nodeType} node ${at}, ${why}`); + + // The fix, identical in both the direct and the linked-run phrasing — + // what has to change is a descriptor, not the call that just failed. + const undeclaredFix = + `and that node type never declares resumeAuthority, so the generic resume route is closed to the ` + + `pauses it creates: an undeclared resumeAuthority resolves to 'service', fail-closed. If that ` + + `route IS the intended door — a screen's collected inputs, a ` + + `signal wait's external producer — declare resumeAuthority: 'any' on its action descriptor; declare ` + + `'service' if resuming is the tail of a decision some service must authorize and record first`; + return { + success: false, + code: 'PERMISSION_DENIED', + error: declared === 'service' + ? direct + ? `Run '${runId}' is paused at a '${nodeType}' node, which only its owning service may resume — ` + + `drive it through that service's API (e.g. an approval decision), not a raw resume` + : `Run '${runId}' is waiting on run '${run.runId}', which is paused at a '${nodeType}' node that ` + + `only its owning service may resume — resuming here would continue past a decision that has not ` + + `been made; drive it through that service's API instead` + : direct + ? `Run '${runId}' is paused at a '${nodeType}' node, ${undeclaredFix}` + : `Run '${runId}' is waiting on run '${run.runId}', which is paused at a '${nodeType}' node, ` + + `${undeclaredFix}`, + }; +} + +/** The body of {@link AutomationEngine.refuseUndeclaredSuspension}, moved unedited; that method's docblock is its contract. */ +export function refuseUndeclaredSuspension(this: Pick, nodeType: string): NodeExecutionResult | null { + const descriptor = this.resolveCanonicalDescriptor(nodeType); + // No descriptor ⇒ no declaration ⇒ nothing to enforce (see above). + if (!descriptor) return null; + if (descriptor.supportsPause === true) return null; + return refuseNode( + `node type '${nodeType}' suspended the run but its action descriptor declares ` + + `supportsPause: false, so the pause is refused — a run that paused here could not be ` + + `continued on the generic resume route anyway: a type that declares no pause declares no ` + + `resumeAuthority either, and an undeclared resumeAuthority resolves to 'service', which the ` + + `generic route refuses. Declare ` + + `supportsPause: true on the descriptor together with the resumeAuthority the pauses need ` + + `('any' if POST /automation/:name/runs/:runId/resume is the intended door, 'service' if ` + + `resuming is the tail of a decision some service must authorize and record first) — or stop ` + + `returning suspend: true from execute(). This is a metadata defect, not a runtime one, so a ` + + `fault edge does not route it.`, + ); +} + +/** The body of {@link AutomationEngine.loadSuspendedRun}, moved unedited; that method's docblock is its contract. */ +export async function loadSuspendedRun(this: Pick, runId: string): Promise { + try { + return await this.loadSuspendedRunStrict(runId); + } catch (err) { + // #6230 — the cause goes to `meta`, never into the message. It 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 / + // #5912, and cloud#971's shape. + // + // This seam is the `warn` half of that family and carries the extra + // harm the module docblock of ./thrown-cause-diagnostics.ts calls + // out: `ObjectLogger` routes `warn` to **stdout**, and `serve`'s + // boot-quiet window wraps `process.stdout.write`, where + // `BootLogCapture.offer()` retains a physical line only when + // `classifyBootLogLine` finds a level head on it. Continuation lines + // are therefore DROPPED outright, not merely misread — and this is + // live during boot: `plugin.ts` `start()` → `rearmSuspendedWaitTimers` + // → `engine.resume()` for an overdue run → the gate + // (`refuseGatedResume` → `resolveEffectiveSuspension`) → here. + // + // SECOND argument, per the `Logger` contract + // (`packages/spec/src/contracts/logger.ts`): `warn(message, meta?)`. + // Unlike `error(message, error?, meta?)` (#5912's seam, PR #6228) + // `warn` has no `Error` slot, so `meta` is the second parameter. + // + // Level stays `warn` on purpose. This is a deliberate FUNCTIONAL + // degradation — the loader's whole contract is "a best-effort answer + // for incidental readers" and `resumeInternal` uses the strict form + // when the difference matters — so #4632's durability rule does not + // apply and raising it to `error` would be that rule's mirror-image + // misuse, an alarm on every gate lookup during an outage. + this.logger.warn( + `[automation] durable suspended-run store unreadable for run '${runId}' — this read is best-effort ` + + `and DEGRADES to null, so its caller (the resume gate, a screen fetch) sees exactly what it ` + + `would see if no suspension existed under that id; the run itself is untouched and stays ` + + `parked. Fix the store failure in this record's meta — a strict read reports it as ` + + `STORE_UNAVAILABLE instead of degrading.`, + describeThrownForLog(err), + ); + return null; + } +} + +/** The body of {@link AutomationEngine.loadSuspendedRunStrict}, moved unedited; that method's docblock is its contract. */ +export async function loadSuspendedRunStrict(this: Pick, runId: string): Promise { + if (!this.store) return this.suspendedRuns.get(runId) ?? null; + const stored = await this.store.load(runId); + if (stored) return stored; + // The store has no row. For every run it ever accepted that IS the + // answer — including the runs this process advanced past, whose stale + // map entries are the whole defect above. The lone exception is a run + // whose durable save failed here: the store was never handed that row, + // so its silence says nothing about it, and `persistSuspendedRun` + // deliberately keeps such a run resumable in-process (it reports the + // lost durability at `error`). + if (this.cacheOnlySuspensions.has(runId)) return this.suspendedRuns.get(runId) ?? null; + // [#15832] The store ANSWERED, and the answer is "no row". Any entry + // this process still holds for that run is a run it parked and another + // replica consumed — the phantom the two listings hand back. This is + // the definitive per-id evidence the paragraph above already rests on, + // so the same reading that refuses to serve it here stops publishing it + // there. A store read that threw never reaches this line. + this.evictConsumedSuspension(runId); + return null; +} + +/** The body of {@link AutomationEngine.creditChildRun}, moved unedited; that method's docblock is its contract. */ +export function creditChildRun(steps: StepLogEntry[], nodeId: string, child: FlowRunSummary | undefined): void { + if (!child) return; + for (let i = steps.length - 1; i >= 0; i--) { + if (steps[i].nodeId !== nodeId) continue; + const prior = steps[i].metrics ?? {}; + steps[i] = { + ...steps[i], + metrics: { + selected: (prior.selected ?? 0) + child.selected, + acted: (prior.acted ?? 0) + child.acted, + // N uncountable effects in the child collapse to ONE flag on + // the parent's step: this execution dispatched something the + // platform cannot count. The child keeps the real count in + // its own run row, and the question this feeds — "is the + // parent's `acted` complete?" — is boolean either way. + ...(prior.unmeasuredEffect || child.unmeasured ? { unmeasuredEffect: true } : {}), + // #15617 — the pausing item's own contained failures, which + // the synchronous path reports through the executor's + // `metrics`. Written only when one of the two sides actually + // tracked a count: an absent `failures` means "not tracked", + // and a `0` written here would claim a measurement of a + // child recorded before the count existed. + ...(prior.failures !== undefined || child.failed !== undefined + ? { failures: (prior.failures ?? 0) + (child.failed ?? 0) } + : {}), + }, + }; + return; + } +} + +/** The body of {@link AutomationEngine.failSuspendedRun}, moved unedited; that method's docblock is its contract. */ +export async function failSuspendedRun( + this: Pick, + run: SuspendedRun, + error: string, + repairableDescendant?: string, +): Promise { + const consumed = repairableDescendant !== undefined + ? this.journalConsumedSuspension(run, run.steps.length, error, run.variables) + : undefined; + await this.forgetSuspendedRun(run, 'failed'); + this.recordLog({ + id: run.runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'failed', + startedAt: run.startedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - run.startTime, + trigger: buildRunTrigger(run.context), + steps: run.steps, + error, + }, run.context, consumed?.run); +} + +/** The body of {@link AutomationEngine.settleUnresumableSuspension}, moved unedited; that method's docblock is its contract. */ +export async function settleUnresumableSuspension(this: Pick, run: SuspendedRun, cause: string): Promise { + const refusal: AutomationResult = { success: false, code: 'RUN_NOT_FOUND', error: cause }; + if (!this.nodeTypeVocabularySealed) return refusal; + const claim = await this.claimAdvance(run); + if (claim.kind === 'lost') { + // [#15832] The store answered and holds no row where this replica + // read one, so whatever snapshot this process still holds is stale. + this.evictConsumedSuspension(run.runId); + return refusal; + } + if (claim.kind === 'unavailable') { + return { + success: false, + code: 'STORE_UNAVAILABLE', + error: + `Run '${run.runId}' can never continue (${cause}), but the durable suspended-run store was ` + + `unreachable while settling it, and whether the store consumed the suspension is UNKNOWN (a ` + + `failure can arrive after a committed delete), so its record may still read paused. Retry once ` + + `the store is available: a retry that finds the run still parked settles it, and one that finds ` + + `the suspension gone answers RUN_NOT_FOUND: ${claim.message}`, + }; + } + await this.forgetSuspendedRun(run, 'failed', claim.kind === 'claimed'); + try { + this.recordLog({ + id: run.runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'failed', + startedAt: run.startedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - run.startTime, + trigger: buildRunTrigger(run.context), + steps: run.steps, + error: cause, + }, run.context); + } catch (bookkeeping) { + // #4632 verdict: DURABILITY, so `error`. The suspension is consumed + // and the caller is answered the truthful refusal, so nothing reads + // as wrong from the outside. But the `failed` row that says why + // never landed, and nothing retries it. Same sink and same level as + // the #16274 completion guard. THIRD argument per + // `error(message, error?, meta?)`; the `Error` slot stays empty + // (#5575) and the thrown text rides the structured slot (#6499). + this.logger.error( + `[Automation] run '${run.runId}' of flow '${run.flowName}' can never continue and its suspension ` + + `was consumed, but recording its terminal 'failed' history threw, so no history row says why: ` + + `the caller was answered RUN_NOT_FOUND, the run no longer reads as paused, and nothing retries ` + + `the write. Fix the history failure in this record's meta.`, + undefined, + describeThrownForLog(bookkeeping), + ); + } + return refusal; +} + +/** The body of {@link AutomationEngine.cancelRun}, moved unedited; that method's docblock is its contract. */ +export async function cancelRun(this: Pick, runId: string, reason?: string): Promise { + let run: SuspendedRun | null = null; + try { + run = await this.loadSuspendedRunStrict(runId); + } catch (err) { + // #6299 — same family, same mechanism as `forgetSuspendedRun` + // above: the driver's uncontrolled text goes to the structured + // slot so the record stays one physical line. + // + // #4632 verdict: DURABILITY — raised from `warn` to `error`. The + // failed read is silently turned into "no such suspended run" + // and this method returns `false`, which its own contract + // documents as idempotent success (already terminal / unknown), + // so the cancellation is SKIPPED while the call reads clean. The + // only in-repo caller measures the cost: plugin-approvals' + // revise-window recall + // (`packages/plugins/plugin-approvals/src/approval-service.ts`) + // never reads the boolean at all — it only catches a THROW, and + // grades that throw `error` with "the run may be stranded" + // (#4420). A store-read failure produces precisely that stranded + // run WITHOUT firing that alarm: the request is marked + // `recalled`, the record lock is released, `resumeError` stays + // undefined — and the run stays parked in the store, to be + // re-armed and resumed by the next restart, inside a flow whose + // approval has already been withdrawn. + // + // This is why #6230's verdict must not be copied here. + // `loadSuspendedRun` is a DECLARED best-effort reader for + // incidental callers (a gate lookup, a screen fetch), and + // `resumeInternal` takes the strict form exactly where the + // difference matters. `cancelRun` has no strict alternative, and + // its degradation decides a WRITE. + // + // THIRD argument (`error(message, error?, meta?)`), `Error` slot + // deliberately empty (#5575). + this.logger.error( + `[automation] cancelRun('${runId}') could not read the durable suspended-run store, so the ` + + `cancellation was SKIPPED and reported as idempotent success — this call returns false, which ` + + `its callers read as "no such suspended run". The run is NOT cancelled: if it is parked in the ` + + `store it stays parked, and the next restart re-arms and resumes it while the caller has ` + + `already recorded the cancellation. Fix the store failure in this record's meta, then re-issue ` + + `cancelRun('${runId}').`, + undefined, + describeThrownForLog(err), + ); + } + if (!run) return false; + await this.forgetSuspendedRun(run, 'cancelled'); + this.recordLog({ + id: run.runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'cancelled', + startedAt: run.startedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - run.startTime, + trigger: buildRunTrigger(run.context), + steps: run.steps, + error: reason, + }, run.context); + return true; +} + +/** The body of {@link AutomationEngine.cloneVariablesAtPause}, moved unedited; that method's docblock is its contract. */ +export function cloneVariablesAtPause(this: Pick, run: SuspendedRun, runId: string): SuspendedRun['variables'] { + try { + return JSON.parse(JSON.stringify(run.variables)) as SuspendedRun['variables']; + } catch (err) { + // #6299 family — the thrown text is not ours to shape, so it rides + // the structured slot and never the message. + this.logger.warn( + `[automation] run '${runId}': could not copy the paused variables before resuming, so if this ` + + `resume strands the run, the suspension an operator restores may carry state this attempt ` + + `wrote rather than the state at the pause. The run resumes normally either way.`, + describeThrownForLog(err), + ); + return run.variables; + } +} + +/** The body of {@link AutomationEngine.failAncestors}, moved unedited; that method's docblock is its contract. */ +export async function failAncestors( + this: Pick, + context: AutomationContext | undefined, + error: string, + strandedRunId?: string, +): Promise { + let parentId = (context as Record | undefined)?.$parentRunId; + let below = strandedRunId; + let hops = 0; + while (typeof parentId === 'string' && parentId && hops++ < 32) { + // [#14332] The DEGRADING loader, by deliberate choice: this walk runs + // inside the catch arm that is already handling a run's failure, so it + // must not throw, and its recorded posture is exactly + // `loadSuspendedRun`'s — a store failure reads as "no ancestor here" + // and stops the walk. What changes is only WHICH suspension is read: + // the shared store's, not this replica's memory of where the parent + // was last parked. The old `??` chain had #13617's own harm shape — + // a stale parent failed at a node it had already left, so + // `forgetSuspendedRun` released the wrong node's pause. The one thing + // gained beyond that: the bare `.catch(() => null)` swallowed a store + // failure in total silence, and the loader records it (at `warn`, + // its declared best-effort level — no new `error` seam here). + const parent = await this.loadSuspendedRun(parentId); + if (!parent) return; + await this.failSuspendedRun( + parent, + `subflow descendant failed: ${error}`, + // [#15222] Earned, not assumed: only a descendant this process + // actually holds a snapshot for makes the ancestor's own + // snapshot a promise the restore verb can keep. + below !== undefined && this.consumedSuspensions.has(below) ? below : undefined, + ); + below = parent.runId; + parentId = (parent.context as Record | undefined)?.$parentRunId; + } +}