From 9047bcb70475ef42a87dc04015a1cdccf0ee7326 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 22:58:20 +0000 Subject: [PATCH] refactor(service-automation): move the run-history bodies into run-history.ts Move-only. The bodies of nine AutomationEngine methods move unedited into a new run-history.ts: listRuns, listRunsPage, runRecordToLogEntry, suspendedRunToLogEntry, getRun, listSuspendedRuns and readSuspendedRuns, which read runs back, and recordLog and finishRefusedRun, which record them. Each method keeps its name, docblock, signature and position in the class and calls its body with a RunHistoryHost literal holding exactly the members the body reads; runRecordToLogEntry and suspendedRunToLogEntry read none and are 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 goes through a getter, because setSuspendedRunStore replaces it. formatRunSummaryLine's import moves with recordLog, its one reader. 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 | 739 ++--------------- .../src/flow-activation-ledger.test.ts | 2 +- .../service-automation/src/run-history.ts | 781 ++++++++++++++++++ 3 files changed, 862 insertions(+), 660 deletions(-) create mode 100644 packages/services/service-automation/src/run-history.ts diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index cba5f7e80be..e870a3bec66 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -66,7 +66,7 @@ export const DEFAULT_BRANCH_LABEL = 'default'; import { runIsUnscopedUserMode, flowTouchesData, runOrgUserIds, runTriggerUserId, type SeededRunContext } from './runtime-identity.js'; import { refuseNode } from './guard-refusal.js'; import { carriedRefusalOfResult, readCarriedRefusal, type CarriedRefusal } from './inner-refusal.js'; -import { summarizeRun, formatRunSummaryLine } from './run-summary.js'; +import { summarizeRun } from './run-summary.js'; // #5660 — the degrade registration reports a FOREIGN failure (a third-party // provider factory's text), so it renders it as structured `meta` rather than // interpolating it into the log message. See ./thrown-cause-diagnostics.ts. @@ -127,6 +127,20 @@ import { executeWithoutRetry, type runUserOf, } from './dag-traversal.js'; +// [#25222] The run history's bodies: recording a run's log entry, finishing a +// refused run, and reading runs and suspended runs back. Each +// `AutomationEngine` method of the same name calls its body. +import { + listRuns, + listRunsPage, + runRecordToLogEntry, + suspendedRunToLogEntry, + getRun, + listSuspendedRuns, + readSuspendedRuns, + recordLog, + finishRefusedRun, +} from './run-history.js'; // ─── Node Executor Interface (Plugin Extension Point) ─────────────── @@ -5586,16 +5600,15 @@ export class AutomationEngine implements IAutomationService { this.logger.info(`Flow '${name}' rolled back to version ${version}`); } - async listRuns( + listRuns( flowName: string, options?: { limit?: number; status?: ExecutionStatus }, ): Promise { - // [#19543] ONE implementation, two projections — `listRunsPage` is the - // whole method and this is its `runs` half. ⛔ Never re-derive the - // listing here: a second copy of the merge/filter/sort would be the - // fork the route-ownership rule refuses, and it is the half that would - // rot, because the page method is the one the REST door calls. - return (await this.listRunsPage(flowName, options)).runs; + // The body lives in ./run-history.ts (#25222). + const host: ThisParameterType = { + listRunsPage: (...a) => this.listRunsPage(...a), + }; + return listRuns.call(host, flowName, options); } /** @@ -5647,229 +5660,28 @@ export class AutomationEngine implements IAutomationService { * this card's; the same paragraph below the merge already records it for * the listing itself. */ - async listRunsPage( + listRunsPage( flowName: string, options?: { limit?: number; status?: ExecutionStatus }, ): Promise { - const limit = options?.limit ?? 20; - const inMem = this.executionLogs.filter(l => l.flowName === flowName); - - // [#8050] Durable PAUSED rows — the arm this merge was missing. - // - // `sys_automation_run` holds TWO disjoint row families: the terminal - // history rows `recordTerminal` writes (id `run_` + runId, status - // completed/failed) and the LIVE suspension rows `save` writes (id = - // the raw runId, status `paused`). The history arm below reads the - // first family; nothing here read the second. Before a restart that is - // invisible, because a paused run is still in `executionLogs` — so - // every in-process test of this method passes. After a restart the ring - // is empty and the paused rows had no reader at all, which is the - // defect: an operator who restarts the process can enumerate what - // FINISHED and what is IN FLIGHT vanishes — the strictly more urgent - // half. It also structurally emptied #7359's just-enforced - // `?status=paused`: with no producer of a `paused` entry after a - // restart, that filter could never match a row, and the one query - // reached for here always answered "nothing pending". - // - // Read-side only, by construction: this rehydrates the row the suspend - // path already writes. No column, prefix or lifecycle changes — the - // paused row is NOT reshaped into a history row, because the two have - // different lifetimes (a paused row is live resumable state, deleted on - // completion and exempt from the age sweep; a history row is a - // tombstone). Unifying them would trade an observability gap for a - // persistence-semantics change. - // - // Skipped when the caller filters for a status a paused row can never - // have: `suspendedRunToLogEntry` always yields `paused`, and a run that - // has since finished is answered by the fresher history/ring entry that - // outranks it in the merge below — so the arm cannot change the result - // of `?status=failed`, only its cost. `store.list()` is a table scan of - // every paused row in the deployment (see its own contract), so not - // paying it on the monitoring queries is worth the one-line guard. - const wantsPaused = options?.status === undefined || options.status === 'paused'; - let durablePaused: ExecutionLogEntry[] = []; - if (this.store && wantsPaused) { - try { - const rows = await this.store.list(); - durablePaused = rows - .filter(r => r.flowName === flowName) - .map(r => this.suspendedRunToLogEntry(r)); - } catch (err) { - // #6499 — driver text to the structured slot; `warn(message, - // meta?)`, meta SECOND (no `Error` slot on `warn`). - // - // #4632 verdict: FUNCTIONAL — `warn`, for the same reason as - // the history arm below and `listSuspendedRunsDurable`. Nothing - // claimed-persisted failed to land: the paused rows are intact - // and still resumable by id (`resume` reads them through - // `loadSuspendedRun`, a different door that is unaffected by - // this failure). What degrades is this observability read, back - // to exactly the pre-#8050 answer. - this.logger.warn( - `[Automation] paused-run read failed for '${flowName}' — the Runs listing DEGRADES to the ` + - `in-memory ring buffer plus terminal history, so runs parked by a previous process are ` + - `missing and '?status=paused' can report an empty result for a flow that has runs ` + - `waiting. The rows themselves are untouched and still resumable by id. Fix the store ` + - `failure in this record's meta.`, - describeThrownForLog(err), - ); - } - } - - // Merge durable run history so the "Runs" view survives a restart and - // ring-buffer eviction. In-memory entries are the freshest (they carry - // full step detail); durable rows backfill runs the process no longer - // holds. Best-effort: a history-read failure degrades to in-memory only. - let durable: ExecutionLogEntry[] = []; - if (this.store?.listHistory) { - try { - // [#19543] `limit + 1`, not `limit` — the over-read that makes - // `hasMore` answerable at all. Asking for exactly `limit` makes a - // saturated window and a complete one identical; one extra row - // tells them apart, and `.slice(0, limit)` below drops it again - // so nothing on the wire widens. - const rows = await this.store.listHistory(flowName, limit + 1); - durable = rows.map(r => this.runRecordToLogEntry(r)); - } catch (err) { - // #6499 — the datasource driver's text to the structured slot; - // see `forgetSuspendedRun`'s catch above for the full mechanism - // (#6299). `warn(message, meta?)` — meta is the SECOND - // argument; `warn` has no `Error` slot. - // - // #4632 verdict: FUNCTIONAL — stays `warn`, the same reasoning - // as `listSuspendedRunsDurable` below: nothing - // claimed-persisted failed to land — the history rows are - // intact — and this read feeds only the observability Runs - // view, which degrades to the in-memory ring buffer. The - // record says the shortfall out loud (#5186's invented-answer - // shape; its propagation remedy is a return-contract change - // outside #6499). - this.logger.warn( - `[Automation] run-history read failed for '${flowName}' — the Runs listing DEGRADES to the ` + - `in-memory ring buffer alone, so terminal runs from before the last restart (or evicted ` + - `from the buffer) are missing and the caller cannot tell a short list from a complete ` + - `one. The rows themselves are untouched. Fix the store failure in this record's meta.`, - describeThrownForLog(err), - ); - } - } - // Dedupe by run id, weakest source first — the same run legitimately - // appears in more than one of these (#8050): - // - // 1. durable PAUSED — the run parked, and the row is still there. - // 2. durable HISTORY — the run reached a terminal state. - // 3. in-memory ring — this process executed it. - // - // The order is a precedence claim, not an accident. Paused loses to - // both because it is the only one that can be STALE while the others - // cannot: `forgetSuspendedRun` deletes the paused row on completion, - // but that delete is best-effort (a store outage swallows it), so a - // finished run can leave its paused row behind. A terminal row or a - // terminal ring entry for the same id is therefore strictly later - // evidence, and letting the paused row win would report a completed run - // as still waiting — the exact defect `run-history.test.ts`'s "latest - // entry wins" block pins for `getRun`. There is no symmetric hazard: - // within a process the ring is written in the same breath as the paused - // row (`persistSuspendedRun` then `recordLog`, and again on re-suspend), - // so it is never the older of the two; across a restart the ring is - // empty and cannot mask anything. - const byId = new Map(); - for (const e of durablePaused) byId.set(e.id, e); - for (const e of durable) byId.set(e.id, e); - for (const e of inMem) byId.set(e.id, e); // freshest wins - - // [#7359] The wire has always declared `?status=`, but it reached - // neither the contract nor this method, so the filter was dropped at - // the HTTP boundary and every run came back as if it matched — a - // monitoring caller paging for failures read the first `limit` runs of - // any status and concluded those were the failures. - // - // Filtering HERE, after the merge, is load-bearing in two ways: - // - // 1. It covers BOTH stores at once — the in-memory ring buffer and the - // durable rows. Narrowing only the buffer would answer "no - // failures" for a flow whose failures are all in durable history - // (i.e. after any restart, which is exactly when someone asks); - // narrowing only the durable rows would hide the live ones. Half a - // filter is the same class of confident wrong answer as none. - // 2. It reads each run's RESOLVED status. `executionLogs` holds more - // than one entry per run id — a run that pauses and later finishes - // appends 'paused' and then its terminal entry — and the merge above - // is what collapses them to the freshest. Filtering the two arms - // BEFORE that collapse would drop the terminal entry for - // `?status=paused` and let the stale 'paused' one survive, so every - // approval / screen / wait run that had since completed would report - // itself as still paused. That is the exact defect - // `run-history.test.ts`'s "latest entry wins" block pins for - // `getRun`, re-introduced one method over. - // - // The durable arm's window is still the store's newest `limit` rows — - // `listHistory(flowName, limit)` has no status slot, so the filter is - // applied to what comes back rather than pushed down. A status filter - // therefore narrows WITHIN that window and can return fewer than - // `limit` matches while older ones exist. That is the merge's - // pre-existing shape (durable was already capped at `limit` before the - // sort-and-slice) and closing it is a store-contract change, not this - // card's. What it never does is return a run of another status. - const status = options?.status; - const merged = status === undefined - ? [...byId.values()] - : [...byId.values()].filter(e => e.status === status); - const ordered = merged - .sort((a, b) => (b.startedAt ?? '').localeCompare(a.startedAt ?? '')); - // [#19543] The comparison is against the ORDERED, FILTERED set, not - // against what any single source returned: a row can reach `ordered` - // from the ring or the paused arm without the history arm knowing, and - // a `?status=` filter can drop the over-read row specifically. Reading - // the overflow here — after every arm has contributed and after the - // filter has run — is what makes the answer true of the response - // actually being sent. - return { runs: ordered.slice(0, limit), hasMore: ordered.length > limit }; + // The body lives in ./run-history.ts (#25222). + const readStore = () => this.store; + const host: ThisParameterType = { + executionLogs: this.executionLogs, + get store() { return readStore(); }, + suspendedRunToLogEntry: (...a) => this.suspendedRunToLogEntry(...a), + logger: this.logger, + runRecordToLogEntry: (...a) => this.runRecordToLogEntry(...a), + }; + return listRunsPage.call(host, flowName, options); } /** Rehydrate a durable {@link RunRecord} into an {@link ExecutionLogEntry} * for the Runs surfaces. Steps carry the bounded persisted step log (rows * written before step persistence have none). */ private runRecordToLogEntry(r: RunRecord): ExecutionLogEntry { - return { - id: r.runId, - flowName: r.flowName, - flowVersion: r.flowVersion, - // [#15223] All four {@link TERMINAL_RUN_STATUSES} members, each a - // valid `ExecutionLog` status — the schema has declared the whole - // vocabulary since before this row existed, and it was the - // persistence layer, not the contract, that reported only two. - status: r.status, - startedAt: r.startedAt, - completedAt: r.finishedAt, - durationMs: r.durationMs, - // #7533 — the persisted trigger attribution, so a run rehydrated - // after a restart still says what fired it and (for a record - // change) which record. `type` falls back to `''` ONLY for rows - // written before those columns existed: the field is required by - // `ExecutionLogSchema`, and `''` is the same "not recorded" this - // method already returned for every row. It is a legacy-row - // default, not an alias — a row written today always carries a - // real type. - trigger: { - type: r.triggerType ?? '', - userId: r.userId ?? undefined, - object: r.triggerObject, - recordId: r.triggerRecordId, - }, - steps: r.steps ?? [], - error: r.error, - // [#15788] The persisted refusal text, so the run a caller opens - // after a restart still says WHY it refused. Without this line the - // status survives the round-trip and its reason does not, which is - // the worst of the two halves to lose: `refused` with no message is - // indistinguishable from a refusal nobody authored. - refusalMessage: r.refusalMessage, - // #4354 — the PERSISTED summary, never re-folded from `r.steps`: - // those are compacted (200 max), so recomputing here would report a - // 5000-row sweep as having acted on a couple of hundred. - summary: r.summary, - }; + // The body lives in ./run-history.ts (#25222). + return runRecordToLogEntry(r); } /** @@ -5901,94 +5713,21 @@ export class AutomationEngine implements IAutomationService { * is read and is not the "time spent executing" the ring entry reports. */ private suspendedRunToLogEntry(run: SuspendedRun): ExecutionLogEntry { - return { - id: run.runId, - flowName: run.flowName, - flowVersion: run.flowVersion, - status: 'paused', - startedAt: run.startedAt, - trigger: buildRunTrigger(run.context), - steps: run.steps ?? [], - variables: run.variables ?? {}, - }; + // The body lives in ./run-history.ts (#25222). + return suspendedRunToLogEntry(run); } - async getRun(runId: string): Promise { - // LAST entry wins, not the first: a run that pauses and later finishes - // records TWO entries under the same run id ('paused', then - // 'completed'/'failed'/'cancelled'). Scanning forwards returned the stale - // 'paused' one, so every suspend-then-finish run — i.e. every approval, - // screen and wait flow — reported itself as still paused forever, both on - // the Runs surface and to the approvals dead-run sweep (#3456). - let inMem: ExecutionLogEntry | undefined; - for (let i = this.executionLogs.length - 1; i >= 0; i--) { - if (this.executionLogs[i].id === runId) { inMem = this.executionLogs[i]; break; } - } - if (inMem) return inMem; - // Durable fallback: after a restart (or ring-buffer eviction) the run's - // terminal history row still answers "what happened, at which node?". - // Best-effort — a store failure degrades to "not found", never throws. - if (this.store?.loadTerminal) { - try { - const rec = await this.store.loadTerminal(runId); - if (rec) return this.runRecordToLogEntry(rec); - } catch (err) { - // #6499 — driver text to the structured slot; see - // `forgetSuspendedRun`'s catch above for the full mechanism - // (#6299). - // - // #4632 verdict: FUNCTIONAL — stays `warn`. Nothing - // claimed-persisted failed to land: the terminal row, if one - // exists, is intact — this read degrades to `null`. What IS - // wrong is #5186's shape: `null` is also this method's honest - // "no such run", so a caller cannot tell an unreadable store - // from a run that never ran (plugin-approvals' - // `inspectStrandedRequests` counts a THROWN getRun as - // `undetermined` — a distinction this swallow denies it). The - // remedy is propagation, a return-contract change outside - // #6499's scope, so the record says the degradation out loud. - this.logger.warn( - `[Automation] durable run lookup failed for '${runId}' — this read DEGRADES to null, so ` + - `the caller sees exactly what it would see if the run had never run and cannot tell ` + - `the two apart. The terminal row, if one exists, is untouched. Fix the store failure ` + - `in this record's meta.`, - describeThrownForLog(err), - ); - } - } - // [#8050] …and the PAUSED fallback, so run-detail and `listRuns` answer - // out of one story. The card measured both surfaces failing together - // after a restart — list returning zero rows AND this method 404ing — - // and fixing only the list would have swapped a visible gap for an - // inconsistency between two reads of the same run. - // - // AFTER the terminal probe, matching the merge order in `listRuns`: a - // paused row can outlive the run it describes (the delete on completion - // is best-effort), so a terminal row for the same id is later evidence - // and must win. Trying this first would resurrect #3456's "paused - // forever" for any run whose cleanup delete was lost. - // - // This does NOT make a nonexistent run findable: `store.load` answers - // `null` for an unknown id exactly as `loadTerminal` does, so the route - // above still returns its 404 `RESOURCE_NOT_FOUND` envelope for one. - if (this.store) { - try { - const suspended = await this.store.load(runId); - if (suspended) return this.suspendedRunToLogEntry(suspended); - } catch (err) { - // #6499 / #4632: same verdict as the terminal probe above — - // FUNCTIONAL, so `warn`. The suspension row is intact and the - // run stays parked and resumable; what degrades is this read. - this.logger.warn( - `[Automation] durable paused-run lookup failed for '${runId}' — this read DEGRADES to null, ` + - `so a run that is parked and resumable reports as if it had never run, and the caller ` + - `cannot tell the two apart. The suspension row is untouched. Fix the store failure in ` + - `this record's meta.`, - describeThrownForLog(err), - ); - } - } - return null; + getRun(runId: string): Promise { + // The body lives in ./run-history.ts (#25222). + const readStore = () => this.store; + const host: ThisParameterType = { + executionLogs: this.executionLogs, + get store() { return readStore(); }, + runRecordToLogEntry: (...a) => this.runRecordToLogEntry(...a), + logger: this.logger, + suspendedRunToLogEntry: (...a) => this.suspendedRunToLogEntry(...a), + }; + return getRun.call(host, runId); } /** @@ -9806,12 +9545,11 @@ export class AutomationEngine implements IAutomationService { * {@link listSuspendedRunsDurable} to include those. */ listSuspendedRuns(): Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }> { - return [...this.suspendedRuns.values()].map(r => ({ - runId: r.runId, - flowName: r.flowName, - nodeId: r.nodeId, - correlation: r.correlation, - })); + // The body lives in ./run-history.ts (#25222). + const host: ThisParameterType = { + suspendedRuns: this.suspendedRuns, + }; + return listSuspendedRuns.call(host); } /** @@ -9850,135 +9588,20 @@ export class AutomationEngine implements IAutomationService { * read a page in which that caller's run could be row 1001. ⛔ Raising the * listing's cap would not have been this: a bigger page is still a page. */ - private async readSuspendedRuns( + private readSuspendedRuns( onEnumerationFailure: 'degrade' | 'throw', flowNames?: ReadonlySet, ): Promise> { - const byId = new Map(); - // [#15832] Did the ENUMERATION answer? The reconcile below is allowed - // only when it did — see the merge comment for why a failed listing is - // silence rather than evidence. - let enumerated = false; - if (this.store) { - try { - const stored = flowNames === undefined - ? await this.store.list() - : this.store.listByFlow - ? await this.store.listByFlow([...flowNames]) - : (await this.store.list()).filter((r) => flowNames.has(r.flowName)); - for (const r of stored) { - byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation }); - } - enumerated = true; - } catch (err) { - // A write-deciding caller is told, and decides nothing: see - // {@link readSuspendedRuns}. Everything below is the LISTING's - // posture. - if (onEnumerationFailure === 'throw') throw err; - // #6299 — driver text to the structured slot, message one line, - // same as the two seams above. The SLOT differs: the `Logger` - // contract declares `warn(message, meta?)`, so `meta` is the - // SECOND argument here — `warn` has no `Error` slot, and a - // `meta` passed third to it is silently ignored. - // - // #4632 verdict: FUNCTIONAL — the level deliberately STAYS - // `warn`, and this is the one of #6299's three sites where that - // is the answer. Nothing the system claims to have persisted - // failed to land: the durable rows are intact, still resumable - // by id, and the next boot's `rearmSuspendedWaitTimers` still - // re-arms them off its OWN `store.list()` — which builtin/ - // wait-node.ts grades `error` precisely because THAT failure - // breaks the promise to resume them. This one breaks no promise, - // so #4632's judgment question ("does something the system - // claims is persisted fail to land while it keeps looking - // healthy?") answers NO. - // - // What IS wrong here is the shape #5186 owns: an answer INVENTED - // for a read that failed — a silently SHORT list. That rule's - // remedy is propagation (rethrow, or a discriminated result), - // which changes this method's return contract and is outside - // #6299's scope, so the record has to say the shortfall out loud - // instead. Its scan roots (`packages/metadata`, - // `metadata-protocol`, `objectql`) do not reach this package, so - // `check:durability-log-level` reports neither rule here. - // Reachability is also the weakest of the three: this method is - // not on the `AutomationService` spec contract (only the - // synchronous `listSuspendedRuns` is); the in-repo caller that - // would ADMIT a write on what this list lacks — the §7.3 - // disable guard — takes the `'throw'` posture above and never - // reaches this line, and plugin-approvals' restored- - // continuation proof refuses on an absent entry, so a short - // list costs it a retry. Raising it to `error` would alarm for - // the duration of an outage on a read no caller acts on - // wrongly — #4632's mirror-image misuse, the trap #6230 - // avoided. - this.logger.warn( - `[automation] the durable suspended-run store could not be listed — this listing DEGRADES to the ` + - `in-memory cache alone, so every run parked by a previous process is missing from the result ` + - `and the caller cannot tell a short list from a complete one (after a restart the cache is ` + - `empty, so this answers []). The runs themselves are untouched — still stored, still resumable ` + - `by id. Fix the store failure in this record's meta.`, - describeThrownForLog(err), - ); - } - } - // [#14332] The DURABLE row wins a collision. The comment this replaces - // said the opposite — "In-memory entries win — they are the freshest - // copy" — which is true of exactly one deployment shape, a single - // process. Put several replicas over one store and this map is a - // per-replica snapshot of the node a run was parked at THE LAST TIME - // THIS REPLICA TOUCHED IT, with no invalidation channel to it at all - // (the mechanism is in {@link loadSuspendedRunStrict}), so preferring it - // reported a run at a node it had already left. - // - // Map entries the durable list does not carry are still appended, and - // that is NOT the strict loader's rule being softened: a LIST is not a - // per-id answer. `store.list()` is a capped, best-effort enumeration - // (`ObjectStoreSuspendedRunStore` reads at most 1000 `paused` rows) and - // this line is also reached on the DEGRADED path above, where the - // enumeration failed outright and `byId` is empty. So "absent from the - // list" is not the evidence "the store answered and has no row" is, - // which is what {@link loadSuspendedRunStrict} rests on when it lets - // only {@link cacheOnlySuspensions} answer out of the map. Applying that - // qualifier here would let a truncated or failed enumeration silently - // drop live runs from an operability listing. - // - // [#15832] What that reasoning leaves open is a run this process parked - // and ANOTHER replica has since consumed: absent from the durable list - // because it is finished, appended here, and published as suspended by - // a listing that also backs the cache-only one. The paragraph above is - // right that list-absence is not evidence — so this asks for the - // evidence instead. `store.load` is the same definitive per-id read - // {@link loadSuspendedRunStrict} rests on, and it is bought only for the - // entries that look suspicious: a healthy process, whose map entries all - // appear in the durable list, buys none. A read that THROWS leaves the - // entry standing (unknown is not gone), and a store that could not be - // enumerated at all is not probed row by row — an outage would answer - // for every live run in the process. - // - // [#20725] A scoped read takes only the named flows' entries — the hot - // map holds every run this process parked, uncapped, so it answers the - // scoped question completely for this process — and probes only those. - for (const r of [...this.suspendedRuns.values()]) { - if (flowNames !== undefined && !flowNames.has(r.flowName)) continue; - if (byId.has(r.runId)) continue; - if (enumerated && !this.cacheOnlySuspensions.has(r.runId)) { - let stored: SuspendedRun | null; - try { - stored = await this.store!.load(r.runId); - } catch { - // Unknown, not gone — keep the entry and publish it, exactly - // as this method did before the reconcile existed. - stored = r; - } - if (stored === null) { - this.evictConsumedSuspension(r.runId); - continue; - } - } - byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation }); - } - return [...byId.values()]; + // The body lives in ./run-history.ts (#25222). + const readStore = () => this.store; + const host: ThisParameterType = { + get store() { return readStore(); }, + logger: this.logger, + suspendedRuns: this.suspendedRuns, + cacheOnlySuspensions: this.cacheOnlySuspensions, + evictConsumedSuspension: (...a) => this.evictConsumedSuspension(...a), + }; + return readSuspendedRuns.call(host, onEnumerationFailure, flowNames); } /** @@ -10034,172 +9657,18 @@ export class AutomationEngine implements IAutomationService { context?: AutomationContext, consumedSuspension?: SuspendedRun, ): ExecutionLogEntry { - // #4354 — fold the run's outcome BEFORE anything downstream trims the - // step log. History compaction keeps 200 steps; the summary must count - // all 5000, or a long sweep's `acted` would shrink with its step log and - // the broken-sweep detector would read a bounded artefact as truth. - entry.summary = summarizeRun(entry.steps); - - this.executionLogs.push(entry); - // Evict oldest logs when exceeding max size - if (this.executionLogs.length > this.maxLogSize) { - this.executionLogs.splice(0, this.executionLogs.length - this.maxLogSize); - } - // Durable run history (observability): mirror every TERMINAL run to the - // store so "did it run / fail, and why?" survives a restart and the - // in-memory ring-buffer eviction. Best-effort + fire-and-forget: a - // history write must NEVER block or break the run that produced it. - // [#15223] ONE vocabulary, not a fourth copy of the list: this - // predicate decides both WHETHER a history row is written and WHAT its - // `status` says. Keeping the narrowed value in a `const` is what makes - // the record below type-check without a cast — and a cast is precisely - // how the fold this card is about survived four members for two. - const terminalStatus = isTerminalRunStatus(entry.status) ? entry.status : undefined; - const terminal = terminalStatus !== undefined; - - // The MVP of #4354, and the half that needs no console: one structured - // line per terminal run. `selected=30 acted=0` in a log file is the - // difference between an invisible failure and a greppable one. Only - // terminal runs — a `paused` run has not finished doing its work yet. - if (terminal && this.runSummaryLog !== 'off') { - const line = formatRunSummaryLine( - { - flowName: entry.flowName, - runId: entry.id, - status: entry.status, - durationMs: entry.durationMs, - }, - entry.summary, - ); - const meta = { - flow: entry.flowName, - runId: entry.id, - status: entry.status, - durationMs: entry.durationMs, - selected: entry.summary.selected, - acted: entry.summary.acted, - skipped: entry.summary.skipped, - unmeasured: entry.summary.unmeasured, - // #14456 — the structured half of the `failed=` token the line - // above now carries, so a log pipeline that reads `meta` and - // never parses the message sees the same fact. - failed: entry.summary.failed, - gates: entry.summary.gates, - }; - if (this.runSummaryLog === 'debug') this.logger.debug(line, meta); - else this.logger.info(line, meta); - } - - if (terminalStatus && this.store?.recordTerminal) { - const lastStep = entry.steps[entry.steps.length - 1]; - const record: RunRecord = { - runId: entry.id, - flowName: entry.flowName, - flowVersion: entry.flowVersion, - // [#15223] The status the run actually reached. This used to be - // `entry.status === 'completed' ? 'completed' : 'failed'` — a - // fold applied at WRITE time, so a cancelled or timed-out run's - // distinction was not merely unshown, it was never stored and - // could not be recovered afterwards. ⛔ Never re-introduce a - // conditional here: whatever the terminal predicate above - // admits is what the row must carry. - status: terminalStatus, - startedAt: entry.startedAt, - finishedAt: entry.completedAt, - durationMs: entry.durationMs, - error: entry.error, - // [#15788] The rendered refusal, carried onto the durable row - // beside `status: 'refused'`. Set by exactly one producer - // ({@link AutomationEngine.finishRefusedRun}) and `undefined` - // on every other terminal record, so the store writes an - // explicit NULL there — an upsert must CLEAR it, or a run id - // reused by a restore would keep a refusal it no longer has. - refusalMessage: entry.refusalMessage, - userId: entry.trigger?.userId, - // [#10101] The two organization-attribution inputs, from the - // run context (see the `recordLog` doc): the acting tenant is - // the ruled FALLBACK, the trigger-record snapshot is what the - // store resolves the SUBJECT organization from — the same - // snapshot the paused row resolves from, so the two rows of - // one run agree by construction. - organizationId: context?.tenantId ?? null, - triggerRecord: context?.record, - // #7533 — the rest of the trigger block, not just its userId. - // The information exists at this exact point (the in-memory log - // entry one line up carries it); it was simply not copied onto - // the record handed to the store, which is where the durable - // history lost the answer to "what fired this run?". - triggerType: entry.trigger?.type || undefined, - triggerObject: entry.trigger?.object, - triggerRecordId: entry.trigger?.recordId, - // [#13937] On a stranded run's record the PAUSE node, not the - // node that threw: the object store rebuilds the snapshot's - // node from this one column, and a restore from the row used - // to re-arm the run at the failed node — whose next resume - // then SKIPPED it and reported the run completed. The throwing - // node is the last entry of `steps` below, and `error`. - nodeId: consumedSuspension?.nodeId ?? lastStep?.nodeId, - steps: this.compactStepsForHistory(entry.steps), - summary: entry.summary, - // [#13909] Present only on the resume-consumed-then-failed - // path. On every other terminal record it is `undefined`, and - // the store writes explicit NULLs for it — so a run that is - // restored and then finishes CLEARS its own snapshot instead of - // leaving a stale one an operator could restore a second time. - consumedSuspension, - }; - // [#13937] The journal entry this record describes, when it is a - // strand's: `journalConsumedSuspension` ran just above in the - // same arm, so the entry under this run id is the one built from - // `consumedSuspension` — checked by identity, because a later - // strand of the same run replaces the entry and must not inherit - // this write's outcome. - const journal = consumedSuspension ? this.consumedSuspensions.get(entry.id) : undefined; - const write = this.store.recordTerminal(record); - if (journal && journal.run === consumedSuspension) { - void write.then( - () => { journal.persisted = 'landed'; }, - () => { journal.persisted = 'failed'; }, - ); - } - void write.catch((err) => { - // #6499 — driver text to the structured slot; see - // `forgetSuspendedRun`'s catch above for the full mechanism - // (#6299). - // - // #4632 verdict: DURABILITY — RAISED from `warn` to `error`. - // This is the WRITE half of the run-history claim whose read - // halves (`listRuns` / `getRun` above) stay `warn`: a TERMINAL - // run's history row failed to land while the run itself - // completed and every caller reads healthy — fire-and-forget, - // nothing retries it, nothing upstream is told. After the next - // restart the run is invisible to the Runs surfaces, and the - // approvals sweeps read exactly that hole: `inspectStranded- - // Requests` (#3456) treats "no suspension + no terminal row" - // as a STRANDED request, so a run that actually completed is - // reported as lost, and `releasePendingForTerminalRuns` - // (#4469) treats "no terminal row" as still-alive, so a - // finished run's leftover pending approvals are never - // auto-released. A write that claims to persist did not, while - // the system keeps looking healthy — #4632's judgment question - // answers YES. `check:durability-log-level` cannot see it - // (`SuspendedRunStore.recordTerminal` is not in its callee - // vocabulary), so the level is pinned by a test instead. - // - // THIRD argument per `error(message, error?, meta?)`; the - // `Error` slot stays empty on purpose (#5575). - this.logger.error( - `[Automation] run-history persist failed for terminal run '${entry.id}' of flow ` + - `'${entry.flowName}' — the run finished and reads healthy everywhere, but its history ` + - `row never landed and nothing retries the write, so after the next restart this run is ` + - `invisible to the Runs surfaces and the approvals sweeps read it as never-finished. ` + - `Fix the store failure in this record's meta.`, - undefined, - describeThrownForLog(err), - ); - }); - } - return entry; + // The body lives in ./run-history.ts (#25222). + const readStore = () => this.store; + const host: ThisParameterType = { + executionLogs: this.executionLogs, + maxLogSize: this.maxLogSize, + runSummaryLog: this.runSummaryLog, + logger: this.logger, + get store() { return readStore(); }, + compactStepsForHistory: (...a) => this.compactStepsForHistory(...a), + consumedSuspensions: this.consumedSuspensions, + }; + return recordLog.call(host, entry, context, consumedSuspension); } /** @@ -10259,60 +9728,12 @@ export class AutomationEngine implements IAutomationService { refusalMessage?: string; context?: AutomationContext; }): AutomationResult { - // The run's declared outputs, collected exactly as the three completion - // exits collect them. A refusal is terminal, and the nodes BEFORE the - // refusing `end` really ran — withholding what they produced would make - // the caller's answer depend on how the run ended rather than on what - // the flow declared. - const output: Record = {}; - if (args.flow.variables) { - for (const v of args.flow.variables) { - if (v.isOutput) output[v.name] = args.variables.get(v.name); - } - } - let logged: ExecutionLogEntry | undefined; - try { - logged = this.recordLog({ - id: args.runId, - flowName: args.flowName, - flowVersion: args.flowVersion, - status: 'refused', - refusalMessage: args.refusalMessage, - startedAt: args.startedAt, - completedAt: new Date().toISOString(), - durationMs: args.durationMs, - trigger: buildRunTrigger(args.context), - steps: args.steps, - output, - }, args.context); - } catch (bookkeeping) { - // #4632 verdict: DURABILITY, so `error` — the same judgement and - // the same consequence as the completion sites', with one word - // changed: the caller is told the truthful thing (the run refused), - // which is exactly what makes the rest invisible from outside. - // Said ONCE per run. THIRD argument per `error(message, error?, - // meta?)`; the `Error` slot stays empty on purpose (#5575). - this.logger.error( - `[Automation] run '${args.runId}' of flow '${args.flowName}' REFUSED (an 'end' node with ` + - `outcome: 'refused', or a node whose child run refused) but its run-history bookkeeping ` + - `threw, so its terminal history row ` + - `never landed — nothing retries it, the caller is told the run refused, and after the next ` + - `restart this run is invisible to the Runs surfaces while the approvals sweeps read it as ` + - `never-finished. The run itself is TERMINAL and must NOT be re-run, retried or resumed. ` + - `Fix the history failure in this record's meta.`, - undefined, - describeThrownForLog(bookkeeping), - ); - } - return { - success: true, - status: 'refused', - refusalMessage: args.refusalMessage, - flowLabel: args.flow.label, - output, - durationMs: args.durationMs, - summary: logged?.summary ?? summarizeRun(args.steps), + // The body lives in ./run-history.ts (#25222). + const host: ThisParameterType = { + recordLog: (...a) => this.recordLog(...a), + logger: this.logger, }; + return finishRefusedRun.call(host, args); } /** 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 7b7cc524fb8..0411afd7f48 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']; + const ENGINE_SOURCES = ['./engine.ts', './flow-registration-checks.ts', './plugin-extension-api.ts', './dag-traversal.ts', './run-history.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-history.ts b/packages/services/service-automation/src/run-history.ts new file mode 100644 index 00000000000..8d6c532fe84 --- /dev/null +++ b/packages/services/service-automation/src/run-history.ts @@ -0,0 +1,781 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The run history of {@link AutomationEngine}: how a run's log entry is + * recorded and a refused run is finished, and how runs, one run and the + * suspended runs are read back. + * + * [#25222] Slice 5 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 RunHistoryHost} 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 { ExecutionStatus, FlowParsed } from '@objectstack/spec/automation'; +import type { AutomationContext, AutomationResult, RunListResult } from '@objectstack/spec/contracts'; +import { summarizeRun, formatRunSummaryLine } from './run-summary.js'; +import { describeThrownForLog } from './thrown-cause-diagnostics.js'; +// The values below (`buildRunTrigger` and `isTerminalRunStatus`) 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, + isTerminalRunStatus, + type AutomationEngine, + type ExecutionLogEntry, + type RunRecord, + 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`. `store` goes through a getter, because `setSuspendedRunStore` + * replaces it: each read in the body reads the engine's current store, as it + * did through `this`. + */ +export interface RunHistoryHost { + readonly listRunsPage: AutomationEngine['listRunsPage']; + readonly executionLogs: AutomationEngine['executionLogs']; + readonly store: AutomationEngine['store']; + readonly suspendedRunToLogEntry: AutomationEngine['suspendedRunToLogEntry']; + readonly logger: AutomationEngine['logger']; + readonly runRecordToLogEntry: AutomationEngine['runRecordToLogEntry']; + readonly suspendedRuns: AutomationEngine['suspendedRuns']; + readonly cacheOnlySuspensions: AutomationEngine['cacheOnlySuspensions']; + readonly evictConsumedSuspension: AutomationEngine['evictConsumedSuspension']; + readonly maxLogSize: AutomationEngine['maxLogSize']; + readonly runSummaryLog: AutomationEngine['runSummaryLog']; + readonly compactStepsForHistory: AutomationEngine['compactStepsForHistory']; + readonly consumedSuspensions: AutomationEngine['consumedSuspensions']; + readonly recordLog: AutomationEngine['recordLog']; +} + +/** The body of {@link AutomationEngine.listRuns}, moved unedited; that method's docblock is its contract. */ +export async function listRuns( + this: Pick, + flowName: string, + options?: { limit?: number; status?: ExecutionStatus }, +): Promise { + // [#19543] ONE implementation, two projections — `listRunsPage` is the + // whole method and this is its `runs` half. ⛔ Never re-derive the + // listing here: a second copy of the merge/filter/sort would be the + // fork the route-ownership rule refuses, and it is the half that would + // rot, because the page method is the one the REST door calls. + return (await this.listRunsPage(flowName, options)).runs; +} + +/** The body of {@link AutomationEngine.listRunsPage}, moved unedited; that method's docblock is its contract. */ +export async function listRunsPage( + this: Pick, + flowName: string, + options?: { limit?: number; status?: ExecutionStatus }, +): Promise { + const limit = options?.limit ?? 20; + const inMem = this.executionLogs.filter(l => l.flowName === flowName); + + // [#8050] Durable PAUSED rows — the arm this merge was missing. + // + // `sys_automation_run` holds TWO disjoint row families: the terminal + // history rows `recordTerminal` writes (id `run_` + runId, status + // completed/failed) and the LIVE suspension rows `save` writes (id = + // the raw runId, status `paused`). The history arm below reads the + // first family; nothing here read the second. Before a restart that is + // invisible, because a paused run is still in `executionLogs` — so + // every in-process test of this method passes. After a restart the ring + // is empty and the paused rows had no reader at all, which is the + // defect: an operator who restarts the process can enumerate what + // FINISHED and what is IN FLIGHT vanishes — the strictly more urgent + // half. It also structurally emptied #7359's just-enforced + // `?status=paused`: with no producer of a `paused` entry after a + // restart, that filter could never match a row, and the one query + // reached for here always answered "nothing pending". + // + // Read-side only, by construction: this rehydrates the row the suspend + // path already writes. No column, prefix or lifecycle changes — the + // paused row is NOT reshaped into a history row, because the two have + // different lifetimes (a paused row is live resumable state, deleted on + // completion and exempt from the age sweep; a history row is a + // tombstone). Unifying them would trade an observability gap for a + // persistence-semantics change. + // + // Skipped when the caller filters for a status a paused row can never + // have: `suspendedRunToLogEntry` always yields `paused`, and a run that + // has since finished is answered by the fresher history/ring entry that + // outranks it in the merge below — so the arm cannot change the result + // of `?status=failed`, only its cost. `store.list()` is a table scan of + // every paused row in the deployment (see its own contract), so not + // paying it on the monitoring queries is worth the one-line guard. + const wantsPaused = options?.status === undefined || options.status === 'paused'; + let durablePaused: ExecutionLogEntry[] = []; + if (this.store && wantsPaused) { + try { + const rows = await this.store.list(); + durablePaused = rows + .filter(r => r.flowName === flowName) + .map(r => this.suspendedRunToLogEntry(r)); + } catch (err) { + // #6499 — driver text to the structured slot; `warn(message, + // meta?)`, meta SECOND (no `Error` slot on `warn`). + // + // #4632 verdict: FUNCTIONAL — `warn`, for the same reason as + // the history arm below and `listSuspendedRunsDurable`. Nothing + // claimed-persisted failed to land: the paused rows are intact + // and still resumable by id (`resume` reads them through + // `loadSuspendedRun`, a different door that is unaffected by + // this failure). What degrades is this observability read, back + // to exactly the pre-#8050 answer. + this.logger.warn( + `[Automation] paused-run read failed for '${flowName}' — the Runs listing DEGRADES to the ` + + `in-memory ring buffer plus terminal history, so runs parked by a previous process are ` + + `missing and '?status=paused' can report an empty result for a flow that has runs ` + + `waiting. The rows themselves are untouched and still resumable by id. Fix the store ` + + `failure in this record's meta.`, + describeThrownForLog(err), + ); + } + } + + // Merge durable run history so the "Runs" view survives a restart and + // ring-buffer eviction. In-memory entries are the freshest (they carry + // full step detail); durable rows backfill runs the process no longer + // holds. Best-effort: a history-read failure degrades to in-memory only. + let durable: ExecutionLogEntry[] = []; + if (this.store?.listHistory) { + try { + // [#19543] `limit + 1`, not `limit` — the over-read that makes + // `hasMore` answerable at all. Asking for exactly `limit` makes a + // saturated window and a complete one identical; one extra row + // tells them apart, and `.slice(0, limit)` below drops it again + // so nothing on the wire widens. + const rows = await this.store.listHistory(flowName, limit + 1); + durable = rows.map(r => this.runRecordToLogEntry(r)); + } catch (err) { + // #6499 — the datasource driver's text to the structured slot; + // see `forgetSuspendedRun`'s catch above for the full mechanism + // (#6299). `warn(message, meta?)` — meta is the SECOND + // argument; `warn` has no `Error` slot. + // + // #4632 verdict: FUNCTIONAL — stays `warn`, the same reasoning + // as `listSuspendedRunsDurable` below: nothing + // claimed-persisted failed to land — the history rows are + // intact — and this read feeds only the observability Runs + // view, which degrades to the in-memory ring buffer. The + // record says the shortfall out loud (#5186's invented-answer + // shape; its propagation remedy is a return-contract change + // outside #6499). + this.logger.warn( + `[Automation] run-history read failed for '${flowName}' — the Runs listing DEGRADES to the ` + + `in-memory ring buffer alone, so terminal runs from before the last restart (or evicted ` + + `from the buffer) are missing and the caller cannot tell a short list from a complete ` + + `one. The rows themselves are untouched. Fix the store failure in this record's meta.`, + describeThrownForLog(err), + ); + } + } + // Dedupe by run id, weakest source first — the same run legitimately + // appears in more than one of these (#8050): + // + // 1. durable PAUSED — the run parked, and the row is still there. + // 2. durable HISTORY — the run reached a terminal state. + // 3. in-memory ring — this process executed it. + // + // The order is a precedence claim, not an accident. Paused loses to + // both because it is the only one that can be STALE while the others + // cannot: `forgetSuspendedRun` deletes the paused row on completion, + // but that delete is best-effort (a store outage swallows it), so a + // finished run can leave its paused row behind. A terminal row or a + // terminal ring entry for the same id is therefore strictly later + // evidence, and letting the paused row win would report a completed run + // as still waiting — the exact defect `run-history.test.ts`'s "latest + // entry wins" block pins for `getRun`. There is no symmetric hazard: + // within a process the ring is written in the same breath as the paused + // row (`persistSuspendedRun` then `recordLog`, and again on re-suspend), + // so it is never the older of the two; across a restart the ring is + // empty and cannot mask anything. + const byId = new Map(); + for (const e of durablePaused) byId.set(e.id, e); + for (const e of durable) byId.set(e.id, e); + for (const e of inMem) byId.set(e.id, e); // freshest wins + + // [#7359] The wire has always declared `?status=`, but it reached + // neither the contract nor this method, so the filter was dropped at + // the HTTP boundary and every run came back as if it matched — a + // monitoring caller paging for failures read the first `limit` runs of + // any status and concluded those were the failures. + // + // Filtering HERE, after the merge, is load-bearing in two ways: + // + // 1. It covers BOTH stores at once — the in-memory ring buffer and the + // durable rows. Narrowing only the buffer would answer "no + // failures" for a flow whose failures are all in durable history + // (i.e. after any restart, which is exactly when someone asks); + // narrowing only the durable rows would hide the live ones. Half a + // filter is the same class of confident wrong answer as none. + // 2. It reads each run's RESOLVED status. `executionLogs` holds more + // than one entry per run id — a run that pauses and later finishes + // appends 'paused' and then its terminal entry — and the merge above + // is what collapses them to the freshest. Filtering the two arms + // BEFORE that collapse would drop the terminal entry for + // `?status=paused` and let the stale 'paused' one survive, so every + // approval / screen / wait run that had since completed would report + // itself as still paused. That is the exact defect + // `run-history.test.ts`'s "latest entry wins" block pins for + // `getRun`, re-introduced one method over. + // + // The durable arm's window is still the store's newest `limit` rows — + // `listHistory(flowName, limit)` has no status slot, so the filter is + // applied to what comes back rather than pushed down. A status filter + // therefore narrows WITHIN that window and can return fewer than + // `limit` matches while older ones exist. That is the merge's + // pre-existing shape (durable was already capped at `limit` before the + // sort-and-slice) and closing it is a store-contract change, not this + // card's. What it never does is return a run of another status. + const status = options?.status; + const merged = status === undefined + ? [...byId.values()] + : [...byId.values()].filter(e => e.status === status); + const ordered = merged + .sort((a, b) => (b.startedAt ?? '').localeCompare(a.startedAt ?? '')); + // [#19543] The comparison is against the ORDERED, FILTERED set, not + // against what any single source returned: a row can reach `ordered` + // from the ring or the paused arm without the history arm knowing, and + // a `?status=` filter can drop the over-read row specifically. Reading + // the overflow here — after every arm has contributed and after the + // filter has run — is what makes the answer true of the response + // actually being sent. + return { runs: ordered.slice(0, limit), hasMore: ordered.length > limit }; +} + +/** The body of {@link AutomationEngine.runRecordToLogEntry}, moved unedited; that method's docblock is its contract. */ +export function runRecordToLogEntry(r: RunRecord): ExecutionLogEntry { + return { + id: r.runId, + flowName: r.flowName, + flowVersion: r.flowVersion, + // [#15223] All four {@link TERMINAL_RUN_STATUSES} members, each a + // valid `ExecutionLog` status — the schema has declared the whole + // vocabulary since before this row existed, and it was the + // persistence layer, not the contract, that reported only two. + status: r.status, + startedAt: r.startedAt, + completedAt: r.finishedAt, + durationMs: r.durationMs, + // #7533 — the persisted trigger attribution, so a run rehydrated + // after a restart still says what fired it and (for a record + // change) which record. `type` falls back to `''` ONLY for rows + // written before those columns existed: the field is required by + // `ExecutionLogSchema`, and `''` is the same "not recorded" this + // method already returned for every row. It is a legacy-row + // default, not an alias — a row written today always carries a + // real type. + trigger: { + type: r.triggerType ?? '', + userId: r.userId ?? undefined, + object: r.triggerObject, + recordId: r.triggerRecordId, + }, + steps: r.steps ?? [], + error: r.error, + // [#15788] The persisted refusal text, so the run a caller opens + // after a restart still says WHY it refused. Without this line the + // status survives the round-trip and its reason does not, which is + // the worst of the two halves to lose: `refused` with no message is + // indistinguishable from a refusal nobody authored. + refusalMessage: r.refusalMessage, + // #4354 — the PERSISTED summary, never re-folded from `r.steps`: + // those are compacted (200 max), so recomputing here would report a + // 5000-row sweep as having acted on a couple of hundred. + summary: r.summary, + }; +} + +/** The body of {@link AutomationEngine.suspendedRunToLogEntry}, moved unedited; that method's docblock is its contract. */ +export function suspendedRunToLogEntry(run: SuspendedRun): ExecutionLogEntry { + return { + id: run.runId, + flowName: run.flowName, + flowVersion: run.flowVersion, + status: 'paused', + startedAt: run.startedAt, + trigger: buildRunTrigger(run.context), + steps: run.steps ?? [], + variables: run.variables ?? {}, + }; +} + +/** The body of {@link AutomationEngine.getRun}, moved unedited; that method's docblock is its contract. */ +export async function getRun(this: Pick, runId: string): Promise { + // LAST entry wins, not the first: a run that pauses and later finishes + // records TWO entries under the same run id ('paused', then + // 'completed'/'failed'/'cancelled'). Scanning forwards returned the stale + // 'paused' one, so every suspend-then-finish run — i.e. every approval, + // screen and wait flow — reported itself as still paused forever, both on + // the Runs surface and to the approvals dead-run sweep (#3456). + let inMem: ExecutionLogEntry | undefined; + for (let i = this.executionLogs.length - 1; i >= 0; i--) { + if (this.executionLogs[i].id === runId) { inMem = this.executionLogs[i]; break; } + } + if (inMem) return inMem; + // Durable fallback: after a restart (or ring-buffer eviction) the run's + // terminal history row still answers "what happened, at which node?". + // Best-effort — a store failure degrades to "not found", never throws. + if (this.store?.loadTerminal) { + try { + const rec = await this.store.loadTerminal(runId); + if (rec) return this.runRecordToLogEntry(rec); + } catch (err) { + // #6499 — driver text to the structured slot; see + // `forgetSuspendedRun`'s catch above for the full mechanism + // (#6299). + // + // #4632 verdict: FUNCTIONAL — stays `warn`. Nothing + // claimed-persisted failed to land: the terminal row, if one + // exists, is intact — this read degrades to `null`. What IS + // wrong is #5186's shape: `null` is also this method's honest + // "no such run", so a caller cannot tell an unreadable store + // from a run that never ran (plugin-approvals' + // `inspectStrandedRequests` counts a THROWN getRun as + // `undetermined` — a distinction this swallow denies it). The + // remedy is propagation, a return-contract change outside + // #6499's scope, so the record says the degradation out loud. + this.logger.warn( + `[Automation] durable run lookup failed for '${runId}' — this read DEGRADES to null, so ` + + `the caller sees exactly what it would see if the run had never run and cannot tell ` + + `the two apart. The terminal row, if one exists, is untouched. Fix the store failure ` + + `in this record's meta.`, + describeThrownForLog(err), + ); + } + } + // [#8050] …and the PAUSED fallback, so run-detail and `listRuns` answer + // out of one story. The card measured both surfaces failing together + // after a restart — list returning zero rows AND this method 404ing — + // and fixing only the list would have swapped a visible gap for an + // inconsistency between two reads of the same run. + // + // AFTER the terminal probe, matching the merge order in `listRuns`: a + // paused row can outlive the run it describes (the delete on completion + // is best-effort), so a terminal row for the same id is later evidence + // and must win. Trying this first would resurrect #3456's "paused + // forever" for any run whose cleanup delete was lost. + // + // This does NOT make a nonexistent run findable: `store.load` answers + // `null` for an unknown id exactly as `loadTerminal` does, so the route + // above still returns its 404 `RESOURCE_NOT_FOUND` envelope for one. + if (this.store) { + try { + const suspended = await this.store.load(runId); + if (suspended) return this.suspendedRunToLogEntry(suspended); + } catch (err) { + // #6499 / #4632: same verdict as the terminal probe above — + // FUNCTIONAL, so `warn`. The suspension row is intact and the + // run stays parked and resumable; what degrades is this read. + this.logger.warn( + `[Automation] durable paused-run lookup failed for '${runId}' — this read DEGRADES to null, ` + + `so a run that is parked and resumable reports as if it had never run, and the caller ` + + `cannot tell the two apart. The suspension row is untouched. Fix the store failure in ` + + `this record's meta.`, + describeThrownForLog(err), + ); + } + } + return null; +} + +/** The body of {@link AutomationEngine.listSuspendedRuns}, moved unedited; that method's docblock is its contract. */ +export function listSuspendedRuns(this: Pick): Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }> { + return [...this.suspendedRuns.values()].map(r => ({ + runId: r.runId, + flowName: r.flowName, + nodeId: r.nodeId, + correlation: r.correlation, + })); +} + +/** The body of {@link AutomationEngine.readSuspendedRuns}, moved unedited; that method's docblock is its contract. */ +export async function readSuspendedRuns( + this: Pick, + onEnumerationFailure: 'degrade' | 'throw', + flowNames?: ReadonlySet, +): Promise> { + const byId = new Map(); + // [#15832] Did the ENUMERATION answer? The reconcile below is allowed + // only when it did — see the merge comment for why a failed listing is + // silence rather than evidence. + let enumerated = false; + if (this.store) { + try { + const stored = flowNames === undefined + ? await this.store.list() + : this.store.listByFlow + ? await this.store.listByFlow([...flowNames]) + : (await this.store.list()).filter((r) => flowNames.has(r.flowName)); + for (const r of stored) { + byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation }); + } + enumerated = true; + } catch (err) { + // A write-deciding caller is told, and decides nothing: see + // {@link readSuspendedRuns}. Everything below is the LISTING's + // posture. + if (onEnumerationFailure === 'throw') throw err; + // #6299 — driver text to the structured slot, message one line, + // same as the two seams above. The SLOT differs: the `Logger` + // contract declares `warn(message, meta?)`, so `meta` is the + // SECOND argument here — `warn` has no `Error` slot, and a + // `meta` passed third to it is silently ignored. + // + // #4632 verdict: FUNCTIONAL — the level deliberately STAYS + // `warn`, and this is the one of #6299's three sites where that + // is the answer. Nothing the system claims to have persisted + // failed to land: the durable rows are intact, still resumable + // by id, and the next boot's `rearmSuspendedWaitTimers` still + // re-arms them off its OWN `store.list()` — which builtin/ + // wait-node.ts grades `error` precisely because THAT failure + // breaks the promise to resume them. This one breaks no promise, + // so #4632's judgment question ("does something the system + // claims is persisted fail to land while it keeps looking + // healthy?") answers NO. + // + // What IS wrong here is the shape #5186 owns: an answer INVENTED + // for a read that failed — a silently SHORT list. That rule's + // remedy is propagation (rethrow, or a discriminated result), + // which changes this method's return contract and is outside + // #6299's scope, so the record has to say the shortfall out loud + // instead. Its scan roots (`packages/metadata`, + // `metadata-protocol`, `objectql`) do not reach this package, so + // `check:durability-log-level` reports neither rule here. + // Reachability is also the weakest of the three: this method is + // not on the `AutomationService` spec contract (only the + // synchronous `listSuspendedRuns` is); the in-repo caller that + // would ADMIT a write on what this list lacks — the §7.3 + // disable guard — takes the `'throw'` posture above and never + // reaches this line, and plugin-approvals' restored- + // continuation proof refuses on an absent entry, so a short + // list costs it a retry. Raising it to `error` would alarm for + // the duration of an outage on a read no caller acts on + // wrongly — #4632's mirror-image misuse, the trap #6230 + // avoided. + this.logger.warn( + `[automation] the durable suspended-run store could not be listed — this listing DEGRADES to the ` + + `in-memory cache alone, so every run parked by a previous process is missing from the result ` + + `and the caller cannot tell a short list from a complete one (after a restart the cache is ` + + `empty, so this answers []). The runs themselves are untouched — still stored, still resumable ` + + `by id. Fix the store failure in this record's meta.`, + describeThrownForLog(err), + ); + } + } + // [#14332] The DURABLE row wins a collision. The comment this replaces + // said the opposite — "In-memory entries win — they are the freshest + // copy" — which is true of exactly one deployment shape, a single + // process. Put several replicas over one store and this map is a + // per-replica snapshot of the node a run was parked at THE LAST TIME + // THIS REPLICA TOUCHED IT, with no invalidation channel to it at all + // (the mechanism is in {@link loadSuspendedRunStrict}), so preferring it + // reported a run at a node it had already left. + // + // Map entries the durable list does not carry are still appended, and + // that is NOT the strict loader's rule being softened: a LIST is not a + // per-id answer. `store.list()` is a capped, best-effort enumeration + // (`ObjectStoreSuspendedRunStore` reads at most 1000 `paused` rows) and + // this line is also reached on the DEGRADED path above, where the + // enumeration failed outright and `byId` is empty. So "absent from the + // list" is not the evidence "the store answered and has no row" is, + // which is what {@link loadSuspendedRunStrict} rests on when it lets + // only {@link cacheOnlySuspensions} answer out of the map. Applying that + // qualifier here would let a truncated or failed enumeration silently + // drop live runs from an operability listing. + // + // [#15832] What that reasoning leaves open is a run this process parked + // and ANOTHER replica has since consumed: absent from the durable list + // because it is finished, appended here, and published as suspended by + // a listing that also backs the cache-only one. The paragraph above is + // right that list-absence is not evidence — so this asks for the + // evidence instead. `store.load` is the same definitive per-id read + // {@link loadSuspendedRunStrict} rests on, and it is bought only for the + // entries that look suspicious: a healthy process, whose map entries all + // appear in the durable list, buys none. A read that THROWS leaves the + // entry standing (unknown is not gone), and a store that could not be + // enumerated at all is not probed row by row — an outage would answer + // for every live run in the process. + // + // [#20725] A scoped read takes only the named flows' entries — the hot + // map holds every run this process parked, uncapped, so it answers the + // scoped question completely for this process — and probes only those. + for (const r of [...this.suspendedRuns.values()]) { + if (flowNames !== undefined && !flowNames.has(r.flowName)) continue; + if (byId.has(r.runId)) continue; + if (enumerated && !this.cacheOnlySuspensions.has(r.runId)) { + let stored: SuspendedRun | null; + try { + stored = await this.store!.load(r.runId); + } catch { + // Unknown, not gone — keep the entry and publish it, exactly + // as this method did before the reconcile existed. + stored = r; + } + if (stored === null) { + this.evictConsumedSuspension(r.runId); + continue; + } + } + byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation }); + } + return [...byId.values()]; +} + +/** The body of {@link AutomationEngine.recordLog}, moved unedited; that method's docblock is its contract. */ +export function recordLog( + this: Pick, + entry: ExecutionLogEntry, + context?: AutomationContext, + consumedSuspension?: SuspendedRun, +): ExecutionLogEntry { + // #4354 — fold the run's outcome BEFORE anything downstream trims the + // step log. History compaction keeps 200 steps; the summary must count + // all 5000, or a long sweep's `acted` would shrink with its step log and + // the broken-sweep detector would read a bounded artefact as truth. + entry.summary = summarizeRun(entry.steps); + + this.executionLogs.push(entry); + // Evict oldest logs when exceeding max size + if (this.executionLogs.length > this.maxLogSize) { + this.executionLogs.splice(0, this.executionLogs.length - this.maxLogSize); + } + // Durable run history (observability): mirror every TERMINAL run to the + // store so "did it run / fail, and why?" survives a restart and the + // in-memory ring-buffer eviction. Best-effort + fire-and-forget: a + // history write must NEVER block or break the run that produced it. + // [#15223] ONE vocabulary, not a fourth copy of the list: this + // predicate decides both WHETHER a history row is written and WHAT its + // `status` says. Keeping the narrowed value in a `const` is what makes + // the record below type-check without a cast — and a cast is precisely + // how the fold this card is about survived four members for two. + const terminalStatus = isTerminalRunStatus(entry.status) ? entry.status : undefined; + const terminal = terminalStatus !== undefined; + + // The MVP of #4354, and the half that needs no console: one structured + // line per terminal run. `selected=30 acted=0` in a log file is the + // difference between an invisible failure and a greppable one. Only + // terminal runs — a `paused` run has not finished doing its work yet. + if (terminal && this.runSummaryLog !== 'off') { + const line = formatRunSummaryLine( + { + flowName: entry.flowName, + runId: entry.id, + status: entry.status, + durationMs: entry.durationMs, + }, + entry.summary, + ); + const meta = { + flow: entry.flowName, + runId: entry.id, + status: entry.status, + durationMs: entry.durationMs, + selected: entry.summary.selected, + acted: entry.summary.acted, + skipped: entry.summary.skipped, + unmeasured: entry.summary.unmeasured, + // #14456 — the structured half of the `failed=` token the line + // above now carries, so a log pipeline that reads `meta` and + // never parses the message sees the same fact. + failed: entry.summary.failed, + gates: entry.summary.gates, + }; + if (this.runSummaryLog === 'debug') this.logger.debug(line, meta); + else this.logger.info(line, meta); + } + + if (terminalStatus && this.store?.recordTerminal) { + const lastStep = entry.steps[entry.steps.length - 1]; + const record: RunRecord = { + runId: entry.id, + flowName: entry.flowName, + flowVersion: entry.flowVersion, + // [#15223] The status the run actually reached. This used to be + // `entry.status === 'completed' ? 'completed' : 'failed'` — a + // fold applied at WRITE time, so a cancelled or timed-out run's + // distinction was not merely unshown, it was never stored and + // could not be recovered afterwards. ⛔ Never re-introduce a + // conditional here: whatever the terminal predicate above + // admits is what the row must carry. + status: terminalStatus, + startedAt: entry.startedAt, + finishedAt: entry.completedAt, + durationMs: entry.durationMs, + error: entry.error, + // [#15788] The rendered refusal, carried onto the durable row + // beside `status: 'refused'`. Set by exactly one producer + // ({@link AutomationEngine.finishRefusedRun}) and `undefined` + // on every other terminal record, so the store writes an + // explicit NULL there — an upsert must CLEAR it, or a run id + // reused by a restore would keep a refusal it no longer has. + refusalMessage: entry.refusalMessage, + userId: entry.trigger?.userId, + // [#10101] The two organization-attribution inputs, from the + // run context (see the `recordLog` doc): the acting tenant is + // the ruled FALLBACK, the trigger-record snapshot is what the + // store resolves the SUBJECT organization from — the same + // snapshot the paused row resolves from, so the two rows of + // one run agree by construction. + organizationId: context?.tenantId ?? null, + triggerRecord: context?.record, + // #7533 — the rest of the trigger block, not just its userId. + // The information exists at this exact point (the in-memory log + // entry one line up carries it); it was simply not copied onto + // the record handed to the store, which is where the durable + // history lost the answer to "what fired this run?". + triggerType: entry.trigger?.type || undefined, + triggerObject: entry.trigger?.object, + triggerRecordId: entry.trigger?.recordId, + // [#13937] On a stranded run's record the PAUSE node, not the + // node that threw: the object store rebuilds the snapshot's + // node from this one column, and a restore from the row used + // to re-arm the run at the failed node — whose next resume + // then SKIPPED it and reported the run completed. The throwing + // node is the last entry of `steps` below, and `error`. + nodeId: consumedSuspension?.nodeId ?? lastStep?.nodeId, + steps: this.compactStepsForHistory(entry.steps), + summary: entry.summary, + // [#13909] Present only on the resume-consumed-then-failed + // path. On every other terminal record it is `undefined`, and + // the store writes explicit NULLs for it — so a run that is + // restored and then finishes CLEARS its own snapshot instead of + // leaving a stale one an operator could restore a second time. + consumedSuspension, + }; + // [#13937] The journal entry this record describes, when it is a + // strand's: `journalConsumedSuspension` ran just above in the + // same arm, so the entry under this run id is the one built from + // `consumedSuspension` — checked by identity, because a later + // strand of the same run replaces the entry and must not inherit + // this write's outcome. + const journal = consumedSuspension ? this.consumedSuspensions.get(entry.id) : undefined; + const write = this.store.recordTerminal(record); + if (journal && journal.run === consumedSuspension) { + void write.then( + () => { journal.persisted = 'landed'; }, + () => { journal.persisted = 'failed'; }, + ); + } + void write.catch((err) => { + // #6499 — driver text to the structured slot; see + // `forgetSuspendedRun`'s catch above for the full mechanism + // (#6299). + // + // #4632 verdict: DURABILITY — RAISED from `warn` to `error`. + // This is the WRITE half of the run-history claim whose read + // halves (`listRuns` / `getRun` above) stay `warn`: a TERMINAL + // run's history row failed to land while the run itself + // completed and every caller reads healthy — fire-and-forget, + // nothing retries it, nothing upstream is told. After the next + // restart the run is invisible to the Runs surfaces, and the + // approvals sweeps read exactly that hole: `inspectStranded- + // Requests` (#3456) treats "no suspension + no terminal row" + // as a STRANDED request, so a run that actually completed is + // reported as lost, and `releasePendingForTerminalRuns` + // (#4469) treats "no terminal row" as still-alive, so a + // finished run's leftover pending approvals are never + // auto-released. A write that claims to persist did not, while + // the system keeps looking healthy — #4632's judgment question + // answers YES. `check:durability-log-level` cannot see it + // (`SuspendedRunStore.recordTerminal` is not in its callee + // vocabulary), so the level is pinned by a test instead. + // + // THIRD argument per `error(message, error?, meta?)`; the + // `Error` slot stays empty on purpose (#5575). + this.logger.error( + `[Automation] run-history persist failed for terminal run '${entry.id}' of flow ` + + `'${entry.flowName}' — the run finished and reads healthy everywhere, but its history ` + + `row never landed and nothing retries the write, so after the next restart this run is ` + + `invisible to the Runs surfaces and the approvals sweeps read it as never-finished. ` + + `Fix the store failure in this record's meta.`, + undefined, + describeThrownForLog(err), + ); + }); + } + return entry; +} + +/** The body of {@link AutomationEngine.finishRefusedRun}, moved unedited; that method's docblock is its contract. */ +export function finishRefusedRun(this: Pick, args: { + runId: string; + flowName: string; + flowVersion?: number; + startedAt: string; + durationMs: number; + steps: StepLogEntry[]; + /** Read for its `isOutput` variable declarations — see `output` below. */ + flow: FlowParsed; + variables: Map; + refusalMessage?: string; + context?: AutomationContext; +}): AutomationResult { + // The run's declared outputs, collected exactly as the three completion + // exits collect them. A refusal is terminal, and the nodes BEFORE the + // refusing `end` really ran — withholding what they produced would make + // the caller's answer depend on how the run ended rather than on what + // the flow declared. + const output: Record = {}; + if (args.flow.variables) { + for (const v of args.flow.variables) { + if (v.isOutput) output[v.name] = args.variables.get(v.name); + } + } + let logged: ExecutionLogEntry | undefined; + try { + logged = this.recordLog({ + id: args.runId, + flowName: args.flowName, + flowVersion: args.flowVersion, + status: 'refused', + refusalMessage: args.refusalMessage, + startedAt: args.startedAt, + completedAt: new Date().toISOString(), + durationMs: args.durationMs, + trigger: buildRunTrigger(args.context), + steps: args.steps, + output, + }, args.context); + } catch (bookkeeping) { + // #4632 verdict: DURABILITY, so `error` — the same judgement and + // the same consequence as the completion sites', with one word + // changed: the caller is told the truthful thing (the run refused), + // which is exactly what makes the rest invisible from outside. + // Said ONCE per run. THIRD argument per `error(message, error?, + // meta?)`; the `Error` slot stays empty on purpose (#5575). + this.logger.error( + `[Automation] run '${args.runId}' of flow '${args.flowName}' REFUSED (an 'end' node with ` + + `outcome: 'refused', or a node whose child run refused) but its run-history bookkeeping ` + + `threw, so its terminal history row ` + + `never landed — nothing retries it, the caller is told the run refused, and after the next ` + + `restart this run is invisible to the Runs surfaces while the approvals sweeps read it as ` + + `never-finished. The run itself is TERMINAL and must NOT be re-run, retried or resumed. ` + + `Fix the history failure in this record's meta.`, + undefined, + describeThrownForLog(bookkeeping), + ); + } + return { + success: true, + status: 'refused', + refusalMessage: args.refusalMessage, + flowLabel: args.flow.label, + output, + durationMs: args.durationMs, + summary: logged?.summary ?? summarizeRun(args.steps), + }; +}