From d3347f07b526583da918c89f31c23443b5f0616b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 24 Oct 2026 02:50:04 +0000 Subject: [PATCH] refactor(service-automation): move 42 engine bodies into five seam modules Move-only. The bodies of 42 AutomationEngine methods move unedited into five new modules: dispatch-claims.ts (claiming, settling and reading a dispatch, minting a run id, releasing a suspension; 7 bodies), flow-definitions.ts (canonicalising, unregistering, refusing and rolling back a flow, its runtime states and trigger bindings, and its function, connector and credential surfaces; 12), subflow-guards.ts (finding and refusing what reaches a disabled subflow; 11), consumed-suspension-restore.ts (resolving, restoring, peeking at and inspecting a consumed suspension; 6) and node-text-rendering.ts (the node-type vocabulary checks, a failed node's error variable, and refusal and flow-text rendering; 6). Each method keeps its name, docblock, signature and position in the class and calls its body with a host literal holding exactly the members the body reads; six bodies 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. Six fields a method reassigns go through getters. The four warned-once flags a moved body sets go through get/set pairs, so each write lands on the engine. claimAdvance, validateNodeInputSchemas, refuseInvalidScreenInput and buildSubflowResumeSignal stay in the class, because each reads a name documented as private to engine.ts. pullConnectorSource stays too: a second top-level function of that name would make the bundler rename the public pullConnectorSource export. Seven helpers the moved bodies read gain export in place; index.ts re-exports none of them. The imports only moved bodies read move with them. The ENGINE_SOURCES grep pins read the new modules too. Claude-Session: https://claude.ai/code/session_01Mnj4nkWuhJ18xEdZvb2cjE Co-Authored-By: Claude --- .../src/consumed-suspension-restore.ts | 243 ++++ .../service-automation/src/dispatch-claims.ts | 203 +++ .../services/service-automation/src/engine.ts | 1136 ++++++----------- .../src/flow-activation-ledger.test.ts | 2 +- .../src/flow-definitions.ts | 370 ++++++ .../src/node-text-rendering.ts | 168 +++ .../service-automation/src/subflow-guards.ts | 284 +++++ 7 files changed, 1634 insertions(+), 772 deletions(-) create mode 100644 packages/services/service-automation/src/consumed-suspension-restore.ts create mode 100644 packages/services/service-automation/src/dispatch-claims.ts create mode 100644 packages/services/service-automation/src/flow-definitions.ts create mode 100644 packages/services/service-automation/src/node-text-rendering.ts create mode 100644 packages/services/service-automation/src/subflow-guards.ts diff --git a/packages/services/service-automation/src/consumed-suspension-restore.ts b/packages/services/service-automation/src/consumed-suspension-restore.ts new file mode 100644 index 00000000000..a7ffaa6a2bb --- /dev/null +++ b/packages/services/service-automation/src/consumed-suspension-restore.ts @@ -0,0 +1,243 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The consumed-suspension restore of {@link AutomationEngine}: how a consumed + * suspension's witnesses are resolved, and how a consumed suspension is + * restored along its chain, peeked at or inspected. + * + * [#25222] Slice 7 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 + * ConsumedSuspensionRestoreHost} 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 { describeThrownForLog } from './thrown-cause-diagnostics.js'; +// The values below (`MAX_RESTORE_CHAIN_HOPS`, `chainEntryOf` and +// `nestedChildRunId`) 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 { + MAX_RESTORE_CHAIN_HOPS, + chainEntryOf, + nestedChildRunId, + type AutomationEngine, + type ChainRestoreEntry, + type ConsumedSuspension, + type ConsumedSuspensionDropNotice, + type ConsumedSuspensionInspection, + type RunRecord, + type SuspensionRestoreResult, +} 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` reassigns it: + * each read in the body reads the engine's current value, as it did through + * `this`. + */ +export interface ConsumedSuspensionRestoreHost { + readonly consumedSuspensions: AutomationEngine['consumedSuspensions']; + readonly rowSupersedesJournal: AutomationEngine['rowSupersedesJournal']; + readonly resolveRestoreChain: AutomationEngine['resolveRestoreChain']; + readonly restoreOneConsumedSuspension: AutomationEngine['restoreOneConsumedSuspension']; + readonly peekConsumedSuspension: AutomationEngine['peekConsumedSuspension']; + readonly loadSuspendedRunStrict: AutomationEngine['loadSuspendedRunStrict']; + readonly store: AutomationEngine['store']; + readonly resolveConsumedSuspensionWitnesses: AutomationEngine['resolveConsumedSuspensionWitnesses']; + readonly logger: AutomationEngine['logger']; +} + +/** The body of {@link AutomationEngine.rowSupersedesJournal}, moved unedited; that method's docblock is its contract. */ +export function rowSupersedesJournal( + hot: ConsumedSuspension, + rowPause: { nodeId?: string; correlation?: string }, +): boolean { + const samePause = + hot.run.nodeId === rowPause.nodeId && + (hot.run.correlation ?? undefined) === (rowPause.correlation ?? undefined); + if (samePause) return false; + return hot.persisted === 'landed'; +} + +/** The body of {@link AutomationEngine.resolveConsumedSuspensionWitnesses}, moved unedited; that method's docblock is its contract. */ +export function resolveConsumedSuspensionWitnesses( + this: Pick, + runId: string, + terminal: RunRecord | null, +): { + consumed?: ConsumedSuspension; + witness?: 'journal' | 'durable'; + dropped?: ConsumedSuspensionDropNotice; + /** The hot copy describes a pause the run has since LEFT. */ + staleHot: boolean; +} { + const hot = this.consumedSuspensions.get(runId); + if (!terminal) { + return hot ? { consumed: hot, witness: 'journal', staleHot: false } : { staleHot: false }; + } + if (terminal.consumedSuspension) { + const durable: ConsumedSuspension = { + run: terminal.consumedSuspension, + consumedAt: terminal.finishedAt ?? terminal.startedAt, + error: terminal.error ?? '', + }; + if (hot && !this.rowSupersedesJournal(hot, durable.run)) { + return { consumed: hot, witness: 'journal', staleHot: false }; + } + return { consumed: durable, witness: 'durable', staleHot: hot !== undefined }; + } + if (terminal.consumedSuspensionDropped) { + const dropped = terminal.consumedSuspensionDropped; + if (hot && !this.rowSupersedesJournal(hot, dropped)) { + return { consumed: hot, witness: 'journal', dropped, staleHot: false }; + } + return { dropped, staleHot: hot !== undefined }; + } + if (hot && hot.persisted !== 'landed') { + // The row predates this strand — this process's own write for it + // never reached the store (in flight, or failed and reported at + // `error`). The store's silence says nothing. + return { consumed: hot, witness: 'journal', staleHot: false }; + } + // A later terminal record with no snapshot and no drop notice: + // completed, cancelled or cascade-failed after any hot copy was + // taken. Stale by definition. + return { staleHot: hot !== undefined }; +} + +/** The body of {@link AutomationEngine.restoreConsumedSuspension}, moved unedited; that method's docblock is its contract. */ +export async function restoreConsumedSuspension( + this: Pick, + runId: string, + options?: { requestedBy?: string; reason?: string }, +): Promise { + // [#15222] Resolve the nested chain FIRST, read-only. A flat run finds + // nothing here and takes the per-run call below unchanged — same + // guards, same answer, and a result with no `chain` key at all. + // + // A nested one is the whole of this card: the ancestors' pauses were + // consumed by the same cascade and journalled by + // {@link failSuspendedRun}, and re-arming only the run the operator + // named would put the deployment back in the condition the verb exists + // to leave — a leaf that completes into an ancestor which never + // continues. So the chain is one unit of repair, and this is the call + // that makes it one. + const { below, above } = await this.resolveRestoreChain(runId); + if (below.length === 0 && above.length === 0) { + return this.restoreOneConsumedSuspension(runId, options); + } + // DEEPEST FIRST, and the order IS the guarantee: an ancestor becomes + // resumable only after the run it is parked awaiting is parked again, + // so a resume arriving mid-repair delegates into a live child instead + // of continuing past a nested node whose child is terminal. + // + // Every member goes through the SAME per-run call — its own + // `restoring` claim, its own strict live-suspension read, its own + // two-witness read, its own durable park. So the verb's idempotence + // and the #14333 advance claim hold PER RUN in the chain rather than + // once for the chain, and a second chain restore finds every member + // parked and answers `RUN_SUSPENDED` without minting a second pause + // anywhere. + const chain: ChainRestoreEntry[] = []; + for (const memberId of below) { + chain.push(chainEntryOf(memberId, await this.restoreOneConsumedSuspension(memberId, options, runId))); + } + const own = await this.restoreOneConsumedSuspension(runId, options); + chain.push(chainEntryOf(runId, own)); + for (const memberId of above) { + chain.push(chainEntryOf(memberId, await this.restoreOneConsumedSuspension(memberId, options, runId))); + } + // The six fields above `chain` keep describing the run the caller + // NAMED, wherever in the chain it sits — a consumer that never heard + // of nesting reads exactly what it always read. + return { ...own, chain }; +} + +/** The body of {@link AutomationEngine.resolveRestoreChain}, moved unedited; that method's docblock is its contract. */ +export async function resolveRestoreChain(this: Pick, runId: string): Promise<{ below: string[]; above: string[] }> { + const below: string[] = []; + const above: string[] = []; + const self = await this.peekConsumedSuspension(runId); + if (!self) return { below, above }; + + let cursor = self; + for (let hops = 0; hops < MAX_RESTORE_CHAIN_HOPS; hops++) { + const childRunId = nestedChildRunId(cursor.run.correlation); + if (childRunId === undefined || childRunId === runId || below.includes(childRunId)) break; + const child = await this.peekConsumedSuspension(childRunId, runId); + if (!child) break; + below.unshift(childRunId); + cursor = child; + } + + cursor = self; + for (let hops = 0; hops < MAX_RESTORE_CHAIN_HOPS; hops++) { + const parentRunId = (cursor.run.context as Record | undefined)?.$parentRunId; + if (typeof parentRunId !== 'string' || !parentRunId) break; + if (parentRunId === runId || above.includes(parentRunId)) break; + const parent = await this.peekConsumedSuspension(parentRunId, runId); + if (!parent) break; + above.push(parentRunId); + cursor = parent; + } + return { below, above }; +} + +/** The body of {@link AutomationEngine.peekConsumedSuspension}, moved unedited; that method's docblock is its contract. */ +export async function peekConsumedSuspension(this: Pick, runId: string, chainOf?: string): Promise { + try { + if (await this.loadSuspendedRunStrict(runId)) return undefined; + const terminal = this.store?.loadTerminal ? await this.store.loadTerminal(runId) : null; + return this.resolveConsumedSuspensionWitnesses(runId, terminal).consumed; + } catch (err) { + if (chainOf !== undefined) { + // #6299 — the driver's own text rides the structured slot. + // #4632 verdict: FUNCTIONAL — the repair is visibly short (the + // `chain` this call returns names only what it reached) and the + // remedy is to re-issue, which is safe per run. + this.logger.warn( + `[automation] restoreConsumedSuspension('${chainOf}') could not read run '${runId}' while ` + + `walking its nested chain, so the walk STOPPED there and nothing beyond it was re-armed ` + + `— the repair this call reports is partial. Nothing was written for that run. Fix the ` + + `store failure in this record's meta, then re-issue the restore: it is idempotent per ` + + `run and puts back only what is still missing.`, + describeThrownForLog(err), + ); + } + return undefined; + } +} + +/** The body of {@link AutomationEngine.inspectConsumedSuspension}, moved unedited; that method's docblock is its contract. */ +export async function inspectConsumedSuspension(this: Pick, runId: string): Promise { + // Already resumable? STRICT read — a store outage throws through. + const live = await this.loadSuspendedRunStrict(runId); + if (live) return { repairable: false, runId, reason: 'RUN_SUSPENDED', nodeId: live.nodeId }; + + let terminal: RunRecord | null = null; + if (this.store?.loadTerminal) terminal = await this.store.loadTerminal(runId); + + const { consumed, witness, dropped } = this.resolveConsumedSuspensionWitnesses(runId, terminal); + if (consumed && witness) { + return { + repairable: true, + runId, + flowName: consumed.run.flowName, + nodeId: consumed.run.nodeId, + ...(consumed.run.correlation !== undefined ? { correlation: consumed.run.correlation } : {}), + witness, + consumedAt: consumed.consumedAt, + }; + } + if (dropped) return { repairable: false, runId, reason: 'SNAPSHOT_DROPPED', dropped }; + return { repairable: false, runId, reason: 'NO_CONSUMED_SUSPENSION' }; +} diff --git a/packages/services/service-automation/src/dispatch-claims.ts b/packages/services/service-automation/src/dispatch-claims.ts new file mode 100644 index 00000000000..958df7bca2b --- /dev/null +++ b/packages/services/service-automation/src/dispatch-claims.ts @@ -0,0 +1,203 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The dispatch and suspension claims of {@link AutomationEngine}: how a + * dispatch is claimed, in the durable store or in process, then settled and + * read back; how a run id is minted; how a degraded advance claim is announced; + * and how a suspension is released. + * + * [#25222] Slice 7 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 DispatchClaimsHost} + * 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 { describeThrownForLog } from './thrown-cause-diagnostics.js'; +import type { AutomationEngine, FlowDispatchClaim, FlowDispatchOutcome, SuspendedRun, SuspensionReleaseReason } 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`. + * `flowDispatchStore` goes through a getter, because `setFlowDispatchStore` + * reassigns it: each read in the body reads the engine's current value, as it + * did through `this`. Three warned-once flags go through a get/set pair + * (#25508), because a moved body sets each: `dispatchClaimDegradationWarned` + * (`claim`), `dispatchOutcomeDegradationWarned` (`readDispatch`) and + * `advanceClaimDegradationWarned` (`warnAdvanceClaimDegraded`). Each write + * lands on the engine, as it did through `this`. + */ +export interface DispatchClaimsHost { + readonly flowDispatchStore: AutomationEngine['flowDispatchStore']; + readonly logger: AutomationEngine['logger']; + readonly claimInProcess: AutomationEngine['claimInProcess']; + dispatchClaimDegradationWarned: AutomationEngine['dispatchClaimDegradationWarned']; + readonly pruneInProcessDispatchClaims: AutomationEngine['pruneInProcessDispatchClaims']; + readonly inProcessDispatchClaims: AutomationEngine['inProcessDispatchClaims']; + dispatchOutcomeDegradationWarned: AutomationEngine['dispatchOutcomeDegradationWarned']; + advanceClaimDegradationWarned: AutomationEngine['advanceClaimDegradationWarned']; + readonly resolveSuspendedNodeType: AutomationEngine['resolveSuspendedNodeType']; + readonly nodeExecutors: AutomationEngine['nodeExecutors']; +} + +/** The body of {@link AutomationEngine.claim}, moved unedited; that method's docblock is its contract. */ +export async function claim(this: Pick, key: string): Promise { + if (this.flowDispatchStore) { + try { + return await this.flowDispatchStore.claim(key); + } catch (err) { + this.logger.warn( + `[automation] flow-dispatch claim '${key}' failed against the persisted ledger — ` + + `falling back to in-process dedup for this key (availability over strict-once: ` + + `the dispatch proceeds unless this process already made it; a kernel rebuild may re-dispatch it). ` + + `The store failure is in this record's meta.`, + describeThrownForLog(err), + ); + return this.claimInProcess(key); + } + } + if (!this.dispatchClaimDegradationWarned) { + this.dispatchClaimDegradationWarned = true; + this.logger.warn( + '[automation] no persisted flow-dispatch ledger (no ObjectQL engine, or sys_flow_dispatch not registered) — ' + + 'trigger dispatch dedup is IN-PROCESS ONLY and will NOT survive a kernel rebuild: ' + + 'the same record/window can be re-dispatched after a restart.', + ); + } + return this.claimInProcess(key); +} + +/** The body of {@link AutomationEngine.claimInProcess}, moved unedited; that method's docblock is its contract. */ +export function claimInProcess(this: Pick, key: string): boolean { + const now = Date.now(); + this.pruneInProcessDispatchClaims(now); + if (this.inProcessDispatchClaims.has(key)) return false; + this.inProcessDispatchClaims.set(key, { at: now, outcome: null }); + return true; +} + +/** The body of {@link AutomationEngine.settleDispatch}, moved unedited; that method's docblock is its contract. */ +export async function settleDispatch(this: Pick, key: string, outcome: FlowDispatchOutcome): Promise { + const store = this.flowDispatchStore; + if (store && typeof store.settle === 'function') { + try { + await store.settle(key, outcome); + return; + } catch (err) { + this.logger.warn( + `[automation] flow-dispatch settle '${key}' → '${outcome}' failed against the persisted ledger — ` + + `the claim stays UNSETTLED, which reads as 'not delivered': a later replay of this window is ` + + `allowed through rather than refused. The store failure is in this record's meta.`, + describeThrownForLog(err), + ); + } + } + const existing = this.inProcessDispatchClaims.get(key); + // Same write rule as the persisted ledger: `succeeded` is absorbing, so + // the fallback cannot reopen a re-delivery door the durable path keeps + // shut. Kept here rather than imported so the engine stays free of a + // dependency on the store module it merely drives. + if (existing && !(existing.outcome === 'succeeded' && outcome === 'failed')) { + this.inProcessDispatchClaims.set(key, { ...existing, outcome }); + } +} + +/** The body of {@link AutomationEngine.readDispatch}, moved unedited; that method's docblock is its contract. */ +export async function readDispatch(this: Pick, key: string): Promise { + const store = this.flowDispatchStore; + if (store && typeof store.read === 'function') { + try { + return await store.read(key); + } catch (err) { + this.logger.warn( + `[automation] flow-dispatch read '${key}' failed against the persisted ledger — ` + + `reporting the key as UNCLAIMED (availability over strict-once: a replay proceeds rather ` + + `than being refused on a reading we could not make). The store failure is in this record's meta.`, + describeThrownForLog(err), + ); + return null; + } + } + if (store && !this.dispatchOutcomeDegradationWarned) { + this.dispatchOutcomeDegradationWarned = true; + // The ledger predates the outcome half of the claim contract. + this.logger.warn( + '[automation] the attached flow-dispatch ledger has no read() — this message describes the ' + + 'ledger that has no read()/settle() AT ALL, where dispatch claims are still deduplicated ' + + 'and the in-process fallback still records outcomes and still refuses WITHIN one process ' + + 'lifetime, but nothing outlives a restart. A ledger that has settle() but no read() is ' + + 'weaker still: outcomes are written durably and never read back, so no replay is ever ' + + 'refused, in this lifetime or any other.', + ); + } + this.pruneInProcessDispatchClaims(Date.now()); + const rec = this.inProcessDispatchClaims.get(key); + if (!rec) return null; + return { + key, + claimedAt: new Date(rec.at).toISOString(), + outcome: rec.outcome, + settledAt: null, + }; +} + +/** The body of {@link AutomationEngine.nextRunId}, moved unedited; that method's docblock is its contract. */ +export function nextRunId(): string { + const g = globalThis as { crypto?: { randomUUID?: () => string } }; + const rand = g.crypto?.randomUUID + ? g.crypto.randomUUID() + : `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`; + return `run_${rand}`; +} + +/** The body of {@link AutomationEngine.warnAdvanceClaimDegraded}, moved unedited; that method's docblock is its contract. */ +export function warnAdvanceClaimDegraded(this: Pick, cause: string): void { + if (this.advanceClaimDegradationWarned) return; + this.advanceClaimDegradationWarned = true; + this.logger.warn( + `[automation] no cross-replica advance guarantee (${cause}) — resume idempotency is ` + + `IN-PROCESS ONLY: two decisions on one run arriving at the same moment on two replicas ` + + `can both advance it, running every downstream side effect twice. Attach a suspended-run ` + + `store that implements claimSuspension() to close it.`, + ); +} + +/** The body of {@link AutomationEngine.releaseSuspension}, moved unedited; that method's docblock is its contract. */ +export async function releaseSuspension(this: Pick, run: SuspendedRun, reason: SuspensionReleaseReason): Promise { + const nodeType = this.resolveSuspendedNodeType(run); + const executor = nodeType ? this.nodeExecutors.get(nodeType) : undefined; + if (!executor?.onSuspensionReleased) return; + try { + await executor.onSuspensionReleased({ + runId: run.runId, + flowName: run.flowName, + nodeId: run.nodeId, + correlation: run.correlation, + reason, + }); + } catch (err) { + // #6499 — the thrown text is PLUGIN-SUPPLIED (the node executor's + // own teardown), so it goes to the structured slot; see + // `forgetSuspendedRun`'s catch above for the full mechanism + // (#6299). The message keeps the correlation handle — that is the + // thing an operator cleans up by hand. + // + // #4632 verdict: FUNCTIONAL — stays `warn`, per the docblock: the + // run continued and nothing claimed-persisted failed to land; the + // leftover is whatever the node armed on entry (a misleading + // `sys_job` row, not a broken run). `warn(message, meta?)` — meta + // is the SECOND argument; `warn` has no `Error` slot. + this.logger.warn( + `[automation] run '${run.runId}': '${nodeType}' node '${run.nodeId}' failed to release its suspension ` + + `(reason: ${reason}, correlation: ${run.correlation ?? 'none'}) — the run continued; whatever the ` + + `node armed on entry may still be scheduled. The executor's own failure is in this record's meta.`, + describeThrownForLog(err), + ); + } +} diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 3e937a874ab..45686efa721 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -21,8 +21,7 @@ import { } from './screen-input-contract.js'; import type { Logger } from '@objectstack/spec/contracts'; import type { FlowUnboundCause } from '@objectstack/spec/contracts'; -import { FLOW_HOOK_SECRET_KEY, flowCredentialClassLabel, flowCredentialPositions } from './flow-credential-projection.js'; -import { FlowSchema, FLOW_RESERVED_NODE_TYPES, FLOW_STRUCTURAL_NODE_TYPES, validateControlFlow, collectFlowGraphs } from '@objectstack/spec/automation'; +import { FLOW_RESERVED_NODE_TYPES, FLOW_STRUCTURAL_NODE_TYPES } from '@objectstack/spec/automation'; // [#14328] The ONE answer to "which trigger kind does this flow ask for?" — // shared with `defineStack`'s trigger-capability refusal and `@objectstack/lint`'s // `validate-flow-trigger-readiness`, so the runtime cannot drift from what @@ -33,7 +32,7 @@ import { scheduledWorkDisabledReason, type ScheduledWorkPolicy, } from '@objectstack/types'; -import { applyConversionsToFlow, CONVERSION_TODO_CODE, type ConversionNotice, type ConversionTodoNotice } from '@objectstack/spec'; +import { CONVERSION_TODO_CODE, type ConversionNotice, type ConversionTodoNotice } from '@objectstack/spec'; // [ADR-0126 §7.3, #20761] "Does a code package ship this flow?" is NOT read // off a definition here. The subflow guards, the arming gate and the // activation door ask {@link AutomationEngine.packagedFlowOwner} — the loader's @@ -70,17 +69,10 @@ import { summarizeRun } from './run-summary.js'; // provider factory's text), so it renders it as structured `meta` rather than // interpolating it into the log message. See ./thrown-cause-diagnostics.ts. import { describeThrownForLog } from './thrown-cause-diagnostics.js'; -// [#15788] The refusing `end` node renders its `message` through the SAME -// interpolator a screen `description` gets — the ruling's own words, so a -// shared function rather than a second template engine. A value import from -// `./builtin/` is safe in this direction: `template.ts` imports only -// `../guard-refusal.js` and package-external contracts, so nothing it pulls in -// reaches back here. -import { renderTextSlot, FlowTextTemplateError } from './builtin/template.js'; // [#22450] The one bundle-locale negotiation rule (exact → case-insensitive → // base language → variant expansion), asked about the locales the i18n service // holds — the same question objectql's `negotiatedMessageLocale` asks it. -import { flowRefusalMessageKey, resolveBundleLocale } from '@objectstack/spec/system'; +import { resolveBundleLocale } from '@objectstack/spec/system'; // [#25222] The registration-time checks' bodies, and the declarations only // they read. Each `AutomationEngine` method of the same name calls its body. import { @@ -161,6 +153,76 @@ import { cloneVariablesAtPause, failAncestors, } from './run-resume.js'; +// [#25222] The dispatch and suspension claims' bodies: claiming, settling and +// reading a dispatch, minting a run id, announcing a degraded advance claim and +// releasing a suspension. Each `AutomationEngine` method of the same name calls +// its body. +import { + claim, + claimInProcess, + settleDispatch, + readDispatch, + nextRunId, + warnAdvanceClaimDegraded, + releaseSuspension, +} from './dispatch-claims.js'; +// [#25222] The flow definitions' bodies: canonicalising, unregistering, +// refusing and rolling back a flow, reporting its runtime states and trigger +// bindings, and the function, connector and credential surfaces a flow reads. +// Each `AutomationEngine` method of the same name calls its body. +import { + resolveFunction, + registerConnectorProvider, + canonicalizeStoredFlow, + canonicalizeFlow, + unregisterFlow, + recordFlowRefusal, + getFlowRuntimeStates, + describeUnboundReason, + getTriggerBindingAudit, + flowCredentialHoldings, + hookSecretResolver, + rollbackFlow, +} from './flow-definitions.js'; +// [#25222] The subflow guards' bodies: finding and describing disabled or +// packaged subflows and their callers, and refusing what would reach a disabled +// subflow. Each `AutomationEngine` method of the same name calls its body. +import { + describeDisabledFlow, + subflowTargets, + packagedSubflowCallers, + parkedRunsOf, + refuseUnderReachingCallers, + disabledPackagedSubflows, + ledgerDisabledChainReaches, + refuseEnableOntoDisabledSubflow, + describeDisabledSubflows, + rejudgeSubflowCallers, + refuseCustomerAuthoredToggle, +} from './subflow-guards.js'; +// [#25222] The consumed-suspension restore's bodies: resolving a consumed +// suspension's witnesses, and restoring, peeking at or inspecting one. Each +// `AutomationEngine` method of the same name calls its body. +import { + rowSupersedesJournal, + resolveConsumedSuspensionWitnesses, + restoreConsumedSuspension, + resolveRestoreChain, + peekConsumedSuspension, + inspectConsumedSuspension, +} from './consumed-suspension-restore.js'; +// [#25222] The DAG traversal's node-level helpers: the node-type vocabulary +// checks a run makes, a failed node's error variable, and rendering refusal +// messages and flow text slots. Each `AutomationEngine` method of the same name +// calls its body. +import { + unknownNodeTypes, + warnIfNodeTypeVocabularyNeverSealed, + setNodeError, + renderRefusalMessage, + renderFlowTextSlot, + translatedFlowTemplate, +} from './node-text-rendering.js'; // ─── Node Executor Interface (Plugin Extension Point) ─────────────── @@ -764,7 +826,7 @@ export const MAX_CONSUMED_SUSPENSIONS = 50; * the repair can reach exactly as deep as the cascade could, and a corrupt * context cannot make either loop run forever. */ -const MAX_RESTORE_CHAIN_HOPS = 32; +export const MAX_RESTORE_CHAIN_HOPS = 32; /** * [#15222] The child run a parked run is awaiting, read off the correlation @@ -777,7 +839,7 @@ const MAX_RESTORE_CHAIN_HOPS = 32; * to. Naming it once here is what keeps the restore walk and the resume walk * reading one fact. */ -function nestedChildRunId(correlation: string | undefined): string | undefined { +export function nestedChildRunId(correlation: string | undefined): string | undefined { if (typeof correlation !== 'string') return undefined; for (const prefix of ['subflow:', 'map:']) { if (correlation.startsWith(prefix)) return correlation.slice(prefix.length) || undefined; @@ -786,7 +848,7 @@ function nestedChildRunId(correlation: string | undefined): string | undefined { } /** [#15222] One chain member's outcome, projected from its own restore result. */ -function chainEntryOf(runId: string, result: SuspensionRestoreResult): ChainRestoreEntry { +export function chainEntryOf(runId: string, result: SuspensionRestoreResult): ChainRestoreEntry { return { runId, restored: result.restored, @@ -1268,7 +1330,7 @@ export interface FlowTextSlotTranslation { * whenever there is nothing to negotiate against (no `getLocales`, a throwing or * empty answer, no match). */ -function negotiatedServiceLocale(service: RefusalI18nService, requested: string): string { +export function negotiatedServiceLocale(service: RefusalI18nService, requested: string): string { if (typeof service.getLocales !== 'function') return requested; let available: unknown; try { @@ -2206,7 +2268,7 @@ export const IN_PROCESS_DISPATCH_CLAIM_TTL_MS = 48 * 60 * 60 * 1000; * door refuses it for the same reason (its declared `^17.0.0` floor is a * dependency range, not an age), in the module named above. */ -const CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION: readonly string[] = [ +export const CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION: readonly string[] = [ 'flow-decision-mode-inclusive-explicit', ]; @@ -2225,7 +2287,7 @@ const FLOW_REGISTRATION_TODO_SURFACE = 'flow — a site AutomationEngine.registe * check's own words follow, as a refused-shape TODO's do: the site stays as * stored, and the flow does not register. */ -function flowRegistrationTodo(flowName: string, site: RegistrationCheckSite): ConversionTodoNotice { +export function flowRegistrationTodo(flowName: string, site: RegistrationCheckSite): ConversionTodoNotice { const reason = 'No conversion rewrites it, and registerFlow refuses it: left as stored, the flow is not ' + `registered at boot and never runs. Flow \`${flowName}\`, ${site.detail}`; return { @@ -2264,7 +2326,7 @@ function flowRegistrationTodo(flowName: string, site: RegistrationCheckSite): Co * so the parse never lowers it, so there is no envelope on the parsed side to * copy and the recursion finds a string facing a string. */ -function graftConditionEnvelopes(converted: unknown, parsed: unknown): unknown { +export function graftConditionEnvelopes(converted: unknown, parsed: unknown): unknown { if (Array.isArray(converted)) { if (!Array.isArray(parsed)) return converted; let changed = false; @@ -2796,39 +2858,29 @@ export class AutomationEngine implements IAutomationService { * it — so a store outage never blocks a dispatch, and never double-fires * within one process lifetime either. */ - async claim(key: string): Promise { - if (this.flowDispatchStore) { - try { - return await this.flowDispatchStore.claim(key); - } catch (err) { - this.logger.warn( - `[automation] flow-dispatch claim '${key}' failed against the persisted ledger — ` + - `falling back to in-process dedup for this key (availability over strict-once: ` + - `the dispatch proceeds unless this process already made it; a kernel rebuild may re-dispatch it). ` + - `The store failure is in this record's meta.`, - describeThrownForLog(err), - ); - return this.claimInProcess(key); - } - } - if (!this.dispatchClaimDegradationWarned) { - this.dispatchClaimDegradationWarned = true; - this.logger.warn( - '[automation] no persisted flow-dispatch ledger (no ObjectQL engine, or sys_flow_dispatch not registered) — ' + - 'trigger dispatch dedup is IN-PROCESS ONLY and will NOT survive a kernel rebuild: ' + - 'the same record/window can be re-dispatched after a restart.', - ); - } - return this.claimInProcess(key); + claim(key: string): Promise { + // The body lives in ./dispatch-claims.ts (#25222). + const readFlowDispatchStore = () => this.flowDispatchStore; + const readDispatchClaimDegradationWarned = () => this.dispatchClaimDegradationWarned; + const writeDispatchClaimDegradationWarned = (v: AutomationEngine['dispatchClaimDegradationWarned']) => { this.dispatchClaimDegradationWarned = v; }; + const host: ThisParameterType = { + get flowDispatchStore() { return readFlowDispatchStore(); }, + logger: this.logger, + claimInProcess: (...a) => this.claimInProcess(...a), + get dispatchClaimDegradationWarned() { return readDispatchClaimDegradationWarned(); }, + set dispatchClaimDegradationWarned(v) { writeDispatchClaimDegradationWarned(v); }, + }; + return claim.call(host, key); } /** In-process half of {@link claim}: TTL-pruned check-and-record. */ private claimInProcess(key: string): boolean { - const now = Date.now(); - this.pruneInProcessDispatchClaims(now); - if (this.inProcessDispatchClaims.has(key)) return false; - this.inProcessDispatchClaims.set(key, { at: now, outcome: null }); - return true; + // The body lives in ./dispatch-claims.ts (#25222). + const host: ThisParameterType = { + pruneInProcessDispatchClaims: (...a) => this.pruneInProcessDispatchClaims(...a), + inProcessDispatchClaims: this.inProcessDispatchClaims, + }; + return claimInProcess.call(host, key); } private pruneInProcessDispatchClaims(now: number): void { @@ -2858,29 +2910,15 @@ export class AutomationEngine implements IAutomationService { * refuses by not writing, never by throwing — a refusal is the invariant * working, not a failure to report. */ - async settleDispatch(key: string, outcome: FlowDispatchOutcome): Promise { - const store = this.flowDispatchStore; - if (store && typeof store.settle === 'function') { - try { - await store.settle(key, outcome); - return; - } catch (err) { - this.logger.warn( - `[automation] flow-dispatch settle '${key}' → '${outcome}' failed against the persisted ledger — ` + - `the claim stays UNSETTLED, which reads as 'not delivered': a later replay of this window is ` + - `allowed through rather than refused. The store failure is in this record's meta.`, - describeThrownForLog(err), - ); - } - } - const existing = this.inProcessDispatchClaims.get(key); - // Same write rule as the persisted ledger: `succeeded` is absorbing, so - // the fallback cannot reopen a re-delivery door the durable path keeps - // shut. Kept here rather than imported so the engine stays free of a - // dependency on the store module it merely drives. - if (existing && !(existing.outcome === 'succeeded' && outcome === 'failed')) { - this.inProcessDispatchClaims.set(key, { ...existing, outcome }); - } + settleDispatch(key: string, outcome: FlowDispatchOutcome): Promise { + // The body lives in ./dispatch-claims.ts (#25222). + const readFlowDispatchStore = () => this.flowDispatchStore; + const host: ThisParameterType = { + get flowDispatchStore() { return readFlowDispatchStore(); }, + logger: this.logger, + inProcessDispatchClaims: this.inProcessDispatchClaims, + }; + return settleDispatch.call(host, key, outcome); } /** @@ -2892,42 +2930,20 @@ export class AutomationEngine implements IAutomationService { * takes everywhere else: the caller re-runs, it never refuses on a reading * it could not make. */ - async readDispatch(key: string): Promise { - const store = this.flowDispatchStore; - if (store && typeof store.read === 'function') { - try { - return await store.read(key); - } catch (err) { - this.logger.warn( - `[automation] flow-dispatch read '${key}' failed against the persisted ledger — ` + - `reporting the key as UNCLAIMED (availability over strict-once: a replay proceeds rather ` + - `than being refused on a reading we could not make). The store failure is in this record's meta.`, - describeThrownForLog(err), - ); - return null; - } - } - if (store && !this.dispatchOutcomeDegradationWarned) { - this.dispatchOutcomeDegradationWarned = true; - // The ledger predates the outcome half of the claim contract. - this.logger.warn( - '[automation] the attached flow-dispatch ledger has no read() — this message describes the ' + - 'ledger that has no read()/settle() AT ALL, where dispatch claims are still deduplicated ' + - 'and the in-process fallback still records outcomes and still refuses WITHIN one process ' + - 'lifetime, but nothing outlives a restart. A ledger that has settle() but no read() is ' + - 'weaker still: outcomes are written durably and never read back, so no replay is ever ' + - 'refused, in this lifetime or any other.', - ); - } - this.pruneInProcessDispatchClaims(Date.now()); - const rec = this.inProcessDispatchClaims.get(key); - if (!rec) return null; - return { - key, - claimedAt: new Date(rec.at).toISOString(), - outcome: rec.outcome, - settledAt: null, + readDispatch(key: string): Promise { + // The body lives in ./dispatch-claims.ts (#25222). + const readFlowDispatchStore = () => this.flowDispatchStore; + const readDispatchOutcomeDegradationWarned = () => this.dispatchOutcomeDegradationWarned; + const writeDispatchOutcomeDegradationWarned = (v: AutomationEngine['dispatchOutcomeDegradationWarned']) => { this.dispatchOutcomeDegradationWarned = v; }; + const host: ThisParameterType = { + get flowDispatchStore() { return readFlowDispatchStore(); }, + logger: this.logger, + get dispatchOutcomeDegradationWarned() { return readDispatchOutcomeDegradationWarned(); }, + set dispatchOutcomeDegradationWarned(v) { writeDispatchOutcomeDegradationWarned(v); }, + pruneInProcessDispatchClaims: (...a) => this.pruneInProcessDispatchClaims(...a), + inProcessDispatchClaims: this.inProcessDispatchClaims, }; + return readDispatch.call(host, key); } /** @@ -2937,11 +2953,8 @@ export class AutomationEngine implements IAutomationService { * a still-suspended durable run). */ private nextRunId(): string { - const g = globalThis as { crypto?: { randomUUID?: () => string } }; - const rand = g.crypto?.randomUUID - ? g.crypto.randomUUID() - : `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`; - return `run_${rand}`; + // The body lives in ./dispatch-claims.ts (#25222). + return nextRunId(); } /** @@ -3306,14 +3319,15 @@ export class AutomationEngine implements IAutomationService { * `error` here would train everyone to skim `error`. */ private warnAdvanceClaimDegraded(cause: string): void { - if (this.advanceClaimDegradationWarned) return; - this.advanceClaimDegradationWarned = true; - this.logger.warn( - `[automation] no cross-replica advance guarantee (${cause}) — resume idempotency is ` + - `IN-PROCESS ONLY: two decisions on one run arriving at the same moment on two replicas ` + - `can both advance it, running every downstream side effect twice. Attach a suspended-run ` + - `store that implements claimSuspension() to close it.`, - ); + // The body lives in ./dispatch-claims.ts (#25222). + const readAdvanceClaimDegradationWarned = () => this.advanceClaimDegradationWarned; + const writeAdvanceClaimDegradationWarned = (v: AutomationEngine['advanceClaimDegradationWarned']) => { this.advanceClaimDegradationWarned = v; }; + const host: ThisParameterType = { + get advanceClaimDegradationWarned() { return readAdvanceClaimDegradationWarned(); }, + set advanceClaimDegradationWarned(v) { writeAdvanceClaimDegradationWarned(v); }, + logger: this.logger, + }; + warnAdvanceClaimDegraded.call(host, cause); } /** @@ -3333,37 +3347,14 @@ export class AutomationEngine implements IAutomationService { * created the pause is the one that owns tearing it down, and nothing else * has to pattern-match correlation strings it does not own. */ - private async releaseSuspension(run: SuspendedRun, reason: SuspensionReleaseReason): Promise { - const nodeType = this.resolveSuspendedNodeType(run); - const executor = nodeType ? this.nodeExecutors.get(nodeType) : undefined; - if (!executor?.onSuspensionReleased) return; - try { - await executor.onSuspensionReleased({ - runId: run.runId, - flowName: run.flowName, - nodeId: run.nodeId, - correlation: run.correlation, - reason, - }); - } catch (err) { - // #6499 — the thrown text is PLUGIN-SUPPLIED (the node executor's - // own teardown), so it goes to the structured slot; see - // `forgetSuspendedRun`'s catch above for the full mechanism - // (#6299). The message keeps the correlation handle — that is the - // thing an operator cleans up by hand. - // - // #4632 verdict: FUNCTIONAL — stays `warn`, per the docblock: the - // run continued and nothing claimed-persisted failed to land; the - // leftover is whatever the node armed on entry (a misleading - // `sys_job` row, not a broken run). `warn(message, meta?)` — meta - // is the SECOND argument; `warn` has no `Error` slot. - this.logger.warn( - `[automation] run '${run.runId}': '${nodeType}' node '${run.nodeId}' failed to release its suspension ` + - `(reason: ${reason}, correlation: ${run.correlation ?? 'none'}) — the run continued; whatever the ` + - `node armed on entry may still be scheduled. The executor's own failure is in this record's meta.`, - describeThrownForLog(err), - ); - } + private releaseSuspension(run: SuspendedRun, reason: SuspensionReleaseReason): Promise { + // The body lives in ./dispatch-claims.ts (#25222). + const host: ThisParameterType = { + resolveSuspendedNodeType: (...a) => this.resolveSuspendedNodeType(...a), + nodeExecutors: this.nodeExecutors, + logger: this.logger, + }; + return releaseSuspension.call(host, run, reason); } /** @@ -4007,10 +3998,12 @@ export class AutomationEngine implements IAutomationService { * no-op'ing. */ resolveFunction(name: string): FlowFunctionRegistration | undefined { - const resolved = this.functionResolver?.(name); - if (!resolved) return undefined; - if (typeof resolved === 'function') return { handler: resolved }; - return typeof resolved.handler === 'function' ? resolved : undefined; + // The body lives in ./flow-definitions.ts (#25222). + const readFunctionResolver = () => this.functionResolver; + const host: ThisParameterType = { + get functionResolver() { return readFunctionResolver(); }, + }; + return resolveFunction.call(host, name); } /** Get all registered connector names. */ @@ -4043,11 +4036,12 @@ export class AutomationEngine implements IAutomationService { * key replaces (mirrors {@link registerNodeExecutor}). */ registerConnectorProvider(providerKey: string, factory: ConnectorProviderFactory): void { - if (this.connectorProviders.has(providerKey)) { - this.logger.warn(`Connector provider '${providerKey}' replaced`); - } - this.connectorProviders.set(providerKey, factory); - this.logger.info(`Connector provider registered: ${providerKey}`); + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + connectorProviders: this.connectorProviders, + logger: this.logger, + }; + registerConnectorProvider.call(host, providerKey, factory); } /** Unregister a connector provider factory (hot-unplug). */ @@ -4189,36 +4183,12 @@ export class AutomationEngine implements IAutomationService { notices: ConversionNotice[]; todos: ConversionTodoNotice[]; } { - const { parsed, storable, notices, todos: chainTodos } = this.canonicalizeFlow(name, definition, options); - // The registration checks run when `todos` is first READ, once: the - // stored pass reads it, and the doors that only want `parsed` or - // `storable` (the save door, a package duplication, the sync's - // "already held?" comparison) pay nothing for a listing they discard. - const listRegistrationRefusals = (): ConversionTodoNotice[] => { - const todos = [...chainTodos]; - // One listing per site: a site a conversion TODO names (the value-slot - // refusal the spec lists under its D3 entry) keeps that TODO, which - // names the retirement; the registration check's line for it is the - // same refusal and is not repeated. - const listed = new Set(chainTodos.map((todo) => todo.path)); - for (const refusal of this.registrationCheckRefusals(name, parsed)) { - for (const site of refusal.sites) { - if (listed.has(site.path)) continue; - todos.push(flowRegistrationTodo(name, site)); - } - } - return todos; - }; - let listedTodos: ConversionTodoNotice[] | undefined; - return { - parsed, - storable, - notices, - get todos(): ConversionTodoNotice[] { - listedTodos ??= listRegistrationRefusals(); - return listedTodos; - }, + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + canonicalizeFlow: (...a) => this.canonicalizeFlow(...a), + registrationCheckRefusals: (...a) => this.registrationCheckRefusals(...a), }; + return canonicalizeStoredFlow.call(host, name, definition, options); } /** @@ -4234,76 +4204,12 @@ export class AutomationEngine implements IAutomationService { notices: ConversionNotice[]; todos: ConversionTodoNotice[]; } { - // ADR-0087 D2 — the runtime load seam. A stored flow authored against an - // old shape (a `webhook`/`http_request` callout node, a `delete_record` - // with `config.filters`) is canonicalized on rehydration, BEFORE parse + - // execution, so the executor only ever sees the canonical shape and a - // dropped-alias upgrade never silently changes behavior (e.g. an empty - // `filter` deleting a whole table). No live executor registry is handed - // to the pass: the one open-namespace rename, the callout rename, - // reserves its old names instead (#23543 F5-http), and - // `registerNodeExecutor` refuses an executor under one, so (#24001) the - // conversion-conflict channel it used to feed is gone. - // - // `includeRetired` (#3903): this seam rehydrates DATA AT REST — flows - // stored in `sys_metadata` outlive any load window, and a row written - // under protocol N has no author for a tombstone to teach. When a flow - // conversion graduates to `retiredFromLoadPath` (retiring at N+1), the - // stored flows that still carry the old shape must keep canonicalizing - // here or the retirement would silently change their behavior — the - // exact hazard this seam exists to prevent. Authored sources keep - // window semantics at their own seam (`normalizeStackInput` applies - // live-window entries only; the schema tombstones the retired shape). - // [#23775] The one caller that closes the window is the metadata save - // door, for a flow with no stored row (an authored body, not one at - // rest), through `canonicalizeStoredFlow`'s option; the rehydration - // passes none and replays the full chain, as it always has. - // - // `excludeConversionIds`: the retired window is opened for a CLASS of - // caller, and one kind of entry inside it is not a rescue but a - // reinterpretation — a DEFAULT FLIP, whose old shape still parses and - // now means something else. This seam cannot say a body predates such - // a flip, so it refuses those entries by id, each with its reason on - // `CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION`. - const notices: ConversionNotice[] = []; - const todos: ConversionTodoNotice[] = []; - const converted = applyConversionsToFlow(definition, { - includeRetired: options.includeRetired ?? true, - excludeConversionIds: CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION, - onNotice: (n) => { - notices.push(n); - this.logger.warn(`[flow '${name}'] ${n.code}: ${n.message}`); - }, - // Collected, not logged: a site listed here is one `registerFlow` - // refuses below with its own located message. - onTodo: (t) => { - todos.push(t); - }, - }); - // #4347 / #4415 — one call, canonical at every depth. `FlowNodeSchema` - // parses its own ADR-0031 regions (`FlowNodeSchema.transform` → - // `parseFlowNodeRegions`), so what comes back here is already normalized - // inside `loop.config.body`, `parallel.config.branches[]` and - // `try_catch.config.try`/`.catch` — recursively. Until #4415 that needed - // a second, separately-remembered call to `normalizeControlFlowRegions` - // right here, and every consumer that took a `FlowParsed` without making - // it held a half-parsed flow that looked finished. - const parsed = FlowSchema.parse(converted); - - // DAG cycle detection - this.detectCycles(parsed); - - // ADR-0031 — validate structured control-flow constructs (loop bodies, - // parallel branches, try/catch regions) are well-formed (single-entry/ - // single-exit, acyclic). Reject the malformed before it can run. - validateControlFlow(parsed); - - return { - parsed, - storable: graftConditionEnvelopes(converted, parsed), - notices, - todos, + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + logger: this.logger, + detectCycles: (...a) => this.detectCycles(...a), }; + return canonicalizeFlow.call(host, name, definition, options); } registerFlow(name: string, definition: unknown): FlowParsed { @@ -4427,12 +4333,15 @@ export class AutomationEngine implements IAutomationService { * @throws `DELETE_RESTRICTED` / 409 as above; nothing is removed. */ unregisterFlow(name: string): void { - const flow = this.flows.get(name); - if (flow && this.isPackagedFlow(name)) { - const callers = this.packagedSubflowCallers(name); - if (callers.length > 0) this.refuseUnderReachingCallers(name, 'remove', callers, new Map()); - } - this.withdrawFlow(name); + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + flows: this.flows, + isPackagedFlow: (...a) => this.isPackagedFlow(...a), + packagedSubflowCallers: (...a) => this.packagedSubflowCallers(...a), + refuseUnderReachingCallers: (...a) => this.refuseUnderReachingCallers(...a), + withdrawFlow: (...a) => this.withdrawFlow(...a), + }; + unregisterFlow.call(host, name); } /** @@ -4491,12 +4400,12 @@ export class AutomationEngine implements IAutomationService { * before it changes anything), and that flow is registered, not refused. */ recordFlowRefusal(name: string, error: unknown): void { - if (this.flows.has(name)) return; - const thrown = (error ?? {}) as { code?: unknown; message?: unknown }; - this.flowRefusals.set(name, { - ...(typeof thrown.code === 'string' ? { code: thrown.code } : {}), - message: typeof thrown.message === 'string' ? thrown.message : String(error), - }); + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + flows: this.flows, + flowRefusals: this.flowRefusals, + }; + recordFlowRefusal.call(host, name, error); } /** @@ -4542,33 +4451,16 @@ export class AutomationEngine implements IAutomationService { armedFrom?: FlowContender; shadowed?: FlowContender[]; }> { - return [...this.flows.keys()].map((name) => { - const resolved = this.resolveTriggerBinding(name); - // [#11997] Attach the shadowing receipt to the row an admin already - // reads. This map holds ONE entry per bare name, so without these - // two fields a displaced contender leaves no trace on this surface. - const shadowing = this.flowShadowing.get(name); - const unbound = this.describeUnboundReason(name, resolved); - return { - name, - enabled: this.isFlowEnabled(name), - bound: this.boundFlowTriggers.has(name), - status: (this.flows.get(name) as { status?: string } | undefined)?.status, - triggerType: resolved?.triggerType, - object: resolved?.binding.object, - // Absent, not `undefined`-valued: a row that is bound, disabled - // or trigger-less has no reason to carry, and `reason: null` on - // the wire would read as "we looked and found nothing". - // [#23842] `unboundCause` rides the SAME spread, so the pair is - // present together or absent together by construction. - ...(unbound === undefined - ? {} - : { reason: unbound.reason, unboundCause: unbound.cause }), - ...(shadowing - ? { armedFrom: shadowing.armed, shadowed: shadowing.shadowed } - : {}), - }; - }); + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + flows: this.flows, + resolveTriggerBinding: (...a) => this.resolveTriggerBinding(...a), + flowShadowing: this.flowShadowing, + describeUnboundReason: (...a) => this.describeUnboundReason(...a), + isFlowEnabled: (...a) => this.isFlowEnabled(...a), + boundFlowTriggers: this.boundFlowTriggers, + }; + return getFlowRuntimeStates.call(host); } /** @@ -4611,26 +4503,15 @@ export class AutomationEngine implements IAutomationService { name: string, resolved: { triggerType: string } | undefined, ): { reason: string; cause: FlowUnboundCause } | undefined { - if (!resolved) return undefined; // manual / screen flow — nothing to bind - if (!this.isFlowEnabled(name)) return undefined; - if (this.boundFlowTriggers.has(name)) return undefined; - const policyReason = this.policyDisabledFlows.get(name); - if (policyReason !== undefined) return { reason: policyReason, cause: 'deployment_policy' }; - // [#20725, ADR-0126 §7.3] The arming gate declined it onto a disabled - // packaged subflow — read from the record, for the policy line's - // reason, and ahead of the trigger branches for the gate's: with the - // child off, the trigger would not arm it either. - const declined = this.subflowDeclinedFlows.get(name); - if (declined !== undefined) return { reason: declined, cause: 'disabled_subflow' }; - return this.triggers.has(resolved.triggerType) - ? { - reason: `trigger '${resolved.triggerType}' is registered but binding failed — see earlier warnings`, - cause: 'binding_failure', - } - : { - reason: `no '${resolved.triggerType}' trigger is registered — add requires: ['triggers'] (record_change/schedule/time_relative/api ship in @objectstack/trigger-*)`, - cause: 'missing_trigger', - }; + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + isFlowEnabled: (...a) => this.isFlowEnabled(...a), + boundFlowTriggers: this.boundFlowTriggers, + policyDisabledFlows: this.policyDisabledFlows, + subflowDeclinedFlows: this.subflowDeclinedFlows, + triggers: this.triggers, + }; + return describeUnboundReason.call(host, name, resolved); } /** @@ -4643,18 +4524,13 @@ export class AutomationEngine implements IAutomationService { * channel in `os dev` / `os start`). */ getTriggerBindingAudit(): Array<{ flowName: string; triggerType: string; reason: string }> { - const audit: Array<{ flowName: string; triggerType: string; reason: string }> = []; - for (const name of this.flows.keys()) { - const resolved = this.resolveTriggerBinding(name); - // [#18235] The eligibility rules (enabled, unbound, declares a - // trigger) and the three-branch vocabulary both live in - // `describeUnboundReason` now, so this audit and the status door - // report the SAME sentence for the same flow by construction. - const unbound = this.describeUnboundReason(name, resolved); - if (unbound === undefined || !resolved) continue; - audit.push({ flowName: name, triggerType: resolved.triggerType, reason: unbound.reason }); - } - return audit; + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + flows: this.flows, + resolveTriggerBinding: (...a) => this.resolveTriggerBinding(...a), + describeUnboundReason: (...a) => this.describeUnboundReason(...a), + }; + return getTriggerBindingAudit.call(host); } /** @@ -4822,23 +4698,13 @@ export class AutomationEngine implements IAutomationService { * copy would share it. */ flowCredentialHoldings(name: string): FlowCredentialHolding[] { - const flow = this.flows.get(name); - if (!flow) return []; - const out: FlowCredentialHolding[] = []; - const seen = new Set(); - for (const position of flowCredentialPositions(flow)) { - const id = JSON.stringify([position.nodeId, position.key]); - if (position.form === 'value') { - seen.add(id); - out.push({ nodeId: position.nodeId, key: position.key, label: flowCredentialClassLabel(position.key), held: 'literal' }); - } - } - for (const { nodeId, key } of this.flowCredentialSource?.held(name) ?? []) { - const id = JSON.stringify([nodeId, key]); - if (seen.has(id)) continue; - out.push({ nodeId, key, label: flowCredentialClassLabel(key), held: 'channel' }); - } - return out; + // The body lives in ./flow-definitions.ts (#25222). + const readFlowCredentialSource = () => this.flowCredentialSource; + const host: ThisParameterType = { + flows: this.flows, + get flowCredentialSource() { return readFlowCredentialSource(); }, + }; + return flowCredentialHoldings.call(host, name); } /** @@ -4867,18 +4733,12 @@ export class AutomationEngine implements IAutomationService { startNodeId: string | undefined, config: Record, ): (() => Promise) | undefined { - const source = this.flowCredentialSource; - const written = Object.prototype.hasOwnProperty.call(config, FLOW_HOOK_SECRET_KEY); - const literal = typeof config.secret === 'string' && config.secret.trim() !== '' ? config.secret : undefined; - if (written && literal === undefined) return undefined; - if (!source || startNodeId === undefined) { - return literal === undefined ? undefined : async () => literal; - } - if (literal === undefined && !source.holds(flowName, startNodeId, FLOW_HOOK_SECRET_KEY)) return undefined; - return async () => { - if (literal !== undefined && !source.holds(flowName, startNodeId, FLOW_HOOK_SECRET_KEY)) return literal; - return (await source.resolve(flowName, startNodeId, FLOW_HOOK_SECRET_KEY)) ?? literal; + // The body lives in ./flow-definitions.ts (#25222). + const readFlowCredentialSource = () => this.flowCredentialSource; + const host: ThisParameterType = { + get flowCredentialSource() { return readFlowCredentialSource(); }, }; + return hookSecretResolver.call(host, flowName, startNodeId, config); } /** @@ -4942,16 +4802,11 @@ export class AutomationEngine implements IAutomationService { * message is the only channel left). */ private describeDisabledFlow(flowName: string): string { - if (this.flowLedgerDisabled.has(flowName)) { - return ( - `Flow '${flowName}' is disabled — it is switched off for this installation in the ` + - `packaged-metadata activation ledger (sys_metadata_activation, ADR-0126 §7.2). ` + - `Re-enable the packaged flow to arm it again, or run a clone of it under a new name.` - ); - } - // Unchanged wording for the status dimension: it is what every existing - // consumer and test reads, and this leg has no reason to move it. - return `Flow '${flowName}' is disabled`; + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + flowLedgerDisabled: this.flowLedgerDisabled, + }; + return describeDisabledFlow.call(host, flowName); } /** @@ -4980,15 +4835,8 @@ export class AutomationEngine implements IAutomationService { * papering over a producer that is already correct. */ private subflowTargets(flow: FlowParsed): string[] { - const nodes = (flow as { nodes?: unknown }).nodes; - if (!Array.isArray(nodes)) return []; - const targets: string[] = []; - for (const node of nodes) { - const n = node as { type?: unknown; config?: { flowName?: unknown } } | null; - if (!n || (n.type !== 'subflow' && n.type !== 'map')) continue; - if (typeof n.config?.flowName === 'string') targets.push(n.config.flowName); - } - return targets; + // The body lives in ./subflow-guards.ts (#25222). + return subflowTargets(flow); } /** @@ -5005,13 +4853,13 @@ export class AutomationEngine implements IAutomationService { * {@link refuseUnderReachingCallers}'s question, not this scan's. */ private packagedSubflowCallers(name: string): string[] { - const callers: string[] = []; - for (const [callerName, flow] of this.flows) { - if (callerName === name) continue; - if (!this.isPackagedFlow(callerName)) continue; - if (this.subflowTargets(flow).includes(name)) callers.push(callerName); - } - return callers; + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + flows: this.flows, + isPackagedFlow: (...a) => this.isPackagedFlow(...a), + subflowTargets: (...a) => this.subflowTargets(...a), + }; + return packagedSubflowCallers.call(host, name); } /** @@ -5041,15 +4889,12 @@ export class AutomationEngine implements IAutomationService { * A run that has ENDED — completed, failed, refused, cancelled — is not in * either store, so it never counts. */ - private async parkedRunsOf(flowNames: ReadonlySet): Promise> { - const parked = new Map(); - if (flowNames.size === 0) return parked; - for (const run of await this.readSuspendedRuns('throw', flowNames)) { - if (!flowNames.has(run.flowName)) continue; - parked.set(run.flowName, [...(parked.get(run.flowName) ?? []), run.runId]); - } - for (const runIds of parked.values()) runIds.sort(); - return parked; + private parkedRunsOf(flowNames: ReadonlySet): Promise> { + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + readSuspendedRuns: (...a) => this.readSuspendedRuns(...a), + }; + return parkedRunsOf.call(host, flowNames); } /** @@ -5107,59 +4952,11 @@ export class AutomationEngine implements IAutomationService { callers: string[], parked: ReadonlyMap, ): void { - const armed = callers.filter((c) => this.isFlowEnabled(c)); - // Removal only: a switched-off caller guards while `name` is enabled, - // because whether it still holds a parked run is not readable here. - const switchOffFirst = act === 'remove' && this.isFlowEnabled(name); - const unread = switchOffFirst ? callers.filter((c) => !armed.includes(c)) : []; - const guarding = callers.filter((c) => armed.includes(c) || parked.has(c) || unread.includes(c)); - if (guarding.length === 0) return; - - const quote = (names: string[]) => names.map((n) => `'${n}'`).join(', '); - const one = guarding.length === 1; - const steps = [ - ...(armed.length > 0 ? [`disable the calling flow${armed.length === 1 ? '' : 's'} ${quote(armed)} first`] : []), - ...[...parked].map(([caller, runIds]) => { - const single = runIds.length === 1; - return ( - `'${caller}' is disabled but still holds ${single ? 'a parked run' : `${runIds.length} parked runs`} ` + - `that ${single ? 'resumes' : 'resume'} into its subflow node (${quote(runIds)}): cancel ` + - `${single ? 'it' : 'each'} through the operator cancel door, ` + - `POST /automation/${caller}/runs/:runId/cancel (ADR-0044), or let ${single ? 'it' : 'them'} finish` - ); - }), - ...(!switchOffFirst - ? [] - : armed.length > 0 - ? [ - `then switch '${name}' off, which reads whether a switched-off caller still holds a parked ` + - `run and names each one to cancel, and remove it after that`, - ] - : [ - `switch '${name}' off first, which reads whether ${quote(unread)} still ` + - `hold${unread.length === 1 ? 's' : ''} a parked run and names each one to cancel, then remove it`, - ]), - ].join('; '); - const [verb, gerund, keep] = act === 'disable' - ? ['disabled', 'Disabling', 'leave this one armed'] - : ['removed', 'Removing', 'leave this one registered']; - throw Object.assign( - new Error( - `Flow '${name}' cannot be ${verb} while ${guarding.length} packaged flow${one ? '' : 's'} ` + - `still call${one ? 's' : ''} it as a subflow: ${quote(guarding)}. ${gerund} it would break ` + - `${one ? 'that caller' : 'those callers'} mid-run at ${one ? 'its' : 'their'} subflow node with a ` + - `late, inexplicable failure (ADR-0126 §7.3). ` + - `${steps.charAt(0).toUpperCase()}${steps.slice(1)} — or ${keep}.`, - ), - // ADR-0112 envelope: code AND status. `DELETE_RESTRICTED` is the - // standard catalog's "cannot do this due to dependencies" member - // (409) — ⛔ no new ledger entry is minted here. Its `DELETE_` - // prefix fits because the toggle ruling (commit 266436a7f, - // recorded at `isFlowAuthoringWrite` in the runtime's automation - // domain) holds that "disabling a shipped flow is functionally - // equivalent to deleting it for as long as it stays off". - { code: 'DELETE_RESTRICTED', status: 409, subflowCallers: guarding }, - ); + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + isFlowEnabled: (...a) => this.isFlowEnabled(...a), + }; + refuseUnderReachingCallers.call(host, name, act, callers, parked); } /** @@ -5200,21 +4997,17 @@ export class AutomationEngine implements IAutomationService { * returns before asking otherwise), so what that door answers is unchanged. */ private disabledPackagedSubflows(name: string): Array<{ name: string; ledger: boolean; status?: string }> { - const flow = this.flows.get(name); - if (!flow) return []; - const loopCanClose = this.flowLedgerDisabled.has(name); - const disabled: Array<{ name: string; ledger: boolean; status?: string }> = []; - for (const target of new Set(this.subflowTargets(flow))) { - const child = this.flows.get(target); - if (!child || !this.isPackagedFlow(target)) continue; - if (this.isFlowEnabled(target)) continue; - const ledger = this.flowLedgerDisabled.has(target); - const statusDisabled = this.flowStatusDisabled.get(target) === true; - if (loopCanClose && ledger && !statusDisabled && this.ledgerDisabledChainReaches(target, name)) continue; - const status = statusDisabled ? String((child as { status?: unknown }).status) : undefined; - disabled.push({ name: target, ledger, ...(status !== undefined ? { status } : {}) }); - } - return disabled; + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + flows: this.flows, + flowLedgerDisabled: this.flowLedgerDisabled, + subflowTargets: (...a) => this.subflowTargets(...a), + isPackagedFlow: (...a) => this.isPackagedFlow(...a), + isFlowEnabled: (...a) => this.isFlowEnabled(...a), + flowStatusDisabled: this.flowStatusDisabled, + ledgerDisabledChainReaches: (...a) => this.ledgerDisabledChainReaches(...a), + }; + return disabledPackagedSubflows.call(host, name); } /** @@ -5223,21 +5016,14 @@ export class AutomationEngine implements IAutomationService { * {@link disabledPackagedSubflows} must not guard on. */ private ledgerDisabledChainReaches(from: string, to: string): boolean { - const seen = new Set([from]); - const pending = [from]; - while (pending.length > 0) { - const flow = this.flows.get(pending.pop()!); - if (!flow) continue; - for (const target of this.subflowTargets(flow)) { - if (target === to) return true; - if (seen.has(target) || !this.flowLedgerDisabled.has(target)) continue; - const next = this.flows.get(target); - if (!next || !this.isPackagedFlow(target)) continue; - seen.add(target); - pending.push(target); - } - } - return false; + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + flows: this.flows, + subflowTargets: (...a) => this.subflowTargets(...a), + flowLedgerDisabled: this.flowLedgerDisabled, + isPackagedFlow: (...a) => this.isPackagedFlow(...a), + }; + return ledgerDisabledChainReaches.call(host, from, to); } /** @@ -5259,22 +5045,14 @@ export class AutomationEngine implements IAutomationService { * is the opposite act. */ private refuseEnableOntoDisabledSubflow(name: string): void { - if (!this.flowLedgerDisabled.has(name)) return; - if (!this.isPackagedFlow(name)) return; - const disabled = this.disabledPackagedSubflows(name); - if (disabled.length === 0) return; - - const { reasons, remedy } = this.describeDisabledSubflows(disabled, 'this switch'); - const many = disabled.length !== 1; - throw Object.assign( - new Error( - `Flow '${name}' cannot be enabled while ${disabled.length} packaged subflow${many ? 's' : ''} it calls ` + - `${many ? 'are' : 'is'} disabled: ${reasons.join(', ')}. Enabled, it would fail at its subflow node ` + - `on a disabled child — the late, inexplicable failure ADR-0126 §7.3 refuses to cause. ` + - `${remedy.charAt(0).toUpperCase()}${remedy.slice(1)} first, then enable this flow — or leave it disabled.`, - ), - { code: 'RESOURCE_CONFLICT', status: 409 }, - ); + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + flowLedgerDisabled: this.flowLedgerDisabled, + isPackagedFlow: (...a) => this.isPackagedFlow(...a), + disabledPackagedSubflows: (...a) => this.disabledPackagedSubflows(...a), + describeDisabledSubflows: (...a) => this.describeDisabledSubflows(...a), + }; + refuseEnableOntoDisabledSubflow.call(host, name); } /** @@ -5292,32 +5070,8 @@ export class AutomationEngine implements IAutomationService { disabled: Array<{ name: string; ledger: boolean; status?: string }>, switchPhrase: 'this switch' | 'the activation switch', ): { reasons: string[]; remedy: string } { - const quote = (names: string[]) => - names.length === 1 - ? `'${names[0]}'` - : `${names.slice(0, -1).map((n) => `'${n}'`).join(', ')} and '${names[names.length - 1]}'`; - const reasons = disabled.map((d) => { - const why = [ - ...(d.ledger ? ['switched off in the activation ledger'] : []), - ...(d.status !== undefined ? [`its definition's status is '${d.status}'`] : []), - ].join(', and '); - return `'${d.name}' (${why})`; - }); - // The remedy follows each child's REAL state: the switch re-arms the - // ledger bit, and only the definition moves a status. - const byStatus = disabled.filter((d) => d.status !== undefined).map((d) => d.name); - const byLedger = disabled.filter((d) => d.ledger).map((d) => d.name); - const remedy = [ - ...(byStatus.length > 0 - ? [ - `publish ${quote(byStatus)} with status 'active' (${switchPhrase} never changes a definition's ` + - `status; for a package that is read-only here, that takes a package version that ships ` + - `${byStatus.length === 1 ? 'it' : 'them'} active)`, - ] - : []), - ...(byLedger.length > 0 ? [`enable ${quote(byLedger)}`] : []), - ].join(' and '); - return { reasons, remedy }; + // The body lives in ./subflow-guards.ts (#25222). + return describeDisabledSubflows(disabled, switchPhrase); } /** @@ -5387,15 +5141,16 @@ export class AutomationEngine implements IAutomationService { * unbound for a reason of its own is not re-offered here. */ private rejudgeSubflowCallers(subflow: string): void { - if (this.boundFlowTriggers.size === 0 && this.subflowDeclinedFlows.size === 0) return; - for (const caller of this.packagedSubflowCallers(subflow)) { - const settled = this.boundFlowTriggers.has(caller) - ? this.disabledPackagedSubflows(caller).length === 0 - : !this.subflowDeclinedFlows.has(caller); - if (settled) continue; - this.deactivateFlowTrigger(caller); - this.activateFlowTrigger(caller); - } + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + boundFlowTriggers: this.boundFlowTriggers, + subflowDeclinedFlows: this.subflowDeclinedFlows, + packagedSubflowCallers: (...a) => this.packagedSubflowCallers(...a), + disabledPackagedSubflows: (...a) => this.disabledPackagedSubflows(...a), + deactivateFlowTrigger: (...a) => this.deactivateFlowTrigger(...a), + activateFlowTrigger: (...a) => this.activateFlowTrigger(...a), + }; + rejudgeSubflowCallers.call(host, subflow); } /** @@ -5457,25 +5212,12 @@ export class AutomationEngine implements IAutomationService { * rows this door should clear is not this refusal's to decide. */ private refuseCustomerAuthoredToggle(name: string, enabled: boolean): void { - if (this.isPackagedFlow(name)) return; - const updateDoor = `its update door, PUT /automation/${name}, which takes the complete definition`; - const ownSwitch = this.flowLedgerDisabled.has(name) - ? `This flow's own switch is its definition's status, published through ${updateDoor} — but it is ` + - `ALSO held off by an activation-ledger row recorded under the name '${name}' (for a packaged flow of ` + - `that name, or by this switch before it refused customer-authored flows), and no status clears that ` + - `row. To run this flow, clone it under a new name through POST /automation/${name}/clone, which ` + - `arms the copy, and remove this one.` - : `This flow's switch is its own definition's status: publish it through ${updateDoor}, with status ` + - `'obsolete' to switch it off or 'active' to arm it.`; - throw Object.assign( - new Error( - `Flow '${name}' cannot be ${enabled ? 'enabled' : 'disabled'} through this switch: the switch turns ` + - `packaged flows on and off, and '${name}' was authored in this deployment, not shipped by a ` + - `package. The switch records an installation's choice about a packaged flow in the activation ` + - `ledger (sys_metadata_activation, ADR-0126 §7.2). ${ownSwitch} Nothing was changed.`, - ), - { code: 'RESOURCE_CONFLICT', status: 409 }, - ); + // The body lives in ./subflow-guards.ts (#25222). + const host: ThisParameterType = { + isPackagedFlow: (...a) => this.isPackagedFlow(...a), + flowLedgerDisabled: this.flowLedgerDisabled, + }; + refuseCustomerAuthoredToggle.call(host, name, enabled); } /** @@ -5608,16 +5350,13 @@ export class AutomationEngine implements IAutomationService { /** Rollback flow to a specific version */ rollbackFlow(name: string, version: number): void { - const history = this.flowVersionHistory.get(name); - if (!history) { - throw new Error(`Flow '${name}' has no version history`); - } - const entry = history.find(h => h.version === version); - if (!entry) { - throw new Error(`Version ${version} not found for flow '${name}'`); - } - this.flows.set(name, entry.definition); - this.logger.info(`Flow '${name}' rolled back to version ${version}`); + // The body lives in ./flow-definitions.ts (#25222). + const host: ThisParameterType = { + flowVersionHistory: this.flowVersionHistory, + flows: this.flows, + logger: this.logger, + }; + rollbackFlow.call(host, name, version); } listRuns( @@ -8423,11 +8162,8 @@ export class AutomationEngine implements IAutomationService { hot: ConsumedSuspension, rowPause: { nodeId?: string; correlation?: string }, ): boolean { - const samePause = - hot.run.nodeId === rowPause.nodeId && - (hot.run.correlation ?? undefined) === (rowPause.correlation ?? undefined); - if (samePause) return false; - return hot.persisted === 'landed'; + // The body lives in ./consumed-suspension-restore.ts (#25222). + return rowSupersedesJournal(hot, rowPause); } /** @@ -8492,38 +8228,12 @@ export class AutomationEngine implements IAutomationService { /** The hot copy describes a pause the run has since LEFT. */ staleHot: boolean; } { - const hot = this.consumedSuspensions.get(runId); - if (!terminal) { - return hot ? { consumed: hot, witness: 'journal', staleHot: false } : { staleHot: false }; - } - if (terminal.consumedSuspension) { - const durable: ConsumedSuspension = { - run: terminal.consumedSuspension, - consumedAt: terminal.finishedAt ?? terminal.startedAt, - error: terminal.error ?? '', - }; - if (hot && !this.rowSupersedesJournal(hot, durable.run)) { - return { consumed: hot, witness: 'journal', staleHot: false }; - } - return { consumed: durable, witness: 'durable', staleHot: hot !== undefined }; - } - if (terminal.consumedSuspensionDropped) { - const dropped = terminal.consumedSuspensionDropped; - if (hot && !this.rowSupersedesJournal(hot, dropped)) { - return { consumed: hot, witness: 'journal', dropped, staleHot: false }; - } - return { dropped, staleHot: hot !== undefined }; - } - if (hot && hot.persisted !== 'landed') { - // The row predates this strand — this process's own write for it - // never reached the store (in flight, or failed and reported at - // `error`). The store's silence says nothing. - return { consumed: hot, witness: 'journal', staleHot: false }; - } - // A later terminal record with no snapshot and no drop notice: - // completed, cancelled or cascade-failed after any hot copy was - // taken. Stale by definition. - return { staleHot: hot !== undefined }; + // The body lives in ./consumed-suspension-restore.ts (#25222). + const host: ThisParameterType = { + consumedSuspensions: this.consumedSuspensions, + rowSupersedesJournal: (...a) => this.rowSupersedesJournal(...a), + }; + return resolveConsumedSuspensionWitnesses.call(host, runId, terminal); } /** Build a refusal from {@link restoreConsumedSuspension}. */ @@ -8634,50 +8344,16 @@ export class AutomationEngine implements IAutomationService { * @returns Never throws: every outcome — including an unreadable store — is * a {@link SuspensionRestoreResult} naming what was observed. */ - async restoreConsumedSuspension( + restoreConsumedSuspension( runId: string, options?: { requestedBy?: string; reason?: string }, ): Promise { - // [#15222] Resolve the nested chain FIRST, read-only. A flat run finds - // nothing here and takes the per-run call below unchanged — same - // guards, same answer, and a result with no `chain` key at all. - // - // A nested one is the whole of this card: the ancestors' pauses were - // consumed by the same cascade and journalled by - // {@link failSuspendedRun}, and re-arming only the run the operator - // named would put the deployment back in the condition the verb exists - // to leave — a leaf that completes into an ancestor which never - // continues. So the chain is one unit of repair, and this is the call - // that makes it one. - const { below, above } = await this.resolveRestoreChain(runId); - if (below.length === 0 && above.length === 0) { - return this.restoreOneConsumedSuspension(runId, options); - } - // DEEPEST FIRST, and the order IS the guarantee: an ancestor becomes - // resumable only after the run it is parked awaiting is parked again, - // so a resume arriving mid-repair delegates into a live child instead - // of continuing past a nested node whose child is terminal. - // - // Every member goes through the SAME per-run call — its own - // `restoring` claim, its own strict live-suspension read, its own - // two-witness read, its own durable park. So the verb's idempotence - // and the #14333 advance claim hold PER RUN in the chain rather than - // once for the chain, and a second chain restore finds every member - // parked and answers `RUN_SUSPENDED` without minting a second pause - // anywhere. - const chain: ChainRestoreEntry[] = []; - for (const memberId of below) { - chain.push(chainEntryOf(memberId, await this.restoreOneConsumedSuspension(memberId, options, runId))); - } - const own = await this.restoreOneConsumedSuspension(runId, options); - chain.push(chainEntryOf(runId, own)); - for (const memberId of above) { - chain.push(chainEntryOf(memberId, await this.restoreOneConsumedSuspension(memberId, options, runId))); - } - // The six fields above `chain` keep describing the run the caller - // NAMED, wherever in the chain it sits — a consumer that never heard - // of nesting reads exactly what it always read. - return { ...own, chain }; + // The body lives in ./consumed-suspension-restore.ts (#25222). + const host: ThisParameterType = { + resolveRestoreChain: (...a) => this.resolveRestoreChain(...a), + restoreOneConsumedSuspension: (...a) => this.restoreOneConsumedSuspension(...a), + }; + return restoreConsumedSuspension.call(host, runId, options); } /** @@ -8700,33 +8376,12 @@ export class AutomationEngine implements IAutomationService { * has to repair it after a restart, which is the one place this has to * work. */ - private async resolveRestoreChain(runId: string): Promise<{ below: string[]; above: string[] }> { - const below: string[] = []; - const above: string[] = []; - const self = await this.peekConsumedSuspension(runId); - if (!self) return { below, above }; - - let cursor = self; - for (let hops = 0; hops < MAX_RESTORE_CHAIN_HOPS; hops++) { - const childRunId = nestedChildRunId(cursor.run.correlation); - if (childRunId === undefined || childRunId === runId || below.includes(childRunId)) break; - const child = await this.peekConsumedSuspension(childRunId, runId); - if (!child) break; - below.unshift(childRunId); - cursor = child; - } - - cursor = self; - for (let hops = 0; hops < MAX_RESTORE_CHAIN_HOPS; hops++) { - const parentRunId = (cursor.run.context as Record | undefined)?.$parentRunId; - if (typeof parentRunId !== 'string' || !parentRunId) break; - if (parentRunId === runId || above.includes(parentRunId)) break; - const parent = await this.peekConsumedSuspension(parentRunId, runId); - if (!parent) break; - above.push(parentRunId); - cursor = parent; - } - return { below, above }; + private resolveRestoreChain(runId: string): Promise<{ below: string[]; above: string[] }> { + // The body lives in ./consumed-suspension-restore.ts (#25222). + const host: ThisParameterType = { + peekConsumedSuspension: (...a) => this.peekConsumedSuspension(...a), + }; + return resolveRestoreChain.call(host, runId); } /** @@ -8748,28 +8403,16 @@ export class AutomationEngine implements IAutomationService { * identical read immediately afterwards and refuses `STORE_UNAVAILABLE` * on it — one degradation, said once. */ - private async peekConsumedSuspension(runId: string, chainOf?: string): Promise { - try { - if (await this.loadSuspendedRunStrict(runId)) return undefined; - const terminal = this.store?.loadTerminal ? await this.store.loadTerminal(runId) : null; - return this.resolveConsumedSuspensionWitnesses(runId, terminal).consumed; - } catch (err) { - if (chainOf !== undefined) { - // #6299 — the driver's own text rides the structured slot. - // #4632 verdict: FUNCTIONAL — the repair is visibly short (the - // `chain` this call returns names only what it reached) and the - // remedy is to re-issue, which is safe per run. - this.logger.warn( - `[automation] restoreConsumedSuspension('${chainOf}') could not read run '${runId}' while ` + - `walking its nested chain, so the walk STOPPED there and nothing beyond it was re-armed ` + - `— the repair this call reports is partial. Nothing was written for that run. Fix the ` + - `store failure in this record's meta, then re-issue the restore: it is idempotent per ` + - `run and puts back only what is still missing.`, - describeThrownForLog(err), - ); - } - return undefined; - } + private peekConsumedSuspension(runId: string, chainOf?: string): Promise { + // The body lives in ./consumed-suspension-restore.ts (#25222). + const readStore = () => this.store; + const host: ThisParameterType = { + loadSuspendedRunStrict: (...a) => this.loadSuspendedRunStrict(...a), + get store() { return readStore(); }, + resolveConsumedSuspensionWitnesses: (...a) => this.resolveConsumedSuspensionWitnesses(...a), + logger: this.logger, + }; + return peekConsumedSuspension.call(host, runId, chainOf); } /** @@ -9068,28 +8711,15 @@ export class AutomationEngine implements IAutomationService { * {@link hasSuspendedRun}; the restore verb answers `STORE_UNAVAILABLE` * for the same observation. */ - async inspectConsumedSuspension(runId: string): Promise { - // Already resumable? STRICT read — a store outage throws through. - const live = await this.loadSuspendedRunStrict(runId); - if (live) return { repairable: false, runId, reason: 'RUN_SUSPENDED', nodeId: live.nodeId }; - - let terminal: RunRecord | null = null; - if (this.store?.loadTerminal) terminal = await this.store.loadTerminal(runId); - - const { consumed, witness, dropped } = this.resolveConsumedSuspensionWitnesses(runId, terminal); - if (consumed && witness) { - return { - repairable: true, - runId, - flowName: consumed.run.flowName, - nodeId: consumed.run.nodeId, - ...(consumed.run.correlation !== undefined ? { correlation: consumed.run.correlation } : {}), - witness, - consumedAt: consumed.consumedAt, - }; - } - if (dropped) return { repairable: false, runId, reason: 'SNAPSHOT_DROPPED', dropped }; - return { repairable: false, runId, reason: 'NO_CONSUMED_SUSPENSION' }; + inspectConsumedSuspension(runId: string): Promise { + // The body lives in ./consumed-suspension-restore.ts (#25222). + const readStore = () => this.store; + const host: ThisParameterType = { + loadSuspendedRunStrict: (...a) => this.loadSuspendedRunStrict(...a), + get store() { return readStore(); }, + resolveConsumedSuspensionWitnesses: (...a) => this.resolveConsumedSuspensionWitnesses(...a), + }; + return inspectConsumedSuspension.call(host, runId); } /** @@ -9358,11 +8988,8 @@ export class AutomationEngine implements IAutomationService { * failure is hardest to place at run time. */ private unknownNodeTypes(flow: FlowParsed, known: Set): string[] { - return [...new Set( - collectFlowGraphs(flow) - .flatMap(g => g.nodes.map(n => n.type)) - .filter(t => !known.has(t)), - )]; + // The body lives in ./node-text-rendering.ts (#25222). + return unknownNodeTypes(flow, known); } /** @@ -9477,18 +9104,17 @@ export class AutomationEngine implements IAutomationService { * not be counted as one. */ private warnIfNodeTypeVocabularyNeverSealed(): void { - if (this.nodeTypeVocabularySealed || this.nodeTypeSealOmissionWarned) return; - this.nodeTypeSealOmissionWarned = true; - this.logger.warn( - `[automation] flow executed on an engine whose node-type vocabulary was never sealed — ` + - `sealNodeTypeVocabulary() has not been called, so the ADR-0018 node-type check never ran and this ` + - `engine has never reported a flow whose node types nothing has registered; such nodes now fail ` + - `mid-run with NO_EXECUTOR instead of being named at startup. ` + - `A host that constructs AutomationEngine directly must call engine.sealNodeTypeVocabulary() once every ` + - `plugin has contributed its executors (AutomationServicePlugin does this at 'kernel:bootstrapped'); ` + - `engine.getUnknownNodeTypeAudit() returns the same finding without closing the vocabulary. ` + - `Reported once per engine instance.`, - ); + // The body lives in ./node-text-rendering.ts (#25222). + const readNodeTypeVocabularySealed = () => this.nodeTypeVocabularySealed; + const readNodeTypeSealOmissionWarned = () => this.nodeTypeSealOmissionWarned; + const writeNodeTypeSealOmissionWarned = (v: AutomationEngine['nodeTypeSealOmissionWarned']) => { this.nodeTypeSealOmissionWarned = v; }; + const host: ThisParameterType = { + get nodeTypeVocabularySealed() { return readNodeTypeVocabularySealed(); }, + get nodeTypeSealOmissionWarned() { return readNodeTypeSealOmissionWarned(); }, + set nodeTypeSealOmissionWarned(v) { writeNodeTypeSealOmissionWarned(v); }, + logger: this.logger, + }; + warnIfNodeTypeVocabularyNeverSealed.call(host); } /** One warning per flow, shared by the boot audit and the post-seal path. */ @@ -9868,11 +9494,8 @@ export class AutomationEngine implements IAutomationService { * Merges into an existing entry so a node's earlier `output` survives. */ private setNodeError(variables: Map, nodeId: string, message: string): void { - const prior = variables.get(nodeId); - const base = prior && typeof prior === 'object' && !Array.isArray(prior) - ? (prior as Record) - : {}; - variables.set(nodeId, { ...base, error: message }); + // The body lives in ./node-text-rendering.ts (#25222). + setNodeError(variables, nodeId, message); } /** @@ -9893,16 +9516,11 @@ export class AutomationEngine implements IAutomationService { variables: Map, context: AutomationContext, ): string | undefined { - return this.renderFlowTextSlot( - { - key: flowRefusalMessageKey(flowName, nodeId), - authored, - subject: `flow '${flowName}' refused at node '${nodeId}'`, - shownAs: 'the refusal', - }, - variables, - context, - ); + // The body lives in ./node-text-rendering.ts (#25222). + const host: ThisParameterType = { + renderFlowTextSlot: (...a) => this.renderFlowTextSlot(...a), + }; + return renderRefusalMessage.call(host, flowName, nodeId, authored, variables, context); } /** @@ -9947,20 +9565,12 @@ export class AutomationEngine implements IAutomationService { variables: Map, context: AutomationContext, ): string | undefined { - const translated = this.translatedFlowTemplate(slot, context.locale); - if (translated !== undefined) { - try { - return renderTextSlot(translated.template, variables); - } catch (err: unknown) { - if (!(err instanceof FlowTextTemplateError)) throw err; - this.logger.warn( - `[automation] ${slot.subject}: the '${translated.locale}' translation at '${slot.key}' does not ` + - `compile (${err.message}), so ${slot.shownAs} shows its authored text instead. Fix the ` + - `translation's holes to match the authored text.`, - ); - } - } - return renderTextSlot(slot.authored, variables); + // The body lives in ./node-text-rendering.ts (#25222). + const host: ThisParameterType = { + translatedFlowTemplate: (...a) => this.translatedFlowTemplate(...a), + logger: this.logger, + }; + return renderFlowTextSlot.call(host, slot, variables, context); } /** @@ -9972,29 +9582,13 @@ export class AutomationEngine implements IAutomationService { slot: FlowTextSlotTranslation, requested: string | undefined, ): { template: string; locale: string } | undefined { - if (typeof requested !== 'string' || requested.length === 0) return undefined; - let service: RefusalI18nService | undefined; - try { - service = this.i18nServiceSource?.(); - } catch { - return undefined; - } - if (!service || typeof service.t !== 'function') return undefined; - const locale = negotiatedServiceLocale(service, requested); - const { key } = slot; - try { - const template = service.t(key, locale); - // II18nService echoes the key back on a miss. - if (typeof template === 'string' && template.length > 0 && template !== key) { - return { template, locale }; - } - } catch (err: unknown) { - this.logger.warn( - `[automation] ${slot.subject}: the i18n service threw reading '${key}' ` + - `(${err instanceof Error ? err.message : String(err)}), so ${slot.shownAs} shows its authored text.`, - ); - } - return undefined; + // The body lives in ./node-text-rendering.ts (#25222). + const readI18nServiceSource = () => this.i18nServiceSource; + const host: ThisParameterType = { + get i18nServiceSource() { return readI18nServiceSource(); }, + logger: this.logger, + }; + return translatedFlowTemplate.call(host, slot, requested); } /** 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 f51619ac115..97f3dda6e39 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', './run-resume.ts']; + const ENGINE_SOURCES = ['./engine.ts', './flow-registration-checks.ts', './plugin-extension-api.ts', './dag-traversal.ts', './run-history.ts', './run-resume.ts', './dispatch-claims.ts', './flow-definitions.ts', './subflow-guards.ts', './consumed-suspension-restore.ts', './node-text-rendering.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/flow-definitions.ts b/packages/services/service-automation/src/flow-definitions.ts new file mode 100644 index 00000000000..1ecaa6084f8 --- /dev/null +++ b/packages/services/service-automation/src/flow-definitions.ts @@ -0,0 +1,370 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The flow definitions of {@link AutomationEngine}: how a stored or authored + * flow is canonicalised, unregistered, refused and rolled back; how its runtime + * states and trigger bindings are reported; and the function, + * connector-provider and credential surfaces a flow reads. + * + * [#25222] Slice 7 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 FlowDefinitionsHost} + * literal holding exactly the members the body reads. + */ + +import type { ConnectorProviderFactory } from '@objectstack/spec/integration'; +import type { FlowUnboundCause } from '@objectstack/spec/contracts'; +import type { ConversionNotice, ConversionTodoNotice } from '@objectstack/spec'; +import { applyConversionsToFlow } from '@objectstack/spec'; +import { FLOW_HOOK_SECRET_KEY, flowCredentialClassLabel, flowCredentialPositions } from './flow-credential-projection.js'; +import type { FlowParsed } from '@objectstack/spec/automation'; +import { FlowSchema, validateControlFlow } from '@objectstack/spec/automation'; +// The values below (`CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION`, +// `flowRegistrationTodo` and `graftConditionEnvelopes`) 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 { + CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION, + flowRegistrationTodo, + graftConditionEnvelopes, + type AutomationEngine, + type FlowContender, + type FlowCredentialHolding, + type FlowFunctionRegistration, +} 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. Two fields go through a getter, because a method + * reassigns each: `functionResolver` (`setFunctionResolver`) and + * `flowCredentialSource` (`setFlowCredentialSource`). Each read in the body + * reads the engine's current value, as it did through `this`. + */ +export interface FlowDefinitionsHost { + readonly functionResolver: AutomationEngine['functionResolver']; + readonly connectorProviders: AutomationEngine['connectorProviders']; + readonly logger: AutomationEngine['logger']; + readonly canonicalizeFlow: AutomationEngine['canonicalizeFlow']; + readonly registrationCheckRefusals: AutomationEngine['registrationCheckRefusals']; + readonly detectCycles: AutomationEngine['detectCycles']; + readonly flows: AutomationEngine['flows']; + readonly isPackagedFlow: AutomationEngine['isPackagedFlow']; + readonly packagedSubflowCallers: AutomationEngine['packagedSubflowCallers']; + readonly refuseUnderReachingCallers: AutomationEngine['refuseUnderReachingCallers']; + readonly withdrawFlow: AutomationEngine['withdrawFlow']; + readonly flowRefusals: AutomationEngine['flowRefusals']; + readonly resolveTriggerBinding: AutomationEngine['resolveTriggerBinding']; + readonly flowShadowing: AutomationEngine['flowShadowing']; + readonly describeUnboundReason: AutomationEngine['describeUnboundReason']; + readonly isFlowEnabled: AutomationEngine['isFlowEnabled']; + readonly boundFlowTriggers: AutomationEngine['boundFlowTriggers']; + readonly policyDisabledFlows: AutomationEngine['policyDisabledFlows']; + readonly subflowDeclinedFlows: AutomationEngine['subflowDeclinedFlows']; + readonly triggers: AutomationEngine['triggers']; + readonly flowCredentialSource: AutomationEngine['flowCredentialSource']; + readonly flowVersionHistory: AutomationEngine['flowVersionHistory']; +} + +/** The body of {@link AutomationEngine.resolveFunction}, moved unedited; that method's docblock is its contract. */ +export function resolveFunction(this: Pick, name: string): FlowFunctionRegistration | undefined { + const resolved = this.functionResolver?.(name); + if (!resolved) return undefined; + if (typeof resolved === 'function') return { handler: resolved }; + return typeof resolved.handler === 'function' ? resolved : undefined; +} + +/** The body of {@link AutomationEngine.registerConnectorProvider}, moved unedited; that method's docblock is its contract. */ +export function registerConnectorProvider(this: Pick, providerKey: string, factory: ConnectorProviderFactory): void { + if (this.connectorProviders.has(providerKey)) { + this.logger.warn(`Connector provider '${providerKey}' replaced`); + } + this.connectorProviders.set(providerKey, factory); + this.logger.info(`Connector provider registered: ${providerKey}`); +} + +/** The body of {@link AutomationEngine.canonicalizeStoredFlow}, moved unedited; that method's docblock is its contract. */ +export function canonicalizeStoredFlow(this: Pick, name: string, definition: unknown, options: { includeRetired?: boolean } = {}): { + parsed: FlowParsed; + storable: unknown; + notices: ConversionNotice[]; + todos: ConversionTodoNotice[]; +} { + const { parsed, storable, notices, todos: chainTodos } = this.canonicalizeFlow(name, definition, options); + // The registration checks run when `todos` is first READ, once: the + // stored pass reads it, and the doors that only want `parsed` or + // `storable` (the save door, a package duplication, the sync's + // "already held?" comparison) pay nothing for a listing they discard. + const listRegistrationRefusals = (): ConversionTodoNotice[] => { + const todos = [...chainTodos]; + // One listing per site: a site a conversion TODO names (the value-slot + // refusal the spec lists under its D3 entry) keeps that TODO, which + // names the retirement; the registration check's line for it is the + // same refusal and is not repeated. + const listed = new Set(chainTodos.map((todo) => todo.path)); + for (const refusal of this.registrationCheckRefusals(name, parsed)) { + for (const site of refusal.sites) { + if (listed.has(site.path)) continue; + todos.push(flowRegistrationTodo(name, site)); + } + } + return todos; + }; + let listedTodos: ConversionTodoNotice[] | undefined; + return { + parsed, + storable, + notices, + get todos(): ConversionTodoNotice[] { + listedTodos ??= listRegistrationRefusals(); + return listedTodos; + }, + }; +} + +/** The body of {@link AutomationEngine.canonicalizeFlow}, moved unedited; that method's docblock is its contract. */ +export function canonicalizeFlow(this: Pick, name: string, definition: unknown, options: { includeRetired?: boolean } = {}): { + parsed: FlowParsed; + storable: unknown; + notices: ConversionNotice[]; + todos: ConversionTodoNotice[]; +} { + // ADR-0087 D2 — the runtime load seam. A stored flow authored against an + // old shape (a `webhook`/`http_request` callout node, a `delete_record` + // with `config.filters`) is canonicalized on rehydration, BEFORE parse + + // execution, so the executor only ever sees the canonical shape and a + // dropped-alias upgrade never silently changes behavior (e.g. an empty + // `filter` deleting a whole table). No live executor registry is handed + // to the pass: the one open-namespace rename, the callout rename, + // reserves its old names instead (#23543 F5-http), and + // `registerNodeExecutor` refuses an executor under one, so (#24001) the + // conversion-conflict channel it used to feed is gone. + // + // `includeRetired` (#3903): this seam rehydrates DATA AT REST — flows + // stored in `sys_metadata` outlive any load window, and a row written + // under protocol N has no author for a tombstone to teach. When a flow + // conversion graduates to `retiredFromLoadPath` (retiring at N+1), the + // stored flows that still carry the old shape must keep canonicalizing + // here or the retirement would silently change their behavior — the + // exact hazard this seam exists to prevent. Authored sources keep + // window semantics at their own seam (`normalizeStackInput` applies + // live-window entries only; the schema tombstones the retired shape). + // [#23775] The one caller that closes the window is the metadata save + // door, for a flow with no stored row (an authored body, not one at + // rest), through `canonicalizeStoredFlow`'s option; the rehydration + // passes none and replays the full chain, as it always has. + // + // `excludeConversionIds`: the retired window is opened for a CLASS of + // caller, and one kind of entry inside it is not a rescue but a + // reinterpretation — a DEFAULT FLIP, whose old shape still parses and + // now means something else. This seam cannot say a body predates such + // a flip, so it refuses those entries by id, each with its reason on + // `CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION`. + const notices: ConversionNotice[] = []; + const todos: ConversionTodoNotice[] = []; + const converted = applyConversionsToFlow(definition, { + includeRetired: options.includeRetired ?? true, + excludeConversionIds: CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION, + onNotice: (n) => { + notices.push(n); + this.logger.warn(`[flow '${name}'] ${n.code}: ${n.message}`); + }, + // Collected, not logged: a site listed here is one `registerFlow` + // refuses below with its own located message. + onTodo: (t) => { + todos.push(t); + }, + }); + // #4347 / #4415 — one call, canonical at every depth. `FlowNodeSchema` + // parses its own ADR-0031 regions (`FlowNodeSchema.transform` → + // `parseFlowNodeRegions`), so what comes back here is already normalized + // inside `loop.config.body`, `parallel.config.branches[]` and + // `try_catch.config.try`/`.catch` — recursively. Until #4415 that needed + // a second, separately-remembered call to `normalizeControlFlowRegions` + // right here, and every consumer that took a `FlowParsed` without making + // it held a half-parsed flow that looked finished. + const parsed = FlowSchema.parse(converted); + + // DAG cycle detection + this.detectCycles(parsed); + + // ADR-0031 — validate structured control-flow constructs (loop bodies, + // parallel branches, try/catch regions) are well-formed (single-entry/ + // single-exit, acyclic). Reject the malformed before it can run. + validateControlFlow(parsed); + + return { + parsed, + storable: graftConditionEnvelopes(converted, parsed), + notices, + todos, + }; +} + +/** The body of {@link AutomationEngine.unregisterFlow}, moved unedited; that method's docblock is its contract. */ +export function unregisterFlow(this: Pick, name: string): void { + const flow = this.flows.get(name); + if (flow && this.isPackagedFlow(name)) { + const callers = this.packagedSubflowCallers(name); + if (callers.length > 0) this.refuseUnderReachingCallers(name, 'remove', callers, new Map()); + } + this.withdrawFlow(name); +} + +/** The body of {@link AutomationEngine.recordFlowRefusal}, moved unedited; that method's docblock is its contract. */ +export function recordFlowRefusal(this: Pick, name: string, error: unknown): void { + if (this.flows.has(name)) return; + const thrown = (error ?? {}) as { code?: unknown; message?: unknown }; + this.flowRefusals.set(name, { + ...(typeof thrown.code === 'string' ? { code: thrown.code } : {}), + message: typeof thrown.message === 'string' ? thrown.message : String(error), + }); +} + +/** The body of {@link AutomationEngine.getFlowRuntimeStates}, moved unedited; that method's docblock is its contract. */ +export function getFlowRuntimeStates(this: Pick): Array<{ + name: string; + enabled: boolean; + bound: boolean; + status?: string; + triggerType?: string; + object?: string; + reason?: string; + unboundCause?: FlowUnboundCause; + armedFrom?: FlowContender; + shadowed?: FlowContender[]; +}> { + return [...this.flows.keys()].map((name) => { + const resolved = this.resolveTriggerBinding(name); + // [#11997] Attach the shadowing receipt to the row an admin already + // reads. This map holds ONE entry per bare name, so without these + // two fields a displaced contender leaves no trace on this surface. + const shadowing = this.flowShadowing.get(name); + const unbound = this.describeUnboundReason(name, resolved); + return { + name, + enabled: this.isFlowEnabled(name), + bound: this.boundFlowTriggers.has(name), + status: (this.flows.get(name) as { status?: string } | undefined)?.status, + triggerType: resolved?.triggerType, + object: resolved?.binding.object, + // Absent, not `undefined`-valued: a row that is bound, disabled + // or trigger-less has no reason to carry, and `reason: null` on + // the wire would read as "we looked and found nothing". + // [#23842] `unboundCause` rides the SAME spread, so the pair is + // present together or absent together by construction. + ...(unbound === undefined + ? {} + : { reason: unbound.reason, unboundCause: unbound.cause }), + ...(shadowing + ? { armedFrom: shadowing.armed, shadowed: shadowing.shadowed } + : {}), + }; + }); +} + +/** The body of {@link AutomationEngine.describeUnboundReason}, moved unedited; that method's docblock is its contract. */ +export function describeUnboundReason( + this: Pick, + name: string, + resolved: { triggerType: string } | undefined, +): { reason: string; cause: FlowUnboundCause } | undefined { + if (!resolved) return undefined; // manual / screen flow — nothing to bind + if (!this.isFlowEnabled(name)) return undefined; + if (this.boundFlowTriggers.has(name)) return undefined; + const policyReason = this.policyDisabledFlows.get(name); + if (policyReason !== undefined) return { reason: policyReason, cause: 'deployment_policy' }; + // [#20725, ADR-0126 §7.3] The arming gate declined it onto a disabled + // packaged subflow — read from the record, for the policy line's + // reason, and ahead of the trigger branches for the gate's: with the + // child off, the trigger would not arm it either. + const declined = this.subflowDeclinedFlows.get(name); + if (declined !== undefined) return { reason: declined, cause: 'disabled_subflow' }; + return this.triggers.has(resolved.triggerType) + ? { + reason: `trigger '${resolved.triggerType}' is registered but binding failed — see earlier warnings`, + cause: 'binding_failure', + } + : { + reason: `no '${resolved.triggerType}' trigger is registered — add requires: ['triggers'] (record_change/schedule/time_relative/api ship in @objectstack/trigger-*)`, + cause: 'missing_trigger', + }; +} + +/** The body of {@link AutomationEngine.getTriggerBindingAudit}, moved unedited; that method's docblock is its contract. */ +export function getTriggerBindingAudit(this: Pick): Array<{ flowName: string; triggerType: string; reason: string }> { + const audit: Array<{ flowName: string; triggerType: string; reason: string }> = []; + for (const name of this.flows.keys()) { + const resolved = this.resolveTriggerBinding(name); + // [#18235] The eligibility rules (enabled, unbound, declares a + // trigger) and the three-branch vocabulary both live in + // `describeUnboundReason` now, so this audit and the status door + // report the SAME sentence for the same flow by construction. + const unbound = this.describeUnboundReason(name, resolved); + if (unbound === undefined || !resolved) continue; + audit.push({ flowName: name, triggerType: resolved.triggerType, reason: unbound.reason }); + } + return audit; +} + +/** The body of {@link AutomationEngine.flowCredentialHoldings}, moved unedited; that method's docblock is its contract. */ +export function flowCredentialHoldings(this: Pick, name: string): FlowCredentialHolding[] { + const flow = this.flows.get(name); + if (!flow) return []; + const out: FlowCredentialHolding[] = []; + const seen = new Set(); + for (const position of flowCredentialPositions(flow)) { + const id = JSON.stringify([position.nodeId, position.key]); + if (position.form === 'value') { + seen.add(id); + out.push({ nodeId: position.nodeId, key: position.key, label: flowCredentialClassLabel(position.key), held: 'literal' }); + } + } + for (const { nodeId, key } of this.flowCredentialSource?.held(name) ?? []) { + const id = JSON.stringify([nodeId, key]); + if (seen.has(id)) continue; + out.push({ nodeId, key, label: flowCredentialClassLabel(key), held: 'channel' }); + } + return out; +} + +/** The body of {@link AutomationEngine.hookSecretResolver}, moved unedited; that method's docblock is its contract. */ +export function hookSecretResolver( + this: Pick, + flowName: string, + startNodeId: string | undefined, + config: Record, +): (() => Promise) | undefined { + const source = this.flowCredentialSource; + const written = Object.prototype.hasOwnProperty.call(config, FLOW_HOOK_SECRET_KEY); + const literal = typeof config.secret === 'string' && config.secret.trim() !== '' ? config.secret : undefined; + if (written && literal === undefined) return undefined; + if (!source || startNodeId === undefined) { + return literal === undefined ? undefined : async () => literal; + } + if (literal === undefined && !source.holds(flowName, startNodeId, FLOW_HOOK_SECRET_KEY)) return undefined; + return async () => { + if (literal !== undefined && !source.holds(flowName, startNodeId, FLOW_HOOK_SECRET_KEY)) return literal; + return (await source.resolve(flowName, startNodeId, FLOW_HOOK_SECRET_KEY)) ?? literal; + }; +} + +/** The body of {@link AutomationEngine.rollbackFlow}, moved unedited; that method's docblock is its contract. */ +export function rollbackFlow(this: Pick, name: string, version: number): void { + const history = this.flowVersionHistory.get(name); + if (!history) { + throw new Error(`Flow '${name}' has no version history`); + } + const entry = history.find(h => h.version === version); + if (!entry) { + throw new Error(`Version ${version} not found for flow '${name}'`); + } + this.flows.set(name, entry.definition); + this.logger.info(`Flow '${name}' rolled back to version ${version}`); +} diff --git a/packages/services/service-automation/src/node-text-rendering.ts b/packages/services/service-automation/src/node-text-rendering.ts new file mode 100644 index 00000000000..54856c71523 --- /dev/null +++ b/packages/services/service-automation/src/node-text-rendering.ts @@ -0,0 +1,168 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The node-level helpers of {@link AutomationEngine}'s DAG traversal: the + * node-type vocabulary checks a run makes, a failed node's error variable, and + * the rendering of refusal messages and flow text slots. + * + * [#25222] Slice 7 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 NodeTextRenderingHost} + * literal holding exactly the members the body reads. + */ + +import type { AutomationContext } from '@objectstack/spec/contracts'; +import type { FlowParsed } from '@objectstack/spec/automation'; +import { collectFlowGraphs } from '@objectstack/spec/automation'; +// [#15788] The refusing `end` node renders its `message` through the SAME +// interpolator a screen `description` gets — the ruling's own words, so a +// shared function rather than a second template engine. A value import from +// `./builtin/` is safe in this direction: `template.ts` imports only +// `../guard-refusal.js` and package-external contracts, so nothing it pulls in +// reaches back here. +import { renderTextSlot, FlowTextTemplateError } from './builtin/template.js'; +import { flowRefusalMessageKey } from '@objectstack/spec/system'; +// The value below (`negotiatedServiceLocale`) 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 { + negotiatedServiceLocale, + type AutomationEngine, + type FlowTextSlotTranslation, + type RefusalI18nService, +} 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. Two fields go through a getter, because a method + * reassigns each: `nodeTypeVocabularySealed` (`sealNodeTypeVocabulary`) and + * `i18nServiceSource` (`setI18nServiceSource`). Each read in the body reads the + * engine's current value, as it did through `this`. + * `nodeTypeSealOmissionWarned` goes through a get/set pair (#25508), because + * the moved `warnIfNodeTypeVocabularyNeverSealed` body sets it: the write lands + * on the engine, as it did through `this`. + */ +export interface NodeTextRenderingHost { + readonly nodeTypeVocabularySealed: AutomationEngine['nodeTypeVocabularySealed']; + nodeTypeSealOmissionWarned: AutomationEngine['nodeTypeSealOmissionWarned']; + readonly logger: AutomationEngine['logger']; + readonly renderFlowTextSlot: AutomationEngine['renderFlowTextSlot']; + readonly translatedFlowTemplate: AutomationEngine['translatedFlowTemplate']; + readonly i18nServiceSource: AutomationEngine['i18nServiceSource']; +} + +/** The body of {@link AutomationEngine.unknownNodeTypes}, moved unedited; that method's docblock is its contract. */ +export function unknownNodeTypes(flow: FlowParsed, known: Set): string[] { + return [...new Set( + collectFlowGraphs(flow) + .flatMap(g => g.nodes.map(n => n.type)) + .filter(t => !known.has(t)), + )]; +} + +/** The body of {@link AutomationEngine.warnIfNodeTypeVocabularyNeverSealed}, moved unedited; that method's docblock is its contract. */ +export function warnIfNodeTypeVocabularyNeverSealed(this: Pick): void { + if (this.nodeTypeVocabularySealed || this.nodeTypeSealOmissionWarned) return; + this.nodeTypeSealOmissionWarned = true; + this.logger.warn( + `[automation] flow executed on an engine whose node-type vocabulary was never sealed — ` + + `sealNodeTypeVocabulary() has not been called, so the ADR-0018 node-type check never ran and this ` + + `engine has never reported a flow whose node types nothing has registered; such nodes now fail ` + + `mid-run with NO_EXECUTOR instead of being named at startup. ` + + `A host that constructs AutomationEngine directly must call engine.sealNodeTypeVocabulary() once every ` + + `plugin has contributed its executors (AutomationServicePlugin does this at 'kernel:bootstrapped'); ` + + `engine.getUnknownNodeTypeAudit() returns the same finding without closing the vocabulary. ` + + `Reported once per engine instance.`, + ); +} + +/** The body of {@link AutomationEngine.setNodeError}, moved unedited; that method's docblock is its contract. */ +export function setNodeError(variables: Map, nodeId: string, message: string): void { + const prior = variables.get(nodeId); + const base = prior && typeof prior === 'object' && !Array.isArray(prior) + ? (prior as Record) + : {}; + variables.set(nodeId, { ...base, error: message }); +} + +/** The body of {@link AutomationEngine.renderRefusalMessage}, moved unedited; that method's docblock is its contract. */ +export function renderRefusalMessage( + this: Pick, + flowName: string, + nodeId: string, + authored: string | undefined, + variables: Map, + context: AutomationContext, +): string | undefined { + return this.renderFlowTextSlot( + { + key: flowRefusalMessageKey(flowName, nodeId), + authored, + subject: `flow '${flowName}' refused at node '${nodeId}'`, + shownAs: 'the refusal', + }, + variables, + context, + ); +} + +/** The body of {@link AutomationEngine.renderFlowTextSlot}, moved unedited; that method's docblock is its contract. */ +export function renderFlowTextSlot( + this: Pick, + slot: FlowTextSlotTranslation, + variables: Map, + context: AutomationContext, +): string | undefined { + const translated = this.translatedFlowTemplate(slot, context.locale); + if (translated !== undefined) { + try { + return renderTextSlot(translated.template, variables); + } catch (err: unknown) { + if (!(err instanceof FlowTextTemplateError)) throw err; + this.logger.warn( + `[automation] ${slot.subject}: the '${translated.locale}' translation at '${slot.key}' does not ` + + `compile (${err.message}), so ${slot.shownAs} shows its authored text instead. Fix the ` + + `translation's holes to match the authored text.`, + ); + } + } + return renderTextSlot(slot.authored, variables); +} + +/** The body of {@link AutomationEngine.translatedFlowTemplate}, moved unedited; that method's docblock is its contract. */ +export function translatedFlowTemplate( + this: Pick, + slot: FlowTextSlotTranslation, + requested: string | undefined, +): { template: string; locale: string } | undefined { + if (typeof requested !== 'string' || requested.length === 0) return undefined; + let service: RefusalI18nService | undefined; + try { + service = this.i18nServiceSource?.(); + } catch { + return undefined; + } + if (!service || typeof service.t !== 'function') return undefined; + const locale = negotiatedServiceLocale(service, requested); + const { key } = slot; + try { + const template = service.t(key, locale); + // II18nService echoes the key back on a miss. + if (typeof template === 'string' && template.length > 0 && template !== key) { + return { template, locale }; + } + } catch (err: unknown) { + this.logger.warn( + `[automation] ${slot.subject}: the i18n service threw reading '${key}' ` + + `(${err instanceof Error ? err.message : String(err)}), so ${slot.shownAs} shows its authored text.`, + ); + } + return undefined; +} diff --git a/packages/services/service-automation/src/subflow-guards.ts b/packages/services/service-automation/src/subflow-guards.ts new file mode 100644 index 00000000000..d603cf84b60 --- /dev/null +++ b/packages/services/service-automation/src/subflow-guards.ts @@ -0,0 +1,284 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The subflow guards of {@link AutomationEngine}: how a disabled or packaged + * subflow and its callers are found and described, and the refusals that keep + * an enable, a toggle or a caller from reaching a disabled subflow. + * + * [#25222] Slice 7 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 SubflowGuardsHost} + * 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 } from '@objectstack/spec/automation'; +import type { AutomationEngine } 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`. + */ +export interface SubflowGuardsHost { + readonly flowLedgerDisabled: AutomationEngine['flowLedgerDisabled']; + readonly flows: AutomationEngine['flows']; + readonly isPackagedFlow: AutomationEngine['isPackagedFlow']; + readonly subflowTargets: AutomationEngine['subflowTargets']; + readonly readSuspendedRuns: AutomationEngine['readSuspendedRuns']; + readonly isFlowEnabled: AutomationEngine['isFlowEnabled']; + readonly flowStatusDisabled: AutomationEngine['flowStatusDisabled']; + readonly ledgerDisabledChainReaches: AutomationEngine['ledgerDisabledChainReaches']; + readonly disabledPackagedSubflows: AutomationEngine['disabledPackagedSubflows']; + readonly describeDisabledSubflows: AutomationEngine['describeDisabledSubflows']; + readonly boundFlowTriggers: AutomationEngine['boundFlowTriggers']; + readonly subflowDeclinedFlows: AutomationEngine['subflowDeclinedFlows']; + readonly packagedSubflowCallers: AutomationEngine['packagedSubflowCallers']; + readonly deactivateFlowTrigger: AutomationEngine['deactivateFlowTrigger']; + readonly activateFlowTrigger: AutomationEngine['activateFlowTrigger']; +} + +/** The body of {@link AutomationEngine.describeDisabledFlow}, moved unedited; that method's docblock is its contract. */ +export function describeDisabledFlow(this: Pick, flowName: string): string { + if (this.flowLedgerDisabled.has(flowName)) { + return ( + `Flow '${flowName}' is disabled — it is switched off for this installation in the ` + + `packaged-metadata activation ledger (sys_metadata_activation, ADR-0126 §7.2). ` + + `Re-enable the packaged flow to arm it again, or run a clone of it under a new name.` + ); + } + // Unchanged wording for the status dimension: it is what every existing + // consumer and test reads, and this leg has no reason to move it. + return `Flow '${flowName}' is disabled`; +} + +/** The body of {@link AutomationEngine.subflowTargets}, moved unedited; that method's docblock is its contract. */ +export function subflowTargets(flow: FlowParsed): string[] { + const nodes = (flow as { nodes?: unknown }).nodes; + if (!Array.isArray(nodes)) return []; + const targets: string[] = []; + for (const node of nodes) { + const n = node as { type?: unknown; config?: { flowName?: unknown } } | null; + if (!n || (n.type !== 'subflow' && n.type !== 'map')) continue; + if (typeof n.config?.flowName === 'string') targets.push(n.config.flowName); + } + return targets; +} + +/** The body of {@link AutomationEngine.packagedSubflowCallers}, moved unedited; that method's docblock is its contract. */ +export function packagedSubflowCallers(this: Pick, name: string): string[] { + const callers: string[] = []; + for (const [callerName, flow] of this.flows) { + if (callerName === name) continue; + if (!this.isPackagedFlow(callerName)) continue; + if (this.subflowTargets(flow).includes(name)) callers.push(callerName); + } + return callers; +} + +/** The body of {@link AutomationEngine.parkedRunsOf}, moved unedited; that method's docblock is its contract. */ +export async function parkedRunsOf(this: Pick, flowNames: ReadonlySet): Promise> { + const parked = new Map(); + if (flowNames.size === 0) return parked; + for (const run of await this.readSuspendedRuns('throw', flowNames)) { + if (!flowNames.has(run.flowName)) continue; + parked.set(run.flowName, [...(parked.get(run.flowName) ?? []), run.runId]); + } + for (const runIds of parked.values()) runIds.sort(); + return parked; +} + +/** The body of {@link AutomationEngine.refuseUnderReachingCallers}, moved unedited; that method's docblock is its contract. */ +export function refuseUnderReachingCallers( + this: Pick, + name: string, + act: 'disable' | 'remove', + callers: string[], + parked: ReadonlyMap, +): void { + const armed = callers.filter((c) => this.isFlowEnabled(c)); + // Removal only: a switched-off caller guards while `name` is enabled, + // because whether it still holds a parked run is not readable here. + const switchOffFirst = act === 'remove' && this.isFlowEnabled(name); + const unread = switchOffFirst ? callers.filter((c) => !armed.includes(c)) : []; + const guarding = callers.filter((c) => armed.includes(c) || parked.has(c) || unread.includes(c)); + if (guarding.length === 0) return; + + const quote = (names: string[]) => names.map((n) => `'${n}'`).join(', '); + const one = guarding.length === 1; + const steps = [ + ...(armed.length > 0 ? [`disable the calling flow${armed.length === 1 ? '' : 's'} ${quote(armed)} first`] : []), + ...[...parked].map(([caller, runIds]) => { + const single = runIds.length === 1; + return ( + `'${caller}' is disabled but still holds ${single ? 'a parked run' : `${runIds.length} parked runs`} ` + + `that ${single ? 'resumes' : 'resume'} into its subflow node (${quote(runIds)}): cancel ` + + `${single ? 'it' : 'each'} through the operator cancel door, ` + + `POST /automation/${caller}/runs/:runId/cancel (ADR-0044), or let ${single ? 'it' : 'them'} finish` + ); + }), + ...(!switchOffFirst + ? [] + : armed.length > 0 + ? [ + `then switch '${name}' off, which reads whether a switched-off caller still holds a parked ` + + `run and names each one to cancel, and remove it after that`, + ] + : [ + `switch '${name}' off first, which reads whether ${quote(unread)} still ` + + `hold${unread.length === 1 ? 's' : ''} a parked run and names each one to cancel, then remove it`, + ]), + ].join('; '); + const [verb, gerund, keep] = act === 'disable' + ? ['disabled', 'Disabling', 'leave this one armed'] + : ['removed', 'Removing', 'leave this one registered']; + throw Object.assign( + new Error( + `Flow '${name}' cannot be ${verb} while ${guarding.length} packaged flow${one ? '' : 's'} ` + + `still call${one ? 's' : ''} it as a subflow: ${quote(guarding)}. ${gerund} it would break ` + + `${one ? 'that caller' : 'those callers'} mid-run at ${one ? 'its' : 'their'} subflow node with a ` + + `late, inexplicable failure (ADR-0126 §7.3). ` + + `${steps.charAt(0).toUpperCase()}${steps.slice(1)} — or ${keep}.`, + ), + // ADR-0112 envelope: code AND status. `DELETE_RESTRICTED` is the + // standard catalog's "cannot do this due to dependencies" member + // (409) — ⛔ no new ledger entry is minted here. Its `DELETE_` + // prefix fits because the toggle ruling (commit 266436a7f, + // recorded at `isFlowAuthoringWrite` in the runtime's automation + // domain) holds that "disabling a shipped flow is functionally + // equivalent to deleting it for as long as it stays off". + { code: 'DELETE_RESTRICTED', status: 409, subflowCallers: guarding }, + ); +} + +/** The body of {@link AutomationEngine.disabledPackagedSubflows}, moved unedited; that method's docblock is its contract. */ +export function disabledPackagedSubflows(this: Pick, name: string): Array<{ name: string; ledger: boolean; status?: string }> { + const flow = this.flows.get(name); + if (!flow) return []; + const loopCanClose = this.flowLedgerDisabled.has(name); + const disabled: Array<{ name: string; ledger: boolean; status?: string }> = []; + for (const target of new Set(this.subflowTargets(flow))) { + const child = this.flows.get(target); + if (!child || !this.isPackagedFlow(target)) continue; + if (this.isFlowEnabled(target)) continue; + const ledger = this.flowLedgerDisabled.has(target); + const statusDisabled = this.flowStatusDisabled.get(target) === true; + if (loopCanClose && ledger && !statusDisabled && this.ledgerDisabledChainReaches(target, name)) continue; + const status = statusDisabled ? String((child as { status?: unknown }).status) : undefined; + disabled.push({ name: target, ledger, ...(status !== undefined ? { status } : {}) }); + } + return disabled; +} + +/** The body of {@link AutomationEngine.ledgerDisabledChainReaches}, moved unedited; that method's docblock is its contract. */ +export function ledgerDisabledChainReaches(this: Pick, from: string, to: string): boolean { + const seen = new Set([from]); + const pending = [from]; + while (pending.length > 0) { + const flow = this.flows.get(pending.pop()!); + if (!flow) continue; + for (const target of this.subflowTargets(flow)) { + if (target === to) return true; + if (seen.has(target) || !this.flowLedgerDisabled.has(target)) continue; + const next = this.flows.get(target); + if (!next || !this.isPackagedFlow(target)) continue; + seen.add(target); + pending.push(target); + } + } + return false; +} + +/** The body of {@link AutomationEngine.refuseEnableOntoDisabledSubflow}, moved unedited; that method's docblock is its contract. */ +export function refuseEnableOntoDisabledSubflow(this: Pick, name: string): void { + if (!this.flowLedgerDisabled.has(name)) return; + if (!this.isPackagedFlow(name)) return; + const disabled = this.disabledPackagedSubflows(name); + if (disabled.length === 0) return; + + const { reasons, remedy } = this.describeDisabledSubflows(disabled, 'this switch'); + const many = disabled.length !== 1; + throw Object.assign( + new Error( + `Flow '${name}' cannot be enabled while ${disabled.length} packaged subflow${many ? 's' : ''} it calls ` + + `${many ? 'are' : 'is'} disabled: ${reasons.join(', ')}. Enabled, it would fail at its subflow node ` + + `on a disabled child — the late, inexplicable failure ADR-0126 §7.3 refuses to cause. ` + + `${remedy.charAt(0).toUpperCase()}${remedy.slice(1)} first, then enable this flow — or leave it disabled.`, + ), + { code: 'RESOURCE_CONFLICT', status: 409 }, + ); +} + +/** The body of {@link AutomationEngine.describeDisabledSubflows}, moved unedited; that method's docblock is its contract. */ +export function describeDisabledSubflows( + disabled: Array<{ name: string; ledger: boolean; status?: string }>, + switchPhrase: 'this switch' | 'the activation switch', +): { reasons: string[]; remedy: string } { + const quote = (names: string[]) => + names.length === 1 + ? `'${names[0]}'` + : `${names.slice(0, -1).map((n) => `'${n}'`).join(', ')} and '${names[names.length - 1]}'`; + const reasons = disabled.map((d) => { + const why = [ + ...(d.ledger ? ['switched off in the activation ledger'] : []), + ...(d.status !== undefined ? [`its definition's status is '${d.status}'`] : []), + ].join(', and '); + return `'${d.name}' (${why})`; + }); + // The remedy follows each child's REAL state: the switch re-arms the + // ledger bit, and only the definition moves a status. + const byStatus = disabled.filter((d) => d.status !== undefined).map((d) => d.name); + const byLedger = disabled.filter((d) => d.ledger).map((d) => d.name); + const remedy = [ + ...(byStatus.length > 0 + ? [ + `publish ${quote(byStatus)} with status 'active' (${switchPhrase} never changes a definition's ` + + `status; for a package that is read-only here, that takes a package version that ships ` + + `${byStatus.length === 1 ? 'it' : 'them'} active)`, + ] + : []), + ...(byLedger.length > 0 ? [`enable ${quote(byLedger)}`] : []), + ].join(' and '); + return { reasons, remedy }; +} + +/** The body of {@link AutomationEngine.rejudgeSubflowCallers}, moved unedited; that method's docblock is its contract. */ +export function rejudgeSubflowCallers(this: Pick, subflow: string): void { + if (this.boundFlowTriggers.size === 0 && this.subflowDeclinedFlows.size === 0) return; + for (const caller of this.packagedSubflowCallers(subflow)) { + const settled = this.boundFlowTriggers.has(caller) + ? this.disabledPackagedSubflows(caller).length === 0 + : !this.subflowDeclinedFlows.has(caller); + if (settled) continue; + this.deactivateFlowTrigger(caller); + this.activateFlowTrigger(caller); + } +} + +/** The body of {@link AutomationEngine.refuseCustomerAuthoredToggle}, moved unedited; that method's docblock is its contract. */ +export function refuseCustomerAuthoredToggle(this: Pick, name: string, enabled: boolean): void { + if (this.isPackagedFlow(name)) return; + const updateDoor = `its update door, PUT /automation/${name}, which takes the complete definition`; + const ownSwitch = this.flowLedgerDisabled.has(name) + ? `This flow's own switch is its definition's status, published through ${updateDoor} — but it is ` + + `ALSO held off by an activation-ledger row recorded under the name '${name}' (for a packaged flow of ` + + `that name, or by this switch before it refused customer-authored flows), and no status clears that ` + + `row. To run this flow, clone it under a new name through POST /automation/${name}/clone, which ` + + `arms the copy, and remove this one.` + : `This flow's switch is its own definition's status: publish it through ${updateDoor}, with status ` + + `'obsolete' to switch it off or 'active' to arm it.`; + throw Object.assign( + new Error( + `Flow '${name}' cannot be ${enabled ? 'enabled' : 'disabled'} through this switch: the switch turns ` + + `packaged flows on and off, and '${name}' was authored in this deployment, not shipped by a ` + + `package. The switch records an installation's choice about a packaged flow in the activation ` + + `ledger (sys_metadata_activation, ADR-0126 §7.2). ${ownSwitch} Nothing was changed.`, + ), + { code: 'RESOURCE_CONFLICT', status: 409 }, + ); +}