diff --git a/.changeset/contained-failure-visibility.md b/.changeset/contained-failure-visibility.md new file mode 100644 index 0000000000..fd9080a7bf --- /dev/null +++ b/.changeset/contained-failure-visibility.md @@ -0,0 +1,16 @@ +--- +"@objectstack/service-automation": minor +--- + +A contained per-iteration failure is now visible at run level, attributed to its iteration, and bound to its row. + +`loop { body: [ try_catch { try, catch } ] }` is the containment spelling for a per-iteration failure that must not end the sweep (there is deliberately no `loop.config.onIterationError` key). Containment already worked — the failure was caught, the loop went on and the run completed — but nothing said what it had contained: a sweep that lost two rows out of five reported `status=completed selected=5 acted=9 skipped=0` and was indistinguishable from one that lost none. The failure was in the step log and in `nodes[].failures`; no run-level number carried it, the failing step named no row, and `$error` bound no row identity. + +Four changes populate the contract `@objectstack/spec` already declares: + +- **`FlowRunSummary.failed`** — `summarizeRun` now folds `failed = Σ nodes[].failures` over the per-node array it publishes, so the run-level count can never disagree with the breakdown it summarizes. It counts every node execution that failed, contained or fatal; on a run that completed, all of them were contained. +- **`failed=N` on the run summary line** — `formatRunSummaryLine` prints the token whenever the count is present, `failed=0` included. That is the opposite of the `unmeasured` rule beside it and deliberate: `unmeasured` qualifies `acted`, while `failed` answers a question a completed run's line otherwise cannot be asked at all. Read `failed=0` precisely: **no node execution of this run failed**. It is the node fold and only that, so a `subflow` child's own contained failures stay on the child's summary rather than rolling up the way `acted` does — see #15617, where the declaration's two paragraphs are being reconciled. +- **Iteration through `try_catch`** — a step that ran in a `try` or `catch` region inside a loop body now carries the enclosing loop's `iteration`, with `regionKind` still `try` / `catch`. The step says which region ran it *and* which row it ran for. `parallel` branch tagging is unchanged. +- **`$error` binds the row** — the value bound to `errorVariable` (default `$error`) is the declared `TryCatchErrorValue`: `nodeId` and `message` as before, plus `iteration` and the loop's current `item` when the failure happened inside a loop body. A `subflow` / `map` child run has its own variable scope and therefore binds neither, so a parent's row identity never leaks into a child's `$error`. + +**`failed` absent means "not tracked", never `0`.** Runs recorded before this change keep it absent — no migration and no default, the same convention `unmeasured` carries. Defaulting it to zero would tell an operator "nothing failed" about a run nobody measured. Absent, the summary line prints no `failed=` token at all; present-and-zero prints `failed=0`. The count rides in the persisted `summary_json`, including on a summary compacted past the size cap, where the per-node `failures` it folds are exactly what gets dropped. diff --git a/content/docs/automation/flows.mdx b/content/docs/automation/flows.mdx index 436379d28f..dddfa96ac9 100644 --- a/content/docs/automation/flows.mdx +++ b/content/docs/automation/flows.mdx @@ -872,7 +872,7 @@ terminal run therefore carries a **summary** — on the `AutomationResult`, on t run in `listRuns` / `getRun`, and in the log: ``` -[automation] run flow=stalled_deal_sweep run=run_a1b2 status=completed durationMs=142 selected=30 acted=0 skipped=30 gate=check_stalled->send_nudge:30 +[automation] run flow=stalled_deal_sweep run=run_a1b2 status=completed durationMs=142 selected=30 acted=0 skipped=30 failed=0 gate=check_stalled->send_nudge:30 ``` | Field | Meaning | @@ -881,7 +881,7 @@ run in `listRuns` / `getRun`, and in the log: | `acted` | Records **created / updated / deleted**, plus effects dispatched (notifications delivered) | | `skipped` | Node executions a **closed gate** prevented — one per loop iteration whose conditional edge evaluated false | | `unmeasured` | Executions that reached something the platform **cannot count** — see below | -| `failed` | Node executions that **failed and were contained** — caught by a `try_catch` or routed down a `fault` edge — so the run went on; the sum of `nodes[].failures`. Absent on a run that did not track it, which is not zero | +| `failed` | Node executions **of this run** that failed — on a completed run every one of them was contained, caught by a `try_catch` or routed down a `fault` edge, so the run went on; the sum of `nodes[].failures`. `failed=0` therefore reads "no node execution of this run failed", which is narrower than "nothing failed anywhere": a `subflow` child's own contained failures are counted on the CHILD's summary, not folded up here the way `acted` is (that inconsistency in the declaration is [#15617](https://github.com/objectstack-ai/objectstack/issues/15617)). Absent on a run that did not track it, which is not zero | | `nodes[]` | Per-node terminal status with `runs` / `failures` / `skipped` and its own selected/acted | | `gates[]` | Which gates closed and how often, most-skipped first | diff --git a/packages/services/service-automation/src/builtin/contained-failure-visibility.test.ts b/packages/services/service-automation/src/builtin/contained-failure-visibility.test.ts new file mode 100644 index 0000000000..2a94ecb64e --- /dev/null +++ b/packages/services/service-automation/src/builtin/contained-failure-visibility.test.ts @@ -0,0 +1,424 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// #14456 — the visibility half of the ruled containment contract (#13681). +// +// `loop { body: [ try_catch { try, catch } ] }` is the containment spelling for +// a per-iteration failure that must not end the sweep (maintainer ruling +// 2026-08-31, branch B; there is deliberately no `loop.config.onIterationError` +// key). Containment already worked. What did not exist was any way to SEE what +// it contained: the measurement these tests reproduce (#13681) found a run that +// lost one row out of five reporting +// +// status=completed selected=5 acted=9 skipped=0 +// +// with nothing at run level naming the loss, the failing step carrying no +// iteration index, and `$error` binding no row identity at all. A sweep that +// lost two rows was indistinguishable from one that lost none. +// +// The fixture is that measurement's, on the real `AutomationEngine`: five rows, +// the THIRD one ownerless, `flag` always succeeding and `notify` failing on the +// ownerless row. + +import { describe, it, expect } from 'vitest'; +import { AutomationEngine } from '../engine.js'; +import type { NodeExecutor } from '../engine.js'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import type { FlowRunSummary } from '@objectstack/spec/automation'; +import { TryCatchErrorValueSchema, FlowRunSummarySchema } from '@objectstack/spec/automation'; +import { registerLoopNode } from './loop-node.js'; +import { registerTryCatchNode } from './try-catch-node.js'; +import { registerLogicNodes } from './logic-nodes.js'; +import { registerParallelNode } from './parallel-node.js'; +import { registerSubflowNode } from './subflow-node.js'; +import { formatRunSummaryLine, summarizeRun } from '../run-summary.js'; + +function silentLogger(): any { + const l: any = { info() {}, warn() {}, error() {}, debug() {} }; + l.child = () => l; + return l; +} +const pluginCtx = (logger: any) => ({ logger, getService() { throw new Error('none'); } }) as any; + +/** The five breached cases; the THIRD (index 2) is the one with no owner. */ +const CASES = [ + { id: 'c1', owner: 'u1' }, + { id: 'c2', owner: 'u2' }, + { id: 'c3', owner: null }, + { id: 'c4', owner: 'u4' }, + { id: 'c5', owner: 'u5' }, +]; + +describe('#14456 — a contained per-iteration failure is visible, attributed and bound', () => { + let engine: AutomationEngine; + let flagged: string[]; + let notified: string[]; + let caught: unknown[]; + + /** Rows whose `owner` is null are the ones `notify` refuses. */ + function setup(cases: Array<{ id: string; owner: string | null }>): void { + const logger = silentLogger(); + engine = new AutomationEngine(logger); + flagged = []; + notified = []; + caught = []; + + registerLoopNode(engine, pluginCtx(logger)); + registerTryCatchNode(engine, pluginCtx(logger)); + registerLogicNodes(engine, pluginCtx(logger)); + registerParallelNode(engine, pluginCtx(logger)); + registerSubflowNode(engine, pluginCtx(logger)); + + // Stands in for the `get_record` sweep query — seeds the collection and + // reports `selected` the way a real query node does (#4354). + engine.registerNodeExecutor({ + type: 'seed', + async execute(_node, variables) { + variables.set('cases', cases); + return { success: true, metrics: { selected: cases.length } }; + }, + } as NodeExecutor); + + // "flag the breach" — always succeeds, writes one row. Its presence is + // how the test reads "every iteration ran", independently of the fold. + engine.registerNodeExecutor({ + type: 'flag', + async execute(_node, variables) { + flagged.push((variables.get('currentCase') as { id: string }).id); + return { success: true, metrics: { acted: 1 } }; + }, + } as NodeExecutor); + + // "notify the owner" — fails on an ownerless row, as the real `notify` + // builtin does when every recipient template resolves to nothing. + engine.registerNodeExecutor({ + type: 'notify', + async execute(_node, variables) { + const c = variables.get('currentCase') as { id: string; owner: string | null }; + if (!c.owner) { + return { success: false, error: `notify: at least one recipient is required (${c.id})` }; + } + notified.push(c.id); + return { success: true, metrics: { acted: 1 } }; + }, + } as NodeExecutor); + + // The catch region's second node: an operator reading the binding. A + // completed run does not hand its variable map back, so the assertion + // has to happen inside the region — this is the only way to observe + // `$error` at the moment the handler sees it. + engine.registerNodeExecutor({ + type: 'capture', + async execute(_node, variables) { + caught.push(variables.get('$error')); + return { success: true }; + }, + } as NodeExecutor); + } + + /** + * The card's shape: a `try_catch` per iteration, `assignment` as the + * handler (the minimal working catch region), plus a `capture` probe. + */ + const sweepFlow = (name: string) => ({ + name, + label: name, + type: 'autolaunched' as const, + runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'query', type: 'seed', label: 'Query' }, + { + id: 'each', + type: 'loop', + label: 'Each case', + config: { + collection: '{cases}', + iteratorVariable: 'currentCase', + body: { + nodes: [ + { + id: 'guard', + type: 'try_catch', + label: 'Guarded', + config: { + try: { + nodes: [ + { id: 'flag', type: 'flag', label: 'Flag' }, + { id: 'notify', type: 'notify', label: 'Notify' }, + ], + edges: [{ id: 't1', source: 'flag', target: 'notify' }], + }, + catch: { + nodes: [ + { id: 'handled', type: 'assignment', label: 'Handled' }, + { id: 'seen', type: 'capture', label: 'Seen' }, + ], + edges: [{ id: 'k1', source: 'handled', target: 'seen' }], + }, + }, + }, + ], + edges: [], + }, + }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'query' }, + { id: 'e2', source: 'query', target: 'each' }, + { id: 'e3', source: 'each', target: 'end' }, + ], + }); + + async function runSweep(cases = CASES) { + setup(cases); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + const runs = await engine.listRuns('sweep'); + return { res, run: runs[0], summary: res.summary as FlowRunSummary }; + } + + // ── The measurement, reproduced ──────────────────────────────────────── + + it('completes all five iterations and reports the one it lost (#13681 comment 5478851960)', async () => { + const { res, run, summary } = await runSweep(); + + // (a)/(b)/(c) of the measurement: contained, and every row processed. + expect(res.success).toBe(true); + expect(run.status).toBe('completed'); + expect(flagged).toEqual(['c1', 'c2', 'c3', 'c4', 'c5']); + expect(notified).toEqual(['c1', 'c2', 'c4', 'c5']); + const bodySteps = run.steps.filter((s) => s.regionKind === 'loop-body'); + expect(bodySteps.map((s) => s.iteration)).toEqual([0, 1, 2, 3, 4]); + + // (d), the half that did not exist: the run-level count. + expect(summary.failed).toBe(1); + // …and it agrees with the breakdown it is declared to fold. + expect(summary.failed).toBe(summary.nodes.reduce((n, node) => n + node.failures, 0)); + expect(summary.nodes.find((n) => n.nodeId === 'notify')).toMatchObject({ runs: 5, failures: 1 }); + // The container recovered, so IT did not fail — which is exactly why a + // run-level counter was needed: `status` alone reports a clean sweep. + expect(summary.nodes.find((n) => n.nodeId === 'guard')).toMatchObject({ runs: 5, failures: 0, status: 'success' }); + }); + + it('counts two contained failures as two (the measurement\'s two-caught variant)', async () => { + const { summary } = await runSweep([ + { id: 'c1', owner: 'u1' }, + { id: 'c2', owner: null }, + { id: 'c3', owner: null }, + { id: 'c4', owner: 'u4' }, + { id: 'c5', owner: 'u5' }, + ]); + expect(summary.failed).toBe(2); + expect(notified).toEqual(['c1', 'c4', 'c5']); + }); + + // ── Item 3: iteration through try_catch → runRegion ──────────────────── + + it('tags the catch region\'s steps with the enclosing loop iteration, regionKind unchanged', async () => { + const { run } = await runSweep(); + + const handled = run.steps.filter((s) => s.nodeId === 'handled'); + expect(handled).toHaveLength(1); + expect(handled[0]).toMatchObject({ parentNodeId: 'guard', regionKind: 'catch', iteration: 2 }); + }); + + it('tags the try region\'s steps the same way — the failing step names its row', async () => { + const { run } = await runSweep(); + + const failing = run.steps.filter((s) => s.nodeId === 'notify' && s.status === 'failure'); + expect(failing).toHaveLength(1); + expect(failing[0]).toMatchObject({ parentNodeId: 'guard', regionKind: 'try', iteration: 2 }); + + // Every try-region step carries the row it ran for, not only the failing one. + expect(run.steps.filter((s) => s.nodeId === 'flag').map((s) => s.iteration)).toEqual([0, 1, 2, 3, 4]); + }); + + // ── Item 4: `$error` binds the row ───────────────────────────────────── + + it('binds $error.iteration and $error.item to the row that failed', async () => { + await runSweep(); + + expect(caught).toHaveLength(1); + expect(caught[0]).toMatchObject({ nodeId: 'guard', iteration: 2, item: CASES[2] }); + // The binding is the declared `TryCatchErrorValue`, not a shape of this + // executor's own invention. + expect(TryCatchErrorValueSchema.safeParse(caught[0]).success).toBe(true); + }); + + it('binds NEITHER outside a loop — absent means "not in a loop", never "row unknown"', async () => { + setup(CASES); + engine.registerFlow('bare', { + name: 'bare', label: 'bare', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { + id: 'guard', type: 'try_catch', label: 'Guarded', + config: { + try: { nodes: [{ id: 'notify', type: 'notify', label: 'Notify' }], edges: [] }, + catch: { nodes: [{ id: 'seen', type: 'capture', label: 'Seen' }], edges: [] }, + }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'guard' }, + { id: 'e2', source: 'guard', target: 'end' }, + ], + } as never); + // No loop, so nothing binds `currentCase`; `notify` refuses it. + await engine.execute('bare', { event: 'manual' } as AutomationContext); + + expect(caught).toHaveLength(1); + expect(caught[0]).not.toHaveProperty('iteration'); + expect(caught[0]).not.toHaveProperty('item'); + }); + + it('does not leak a parent run\'s row identity into a subflow child (#14456)', async () => { + setup(CASES); + // The child owns the try_catch; the PARENT owns the loop. The child run + // gets its own variable scope, so the frame the parent published is not + // the child's — the binding must report no row. + engine.registerFlow('child', { + name: 'child', label: 'child', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { + id: 'guard', type: 'try_catch', label: 'Guarded', + config: { + try: { nodes: [{ id: 'notify', type: 'notify', label: 'Notify' }], edges: [] }, + catch: { nodes: [{ id: 'seen', type: 'capture', label: 'Seen' }], edges: [] }, + }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'guard' }, + { id: 'e2', source: 'guard', target: 'end' }, + ], + } as never); + engine.registerFlow('parent', { + name: 'parent', label: 'parent', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'query', type: 'seed', label: 'Query' }, + { + id: 'each', type: 'loop', label: 'Each', + config: { + collection: '{cases}', iteratorVariable: 'currentCase', + body: { + nodes: [{ id: 'call', type: 'subflow', label: 'Call', config: { flowName: 'child' } }], + edges: [], + }, + }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'query' }, + { id: 'e2', source: 'query', target: 'each' }, + { id: 'e3', source: 'each', target: 'end' }, + ], + } as never); + + await engine.execute('parent', { event: 'schedule' } as AutomationContext); + + expect(caught.length).toBeGreaterThan(0); + for (const bound of caught) { + expect(bound).not.toHaveProperty('iteration'); + expect(bound).not.toHaveProperty('item'); + } + }); + + // ── The fence: `parallel` is untouched ───────────────────────────────── + + it('leaves `parallel` branch tagging exactly as it was — the #14414 fence', async () => { + setup(CASES); + engine.registerFlow('par', { + name: 'par', label: 'par', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'query', type: 'seed', label: 'Query' }, + { + id: 'each', type: 'loop', label: 'Each', + config: { + collection: '{cases}', iteratorVariable: 'currentCase', + body: { + nodes: [{ + id: 'fan', type: 'parallel', label: 'Fan', + config: { + branches: [ + { name: 'a', nodes: [{ id: 'flag', type: 'flag', label: 'Flag' }], edges: [] }, + { name: 'b', nodes: [{ id: 'noop', type: 'assignment', label: 'Noop' }], edges: [] }, + ], + }, + }], + edges: [], + }, + }, + }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'query' }, + { id: 'e2', source: 'query', target: 'each' }, + { id: 'e3', source: 'each', target: 'end' }, + ], + } as never); + + await engine.execute('par', { event: 'schedule' } as AutomationContext); + const run = (await engine.listRuns('par'))[0]; + + // A branch step still carries its BRANCH index on `iteration` and + // `regionKind: 'parallel-branch'`, five times over (once per row) — the + // pre-existing overload #14414 owns. Nothing here changed it. + const branchSteps = run.steps.filter((s) => s.regionKind === 'parallel-branch'); + expect(branchSteps).toHaveLength(10); + expect(branchSteps.filter((s) => s.nodeId === 'flag').every((s) => s.iteration === 0)).toBe(true); + expect(branchSteps.filter((s) => s.nodeId === 'noop').every((s) => s.iteration === 1)).toBe(true); + }); +}); + +// ── Item 5 + the `unmeasured` convention ─────────────────────────────────── + +describe('#14456 — `failed=` on the summary line, and the absent convention', () => { + const AT = '2026-09-04T00:00:00.000Z'; + const params = { flowName: 'sweep', runId: 'run_1', status: 'completed' }; + + it('prints `failed=N` for a summary that carries the count', () => { + const line = formatRunSummaryLine(params, summarizeRun([ + { nodeId: 'q', nodeType: 'seed', status: 'success', startedAt: AT, metrics: { selected: 5 } }, + { nodeId: 'notify', nodeType: 'notify', status: 'failure', startedAt: AT }, + ])); + expect(line).toBe('[automation] run flow=sweep run=run_1 status=completed selected=5 acted=0 skipped=0 failed=1'); + }); + + it('prints `failed=0` when the count is present and zero — "nothing failed" is a reading', () => { + const line = formatRunSummaryLine(params, summarizeRun([ + { nodeId: 'w', nodeType: 'assignment', status: 'success', startedAt: AT, metrics: { acted: 1 } }, + ])); + expect(line).toContain('failed=0'); + }); + + it('prints NOTHING for a summary recorded before the count existed', () => { + // Absent is "not tracked", not zero — the same convention `unmeasured` + // carries, and the one a default would erase. + const older: FlowRunSummary = { selected: 5, acted: 9, skipped: 0, unmeasured: 0, nodes: [], gates: [] }; + expect(formatRunSummaryLine(params, older)).not.toContain('failed'); + }); + + it('a row persisted before this change still parses, with `failed` ABSENT — no migration, no default', () => { + const stored = { selected: 5, acted: 9, skipped: 0, unmeasured: 0, nodes: [], gates: [] }; + const parsed = FlowRunSummarySchema.safeParse(stored); + expect(parsed.success).toBe(true); + expect(parsed.data?.failed).toBeUndefined(); + expect('failed' in (parsed.data ?? {})).toBe(false); + }); + + it('summarizeRun always emits the count, `0` included — only stored rows are absent', () => { + const s = summarizeRun([{ nodeId: 'w', nodeType: 'assignment', status: 'success', startedAt: AT }]); + expect(s.failed).toBe(0); + expect(FlowRunSummarySchema.safeParse(s).success).toBe(true); + }); +}); diff --git a/packages/services/service-automation/src/builtin/loop-frame.ts b/packages/services/service-automation/src/builtin/loop-frame.ts new file mode 100644 index 0000000000..a10242cba2 --- /dev/null +++ b/packages/services/service-automation/src/builtin/loop-frame.ts @@ -0,0 +1,73 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import { AsyncLocalStorage } from 'node:async_hooks'; + +/** + * The enclosing `loop` iteration a nested container is running for (#14456). + * + * `loop { body: [ try_catch { try, catch } ] }` is the ruled containment + * spelling for a per-iteration failure that must not end the sweep (maintainer + * ruling 2026-08-31, branch B). The catch region therefore has to know WHICH + * ROW failed — to record it, to notify about it, to retry it later — and the + * `try_catch` executor is handed only its own node, the shared variable scope + * and the trigger context. None of the three names the row: `iteratorVariable` + * is the loop's config, not the container's, and the shared scope holds the + * item under a name only the loop knows. + * + * So the loop publishes the row identity for the duration of the body region + * and a nested container reads it. Two properties make that safe: + * + * - **`AsyncLocalStorage`, not a field on the engine.** A body region is + * `await`ed, and two `parallel` branches — or two whole loops inside one + * `parallel` — run their bodies CONCURRENTLY over the same engine. A stored + * "current iteration" would be read by whichever continuation resumed next; + * an ALS frame is scoped to the async subtree that established it, so each + * iteration reads its own. Same primitive the ObjectQL engine uses for its + * ambient transaction, for the same reason. + * - **The frame names the SCOPE it belongs to**, and a reader must present a + * matching one. A region runs in the ENCLOSING variable scope — the same + * `Map` identity — so a `try_catch` that receives the loop's map is inside + * that loop's body. A `subflow` / `map` child run gets a FRESH map, and its + * executors therefore read no frame even though the ALS context propagates + * into the child: a parent's row identity never leaks onto a child run's + * `$error`, which nothing else here would have prevented. + * + * Innermost wins, for free: nested loops nest their frames, so the inner + * loop's iteration is the one a `try_catch` between them reads — which is the + * `iteration` the run log's own "innermost container" tagging rule records. + */ +export interface LoopIterationFrame { + /** Zero-based index of the iteration whose body is running. */ + readonly iteration: number; + /** The value bound to the loop's `iteratorVariable` for this iteration. */ + readonly item: unknown; + /** + * The variable scope the loop bound the item in. A reader passes the scope + * it was itself handed and gets the frame only when the two are the same + * `Map` — see the class doc. + */ + readonly scope: Map; +} + +const frames = new AsyncLocalStorage(); + +/** + * Run one loop iteration's body with `frame` published to everything it + * `await`s. The return value and any throw pass through untouched: this wraps + * the body region's execution, it does not participate in it. + */ +export function runInLoopIteration(frame: LoopIterationFrame, body: () => Promise): Promise { + return frames.run(frame, body); +} + +/** + * The enclosing loop iteration for a node executing in `scope`, or `undefined` + * when there is none — no loop above it, or a loop whose scope this is not + * (a `subflow` / `map` child run). "Absent" is the answer "not inside a loop + * body", which is exactly what a `TryCatchErrorValue` without `iteration` / + * `item` means. + */ +export function currentLoopIteration(scope: Map): LoopIterationFrame | undefined { + const frame = frames.getStore(); + return frame !== undefined && frame.scope === scope ? frame : undefined; +} diff --git a/packages/services/service-automation/src/builtin/loop-node.ts b/packages/services/service-automation/src/builtin/loop-node.ts index 85a97c8618..f3a31c80b9 100644 --- a/packages/services/service-automation/src/builtin/loop-node.ts +++ b/packages/services/service-automation/src/builtin/loop-node.ts @@ -8,6 +8,7 @@ import type { AutomationEngine, StepLogEntry } from '../engine.js'; import { interpolate } from './template.js'; import { parseNodeConfig } from './parse-config.js'; import { attachPartialSteps } from '../partial-steps.js'; +import { runInLoopIteration } from './loop-frame.js'; /** * `loop` built-in node — a **structured iteration container** (ADR-0031). @@ -131,16 +132,25 @@ export function registerLoopNode(engine: AutomationEngine, ctx: PluginContext): // FAILING iteration's steps land here too — `runRegion` tags them and // pushes them in before it rethrows. Success returns them instead, so // nothing is counted twice. - const iterSteps = await engine.runRegion( - body, - variables, - context ?? ({} as AutomationContext), - { - parentNodeId: node.id, - iteration: i, - regionKind: 'loop-body', - }, - childSteps, + // #14456: publish the row identity for the duration of this + // iteration's body, so a `try_catch` nested in it can attribute a + // CONTAINED failure to the row that produced it — `$error.iteration` + // / `$error.item`, and the loop's `iteration` on the try/catch + // region's own steps. Nothing about the iteration itself changes: + // `runInLoopIteration` returns and throws exactly what the body does. + const iterSteps = await runInLoopIteration( + { iteration: i, item: collection[i], scope: variables }, + () => engine.runRegion( + body, + variables, + context ?? ({} as AutomationContext), + { + parentNodeId: node.id, + iteration: i, + regionKind: 'loop-body', + }, + childSteps, + ), ); childSteps.push(...iterSteps); iterations++; diff --git a/packages/services/service-automation/src/builtin/try-catch-node.ts b/packages/services/service-automation/src/builtin/try-catch-node.ts index d13299106f..04359f3391 100644 --- a/packages/services/service-automation/src/builtin/try-catch-node.ts +++ b/packages/services/service-automation/src/builtin/try-catch-node.ts @@ -2,10 +2,11 @@ import type { PluginContext } from '@objectstack/core'; import { defineActionDescriptor, TryCatchConfigSchema } from '@objectstack/spec/automation'; -import type { TryCatchConfigParsed } from '@objectstack/spec/automation'; +import type { TryCatchConfigParsed, TryCatchErrorValue } from '@objectstack/spec/automation'; import type { AutomationContext } from '@objectstack/spec/contracts'; import type { AutomationEngine, StepLogEntry } from '../engine.js'; import { parseNodeConfig } from './parse-config.js'; +import { currentLoopIteration } from './loop-frame.js'; /** * `try_catch` built-in node — **structured try/catch/retry** (ADR-0031 §Decision 3). @@ -17,7 +18,9 @@ import { parseNodeConfig } from './parse-config.js'; * message }`, plus `code` (#14419) when the failing node's own result set a * platform-classified one (e.g. `create_record`'s `DUPLICATE_RECORD`), so the * catch region can branch on `{$error.code}` instead of only ever seeing a - * string. Both regions are self-contained single-entry/single-exit sub-graphs + * string, plus `iteration` and `item` (#14456) when the container is running + * inside a `loop` body, so a caught per-row failure names the ROW it lost and + * not merely that one was lost. Both regions are self-contained single-entry/single-exit sub-graphs * validated at `registerFlow()`, executed in the **enclosing variable scope** * via {@link AutomationEngine.runRegion}. * @@ -67,7 +70,7 @@ export function registerTryCatchNode(engine: AutomationEngine, ctx: PluginContex description: 'Handler region run when the try region fails', properties: { nodes: { type: 'array' }, edges: { type: 'array' } }, }, - errorVariable: { type: 'string', description: 'Variable holding the caught error in the catch region' }, + errorVariable: { type: 'string', description: 'Variable holding the caught error in the catch region — a `TryCatchErrorValue`: `nodeId`, `message`, and `iteration` / `item` when the failure happened inside a loop body' }, retry: { type: 'object', properties: { @@ -101,6 +104,14 @@ export function registerTryCatchNode(engine: AutomationEngine, ctx: PluginContex const retry = cfg.retry; const ctxOrEmpty = context ?? ({} as AutomationContext); + // #14456 — the enclosing loop's row identity, when this container is a + // loop body's containment wrapper. `undefined` outside a loop, and that + // absence is the contract's own answer ("not in a loop"), not a gap: + // `TryCatchErrorValueSchema` declares `iteration` / `item` present only + // for a failure that happened inside a loop body. Read ONCE here rather + // than per region: the frame cannot change while this node executes, and + // one read is one place for the next reader to look. + const loopFrame = currentLoopIteration(variables); const maxRetries = retry?.maxRetries ?? 0; const baseDelay = retry?.backoffMs ?? 0; const multiplier = retry?.backoffMultiplier ?? 1; @@ -169,6 +180,15 @@ export function registerTryCatchNode(engine: AutomationEngine, ctx: PluginContex { parentNodeId: node.id, regionKind: 'try', + // #14456 — forward the ENCLOSING loop's iteration so a step this + // region ran says which region ran it AND which row it ran for. + // `runRegion`'s tagger fills only fields the INNERMOST tagger + // left undefined, so a loop's own tagger can never reach past + // this one to a try/catch step; forwarding at this call site is + // what closes that, and it leaves both the tagger and `parallel` + // untouched (a branch step already carries its own `iteration`, + // and nothing here changes what `parallel` writes). + ...(loopFrame ? { iteration: loopFrame.iteration } : {}), // Only tag the attempt index when a retry ladder is actually // declared: on a plain `try_catch` every step would carry a // constant `retryAttempt: 0`, which is noise rather than signal. @@ -198,11 +218,27 @@ export function registerTryCatchNode(engine: AutomationEngine, ctx: PluginContex // The try region (and any retries) failed. Run the catch handler if present. if (catchRegion != null) { - variables.set(errorVariable, { + // #14456 — the caught error is a `TryCatchErrorValue` (declared in + // `packages/spec`, PR #14452): `nodeId` + `message`, plus the ROW + // IDENTITY when the failure happened inside a loop body. Without + // `iteration` / `item` a caught per-row failure is attributable to no + // row — the catch region can record THAT something failed and never + // WHICH thing, and `message` names one only when it happens to echo + // the template. + // + // `code` (#14419) is bound alongside but is NOT declared on + // `TryCatchErrorValueSchema`, so it is spelled as an explicit widening + // of the declared type rather than dropped — dropping it would regress + // a catch region's ability to branch on `{$error.code}`. The + // divergence is filed as #14954 against the spec lane; this file is + // not the place to change the contract. + const errorValue: TryCatchErrorValue & { code?: string } = { nodeId: node.id, message: lastError, ...(lastErrorCode ? { code: lastErrorCode } : {}), - }); + ...(loopFrame ? { iteration: loopFrame.iteration, item: loopFrame.item } : {}), + }; + variables.set(errorVariable, errorValue); // #14222: sink for the catch region's OWN partial steps, filled by // `runRegion` only if the handler itself throws. Without it the catch // region's completed steps unwound with the stack exactly as the try @@ -217,6 +253,10 @@ export function registerTryCatchNode(engine: AutomationEngine, ctx: PluginContex { parentNodeId: node.id, regionKind: 'catch', + // #14456 — same forwarding as the try region above: the handler's + // steps carry the row they handled, with `regionKind` still + // naming the region. + ...(loopFrame ? { iteration: loopFrame.iteration } : {}), }, catchAttemptSteps, ); diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index c5f075e207..d765ab2e0c 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -6575,6 +6575,10 @@ export class AutomationEngine implements IAutomationService { acted: entry.summary.acted, skipped: entry.summary.skipped, unmeasured: entry.summary.unmeasured, + // #14456 — the structured half of the `failed=` token the line + // above now carries, so a log pipeline that reads `meta` and + // never parses the message sees the same fact. + failed: entry.summary.failed, gates: entry.summary.gates, }; if (this.runSummaryLog === 'debug') this.logger.debug(line, meta); diff --git a/packages/services/service-automation/src/run-summary.test.ts b/packages/services/service-automation/src/run-summary.test.ts index e78f85e9c5..4434abe75a 100644 --- a/packages/services/service-automation/src/run-summary.test.ts +++ b/packages/services/service-automation/src/run-summary.test.ts @@ -150,7 +150,10 @@ describe('formatRunSummaryLine', () => { expect(line).not.toContain('\n'); expect(line).toBe( '[automation] run flow=stalled_deal_sweep run=run_1 status=completed durationMs=42 ' + - 'selected=30 acted=0 skipped=30 gate=gate->nudge:30', + // `failed=0` rides between `skipped` and the optional tokens + // (#14456): a summary this fold produced always carries the count, + // so the always-present prefix stays greppable as one unit. + 'selected=30 acted=0 skipped=30 failed=0 gate=gate->nudge:30', ); }); }); @@ -891,7 +894,7 @@ describe('uncountable effects (#4354 follow-up)', () => { ); // The line an operator greps for `acted=0` must carry the qualifier that // says the zero is incomplete. - expect(murky).toContain('selected=9 acted=0 skipped=0 unmeasured=1'); + expect(murky).toContain('selected=9 acted=0 skipped=0 failed=0 unmeasured=1'); }); it('an http node reports by METHOD — the one case the platform CAN count', async () => { @@ -1019,7 +1022,14 @@ describe('uncountable effects (#4354 follow-up)', () => { expect(res.summary).toMatchObject({ acted: 0, unmeasured: 1 }); }); - it('persists as a queryable column, null when never tracked', async () => { + /** + * ONE engine double for the three `recordTerminal` pins in this describe, + * rather than one apiece. `check:engine-double-contract` counts unguarded + * doubles PER FILE against a shrink-only baseline, so three copies of the + * same four members would have been three new ledger rows for one fact — + * and the gate's own advice is to reuse the double the file already has. + */ + function recordingRunStore(): { store: ObjectStoreSuspendedRunStore; rows: any[] } { const rows: any[] = []; const engine: any = { async find() { return []; }, @@ -1027,7 +1037,11 @@ describe('uncountable effects (#4354 follow-up)', () => { async update() { return 1; }, async delete() { return 1; }, }; - const store = new ObjectStoreSuspendedRunStore(engine); + return { store: new ObjectStoreSuspendedRunStore(engine), rows }; + } + + it('persists as a queryable column, null when never tracked', async () => { + const { store, rows } = recordingRunStore(); await store.recordTerminal({ runId: 'r1', flowName: 'f', status: 'completed', startedAt: AT, summary: { selected: 9, acted: 0, skipped: 0, unmeasured: 3, nodes: [], gates: [] }, @@ -1037,6 +1051,50 @@ describe('uncountable effects (#4354 follow-up)', () => { expect(rows[0].unmeasured_count).toBe(3); expect(rows[1].unmeasured_count).toBeNull(); // not measured ≠ measured zero }); + + // #14456 — the run-level failure count rides in `summary_json` with the + // rest of the fold. It gets no column of its own here: `selected/acted/ + // skipped/unmeasured` are columns because the broken-sweep FILTER queries + // them, and a contained failure is read from the run row, not alerted on + // by that filter. + it('carries the #14456 failure count through the persisted summary, absent when never tracked', async () => { + const { store, rows } = recordingRunStore(); + await store.recordTerminal({ + runId: 'r1', flowName: 'f', status: 'completed', startedAt: AT, + summary: summarizeRun([ + step({ nodeId: 'q', nodeType: 'get_record', metrics: { selected: 5 } }), + step({ nodeId: 'notify', nodeType: 'notify', status: 'failure' }), + ]), + }); + // A row written before the count existed — no `failed`, and nothing + // may invent one for it. + await store.recordTerminal({ + runId: 'r2', flowName: 'f', status: 'completed', startedAt: AT, + summary: { selected: 5, acted: 9, skipped: 0, unmeasured: 0, nodes: [], gates: [] }, + }); + + expect(JSON.parse(rows[0].summary_json).failed).toBe(1); + expect('failed' in JSON.parse(rows[1].summary_json)).toBe(false); + }); + + it('keeps the failure count when the detail is dropped for size', async () => { + const { store, rows } = recordingRunStore(); + // A pathological flow: enough nodes to blow the 16 KiB summary cap, so + // the per-node `failures` this count folds is exactly what gets + // dropped. The total has to survive the drop or the compacted row goes + // from "some rows failed" to silence. + const many = Array.from({ length: 600 }, (_, i) => + step({ nodeId: `n_${i}_padding_to_blow_the_summary_cap`, nodeType: 'assignment', status: i % 3 === 0 ? 'failure' : 'success' }), + ); + await store.recordTerminal({ + runId: 'r3', flowName: 'f', status: 'completed', startedAt: AT, summary: summarizeRun(many), + }); + + const stored = JSON.parse(rows[0].summary_json); + expect(stored.detailOmitted).toBe(true); + expect(stored.nodes).toEqual([]); + expect(stored.failed).toBe(200); + }); }); // ── The engine's own contract with executors ──────────────────────────────── diff --git a/packages/services/service-automation/src/run-summary.ts b/packages/services/service-automation/src/run-summary.ts index b6b13a7169..aea3f2b90f 100644 --- a/packages/services/service-automation/src/run-summary.ts +++ b/packages/services/service-automation/src/run-summary.ts @@ -46,6 +46,19 @@ import type { StepLogEntry } from './engine.js'; * its own metrics. The parent therefore answers "what did this run cause", * including through its subflows — otherwise a sweep that delegates its writes * would report `acted: 0` and trip the very detector this exists to feed. + * + * #14456 adds the run-level `failed` counter — `Σ nodes[].failures` — which is + * the count a GREEN run hides. `loop { body: [ try_catch { try, catch } ] }` is + * the containment spelling for a per-iteration failure that must not end the + * sweep (maintainer ruling 2026-08-31, branch B): the failure is caught, the + * loop goes on and the run completes. Until this counter existed the run row + * said nothing about it — the caught failure was in the step log and in + * `nodes[].failures`, but no run-level number carried it, so a run that lost + * two rows out of five was indistinguishable from one that lost none. + * + * It is folded from the per-node array rather than accumulated alongside it, + * so the run-level count can never disagree with the breakdown it summarizes; + * `unmeasured` above is a genuinely independent tally and stays one. */ export function summarizeRun(steps: readonly StepLogEntry[]): FlowRunSummary { // Mutable accumulators; `status` is decided once the counts are final. @@ -119,11 +132,16 @@ export function summarizeRun(steps: readonly StepLogEntry[]): FlowRunSummary { } } + // #14456 — `failed = Σ nodes[].failures`, stated as a fold over the SAME + // array the summary publishes rather than as a second accumulator in the + // step loop above. Two counters over one fact drift; one addition cannot. + let failed = 0; for (const node of nodes.values()) { // Worst outcome wins: one failed iteration makes the node's run-level // status `failure`, and `runs`/`failures` carry the nuance. A node that // only ever got skipped never ran at all. node.status = node.failures > 0 ? 'failure' : node.runs > 0 ? 'success' : 'skipped'; + failed += node.failures; } return { @@ -131,6 +149,11 @@ export function summarizeRun(steps: readonly StepLogEntry[]): FlowRunSummary { acted, skipped, unmeasured, + // Every node execution that failed, contained or fatal. A summary this + // function produced ALWAYS carries it, `0` included; the declared + // `undefined` case belongs to rows persisted before the counter + // existed, and means "not tracked", never "nothing failed". + failed, nodes: [...nodes.values()], gates: [...gates.values()].sort((a, b) => b.skipped - a.skipped), }; @@ -171,6 +194,29 @@ export function formatRunSummaryLine( // Only when non-zero, like `gate=`: its absence is the common case and its // PRESENCE is the thing a reader must not miss — `acted=0` on a line that // also says `unmeasured=3` means "cannot tell", not "did nothing". + // #14456 — printed whenever PRESENT, `failed=0` included, which is the + // opposite of the `unmeasured` rule directly above and deliberately so. + // `unmeasured` is a qualifier on `acted`: absent, the zero beside it is + // trustworthy. `failed` answers a question the line otherwise cannot be + // asked at all — a completed run says nothing about the rows it lost — so + // the token has to be there to be read. + // + // What `failed=0` says, exactly: NO NODE EXECUTION OF THIS RUN FAILED. + // That is narrower than "nothing failed", and the difference is a + // `subflow`: the fold this prints is `Sigma nodes[].failures` over THIS + // run's own nodes, so a child run that CONTAINED failures of its own + // reports them on the child's summary and the parent still prints + // `failed=0` — measured, alongside the control where a child that FAILS + // rather than contains does reach the parent's count through the + // `subflow` node's own failure step. `acted` rolls a child's totals up + // and this does not; the declaration says both things in two paragraphs + // and is being reconciled in #15617. Until it is, this line is the node + // fold, and only that. + // + // A line with no token at all is a different reading again — the older + // "not tracked". A run summarized by `summarizeRun` always carries the + // count; only a summary persisted before this existed prints nothing here. + if (summary.failed !== undefined) parts.push(`failed=${summary.failed}`); if (summary.unmeasured) parts.push(`unmeasured=${summary.unmeasured}`); const topGate = summary.gates[0]; if (topGate) { diff --git a/packages/services/service-automation/src/suspended-run-store.ts b/packages/services/service-automation/src/suspended-run-store.ts index 78f65ceddb..03b860aa39 100644 --- a/packages/services/service-automation/src/suspended-run-store.ts +++ b/packages/services/service-automation/src/suspended-run-store.ts @@ -853,6 +853,13 @@ function serializeSummaryBounded(summary: NonNullable): st acted: summary.acted, skipped: summary.skipped, unmeasured: summary.unmeasured, + // #14456 — a TOTAL, so it survives the detail drop with the others. It is + // also the one this branch would hurt most by losing: the per-node + // `failures` it folds are exactly what gets dropped here, so a summary + // over the cap would otherwise go from "two rows failed" to silence. + // `undefined` (an older summary that never tracked it) stringifies away, + // which keeps absent meaning "not tracked" on the compacted row too. + failed: summary.failed, nodes: [], gates: [], detailOmitted: true,