From 95b29eae5aa0e432b96d1fd8371041aaad34498d Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 31 Jul 2026 09:56:06 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat(spec,service-automation):=20surface=20?= =?UTF-8?q?per-run=20flow=20summaries=20=E2=80=94=20selected=20/=20acted?= =?UTF-8?q?=20/=20skipped=20(#4354)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `success: true` never meant "it did its job". A scheduled sweep that selects thirty records and writes none is, from outside, identical to one with nothing to do: same green status, same empty output, same silence, same schedule tomorrow. Nothing anywhere separated "nothing to do" from "broken" — which is how #4347 left three hotcrm production flows completely inert, running daily and green for as long as they had existed. Every terminal run now carries a `FlowRunSummary`: `selected` / `acted` / `skipped` totals, a per-node breakdown, and which gates closed. It is surfaced on `AutomationResult`, in `listRuns` / `getRun`, as one greppable log line, and as queryable `selected_count` / `acted_count` / `skipped_count` columns on `sys_automation_run` — so `selected > 0 AND acted = 0` over consecutive runs is something an operator can alert on rather than notice. The counts are declared by the executors (`NodeExecutionResult.metrics`), not sniffed from node output: only the node knows whether its result is a row count, a record, or a boolean. A closed conditional edge now records a `skipped` step naming the gate — the event that previously left no trace at all — and that step is explicitly not a run, so the ADR-0044 re-entry guard and the per-node `runs` counts are unchanged. Two details decide whether the numbers can be trusted: the summary is folded from the full step log before history compaction (and rehydrated from `summary_json`, never re-folded from the compacted steps), and `subflow` / `map` children roll up into their parent — synchronously via `metrics`, and via `creditChildRun` when the child paused and bubbles back. Additive throughout: `summary` is optional everywhere, legacy rows keep `null` counts rather than `0`, and no execution behaviour changes. The log line defaults to `info`, with `runSummaryLog: 'debug' | 'off'` to lower the volume without turning the measurement off. Verified: service-automation 525/46 (26 new), spec 7165/279 (5 new), runtime 974/68, plugin-approvals 330/13; all 8 spec `check:generated` gates plus `check:liveness` and `check:exported-any`; eslint clean; `tsc --noEmit` at the ledgered 2 pre-existing errors. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01SzJqHp1yXjri1tqKP8WoL4 --- .changeset/flow-run-summaries.md | 99 +++ content/docs/automation/flows.mdx | 55 ++ .../docs/references/automation/execution.mdx | 82 ++- content/docs/releases/v17.mdx | 11 + .../src/builtin/crud-nodes.ts | 56 +- .../src/builtin/map-node.ts | 26 +- .../src/builtin/notify-node.ts | 9 + .../src/builtin/subflow-node.ts | 15 +- .../service-automation/src/engine.test.ts | 16 +- .../services/service-automation/src/engine.ts | 282 +++++++- .../services/service-automation/src/index.ts | 7 + .../services/service-automation/src/plugin.ts | 14 + .../src/run-summary.test.ts | 661 ++++++++++++++++++ .../service-automation/src/run-summary.ts | Bin 0 -> 5451 bytes .../src/suspended-run-store.ts | 42 ++ .../src/sys-automation-run.object.ts | 39 +- packages/spec/api-surface.json | 11 + packages/spec/authorable-surface.json | 28 + packages/spec/json-schema.manifest.json | 5 + .../spec/src/automation/execution.test.ts | 64 ++ packages/spec/src/automation/execution.zod.ts | 111 +++ .../spec/src/contracts/automation-service.ts | 13 +- 22 files changed, 1607 insertions(+), 39 deletions(-) create mode 100644 .changeset/flow-run-summaries.md create mode 100644 packages/services/service-automation/src/run-summary.test.ts create mode 100644 packages/services/service-automation/src/run-summary.ts diff --git a/.changeset/flow-run-summaries.md b/.changeset/flow-run-summaries.md new file mode 100644 index 0000000000..10524795a4 --- /dev/null +++ b/.changeset/flow-run-summaries.md @@ -0,0 +1,99 @@ +--- +"@objectstack/spec": minor +"@objectstack/service-automation": minor +--- + +feat(spec,service-automation): every flow run reports what it actually did — selected / acted / skipped (#4354) + +`success: true` never meant "it did its job". A scheduled sweep that selects +thirty records and writes none is, from outside, **identical** to one with +nothing to do: same green status, same empty output, same silence, same schedule +tomorrow. There was no signal anywhere that separated "nothing to do" from +"broken". + +That is not theoretical. #4347 left three hotcrm production flows completely +inert — the stalled-deal sweep found every stalled deal and nudged nobody, the +renewal sweep booked nothing, the campaign action enrolled no leads. They ran +daily, on time, green, for as long as they had existed, and were caught only by +adding tests that assert on records written. Automation is exactly the category +where nobody is watching: a UI bug files a ticket within the hour, a dead sweep +files nothing, and the longer it runs the more normal the silence looks. + +**Every terminal run now carries a `FlowRunSummary`** — on the +`AutomationResult`, on the run in `listRuns` / `getRun`, in the log, and in the +database: + +``` +[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 +``` + +- `selected` — records read by the run's data nodes +- `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 +- `nodes[]` — per-node terminal status with `runs` / `failures` / `skipped` +- `gates[]` — which gates closed and how often, most-skipped first + +**The counts are declared, not sniffed.** Executors report +`NodeExecutionResult.metrics`, because only the node knows what its result +*means*: `update_record`'s is a row count on a bulk write and a record on a by-id +one, `delete_record`'s can be a boolean, `notify`'s is a delivery count. An +engine inferring from output shapes would be guessing, and a machine-readable +count that guesses is worse than none. A node that touches no records +(`decision`, `assignment`) reports nothing — absent is not `0`. + +**The gate is named.** A conditional out-edge that evaluates false now records a +`skipped` step tagged with the gate that closed. That event previously left no +trace at all, which is why #4347 was invisible: the flow selected every row and +the loop-body edge never opened. A skipped step is explicitly *not* a run — the +ADR-0044 re-entry guard, per-node `runs`, and node status all exclude it, so a +new observability signal cannot change execution semantics. + +**Queryable, so it can be alerted on rather than noticed.** +`sys_automation_run` gains `selected_count` / `acted_count` / `skipped_count` +columns plus a `summary_json` breakdown: + +```typescript +const suspect = await engine.find('sys_automation_run', { + where: { status: 'completed', selected_count: { $gt: 0 }, acted_count: 0 }, + orderBy: [{ field: 'started_at', order: 'desc' }], +}); +``` + +`selected > 0 && acted == 0` over consecutive runs is a near-perfect +broken-sweep detector. Columns, not JSON: an operator can only alert on what is +filterable. Rows written before this carry `null`, never `0` — "not measured" +must not read as "measured zero", or every legacy row is a false alarm the first +time someone writes that query. + +Two details that decide whether the numbers can be trusted. The summary is +folded from the **full** step log before history compaction, so a +5000-iteration sweep does not silently report the ~200 steps that fit in +`steps_json`; and rehydration reads the persisted `summary_json` rather than +re-folding those compacted steps. A `subflow` rolls its child's totals into its +parent, so a sweep that delegates its writes is not read as inert — the child +keeps its own run row, and the parent's summary answers "what did this run +cause". + +Additive throughout: `summary` is optional everywhere it appears, existing runs +and stores keep working, and no execution behaviour changes. The one-line log +defaults to `info` — a line nobody sees at their production level is the same +non-signal this closes — with `AutomationServicePlugin`'s +`runSummaryLog: 'debug' | 'off'` to turn the volume down on a very +high-frequency flow without turning the measurement off. + +New spec exports: `FlowRunSummarySchema`, `FlowRunNodeSummarySchema`, +`FlowRunGateSummarySchema`, `ExecutionStepMetricsSchema`, +`ExecutionStepSkipReasonSchema` (+ inferred types); `ExecutionLog.summary` and +`ExecutionStepLog.metrics` / `.skippedBy`. `service-automation` exports +`summarizeRun` / `formatRunSummaryLine` so a host building its own surface +reuses the platform's definition instead of re-deriving one. + +Does not fix #4347 itself — this is the instrument that would have caught it. + +Verified: `@objectstack/service-automation` **522 tests / 46 files** (23 new), +`@objectstack/spec` **7165 / 279** (5 new), `@objectstack/runtime` **974 / 68**, +`@objectstack/plugin-approvals` **330 / 13**; all eight `@objectstack/spec` +`check:generated` gates plus `check:liveness` and `check:exported-any`; and +`tsc --noEmit` on service-automation at its ledgered 2 pre-existing errors. diff --git a/content/docs/automation/flows.mdx b/content/docs/automation/flows.mdx index ce7c943be3..eb588da6b0 100644 --- a/content/docs/automation/flows.mdx +++ b/content/docs/automation/flows.mdx @@ -579,6 +579,61 @@ panel. Recent runs are held in an in-memory ring buffer; terminal runs history with a bounded step log, so `listRuns` / `getRun` still report a run's status, steps, and failure reason after a restart or ring-buffer eviction. +### Run summaries + +A run that reports `success: true` has not told you it did its job. A scheduled +sweep that selects thirty records and writes none looks *identical* to one with +nothing to do: same green status, same empty output, same silence. Every +terminal run therefore carries a **summary** — on the `AutomationResult`, on the +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 +``` + +| Field | Meaning | +| :--- | :--- | +| `selected` | Records **read** by the run's data nodes | +| `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 | +| `nodes[]` | Per-node terminal status with `runs` / `failures` / `skipped` and its own selected/acted | +| `gates[]` | Which gates closed and how often, most-skipped first | + +The counts come from the node executors themselves — `get_record` reports what +it matched, the write nodes report what the data engine actually changed — not +from the engine guessing at a node's output shape. A node that touches no +records (`decision`, `assignment`) reports nothing at all, which is different +from reporting zero. + +The same counts land on `sys_automation_run` as **queryable columns** +(`selected_count`, `acted_count`, `skipped_count`, plus a `summary_json` +breakdown), so a broken sweep is something you can alert on rather than notice: + +```typescript +// Runs that selected work and did none of it, newest first. +const suspect = await engine.find('sys_automation_run', { + where: { + status: 'completed', + selected_count: { $gt: 0 }, + acted_count: 0, + started_at: { $gte: since }, + }, + orderBy: [{ field: 'started_at', order: 'desc' }], +}); +``` + +`selected > 0 && acted == 0` over several consecutive runs is a near-perfect +broken-sweep detector — the case that is otherwise invisible, because nobody is +watching automation until it has already been dead for a month. A single such +run is not proof of anything: a sweep whose work is all already done reports the +same thing legitimately, which is why the signal is *consecutive* runs, and why +the platform reports the counts rather than raising the alarm itself. + +Rows written before summaries existed carry `null` counts, not `0` — "not +measured" must not read as "measured zero". The log line defaults to `info`; +`AutomationServicePlugin`'s `runSummaryLog: 'debug' | 'off'` turns the volume +down on a very high-frequency flow without turning the measurement off. + ## Edges Edges connect nodes and define the execution path: diff --git a/content/docs/references/automation/execution.mdx b/content/docs/references/automation/execution.mdx index 98cebfe215..3eb479551c 100644 --- a/content/docs/references/automation/execution.mdx +++ b/content/docs/references/automation/execution.mdx @@ -22,8 +22,8 @@ AWS Step Functions execution logs. ## TypeScript Usage ```typescript -import { Checkpoint, ConcurrencyPolicy, ExecutionError, ExecutionErrorSeverity, ExecutionLog, ExecutionStatus, ExecutionStepLog, ScheduleState } from '@objectstack/spec/automation'; -import type { Checkpoint, ConcurrencyPolicy, ExecutionError, ExecutionErrorSeverity, ExecutionLog, ExecutionStatus, ExecutionStepLog, ScheduleState } from '@objectstack/spec/automation'; +import { Checkpoint, ConcurrencyPolicy, ExecutionError, ExecutionErrorSeverity, ExecutionLog, ExecutionStatus, ExecutionStepLog, ExecutionStepMetrics, ExecutionStepSkipReason, FlowRunGateSummary, FlowRunNodeSummary, FlowRunSummary, ScheduleState } from '@objectstack/spec/automation'; +import type { Checkpoint, ConcurrencyPolicy, ExecutionError, ExecutionErrorSeverity, ExecutionLog, ExecutionStatus, ExecutionStepLog, ExecutionStepMetrics, ExecutionStepSkipReason, FlowRunGateSummary, FlowRunNodeSummary, FlowRunSummary, ScheduleState } from '@objectstack/spec/automation'; // Validate data const result = Checkpoint.parse(data); @@ -108,6 +108,7 @@ const result = Checkpoint.parse(data); | **status** | `Enum<'pending' \| 'running' \| 'paused' \| 'completed' \| 'failed' \| 'cancelled' \| 'timed_out' \| 'retrying'>` | ✅ | Current execution status | | **trigger** | `{ type: string; recordId?: string; object?: string; userId?: string; … }` | ✅ | What triggered this execution | | **steps** | `{ nodeId: string; nodeType: string; nodeLabel?: string; status: Enum<'success' \| 'failure' \| 'skipped'>; … }[]` | ✅ | Ordered list of executed steps | +| **summary** | `{ selected: integer; acted: integer; skipped: integer; nodes: { nodeId: string; nodeType: string; nodeLabel?: string; status: Enum<'success' \| 'failure' \| 'skipped'>; … }[]; … }` | optional | Per-run rollup: records selected / acted on, gate skips, per-node status | | **variables** | `Record` | optional | Final state of flow variables | | **startedAt** | `string` | ✅ | Execution start timestamp | | **completedAt** | `string` | optional | Execution completion timestamp | @@ -154,6 +155,83 @@ const result = Checkpoint.parse(data); | **parentNodeId** | `string` | optional | Enclosing structured-region container node ID (loop/parallel/try_catch) | | **iteration** | `integer` | optional | Zero-based loop iteration or parallel branch index of the enclosing region | | **regionKind** | `string` | optional | Region kind the step ran in: loop-body \| parallel-branch \| try \| catch | +| **metrics** | `{ selected?: integer; acted?: integer }` | optional | Records this step selected / acted on, as reported by the node executor | +| **skippedBy** | `{ nodeId: string; edgeId?: string; label?: string }` | optional | The gate that closed, when `status` is `skipped` | + + +--- + +## ExecutionStepMetrics + +### Properties + +| Property | Type | Required | Description | +| :--- | :--- | :--- | :--- | +| **selected** | `integer` | optional | Records this node READ or matched (a `get_record` query, a lookup) | +| **acted** | `integer` | optional | Records this node WROTE (created / updated / deleted) or effects it dispatched (notifications delivered) | + + +--- + +## ExecutionStepSkipReason + +### Properties + +| Property | Type | Required | Description | +| :--- | :--- | :--- | :--- | +| **nodeId** | `string` | ✅ | Node whose out-edge did not open (the gate) | +| **edgeId** | `string` | optional | Edge whose condition evaluated false | +| **label** | `string` | optional | Edge label, when the flow names its branches | + + +--- + +## FlowRunGateSummary + +### Properties + +| Property | Type | Required | Description | +| :--- | :--- | :--- | :--- | +| **nodeId** | `string` | ✅ | Node whose out-edge did not open (the gate) | +| **targetNodeId** | `string` | ✅ | Node the closed edge would have run | +| **edgeId** | `string` | optional | Edge whose condition evaluated false | +| **label** | `string` | optional | Edge label, when the flow names its branches | +| **skipped** | `integer` | ✅ | Times this gate evaluated false (once per loop iteration) | + + +--- + +## FlowRunNodeSummary + +### Properties + +| Property | Type | Required | Description | +| :--- | :--- | :--- | :--- | +| **nodeId** | `string` | ✅ | Node ID | +| **nodeType** | `string` | ✅ | Node action type (e.g., "get_record", "decision") | +| **nodeLabel** | `string` | optional | Human-readable node label | +| **status** | `Enum<'success' \| 'failure' \| 'skipped'>` | ✅ | Terminal status of the node across the run — `failure` if any execution failed, else `success` if any succeeded, else `skipped` | +| **runs** | `integer` | ✅ | Times the node executed (loop iterations and parallel branches each count) | +| **failures** | `integer` | ✅ | Executions that failed | +| **skipped** | `integer` | ✅ | Times a closed gate kept this node from running at all | +| **selected** | `integer` | optional | Records read across every execution — omitted for a node that reads none | +| **acted** | `integer` | optional | Records written / effects dispatched across every execution — omitted for a node that writes none | + + +--- + +## FlowRunSummary + +### Properties + +| Property | Type | Required | Description | +| :--- | :--- | :--- | :--- | +| **selected** | `integer` | ✅ | Total records read by the run | +| **acted** | `integer` | ✅ | Total records written / effects dispatched by the run | +| **skipped** | `integer` | ✅ | Total node executions a closed gate prevented | +| **nodes** | `{ nodeId: string; nodeType: string; nodeLabel?: string; status: Enum<'success' \| 'failure' \| 'skipped'>; … }[]` | ✅ | Per-node breakdown, in first-execution order | +| **gates** | `{ nodeId: string; targetNodeId: string; edgeId?: string; label?: string; … }[]` | ✅ | Gates that closed during the run, most-skipped first | +| **detailOmitted** | `boolean` | optional | Set when persistence dropped `nodes`/`gates` to keep the stored row bounded — the totals are still exact. Declared so empty arrays are never mistaken for "nothing ran". | --- diff --git a/content/docs/releases/v17.mdx b/content/docs/releases/v17.mdx index 99bc1c7d53..995cd7fdf3 100644 --- a/content/docs/releases/v17.mdx +++ b/content/docs/releases/v17.mdx @@ -1152,6 +1152,17 @@ is gone. ### Automation & flows +- **Every terminal run reports what it did** (#4354): `selected` / `acted` / + `skipped` totals, a per-node breakdown, and *which gate closed* — on the run + result, in `listRuns` / `getRun`, as one greppable log line + (`selected=30 acted=0 skipped=30 gate=check_stalled->send_nudge:30`), and as + queryable `selected_count` / `acted_count` / `skipped_count` columns on + `sys_automation_run`. `success: true` never meant "it did its job": a sweep + that selected thirty records and wrote none was indistinguishable from one + with nothing to do, which is how three inert production flows ran green for + as long as they had existed. `selected > 0 && acted == 0` over consecutive + runs is now the detector. See + [Run summaries](/docs/automation/flows#run-summaries). - **`record-after-write`** fires one flow on create **or** update (#3427), with `previous` bound as `null` on the create leg so start conditions can discriminate. diff --git a/packages/services/service-automation/src/builtin/crud-nodes.ts b/packages/services/service-automation/src/builtin/crud-nodes.ts index b3cd596102..3083eab935 100644 --- a/packages/services/service-automation/src/builtin/crud-nodes.ts +++ b/packages/services/service-automation/src/builtin/crud-nodes.ts @@ -121,6 +121,28 @@ function droppedFieldsWarning(nodeType: string, e: DroppedFieldsEvent): string { return `${nodeType}(${e.object}): requested field(s) [${e.fields.join(', ')}] were NOT written — ${DROPPED_REASON_LABEL[e.reason] ?? e.reason}. The write succeeded without them.`; } +/** + * How many rows an `update` / `delete` actually touched (#4354). + * + * The data engine returns a DIFFERENT shape per route, by design: a bulk write + * lands on `driver.updateMany` / `deleteMany`, whose contract is + * `Promise` (the row count), while a by-id write returns the updated + * record — or, for delete, whatever the driver reports (typically a boolean). + * The executor is the only place that knows which route it asked for, which is + * exactly why {@link NodeExecutionResult.metrics} is declared by the node and + * not sniffed by the engine. + * + * `null` / `false` / `undefined` ⇒ nothing was written. Anything else the driver + * hands back that is neither a count nor a list is one row — the write returned, + * so a row was touched. + */ +function writtenRowCount(result: unknown): number { + if (typeof result === 'number') return Number.isFinite(result) && result > 0 ? Math.trunc(result) : 0; + if (Array.isArray(result)) return result.length; + if (result === null || result === undefined || result === false) return 0; + return 1; +} + /** * CRUD built-in nodes — `get_record` / `create_record` / `update_record` / * `delete_record`, wired to the runtime data layer (ObjectQL / IDataEngine). @@ -205,7 +227,7 @@ export function registerCrudNodes(engine: AutomationEngine, ctx: PluginContext): const data = getData(); if (!data) { ctx.logger.warn(`[get_record] no data engine; skipping ${objectName}`); - return { success: true, output: { records: [], object: objectName } }; + return { success: true, output: { records: [], object: objectName }, metrics: { selected: 0 } }; } // #1888 — honor flow.runAs: read under the run's effective identity @@ -215,11 +237,22 @@ export function registerCrudNodes(engine: AutomationEngine, ctx: PluginContext): if (limit && limit > 1) { const records = await data.find(objectName, { where: filter, fields, limit, context: dataCtx }); if (outputVariable) variables.set(outputVariable, records); - return { success: true, output: { records, object: objectName } }; + // #4354 — the `selected` half of the broken-sweep signal: + // this is the count that made #4347 diagnosable at all + // ("found every stalled deal and nudged nobody"). + return { + success: true, + output: { records, object: objectName }, + metrics: { selected: Array.isArray(records) ? records.length : 0 }, + }; } const record = await data.findOne(objectName, { where: filter, fields, context: dataCtx }); if (outputVariable) variables.set(outputVariable, record); - return { success: true, output: { record, id: record?.id, object: objectName } }; + return { + success: true, + output: { record, id: record?.id, object: objectName }, + metrics: { selected: record ? 1 : 0 }, + }; } catch (err) { return { success: false, error: `get_record(${objectName}) failed: ${(err as Error).message}` }; } @@ -260,7 +293,10 @@ export function registerCrudNodes(engine: AutomationEngine, ctx: PluginContext): ctx.logger.warn(`[create_record] no data engine; skipping ${objectName}`); const mockId = `mock-${objectName}-${Date.now()}`; if (outputVariable) variables.set(outputVariable, { id: mockId }); - return { success: true, output: { id: mockId, object: objectName } }; + // `acted: 0` — the mock id is a placeholder for downstream + // templates, not a row. A summary that counted it would + // report writes that never happened (#4354). + return { success: true, output: { id: mockId, object: objectName }, metrics: { acted: 0 } }; } // #1888 — honor flow.runAs (system → RLS-bypassing; user → trigger user). @@ -302,6 +338,7 @@ export function registerCrudNodes(engine: AutomationEngine, ctx: PluginContext): ...(dropped.length > 0 ? { warnings: dropped.map((e) => droppedFieldsWarning('create_record', e)) } : {}), + metrics: { acted: 1 }, }; } catch (err) { return { success: false, error: `create_record(${objectName}) failed: ${(err as Error).message}` }; @@ -349,7 +386,7 @@ export function registerCrudNodes(engine: AutomationEngine, ctx: PluginContext): const data = getData(); if (!data) { ctx.logger.warn(`[update_record] no data engine; skipping ${objectName}`); - return { success: true }; + return { success: true, metrics: { acted: 0 } }; } // #1888 — honor flow.runAs (system → RLS-bypassing; user → trigger user). @@ -378,6 +415,7 @@ export function registerCrudNodes(engine: AutomationEngine, ctx: PluginContext): ...(dropped.length > 0 ? { warnings: dropped.map((e) => droppedFieldsWarning('update_record', e)) } : {}), + metrics: { acted: writtenRowCount(result) }, }; } catch (err) { return { success: false, error: `update_record(${objectName}) failed: ${(err as Error).message}` }; @@ -421,13 +459,17 @@ export function registerCrudNodes(engine: AutomationEngine, ctx: PluginContext): const filter = filterResult.filter; const data = getData(); - if (!data) return { success: true }; + if (!data) return { success: true, metrics: { acted: 0 } }; // #1888 — honor flow.runAs (system → RLS-bypassing; user → trigger user). const dataCtx = resolveRunDataContext(context); try { const result = await data.delete(objectName, { where: filter, context: dataCtx }); - return { success: true, output: { result, object: objectName } }; + return { + success: true, + output: { result, object: objectName }, + metrics: { acted: writtenRowCount(result) }, + }; } catch (err) { return { success: false, error: `delete_record(${objectName}) failed: ${(err as Error).message}` }; } diff --git a/packages/services/service-automation/src/builtin/map-node.ts b/packages/services/service-automation/src/builtin/map-node.ts index 7c02bae101..62888eb7d1 100644 --- a/packages/services/service-automation/src/builtin/map-node.ts +++ b/packages/services/service-automation/src/builtin/map-node.ts @@ -131,6 +131,14 @@ export function registerMapNode(engine: AutomationEngine, ctx: PluginContext): v const parentRunId = variables.get('$runId'); + // #4354 — totals of the items completed SYNCHRONOUSLY during THIS entry. + // Per-entry, not cumulative: the node re-runs (and logs a fresh step) on + // every re-entry, so a running total would be counted once per item. An + // item that PAUSED is credited to this entry's step by the engine when its + // child run bubbles back (AutomationEngine.creditChildRun). + let selected = 0; + let acted = 0; + // Drive items in order. Synchronous items advance inline; a pausing item // suspends the run and is resumed via re-entry. while (state.started < collection.length) { @@ -167,20 +175,32 @@ export function registerMapNode(engine: AutomationEngine, ctx: PluginContext): v // Mark this item started and suspend; the engine re-enters on bubble. state.started = idx + 1; variables.set(stateKey, state); - return { success: true, suspend: true, correlation: `map:${child.runId}` }; + return { success: true, suspend: true, correlation: `map:${child.runId}`, metrics: { selected, acted } }; } if (!child.success) { - return { success: false, error: `map '${node.id}': item ${idx} (subflow '${flowName}') failed: ${child.error ?? 'unknown error'}` }; + return { + success: false, + error: `map '${node.id}': item ${idx} (subflow '${flowName}') failed: ${child.error ?? 'unknown error'}`, + // Items that already succeeded wrote real rows; a later item's + // failure must not erase them from the run's totals. + metrics: { selected: selected + (child.summary?.selected ?? 0), acted: acted + (child.summary?.acted ?? 0) }, + }; } // Synchronous completion — record and advance. state.started = idx + 1; state.results.push(child.output ?? null); + selected += child.summary?.selected ?? 0; + acted += child.summary?.acted ?? 0; } // All items done. variables.set(stateKey, state); if (outVar) variables.set(outVar, state.results); - return { success: true, output: { results: state.results, count: state.results.length } }; + return { + success: true, + output: { results: state.results, count: state.results.length }, + metrics: { selected, acted }, + }; }, }); diff --git a/packages/services/service-automation/src/builtin/notify-node.ts b/packages/services/service-automation/src/builtin/notify-node.ts index 27803912dc..8987fcc3c2 100644 --- a/packages/services/service-automation/src/builtin/notify-node.ts +++ b/packages/services/service-automation/src/builtin/notify-node.ts @@ -250,6 +250,10 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) return { success: true, output: { delivered: 0, failed: 0, skipped: true }, + // #4354 — nothing was delivered, and the run summary must say + // so: a nudge sweep whose messaging service is absent is + // precisely the "green but inert" case this counter exists for. + metrics: { acted: 0 }, }; } @@ -274,6 +278,11 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) delivered: result.delivered, failed: result.failed, }, + // A notification IS the action for a nudge/alert sweep, so it + // counts toward `acted` (#4354) — otherwise the flow whose + // whole job is to notify would report acting on nothing, and + // the broken-sweep detector would fire on every healthy run. + metrics: { acted: Number(result.delivered) || 0 }, }; } catch (err) { return { success: false, error: `notify failed: ${(err as Error).message}` }; diff --git a/packages/services/service-automation/src/builtin/subflow-node.ts b/packages/services/service-automation/src/builtin/subflow-node.ts index 4997ef709b..d18c5169ad 100644 --- a/packages/services/service-automation/src/builtin/subflow-node.ts +++ b/packages/services/service-automation/src/builtin/subflow-node.ts @@ -113,15 +113,26 @@ export function registerSubflowNode(engine: AutomationEngine, ctx: PluginContext screen: child.screen, }; } + // #4354 — roll the child's totals up into this node's step. The child's + // steps live in the CHILD's run log, so without this a parent that + // delegates its writes to a subflow reports `acted: 0` and reads as a + // broken sweep. The child keeps its own run row with its own summary; + // the parent's answers "what did this run cause", subflows included. + const rolled = child.summary + ? { metrics: { selected: child.summary.selected, acted: child.summary.acted } } + : {}; + if (!child.success) { - return { success: false, error: `subflow '${flowName}' failed: ${child.error ?? 'unknown error'}` }; + // A failed child may still have written rows before it died — carry its + // counts so the parent's summary does not understate what happened. + return { success: false, error: `subflow '${flowName}' failed: ${child.error ?? 'unknown error'}`, ...rolled }; } // Bare output variable (like the assignment node, the executor may write // directly to the parent variable map). if (outVar) variables.set(outVar, child.output ?? null); - return { success: true, output: { output: child.output ?? null } }; + return { success: true, output: { output: child.output ?? null }, ...rolled }; }, }); diff --git a/packages/services/service-automation/src/engine.test.ts b/packages/services/service-automation/src/engine.test.ts index dc7b92fe8d..f3c5cf9615 100644 --- a/packages/services/service-automation/src/engine.test.ts +++ b/packages/services/service-automation/src/engine.test.ts @@ -44,9 +44,12 @@ describe('AutomationEngine', () => { expect((e as unknown as { maxLogSize: number }).maxLogSize).toBe(3); // Drive the private ring buffer directly: push 5, keep newest 3. + // `steps` is required on the entry (recordLog folds it into the run + // summary, #4354) — an empty log is the honest value for a synthetic + // entry that never ran a node. const rec = (e as any).recordLog.bind(e); for (let i = 0; i < 5; i++) { - rec({ id: `run_${i}`, flowName: 'f', status: 'success' }); + rec({ id: `run_${i}`, flowName: 'f', status: 'success', steps: [] }); } const buf = (e as unknown as { executionLogs: Array<{ id: string }> }).executionLogs; expect(buf).toHaveLength(3); @@ -1727,7 +1730,16 @@ describe('AutomationEngine - Back-edge re-entry (ADR-0044)', () => { expect(getCount()).toBe(3); // Every visit is its own step entry — observability shows each round. const run = (await engine.listRuns('counter_flow'))[0]; - expect(run.steps.filter(s => s.nodeId === 'inc')).toHaveLength(3); + const visits = run.steps.filter(s => s.nodeId === 'inc' && s.status !== 'skipped'); + expect(visits).toHaveLength(3); + // #4354 — the round that CLOSED the back-edge is recorded too, as a + // `skipped` step naming the gate. It is not a visit: the re-entry guard + // and the summary's per-node `runs` both exclude it, so recording the + // non-event must not inflate either. + const closed = run.steps.filter(s => s.nodeId === 'inc' && s.status === 'skipped'); + expect(closed).toHaveLength(1); + expect(closed[0].skippedBy).toMatchObject({ nodeId: 'inc', edgeId: 'e_back' }); + expect(run.summary?.nodes.find(n => n.nodeId === 'inc')).toMatchObject({ runs: 3, skipped: 1 }); }); it('aborts a runaway back-edge loop at the re-entry cap', async () => { diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 9ab2adc742..02f32f25e7 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1,7 +1,13 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import type { FlowParsed, FlowNodeParsed, FlowEdgeParsed } from '@objectstack/spec/automation'; -import type { ExecutionLog, ActionDescriptor } from '@objectstack/spec/automation'; +import type { + ExecutionLog, + ActionDescriptor, + ExecutionStepMetrics, + ExecutionStepSkipReason, + FlowRunSummary, +} from '@objectstack/spec/automation'; import type { AutomationContext, AutomationResult, ResumeSignal, IAutomationService, ScreenSpec } from '@objectstack/spec/contracts'; import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; import type { Logger } from '@objectstack/spec/contracts'; @@ -67,6 +73,7 @@ const FLOW_NODE_UNKNOWN_KEY_GUIDANCE: Record> = { }; import { runIsUnscopedUserMode, flowTouchesData } from './runtime-identity.js'; import { isGuardRefusal } from './guard-refusal.js'; +import { summarizeRun, formatRunSummaryLine } from './run-summary.js'; // ─── Node Executor Interface (Plugin Extension Point) ─────────────── @@ -172,6 +179,23 @@ export interface NodeExecutionResult { * so per-iteration / per-branch body steps surface in run observability. */ childSteps?: StepLogEntry[]; + /** + * #4354: how many records this execution **read** and **wrote**, reported by + * the executor and folded into the run summary by {@link summarizeRun}. + * + * Declared by the node rather than inferred by the engine from `output`, + * because only the node knows what its result *means*: `update_record`'s + * `result` is a row count on a bulk write and the updated record on a + * by-id one, `delete_record`'s can be a boolean, `notify`'s is a delivery + * count. An engine that sniffed those shapes would be guessing, and a + * machine-readable count that guesses is worse than none (ADR-0076 D12 — + * "machine-readable surfaces must not lie"). + * + * Omit it entirely for a node that touches no records (`decision`, + * `assignment`): absent means "reads/writes nothing", which is a different + * fact from `0`. + */ + metrics?: ExecutionStepMetrics; } // ─── Trigger Interface (Plugin Extension Point) ───────────────────── @@ -395,6 +419,18 @@ export const DEFAULT_MAX_EXECUTION_LOG_SIZE = 1000; */ export const MAX_PERSISTED_HISTORY_STEPS = 200; +/** + * Level the one-line-per-terminal-run summary is logged at (#4354), or `'off'`. + * + * Defaults to `'info'` — deliberately. The whole premise of #4354 is that a + * scheduled flow which silently stopped working emits **no signal at all**, and + * a line nobody sees at their production log level is the same non-signal. A + * host running very high-frequency record-change flows can turn the volume down + * to `'debug'` (or off) via {@link AutomationServicePluginOptions.runSummaryLog}, + * which is a decision about noise, not about whether the platform measures. + */ +export type RunSummaryLogLevel = 'info' | 'debug' | 'off'; + /** Construction options for {@link AutomationEngine}. */ export interface AutomationEngineOptions { /** @@ -402,6 +438,11 @@ export interface AutomationEngineOptions { * Defaults to {@link DEFAULT_MAX_EXECUTION_LOG_SIZE}. Must be > 0. */ maxLogSize?: number; + /** + * Level for the per-terminal-run summary line (#4354). Defaults to `'info'`. + * See {@link RunSummaryLogLevel}. + */ + runSummaryLog?: RunSummaryLogLevel; } /** @@ -437,6 +478,19 @@ export interface StepLogEntry { iteration?: number; /** Which region kind the step ran in: `loop-body` | `parallel-branch` | `try` | `catch`. */ regionKind?: string; + /** + * #4354: records this step read / wrote, copied from + * {@link NodeExecutionResult.metrics}. Folded into the run summary. + */ + metrics?: ExecutionStepMetrics; + /** + * #4354: the gate that closed in front of this node, on a `skipped` step. + * Written by {@link AutomationEngine.traverseNext} when a conditional + * out-edge evaluates false — the shape #4347 could not surface: a loop-body + * edge that never opened left no trace at all, so a sweep that nudged + * nobody looked exactly like a sweep with nobody to nudge. + */ + skippedBy?: ExecutionStepSkipReason; } /** @@ -537,6 +591,13 @@ interface ExecutionLogEntry { variables?: Record; output?: unknown; error?: string; + /** + * #4354: what the run did, folded out of the FULL step log by + * {@link AutomationEngine.recordLog} — before history compaction, so a + * 5000-iteration loop's counts are exact even though only 200 of its steps + * are persisted. + */ + summary?: FlowRunSummary; } /** @@ -717,6 +778,15 @@ export interface RunRecord { * written before this field existed have none. */ steps?: StepLogEntry[]; + /** + * #4354: the run's selected / acted / skipped rollup, computed from the + * un-compacted step log. Persisted alongside `steps` so the Runs surface + * and an operator's `selected > 0 AND acted = 0` alert both read exact + * counts, not counts inferred from the 200 steps that survived compaction. + * Optional — rows written before this field existed have none, and an + * absent summary must never be read as "this run did nothing". + */ + summary?: FlowRunSummary; } export interface SuspendedRunStore { @@ -808,6 +878,8 @@ export class AutomationEngine implements IAutomationService { private recordExpander: FlowRecordExpander | null = null; private executionLogs: ExecutionLogEntry[] = []; private readonly maxLogSize: number; + /** Level for the per-run summary line (#4354). See {@link RunSummaryLogLevel}. */ + private readonly runSummaryLog: RunSummaryLogLevel; private logger: Logger; /** * Runs paused at a node, keyed by runId (ADR-0019). In-memory hot cache — @@ -831,6 +903,7 @@ export class AutomationEngine implements IAutomationService { this.logger = logger; this.store = store; this.maxLogSize = options?.maxLogSize ?? DEFAULT_MAX_EXECUTION_LOG_SIZE; + this.runSummaryLog = options?.runSummaryLog ?? 'info'; } /** @@ -1676,6 +1749,10 @@ export class AutomationEngine implements IAutomationService { trigger: { type: '', userId: r.userId ?? undefined }, steps: r.steps ?? [], error: r.error, + // #4354 — the PERSISTED summary, never re-folded from `r.steps`: + // those are compacted (200 max), so recomputing here would report a + // 5000-row sweep as having acted on a couple of hundred. + summary: r.summary, }; } @@ -1974,7 +2051,7 @@ export class AutomationEngine implements IAutomationService { const durationMs = Date.now() - startTime; // Record execution log - this.recordLog({ + const logged = this.recordLog({ id: runId, flowName, flowVersion: flow.version, @@ -1995,6 +2072,10 @@ export class AutomationEngine implements IAutomationService { success: true, output, durationMs, + // #4354 — hand the counts back synchronously so a caller + // (a `subflow` roll-up, a runtime test asserting the sweep wrote + // something) never has to re-read the run to learn what it did. + summary: logged.summary, }; } catch (err: unknown) { // A node asked to suspend the run (ADR-0019 durable pause). Snapshot @@ -2043,7 +2124,7 @@ export class AutomationEngine implements IAutomationService { // Record failed execution log const durationMs = Date.now() - startTime; - this.recordLog({ + const logged = this.recordLog({ id: runId, flowName, flowVersion: flow.version, @@ -2068,6 +2149,9 @@ export class AutomationEngine implements IAutomationService { success: false, error: errorMessage, durationMs, + // A failed run's counts matter MORE, not less: they say how far + // it got before dying — how many rows it had already written. + summary: logged.summary, }; } finally { // Release the re-entrancy guard for this (flow, record). Runs before @@ -2244,12 +2328,50 @@ export class AutomationEngine implements IAutomationService { } } + /** + * Credit a completed child run's totals to the parent step waiting on it + * (#4354). + * + * A `subflow` / `map` child that PAUSED cannot report through + * `NodeExecutionResult.metrics` the way a synchronous one does: the parent's + * step for that node was written at suspend time, before the child had done + * anything. Without this, a sweep whose writes all happen inside a paused + * child would report `acted: 0` — a healthy run indistinguishable from a + * dead one, which is the exact confusion this feature exists to remove. + * + * The credit lands on the LAST step for the node, which is the entry that + * suspended awaiting this child — so a `map` re-entering once per item + * credits each item to its own step and nothing is counted twice. + */ + private creditChildRun(steps: StepLogEntry[], nodeId: string, child: FlowRunSummary | undefined): void { + if (!child) return; + for (let i = steps.length - 1; i >= 0; i--) { + if (steps[i].nodeId !== nodeId) continue; + const prior = steps[i].metrics ?? {}; + steps[i] = { + ...steps[i], + metrics: { + selected: (prior.selected ?? 0) + child.selected, + acted: (prior.acted ?? 0) + child.acted, + }, + }; + return; + } + } + /** * @param skipBubble - Set when the caller is the subflow DELEGATION path, * which continues the parent itself after the child completes — the * child's own up-bubble must stay off so the parent isn't resumed twice. - */ - private async resumeInternal(runId: string, signal: ResumeSignal | undefined, skipBubble: boolean): Promise { + * @param childSummary - #4354: totals of the child run whose completion + * triggered this resume (the up-bubble path), credited to the awaiting step. + */ + private async resumeInternal( + runId: string, + signal: ResumeSignal | undefined, + skipBubble: boolean, + childSummary?: FlowRunSummary, + ): Promise { // Idempotency guard (set synchronously, before any await): reject a // concurrent duplicate resume of the same run so side effects can't run // twice. A duplicate that arrives *after* this one finishes finds no @@ -2274,6 +2396,11 @@ export class AutomationEngine implements IAutomationService { return { success: false, error: `Suspended node '${run.nodeId}' no longer exists in flow '${run.flowName}'` }; } + // #4354 — up-bubble: the child that just finished did work this run + // is accountable for. Credit it to the step that suspended awaiting + // it, before traversal appends anything further. + this.creditChildRun(run.steps, run.nodeId, childSummary); + // ── Subflow delegation (nested pause): this run is paused at a // `subflow` node whose child run itself suspended. The caller's // signal is meant for the node the CHILD paused on (its screen / @@ -2312,6 +2439,9 @@ export class AutomationEngine implements IAutomationService { // resume signal (replaces the caller's signal, which the // child already consumed). signal = this.buildSubflowResumeSignal(childRun.context, childRes.output); + // #4354 — down-delegation is the other way a child's work + // lands under a parent step written at suspend time. + this.creditChildRun(run.steps, run.nodeId, childRes.summary); } else { this.logger.warn( `[automation] run '${runId}' is paused at subflow node '${run.nodeId}' but child run '${childRunId}' ` + @@ -2369,7 +2499,7 @@ export class AutomationEngine implements IAutomationService { } } const durationMs = Date.now() - run.startTime; - this.recordLog({ + const logged = this.recordLog({ id: runId, flowName: run.flowName, flowVersion: run.flowVersion, @@ -2393,12 +2523,20 @@ export class AutomationEngine implements IAutomationService { // continues the parent itself). Best-effort: the child's own // completion stands even if the parent continuation fails. if (!skipBubble) { - await this.bubbleToParent(run, output); + await this.bubbleToParent(run, output, logged.summary); } // Surface the flow's friendly completion message so a screen-flow - // runner shows it instead of a generic "Done". - return { success: true, output, durationMs, successMessage: flow.successMessage }; + // runner shows it instead of a generic "Done". `summary` (#4354) + // covers the WHOLE run — the steps before the pause and after it + // are one log, so a resumed approval reports what it did in total. + return { + success: true, + output, + durationMs, + successMessage: flow.successMessage, + summary: logged.summary, + }; } catch (err: unknown) { // Re-suspended at a downstream node: persist a fresh continuation. if (isSuspendSignal(err)) { @@ -2431,7 +2569,7 @@ export class AutomationEngine implements IAutomationService { const errorMessage = err instanceof Error ? err.message : String(err); const durationMs = Date.now() - run.startTime; - this.recordLog({ + const logged = this.recordLog({ id: runId, flowName: run.flowName, flowVersion: run.flowVersion, @@ -2455,7 +2593,13 @@ export class AutomationEngine implements IAutomationService { } // Surface the flow's friendly error message (the raw error stays // in `error` for logs/diagnostics). - return { success: false, error: errorMessage, durationMs, errorMessage: flow.errorMessage }; + return { + success: false, + error: errorMessage, + durationMs, + errorMessage: flow.errorMessage, + summary: logged.summary, + }; } } finally { this.resuming.delete(runId); @@ -2488,7 +2632,12 @@ export class AutomationEngine implements IAutomationService { * a failed parent continuation is logged, never thrown back at the * caller who resumed the child. */ - private async bubbleToParent(run: SuspendedRun, output: Record): Promise { + private async bubbleToParent( + run: SuspendedRun, + output: Record, + /** #4354 — this child's totals, credited to the parent's awaiting step. */ + summary?: FlowRunSummary, + ): Promise { const ctx = run.context as Record | undefined; const parentRunId = ctx?.$parentRunId; if (typeof parentRunId !== 'string' || !parentRunId) return; @@ -2502,7 +2651,7 @@ export class AutomationEngine implements IAutomationService { // the one writer allowed to set them (#3853 follow-up). ? engineBuilt({ variables: { [`${mapNode}.$mapItemOutput`]: output ?? null, [`${mapNode}.$mapItemDone`]: true } }) : this.buildSubflowResumeSignal(run.context, output); - const parentRes = await this.resumeInternal(parentRunId, sig, false); + const parentRes = await this.resumeInternal(parentRunId, sig, false, summary); if (!parentRes.success) { this.logger.warn( `[automation] subflow run '${run.runId}' completed but resuming parent '${parentRunId}' failed: ${parentRes.error}`, @@ -2652,7 +2801,22 @@ export class AutomationEngine implements IAutomationService { // ── DAG Traversal Core ────────────────────────────────── - private recordLog(entry: ExecutionLogEntry): void { + /** + * Append a run to the in-memory ring buffer, fold its {@link FlowRunSummary}, + * log the one-line-per-run summary (#4354) and mirror a terminal run to + * durable history. + * + * @returns the same entry, now carrying `summary` — so a caller returning an + * {@link AutomationResult} hands the counts straight back without a second + * fold or a `getRun` round-trip. + */ + private recordLog(entry: ExecutionLogEntry): ExecutionLogEntry { + // #4354 — fold the run's outcome BEFORE anything downstream trims the + // step log. History compaction keeps 200 steps; the summary must count + // all 5000, or a long sweep's `acted` would shrink with its step log and + // the broken-sweep detector would read a bounded artefact as truth. + entry.summary = summarizeRun(entry.steps); + this.executionLogs.push(entry); // Evict oldest logs when exceeding max size if (this.executionLogs.length > this.maxLogSize) { @@ -2667,6 +2831,35 @@ export class AutomationEngine implements IAutomationService { entry.status === 'failed' || entry.status === 'cancelled' || entry.status === 'timed_out'; + + // The MVP of #4354, and the half that needs no console: one structured + // line per terminal run. `selected=30 acted=0` in a log file is the + // difference between an invisible failure and a greppable one. Only + // terminal runs — a `paused` run has not finished doing its work yet. + if (terminal && this.runSummaryLog !== 'off') { + const line = formatRunSummaryLine( + { + flowName: entry.flowName, + runId: entry.id, + status: entry.status, + durationMs: entry.durationMs, + }, + entry.summary, + ); + const meta = { + flow: entry.flowName, + runId: entry.id, + status: entry.status, + durationMs: entry.durationMs, + selected: entry.summary.selected, + acted: entry.summary.acted, + skipped: entry.summary.skipped, + gates: entry.summary.gates, + }; + if (this.runSummaryLog === 'debug') this.logger.debug(line, meta); + else this.logger.info(line, meta); + } + if (terminal && this.store?.recordTerminal) { const lastStep = entry.steps[entry.steps.length - 1]; const record: RunRecord = { @@ -2681,6 +2874,7 @@ export class AutomationEngine implements IAutomationService { userId: entry.trigger?.userId, nodeId: lastStep?.nodeId, steps: this.compactStepsForHistory(entry.steps), + summary: entry.summary, }; void this.store.recordTerminal(record).catch((err) => { this.logger.warn( @@ -2688,6 +2882,7 @@ export class AutomationEngine implements IAutomationService { ); }); } + return entry; } /** @@ -3120,8 +3315,17 @@ export class AutomationEngine implements IAutomationService { // 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 ? n + 1 : n), 0, + (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( @@ -3213,6 +3417,9 @@ export class AutomationEngine implements IAutomationService { 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 } : {}), }); // Write error output to variable context for downstream nodes @@ -3238,7 +3445,10 @@ export class AutomationEngine implements IAutomationService { } // Log successful step (#3407: advisory executor warnings ride along - // so a legal-but-partial outcome never reads as a clean success). + // 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, @@ -3247,6 +3457,7 @@ export class AutomationEngine implements IAutomationService { 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 @@ -3329,11 +3540,38 @@ export class AutomationEngine implements IAutomationService { // Conditional edges: evaluate sequentially (mutually exclusive) for (const edge of conditionalEdges) { + const nextNode = flow.nodes.find(n => n.id === edge.target); if (this.evaluateCondition(edge.condition!, variables)) { - const nextNode = flow.nodes.find(n => n.id === edge.target); if (nextNode) { await this.executeNode(nextNode, flow, variables, context, steps); } + } else if (nextNode) { + // #4354 — the gate closed. 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 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 } : {}), + }, + }); } } @@ -3666,7 +3904,7 @@ export class AutomationEngine implements IAutomationService { } const durationMs = Date.now() - startTime; - this.recordLog({ + const logged = this.recordLog({ id: runId, flowName, flowVersion: flow.version, @@ -3683,11 +3921,13 @@ export class AutomationEngine implements IAutomationService { output, }); - return { success: true, output, durationMs }; + // #4354 — a retried run reports its own attempt's counts, not the + // failed one's: `retryExecution` returns THIS result on success. + return { success: true, output, durationMs, summary: logged.summary }; } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); const durationMs = Date.now() - startTime; - this.recordLog({ + const logged = this.recordLog({ id: runId, flowName, flowVersion: flow.version, @@ -3703,7 +3943,7 @@ export class AutomationEngine implements IAutomationService { steps, error: errorMessage, }); - return { success: false, error: errorMessage, durationMs }; + return { success: false, error: errorMessage, durationMs, summary: logged.summary }; } } } diff --git a/packages/services/service-automation/src/index.ts b/packages/services/service-automation/src/index.ts index 0c29588970..a181a8cc7a 100644 --- a/packages/services/service-automation/src/index.ts +++ b/packages/services/service-automation/src/index.ts @@ -4,6 +4,7 @@ export { AutomationEngine, DEFAULT_MAX_EXECUTION_LOG_SIZE, MAX_PERSISTED_HISTORY_STEPS } from './engine.js'; export type { AutomationEngineOptions, + RunSummaryLogLevel, NodeExecutor, NodeExecutionResult, FlowTrigger, @@ -17,6 +18,12 @@ export type { StepLogEntry, } from './engine.js'; +// Per-run summary (#4354): the fold that turns a run's step log into +// "selected N, acted M, skipped K by ". Exported so a host building its +// own observability surface (or a test asserting a sweep actually wrote +// something) reuses the platform's definition instead of re-deriving one. +export { summarizeRun, formatRunSummaryLine } from './run-summary.js'; + // Connector provider contract (ADR-0097) and the registry vocabulary that goes // with it — re-exported from @objectstack/spec so hosts/tests can reach them via // this package too. Connector plugins should import them directly from diff --git a/packages/services/service-automation/src/plugin.ts b/packages/services/service-automation/src/plugin.ts index b3679a0414..ecf987ae56 100644 --- a/packages/services/service-automation/src/plugin.ts +++ b/packages/services/service-automation/src/plugin.ts @@ -11,6 +11,7 @@ import type { } from '@objectstack/spec/integration'; import { isConnectorUpstreamUnavailable } from '@objectstack/spec/integration'; import { AutomationEngine } from './engine.js'; +import type { RunSummaryLogLevel } from './engine.js'; import { installBuiltinNodes, rearmSuspendedWaitTimers } from './builtin/index.js'; import { resolveRunDataContext } from './runtime-identity.js'; import { SysAutomationRun } from './sys-automation-run.object.js'; @@ -75,6 +76,18 @@ export interface AutomationServicePluginOptions { * to {@link DEFAULT_MAX_EXECUTION_LOG_SIZE} (1000). */ maxLogSize?: number; + /** + * Level for the one-line-per-terminal-run summary (#4354) — the line that + * makes `selected=30 acted=0` greppable in a plain log file. Defaults to + * `'info'`; a host running very high-frequency record-change flows can drop + * it to `'debug'` or `'off'`. + * + * Turning it down is a decision about log volume only: the summary is still + * computed, still returned on {@link AutomationResult}, still persisted to + * `sys_automation_run`, and still queryable. The measurement is not optional + * — only its default-level narration is. + */ + runSummaryLog?: RunSummaryLogLevel; /** * Per-flow cap on terminal run-history rows, enforced at write time (the * "or 100 runs/flow, whichever first" half of the #2585 retention @@ -368,6 +381,7 @@ export class AutomationServicePlugin implements Plugin { async init(ctx: PluginContext): Promise { this.engine = new AutomationEngine(ctx.logger, undefined, { maxLogSize: this.options.maxLogSize, + runSummaryLog: this.options.runSummaryLog, }); // Register as global service — other plugins access via ctx.getService('automation') diff --git a/packages/services/service-automation/src/run-summary.test.ts b/packages/services/service-automation/src/run-summary.test.ts new file mode 100644 index 0000000000..e9fa5a4552 --- /dev/null +++ b/packages/services/service-automation/src/run-summary.test.ts @@ -0,0 +1,661 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// #4354 — per-run summaries. The premise: a scheduled flow that selects records +// and acts on none is INDISTINGUISHABLE from one with nothing to do. Both report +// success, both write nothing, both stay silent. These tests pin the three +// surfaces that separate them — a structured log line, a value on the run +// result, and queryable columns on `sys_automation_run` — and the counters they +// all read. + +import { describe, it, expect } from 'vitest'; +import { AutomationEngine } from './engine.js'; +import type { StepLogEntry, NodeExecutor, RunRecord } from './engine.js'; +import { summarizeRun, formatRunSummaryLine } from './run-summary.js'; +import { InMemorySuspendedRunStore, ObjectStoreSuspendedRunStore } from './suspended-run-store.js'; +import { registerCrudNodes } from './builtin/crud-nodes.js'; +import { registerLoopNode } from './builtin/loop-node.js'; +import { registerLogicNodes } from './builtin/logic-nodes.js'; +import { registerNotifyNode } from './builtin/notify-node.js'; +import { registerSubflowNode } from './builtin/subflow-node.js'; +import { registerMapNode } from './builtin/map-node.js'; +import type { AutomationContext } from '@objectstack/spec/contracts'; + +const AT = '2026-07-31T00:00:00.000Z'; + +function makeLogger(): any { + const info: Array<{ msg: string; meta?: any }> = []; + const debug: Array<{ msg: string; meta?: any }> = []; + const l: any = { + info(msg: string, meta?: any) { info.push({ msg, meta }); }, + debug(msg: string, meta?: any) { debug.push({ msg, meta }); }, + warn() {}, + error() {}, + }; + l.child = () => l; + l.lines = info; + l.debugLines = debug; + return l; +} + +const step = (over: Partial & { nodeId: string }): StepLogEntry => ({ + nodeType: 'noop', + status: 'success', + startedAt: AT, + ...over, +}); + +// ── The fold ──────────────────────────────────────────────────────────────── + +describe('summarizeRun', () => { + it('sums selected / acted across the run', () => { + const s = summarizeRun([ + step({ nodeId: 'query', nodeType: 'get_record', metrics: { selected: 30 } }), + step({ nodeId: 'write', nodeType: 'create_record', metrics: { acted: 1 } }), + step({ nodeId: 'write2', nodeType: 'update_record', metrics: { acted: 4 } }), + ]); + expect(s.selected).toBe(30); + expect(s.acted).toBe(5); + expect(s.skipped).toBe(0); + }); + + it('folds repeated executions of one node into a single entry', () => { + // A loop body node that ran 30 times is ONE node entry with runs: 30 — + // not 30 entries the reader has to add up. + const steps = Array.from({ length: 30 }, (_, i) => + step({ nodeId: 'write', nodeType: 'create_record', metrics: { acted: 1 }, parentNodeId: 'each', iteration: i }), + ); + const s = summarizeRun(steps); + expect(s.nodes).toHaveLength(1); + expect(s.nodes[0]).toMatchObject({ nodeId: 'write', runs: 30, acted: 30, status: 'success' }); + expect(s.acted).toBe(30); + }); + + it('omits selected / acted for a node that reports no metrics', () => { + // Absent means "this kind of node touches no records" — different from + // `0`, which means "it looked and found none". + const s = summarizeRun([step({ nodeId: 'gate', nodeType: 'decision' })]); + expect(s.nodes[0].selected).toBeUndefined(); + expect(s.nodes[0].acted).toBeUndefined(); + }); + + it('counts gate skips per gate, most-skipped first, and excludes them from runs', () => { + const steps: StepLogEntry[] = []; + for (let i = 0; i < 3; i++) { + steps.push(step({ + nodeId: 'nudge', nodeType: 'notify', status: 'skipped', + skippedBy: { nodeId: 'gate', edgeId: 'b1', label: 'Go' }, + })); + } + steps.push(step({ + nodeId: 'other', nodeType: 'notify', status: 'skipped', + skippedBy: { nodeId: 'gate2', edgeId: 'b2' }, + })); + + const s = summarizeRun(steps); + expect(s.skipped).toBe(4); + expect(s.gates).toEqual([ + { nodeId: 'gate', targetNodeId: 'nudge', edgeId: 'b1', label: 'Go', skipped: 3 }, + { nodeId: 'gate2', targetNodeId: 'other', edgeId: 'b2', skipped: 1 }, + ]); + // A node that only ever got skipped never RAN. + expect(s.nodes.find((n) => n.nodeId === 'nudge')).toMatchObject({ runs: 0, skipped: 3, status: 'skipped' }); + }); + + it('reports a node as failed when any of its executions failed', () => { + const s = summarizeRun([ + step({ nodeId: 'call', nodeType: 'http' }), + step({ nodeId: 'call', nodeType: 'http', status: 'failure' }), + step({ nodeId: 'call', nodeType: 'http' }), + ]); + expect(s.nodes[0]).toMatchObject({ nodeId: 'call', runs: 3, failures: 1, status: 'failure' }); + }); + + it('keeps the counts of a node that failed part way through writing', () => { + const s = summarizeRun([ + step({ nodeId: 'w', nodeType: 'update_record', status: 'failure', metrics: { acted: 7 } }), + ]); + expect(s.acted).toBe(7); + }); +}); + +describe('formatRunSummaryLine', () => { + it('is one greppable line carrying selected / acted and the top gate', () => { + const line = formatRunSummaryLine( + { flowName: 'stalled_deal_sweep', runId: 'run_1', status: 'completed', durationMs: 42 }, + summarizeRun([ + step({ nodeId: 'q', nodeType: 'get_record', metrics: { selected: 30 } }), + ...Array.from({ length: 30 }, () => step({ + nodeId: 'nudge', nodeType: 'notify', status: 'skipped', + skippedBy: { nodeId: 'gate', edgeId: 'b1' }, + })), + ]), + ); + 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', + ); + }); +}); + +// ── End to end: the shape #4347 shipped ───────────────────────────────────── + +/** + * A sweep in the shape that shipped inert three times over: query rows, loop + * them, and act only when a gate opens. `shouldRun` decides whether the gate + * opens, which is the difference between "nothing to do" and "broken" — the + * two states that used to look identical from outside. + */ +function sweepFlow(name: string) { + const condition = { dialect: 'cel', source: 'row.shouldRun == true' }; + return { + name, label: name, type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { + id: 'query', type: 'get_record', label: 'Query', + config: { objectName: 'deal', filter: { stalled: true }, limit: 10, outputVariable: 'rows' }, + }, + { + id: 'each', type: 'loop', label: 'Each', + config: { + collection: '{rows}', iteratorVariable: 'row', + body: { + nodes: [ + { id: 'gate', type: 'decision', label: 'Gate' }, + { id: 'nudge', type: 'create_record', label: 'Nudge', config: { objectName: 'nudge', fields: { note: 'x' } } }, + ], + edges: [{ id: 'b1', source: 'gate', target: 'nudge', type: 'conditional', condition, label: 'Go' }], + }, + }, + }, + { 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' }, + ], + }; +} + +function sweepEngine(rows: Array>, logger = makeLogger(), store?: any) { + const written: Array> = []; + let seq = 0; + const data: any = { + async find() { return rows; }, + async findOne() { return rows[0] ?? null; }, + async insert(obj: string, fields: any) { seq += 1; const r = { id: `${obj}_${seq}`, ...fields }; written.push(r); return r; }, + async update() { return 0; }, + async delete() { return 0; }, + }; + const ctx: any = { logger, getService: (n: string) => (n === 'data' ? data : undefined) }; + const engine = new AutomationEngine(logger, store); + registerCrudNodes(engine, ctx); + registerLogicNodes(engine, ctx); + registerLoopNode(engine, ctx); + return { engine, written, logger }; +} + +describe('flow run summary — a sweep that selects rows and writes none (#4354)', () => { + it('reports selected 3 / acted 0 and NAMES the gate that closed', async () => { + const rows = [1, 2, 3].map((i) => ({ id: `d${i}`, shouldRun: false })); + const { engine, written } = sweepEngine(rows); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + + // The old world stopped here: success, nothing written, no signal. + expect(res.success).toBe(true); + expect(written).toHaveLength(0); + + // The new one keeps going. + expect(res.summary).toBeDefined(); + expect(res.summary!.selected).toBe(3); + expect(res.summary!.acted).toBe(0); + expect(res.summary!.skipped).toBe(3); + expect(res.summary!.gates).toEqual([ + { nodeId: 'gate', targetNodeId: 'nudge', edgeId: 'b1', label: 'Go', skipped: 3 }, + ]); + expect(res.summary!.nodes.find((n) => n.nodeId === 'nudge')).toMatchObject({ + runs: 0, skipped: 3, status: 'skipped', + }); + }); + + it('reports selected 3 / acted 3 for the SAME flow once the gate opens', async () => { + // Same graph, same nodes, same query — only the data differs. If the + // summary could not tell these two runs apart it would be worthless. + const rows = [1, 2, 3].map((i) => ({ id: `d${i}`, shouldRun: true })); + const { engine, written } = sweepEngine(rows); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + expect(res.success).toBe(true); + expect(written).toHaveLength(3); + expect(res.summary).toMatchObject({ selected: 3, acted: 3, skipped: 0, gates: [] }); + }); + + it('distinguishes "nothing to do" — selected 0, acted 0, no gate blamed', async () => { + const { engine } = sweepEngine([]); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + expect(res.summary).toMatchObject({ selected: 0, acted: 0, skipped: 0, gates: [] }); + }); + + it('logs ONE structured line per terminal run', async () => { + const rows = [1, 2, 3].map((i) => ({ id: `d${i}`, shouldRun: false })); + const { engine, logger } = sweepEngine(rows); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + + const lines = logger.lines.filter((l: any) => l.msg.startsWith('[automation] run flow=')); + expect(lines).toHaveLength(1); + expect(lines[0].msg).toContain('selected=3 acted=0 skipped=3'); + expect(lines[0].msg).toContain('gate=gate->nudge:3'); + expect(lines[0].meta).toMatchObject({ flow: 'sweep', selected: 3, acted: 0, skipped: 3 }); + }); + + it('runSummaryLog:"off" silences the line but still MEASURES the run', async () => { + const rows = [{ id: 'd1', shouldRun: false }]; + const logger = makeLogger(); + const { engine } = sweepEngine(rows, logger); + (engine as unknown as { runSummaryLog: string }).runSummaryLog = 'off'; + engine.registerFlow('sweep', sweepFlow('sweep') as never); + + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + expect(logger.lines.filter((l: any) => l.msg.startsWith('[automation] run flow='))).toHaveLength(0); + // Turning the narration down must never turn the measurement off. + expect(res.summary).toMatchObject({ selected: 1, acted: 0, skipped: 1 }); + }); + + it('records a gate that closes on every one of 40 iterations without inflating runs', async () => { + // The observability signal must not change execution semantics: 40 skips + // are 40 non-events, so `runs` for the skipped node stays 0 and the run + // still completes. (The top-level twin of this — a closed back-edge not + // counting toward the ADR-0044 re-entry cap — is pinned in engine.test.ts.) + const rows = Array.from({ length: 40 }, (_, i) => ({ id: `d${i}`, shouldRun: false })); + const { engine } = sweepEngine(rows); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + expect(res.success).toBe(true); // not aborted as a runaway loop + expect(res.summary!.skipped).toBe(40); + expect(res.summary!.nodes.find((n) => n.nodeId === 'nudge')!.runs).toBe(0); + }); +}); + +// ── Durability ────────────────────────────────────────────────────────────── + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +describe('run summary durability', () => { + it('survives a restart via listRuns / getRun', async () => { + const rows = [1, 2, 3].map((i) => ({ id: `d${i}`, shouldRun: false })); + const store = new InMemorySuspendedRunStore(); + const { engine } = sweepEngine(rows, makeLogger(), store); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + await flush(); // recordTerminal is fire-and-forget + + // Fresh engine, empty in-memory logs, same durable store. + const fresh = new AutomationEngine(makeLogger(), store); + const [listed] = await fresh.listRuns('sweep', { limit: 1 }); + expect(listed.summary).toMatchObject({ selected: 3, acted: 0, skipped: 3 }); + const run = await fresh.getRun(listed.id); + expect(run!.summary!.gates[0]).toMatchObject({ nodeId: 'gate', skipped: 3 }); + }); + + it('counts stay EXACT when the step log is compacted past its cap', async () => { + // The counts are folded before compaction, so a 300-iteration sweep does + // not silently shrink to the 200 steps that fit in the history row. + const rows = Array.from({ length: 300 }, (_, i) => ({ id: `d${i}`, shouldRun: true })); + const store = new InMemorySuspendedRunStore(); + const { engine } = sweepEngine(rows, makeLogger(), store); + engine.registerFlow('sweep', sweepFlow('sweep') as never); + await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + await flush(); + + const fresh = new AutomationEngine(makeLogger(), store); + const [listed] = await fresh.listRuns('sweep', { limit: 1 }); + expect(listed.steps.length).toBeLessThanOrEqual(200); // compacted… + expect(listed.summary).toMatchObject({ selected: 300, acted: 300 }); // …counts are not + }); + + it('writes the counts as QUERYABLE columns on sys_automation_run', async () => { + // `selected_count > 0 AND acted_count = 0` has to be a filter an operator + // can alert on, not a value buried inside a JSON blob. + const rows: any[] = []; + const engine: any = { + async find() { return []; }, + async insert(_obj: string, row: any) { rows.push(row); return row; }, + async update() { return 1; }, + async delete() { return 1; }, + }; + const store = new ObjectStoreSuspendedRunStore(engine); + const record: RunRecord = { + runId: 'r1', flowName: 'sweep', status: 'completed', startedAt: AT, + summary: { + selected: 30, acted: 0, skipped: 30, + nodes: [{ nodeId: 'q', nodeType: 'get_record', status: 'success', runs: 1, failures: 0, skipped: 0, selected: 30 }], + gates: [{ nodeId: 'gate', targetNodeId: 'nudge', skipped: 30 }], + }, + }; + await store.recordTerminal(record); + + expect(rows).toHaveLength(1); + expect(rows[0].selected_count).toBe(30); + expect(rows[0].acted_count).toBe(0); + expect(rows[0].skipped_count).toBe(30); + expect(JSON.parse(rows[0].summary_json).gates[0].nodeId).toBe('gate'); + }); + + it('leaves the count columns NULL — never 0 — when no summary was computed', async () => { + // "Not measured" must not read as "measured zero", or every legacy row + // becomes a false alarm the first time someone writes the query. + const rows: any[] = []; + const engine: any = { + async find() { return []; }, + async insert(_obj: string, row: any) { rows.push(row); return row; }, + async update() { return 1; }, + async delete() { return 1; }, + }; + await new ObjectStoreSuspendedRunStore(engine).recordTerminal({ + runId: 'r1', flowName: 'sweep', status: 'completed', startedAt: AT, + }); + expect(rows[0].selected_count).toBeNull(); + expect(rows[0].acted_count).toBeNull(); + expect(rows[0].summary_json).toBeNull(); + }); +}); + +// ── Executor-reported metrics ─────────────────────────────────────────────── + +describe('node executors report what they touched', () => { + const flowOf = (nodes: any[], edges: any[]) => ({ + name: 'f', label: 'f', type: 'autolaunched', runAs: 'system', nodes, edges, + }); + + it('update_record reports the row COUNT a bulk write returned', async () => { + const logger = makeLogger(); + const data: any = { + async find() { return []; }, + async findOne() { return null; }, + async insert(_o: string, f: any) { return { id: 'x', ...f }; }, + async update() { return 7; }, // driver.updateMany contract: Promise + async delete() { return 0; }, + }; + const engine = new AutomationEngine(logger); + registerCrudNodes(engine, { logger, getService: (n: string) => (n === 'data' ? data : undefined) } as never); + engine.registerFlow('f', flowOf( + [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'u', type: 'update_record', label: 'U', config: { objectName: 'deal', filter: { stalled: true }, fields: { nudged: true } } }, + { id: 'end', type: 'end', label: 'E' }, + ], + [{ id: 'e1', source: 'start', target: 'u' }, { id: 'e2', source: 'u', target: 'end' }], + ) as never); + + const res = await engine.execute('f', { event: 'schedule' } as AutomationContext); + expect(res.summary).toMatchObject({ acted: 7, selected: 0 }); + }); + + it('delete_record reports 0 when the driver reports nothing deleted', async () => { + const logger = makeLogger(); + const data: any = { + async find() { return []; }, + async findOne() { return null; }, + async delete() { return false; }, + }; + const engine = new AutomationEngine(logger); + registerCrudNodes(engine, { logger, getService: (n: string) => (n === 'data' ? data : undefined) } as never); + engine.registerFlow('f', flowOf( + [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'd', type: 'delete_record', label: 'D', config: { objectName: 'deal', filter: { stale: true } } }, + { id: 'end', type: 'end', label: 'E' }, + ], + [{ id: 'e1', source: 'start', target: 'd' }, { id: 'e2', source: 'd', target: 'end' }], + ) as never); + + const res = await engine.execute('f', { event: 'schedule' } as AutomationContext); + expect(res.summary!.acted).toBe(0); + }); + + it('notify counts DELIVERED notifications as acted — a nudge sweep acts by notifying', async () => { + const logger = makeLogger(); + const messaging: any = { async emit() { return { notificationId: 'n1', delivered: 4, failed: 0 }; } }; + const engine = new AutomationEngine(logger); + registerNotifyNode(engine, { logger, getService: (n: string) => (n === 'messaging' ? messaging : undefined) } as never); + engine.registerFlow('f', flowOf( + [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'n', type: 'notify', label: 'N', config: { recipients: ['u1'], title: 'Stalled deal' } }, + { id: 'end', type: 'end', label: 'E' }, + ], + [{ id: 'e1', source: 'start', target: 'n' }, { id: 'e2', source: 'n', target: 'end' }], + ) as never); + + const res = await engine.execute('f', { event: 'schedule' } as AutomationContext); + expect(res.summary!.acted).toBe(4); + }); + + it('notify with NO messaging service reports acted 0 — green, but inert', async () => { + const logger = makeLogger(); + const engine = new AutomationEngine(logger); + registerNotifyNode(engine, { logger, getService: () => undefined } as never); + engine.registerFlow('f', flowOf( + [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'n', type: 'notify', label: 'N', config: { recipients: ['u1'], title: 'Stalled deal' } }, + { id: 'end', type: 'end', label: 'E' }, + ], + [{ id: 'e1', source: 'start', target: 'n' }, { id: 'e2', source: 'n', target: 'end' }], + ) as never); + + const res = await engine.execute('f', { event: 'schedule' } as AutomationContext); + expect(res.success).toBe(true); + expect(res.summary!.acted).toBe(0); + }); + + it('subflow rolls its child run up, so a parent that delegates is not read as inert', async () => { + const logger = makeLogger(); + const written: any[] = []; + const data: any = { + async find() { return []; }, + async findOne() { return null; }, + async insert(o: string, f: any) { const r = { id: `${o}_1`, ...f }; written.push(r); return r; }, + }; + const ctx: any = { logger, getService: (n: string) => (n === 'data' ? data : undefined) }; + const engine = new AutomationEngine(logger); + registerCrudNodes(engine, ctx); + registerSubflowNode(engine, ctx); + + engine.registerFlow('child', flowOf( + [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'w', type: 'create_record', label: 'W', config: { objectName: 'nudge', fields: { note: 'x' } } }, + { id: 'end', type: 'end', label: 'E' }, + ], + [{ id: 'e1', source: 'start', target: 'w' }, { id: 'e2', source: 'w', target: 'end' }], + ) as never); + engine.registerFlow('parent', { + name: 'parent', label: 'parent', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'sub', type: 'subflow', label: 'Sub', config: { flowName: 'child' } }, + { id: 'end', type: 'end', label: 'E' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'sub' }, { id: 'e2', source: 'sub', target: 'end' }], + } as never); + + const res = await engine.execute('parent', { event: 'schedule' } as AutomationContext); + expect(written).toHaveLength(1); + // Without the roll-up the parent would report `acted: 0` — a healthy + // run indistinguishable from a broken one, which is the whole bug. + expect(res.summary!.acted).toBe(1); + }); +}); + +describe('map rolls its per-item child runs up', () => { + // `map` is a sweep construct ("process each row"), so a map whose items all + // write while the parent reports `acted: 0` would make the detector fire on + // healthy runs — worse than no signal, because it trains people to ignore it. + it('counts every item its per-item flow wrote', async () => { + const logger = makeLogger(); + const written: any[] = []; + const data: any = { + async find() { return []; }, + async findOne() { return null; }, + async insert(o: string, f: any) { const r = { id: `${o}_${written.length + 1}`, ...f }; written.push(r); return r; }, + }; + const ctx: any = { logger, getService: (n: string) => (n === 'data' ? data : undefined) }; + const engine = new AutomationEngine(logger, new InMemorySuspendedRunStore()); + registerCrudNodes(engine, ctx); + registerMapNode(engine, ctx); + + engine.registerFlow('per_item', { + name: 'per_item', label: 'per_item', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'w', type: 'create_record', label: 'W', config: { objectName: 'nudge', fields: { note: 'x' } } }, + { id: 'end', type: 'end', label: 'E' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'w' }, { id: 'e2', source: 'w', target: 'end' }], + } as never); + engine.registerFlow('sweep', { + name: 'sweep', label: 'sweep', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'S' }, + { + id: 'each', type: 'map', label: 'Each', + config: { collection: [{ id: 'a' }, { id: 'b' }, { id: 'c' }], flowName: 'per_item', iteratorVariable: 'row' }, + }, + { id: 'end', type: 'end', label: 'E' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'each' }, { id: 'e2', source: 'each', target: 'end' }], + } as never); + + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + expect(res.success).toBe(true); + expect(written).toHaveLength(3); + expect(res.summary!.acted).toBe(3); + }); +}); + +// ── Children that pause ───────────────────────────────────────────────────── + +describe('a child run that PAUSED still counts toward its parent', () => { + // The synchronous roll-up rides on NodeExecutionResult.metrics, which a + // paused child cannot use: the parent's step for the node was written at + // suspend time, before the child had done anything. The engine credits the + // child's totals to that step when it bubbles back. + + function makeEngine() { + const logger = makeLogger(); + const written: any[] = []; + const data: any = { + async find() { return []; }, + async findOne() { return null; }, + async insert(o: string, f: any) { const r = { id: `${o}_${written.length + 1}`, ...f }; written.push(r); return r; }, + }; + const ctx: any = { logger, getService: (n: string) => (n === 'data' ? data : undefined) }; + const engine = new AutomationEngine(logger, new InMemorySuspendedRunStore()); + registerCrudNodes(engine, ctx); + registerSubflowNode(engine, ctx); + engine.registerNodeExecutor({ + type: 'hold', + async execute() { return { success: true, suspend: true, correlation: 'held' }; }, + } as NodeExecutor); + // A child that pauses, then writes a row once resumed. + engine.registerFlow('child', { + name: 'child', label: 'child', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'hold', type: 'hold', label: 'H' }, + { id: 'w', type: 'create_record', label: 'W', config: { objectName: 'nudge', fields: { note: 'x' } } }, + { id: 'end', type: 'end', label: 'E' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'hold' }, + { id: 'e2', source: 'hold', target: 'w' }, + { id: 'e3', source: 'w', target: 'end' }, + ], + } as never); + engine.registerFlow('parent', { + name: 'parent', label: 'parent', type: 'autolaunched', runAs: 'system', + nodes: [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'sub', type: 'subflow', label: 'Sub', config: { flowName: 'child' } }, + { id: 'end', type: 'end', label: 'E' }, + ], + edges: [{ id: 'e1', source: 'start', target: 'sub' }, { id: 'e2', source: 'sub', target: 'end' }], + } as never); + return { engine, written }; + } + + it('credits the child when resuming the CHILD bubbles up to the parent', async () => { + const { engine, written } = makeEngine(); + const paused = await engine.execute('parent', { event: 'schedule' } as AutomationContext); + expect(paused.status).toBe('paused'); + + // Resume the child directly (what an approval service / wait timer does). + const [childRun] = await engine.listRuns('child', { limit: 1 }); + const done = await engine.resume(childRun.id); + expect(done.success).toBe(true); + expect(written).toHaveLength(1); + + const parent = await engine.getRun(paused.runId!); + expect(parent!.status).toBe('completed'); + expect(parent!.summary!.acted).toBe(1); + }); + + it('credits the child when resuming the PARENT delegates down', async () => { + const { engine, written } = makeEngine(); + const paused = await engine.execute('parent', { event: 'schedule' } as AutomationContext); + expect(paused.status).toBe('paused'); + + // Resume the parent (what a UI holding the launch run id does). + const done = await engine.resume(paused.runId!); + expect(done.success).toBe(true); + expect(written).toHaveLength(1); + expect(done.summary!.acted).toBe(1); + }); +}); + +// ── The engine's own contract with executors ──────────────────────────────── + +describe('engine ↔ executor metrics plumbing', () => { + it('copies an executor\'s metrics onto its step, and no further', async () => { + const logger = makeLogger(); + const engine = new AutomationEngine(logger); + engine.registerNodeExecutor({ + type: 'counter', + async execute() { return { success: true, metrics: { selected: 5, acted: 2 } }; }, + } as NodeExecutor); + engine.registerNodeExecutor({ + type: 'quiet', + async execute() { return { success: true }; }, + } as NodeExecutor); + engine.registerFlow('f', { + name: 'f', label: 'f', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'S' }, + { id: 'c', type: 'counter', label: 'C' }, + { id: 'q', type: 'quiet', label: 'Q' }, + { id: 'end', type: 'end', label: 'E' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'c' }, + { id: 'e2', source: 'c', target: 'q' }, + { id: 'e3', source: 'q', target: 'end' }, + ], + } as never); + + const res = await engine.execute('f', {} as AutomationContext); + expect(res.summary).toMatchObject({ selected: 5, acted: 2 }); + const run = (await engine.listRuns('f'))[0]; + expect(run.steps.find((s) => s.nodeId === 'c')!.metrics).toEqual({ selected: 5, acted: 2 }); + expect(run.steps.find((s) => s.nodeId === 'q')!.metrics).toBeUndefined(); + }); +}); diff --git a/packages/services/service-automation/src/run-summary.ts b/packages/services/service-automation/src/run-summary.ts new file mode 100644 index 0000000000000000000000000000000000000000..b3e58d8df7a121e66dd96475f0f319ba102adb77 GIT binary patch literal 5451 zcmai2+j85;5zVu{qSx74Qd$Y*dN+BHLwTKKz2%fGr)*c{L8S^nU`WCOF$@L}&DF9h zACWK2m*kwD!G*eH@j_yzr>Fb$9g~Ut#cp<0zFgHZPLJjJ^!Xp<{rs~@>yLGk-bC^? zPfh7eCQX@{O6rx7SDPeVnX~87R0_Wv9UbNCjjd{_cN-&jM}p^dVYeTe^5>*BADi_$ zsdlHs^-G&Qy57<5Wl`C+jDEB#kE1*jw=wA?X==Mp>fDy27l#+i$J%V(+T~AWUG1dh z5KTxS7RJSl+uc8szs72E${S%bF;9OIhTZ-91zW# z0yu*SM5$LvEzT4$gp4F`yk3E{ zUghPITbY&r{`)^N$BHZ~W6D$i-{l23gcsgG`f!fAO6*0!WfwA)Z@!Tvt?@FO&14#j zy@lSXg`1>Jo!=T_3P{Z+sbG9Zr&+gew)>V_%V3Gl+B>;DTRAOMWiDe=&g(wPB=nK5!pJ`I3tC;fE z37n$BF&u}$dF{w>ZnK@<4(Xi4=_)TWx|NI702I)>lei82-|Occhnp^0(=^LdahCiH z(Aj~x0T}cLYF$(&!7U(%@-zue%fp>i@JYKZJql@3Rk<3S{MM$f^3;)iYa{Z^Y%?#+ znRmx5lE0hktVE1KSkU9NVw|Lvb*iTMqSgv{snSZWo>wKMd_R>sU*jN{zrXxR`Rqhp zDyD(YK@qoRja@TeAmC$CQxW>7`I4Fqbv{-7@>=CIMLs7MbbV2D2Qlc7Qf>{^%u5dl z;SXR0P<6e(P^GR6(~rMedE!WI>!hF&l^r$-Xh1L0*oAKZ9rCL*7$GKO*vYdk<1fo; z6et9BB~2P$P0B+H;4?# zl9?2V2w_|z(@|b^Mc~i)&da2LczBp6YB+8N5N1-Et-MP%e{%5A5+M_c#${g|F2=_% z1_wQQPaYgpo9uQU^aS4;`7Ax;l?7#3*>CL;v<`Z z--Erqw_K|>Rd$g(MEX=U8Q{G?QY-Qhp{EHWFS(YlPWb!gfX}}YdYhz$WdnqSgoXs^ z0uG?Trv_v4Mq(?ONRDGiD;Xtnczu>2R$@BuMmPe(g6obbk7n$pMQ;#FewV`q zy&q_X;Ujy;p4{U2HYRAGo6 zhUiZ)okQn_&JZnJJBVR+X5$Mpui5x%|BU9{wez#K$h11P^L4Y9+nmiMJ7JVj$0?~T zvI~O-vohY~JVHOx8D(p$8}!hA#MPQ**`KQE&dDmqq`3j&A>d~t`IS>ImK4m{bQFu9 z{4nk0Sr^67G zs7w4fm`F^IBlQESYN4ltZ-I87sQW*GA!pzm&;wR777awAqw`tKJQrdtRvrao`X zFq$aWO=WNK4Z;zWhm!}=20qh#Xv*;YZ%MAbF02&cj$YsU6-`BtHjP`2 z^x1|bVc?Ab-XG;H4X=lyyRwL`|&!;q-`6`yThjz= 0 AND acted_count = 0`; the per-node / per-gate detail + // rides in the JSON blob. Null (not 0) when the engine computed no + // summary: "not measured" and "measured zero" are different answers, and + // only one of them should trip an alarm. + selected_count: record.summary?.selected ?? null, + acted_count: record.summary?.acted ?? null, + skipped_count: record.summary?.skipped ?? null, + summary_json: record.summary ? serializeSummaryBounded(record.summary) : null, }; const existing = await this.engine.find(TABLE, { where: { id }, limit: 1, context: SYSTEM_CTX, @@ -345,6 +359,10 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { organizationId: row.organization_id ?? null, userId: row.user_id ?? undefined, steps: parseJson(row.steps_json, undefined), + // #4354 — rehydrate from `summary_json`, never re-fold `steps_json`: those + // steps are compacted (200 max), so recomputing would report a + // 5000-iteration sweep as having acted a couple of hundred times. + summary: parseJson(row.summary_json, undefined), }; } @@ -409,3 +427,27 @@ function serializeStepsBounded(steps: RunRecord['steps']): string | null { } return null; } + +/** + * JSON-encode a run summary under {@link MAX_SUMMARY_JSON_BYTES} (#4354). + * + * The detail arrays are bounded by the flow's STATIC shape — one entry per node + * that ran, one per gate that closed — not by iteration count, so a 5000-row + * sweep over a 6-node flow serializes six entries. A pathological flow with + * thousands of nodes is the only way over the cap; there the detail is dropped + * and the TOTALS are kept, because the totals are what the broken-sweep alert + * queries and losing them to a size limit would be the one unacceptable outcome. + * The dropped detail stays visible as `detailOmitted`, never silently absent. + */ +function serializeSummaryBounded(summary: NonNullable): string { + const json = JSON.stringify(summary); + if (json.length <= MAX_SUMMARY_JSON_BYTES) return json; + return JSON.stringify({ + selected: summary.selected, + acted: summary.acted, + skipped: summary.skipped, + nodes: [], + gates: [], + detailOmitted: true, + }); +} diff --git a/packages/services/service-automation/src/sys-automation-run.object.ts b/packages/services/service-automation/src/sys-automation-run.object.ts index 1c9098c5a4..c243b14e4c 100644 --- a/packages/services/service-automation/src/sys-automation-run.object.ts +++ b/packages/services/service-automation/src/sys-automation-run.object.ts @@ -59,7 +59,10 @@ export const SysAutomationRun = ObjectSchema.create({ displayNameField: 'id', nameField: 'id', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) titleFormat: '{flow_name} · {node_id}', - highlightFields: ['flow_name', 'node_id', 'status', 'correlation', 'started_at', 'updated_at'], + // `selected_count`/`acted_count` sit in the highlight set on purpose (#4354): + // "selected 30, acted 0" has to be visible on the run row itself, not one + // drill-down away — a signal you must click to find is a signal nobody sees. + highlightFields: ['flow_name', 'node_id', 'status', 'selected_count', 'acted_count', 'correlation', 'started_at', 'updated_at'], fields: { id: Field.text({ label: 'Run ID', required: true, readonly: true, group: 'System' }), @@ -182,6 +185,40 @@ export const SysAutomationRun = ObjectSchema.create({ group: 'Outcome', }), + // ── Run summary (#4354) ──────────────────────────────────────────────── + // COLUMNS, not just a blob: `selected_count > 0 AND acted_count = 0` over N + // consecutive runs is a near-perfect broken-sweep detector, and an operator + // can only alert on what is filterable. Buried inside `summary_json` these + // would be readable but not queryable — the difference between a dashboard + // and an alarm. + selected_count: Field.number({ + label: 'Records Selected', + required: false, + description: 'Records this run READ across its data nodes. Null on rows written before run summaries existed — which is NOT the same as zero.', + group: 'Outcome', + }), + + acted_count: Field.number({ + label: 'Records Acted On', + required: false, + description: 'Records this run created / updated / deleted, plus effects dispatched (notifications delivered). `selected_count > 0 AND acted_count = 0` over consecutive runs is the broken-sweep signal.', + group: 'Outcome', + }), + + skipped_count: Field.number({ + label: 'Gate Skips', + required: false, + description: 'Node executions a closed gate prevented — one per loop iteration whose conditional edge evaluated false. Many skips with no writes names the gate as the suspect.', + group: 'Outcome', + }), + + summary_json: Field.textarea({ + label: 'Run Summary', + required: false, + description: 'JSON per-node breakdown (terminal status, runs, failures, selected/acted) plus which gates closed and how often. Folded from the FULL step log, so its counts stay exact even when `steps_json` is compacted.', + group: 'Outcome', + }), + created_at: Field.datetime({ label: 'Created At', required: true, diff --git a/packages/spec/api-surface.json b/packages/spec/api-surface.json index aff1d07f2e..c9281af88b 100644 --- a/packages/spec/api-surface.json +++ b/packages/spec/api-surface.json @@ -2128,6 +2128,10 @@ "ExecutionStepLog (type)", "ExecutionStepLogParsed (type)", "ExecutionStepLogSchema (const)", + "ExecutionStepMetrics (type)", + "ExecutionStepMetricsSchema (const)", + "ExecutionStepSkipReason (type)", + "ExecutionStepSkipReasonSchema (const)", "FLOW_BUILTIN_NODE_TYPES (const)", "FLOW_NODE_EXPRESSION_PATHS (const)", "FLOW_STRUCTURAL_NODE_TYPES (const)", @@ -2145,6 +2149,13 @@ "FlowRegion (type)", "FlowRegionParsed (type)", "FlowRegionSchema (const)", + "FlowRunGateSummary (type)", + "FlowRunGateSummarySchema (const)", + "FlowRunNodeSummary (type)", + "FlowRunNodeSummarySchema (const)", + "FlowRunSummary (type)", + "FlowRunSummaryParsed (type)", + "FlowRunSummarySchema (const)", "FlowSchema (const)", "FlowVariableSchema (const)", "FlowVersionHistory (type)", diff --git a/packages/spec/authorable-surface.json b/packages/spec/authorable-surface.json index 036e34c53b..cde48c6e1a 100644 --- a/packages/spec/authorable-surface.json +++ b/packages/spec/authorable-surface.json @@ -2353,6 +2353,7 @@ "automation/ExecutionLog:startedAt", "automation/ExecutionLog:status", "automation/ExecutionLog:steps", + "automation/ExecutionLog:summary", "automation/ExecutionLog:tenantId", "automation/ExecutionLog:trigger", "automation/ExecutionLog:variables", @@ -2361,6 +2362,7 @@ "automation/ExecutionStepLog:error", "automation/ExecutionStepLog:input", "automation/ExecutionStepLog:iteration", + "automation/ExecutionStepLog:metrics", "automation/ExecutionStepLog:nodeId", "automation/ExecutionStepLog:nodeLabel", "automation/ExecutionStepLog:nodeType", @@ -2368,8 +2370,14 @@ "automation/ExecutionStepLog:parentNodeId", "automation/ExecutionStepLog:regionKind", "automation/ExecutionStepLog:retryAttempt", + "automation/ExecutionStepLog:skippedBy", "automation/ExecutionStepLog:startedAt", "automation/ExecutionStepLog:status", + "automation/ExecutionStepMetrics:acted", + "automation/ExecutionStepMetrics:selected", + "automation/ExecutionStepSkipReason:edgeId", + "automation/ExecutionStepSkipReason:label", + "automation/ExecutionStepSkipReason:nodeId", "automation/Flow:_lock", "automation/Flow:_lockDocsUrl", "automation/Flow:_lockReason", @@ -2413,6 +2421,26 @@ "automation/FlowNode:waitEventConfig", "automation/FlowRegion:edges", "automation/FlowRegion:nodes", + "automation/FlowRunGateSummary:edgeId", + "automation/FlowRunGateSummary:label", + "automation/FlowRunGateSummary:nodeId", + "automation/FlowRunGateSummary:skipped", + "automation/FlowRunGateSummary:targetNodeId", + "automation/FlowRunNodeSummary:acted", + "automation/FlowRunNodeSummary:failures", + "automation/FlowRunNodeSummary:nodeId", + "automation/FlowRunNodeSummary:nodeLabel", + "automation/FlowRunNodeSummary:nodeType", + "automation/FlowRunNodeSummary:runs", + "automation/FlowRunNodeSummary:selected", + "automation/FlowRunNodeSummary:skipped", + "automation/FlowRunNodeSummary:status", + "automation/FlowRunSummary:acted", + "automation/FlowRunSummary:detailOmitted", + "automation/FlowRunSummary:gates", + "automation/FlowRunSummary:nodes", + "automation/FlowRunSummary:selected", + "automation/FlowRunSummary:skipped", "automation/FlowVariable:isInput", "automation/FlowVariable:isOutput", "automation/FlowVariable:name", diff --git a/packages/spec/json-schema.manifest.json b/packages/spec/json-schema.manifest.json index f789d534fa..f0061d0498 100644 --- a/packages/spec/json-schema.manifest.json +++ b/packages/spec/json-schema.manifest.json @@ -550,11 +550,16 @@ "automation/ExecutionLog", "automation/ExecutionStatus", "automation/ExecutionStepLog", + "automation/ExecutionStepMetrics", + "automation/ExecutionStepSkipReason", "automation/Flow", "automation/FlowEdge", "automation/FlowNode", "automation/FlowNodeAction", "automation/FlowRegion", + "automation/FlowRunGateSummary", + "automation/FlowRunNodeSummary", + "automation/FlowRunSummary", "automation/FlowVariable", "automation/FlowVersionHistory", "automation/GetRecordConfig", diff --git a/packages/spec/src/automation/execution.test.ts b/packages/spec/src/automation/execution.test.ts index a9203be241..261662a7ca 100644 --- a/packages/spec/src/automation/execution.test.ts +++ b/packages/spec/src/automation/execution.test.ts @@ -8,6 +8,7 @@ import { CheckpointSchema, ConcurrencyPolicySchema, ScheduleStateSchema, + FlowRunSummarySchema, } from './execution.zod'; // ========================================== @@ -169,6 +170,69 @@ describe('ExecutionLogSchema', () => { }); }); +// ========================================== +// Flow Run Summary (#4354) +// ========================================== + +describe('FlowRunSummarySchema', () => { + const brokenSweep = { + selected: 30, + acted: 0, + skipped: 30, + nodes: [ + { nodeId: 'query', nodeType: 'get_record', status: 'success' as const, runs: 1, failures: 0, skipped: 0, selected: 30 }, + { nodeId: 'nudge', nodeType: 'notify', status: 'skipped' as const, runs: 0, failures: 0, skipped: 30 }, + ], + gates: [{ nodeId: 'gate', targetNodeId: 'nudge', edgeId: 'b1', label: 'Go', skipped: 30 }], + }; + + it('accepts the shape that separates a broken sweep from an idle one', () => { + const summary = FlowRunSummarySchema.parse(brokenSweep); + expect(summary.selected).toBe(30); + expect(summary.acted).toBe(0); + expect(summary.gates[0].nodeId).toBe('gate'); + }); + + it('leaves selected / acted absent for a node that touches no records', () => { + const summary = FlowRunSummarySchema.parse({ + selected: 0, acted: 0, skipped: 0, gates: [], + nodes: [{ nodeId: 'gate', nodeType: 'decision', status: 'success' as const, runs: 1, failures: 0, skipped: 0 }], + }); + // Absent ≠ 0: "this kind of node reads nothing" is not "it read nothing". + expect(summary.nodes[0].selected).toBeUndefined(); + expect(summary.nodes[0].acted).toBeUndefined(); + }); + + it('rejects negative counts', () => { + expect(() => FlowRunSummarySchema.parse({ ...brokenSweep, acted: -1 })).toThrow(); + }); + + it('rides on an execution log', () => { + const log = ExecutionLogSchema.parse({ + id: 'exec_004', + flowName: 'stalled_deal_sweep', + status: 'completed', + trigger: { type: 'schedule' }, + steps: [], + startedAt: '2026-02-01T10:00:00Z', + summary: brokenSweep, + }); + expect(log.summary?.acted).toBe(0); + }); + + it('is optional — an un-summarized run is not a run that did nothing', () => { + const log = ExecutionLogSchema.parse({ + id: 'exec_005', + flowName: 'legacy_flow', + status: 'completed', + trigger: { type: 'schedule' }, + steps: [], + startedAt: '2026-02-01T10:00:00Z', + }); + expect(log.summary).toBeUndefined(); + }); +}); + // ========================================== // Execution Error // ========================================== diff --git a/packages/spec/src/automation/execution.zod.ts b/packages/spec/src/automation/execution.zod.ts index 131ea0a2ae..0daa5c989a 100644 --- a/packages/spec/src/automation/execution.zod.ts +++ b/packages/spec/src/automation/execution.zod.ts @@ -38,6 +38,43 @@ export type ExecutionStatus = z.infer; // 2. Execution Log // ========================================== +/** + * What one node execution did to the data, reported by the node executor + * itself (#4354). + * + * A scheduled sweep that selects records and then writes none looks — in every + * surface the platform had — exactly like a sweep with nothing to do: both + * report success, emit no log and write nothing. These two counters are what + * tells them apart, so they are declared by the executor rather than inferred + * by the engine from a node's output shape: only the node knows whether its + * `result` was a row count, a record, or a boolean. + * + * Absent ⇒ the node touched no records (a `decision`, an `assignment`), which + * is different from `0` — "read nothing" is a fact, "reads nothing" is a kind. + */ +export const ExecutionStepMetricsSchema = lazySchema(() => z.object({ + selected: z.number().int().min(0).optional() + .describe('Records this node READ or matched (a `get_record` query, a lookup)'), + acted: z.number().int().min(0).optional() + .describe('Records this node WROTE (created / updated / deleted) or effects it dispatched (notifications delivered)'), +})); +export type ExecutionStepMetrics = z.infer; + +/** + * The gate that kept a step from running — recorded on a `skipped` step so the + * run trace names *which* condition closed, not merely that something did. + * + * This is the signal #4347 had no way to emit: a conditional edge inside a + * `loop` body evaluated false on every iteration, so the flow selected every + * stalled deal and nudged nobody, silently and green. + */ +export const ExecutionStepSkipReasonSchema = lazySchema(() => z.object({ + nodeId: z.string().describe('Node whose out-edge did not open (the gate)'), + edgeId: z.string().optional().describe('Edge whose condition evaluated false'), + label: z.string().optional().describe('Edge label, when the flow names its branches'), +})); +export type ExecutionStepSkipReason = z.infer; + /** * Execution Step Log Entry * Records the result of executing a single node in the flow graph. @@ -65,9 +102,73 @@ export const ExecutionStepLogSchema = lazySchema(() => z.object({ parentNodeId: z.string().optional().describe('Enclosing structured-region container node ID (loop/parallel/try_catch)'), iteration: z.number().int().min(0).optional().describe('Zero-based loop iteration or parallel branch index of the enclosing region'), regionKind: z.string().optional().describe('Region kind the step ran in: loop-body | parallel-branch | try | catch'), + // #4354: what the step did to the data, and — for a `skipped` step — which + // gate stopped it. Both feed the run summary aggregated on ExecutionLog. + metrics: ExecutionStepMetricsSchema.optional() + .describe('Records this step selected / acted on, as reported by the node executor'), + skippedBy: ExecutionStepSkipReasonSchema.optional() + .describe('The gate that closed, when `status` is `skipped`'), })); export type ExecutionStepLog = z.infer; +// ========================================== +// 2b. Flow Run Summary (#4354) +// ========================================== + +/** + * One node's contribution to a run, folded across every time it ran — a loop + * body node that ran 30 times is ONE entry with `runs: 30`, not 30 entries. + */ +export const FlowRunNodeSummarySchema = lazySchema(() => z.object({ + nodeId: z.string().describe('Node ID'), + nodeType: z.string().describe('Node action type (e.g., "get_record", "decision")'), + nodeLabel: z.string().optional().describe('Human-readable node label'), + status: z.enum(['success', 'failure', 'skipped']) + .describe('Terminal status of the node across the run — `failure` if any execution failed, else `success` if any succeeded, else `skipped`'), + runs: z.number().int().min(0).describe('Times the node executed (loop iterations and parallel branches each count)'), + failures: z.number().int().min(0).describe('Executions that failed'), + skipped: z.number().int().min(0).describe('Times a closed gate kept this node from running at all'), + selected: z.number().int().min(0).optional().describe('Records read across every execution — omitted for a node that reads none'), + acted: z.number().int().min(0).optional().describe('Records written / effects dispatched across every execution — omitted for a node that writes none'), +})); +export type FlowRunNodeSummary = z.infer; + +/** A gate that closed during the run, and how often. */ +export const FlowRunGateSummarySchema = lazySchema(() => z.object({ + nodeId: z.string().describe('Node whose out-edge did not open (the gate)'), + targetNodeId: z.string().describe('Node the closed edge would have run'), + edgeId: z.string().optional().describe('Edge whose condition evaluated false'), + label: z.string().optional().describe('Edge label, when the flow names its branches'), + skipped: z.number().int().min(1).describe('Times this gate evaluated false (once per loop iteration)'), +})); +export type FlowRunGateSummary = z.infer; + +/** + * Per-run rollup of what a flow execution actually *did* (#4354). + * + * The counters exist to answer one question no other surface could: is a green + * run doing its job, or has it silently stopped? `selected > 0 && acted == 0` + * over consecutive runs is the broken-sweep signal — the platform ships the + * measurement so every flow gets it, rather than each app rebuilding a detector + * out of the same primitives that fail silently. + * + * Totals are sums over `nodes`, which is itself a fold of the run's step log, + * so a loop that ran a write 30 times contributes 30 to `acted`. A `subflow` + * node rolls its child run's totals up into this one — the child keeps its own + * run row, so the child's work is counted there too, deliberately: this summary + * answers "what did this run cause", not "what did this run's own nodes do". + */ +export const FlowRunSummarySchema = lazySchema(() => z.object({ + selected: z.number().int().min(0).describe('Total records read by the run'), + acted: z.number().int().min(0).describe('Total records written / effects dispatched by the run'), + skipped: z.number().int().min(0).describe('Total node executions a closed gate prevented'), + nodes: z.array(FlowRunNodeSummarySchema).describe('Per-node breakdown, in first-execution order'), + gates: z.array(FlowRunGateSummarySchema).describe('Gates that closed during the run, most-skipped first'), + detailOmitted: z.boolean().optional() + .describe('Set when persistence dropped `nodes`/`gates` to keep the stored row bounded — the totals are still exact. Declared so empty arrays are never mistaken for "nothing ran".'), +})); +export type FlowRunSummary = z.infer; + /** * Execution Log Schema * Full execution history for a single flow run. @@ -111,6 +212,15 @@ export const ExecutionLogSchema = lazySchema(() => z.object({ /** Step-by-step execution history */ steps: z.array(ExecutionStepLogSchema).describe('Ordered list of executed steps'), + /** + * #4354: what the run did, folded out of `steps`. Present on terminal runs; + * absent on a run written before the summary existed (or by an engine that + * does not compute one) — which is why it is optional rather than defaulted + * to zeros: an absent summary must not read as "this run did nothing". + */ + summary: FlowRunSummarySchema.optional() + .describe('Per-run rollup: records selected / acted on, gate skips, per-node status'), + /** Execution variables snapshot */ variables: z.record(z.string(), z.unknown()).optional().describe('Final state of flow variables'), @@ -275,6 +385,7 @@ export type ScheduleState = z.infer; export type ExecutionStepLogParsed = z.infer; export type ExecutionLogParsed = z.infer; +export type FlowRunSummaryParsed = z.infer; export type ExecutionErrorParsed = z.infer; export type CheckpointParsed = z.infer; export type ConcurrencyPolicyParsed = z.infer; diff --git a/packages/spec/src/contracts/automation-service.ts b/packages/spec/src/contracts/automation-service.ts index 170d1858b5..d20244601d 100644 --- a/packages/spec/src/contracts/automation-service.ts +++ b/packages/spec/src/contracts/automation-service.ts @@ -14,7 +14,7 @@ */ import type { FlowParsed } from '../automation/flow.zod'; -import type { ExecutionLog } from '../automation/execution.zod'; +import type { ExecutionLog, FlowRunSummary } from '../automation/execution.zod'; import type { ActionDescriptor } from '../automation/node-executor.zod'; import type { ConnectorDescriptor } from '../integration/connector-descriptor'; @@ -206,6 +206,17 @@ export interface AutomationResult { */ successMessage?: string; errorMessage?: string; + /** + * #4354: what the run did — records selected / acted on, gate skips, + * per-node status. Set on a TERMINAL result (a paused run has not finished + * doing it yet). + * + * On the result rather than only in the run log because a caller that + * invokes a flow synchronously — the `subflow` node rolling a child run up + * into its parent, a test asserting a sweep wrote something — needs the + * answer without a second round-trip through `getRun`. + */ + summary?: FlowRunSummary; } /** From 45c1c17cfef47d5654a8ed61f6a9a6c26e44caab Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 31 Jul 2026 10:01:44 +0000 Subject: [PATCH 2/2] fix(service-automation,docs): raw NUL in the gate key, and the strictness-ledger count (#4354) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two CI gates on the previous commit, both real: - `check:nul-bytes` — the gate-dedup key joined its parts with LITERAL NUL bytes instead of the escape sequence. Byte-identical at runtime, but a raw NUL makes ripgrep treat the whole file as binary and silently return zero matches, so `run-summary.ts` would have dropped out of code search and every grep-based lint. Written as the escape now, with the reason the joiner is NUL at all spelled out (an edge id is author-supplied text, so a printable separator could collide two distinct gates). - `check:strictness-ledger` — `automation/execution.zod.ts` gained 5 `z.object(` sites (the run-summary family), so its ledger row and the section total were stale. Re-read: they are engine-emitted telemetry that nobody authors, so the row's `wire` verdict covers them unchanged; count updated to 13 (98 for the section) with the additions named. Verified: service-automation 525/46; `pnpm lint`, `check:nul-bytes` and `check:slot-lookup` clean; all seven pure spec audits (liveness, empty-state, variant-docs, strictness-ledger, react-conformance, skill-examples, exported-any) green. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01SzJqHp1yXjri1tqKP8WoL4 --- .../2026-07-unknown-key-strictness-ledger.md | 4 ++-- .../service-automation/src/run-summary.ts | Bin 5451 -> 5693 bytes 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/audits/2026-07-unknown-key-strictness-ledger.md b/docs/audits/2026-07-unknown-key-strictness-ledger.md index c02b07efc6..2654d2afb5 100644 --- a/docs/audits/2026-07-unknown-key-strictness-ledger.md +++ b/docs/audits/2026-07-unknown-key-strictness-ledger.md @@ -165,14 +165,14 @@ tightening (the #4001 "sharing-rule lesson": candidates, not verdicts). | `external-catalog.zod.ts` | 4 | wire (p) | | | `field-value.zod.ts` / `seed.zod.ts` / `validation.zod.ts` | 1 ea | mixed (p) | | -### `automation/` — 93 sites +### `automation/` — 98 sites | File | Sites | Class | Note | |---|---|---|---| | `flow.zod.ts` | 11 | authorable | **strict as of #4001** (4 schemas; `FlowVersionHistorySchema` is runtime — stays tolerant) | | `sync.zod.ts` / `etl.zod.ts` | 12+10 | authorable (p) | authored pipelines — **candidates** | | `trigger-registry.zod.ts` | 11 | mixed | descriptors are code-registered (wire-ish); bindings authored | -| `execution.zod.ts` | 8 | wire | run-state envelopes — never strict | +| `execution.zod.ts` | 13 | wire | run-state envelopes — never strict. +5 at #4354 (the run-summary family: step metrics / skip reason / per-node / per-gate / the summary itself) — engine-emitted telemetry read by the Console and by operator queries, nobody authors them, so the `wire` verdict covers them unchanged | | `state-machine.zod.ts` | 7 | authorable (p) | | | `control-flow.zod.ts` | 6 | authorable (p) | validated structurally by `validateControlFlow` | | `bpmn-interop.zod.ts` | 5 | wire (p) | interop import shapes | diff --git a/packages/services/service-automation/src/run-summary.ts b/packages/services/service-automation/src/run-summary.ts index b3e58d8df7a121e66dd96475f0f319ba102adb77..c19f910495bb2cd2ef43abf641f9b3c21ee12fb2 100644 GIT binary patch delta 264 zcmZvWy-EW?6oos5^#SbU6heMj6Kk6^HiD!RQrO(xxg<9XGsE1wS(gxeh`fkk?GyM? z&SEE??wrHlOBPZ> delta 37 scmdn1b6RVIIM?L=T#_OTD%DAqdZ{Vtsh%kc_Vxa`4;&A8dQ0q%$j@&Et;