diff --git a/packages/services/service-automation/src/dag-traversal.ts b/packages/services/service-automation/src/dag-traversal.ts new file mode 100644 index 00000000000..da761e0a1ed --- /dev/null +++ b/packages/services/service-automation/src/dag-traversal.ts @@ -0,0 +1,1058 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The DAG traversal of {@link AutomationEngine}: how a run executes a node, + * follows its out-edges, runs a region, and evaluates a condition or a value + * envelope. + * + * [#25222] Slice 3 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 DagTraversalHost} + * literal holding exactly the members the body reads; a body that reads no + * member is called directly. An async method's stub is no longer `async`: it + * returns the promise the moved function makes. + */ + +import type { FlowParsed, FlowNodeParsed, FlowEdgeParsed, FlowRegionParsed } from '@objectstack/spec/automation'; +import { findRegionEntry } from '@objectstack/spec/automation'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import { isGuardRefusal } from './guard-refusal.js'; +import { attachCarriedRefusal } from './inner-refusal.js'; +// The values below (`AutomationEngine` for its public static +// `MAX_NODE_REENTRIES`) 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 { + AutomationEngine, + DEFAULT_BRANCH_LABEL, + FlowRefusalSignal, + FlowSuspendSignal, + isRefusalSignal, + isSuspendSignal, + type NodeExecutionResult, + type StepLogEntry, +} from './engine.js'; +import { createEvalUser, type EvalUserParsed } from '@objectstack/spec/identity'; +import { structuralConditionRefusal } from '@objectstack/spec/automation'; +// [#15137] The `value`-role half of the ledger. Both halves of "is this envelope +// well-formed?" are IMPORTED, never re-spelled here: the shape rule is +// `FlowValueSlotSchema` (spec, #14149; slot-neutral since #19938 — it refuses a +// non-`cel` dialect and the `{ dialect: 'cel' }` with no `source` that +// `validateExpression` reads as "not authored"), and the CEL rule is +// `validateExpression('value', …)` (formula). `registerFlow` and the run-time +// evaluator call the SAME composition +// ({@link AutomationEngine.valueEnvelopeRefusals}), so a flow that registers can +// never be refused at run time and vice versa — for every value slot the ledger +// declares (`assignment.assignments.*`, `create_record` / `update_record` +// `fields.*`). +import { FlowValueSlotSchema, VALUE_ENVELOPE_REFUSAL } from '@objectstack/spec/automation'; +// Static import (not a lazy `require`): the engine ships as ESM ("type":"module"), +// where a CommonJS `require('@objectstack/formula')` resolves to tsup's throwing +// `__require` stub. That threw on every CEL evaluation and the catch below +// silently returned `false`, so EVERY start-node / edge condition (record-change +// `previous.*`, `budget > 100000`, …) skipped its flow. A static import binds the +// engine at module load in both ESM and CJS builds. +import { ExpressionEngine, validateExpression } from '@objectstack/formula'; +import { readPartialSteps } from './partial-steps.js'; +import { isRegionSuspensionRefusal, refuseRegionSuspension } from './region-suspension-refusal.js'; + +/** + * The value the flow CEL scope binds as `current_user` for a run (#19939 — + * the maintainer's ruling on its second pass, Q1 A and Q2 A): the run's + * `EvalUser` when the run has a user, else `null`. + * + * Built through `createEvalUser`, the one factory every surface builds the + * canonical user with (ADR-0068), from what the run context holds and + * nothing else: `id` from `userId`, `positions`, and `organizationId` from + * `tenantId`; `isPlatformAdmin` is derived from the positions there. Neither + * an email nor a name is bound (Q2 A enumerates the four above). A + * `runAs:'user'` run's context does carry the user's email + * (`AutomationContext.email`, seeded at run setup), but for its data + * operations' row policies, which read it as `current_user.email` through + * `resolveRunDataContext`; it is not a CEL fact here, and a `runAs:'system'` + * run carries none. A flow that needs either reads the user record + * (`current_user.id`). + * + * A run with no user — a schedule, a record change made by a system write, a + * caller that passed no context — answers `null`, never a pseudo-user + * (ADR-0118 D1, D4): `current_user.id` then fails the run loudly, and the + * author's guard `current_user != null ? current_user.id : null` can be + * written. + */ +export function runUserOf(context: AutomationContext | undefined): EvalUserParsed | null { + const userId = context?.userId; + if (typeof userId !== 'string' || userId === '') return null; + return createEvalUser({ + id: userId, + positions: Array.isArray(context?.positions) ? context.positions : [], + ...(typeof context?.tenantId === 'string' ? { organizationId: context.tenantId } : {}), + }); +} + +/** + * Does this `decision` take EVERY out-edge whose condition holds (#15429)? + * Only an explicit `config.mode: 'inclusive'` says so; an omitted `mode` and + * `'exclusive'` both mean the first true edge in declaration order wins. Any + * other value never reaches here — {@link AutomationEngine.registerFlow} + * refuses it with the spec's prescription. + */ +function decisionTakesEveryBranch(node: FlowNodeParsed): boolean { + const config = node.config as { mode?: unknown } | undefined; + return config?.mode === 'inclusive'; +} + +/** + * 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. Fields go by value: each + * is assigned once, at construction, and never reassigned, so the value passed + * is the field, including where an async body reads it after an `await`. + */ +export interface DagTraversalHost { + readonly renderRefusalMessage: AutomationEngine['renderRefusalMessage']; + readonly nodeExecutors: AutomationEngine['nodeExecutors']; + readonly executeWithTimeout: AutomationEngine['executeWithTimeout']; + readonly setNodeError: AutomationEngine['setNodeError']; + readonly executeNode: AutomationEngine['executeNode']; + readonly refuseUndeclaredSuspension: AutomationEngine['refuseUndeclaredSuspension']; + readonly traverseNext: AutomationEngine['traverseNext']; + readonly logger: AutomationEngine['logger']; + readonly evaluateCondition: AutomationEngine['evaluateCondition']; + readonly valueEnvelopeRefusals: AutomationEngine['valueEnvelopeRefusals']; + readonly celScope: AutomationEngine['celScope']; +} + +/** The body of {@link AutomationEngine.executeNode}, moved unedited; that method's docblock is its contract. */ +export async function executeNode( + this: Pick, + node: FlowNodeParsed, + flow: FlowParsed, + variables: Map, + context: AutomationContext, + steps: StepLogEntry[], +): Promise { + if (node.type === 'end') { + // [#15788] …unless the author declared a REFUSAL here (#14945 + // ruling 2′). `end` has no executor and no descriptor — it is + // structural (`FLOW_STRUCTURAL_NODE_TYPES`), which is why this + // method opens by returning on it — so this is the one place the + // outcome can be read, and until now nothing read it: a flow + // declaring `outcome: 'refused'` ran as a plain completion and the + // author's reason reached nobody. + // + // The config arrives PARSED. `EndConfigSchema` is applied by + // `FlowNodeSchema`'s own transform (`parseEndNodeConfig`, spec + // lane 1), which runs on every node including the ones nested in a + // region, so `outcome` is defaulted and a `refused` without a + // `message` was already refused at the flow parse. ⛔ No second + // door, no `??` default, no re-parse: this reads a contract that is + // already enforced rather than defending against it. + const endConfig = node.config as { outcome?: string; message?: string } | undefined; + if (endConfig?.outcome === 'refused') { + throw new FlowRefusalSignal( + node.id, + // The SAME rendering a screen `description` gets — the + // ruling's words, one implementation (`renderTextSlot`, the + // `{{ }}` text renderer since #22110). + // Rendered HERE, against the live variable map, because + // that is what makes the text per-record; a template on the + // wire would put the rendering in every runner. + // [#22450] …in the run's language: the TEMPLATE is picked + // before it is rendered, because the run stores the + // rendered text and nothing downstream can translate it. + this.renderRefusalMessage(flow.name, node.id, endConfig.message, variables, context), + ); + } + return; + } + + // ADR-0044 runaway guard: declared back-edges make re-entering a node + // legal, so a misauthored unconditional loop could otherwise spin + // forever. Count this node's prior *top-level* visits in the run's step + // log (region body steps carry `parentNodeId` and are excluded — a + // 200-iteration `loop` region is legitimate) and fail the run loudly + // past the cap. Product-level guards (e.g. an approval node's + // `maxRevisions`) terminate far earlier; this is the engine backstop. + // + // A `skipped` step is NOT a visit (#4354): those entries record a gate + // that closed in FRONT of a node, so counting them would let a flow whose + // gate refuses often abort itself as a runaway — a new observability + // signal changing execution semantics, which it must never do. + const priorVisits = steps.reduce( + (n, s) => + s.nodeId === node.id && s.parentNodeId === undefined && s.status !== 'skipped' + ? n + 1 + : n, + 0, + ); + if (priorVisits >= AutomationEngine.MAX_NODE_REENTRIES) { + throw new Error( + `Node '${node.id}' was entered ${priorVisits} times in one run — aborting as a runaway loop ` + + `(back-edge cycles must terminate; see ADR-0044)`, + ); + } + + const stepStart = Date.now(); + const stepStartedAt = new Date().toISOString(); + + // Find executor + const executor = this.nodeExecutors.get(node.type); + if (!executor) { + // start node without executor is fine — just skip + if (node.type !== 'start') { + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'failure', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + error: { code: 'NO_EXECUTOR', message: `No executor registered for node type '${node.type}'` }, + }); + throw new Error(`No executor registered for node type '${node.type}'`); + } + // Log start node step + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'success', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + }); + } else { + // Execute node with optional timeout + let result: NodeExecutionResult; + try { + if (node.timeoutMs && node.timeoutMs > 0) { + result = await this.executeWithTimeout( + executor.execute(node, variables, context), + node.timeoutMs, + node.id, + ); + } else { + result = await executor.execute(node, variables, context); + } + } catch (execErr: unknown) { + const errMsg = execErr instanceof Error ? execErr.message : String(execErr); + // [#18881] ONE region refusal is ONE failure. The refusal names + // the region node whose body could not carry the pause, and + // that node's own frame records it exactly as any other thrown + // failure does. Every ENCLOSING container the unwind passes + // through — the `loop` around the `try_catch` in the card's + // reproduction — records nothing: it did not fail, it is the + // frame a failure is travelling out through, and a step for it + // would make `summary.failed` count the NESTING DEPTH rather + // than the fault. `summary.failed` is `Σ nodes[].failures` + // (#14456), so a second step here reads as a second lost row to + // every operator and every #4354 reader. + // + // Keyed on the refusal's OWN `regionNodeId` rather than on a + // mutable "already reported" flag: the identity is decided once + // at the boundary that raised it and cannot drift as the error + // travels. + const enclosingFrameOfRegionRefusal = + isRegionSuspensionRefusal(execErr) && execErr.regionNodeId !== node.id; + if (!enclosingFrameOfRegionRefusal) { + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'failure', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + error: { code: 'EXECUTION_ERROR', message: errMsg }, + }); + } + + // #13803 — a structured container that DIED mid-body still did + // whatever its completed iterations did, and those writes are + // already in the database. Fold the body steps it carried out + // on the throw into the run log, in the same position the + // success path splices `childSteps`: right behind the + // container's own step, ahead of any `fault` handler's steps. + // + // Without this the record was lost twice over — the run log + // kept no trace of the completed iterations, and the #4354 + // summary folded over that log reported `acted: 0` for a sweep + // that had genuinely written rows. `acted: 0` on a failed run + // reads as "nothing happened, safe to re-run", so the summary + // was wrong in the one direction that causes double-execution. + // + // Purely additive to the record: the error is rethrown below + // untouched, so which failures route, what `$error` holds and + // what the run reports as its error are all unchanged. Every + // step folded here carries a `parentNodeId` (set by + // `runRegion`'s tagger), so the ADR-0044 runaway guard — which + // counts only top-level visits — cannot see them either. + const carriedSteps = readPartialSteps(execErr); + if (carriedSteps?.length) { + steps.push(...carriedSteps); + } + + // #14955 — publish the failure BEFORE deciding whether it + // routes, exactly as the returned-failure arm below does. + // + // This write used to sit inside `if (faultEdge)`, and no reason + // for that was ever recorded: both arms were written in one + // commit, and only this one conflated "publish the failure" with + // "route the failure" — the arm below already separated them. + // The consequence was invisible rather than loud. A node inside a + // region never has a `fault` edge of its own (the region's + // synthetic sub-flow carries only the region's own edges — see + // {@link AutomationEngine.runRegion}), so EVERY thrown failure + // inside a region left `$error` naming an earlier, unrelated + // failure. The first reader that cared about `$error`'s freshness + // — `try_catch`'s `code` binding (commit c5a7448d5) — met it immediately + // and bound a message and a code that came from two different + // failures: `{ code: 'DUPLICATE_RECORD', message: "Node 'mk' + // timed out after 20ms" }`, swallowed by a catch region reading + // it as "the row is already there", the run reporting success. + // + // `{ nodeId, message }` and nothing more: there is no + // `NodeExecutionResult` on this path, so no `output` and no + // classified `code` exist to carry — and that absence is the + // correct answer for a throw, not a reason to leave a stale + // `code` standing. This is what the documented contract already + // promised — `{$error}` names the most recent failure + // (`content/docs/automation/flows.mdx`) — and now holds. + // + // Publishing is not routing: the guard-refusal rule below is + // untouched and still decides, alone, which failures a `fault` + // edge may carry. Nor is the thrown value touched — `execErr` is + // rethrown below exactly as caught. + // + // [#18881] …and it is published for the SAME frames that record + // a step, for the same reason: an enclosing container the + // region refusal is travelling out through did not fail, so + // `{$error}` naming it would be a false sentence about which + // node produced the run's failure. The region node's own frame + // publishes, as any failing node does. + if (!enclosingFrameOfRegionRefusal) { + variables.set('$error', { nodeId: node.id, message: errMsg }); + this.setNodeError(variables, node.id, errMsg); + } + + // #3863 — a guard that THROWS is as un-routable as one that + // returns: `UnscopedRunDataAccessError` (ADR-0049/#1888) reports + // that the metadata would run unscoped, and rerouting it would + // let a `fault` edge disable the elevation check. + const faultEdge = isGuardRefusal(execErr) + ? undefined + : flow.edges.find(e => e.source === node.id && e.type === 'fault'); + if (faultEdge) { + const faultTarget = flow.nodes.find(n => n.id === faultEdge.target); + if (faultTarget) { + await this.executeNode(faultTarget, flow, variables, context, steps); + return; + } + } + throw execErr; + } + + // #6667 — declared = enforced for `supportsPause`, at the ONE seam + // every suspension passes through. + // + // Placed here rather than beside the `throw new FlowSuspendSignal` + // below on purpose: converting the mismatch into an ordinary guard + // refusal *before* the success bookkeeping means the run records a + // `failure` step for the offending node (not a `success` step + // followed by an unexplained failed run), sets `$error` like any + // other refusal, and inherits #3863's un-routability — a `fault` + // edge must not be able to swallow a declaration defect, since + // re-running the flow unchanged can never fix one. + // + // Exactly once, and nothing bypasses it: this is the only call site + // of any `executor.execute()` that the engine acts on — the ADR-0018 + // alias path delegates and RETURNS its target's result here rather + // than suspending on its own, `resume()` re-enters through + // {@link executeNode}, and region bodies ({@link runRegion}) do too. + if (result.success && result.suspend === true) { + result = this.refuseUndeclaredSuspension(node.type) ?? result; + } + + if (!result.success) { + const errMsg = result.error ?? 'Unknown error'; + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'failure', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + error: { code: 'NODE_FAILURE', message: errMsg }, + ...(result.warnings?.length ? { warnings: result.warnings } : {}), + // #4354 — a node that failed PART WAY may still have written + // rows; dropping its counts would understate what the run did. + ...(result.metrics ? { metrics: result.metrics } : {}), + }); + + // #14184 — the returned-failure half of #13803's fold. A + // container that RETURNS its failure (rather than throwing it) + // never reached the splice below, because that one runs only + // after a successful result. `try_catch` with no `catch` region + // is exactly that shape: it catches the try region's failure and + // returns it, so its completed try-region steps were dropped on + // the floor and the #4354 summary reported `acted: 0` over rows + // that had genuinely been written. + // + // Same position as the throw arm's fold and the success path's + // splice: right behind the container's own step, ahead of any + // `fault` handler's steps. #13803 left this branch alone because + // no executor then returned `childSteps` on a failing result; + // `try_catch` is that producer, and it is still the only one. + // + // Additive to the record only — the failure step above, the + // `$error` write below, the guard-refusal routing rule and the + // rethrow are all untouched. Every folded step carries a + // `parentNodeId` (set by `runRegion`'s tagger), so the ADR-0044 + // runaway guard, which counts only top-level visits, cannot see + // them either. + if (result.childSteps?.length) { + steps.push(...result.childSteps); + } + + // Write error output to variable context for downstream nodes. + // Commit c5a7448d5 — carry the executor's classified `code` (when it set + // one) alongside `message`, so a `fault` edge handler reading + // `{$error.code}` can branch on it; a `try_catch` region reads + // this same node-level `$error` before its OWN `errorVariable` + // write can shadow it (see try-catch-node.ts). + variables.set('$error', { + nodeId: node.id, + message: errMsg, + output: result.output, + ...(result.code ? { code: result.code } : {}), + }); + this.setNodeError(variables, node.id, errMsg); + + // #3863 — only a `runtime` failure may be routed. A `guard` + // refusal says the METADATA is wrong (#3810 erased a filter + // condition, a data node names no object); rerouting it would + // make a single `fault` edge a switch that turns the guard off + // while the run still reports success. + const faultEdge = result.errorClass === 'guard' + ? undefined + : flow.edges.find(e => e.source === node.id && e.type === 'fault'); + if (faultEdge) { + const faultTarget = flow.nodes.find(n => n.id === faultEdge.target); + if (faultTarget) { + await this.executeNode(faultTarget, flow, variables, context, steps); + return; + } + } + // [#24914, ruling B′] The text is unchanged — it is the run's + // diagnostic, read by the run log, `failAncestors`, approvals + // and every door's `message`. What the node's data call met is + // carried BESIDE it, on a brand (`inner-refusal.ts`), to the + // terminal exit that builds the run's result. + throw attachCarriedRefusal(new Error(`Node '${node.id}' failed: ${errMsg}`), result); + } + + // Log successful step (#3407: advisory executor warnings ride along + // so a legal-but-partial outcome never reads as a clean success; + // #4354: the executor's own record counts ride along too, so the run + // summary can tell a sweep that had nothing to do from one that did + // nothing). + steps.push({ + nodeId: node.id, + nodeType: node.type, + status: 'success', + startedAt: stepStartedAt, + completedAt: new Date().toISOString(), + durationMs: Date.now() - stepStart, + ...(result.warnings?.length ? { warnings: result.warnings } : {}), + ...(result.metrics ? { metrics: result.metrics } : {}), + }); + + // #1479: fold a structured-region container's body/branch/handler + // steps into the run log, right after the container's own step. + if (result.childSteps?.length) { + steps.push(...result.childSteps); + } + + // Write back output variables + if (result.output) { + for (const [key, value] of Object.entries(result.output)) { + variables.set(`${node.id}.${key}`, value); + } + } + + // [#18110 / #18555] Terminal refusal: the node evaluated and the + // answer is no. Thrown from HERE — the position the suspend signal + // below is thrown from — and that position is the point of the + // design, not a convenience: the node's success step is already + // pushed, its `childSteps` are already folded and its output is + // already written back, so a refusing `subflow` / `map` keeps the + // child's #4354 rollup (`selected` / `acted` / `unmeasuredEffect`) + // in the run summary. An unwind that began any earlier would drop + // exactly those counts — a refusing child really can have written + // rows before it said no. + // + // The step stays a SUCCESS on purpose: the node did evaluate, and + // what it evaluated to is the run's outcome, not this step's. The + // three terminal exits (`execute` / `resumeInternal` / + // `executeWithoutRetry`) already convert the signal into a + // `refused` run through the one `finishRefusedRun` chokepoint, so + // nothing downstream of here needed a second arm. + // + // Ahead of `suspend` deliberately — see `NodeExecutionResult.refuse` + // for why a result carrying both refuses rather than pausing. + if (result.refuse) { + throw new FlowRefusalSignal(node.id, result.refusalMessage); + } + + // ADR-0019 durable pause: the node did its on-entry work and asked to + // suspend here. Output is already written above; unwind the recursion + // up to execute()/resume(), which persists a continuation. Traversal + // of this node's out-edges happens on resume, not now. + if (result.suspend) { + throw new FlowSuspendSignal(node.id, result.correlation, result.screen, node.type); + } + + // #3447 P2: an executor may pick its own out-edge without suspending + // (e.g. an approval node auto-approving an empty slate walks its + // `approve` edge directly). The field predates this — it was declared + // on NodeExecutionResult "for decision nodes" but never consumed on + // the synchronous path; only resume() honoured its signal twin. On a + // labelled-edge node, falling through to the unlabelled traversal + // would walk EVERY unconditional out-edge (approve AND reject), so + // the label must be honoured here. + if (result.branchLabel) { + await this.traverseNext(node, flow, variables, context, steps, result.branchLabel); + return; + } + } + + // Continue to the node's successors. + await this.traverseNext(node, flow, variables, context, steps); +} + +/** The body of {@link AutomationEngine.traverseNext}, moved unedited; that method's docblock is its contract. */ +export async function traverseNext( + this: Pick, + node: FlowNodeParsed, + flow: FlowParsed, + variables: Map, + context: AutomationContext, + steps: StepLogEntry[], + branchLabel?: string, +): Promise { + // Find next nodes — separate conditional, default and unconditional edges + const allOutEdges = flow.edges.filter( + e => e.source === node.id && e.type !== 'fault', + ); + let outEdges = allOutEdges; + + // Branch selection: prefer edges tagged with the decision label. + if (branchLabel) { + let claimed = outEdges.filter(e => e.label === branchLabel); + // The `default` sentinel is also claimed by the BPMN default edge, so + // "none of my conditions matched" routes to the declared fallback + // without the author having to ALSO label that edge 'default'. + if (claimed.length === 0 && branchLabel === DEFAULT_BRANCH_LABEL) { + claimed = outEdges.filter(e => e.isDefault); + } + if (claimed.length > 0) { + outEdges = claimed; + } else { + // #4414 — do not fall back silently. The node computed a branch + // and no out-edge claims it, so every out-edge is about to be + // considered: the guard the author wrote is not guarding. + // + // #6654 — the computed branch label is potentially + // RECORD-DERIVED and the edge labels are FLOW-AUTHOR metadata + // (neither is schema-constrained against newlines), so both + // ride the structured slot, never the message; see + // `forgetSuspendedRun`'s catch for the full mechanism (#6299). + // #4632: FUNCTIONAL — stays `warn`. + this.logger.warn( + // `flow.name` is absent on the synthetic view `runRegion` builds. + `Flow '${flow.name ?? '(region)'}' node '${node.id}' (${node.type}) selected a branch, ` + + `but no out-edge carries that label — the computed branch and the out-edge labels are ` + + `in this record's meta. The branch selection is IGNORED and every out-edge is ` + + `evaluated instead, so unconditional siblings run regardless of the decision. ` + + `Make an out-edge's \`label\` match the branch, or mark the fallback edge ` + + `\`isDefault: true\`.`, + { + branchLabel, + outEdges: allOutEdges.map(e => ({ id: e.id, label: e.label ?? null })), + }, + ); + } + } + + const conditionalEdges: FlowEdgeParsed[] = []; + const defaultEdges: FlowEdgeParsed[] = []; + const unconditionalEdges: FlowEdgeParsed[] = []; + for (const edge of outEdges) { + if (edge.condition) { + conditionalEdges.push(edge); + } else if (edge.isDefault) { + defaultEdges.push(edge); + } else { + unconditionalEdges.push(edge); + } + } + + // #4354 — a gate that did not open leaves a trace. Record it: this is + // THE event that had no trace anywhere, and the reason #4347 shipped + // three inert production flows. A closed gate inside a loop body is + // logged once per iteration (region tagging in `runRegion` attaches the + // container + iteration), so the run summary can say "selected 30, + // acted 0, skipped 30 by " instead of reporting a green run that + // did nothing. + // + // The step is `skipped`, never a run: the re-entrancy guard, per-node + // `runs` counts and node status all exclude it, so recording a + // non-event stays a non-event to execution. + const recordSkipped = (nextNode: FlowNodeParsed, edge: FlowEdgeParsed): void => { + const at = new Date().toISOString(); + steps.push({ + nodeId: nextNode.id, + nodeType: nextNode.type, + ...(nextNode.label ? { nodeLabel: nextNode.label } : {}), + status: 'skipped', + startedAt: at, + completedAt: at, + durationMs: 0, + skippedBy: { + nodeId: node.id, + ...(edge.id ? { edgeId: edge.id } : {}), + ...(edge.label ? { label: edge.label } : {}), + }, + }); + }; + + // Conditional edges: evaluated sequentially, in the order the flow's + // `edges` array declares them — `FlowSchema.parse` and every conversion + // are copy-on-write maps that never reorder, so declaration order here + // IS the author's. + // + // On a `decision` node they are the gateway's branches, and the gateway + // is EXCLUSIVE unless it declares `config.mode: 'inclusive'` (#15429, + // maintainer ruling 「跟主流对齐」): the FIRST edge whose condition holds + // is the branch — the BPMN exclusive gateway, Salesforce Flow's + // Decision, n8n's Switch default — and its later siblings are not + // evaluated at all; they record the same `skipped` step a closed gate + // does, so the run log says which branch won and which were passed + // over. Until this change every true edge ran, one after another, under + // a comment calling that "mutually exclusive": hotcrm#1555 rendered a + // refusal screen AND ran the conversion in one execution. Authored + // sources written against that behaviour are carried across by the + // ADR-0087 conversion `flow-decision-mode-inclusive-explicit` (`os + // migrate meta --from 17`), which writes the inclusive declaration onto + // them — see `CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION` for why this seam + // does not. A flow stored in `sys_metadata` arrives here as stored and + // takes THIS reading (maintainer ruling letter C on #15429: no + // stored-row rewrite, no read-path completion); `os migrate meta + // --stored` lists such decisions for an operator to review. + // + // `mode: 'inclusive'` is the BPMN inclusive gateway: every edge whose + // condition holds runs, one successor at a time (never `Promise.all` — + // that fan-out belongs to the UNCONDITIONAL bucket below alone). + // + // Scoped to `decision` on purpose: `mode` is a decision-config key, and + // conditioned out-edges of any other node type keep the traversal they + // had — every one whose condition holds, sequentially. The ruling and + // its migration cover the gateway, and nothing else was measured to + // carry two conditioned out-edges (the corpus census on #15429). + const exclusive = node.type === 'decision' && !decisionTakesEveryBranch(node); + let anyConditionMet = false; + for (const edge of conditionalEdges) { + const nextNode = flow.nodes.find(n => n.id === edge.target); + if (exclusive && anyConditionMet) { + // A sibling already won: passed over, not evaluated. + if (nextNode) recordSkipped(nextNode, edge); + continue; + } + if (this.evaluateCondition(edge.condition!, variables, context)) { + anyConditionMet = true; + if (nextNode) { + await this.executeNode(nextNode, flow, variables, context, steps); + } + } else if (nextNode) { + recordSkipped(nextNode, edge); + } + } + + // Default edges (BPMN default flow, #4414): the fallback, taken only + // when NO conditional sibling matched. `isDefault` is what makes this an + // "otherwise" rather than a second unconditional path — without it the + // author's only spelling of "otherwise" was to hand-write the negation + // of every sibling condition, and forgetting to do that ran both + // branches. A default edge passed over because a real branch won records + // the same `skipped` trace a closed gate does (#4354). + for (const edge of defaultEdges) { + const nextNode = flow.nodes.find(n => n.id === edge.target); + if (!anyConditionMet) { + if (nextNode) { + await this.executeNode(nextNode, flow, variables, context, steps); + } + } else if (nextNode) { + const at = new Date().toISOString(); + steps.push({ + nodeId: nextNode.id, + nodeType: nextNode.type, + ...(nextNode.label ? { nodeLabel: nextNode.label } : {}), + status: 'skipped', + startedAt: at, + completedAt: at, + durationMs: 0, + skippedBy: { + nodeId: node.id, + ...(edge.id ? { edgeId: edge.id } : {}), + ...(edge.label ? { label: edge.label } : {}), + }, + }); + } + } + + // Unconditional edges: execute in parallel (Promise.all) + if (unconditionalEdges.length > 0) { + const parallelTasks = unconditionalEdges + .map(edge => flow.nodes.find(n => n.id === edge.target)) + .filter((n): n is FlowNodeParsed => n != null) + .map(nextNode => this.executeNode(nextNode, flow, variables, context, steps)); + + await Promise.all(parallelTasks); + } +} + +/** The body of {@link AutomationEngine.runRegion}, moved unedited; that method's docblock is its contract. */ +export async function runRegion( + this: Pick, + region: FlowRegionParsed, + variables: Map, + context: AutomationContext, + grouping?: { parentNodeId: string; iteration?: number; branch?: number; regionKind?: string; retryAttempt?: number }, + partialSteps?: StepLogEntry[], +): Promise { + const entryId = findRegionEntry(region); + const entry = region.nodes.find(n => n.id === entryId); + if (!entry) { + throw new Error(`region entry node '${entryId}' not found`); + } + // A synthetic flow view — executeNode/traverseNext only read `nodes`/`edges`. + const subFlow = { nodes: region.nodes, edges: region.edges ?? [] } as unknown as FlowParsed; + const regionSteps: StepLogEntry[] = []; + // Tag this region's steps with their container. Shared by the success + // and failure paths (#7546) so a failed attempt's steps are + // indistinguishable in SHAPE from a successful one's — they differ only + // in their own `status`. + // + // #15230 splits what "innermost wins" governs, because the two halves + // answer different questions: + // + // IDENTITY (`parentNodeId` / `regionKind` / `retryAttempt`) answers + // WHICH REGION RAN THIS STEP. Innermost wins outright: a step a + // nested region already claimed keeps naming that region, and an + // enclosing region must never relabel it. + // + // INDEX (`iteration` / `branch`) answers WHICH PASS OF WHICH REGION. + // Nested regions contribute DIFFERENT indices — the loop's row and + // the parallel's branch are both true of the same step — so an + // enclosing region fills the index the inner one left undefined + // instead of being skipped along with the identity fields. That skip + // is the defect: `loop { parallel }` used to record the branch and + // discard the row. + // + // Still "only fills what is undefined" in both halves, so an index a + // nested region DID set wins: for `loop { loop }` the inner loop's + // `iteration` stands, exactly as before. + const tag = (): void => { + if (!grouping) return; + for (const step of regionSteps) { + if (step.parentNodeId === undefined) { + step.parentNodeId = grouping.parentNodeId; + if (grouping.regionKind !== undefined) step.regionKind = grouping.regionKind; + if (grouping.retryAttempt !== undefined) step.retryAttempt = grouping.retryAttempt; + } + if (grouping.iteration !== undefined && step.iteration === undefined) { + step.iteration = grouping.iteration; + } + if (grouping.branch !== undefined && step.branch === undefined) { + step.branch = grouping.branch; + } + } + }; + try { + await this.executeNode(entry, subFlow, variables, context, regionSteps); + } catch (err) { + // #7546: surface what the failed attempt DID get through before + // rethrowing. Tagged first so the caller receives finished records, + // and pushed into the caller's sink rather than returned because + // this path's contract is (still) to throw. + tag(); + partialSteps?.push(...regionSteps); + // [#18881] A refusal raised at an INNER region boundary is already + // the named one, and it is re-thrown untouched. Re-wrapping it here + // would rename the region: for `loop { try_catch { map } }` the + // author's fault is the try region, and the loop is only the frame + // the unwind passes through. Tested before the suspend arm because + // this error is not a suspend signal and must not reach the generic + // rethrow below with an enclosing region's identity stamped on it. + if (isRegionSuspensionRefusal(err)) throw err; + // [#18881] The runtime half of #15646's ruling D: a region body + // cannot carry a durable pause, and the refusal is now NAMED + // (region node, suspending node, sub-flow) instead of a plain + // `Error` that an enclosing `try_catch` read as an ordinary region + // failure and handed to its catch handler. See + // `region-suspension-refusal.ts` for the measurement that is. + // + // The sub-flow is read off the suspending node's own `config` + // (`map` / `subflow` name their child there). The node is looked up + // in THIS region's `nodes` because this is the innermost boundary + // the signal crosses — a deeper suspension was already converted by + // the arm above, so `err.nodeId` always names a node of this body. + if (isSuspendSignal(err)) { + const suspended = region.nodes.find(n => n.id === err.nodeId); + const flowName = (suspended?.config as { flowName?: unknown } | undefined)?.flowName; + throw refuseRegionSuspension({ + regionNodeId: grouping?.parentNodeId ?? entryId, + regionKind: grouping?.regionKind ?? 'region', + suspendedNodeId: err.nodeId, + ...(typeof flowName === 'string' && flowName ? { subFlowName: flowName } : {}), + }); + } + // [#15788] The refusing `end` node's signal, converted at exactly + // the same boundary and for the same reason: a control signal must + // not cross a region edge silently. It would not merely leak — the + // `try_catch` executor's own `catch (err)` arm would read it as the + // try region FAILING and hand it to the catch handler, so an + // author's refusal would run the error path and the run would + // record `completed`. That is the pre-#15788 silence with an extra + // step, which is worse than a refusal that says so. + // + // ⛔ Not a narrowing of anything an author had: today an `end` in a + // region is a no-op whatever its `outcome`, so the shape being made + // loud has never once been honoured. Whether a refusal should + // instead PROPAGATE out of a region and terminate the run is a real + // question and ⛔ not one this lane rules on — the #14945 ruling + // says nothing about regions, and "prefer failing to falling back" + // decides the interim. + // + // [#18110 / #18555] The sentence NAMES the node that carried the + // refusal and nothing more, because there are now two producers: an + // `end` declaring the refusal itself, and a `subflow` / `map` whose + // CHILD run refused. Hard-wired to the first, it told an author + // inside a region to go find an `end` node that is not in their + // region at all, and handed them a prescription they could not + // follow. ⛔ TEXT only — region SEMANTICS are untouched (#18112's + // option B is not implemented, no container is taught to rethrow), + // and the authoring-time half of this boundary is #15646's, ⛔ not + // this change's. + if (isRefusalSignal(err)) { + throw new Error( + `a refusal inside a structured region (node '${err.nodeId}') is not supported — a ` + + `refusal terminates the RUN, and a region body cannot end one. The refusing node is ` + + `either an 'end' declaring outcome: 'refused', or a node whose own child run refused ` + + `(a 'subflow' / 'map'). Move that node onto the top-level graph and route the ` + + `region's exit to it.`, + ); + } + throw err; + } + tag(); + return regionSteps; +} + +/** The body of {@link AutomationEngine.executeWithTimeout}, moved unedited; that method's docblock is its contract. */ +export async function executeWithTimeout( + promise: Promise, + timeoutMs: number, + nodeId: string, +): Promise { + let guard: ReturnType | undefined; + + const timeoutPromise = new Promise((_, reject) => { + guard = setTimeout(() => { + reject(new Error(`Node '${nodeId}' timed out after ${timeoutMs}ms`)); + }, timeoutMs); + }); + + try { + return await Promise.race([promise, timeoutPromise]); + } finally { + clearTimeout(guard); + } +} + +/** The body of {@link AutomationEngine.celScope}, moved unedited; that method's docblock is its contract. */ +export function celScope( + variables: Map, + context?: AutomationContext, +): { extra: Record } { + const vars: Record = {}; + for (const [key, value] of variables) { + // Convert "step.result" keys into nested object paths. + const segs = key.split('.'); + let cursor = vars; + for (let i = 0; i < segs.length - 1; i++) { + if (typeof cursor[segs[i]] !== 'object' || cursor[segs[i]] === null) { + cursor[segs[i]] = {}; + } + cursor = cursor[segs[i]] as Record; + } + cursor[segs[segs.length - 1]] = value; + } + // [#22642] No `record` slot: `record` is bound by the spread when the + // run holds one, and unbound otherwise (see the docblock). + return { extra: { ...vars, vars, current_user: runUserOf(context) } }; +} + +/** The body of {@link AutomationEngine.valueEnvelopeRefusals}, moved unedited; that method's docblock is its contract. */ +export function valueEnvelopeRefusals(value: unknown): { message: string; source: string }[] { + // [#16439] Presence, asked before either primitive, because neither can + // answer it (see rule 0 above). It is stated HERE and not as a guard in + // {@link evaluateValueEnvelope} deliberately: a reject reason living + // only on the evaluation side would end the property this method exists + // for — registration and evaluation refuse ONE set, derived by one call + // — and the docblock that says so would stop being true. + // + // REFUSED rather than admitted, and the asymmetry with the predicate + // side is deliberate, not an oversight to harmonise away: + // `structuralConditionRefusal` admits `null` / `undefined` because the + // condition FIELD is optional, so absence there means "the author wrote + // no predicate". A value slot's envelope IS the value, so an absent one + // is a caller handing nothing where a value was required: there is + // nothing to compute, and — per this method's own ADR-0032 §1c + // argument — no falsy default for it to hide behind. + // + // Measured, so the registration half is not an assumption: the + // value-role feeder (`resolveFlowNodeExpressions`) emits ONLY + // envelope-shaped objects and `isExpressionEnvelopeShaped` is false for + // both shapes, so `registerFlow` never presents a nullish value here. + // An authored `null` in an `assignments` slot stays what it always was + // — a literal that parses (`FlowSchema.parse` accepts it) and still + // registers. This adds nothing to the registration reject set. + if (value == null) { + return [{ + message: + `${VALUE_ENVELOPE_REFUSAL} no envelope was passed: the argument is ` + + `\`${value === null ? 'null' : 'undefined'}\`, so there is nothing to evaluate. An absent ` + + 'envelope is not "not authored" — the predicate side admits absence because the condition ' + + "field is optional, but a value slot's envelope IS the value. Write " + + "`{ dialect: 'cel', source: '…' }`.", + source: '', + }]; + } + const source = (value as { source?: unknown })?.source; + const sourceText = typeof source === 'string' ? source : ''; + const shape = FlowValueSlotSchema.safeParse(value); + if (!shape.success) { + // Already prefixed with the refusal sentence by the spec's own + // `superRefine` — re-prefixing would say it twice. + return shape.error.issues.map((issue) => ({ message: issue.message, source: sourceText })); + } + const parsed = validateExpression('value', value as { dialect?: string; source?: string }); + return parsed.errors.map((e) => ({ + message: `${VALUE_ENVELOPE_REFUSAL} ${e.message}`, + source: e.source, + })); +} + +/** The body of {@link AutomationEngine.evaluateValueEnvelope}, moved unedited; that method's docblock is its contract. */ +export function evaluateValueEnvelope( + this: Pick, + envelope: { dialect?: string; source?: string; ast?: unknown }, + variables: Map, + where: string, + context?: AutomationContext, +): unknown { + const refusals = this.valueEnvelopeRefusals(envelope); + if (refusals.length > 0) { + throw new Error( + `${where}: ${refusals.map((r) => r.message).join(' ')} — source: \`${refusals[0].source}\``, + ); + } + const source = envelope.source ?? ''; + const result = ExpressionEngine.evaluate({ dialect: 'cel', source }, this.celScope(variables, context)); + if (!result.ok) { + // Reached by the shapes the two validators above cannot judge — an + // `ast`-only envelope (`ExpressionSchema` accepts `source`-or-`ast`; + // the CEL engine evaluates `source` only) and a source that parses + // but faults on the live values. Loud, with the source attached. + throw new Error( + `${where}: value expression failed to evaluate as CEL: ${result.error?.message ?? 'unknown error'} — ` + + `source: \`${source}\`.`, + ); + } + return result.value; +} + +/** The body of {@link AutomationEngine.evaluateCondition}, moved unedited; that method's docblock is its contract. */ +export function evaluateCondition( + this: Pick, + expression: string | { dialect?: string; source?: string; ast?: unknown }, + variables: Map, + context?: AutomationContext, +): boolean { + const shapeRefusal = structuralConditionRefusal(expression); + if (shapeRefusal) { + // ADR-0032 §1d — the error carries its source. `structuralConditionRefusal` + // attributes an empty one for the shape it is refusing here (a non-string + // `source` cannot be the attribution), which is its documented choice. + throw new Error( + `condition evaluation error: ${shapeRefusal.message} — source: \`${shapeRefusal.source}\``, + ); + } + + const isEnvelope = typeof expression === 'object' && expression != null && 'dialect' in expression; + const dialect = isEnvelope ? (expression as { dialect?: string }).dialect : undefined; + const exprStr = typeof expression === 'string' ? expression : ((expression as { source?: string })?.source ?? ''); + + if (dialect === 'template' || dialect === 'flow') { + // [#19939] The legacy `{var}` template dialect is deleted (see the + // doc comment): the same set `registerFlow` refuses at every + // condition slot, refused here for a caller that skipped it. + throw new Error( + `condition evaluation error: a condition is CEL, and the \`${dialect}\` dialect is not evaluated — ` + + `source: \`${exprStr}\`. Write the predicate as CEL without \`{…}\` template braces ` + + `(e.g. \`record.status == 'open'\`), as bare text or as \`{ dialect: 'cel', source }\`.`, + ); + } + if (isEnvelope && dialect && dialect !== 'cel') { + // Other dialects (cron, js) are not boolean predicates here. + return false; + } + + // An absent / empty condition is not a predicate to evaluate. Callers that + // mean "unconditional" guard before calling; this is the one that does not + // (a `decision` node whose `conditions[]` entry has no `expression`), and + // an unauthored branch must not open. + if (exprStr.trim() === '') return false; + + // Bind the `vars` scope, so a `$`-named variable or a node output's + // flat dotted key is read as `vars["…"]`. + try { + const result = ExpressionEngine.evaluate( + { dialect: 'cel', source: exprStr }, + this.celScope(variables, context), + ); + // ADR-0032 §Decision 1c — NO silent fallback. A non-`ok` result is a + // real fault (malformed predicate, or — pre build-validation — a + // `{…}` template mistakenly written into a CEL condition). Surfacing + // it as a thrown, attributed error makes execute()'s catch record a + // loud flow failure, instead of the old `return false` that made a + // broken condition indistinguishable from "condition not met" (#1491). + if (!result.ok) { + throw new Error( + `condition failed to evaluate as CEL: ${result.error?.message ?? 'unknown error'} — ` + + `source: \`${exprStr}\`. Conditions are bare CEL (e.g. \`record.rating >= 4\`); ` + + `do not wrap field references in \`{…}\` template braces.`, + ); + } + return Boolean(result.value); + } catch (err) { + // Re-throw with the source attached (ADR-0032 §1d — errors written + // for self-correction). Never swallow to `false`. + const msg = (err as Error)?.message ?? String(err); + throw new Error( + msg.includes('source:') ? msg : `condition evaluation error: ${msg} — source: \`${exprStr}\``, + ); + } +} diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 4e968bfe5e8..c63d69f46d5 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1,6 +1,6 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. -import type { FlowParsed, FlowNodeParsed, FlowEdgeParsed } from '@objectstack/spec/automation'; +import type { FlowParsed, FlowNodeParsed } from '@objectstack/spec/automation'; import type { ExecutionLog, ExecutionStatus, @@ -13,7 +13,6 @@ import type { import type { AutomationContext, AutomationResult, ResumeSignal, IAutomationService, RunListResult, ScreenSpec, ScreenFieldSpec, ConnectorSourcePullRequest, ConnectorSourcePullResult } from '@objectstack/spec/contracts'; import type { FlowRegistration, FlowRegistrationRefusal, FlowInnerRefusal } from '@objectstack/spec/api'; import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; -import { createEvalUser, type EvalUserParsed } from '@objectstack/spec/identity'; import { validateScreenInputs, screenDeclaresInputContract, @@ -23,7 +22,7 @@ import { 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, findRegionEntry } from '@objectstack/spec/automation'; +import { FlowSchema, FLOW_RESERVED_NODE_TYPES, FLOW_STRUCTURAL_NODE_TYPES, validateControlFlow, collectFlowGraphs } 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 @@ -34,19 +33,6 @@ import { scheduledWorkDisabledReason, type ScheduledWorkPolicy, } from '@objectstack/types'; -import { structuralConditionRefusal } from '@objectstack/spec/automation'; -// [#15137] The `value`-role half of the ledger. Both halves of "is this envelope -// well-formed?" are IMPORTED, never re-spelled here: the shape rule is -// `FlowValueSlotSchema` (spec, #14149; slot-neutral since #19938 — it refuses a -// non-`cel` dialect and the `{ dialect: 'cel' }` with no `source` that -// `validateExpression` reads as "not authored"), and the CEL rule is -// `validateExpression('value', …)` (formula). `registerFlow` and the run-time -// evaluator call the SAME composition -// ({@link AutomationEngine.valueEnvelopeRefusals}), so a flow that registers can -// never be refused at run time and vice versa — for every value slot the ledger -// declares (`assignment.assignments.*`, `create_record` / `update_record` -// `fields.*`). -import { FlowValueSlotSchema, VALUE_ENVELOPE_REFUSAL } from '@objectstack/spec/automation'; import { applyConversionsToFlow, 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 @@ -63,46 +49,6 @@ import type { ConnectorDescriptor, } from '@objectstack/spec/integration'; import { ConnectorSchema } from '@objectstack/spec/integration'; -// Static import (not a lazy `require`): the engine ships as ESM ("type":"module"), -// where a CommonJS `require('@objectstack/formula')` resolves to tsup's throwing -// `__require` stub. That threw on every CEL evaluation and the catch below -// silently returned `false`, so EVERY start-node / edge condition (record-change -// `previous.*`, `budget > 100000`, …) skipped its flow. A static import binds the -// engine at module load in both ESM and CJS builds. -import { ExpressionEngine, validateExpression } from '@objectstack/formula'; - -/** - * The value the flow CEL scope binds as `current_user` for a run (#19939 — - * the maintainer's ruling on its second pass, Q1 A and Q2 A): the run's - * `EvalUser` when the run has a user, else `null`. - * - * Built through `createEvalUser`, the one factory every surface builds the - * canonical user with (ADR-0068), from what the run context holds and - * nothing else: `id` from `userId`, `positions`, and `organizationId` from - * `tenantId`; `isPlatformAdmin` is derived from the positions there. Neither - * an email nor a name is bound (Q2 A enumerates the four above). A - * `runAs:'user'` run's context does carry the user's email - * (`AutomationContext.email`, seeded at run setup), but for its data - * operations' row policies, which read it as `current_user.email` through - * `resolveRunDataContext`; it is not a CEL fact here, and a `runAs:'system'` - * run carries none. A flow that needs either reads the user record - * (`current_user.id`). - * - * A run with no user — a schedule, a record change made by a system write, a - * caller that passed no context — answers `null`, never a pseudo-user - * (ADR-0118 D1, D4): `current_user.id` then fails the run loudly, and the - * author's guard `current_user != null ? current_user.id : null` can be - * written. - */ -function runUserOf(context: AutomationContext | undefined): EvalUserParsed | null { - const userId = context?.userId; - if (typeof userId !== 'string' || userId === '') return null; - return createEvalUser({ - id: userId, - positions: Array.isArray(context?.positions) ? context.positions : [], - ...(typeof context?.tenantId === 'string' ? { organizationId: context.tenantId } : {}), - }); -} /** * The branch a `decision` node reports when it DECLARED `config.conditions` and @@ -118,10 +64,8 @@ function runUserOf(context: AutomationContext | undefined): EvalUserParsed | nul export const DEFAULT_BRANCH_LABEL = 'default'; import { runIsUnscopedUserMode, flowTouchesData, runOrgUserIds, runTriggerUserId, type SeededRunContext } from './runtime-identity.js'; -import { isGuardRefusal, refuseNode } from './guard-refusal.js'; -import { readPartialSteps } from './partial-steps.js'; -import { attachCarriedRefusal, carriedRefusalOfResult, readCarriedRefusal, type CarriedRefusal } from './inner-refusal.js'; -import { isRegionSuspensionRefusal, refuseRegionSuspension } from './region-suspension-refusal.js'; +import { refuseNode } from './guard-refusal.js'; +import { carriedRefusalOfResult, readCarriedRefusal, type CarriedRefusal } from './inner-refusal.js'; import { summarizeRun, formatRunSummaryLine } from './run-summary.js'; // #5660 — the degrade registration reports a FOREIGN failure (a third-party // provider factory's text), so it renders it as structured `meta` rather than @@ -166,18 +110,20 @@ import { registerDegradedConnector, assertSameOriginOrFree, } from './plugin-extension-api.js'; - -/** - * Does this `decision` take EVERY out-edge whose condition holds (#15429)? - * Only an explicit `config.mode: 'inclusive'` says so; an omitted `mode` and - * `'exclusive'` both mean the first true edge in declaration order wins. Any - * other value never reaches here — {@link AutomationEngine.registerFlow} - * refuses it with the spec's prescription. - */ -function decisionTakesEveryBranch(node: FlowNodeParsed): boolean { - const config = node.config as { mode?: unknown } | undefined; - return config?.mode === 'inclusive'; -} +// [#25222] The DAG traversal's bodies: node execution, edge traversal, regions, +// and condition and value evaluation. Each `AutomationEngine` +// method of the same name calls its body. +import { + executeNode, + traverseNext, + runRegion, + executeWithTimeout, + celScope, + valueEnvelopeRefusals, + evaluateValueEnvelope, + evaluateCondition, + type runUserOf, +} from './dag-traversal.js'; // ─── Node Executor Interface (Plugin Extension Point) ─────────────── @@ -1178,7 +1124,7 @@ function buildRunTrigger( * sibling parallel branches already in flight are not cancelled — durable * pause across parallel gateways is out of scope for ADR-0019 M1. */ -class FlowSuspendSignal { +export class FlowSuspendSignal { readonly __flowSuspend = true as const; constructor( readonly nodeId: string, @@ -1189,7 +1135,7 @@ class FlowSuspendSignal { ) {} } -function isSuspendSignal(err: unknown): err is FlowSuspendSignal { +export function isSuspendSignal(err: unknown): err is FlowSuspendSignal { return typeof err === 'object' && err !== null && (err as FlowSuspendSignal).__flowSuspend === true; } @@ -1218,7 +1164,7 @@ function isSuspendSignal(err: unknown): err is FlowSuspendSignal { * exists to do, so this reuses the mechanism rather than inventing a second * unwinding protocol. (Not exported — callers see `status: 'refused'`.) */ -class FlowRefusalSignal { +export class FlowRefusalSignal { readonly __flowRefused = true as const; constructor( /** @@ -1239,7 +1185,7 @@ class FlowRefusalSignal { ) {} } -function isRefusalSignal(err: unknown): err is FlowRefusalSignal { +export function isRefusalSignal(err: unknown): err is FlowRefusalSignal { return typeof err === 'object' && err !== null && (err as FlowRefusalSignal).__flowRefused === true; } @@ -11044,408 +10990,24 @@ export class AutomationEngine implements IAutomationService { /** * Execute a node with timeout support, fault edge handling, and step logging. */ - private async executeNode( + private executeNode( node: FlowNodeParsed, flow: FlowParsed, variables: Map, context: AutomationContext, steps: StepLogEntry[], ): Promise { - if (node.type === 'end') { - // [#15788] …unless the author declared a REFUSAL here (#14945 - // ruling 2′). `end` has no executor and no descriptor — it is - // structural (`FLOW_STRUCTURAL_NODE_TYPES`), which is why this - // method opens by returning on it — so this is the one place the - // outcome can be read, and until now nothing read it: a flow - // declaring `outcome: 'refused'` ran as a plain completion and the - // author's reason reached nobody. - // - // The config arrives PARSED. `EndConfigSchema` is applied by - // `FlowNodeSchema`'s own transform (`parseEndNodeConfig`, spec - // lane 1), which runs on every node including the ones nested in a - // region, so `outcome` is defaulted and a `refused` without a - // `message` was already refused at the flow parse. ⛔ No second - // door, no `??` default, no re-parse: this reads a contract that is - // already enforced rather than defending against it. - const endConfig = node.config as { outcome?: string; message?: string } | undefined; - if (endConfig?.outcome === 'refused') { - throw new FlowRefusalSignal( - node.id, - // The SAME rendering a screen `description` gets — the - // ruling's words, one implementation (`renderTextSlot`, the - // `{{ }}` text renderer since #22110). - // Rendered HERE, against the live variable map, because - // that is what makes the text per-record; a template on the - // wire would put the rendering in every runner. - // [#22450] …in the run's language: the TEMPLATE is picked - // before it is rendered, because the run stores the - // rendered text and nothing downstream can translate it. - this.renderRefusalMessage(flow.name, node.id, endConfig.message, variables, context), - ); - } - return; - } - - // ADR-0044 runaway guard: declared back-edges make re-entering a node - // legal, so a misauthored unconditional loop could otherwise spin - // forever. Count this node's prior *top-level* visits in the run's step - // log (region body steps carry `parentNodeId` and are excluded — a - // 200-iteration `loop` region is legitimate) and fail the run loudly - // past the cap. Product-level guards (e.g. an approval node's - // `maxRevisions`) terminate far earlier; this is the engine backstop. - // - // A `skipped` step is NOT a visit (#4354): those entries record a gate - // that closed in FRONT of a node, so counting them would let a flow whose - // gate refuses often abort itself as a runaway — a new observability - // signal changing execution semantics, which it must never do. - const priorVisits = steps.reduce( - (n, s) => - s.nodeId === node.id && s.parentNodeId === undefined && s.status !== 'skipped' - ? n + 1 - : n, - 0, - ); - if (priorVisits >= AutomationEngine.MAX_NODE_REENTRIES) { - throw new Error( - `Node '${node.id}' was entered ${priorVisits} times in one run — aborting as a runaway loop ` + - `(back-edge cycles must terminate; see ADR-0044)`, - ); - } - - const stepStart = Date.now(); - const stepStartedAt = new Date().toISOString(); - - // Find executor - const executor = this.nodeExecutors.get(node.type); - if (!executor) { - // start node without executor is fine — just skip - if (node.type !== 'start') { - steps.push({ - nodeId: node.id, - nodeType: node.type, - status: 'failure', - startedAt: stepStartedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - stepStart, - error: { code: 'NO_EXECUTOR', message: `No executor registered for node type '${node.type}'` }, - }); - throw new Error(`No executor registered for node type '${node.type}'`); - } - // Log start node step - steps.push({ - nodeId: node.id, - nodeType: node.type, - status: 'success', - startedAt: stepStartedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - stepStart, - }); - } else { - // Execute node with optional timeout - let result: NodeExecutionResult; - try { - if (node.timeoutMs && node.timeoutMs > 0) { - result = await this.executeWithTimeout( - executor.execute(node, variables, context), - node.timeoutMs, - node.id, - ); - } else { - result = await executor.execute(node, variables, context); - } - } catch (execErr: unknown) { - const errMsg = execErr instanceof Error ? execErr.message : String(execErr); - // [#18881] ONE region refusal is ONE failure. The refusal names - // the region node whose body could not carry the pause, and - // that node's own frame records it exactly as any other thrown - // failure does. Every ENCLOSING container the unwind passes - // through — the `loop` around the `try_catch` in the card's - // reproduction — records nothing: it did not fail, it is the - // frame a failure is travelling out through, and a step for it - // would make `summary.failed` count the NESTING DEPTH rather - // than the fault. `summary.failed` is `Σ nodes[].failures` - // (#14456), so a second step here reads as a second lost row to - // every operator and every #4354 reader. - // - // Keyed on the refusal's OWN `regionNodeId` rather than on a - // mutable "already reported" flag: the identity is decided once - // at the boundary that raised it and cannot drift as the error - // travels. - const enclosingFrameOfRegionRefusal = - isRegionSuspensionRefusal(execErr) && execErr.regionNodeId !== node.id; - if (!enclosingFrameOfRegionRefusal) { - steps.push({ - nodeId: node.id, - nodeType: node.type, - status: 'failure', - startedAt: stepStartedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - stepStart, - error: { code: 'EXECUTION_ERROR', message: errMsg }, - }); - } - - // #13803 — a structured container that DIED mid-body still did - // whatever its completed iterations did, and those writes are - // already in the database. Fold the body steps it carried out - // on the throw into the run log, in the same position the - // success path splices `childSteps`: right behind the - // container's own step, ahead of any `fault` handler's steps. - // - // Without this the record was lost twice over — the run log - // kept no trace of the completed iterations, and the #4354 - // summary folded over that log reported `acted: 0` for a sweep - // that had genuinely written rows. `acted: 0` on a failed run - // reads as "nothing happened, safe to re-run", so the summary - // was wrong in the one direction that causes double-execution. - // - // Purely additive to the record: the error is rethrown below - // untouched, so which failures route, what `$error` holds and - // what the run reports as its error are all unchanged. Every - // step folded here carries a `parentNodeId` (set by - // `runRegion`'s tagger), so the ADR-0044 runaway guard — which - // counts only top-level visits — cannot see them either. - const carriedSteps = readPartialSteps(execErr); - if (carriedSteps?.length) { - steps.push(...carriedSteps); - } - - // #14955 — publish the failure BEFORE deciding whether it - // routes, exactly as the returned-failure arm below does. - // - // This write used to sit inside `if (faultEdge)`, and no reason - // for that was ever recorded: both arms were written in one - // commit, and only this one conflated "publish the failure" with - // "route the failure" — the arm below already separated them. - // The consequence was invisible rather than loud. A node inside a - // region never has a `fault` edge of its own (the region's - // synthetic sub-flow carries only the region's own edges — see - // {@link AutomationEngine.runRegion}), so EVERY thrown failure - // inside a region left `$error` naming an earlier, unrelated - // failure. The first reader that cared about `$error`'s freshness - // — `try_catch`'s `code` binding (commit c5a7448d5) — met it immediately - // and bound a message and a code that came from two different - // failures: `{ code: 'DUPLICATE_RECORD', message: "Node 'mk' - // timed out after 20ms" }`, swallowed by a catch region reading - // it as "the row is already there", the run reporting success. - // - // `{ nodeId, message }` and nothing more: there is no - // `NodeExecutionResult` on this path, so no `output` and no - // classified `code` exist to carry — and that absence is the - // correct answer for a throw, not a reason to leave a stale - // `code` standing. This is what the documented contract already - // promised — `{$error}` names the most recent failure - // (`content/docs/automation/flows.mdx`) — and now holds. - // - // Publishing is not routing: the guard-refusal rule below is - // untouched and still decides, alone, which failures a `fault` - // edge may carry. Nor is the thrown value touched — `execErr` is - // rethrown below exactly as caught. - // - // [#18881] …and it is published for the SAME frames that record - // a step, for the same reason: an enclosing container the - // region refusal is travelling out through did not fail, so - // `{$error}` naming it would be a false sentence about which - // node produced the run's failure. The region node's own frame - // publishes, as any failing node does. - if (!enclosingFrameOfRegionRefusal) { - variables.set('$error', { nodeId: node.id, message: errMsg }); - this.setNodeError(variables, node.id, errMsg); - } - - // #3863 — a guard that THROWS is as un-routable as one that - // returns: `UnscopedRunDataAccessError` (ADR-0049/#1888) reports - // that the metadata would run unscoped, and rerouting it would - // let a `fault` edge disable the elevation check. - const faultEdge = isGuardRefusal(execErr) - ? undefined - : flow.edges.find(e => e.source === node.id && e.type === 'fault'); - if (faultEdge) { - const faultTarget = flow.nodes.find(n => n.id === faultEdge.target); - if (faultTarget) { - await this.executeNode(faultTarget, flow, variables, context, steps); - return; - } - } - throw execErr; - } - - // #6667 — declared = enforced for `supportsPause`, at the ONE seam - // every suspension passes through. - // - // Placed here rather than beside the `throw new FlowSuspendSignal` - // below on purpose: converting the mismatch into an ordinary guard - // refusal *before* the success bookkeeping means the run records a - // `failure` step for the offending node (not a `success` step - // followed by an unexplained failed run), sets `$error` like any - // other refusal, and inherits #3863's un-routability — a `fault` - // edge must not be able to swallow a declaration defect, since - // re-running the flow unchanged can never fix one. - // - // Exactly once, and nothing bypasses it: this is the only call site - // of any `executor.execute()` that the engine acts on — the ADR-0018 - // alias path delegates and RETURNS its target's result here rather - // than suspending on its own, `resume()` re-enters through - // {@link executeNode}, and region bodies ({@link runRegion}) do too. - if (result.success && result.suspend === true) { - result = this.refuseUndeclaredSuspension(node.type) ?? result; - } - - if (!result.success) { - const errMsg = result.error ?? 'Unknown error'; - steps.push({ - nodeId: node.id, - nodeType: node.type, - status: 'failure', - startedAt: stepStartedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - stepStart, - error: { code: 'NODE_FAILURE', message: errMsg }, - ...(result.warnings?.length ? { warnings: result.warnings } : {}), - // #4354 — a node that failed PART WAY may still have written - // rows; dropping its counts would understate what the run did. - ...(result.metrics ? { metrics: result.metrics } : {}), - }); - - // #14184 — the returned-failure half of #13803's fold. A - // container that RETURNS its failure (rather than throwing it) - // never reached the splice below, because that one runs only - // after a successful result. `try_catch` with no `catch` region - // is exactly that shape: it catches the try region's failure and - // returns it, so its completed try-region steps were dropped on - // the floor and the #4354 summary reported `acted: 0` over rows - // that had genuinely been written. - // - // Same position as the throw arm's fold and the success path's - // splice: right behind the container's own step, ahead of any - // `fault` handler's steps. #13803 left this branch alone because - // no executor then returned `childSteps` on a failing result; - // `try_catch` is that producer, and it is still the only one. - // - // Additive to the record only — the failure step above, the - // `$error` write below, the guard-refusal routing rule and the - // rethrow are all untouched. Every folded step carries a - // `parentNodeId` (set by `runRegion`'s tagger), so the ADR-0044 - // runaway guard, which counts only top-level visits, cannot see - // them either. - if (result.childSteps?.length) { - steps.push(...result.childSteps); - } - - // Write error output to variable context for downstream nodes. - // Commit c5a7448d5 — carry the executor's classified `code` (when it set - // one) alongside `message`, so a `fault` edge handler reading - // `{$error.code}` can branch on it; a `try_catch` region reads - // this same node-level `$error` before its OWN `errorVariable` - // write can shadow it (see try-catch-node.ts). - variables.set('$error', { - nodeId: node.id, - message: errMsg, - output: result.output, - ...(result.code ? { code: result.code } : {}), - }); - this.setNodeError(variables, node.id, errMsg); - - // #3863 — only a `runtime` failure may be routed. A `guard` - // refusal says the METADATA is wrong (#3810 erased a filter - // condition, a data node names no object); rerouting it would - // make a single `fault` edge a switch that turns the guard off - // while the run still reports success. - const faultEdge = result.errorClass === 'guard' - ? undefined - : flow.edges.find(e => e.source === node.id && e.type === 'fault'); - if (faultEdge) { - const faultTarget = flow.nodes.find(n => n.id === faultEdge.target); - if (faultTarget) { - await this.executeNode(faultTarget, flow, variables, context, steps); - return; - } - } - // [#24914, ruling B′] The text is unchanged — it is the run's - // diagnostic, read by the run log, `failAncestors`, approvals - // and every door's `message`. What the node's data call met is - // carried BESIDE it, on a brand (`inner-refusal.ts`), to the - // terminal exit that builds the run's result. - throw attachCarriedRefusal(new Error(`Node '${node.id}' failed: ${errMsg}`), result); - } - - // Log successful step (#3407: advisory executor warnings ride along - // so a legal-but-partial outcome never reads as a clean success; - // #4354: the executor's own record counts ride along too, so the run - // summary can tell a sweep that had nothing to do from one that did - // nothing). - steps.push({ - nodeId: node.id, - nodeType: node.type, - status: 'success', - startedAt: stepStartedAt, - completedAt: new Date().toISOString(), - durationMs: Date.now() - stepStart, - ...(result.warnings?.length ? { warnings: result.warnings } : {}), - ...(result.metrics ? { metrics: result.metrics } : {}), - }); - - // #1479: fold a structured-region container's body/branch/handler - // steps into the run log, right after the container's own step. - if (result.childSteps?.length) { - steps.push(...result.childSteps); - } - - // Write back output variables - if (result.output) { - for (const [key, value] of Object.entries(result.output)) { - variables.set(`${node.id}.${key}`, value); - } - } - - // [#18110 / #18555] Terminal refusal: the node evaluated and the - // answer is no. Thrown from HERE — the position the suspend signal - // below is thrown from — and that position is the point of the - // design, not a convenience: the node's success step is already - // pushed, its `childSteps` are already folded and its output is - // already written back, so a refusing `subflow` / `map` keeps the - // child's #4354 rollup (`selected` / `acted` / `unmeasuredEffect`) - // in the run summary. An unwind that began any earlier would drop - // exactly those counts — a refusing child really can have written - // rows before it said no. - // - // The step stays a SUCCESS on purpose: the node did evaluate, and - // what it evaluated to is the run's outcome, not this step's. The - // three terminal exits (`execute` / `resumeInternal` / - // `executeWithoutRetry`) already convert the signal into a - // `refused` run through the one `finishRefusedRun` chokepoint, so - // nothing downstream of here needed a second arm. - // - // Ahead of `suspend` deliberately — see `NodeExecutionResult.refuse` - // for why a result carrying both refuses rather than pausing. - if (result.refuse) { - throw new FlowRefusalSignal(node.id, result.refusalMessage); - } - - // ADR-0019 durable pause: the node did its on-entry work and asked to - // suspend here. Output is already written above; unwind the recursion - // up to execute()/resume(), which persists a continuation. Traversal - // of this node's out-edges happens on resume, not now. - if (result.suspend) { - throw new FlowSuspendSignal(node.id, result.correlation, result.screen, node.type); - } - - // #3447 P2: an executor may pick its own out-edge without suspending - // (e.g. an approval node auto-approving an empty slate walks its - // `approve` edge directly). The field predates this — it was declared - // on NodeExecutionResult "for decision nodes" but never consumed on - // the synchronous path; only resume() honoured its signal twin. On a - // labelled-edge node, falling through to the unlabelled traversal - // would walk EVERY unconditional out-edge (approve AND reject), so - // the label must be honoured here. - if (result.branchLabel) { - await this.traverseNext(node, flow, variables, context, steps, result.branchLabel); - return; - } - } - - // Continue to the node's successors. - await this.traverseNext(node, flow, variables, context, steps); + // The body lives in ./dag-traversal.ts (#25222). + const host: ThisParameterType = { + renderRefusalMessage: (...a) => this.renderRefusalMessage(...a), + nodeExecutors: this.nodeExecutors, + executeWithTimeout: (...a) => this.executeWithTimeout(...a), + setNodeError: (...a) => this.setNodeError(...a), + executeNode: (...a) => this.executeNode(...a), + refuseUndeclaredSuspension: (...a) => this.refuseUndeclaredSuspension(...a), + traverseNext: (...a) => this.traverseNext(...a), + }; + return executeNode.call(host, node, flow, variables, context, steps); } /** @@ -11490,7 +11052,7 @@ export class AutomationEngine implements IAutomationService { * `label` matches — this is how an Approval node's `approve`/`reject` * decision selects its downstream branch. */ - private async traverseNext( + private traverseNext( node: FlowNodeParsed, flow: FlowParsed, variables: Map, @@ -11498,185 +11060,13 @@ export class AutomationEngine implements IAutomationService { steps: StepLogEntry[], branchLabel?: string, ): Promise { - // Find next nodes — separate conditional, default and unconditional edges - const allOutEdges = flow.edges.filter( - e => e.source === node.id && e.type !== 'fault', - ); - let outEdges = allOutEdges; - - // Branch selection: prefer edges tagged with the decision label. - if (branchLabel) { - let claimed = outEdges.filter(e => e.label === branchLabel); - // The `default` sentinel is also claimed by the BPMN default edge, so - // "none of my conditions matched" routes to the declared fallback - // without the author having to ALSO label that edge 'default'. - if (claimed.length === 0 && branchLabel === DEFAULT_BRANCH_LABEL) { - claimed = outEdges.filter(e => e.isDefault); - } - if (claimed.length > 0) { - outEdges = claimed; - } else { - // #4414 — do not fall back silently. The node computed a branch - // and no out-edge claims it, so every out-edge is about to be - // considered: the guard the author wrote is not guarding. - // - // #6654 — the computed branch label is potentially - // RECORD-DERIVED and the edge labels are FLOW-AUTHOR metadata - // (neither is schema-constrained against newlines), so both - // ride the structured slot, never the message; see - // `forgetSuspendedRun`'s catch for the full mechanism (#6299). - // #4632: FUNCTIONAL — stays `warn`. - this.logger.warn( - // `flow.name` is absent on the synthetic view `runRegion` builds. - `Flow '${flow.name ?? '(region)'}' node '${node.id}' (${node.type}) selected a branch, ` + - `but no out-edge carries that label — the computed branch and the out-edge labels are ` + - `in this record's meta. The branch selection is IGNORED and every out-edge is ` + - `evaluated instead, so unconditional siblings run regardless of the decision. ` + - `Make an out-edge's \`label\` match the branch, or mark the fallback edge ` + - `\`isDefault: true\`.`, - { - branchLabel, - outEdges: allOutEdges.map(e => ({ id: e.id, label: e.label ?? null })), - }, - ); - } - } - - const conditionalEdges: FlowEdgeParsed[] = []; - const defaultEdges: FlowEdgeParsed[] = []; - const unconditionalEdges: FlowEdgeParsed[] = []; - for (const edge of outEdges) { - if (edge.condition) { - conditionalEdges.push(edge); - } else if (edge.isDefault) { - defaultEdges.push(edge); - } else { - unconditionalEdges.push(edge); - } - } - - // #4354 — a gate that did not open leaves a trace. Record it: this is - // THE event that had no trace anywhere, and the reason #4347 shipped - // three inert production flows. A closed gate inside a loop body is - // logged once per iteration (region tagging in `runRegion` attaches the - // container + iteration), so the run summary can say "selected 30, - // acted 0, skipped 30 by " instead of reporting a green run that - // did nothing. - // - // The step is `skipped`, never a run: the re-entrancy guard, per-node - // `runs` counts and node status all exclude it, so recording a - // non-event stays a non-event to execution. - const recordSkipped = (nextNode: FlowNodeParsed, edge: FlowEdgeParsed): void => { - const at = new Date().toISOString(); - steps.push({ - nodeId: nextNode.id, - nodeType: nextNode.type, - ...(nextNode.label ? { nodeLabel: nextNode.label } : {}), - status: 'skipped', - startedAt: at, - completedAt: at, - durationMs: 0, - skippedBy: { - nodeId: node.id, - ...(edge.id ? { edgeId: edge.id } : {}), - ...(edge.label ? { label: edge.label } : {}), - }, - }); + // The body lives in ./dag-traversal.ts (#25222). + const host: ThisParameterType = { + logger: this.logger, + evaluateCondition: (...a) => this.evaluateCondition(...a), + executeNode: (...a) => this.executeNode(...a), }; - - // Conditional edges: evaluated sequentially, in the order the flow's - // `edges` array declares them — `FlowSchema.parse` and every conversion - // are copy-on-write maps that never reorder, so declaration order here - // IS the author's. - // - // On a `decision` node they are the gateway's branches, and the gateway - // is EXCLUSIVE unless it declares `config.mode: 'inclusive'` (#15429, - // maintainer ruling 「跟主流对齐」): the FIRST edge whose condition holds - // is the branch — the BPMN exclusive gateway, Salesforce Flow's - // Decision, n8n's Switch default — and its later siblings are not - // evaluated at all; they record the same `skipped` step a closed gate - // does, so the run log says which branch won and which were passed - // over. Until this change every true edge ran, one after another, under - // a comment calling that "mutually exclusive": hotcrm#1555 rendered a - // refusal screen AND ran the conversion in one execution. Authored - // sources written against that behaviour are carried across by the - // ADR-0087 conversion `flow-decision-mode-inclusive-explicit` (`os - // migrate meta --from 17`), which writes the inclusive declaration onto - // them — see `CONVERSIONS_NOT_REPLAYED_AT_REHYDRATION` for why this seam - // does not. A flow stored in `sys_metadata` arrives here as stored and - // takes THIS reading (maintainer ruling letter C on #15429: no - // stored-row rewrite, no read-path completion); `os migrate meta - // --stored` lists such decisions for an operator to review. - // - // `mode: 'inclusive'` is the BPMN inclusive gateway: every edge whose - // condition holds runs, one successor at a time (never `Promise.all` — - // that fan-out belongs to the UNCONDITIONAL bucket below alone). - // - // Scoped to `decision` on purpose: `mode` is a decision-config key, and - // conditioned out-edges of any other node type keep the traversal they - // had — every one whose condition holds, sequentially. The ruling and - // its migration cover the gateway, and nothing else was measured to - // carry two conditioned out-edges (the corpus census on #15429). - const exclusive = node.type === 'decision' && !decisionTakesEveryBranch(node); - let anyConditionMet = false; - for (const edge of conditionalEdges) { - const nextNode = flow.nodes.find(n => n.id === edge.target); - if (exclusive && anyConditionMet) { - // A sibling already won: passed over, not evaluated. - if (nextNode) recordSkipped(nextNode, edge); - continue; - } - if (this.evaluateCondition(edge.condition!, variables, context)) { - anyConditionMet = true; - if (nextNode) { - await this.executeNode(nextNode, flow, variables, context, steps); - } - } else if (nextNode) { - recordSkipped(nextNode, edge); - } - } - - // Default edges (BPMN default flow, #4414): the fallback, taken only - // when NO conditional sibling matched. `isDefault` is what makes this an - // "otherwise" rather than a second unconditional path — without it the - // author's only spelling of "otherwise" was to hand-write the negation - // of every sibling condition, and forgetting to do that ran both - // branches. A default edge passed over because a real branch won records - // the same `skipped` trace a closed gate does (#4354). - for (const edge of defaultEdges) { - const nextNode = flow.nodes.find(n => n.id === edge.target); - if (!anyConditionMet) { - if (nextNode) { - await this.executeNode(nextNode, flow, variables, context, steps); - } - } else if (nextNode) { - const at = new Date().toISOString(); - steps.push({ - nodeId: nextNode.id, - nodeType: nextNode.type, - ...(nextNode.label ? { nodeLabel: nextNode.label } : {}), - status: 'skipped', - startedAt: at, - completedAt: at, - durationMs: 0, - skippedBy: { - nodeId: node.id, - ...(edge.id ? { edgeId: edge.id } : {}), - ...(edge.label ? { label: edge.label } : {}), - }, - }); - } - } - - // Unconditional edges: execute in parallel (Promise.all) - if (unconditionalEdges.length > 0) { - const parallelTasks = unconditionalEdges - .map(edge => flow.nodes.find(n => n.id === edge.target)) - .filter((n): n is FlowNodeParsed => n != null) - .map(nextNode => this.executeNode(nextNode, flow, variables, context, steps)); - - await Promise.all(parallelTasks); - } + return traverseNext.call(host, node, flow, variables, context, steps, branchLabel); } /** @@ -11736,140 +11126,18 @@ export class AutomationEngine implements IAutomationService { * that had processed nothing. The container executors test for the named * type and re-throw, so no region can contain it. */ - async runRegion( + runRegion( region: FlowRegionParsed, variables: Map, context: AutomationContext, grouping?: { parentNodeId: string; iteration?: number; branch?: number; regionKind?: string; retryAttempt?: number }, partialSteps?: StepLogEntry[], ): Promise { - const entryId = findRegionEntry(region); - const entry = region.nodes.find(n => n.id === entryId); - if (!entry) { - throw new Error(`region entry node '${entryId}' not found`); - } - // A synthetic flow view — executeNode/traverseNext only read `nodes`/`edges`. - const subFlow = { nodes: region.nodes, edges: region.edges ?? [] } as unknown as FlowParsed; - const regionSteps: StepLogEntry[] = []; - // Tag this region's steps with their container. Shared by the success - // and failure paths (#7546) so a failed attempt's steps are - // indistinguishable in SHAPE from a successful one's — they differ only - // in their own `status`. - // - // #15230 splits what "innermost wins" governs, because the two halves - // answer different questions: - // - // IDENTITY (`parentNodeId` / `regionKind` / `retryAttempt`) answers - // WHICH REGION RAN THIS STEP. Innermost wins outright: a step a - // nested region already claimed keeps naming that region, and an - // enclosing region must never relabel it. - // - // INDEX (`iteration` / `branch`) answers WHICH PASS OF WHICH REGION. - // Nested regions contribute DIFFERENT indices — the loop's row and - // the parallel's branch are both true of the same step — so an - // enclosing region fills the index the inner one left undefined - // instead of being skipped along with the identity fields. That skip - // is the defect: `loop { parallel }` used to record the branch and - // discard the row. - // - // Still "only fills what is undefined" in both halves, so an index a - // nested region DID set wins: for `loop { loop }` the inner loop's - // `iteration` stands, exactly as before. - const tag = (): void => { - if (!grouping) return; - for (const step of regionSteps) { - if (step.parentNodeId === undefined) { - step.parentNodeId = grouping.parentNodeId; - if (grouping.regionKind !== undefined) step.regionKind = grouping.regionKind; - if (grouping.retryAttempt !== undefined) step.retryAttempt = grouping.retryAttempt; - } - if (grouping.iteration !== undefined && step.iteration === undefined) { - step.iteration = grouping.iteration; - } - if (grouping.branch !== undefined && step.branch === undefined) { - step.branch = grouping.branch; - } - } + // The body lives in ./dag-traversal.ts (#25222). + const host: ThisParameterType = { + executeNode: (...a) => this.executeNode(...a), }; - try { - await this.executeNode(entry, subFlow, variables, context, regionSteps); - } catch (err) { - // #7546: surface what the failed attempt DID get through before - // rethrowing. Tagged first so the caller receives finished records, - // and pushed into the caller's sink rather than returned because - // this path's contract is (still) to throw. - tag(); - partialSteps?.push(...regionSteps); - // [#18881] A refusal raised at an INNER region boundary is already - // the named one, and it is re-thrown untouched. Re-wrapping it here - // would rename the region: for `loop { try_catch { map } }` the - // author's fault is the try region, and the loop is only the frame - // the unwind passes through. Tested before the suspend arm because - // this error is not a suspend signal and must not reach the generic - // rethrow below with an enclosing region's identity stamped on it. - if (isRegionSuspensionRefusal(err)) throw err; - // [#18881] The runtime half of #15646's ruling D: a region body - // cannot carry a durable pause, and the refusal is now NAMED - // (region node, suspending node, sub-flow) instead of a plain - // `Error` that an enclosing `try_catch` read as an ordinary region - // failure and handed to its catch handler. See - // `region-suspension-refusal.ts` for the measurement that is. - // - // The sub-flow is read off the suspending node's own `config` - // (`map` / `subflow` name their child there). The node is looked up - // in THIS region's `nodes` because this is the innermost boundary - // the signal crosses — a deeper suspension was already converted by - // the arm above, so `err.nodeId` always names a node of this body. - if (isSuspendSignal(err)) { - const suspended = region.nodes.find(n => n.id === err.nodeId); - const flowName = (suspended?.config as { flowName?: unknown } | undefined)?.flowName; - throw refuseRegionSuspension({ - regionNodeId: grouping?.parentNodeId ?? entryId, - regionKind: grouping?.regionKind ?? 'region', - suspendedNodeId: err.nodeId, - ...(typeof flowName === 'string' && flowName ? { subFlowName: flowName } : {}), - }); - } - // [#15788] The refusing `end` node's signal, converted at exactly - // the same boundary and for the same reason: a control signal must - // not cross a region edge silently. It would not merely leak — the - // `try_catch` executor's own `catch (err)` arm would read it as the - // try region FAILING and hand it to the catch handler, so an - // author's refusal would run the error path and the run would - // record `completed`. That is the pre-#15788 silence with an extra - // step, which is worse than a refusal that says so. - // - // ⛔ Not a narrowing of anything an author had: today an `end` in a - // region is a no-op whatever its `outcome`, so the shape being made - // loud has never once been honoured. Whether a refusal should - // instead PROPAGATE out of a region and terminate the run is a real - // question and ⛔ not one this lane rules on — the #14945 ruling - // says nothing about regions, and "prefer failing to falling back" - // decides the interim. - // - // [#18110 / #18555] The sentence NAMES the node that carried the - // refusal and nothing more, because there are now two producers: an - // `end` declaring the refusal itself, and a `subflow` / `map` whose - // CHILD run refused. Hard-wired to the first, it told an author - // inside a region to go find an `end` node that is not in their - // region at all, and handed them a prescription they could not - // follow. ⛔ TEXT only — region SEMANTICS are untouched (#18112's - // option B is not implemented, no container is taught to rethrow), - // and the authoring-time half of this boundary is #15646's, ⛔ not - // this change's. - if (isRefusalSignal(err)) { - throw new Error( - `a refusal inside a structured region (node '${err.nodeId}') is not supported — a ` + - `refusal terminates the RUN, and a region body cannot end one. The refusing node is ` + - `either an 'end' declaring outcome: 'refused', or a node whose own child run refused ` + - `(a 'subflow' / 'map'). Move that node onto the top-level graph and route the ` + - `region's exit to it.`, - ); - } - throw err; - } - tag(); - return regionSteps; + return runRegion.call(host, region, variables, context, grouping, partialSteps); } /** @@ -11896,24 +11164,13 @@ export class AutomationEngine implements IAutomationService { * health-monitor helpers, whose hooks may be synchronous): * `NodeExecutor.execute` is declared `Promise`-returning. */ - private async executeWithTimeout( + private executeWithTimeout( promise: Promise, timeoutMs: number, nodeId: string, ): Promise { - let guard: ReturnType | undefined; - - const timeoutPromise = new Promise((_, reject) => { - guard = setTimeout(() => { - reject(new Error(`Node '${nodeId}' timed out after ${timeoutMs}ms`)); - }, timeoutMs); - }); - - try { - return await Promise.race([promise, timeoutPromise]); - } finally { - clearTimeout(guard); - } + // The body lives in ./dag-traversal.ts (#25222). + return executeWithTimeout(promise, timeoutMs, nodeId); } /** @@ -11982,22 +11239,8 @@ export class AutomationEngine implements IAutomationService { variables: Map, context?: AutomationContext, ): { extra: Record } { - const vars: Record = {}; - for (const [key, value] of variables) { - // Convert "step.result" keys into nested object paths. - const segs = key.split('.'); - let cursor = vars; - for (let i = 0; i < segs.length - 1; i++) { - if (typeof cursor[segs[i]] !== 'object' || cursor[segs[i]] === null) { - cursor[segs[i]] = {}; - } - cursor = cursor[segs[i]] as Record; - } - cursor[segs[segs.length - 1]] = value; - } - // [#22642] No `record` slot: `record` is bound by the spread when the - // run holds one, and unbound otherwise (see the docblock). - return { extra: { ...vars, vars, current_user: runUserOf(context) } }; + // The body lives in ./dag-traversal.ts (#25222). + return celScope(variables, context); } /** @@ -12039,53 +11282,8 @@ export class AutomationEngine implements IAutomationService { * executor would have run. */ private valueEnvelopeRefusals(value: unknown): { message: string; source: string }[] { - // [#16439] Presence, asked before either primitive, because neither can - // answer it (see rule 0 above). It is stated HERE and not as a guard in - // {@link evaluateValueEnvelope} deliberately: a reject reason living - // only on the evaluation side would end the property this method exists - // for — registration and evaluation refuse ONE set, derived by one call - // — and the docblock that says so would stop being true. - // - // REFUSED rather than admitted, and the asymmetry with the predicate - // side is deliberate, not an oversight to harmonise away: - // `structuralConditionRefusal` admits `null` / `undefined` because the - // condition FIELD is optional, so absence there means "the author wrote - // no predicate". A value slot's envelope IS the value, so an absent one - // is a caller handing nothing where a value was required: there is - // nothing to compute, and — per this method's own ADR-0032 §1c - // argument — no falsy default for it to hide behind. - // - // Measured, so the registration half is not an assumption: the - // value-role feeder (`resolveFlowNodeExpressions`) emits ONLY - // envelope-shaped objects and `isExpressionEnvelopeShaped` is false for - // both shapes, so `registerFlow` never presents a nullish value here. - // An authored `null` in an `assignments` slot stays what it always was - // — a literal that parses (`FlowSchema.parse` accepts it) and still - // registers. This adds nothing to the registration reject set. - if (value == null) { - return [{ - message: - `${VALUE_ENVELOPE_REFUSAL} no envelope was passed: the argument is ` - + `\`${value === null ? 'null' : 'undefined'}\`, so there is nothing to evaluate. An absent ` - + 'envelope is not "not authored" — the predicate side admits absence because the condition ' - + "field is optional, but a value slot's envelope IS the value. Write " - + "`{ dialect: 'cel', source: '…' }`.", - source: '', - }]; - } - const source = (value as { source?: unknown })?.source; - const sourceText = typeof source === 'string' ? source : ''; - const shape = FlowValueSlotSchema.safeParse(value); - if (!shape.success) { - // Already prefixed with the refusal sentence by the spec's own - // `superRefine` — re-prefixing would say it twice. - return shape.error.issues.map((issue) => ({ message: issue.message, source: sourceText })); - } - const parsed = validateExpression('value', value as { dialect?: string; source?: string }); - return parsed.errors.map((e) => ({ - message: `${VALUE_ENVELOPE_REFUSAL} ${e.message}`, - source: e.source, - })); + // The body lives in ./dag-traversal.ts (#25222). + return valueEnvelopeRefusals(value); } /** @@ -12137,25 +11335,12 @@ export class AutomationEngine implements IAutomationService { where: string, context?: AutomationContext, ): unknown { - const refusals = this.valueEnvelopeRefusals(envelope); - if (refusals.length > 0) { - throw new Error( - `${where}: ${refusals.map((r) => r.message).join(' ')} — source: \`${refusals[0].source}\``, - ); - } - const source = envelope.source ?? ''; - const result = ExpressionEngine.evaluate({ dialect: 'cel', source }, this.celScope(variables, context)); - if (!result.ok) { - // Reached by the shapes the two validators above cannot judge — an - // `ast`-only envelope (`ExpressionSchema` accepts `source`-or-`ast`; - // the CEL engine evaluates `source` only) and a source that parses - // but faults on the live values. Loud, with the source attached. - throw new Error( - `${where}: value expression failed to evaluate as CEL: ${result.error?.message ?? 'unknown error'} — ` + - `source: \`${source}\`.`, - ); - } - return result.value; + // The body lives in ./dag-traversal.ts (#25222). + const host: ThisParameterType = { + valueEnvelopeRefusals: (...a) => this.valueEnvelopeRefusals(...a), + celScope: (...a) => this.celScope(...a), + }; + return evaluateValueEnvelope.call(host, envelope, variables, where, context); } /** @@ -12228,70 +11413,11 @@ export class AutomationEngine implements IAutomationService { variables: Map, context?: AutomationContext, ): boolean { - const shapeRefusal = structuralConditionRefusal(expression); - if (shapeRefusal) { - // ADR-0032 §1d — the error carries its source. `structuralConditionRefusal` - // attributes an empty one for the shape it is refusing here (a non-string - // `source` cannot be the attribution), which is its documented choice. - throw new Error( - `condition evaluation error: ${shapeRefusal.message} — source: \`${shapeRefusal.source}\``, - ); - } - - const isEnvelope = typeof expression === 'object' && expression != null && 'dialect' in expression; - const dialect = isEnvelope ? (expression as { dialect?: string }).dialect : undefined; - const exprStr = typeof expression === 'string' ? expression : ((expression as { source?: string })?.source ?? ''); - - if (dialect === 'template' || dialect === 'flow') { - // [#19939] The legacy `{var}` template dialect is deleted (see the - // doc comment): the same set `registerFlow` refuses at every - // condition slot, refused here for a caller that skipped it. - throw new Error( - `condition evaluation error: a condition is CEL, and the \`${dialect}\` dialect is not evaluated — ` + - `source: \`${exprStr}\`. Write the predicate as CEL without \`{…}\` template braces ` + - `(e.g. \`record.status == 'open'\`), as bare text or as \`{ dialect: 'cel', source }\`.`, - ); - } - if (isEnvelope && dialect && dialect !== 'cel') { - // Other dialects (cron, js) are not boolean predicates here. - return false; - } - - // An absent / empty condition is not a predicate to evaluate. Callers that - // mean "unconditional" guard before calling; this is the one that does not - // (a `decision` node whose `conditions[]` entry has no `expression`), and - // an unauthored branch must not open. - if (exprStr.trim() === '') return false; - - // Bind the `vars` scope, so a `$`-named variable or a node output's - // flat dotted key is read as `vars["…"]`. - try { - const result = ExpressionEngine.evaluate( - { dialect: 'cel', source: exprStr }, - this.celScope(variables, context), - ); - // ADR-0032 §Decision 1c — NO silent fallback. A non-`ok` result is a - // real fault (malformed predicate, or — pre build-validation — a - // `{…}` template mistakenly written into a CEL condition). Surfacing - // it as a thrown, attributed error makes execute()'s catch record a - // loud flow failure, instead of the old `return false` that made a - // broken condition indistinguishable from "condition not met" (#1491). - if (!result.ok) { - throw new Error( - `condition failed to evaluate as CEL: ${result.error?.message ?? 'unknown error'} — ` + - `source: \`${exprStr}\`. Conditions are bare CEL (e.g. \`record.rating >= 4\`); ` + - `do not wrap field references in \`{…}\` template braces.`, - ); - } - return Boolean(result.value); - } catch (err) { - // Re-throw with the source attached (ADR-0032 §1d — errors written - // for self-correction). Never swallow to `false`. - const msg = (err as Error)?.message ?? String(err); - throw new Error( - msg.includes('source:') ? msg : `condition evaluation error: ${msg} — source: \`${exprStr}\``, - ); - } + // The body lives in ./dag-traversal.ts (#25222). + const host: ThisParameterType = { + celScope: (...a) => this.celScope(...a), + }; + return evaluateCondition.call(host, expression, variables, context); } /** 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 7eae1b1e383..7b7cc524fb8 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']; + const ENGINE_SOURCES = ['./engine.ts', './flow-registration-checks.ts', './plugin-extension-api.ts', './dag-traversal.ts']; const engineSource = ENGINE_SOURCES.map((path) => readFileSync(fileURLToPath(new URL(path, import.meta.url)), 'utf8'), ).join('\n');