diff --git a/.changeset/broken-sweep-filter-not-detector.md b/.changeset/broken-sweep-filter-not-detector.md new file mode 100644 index 0000000000..2a1c24938a --- /dev/null +++ b/.changeset/broken-sweep-filter-not-detector.md @@ -0,0 +1,50 @@ +--- +"@objectstack/service-automation": patch +--- + +fix(service-automation): the documented broken-sweep predicate is a first FILTER, not the detector (#12685) + +`patch`, and not empty: `sys_automation_run`'s field descriptions are shipped, +translated, operator-facing text — they are what an admin reads in Setup while +wiring an alert they will then trust for months. No counter, no schema and no +engine behaviour changes here; the run summary measured by #4354 is correct and +untouched. + +## The wrong claim + +`acted_count` advertised `selected_count > 0 AND acted_count = 0 AND +unmeasured_count = 0` as *the* broken-sweep signal, unqualified. Measured A/B on +one graph pair through the real engine — a healthy idempotent sweep (re-select +the same records, gate each one on "was this already handled") and a dead gate +(#4347's shape, the gate sitting in front of the lookup) — **both** report +`selected > 0, acted 0, unmeasured 0`. The predicate cannot make the one +distinction it was advertised to make. + +"Over N consecutive runs" does not rescue it either: the healthy steady state +trips it on *every* run for as long as the outstanding work stands, so it is +persistent rather than transient. Consecutiveness filters flapping, which is a +different failure. + +Why a wrong sentence here is worse than a wrong sentence elsewhere: a detector +that fires during normal operation gets muted, and a muted broken-sweep detector +is the same silence #4347 produced — with the added cost that it now *looks* +monitored. + +## What the descriptions say now + +- `acted_count` states the predicate as the **first filter** and names the + discriminator: a healthy skip is accounted for by a read the run performed + (the lookup the gate depends on shows `runs > 0` and `selected > 0` in + `summary_json.nodes[]`), while a dead gate skips just as often with nothing + behind it (`runs: 0`, or `selected: 0`). +- `skipped_count` points at the same fold — `gates[]` names which edge closed + and how often, `nodes[]` says whether the lookup behind it found anything. +- `unmeasured_count` keeps its own point (why the third clause exists) and now + calls the query a filter rather than an alert. + +The discriminating data was already shipped by #4354; nothing new is measured +and no detector is implemented in the platform. `run-summary.test.ts` pins the +pair as executable evidence: both shapes match the filter, and the per-node fold +separates them. `content/docs/automation/flows.mdx` carries the same correction +with the measured table and the two authoring shapes that make a sweep's signal +quiet in its healthy steady state. diff --git a/content/docs/automation/flows.mdx b/content/docs/automation/flows.mdx index 6d7e581fdb..433afbb136 100644 --- a/content/docs/automation/flows.mdx +++ b/content/docs/automation/flows.mdx @@ -290,8 +290,8 @@ defineStack({ ``` A step that calls a declared writer is counted as an effect the platform cannot -measure (`unmeasured`), never as zero — so the broken-sweep query -`selected > 0 AND acted = 0 AND unmeasured = 0` stops firing on that flow, and +measure (`unmeasured`), never as zero — so the broken-sweep filter +`selected > 0 AND acted = 0 AND unmeasured = 0` stops matching that flow, and keeps working on every other flow that calls a function. Declaring changes what is *reported*, not what is *allowed*: an undeclared writer is still counted as having written nothing, and no runtime check can catch it — a function is @@ -785,17 +785,18 @@ child dispatched an uncountable effect knows its own `acted` is incomplete. The same counts land on `sys_automation_run` as **queryable columns** (`selected_count`, `acted_count`, `skipped_count`, `unmeasured_count`, plus a -`summary_json` breakdown), so a broken sweep is something you can alert on +`summary_json` breakdown), so a broken sweep is something you can query for rather than notice: ```typescript -// Runs that selected work and did none of it, newest first. -const suspect = await engine.find('sys_automation_run', { +// CANDIDATE runs: selected work, did none of it, and measured everything they +// did do. The first filter of the detector — not the detector; see below. +const candidates = await engine.find('sys_automation_run', { where: { status: 'completed', selected_count: { $gt: 0 }, acted_count: 0, - // Without this clause the alert fires on every healthy connector-driven + // Without this clause the filter matches every healthy connector-driven // flow: those runs report acted 0 because the count is INCOMPLETE, not zero. unmeasured_count: 0, started_at: { $gte: since }, @@ -804,13 +805,74 @@ const suspect = await engine.find('sys_automation_run', { }); ``` -`selected > 0 && acted == 0 && unmeasured == 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. +#### The filter is not the detector + + +Do not wire an alert straight to that query. It matches a **healthy idempotent +sweep** exactly as it matches a dead gate, and it does so on every run — so the +alert gets muted, and a muted broken-sweep detector is the same silence these +counters exist to end, with the added cost that it now *looks* monitored. + + +Any flow that re-selects the same records each run and gates each one on "was +this already handled" satisfies the filter in its healthy steady state, for as +long as the prior work stands — and that is the ordinary way to write an +idempotent sweep, not an exotic shape. Measured on one graph pair driven through +the engine (pinned in `run-summary.test.ts`), where the two runs differ only in +where the gate sits relative to the lookup: + +| run | totals | the filter | the lookup behind the gate | gate skips | +| :--- | :--- | :--- | :--- | :--- | +| **healthy steady state** — every stalled deal already nudged | selected 6, acted 0 | **matches** | `runs: 3`, `selected: 3` | 3 | +| **dead gate** — gate in front of the lookup, never opens | selected 3, acted 0 | **matches** | `runs: 0` | 3 | +| **wrong gate** — lookup runs, finds nothing, gate closes anyway | selected 3, acted 0 | **matches** | `runs: 3`, `selected: 0` | 3 | +| genuinely idle — nothing stalled | selected 0, acted 0 | quiet | — | 0 | + +Requiring the match **over N consecutive runs does not rescue it**: a healthy +idempotent sweep trips it on *every* run while the outstanding work stands, so +the steady state is persistent rather than transient. Consecutiveness filters +flapping, which is a different failure. + +#### What separates them: the per-node fold + +`summary_json` answers the question the totals cannot — **were the skips +accounted for?** A gate that closes is not a defect; a gate that closes on +nothing is. + +- **Healthy** — the run reached the state that justifies each skip. The lookup + the gate depends on ran (`runs` covering the gate's `skipped`) and *found* + something (`selected > 0`). Every skip has a find behind it. +- **Broken** — the same gate closed just as often with nothing behind it: the + lookup never ran (`runs: 0` — the gate sits upstream of it) or ran and found + nothing (`selected: 0`). Nothing the run measured justifies a single skip. + +```typescript +// Convict a candidate: name the node whose read your gate depends on. +const summary = JSON.parse(run.summary_json as string) as FlowRunSummary; +const skips = summary.gates.reduce((n, g) => n + g.skipped, 0); +const lookup = summary.nodes.find((n) => n.nodeId === 'find_existing_task'); +const broken = skips > 0 + && (!lookup || lookup.runs === 0 || (lookup.selected ?? 0) < skips); +``` + +`gates[]` names *which* edge closed and how often, so the alert can point at a +node instead of at a flow. `detailOmitted` marks the one case this read cannot +work with: persistence dropped `nodes` / `gates` to keep the row bounded, and +only the totals survive. + +How you author the sweep decides how good its signal can be: + +1. **Exclude the already-handled records in the query** wherever the filter can + express it. The healthy steady state then reports `selected: 0` and never + matches at all — the quietest detector available. +2. Where the "already handled?" answer lives in another object (an open task, a + sent notification), **make it a real lookup node**. The run then records the + find that accounts for the skip, which is the evidence the rule above reads. + +A gate deciding on a field of a record the run already read leaves no second +read to account for its skips; there the summary cannot separate a correct +decision from a stuck one, because the difference is in the data rather than in +the counts. Rows written before summaries existed carry `null` counts, not `0` — "not measured" must not read as "measured zero". The log line defaults to `info`; diff --git a/packages/services/service-automation/src/run-summary.test.ts b/packages/services/service-automation/src/run-summary.test.ts index 62f0979104..e78f85e9c5 100644 --- a/packages/services/service-automation/src/run-summary.test.ts +++ b/packages/services/service-automation/src/run-summary.test.ts @@ -23,6 +23,7 @@ import { registerHttpNodes } from './builtin/http-nodes.js'; import { registerConnectorNodes } from './builtin/connector-nodes.js'; import type { AutomationContext } from '@objectstack/spec/contracts'; import { defineActionDescriptor } from '@objectstack/spec/automation'; +import type { FlowRunSummary } from '@objectstack/spec/automation'; /** * `resumeAuthority: 'any'` is required of a pausing fixture since #5561: these @@ -195,13 +196,24 @@ function sweepFlow(name: string) { }; } -function sweepEngine(rows: Array>, logger = makeLogger(), store?: any) { +function sweepEngine( + rows: Array>, + logger = makeLogger(), + store?: any, + // #12685: what a per-record idempotency lookup FINDS. Absent keeps the + // long-standing answer (the first selected row); `null` is the lookup that + // ran and found nothing, which is a distinct run shape from one that never + // ran at all. Taught to the existing double on purpose — a second fake data + // engine in this file would be a new double to pin for no new capability. + opts?: { findOne?: Record | null }, +) { const written: Array> = []; let seq = 0; const data: any = { async find() { return rows; }, async findOne(object: string, query?: EngineFindOneQueryInput) { - assertEngineFindOnePredicate(object, query); return rows[0] ?? null; }, + assertEngineFindOnePredicate(object, query); + return opts && 'findOne' in opts ? opts.findOne : (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; }, @@ -302,6 +314,172 @@ describe('flow run summary — a sweep that selects rows and writes none (#4354) }); }); +// ── The run-level filter is not the detector (#12685) ─────────────────────── + +/** + * The two runs the run-level filter CANNOT tell apart, as one graph pair. + * + * `selected > 0 AND acted = 0 AND unmeasured = 0` is documented on + * `sys_automation_run.acted_count` as the FIRST FILTER for a broken sweep, and + * this pair is why the description says "filter" and not "signal": an + * idempotent sweep in its healthy steady state — re-select the same records, + * gate each one on "was this already handled" — trips it exactly as a dead gate + * does. Measured downstream through a real app's sweep first (hotcrm's + * `flow-run-summary.test.ts`); pinned here on the platform's own engine so the + * shipped sentence has an executable twin in the repo that ships it. + * + * ⚠️ Adding "over N consecutive runs" does not separate them, and no test can + * pin that by running once: the healthy steady state trips the filter on EVERY + * run for as long as the outstanding work stands, so consecutiveness filters + * flapping rather than this. What separates them is the per-node fold, which is + * what {@link skipsAreAccountedFor} reads. + * + * Both shapes share a node set, a closed gate and a skip count. They differ in + * exactly one thing — whether the run performed the read that justifies each + * skip: + * - `healthy` — the lookup runs FIRST and answers the gate. Every skip has a + * find behind it. + * - `dead_gate` — the #4347 shape: the gate sits IN FRONT of the lookup, so + * nothing behind it ever runs and no read justifies a skip. + */ +function idempotentSweepFlow(name: string, shape: 'healthy' | 'dead_gate') { + // Never opens on this fixture's rows — the gate is closed in both shapes. + const condition = { dialect: 'cel', source: 'row.shouldRun == true' }; + const lookup = { + id: 'find_existing', type: 'get_record', label: 'Already handled?', + // No `limit` ⇒ findOne ⇒ `selected: 1` when it finds one, `0` when not. + config: { objectName: 'nudge', filter: { deal: '{row.id}' }, outputVariable: 'existing' }, + }; + const gate = { id: 'gate', type: 'decision', label: 'Gate' }; + const nudge = { id: 'nudge', type: 'create_record', label: 'Nudge', config: { objectName: 'nudge', fields: { note: 'x' } } }; + const body = shape === 'healthy' + ? { + nodes: [lookup, gate, nudge], + edges: [ + { id: 'b0', source: 'find_existing', target: 'gate' }, + { id: 'b1', source: 'gate', target: 'nudge', type: 'conditional', condition, label: 'Go' }, + ], + } + : { + nodes: [gate, lookup, nudge], + edges: [ + { id: 'b1', source: 'gate', target: 'find_existing', type: 'conditional', condition, label: 'Go' }, + { id: 'b2', source: 'find_existing', target: 'nudge' }, + ], + }; + 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 } }, + { 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' }, + ], + }; +} + +/** The queryable-column filter, exactly as `acted_count`'s description states it. */ +const tripsRunLevelFilter = (s: FlowRunSummary): boolean => + s.selected > 0 && s.acted === 0 && (s.unmeasured ?? 0) === 0; + +/** + * The discriminator `acted_count`'s description sends an operator to: are the + * closed gate's skips accounted for by a read this run actually performed? + * Local to this test on purpose — the platform documents the rule and ships the + * data; it does not ship a detector. + */ +const skipsAreAccountedFor = (s: FlowRunSummary, lookupNodeId: string): boolean => { + const lookup = s.nodes.find((n) => n.nodeId === lookupNodeId); + const skips = s.gates.reduce((n, g) => n + g.skipped, 0); + return !!lookup && lookup.runs > 0 && (lookup.selected ?? 0) >= skips; +}; + +describe('the run-level filter cannot separate a healthy idempotent sweep from a dead gate (#12685)', () => { + const stalled = () => [1, 2, 3].map((i) => ({ id: `d${i}`, shouldRun: false })); + + async function run(shape: 'healthy' | 'dead_gate', opts?: { findOne?: Record | null }) { + const { engine, written } = sweepEngine(stalled(), makeLogger(), undefined, opts); + engine.registerFlow('sweep', idempotentSweepFlow('sweep', shape) as never); + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + return { summary: res.summary as FlowRunSummary, written, success: res.success }; + } + + it('FIRES on a healthy idempotent sweep — the false positive the filter cannot avoid', async () => { + const { summary, written, success } = await run('healthy', { findOne: { id: 'nudge_prior' } }); + + expect(success).toBe(true); + expect(written).toHaveLength(0); // nothing to do: every deal already nudged + // 3 deals selected by the query + one find per iteration = 6 reads, no writes. + expect(summary).toMatchObject({ selected: 6, acted: 0, skipped: 3, unmeasured: 0 }); + expect(tripsRunLevelFilter(summary)).toBe(true); + + // …and it is healthy: the lookup ran once per iteration and found the + // prior work each time. That is what accounts for the skips. + expect(summary.nodes.find((n) => n.nodeId === 'find_existing')).toMatchObject({ runs: 3, selected: 3 }); + expect(summary.gates).toEqual([ + { nodeId: 'gate', targetNodeId: 'nudge', edgeId: 'b1', label: 'Go', skipped: 3 }, + ]); + }); + + it('FIRES identically on a dead gate — same filter verdict, opposite health', async () => { + const { summary, written } = await run('dead_gate'); + + expect(written).toHaveLength(0); + // Only the query read anything — the body never ran at all. + expect(summary).toMatchObject({ selected: 3, acted: 0, skipped: 3, unmeasured: 0 }); + expect(tripsRunLevelFilter(summary)).toBe(true); + + // The gate is in front of the lookup, so nothing behind it ever ran and + // no read justifies a single skip. + expect(summary.nodes.find((n) => n.nodeId === 'find_existing')).toMatchObject({ + runs: 0, skipped: 3, status: 'skipped', + }); + expect(summary.gates[0]).toMatchObject({ nodeId: 'gate', targetNodeId: 'find_existing', skipped: 3 }); + }); + + it('`summary_json` DOES separate them — the discriminator the description documents', async () => { + const healthy = (await run('healthy', { findOne: { id: 'nudge_prior' } })).summary; + const deadGate = (await run('dead_gate')).summary; + + // Same run-level verdict… + expect([tripsRunLevelFilter(healthy), tripsRunLevelFilter(deadGate)]).toEqual([true, true]); + // …opposite per-node verdict. This pair is the whole point of the fold. + expect(skipsAreAccountedFor(healthy, 'find_existing')).toBe(true); + expect(skipsAreAccountedFor(deadGate, 'find_existing')).toBe(false); + }); + + it('catches the OTHER dead gate: the lookup ran and found nothing to justify the skips', async () => { + // The subtler #4347 shape — the gate's condition is simply wrong, so the + // lookup runs, finds nothing, and the gate closes anyway. `runs > 0` is + // therefore not enough on its own; the description says the lookup must + // have FOUND something, and this is the run that proves the clause earns + // its place. + const { summary } = await run('healthy', { findOne: null }); + + expect(summary).toMatchObject({ selected: 3, acted: 0, skipped: 3 }); + expect(tripsRunLevelFilter(summary)).toBe(true); + expect(summary.nodes.find((n) => n.nodeId === 'find_existing')).toMatchObject({ runs: 3, selected: 0 }); + expect(skipsAreAccountedFor(summary, 'find_existing')).toBe(false); + }); + + it('stays quiet on a genuinely idle sweep — nothing selected, nothing to explain', async () => { + const { engine } = sweepEngine([]); + engine.registerFlow('sweep', idempotentSweepFlow('sweep', 'healthy') as never); + const res = await engine.execute('sweep', { event: 'schedule' } as AutomationContext); + + expect(tripsRunLevelFilter(res.summary as FlowRunSummary)).toBe(false); + expect(res.summary).toMatchObject({ selected: 0, acted: 0, skipped: 0, gates: [] }); + }); +}); + // ── Durability ────────────────────────────────────────────────────────────── const flush = () => new Promise((r) => setTimeout(r, 0)); 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 56d99714db..1fec35a9bd 100644 --- a/packages/services/service-automation/src/sys-automation-run.object.ts +++ b/packages/services/service-automation/src/sys-automation-run.object.ts @@ -286,11 +286,27 @@ export const SysAutomationRun = ObjectSchema.create({ }), // ── 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. + // COLUMNS, not just a blob: `selected_count > 0 AND acted_count = 0` is the + // first FILTER of the 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. + // + // [#12685] A filter, not the detector — and the description used to say + // otherwise. Measured A/B through the real engine (pinned in + // `run-summary.test.ts`, and downstream in hotcrm's + // `flow-run-summary.test.ts`): a healthy idempotent sweep — re-select the + // same records, gate each one on "already handled" — and a dead gate BOTH + // report `selected > 0, acted 0, unmeasured 0`. "Over N consecutive runs" + // does not separate them: the healthy steady state is persistent for as + // long as the outstanding work stands, so it trips on EVERY run; + // consecutiveness filters flapping, which is a different failure. The + // discrimination lives in the per-node fold (`summary_json.nodes[]` / + // `gates[]`) and is spelled out in `acted_count`'s description, which is + // what an operator wiring an alert actually reads. Worth the words because + // the failure mode is silent: a detector that fires during normal operation + // gets muted, and a muted broken-sweep detector is the same silence #4347 + // produced — except it now looks monitored. selected_count: Field.number({ label: 'Records Selected', required: false, @@ -301,21 +317,21 @@ export const SysAutomationRun = ObjectSchema.create({ 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.', + description: 'Records this run created / updated / deleted, plus effects dispatched (notifications delivered). `selected_count > 0 AND acted_count = 0 AND unmeasured_count = 0` is the FIRST FILTER for a broken sweep, not a verdict: a healthy idempotent sweep that re-selects the same records and gates each one on "already handled" satisfies it on every run while that work stands, so "over N consecutive runs" does not separate the two. `summary_json` does: a healthy skip is accounted for by a read this run performed — the lookup the gate depends on shows `runs > 0` and `selected > 0` in `nodes[]` — while a dead gate skips just as often with nothing behind it (`runs: 0`, or `selected: 0`).', 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.', + 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; `summary_json` is what convicts or clears it — `gates[]` names which edge closed and how often, and `nodes[]` says whether the lookup that gate depends on ran and found anything (see `acted_count`).', group: 'Outcome', }), unmeasured_count: Field.number({ label: 'Uncountable Effects', required: false, - description: 'Executions that reached something the platform cannot count (a `connector_action`, a mutating `http` call whose response was lost). The qualifier `acted_count` needs to be trusted: the broken-sweep alert is `selected_count > 0 AND acted_count = 0 AND unmeasured_count = 0`, because a run with uncountable effects has an INCOMPLETE acted count, not a zero one. Null on rows written before this was tracked.', + description: 'Executions that reached something the platform cannot count (a `connector_action`, a mutating `http` call whose response was lost). The qualifier `acted_count` needs to be trusted: the broken-sweep filter is `selected_count > 0 AND acted_count = 0 AND unmeasured_count = 0` (a filter, not a verdict — see `acted_count`), because a run with uncountable effects has an INCOMPLETE acted count, not a zero one. Null on rows written before this was tracked.', group: 'Outcome', }),