From 5659aafdc69031f347c8ce0e189256d1c546cb4b Mon Sep 17 00:00:00 2001 From: breis Date: Sat, 18 Jul 2026 06:24:45 -0400 Subject: [PATCH 1/3] Add stateful multi-signal inputs + explicit output to the script stage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The script stage's object form now takes named `inputs` (selectors over signal id/name, topic filter, and envelope identity) and an optional `output`. The new MultiScriptStage caches the latest value/quality/ timestamps per input, partitioned by source device, and evaluates the script when a matched input's value or quality changes — binding the full snapshot as `inputs` and the firing input as `trigger` in both engines — gated until every required input is initialized. With `output.topic`, each successful result is published as a new envelope (producer = the processor with instance = route id, correlation_id = the triggering message's uuid) instead of mutating the trigger in place; without it the in-place contract is unchanged, as are both legacy script forms. Startup validation rejects reserved-class output topics, output topics the route's own subscribe filters would re-consume, unknown selector fields, ambiguous duplicate selectors, and a route publish.topic alongside a stage output topic. Docs: multi-signal section in the scripting guide, selector/output reference tables, ScriptResult envelope reference, an OEE sample route, and a how-to. Closes #5 Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_019emrYEw39h1jJbsA6C8oBZ --- docs/README.md | 2 +- docs/explanation.md | 21 +- docs/how-to-guides.md | 46 ++ docs/reference/configuration.md | 44 +- docs/reference/messaging-interface.md | 21 + docs/sample-configurations.md | 72 +++ docs/scripting.mdx | 123 ++++- src/app.rs | 116 ++++- src/config.rs | 156 ++++++- src/proc/mod.rs | 23 +- src/proc/multi.rs | 633 ++++++++++++++++++++++++++ src/proc/script.rs | 62 ++- 12 files changed, 1279 insertions(+), 40 deletions(-) create mode 100644 src/proc/multi.rs diff --git a/docs/README.md b/docs/README.md index f0bdc97..6cd2d0e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -13,7 +13,7 @@ component, a standalone process, or a Kubernetes pod. It appears on the bus as | Doc | Start here when you want to… | |-----|------------------------------| | **[Tutorial](tutorial.md)** | learn by doing — bring the processor up against a local broker and watch it downsample and archive telemetry, end to end | -| **[How-to guides](how-to-guides.md)** | accomplish a specific task — filter, downsample, window-aggregate, handle array signals, script, archive to Parquet, forward alarms northbound, deploy | +| **[How-to guides](how-to-guides.md)** | accomplish a specific task — filter, downsample, window-aggregate, handle array signals, script, derive a multi-signal KPI, archive to Parquet, forward alarms northbound, deploy | | **[Scripting](scripting.mdx)** | write `filter`/`script` logic in **Rhai or Lua** (runtime-selectable) — engine selection, the shared scope, return semantics, sandbox/budget, and a cookbook shown in both engines | | **[Reference](reference/)** | look up an exact option, topic, payload, or column type | | **[Explanation](explanation.md)** | understand how it works and why — the route/worker model, the processing-and-timing pipeline, targets and the file sink | diff --git a/docs/explanation.md b/docs/explanation.md index 44f5f80..6c57afc 100644 --- a/docs/explanation.md +++ b/docs/explanation.md @@ -75,7 +75,7 @@ the later stages' `process`. | `sample` | Per-key downsampling: keep one message per `everyMs` time window, or one in every `everyN`. The key path is `by`, falling back to the route key (`body.signal.id`). | yes (per key) | | `aggregate` | Tumbling-window reduction. The window is time (`"10s"`, `"500ms"`) or a bare count (`"100"`); state is keyed by `by`/route key; the folded value is `value` (default `body.samples[].value`); reducers are `avg` `max` `min` `sum` `count` `first` `last`. Emits one `ProcessedTelemetry` message per `(key, window)` when the window closes. | yes (per key) | | `project` | Reshape the body: `keep` a whitelist of **top-level** body keys (the first segment of each dotted path — so `keep: ["signal.id"]` retains the whole `signal` object), and/or `set` literal fields onto the body. | no | -| `script` | A Rhai or Lua program (inline or from a file) that returns a new body map, or a "nothing" value to drop the message. Its scope exposes `topic`, `header`, `body`, `tags`, `identity` (the source publisher's UNS identity), `samples`, and the conveniences `value`/`quality`. See [Scripting](#scripting). | no | +| `script` | A Rhai or Lua program (inline or from a file) that returns a new body map, or a "nothing" value to drop the message. Its scope exposes `topic`, `header`, `body`, `tags`, `identity` (the source publisher's UNS identity), `samples`, and the conveniences `value`/`quality`. The **multi-signal form** declares named `inputs` (cached latest values across independent signals, evaluated on change with an `inputs`/`trigger` scope) and an optional `output` topic that publishes each result as a new envelope. See [Scripting](#scripting). | no (multi-signal form: yes — per-device input cache) | Rhai is **always compiled in** — there is no feature gate, and the runtime cost is negligible when no route uses a script. One engine is shared by every `filter`/`script` stage, bounded to a million @@ -93,21 +93,26 @@ with `scriptEngine`: **[Rhai](https://rhai.rs)** (pure-Rust, always compiled in, is compiled **once at startup**, sandboxed, and bounded, so it can shape data but can't reach outside the pipeline. -Scripting appears in **two roles**, both backed by the same scope: +Scripting appears in **three roles**, all backed by the same scope: - a **`filter` `script`** — a predicate; a truthy result keeps the message (it fails *closed* — an error drops). - a **`script` stage** — a transform that returns the **new body**, or a "nothing" value (`()` in Rhai, `nil` in Lua) to **drop** the message. +- a **multi-signal `script` stage** — the same transform contract computed over **several + independent signals**: the stage caches the latest value of each named input and re-runs the + script whenever one changes, binding the snapshot as `inputs` and the firing input as `trigger`; + with an `output` topic the result becomes a **new derived signal** rather than an edit of the + triggering message. A script sees the **message view** (`topic`, the `header`/`body`/`tags` maps, the source publisher's `identity`, `samples`, and the first-sample conveniences `value`/`quality`) plus the **runtime context** (`thingName`, `componentName`, `componentFullName`, `routeId`, `recvMs`) so a generic, reusable script can branch on which component/route/thing it runs in, or on the source device/adapter -(`identity.device` / `identity.component`). Scripts are **stateless** — -each evaluation sees only the current -message; cross-message state belongs in `sample`/`aggregate`. Array-valued fields arrive as native -arrays, so a script can iterate/reduce over them like any collection. +(`identity.device` / `identity.component`). A script itself holds **no state** — each evaluation is +a pure function of its bindings; cross-message state belongs to the stages (`sample`, `aggregate`, +and the multi-signal `script` stage's input cache). Array-valued fields arrive as native arrays, so +a script can iterate/reduce over them like any collection. > **Scripting has its own guide.** The dedicated **[Scripting page](scripting.mdx)** is the full > treatment: every scope binding, return and error semantics, a Rhai language primer (functions, @@ -225,7 +230,9 @@ can be declared column-by-column. A payload that is *not* southbound-shaped is n the defaults, the aggregate stage folds the whole body as one value. `filter`, `sample`, `project`, and `script` preserve the envelope (filter passes it untouched; project -and script rewrite the body). The `aggregate` stage is the one that **emits a new message shape**, +and script rewrite the body) — except a multi-signal `script` with an `output`, which mints a fresh +`ScriptResult` envelope produced by the processor itself. The `aggregate` stage is the other one that +**emits a new message shape**, `ProcessedTelemetry`: it reuses the first message of the window as the base (so the envelope `tags` and the source `signal` carry through) and rewrites the body to diff --git a/docs/how-to-guides.md b/docs/how-to-guides.md index ff9ec51..314c6f9 100644 --- a/docs/how-to-guides.md +++ b/docs/how-to-guides.md @@ -201,6 +201,52 @@ if celsius == () { return (); } // no reading → drop --- + +## Compute a derived signal from several inputs (multi-signal script) + +**Goal:** calculate a value whose operands arrive as **independent signals** — a KPI like OEE, a +ratio of two counters, an interlock across states — and publish it as a **new signal** on its own +topic. + +Give a `script` stage named `inputs` (one selector per operand) and an `output.topic`. The stage +caches the latest value of every input, runs the script whenever one of them **changes**, and +publishes each result as a fresh envelope: + +```jsonc +"instances": [ + { "id": "fill-ratio", "subscribe": ["ecv1/gw-fill-01/opcua-adapter/+/data/#"], + "pipeline": [ + { "script": { + "source": "return { ratio = inputs.good.value / inputs.total.value, by = trigger.name }", + "inputs": { + "good": { "device": "gw-fill-01", "signalId": "GoodBottleCount" }, + "total": { "device": "gw-fill-01", "signalId": "TotalBottleCount" } + }, + "output": { "topic": "ecv1/gw-fill-01/telemetry-processor/fill-ratio/data/current" } + } } + ], + "scriptEngine": "lua", + "target": "local" } +] +``` + +- The script sees `inputs..value` / `.quality` / `.timestamp` / `.recvMs` / `.topic` for every + operand, and `trigger` for the one whose change fired the evaluation. It does not run until every + `required` input (the default) has been observed. +- Input state is isolated **per source device**, so two lines with the same signal ids never mix. +- The published result is a new envelope — producer = the processor (instance = the route id), + `correlation_id` = the triggering message's `uuid`. The output topic must not fall under the + route's own `subscribe` filters (startup error: feedback loop). +- Repeated identical values don't re-evaluate; a quality flip does, and the script decides + (`if inputs.total.quality ~= "GOOD" then return nil end` holds the last output). + +See [Scripting — multi-signal inputs](scripting.mdx#multi-signal-inputs) for the full semantics, the +[configuration reference](reference/configuration.md#script-inputs) for every selector field, and +[sample §12](sample-configurations.md#12-multi-signal-oee-from-named-inputs-lua) for a complete OEE +route. + +--- + ## Aggregate a non-southbound payload diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 8b224b2..26a564c 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -140,12 +140,52 @@ follows the engine. The source is given **inline** or from an **external file**: |------|---------| | `{"script": ""}` | Inline source. Good for a one-liner. | | `{"script": {"file": "rules/x.rhai"}}` | Read the program from a `.rhai`/`.lua` file at startup. The path resolves against [`global.defaults.scriptsDir`](#componentglobaldefaults) when relative, or is used as-is when absolute. Use this for anything beyond a one-liner — see [Use an external script file](../how-to-guides.md#use-an-external-script-file). | +| `{"script": {"file"\|"source": …, "inputs": {…}, "output": {…}}}` | The **multi-signal form**: named stateful inputs and/or an explicit output topic — see below. `source` is the object-form spelling of an inline script; exactly one of `file`/`source` is required. | -Both forms are **compiled once at startup** (a bad path or a compile error fails fast, before any +All forms are **compiled once at startup** (a bad path or a compile error fails fast, before any message flows), sandboxed, and bounded (1,000,000 ops/eval) so a runaway script cannot wedge a worker. For the full scripting model — engine selection, scope, state, return values, both languages, and a cookbook of worked examples in **both engines** — see the dedicated **[Scripting guide](../scripting.mdx)**. +**`inputs` — named multi-signal inputs.** A map of input name → +**selector**. The stage caches the latest observation of every input and evaluates the script when a +matched input's **value or quality changes**, binding the full snapshot as +[`inputs` and the firing input as `trigger`](../scripting.mdx#multi-signal-inputs). The script does +not run until every required input has been observed. Each selector needs at least one of +`signalId`/`signalName`/`topic`; unknown selector fields fail the route at build time, as do two +inputs with identical selectors. + +| Selector field | Type | Meaning | +|----------------|------|---------| +| `signalId` | string | Match `body.signal.id`. | +| `signalName` | string | Match `body.signal.name`. | +| `topic` | string (MQTT filter) | Match the arriving topic (`+`/`#` wildcards supported). The way to select identity-less (non-EdgeCommons) publishers. | +| `device` | string | Match the source envelope identity's device. Identity-based fields never match a message without an envelope identity. | +| `component` | string | Match the source envelope identity's component token. | +| `instance` | string | Match the source envelope identity's instance token. | +| `required` | boolean (default `true`) | Whether the script waits for this input before its first evaluation. An optional input is absent from the `inputs` snapshot until it arrives. | + +Cached input state is **partitioned by the source device** (the envelope identity), so two devices +publishing the same signal ids never mix into one snapshot. State is in-memory and empty at startup: +the stage re-awaits every required input before the first evaluation. A message that matches no +input is consumed by the stage. + +**`output` — an explicit output topic.** Without `output`, the script result replaces the triggering +message's body in place (the classic behavior). With `output`, each successful evaluation is +published as a **new** EdgeCommons envelope and the triggering message is consumed, never +republished: + +| Field | Type | Default | Meaning | +|-------|------|---------|---------| +| `topic` | string (template) | — (required) | The output topic. Template-resolved at startup. A reserved UNS class (`state`/`metric`/`cfg`/`log`) or a topic matching one of the route's own `subscribe` filters (a feedback loop) fails the route at build time, as does combining `output.topic` with a route-level `publish.topic`. | +| `name` | string | `ScriptResult` | The envelope header `name` of the derived message. | +| `version` | string | `1.0` | The envelope header `version` of the derived message. | + +The derived envelope's producer is the **processor itself** (identity instance = the route id), and +its header `correlation_id` carries the triggering message's `uuid` so a consumer can trace each +result back to the update that fired it. The message flows to the route's `target` like any other +stage output. + **Script scope (Rhai or Lua)** (available to both `filter` `script` and the `script` stage; identical in both engines) — the per-message **message view** plus the constant **runtime context**: @@ -164,6 +204,8 @@ the per-message **message view** plus the constant **runtime context**: | `componentFullName` | string | the fully-qualified component name (`{ComponentFullName}`) | | `routeId` | string | the id of the route running the script | | `recvMs` | integer | this message's broker receive time (Unix ms) | +| `inputs` | map | **multi-signal `script` stage only** — `{name: {value, quality, timestamp, recvMs, topic}}` for every observed input; unbound/`nil` elsewhere | +| `trigger` | map | **multi-signal `script` stage only** — `{name, value, quality, timestamp, recvMs, topic}` of the input that fired this evaluation; unbound/`nil` elsewhere | > **Key paths** are dotted paths over the message: roots `body.` (the default when no known root diff --git a/docs/reference/messaging-interface.md b/docs/reference/messaging-interface.md index a49b93c..7d5e916 100644 --- a/docs/reference/messaging-interface.md +++ b/docs/reference/messaging-interface.md @@ -154,6 +154,9 @@ The output target is per route (`target`). Route outputs must land on a non-rese - Set an explicit `publish.topic` to a UNS `data`/`evt`/`app` topic template, e.g. `ecv1/{ThingName}/telemetry-processor/data/downsampled` or `ecv1/{ThingName}/telemetry-processor/evt/alarms`. Templates are resolved at startup. +- A multi-signal `script` stage with an `output.topic` carries its own topic: the derived message + reaches the target on that topic, and combining it with a route-level `publish.topic` is a + startup error (the route topic would override the stage's). - **`northbound`** publishes to IoT Core via the mqttproxy with `qos` = `atLeastOnce` (default) or `atMostOnce`. - **`stream:`** appends the EdgeCommons protobuf envelope as one record; the stream's @@ -203,6 +206,24 @@ except on a `local` target where it is restamped): > Numeric reducers (`avg`/`max`/`min`/`sum`) are emitted only when ≥1 sample in the window was > numeric; otherwise that reducer is `null`. +## Script output (`ScriptResult`) + +A multi-signal `script` stage with a configured `output` publishes one **new** envelope per +successful evaluation on `output.topic` ([configuration +reference](configuration.md#script-inputs)): + +| Envelope field | Value | +|----------------|-------| +| `header.name` / `header.version` | `output.name` / `output.version` — default `ScriptResult` / `1.0`. | +| `header.uuid` / `header.timestamp` | fresh, minted for the derived message. | +| `header.correlation_id` | the **triggering message's `uuid`** — trace a result back to the exact update that fired it. | +| `identity` | the **processor's own** identity with `instance` = the route id — the derived signal's producer is the processor, on every target (not only `local`). | +| `body` | exactly what the script returned (map/table → JSON object). | + +The triggering input message is consumed, never republished on the output topic. Without a +configured `output`, a `script` stage keeps its in-place contract: the result replaces the body of +the triggering message, which continues on the source topic. + ## Command verbs The processor subscribes its own command inbox `ecv1/{device}/telemetry-processor/cmd/#` (wired diff --git a/docs/sample-configurations.md b/docs/sample-configurations.md index d99911e..dfd807b 100644 --- a/docs/sample-configurations.md +++ b/docs/sample-configurations.md @@ -85,6 +85,7 @@ to 0 or 1, an `aggregate` accumulates and emits on window close, the rest pass 1 | `aggregate` | `{ "aggregate": { "window": "10s", "by": "body.signal.id", "fn": ["avg","max","min","sum","count","first","last"] } }` | Tumbling-window reduction per key. `window` is time (`"10s"` / `"500ms"`) or a bare record count (`"100"`). Emits one `ProcessedTelemetry` message per `(key, window)` on close (§2). | | `project` | `{ "project": { "keep": ["signal","samples"], "set": { "origin": "processor" } } }` | `keep` whitelists **top-level body keys** (the first segment of each listed path); `set` overlays literal fields onto the body. With neither, the body passes through. | | `script` | `{ "script": "#{ \"scaled\": value * 0.1 }" }` | A Rhai program that returns a new body map; `()` drops the message (§9). | +| `script` (multi-signal) | `{ "script": { "file": "oee.lua", "inputs": { … }, "output": { "topic": "…" } } }` | Stateful named inputs: cache the latest value of every selected signal, evaluate on any change with the full snapshot bound as `inputs`, and publish the result as a new envelope on `output.topic` (§12). | Scripts run in the route's `scriptEngine` — `rhai` (default, always compiled in) or `lua` (needs the `scripting-lua` build); the scope and return contract are **identical** in both engines. A @@ -910,6 +911,77 @@ What each lever does here: --- +## 12. Multi-signal OEE from named inputs (Lua) + +A derived KPI whose operands are **independent signals**: the script declares named `inputs`, the +stage caches the latest value of each, and every change to any operand refreshes the calculation. +The result is published as a **new signal** on its own UNS topic via `output.topic` — not spliced +into whichever source message happened to arrive last. See +[Scripting — multi-signal inputs](scripting.mdx#multi-signal-inputs) for the semantics and the +[configuration reference](reference/configuration.md#script-inputs) for every selector field. + +```jsonc +// config.json — component section +"component": { + "global": { "defaults": { "scriptsDir": "./scripts", "scriptEngine": "lua" } }, + "instances": [ + { + "id": "oee-filler", + "subscribe": [ "ecv1/gw-fill-01/opcua-adapter/+/data/#" ], + "pipeline": [ + { "script": { + "file": "oee.lua", + "inputs": { + "running": { "device": "gw-fill-01", "signalId": "FillerRunning" }, + "idealCycleS": { "device": "gw-fill-01", "signalId": "IdealCycleSeconds" }, + "plannedRunS": { "device": "gw-fill-01", "signalId": "PlannedRunSeconds" }, + "totalCount": { "device": "gw-fill-01", "signalId": "TotalBottleCount" }, + "goodCount": { "device": "gw-fill-01", "signalId": "GoodBottleCount" } + }, + "output": { + "topic": "ecv1/gw-fill-01/telemetry-processor/oee/data/current", + "name": "OeeSnapshot" + } + } } + ], + "target": "local" + } + ] +} +``` + +```lua +-- scripts/oee.lua +if inputs.running.value ~= true then return nil end -- line stopped → hold the last output + +local perf = (inputs.idealCycleS.value * inputs.totalCount.value) / inputs.plannedRunS.value +local qual = inputs.goodCount.value / inputs.totalCount.value + +return { + oee = perf * qual, + performance = perf, + quality = qual, + basedOn = trigger.name -- which operand refreshed this result +} +``` + +What each lever does here: + +- **`inputs` selectors** — each operand is pinned by `device` + `signalId`, so an unrelated signal + under the same subscribe filter is consumed without effect. The stage evaluates only after **all + five** inputs have been observed (they are all `required` by default) and then on every + value/quality change of any one of them. Input state is partitioned per source device. +- **`output.topic`** — every successful evaluation publishes a fresh `OeeSnapshot` envelope on the + `oee` instance's `data/current` topic, produced by the processor (identity instance = + `oee-filler`) and correlated (`correlation_id`) to the triggering update. The topic must not fall + under the route's own `subscribe` filters — here the route listens on `opcua-adapter` topics and + publishes on the processor's own, so there is no feedback loop. +- **`return nil` on a stopped line** — a `nil`/`()` result publishes nothing, so downstream + consumers keep the last computed OEE while `running` is false. A bad-quality operand can be held + the same way (`if inputs.goodCount.quality ~= "GOOD" then return nil end`). + +--- + ## Where settings resolve from (precedence) Most route settings resolve from the most specific source that provides them: diff --git a/docs/scripting.mdx b/docs/scripting.mdx index b0d183f..4fa51b0 100644 --- a/docs/scripting.mdx +++ b/docs/scripting.mdx @@ -3,7 +3,8 @@ The built-in stages (`filter`, `sample`, `aggregate`, `project`) cover the shapes most routes need, and they are the right tool when they fit. **Scripting is the escape hatch** for logic they don't express: a derived engineering unit, a reading you only care about when it moves, a vendor payload that doesn't -match the southbound shape, a decision that depends on the *relationship* between several samples. +match the southbound shape, a decision that depends on the *relationship* between several samples, or +a KPI computed from several independent signals ([multi-signal inputs](#multi-signal-inputs)). The processor embeds **two** script engines, and a route picks one at runtime: @@ -119,12 +120,119 @@ The return value is the whole contract — a script never mutates the message in envelope is preserved. Return **`()`** (Rhai) / **`nil`** (Lua) to **drop**. A result that can't convert to JSON, or a runtime error, also drops (logged). -## Scripts are stateless + +## Multi-signal inputs -Each evaluation gets a **fresh view** of only the current message — no variable survives between -messages in either engine. This keeps a script a pure function and lets one engine instance serve a -route safely. **Cross-message state lives in the built-in stages** — `sample` for rate limiting, -`aggregate` for windowed counters/min/max/avg. A common pattern is a `script` that *shapes* each message +A plain `script` stage sees one message at a time. Many industrial calculations don't work that way: +OEE, a ratio of two counters, an interlock over several states — the operands are **independent +signals**, each arriving on its own cadence, and the result should refresh whenever *any* of them +moves. The multi-signal form of the `script` stage does exactly that. The stage declares **named +inputs**, caches the latest observation of each, and runs the script with the whole snapshot +whenever a matched input's **value or quality changes**: + +```jsonc +{ "script": { + "file": "oee.lua", + "inputs": { + "running": { "device": "gw-fill-01", "signalId": "FillerRunning" }, + "idealCycleS": { "device": "gw-fill-01", "signalId": "IdealCycleSeconds" }, + "plannedRunS": { "device": "gw-fill-01", "signalId": "PlannedRunSeconds" }, + "totalCount": { "device": "gw-fill-01", "signalId": "TotalBottleCount" }, + "goodCount": { "device": "gw-fill-01", "signalId": "GoodBottleCount" } + }, + "output": { "topic": "ecv1/gw-fill-01/telemetry-processor/oee/data/current" } +} } +``` + +Each selector names a signal (`signalId`/`signalName`), a topic filter (`topic`, with `+`/`#` +wildcards — the way to bind identity-less publishers), and/or the source identity +(`device`/`component`/`instance`, matched against the message envelope). The +[configuration reference](reference/configuration.md#script-inputs) has the full selector table and +validation rules. + +Two bindings join the ordinary scope (which still describes the *triggering* message): + +| Binding | Value | +|---------|-------| +| `inputs` | `{name: {value, quality, timestamp, recvMs, topic}}` — the current snapshot of every observed input. `value`/`quality`/`timestamp` come from the signal's first sample (`timestamp` is the source timestamp; `recvMs` the broker receive time). | +| `trigger` | `{name, value, quality, timestamp, recvMs, topic}` — the input whose change fired this evaluation. | + +The semantics, precisely: + +- **Initialization gate.** The script does not run until every input marked `required` (the default) + has been observed at least once — a half-initialized OEE is never computed. An input with + `"required": false` doesn't gate; it is simply absent from `inputs` until it arrives, so the + script checks for it (`inputs.x != ()` in Rhai, `inputs.x ~= nil` in Lua… or Rhai's `"x" in inputs`). +- **Change detection.** A message that repeats an input's current value *and* quality refreshes its + timestamps but does not re-evaluate. A quality flip alone (GOOD → BAD) *is* a change — the script + decides what to do with a bad-quality operand (return the drop value to hold the last output). +- **Per-device isolation.** The input cache is partitioned by the source device of the envelope + identity: two lines publishing the same signal ids each get their own snapshot, and one device's + values can never leak into another's calculation. +- **Consumed, not forwarded.** A multi-input stage is a *sink* for its matched messages: the + triggering message is consumed (its data lives on in the snapshot), and a subscribed message that + matches no input is consumed silently. Put a multi-input `script` last in its pipeline. +- **Restart-empty.** The cache is in-memory; after a restart the stage deterministically re-awaits + its required inputs before the first evaluation (each initialization is logged). + +**The output side.** Without `output`, the result replaces the triggering message's body in place — +which couples the derived value to whichever source topic happened to move last. A derived signal +like OEE is its own signal with its own UNS topic, so the multi-signal form usually configures +`output.topic`: every successful evaluation then publishes a **new** envelope there (header `name` +`ScriptResult` by default, configurable), produced by the processor itself (identity instance = the +route id) and carrying the triggering message's `uuid` as its `correlation_id` for provenance. The +route's startup validation rejects an output topic on a reserved UNS class or one the route's own +`subscribe` filters would re-consume (a feedback loop), and the processor's self-echo guard drops +any re-consumed copy of its own output as a second line of defense. + +A working OEE script over those inputs: + + + + +```rhai +// Availability × Performance × Quality, refreshed on any operand change. +if inputs.running.value != true { return (); } // line stopped → hold the last value + +let perf = (inputs.idealCycleS.value * inputs.totalCount.value) / inputs.plannedRunS.value; +let qual = inputs.goodCount.value / inputs.totalCount.value; + +#{ + "oee": perf * qual, + "performance": perf, + "quality": qual, + "basedOn": trigger.name // which operand refreshed this result +} +``` + + + + +```lua +-- Availability × Performance × Quality, refreshed on any operand change. +if inputs.running.value ~= true then return nil end -- line stopped → hold the last value + +local perf = (inputs.idealCycleS.value * inputs.totalCount.value) / inputs.plannedRunS.value +local qual = inputs.goodCount.value / inputs.totalCount.value + +return { + oee = perf * qual, + performance = perf, + quality = qual, + basedOn = trigger.name -- which operand refreshed this result +} +``` + + + + +## Script state + +Each evaluation is a **pure function of its bindings** — no script variable survives between +messages in either engine, which is what lets one engine instance serve a route safely. +**Cross-message state lives in the stages, not the script**: `sample` for rate limiting, `aggregate` +for windowed counters/min/max/avg, and the multi-signal `script` stage's input cache for +latest-value snapshots across signals. A common pattern is a `script` that *shapes* each message feeding an `aggregate` that *accumulates*. ## Sandbox & limits @@ -484,7 +592,8 @@ first-class facts about a message, not implementation detail. ## Limits and gotchas -- **Stateless** — no memory across messages; use `sample`/`aggregate` for cross-message state. +- **No script-held state** — no script variable survives across messages; cross-message state comes + from the stages (`sample`, `aggregate`, or the [multi-signal input cache](#multi-signal-inputs)). - **Fail-closed / drop-on-error** — a filter that errors drops the message; a transform that errors or returns a non-JSON value drops it. Scripts never crash the route. - **The op budget** bounds each evaluation in both engines — deep loops over large arrays on every diff --git a/src/app.rs b/src/app.rs index 8df5ae7..2320ef7 100644 --- a/src/app.rs +++ b/src/app.rs @@ -30,7 +30,7 @@ use std::sync::Arc; use edgecommons::config::model::Config; use edgecommons::config::template::resolve; use edgecommons::messaging::message::{Message, MessageIdentity}; -use edgecommons::messaging::MessagingService; +use edgecommons::messaging::{topic_matches, MessagingService}; use edgecommons::prelude::*; use edgecommons::uns::reserved_class_of; use rhai::Engine; @@ -38,7 +38,9 @@ use serde_json::{json, Value}; use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinHandle; -use crate::config::{parse_target, GlobalDefaults, RouteConfig, ScriptEngineKind}; +use crate::config::{ + parse_target, GlobalDefaults, RouteConfig, ScriptEngineKind, ScriptStageSpec, StageConfig, +}; use crate::observe::{spawn_metric_emitter, EvtEmitter, RouteStats}; use crate::proc::route::{run_worker, Dispatcher}; use crate::proc::{now_ms, Control, Pipeline, ProcMsg}; @@ -235,7 +237,7 @@ impl ProcessorApp { config: &Config, ctx: &RouteBuildCtx<'_>, evt: &Arc, - route: RouteConfig, + mut route: RouteConfig, ) -> anyhow::Result { let _ = gg; // used only under the `streaming` feature (stream targets) let route_key = route.key.clone().unwrap_or_else(|| ctx.default_key.to_string()); @@ -247,6 +249,7 @@ impl ProcessorApp { let target = parse_target(&target_str)?; anyhow::ensure!(!route.subscribe.is_empty(), "route '{}' has no subscribe topics", route.id); + let filters: Vec = route.subscribe.iter().map(|f| resolve(config, f)).collect(); let mut publish = route.publish.clone().unwrap_or_default(); if let Some(t) = &publish.topic { @@ -265,6 +268,23 @@ impl ProcessorApp { publish.topic = Some(resolved); } + // Resolve + validate every script stage's explicit output topic: reserved classes and + // subscribe-overlap feedback loops are startup errors, and a route-level `publish.topic` + // may not silently override a stage output topic. + for sc in route.pipeline.iter_mut() { + let StageConfig::Script(ScriptStageSpec::Spec(sp)) = sc else { continue }; + let Some(out) = sp.output.as_mut() else { continue }; + let resolved = resolve(config, &out.topic); + validate_script_output_topic( + &route.id, + &resolved, + config.effective_include_root(), + &filters, + publish.topic.as_deref(), + )?; + out.topic = resolved; + } + // Restamp policy: `local` output carries the processor's own identity (instance = route id) // — loop-safety for the self-echo guard + correct provenance for the processor's product. let restamp: Option = match &target { @@ -292,6 +312,8 @@ impl ProcessorApp { component_name: ctx.component_name.to_string(), component_full_name: ctx.component_full_name.to_string(), route_id: route.id.clone(), + // The producer identity for multi-signal script output envelopes (instance = route id). + identity: config.identity().with_instance(&route.id).ok(), }); let engine_kind = route.script_engine.unwrap_or(ctx.default_script_engine); let pipeline = Pipeline::build( @@ -322,7 +344,6 @@ impl ProcessorApp { let (control_tx, control_rx) = mpsc::channel::(CONTROL_QUEUE); self.workers.push(tokio::spawn(run_worker(pipeline, rx, control_rx, dispatcher))); - let filters: Vec = route.subscribe.iter().map(|f| resolve(config, f)).collect(); for f in &filters { tracing::info!(route = %route.id, filter = %f, "route wired"); } @@ -515,3 +536,90 @@ fn set_paused(handles: &[RouteHandle], request: &Message, paused: bool) -> Value let key = if paused { "paused" } else { "resumed" }; json!({ key: affected }) } + +/// Validate a script stage's resolved `output.topic` for one route: reject reserved UNS classes, +/// reject an output that any of the route's own subscribe filters would re-consume (a direct +/// feedback loop — cross-route re-consumption is stopped at runtime by the self-echo guard), and +/// reject a route-level `publish.topic` alongside it (the dispatcher's per-route topic would +/// silently override the per-stage output topic). +fn validate_script_output_topic( + route_id: &str, + topic: &str, + include_root: bool, + filters: &[String], + route_publish_topic: Option<&str>, +) -> anyhow::Result<()> { + if let Some(cls) = reserved_class_of(topic, include_root) { + anyhow::bail!( + "route '{route_id}': script output.topic '{topic}' targets the RESERVED UNS class \ + '{}' — target a data/evt/app class instead", + cls.token() + ); + } + for f in filters { + anyhow::ensure!( + !topic_matches(f, topic), + "route '{route_id}': script output.topic '{topic}' overlaps this route's subscribe \ + filter '{f}' — a feedback loop; publish the derived signal outside the route's input \ + filters" + ); + } + anyhow::ensure!( + route_publish_topic.is_none(), + "route '{route_id}': `publish.topic` and a script stage `output.topic` are mutually \ + exclusive — the route-level topic would override the stage output; drop one of the two" + ); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn script_output_topic_validation() { + let filters = vec!["ecv1/+/+/+/data/#".to_string()]; + // A data-class topic under the subscribed fleet filter is a feedback loop. + let err = validate_script_output_topic( + "r1", + "ecv1/gw-1/telemetry-processor/r1/data/derived", + true, + &filters, + None, + ) + .unwrap_err(); + assert!(err.to_string().contains("feedback loop"), "{err}"); + + // A reserved class (`metric`) is rejected outright. + let err = validate_script_output_topic( + "r1", + "ecv1/gw-1/telemetry-processor/r1/metric/derived", + true, + &[], + None, + ) + .unwrap_err(); + assert!(err.to_string().contains("RESERVED"), "{err}"); + + // A route publish.topic alongside a stage output topic is ambiguous. + let err = validate_script_output_topic( + "r1", + "ecv1/gw-1/telemetry-processor/r1/data/derived", + true, + &[], + Some("ecv1/gw-1/telemetry-processor/r1/data/other"), + ) + .unwrap_err(); + assert!(err.to_string().contains("mutually"), "{err}"); + + // A non-overlapping data topic with no publish.topic passes. + validate_script_output_topic( + "r1", + "ecv1/gw-1/telemetry-processor/r1/data/derived", + true, + &["ecv1/+/opcua-adapter/+/data/#".to_string()], + None, + ) + .unwrap(); + } +} diff --git a/src/config.rs b/src/config.rs index fbc0674..7ee3c1b 100644 --- a/src/config.rs +++ b/src/config.rs @@ -59,6 +59,95 @@ pub enum ScriptSource { File { file: String }, } +/// The `script` stage config: the bare inline form (`{"script": ""}`), or the object form +/// (`{"script": {...}}`) carrying the source plus the optional multi-signal `inputs` map and the +/// optional explicit `output`. +#[derive(Debug, Clone, Deserialize)] +#[serde(untagged)] +pub enum ScriptStageSpec { + /// `{"script": ""}` — inline source, result replaces the message body in place. + Inline(String), + /// `{"script": {"file"|"source": …, "inputs": {...}, "output": {...}}}`. + Spec(ScriptSpec), +} + +/// The object form of a `script` stage: the script source (exactly one of `file`/`source`), plus +/// the optional stateful multi-signal `inputs` and the optional explicit `output`. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(rename_all = "camelCase", default, deny_unknown_fields)] +pub struct ScriptSpec { + /// Path to a script file (relative to `global.defaults.scriptsDir`, or absolute), read once at + /// startup. + pub file: Option, + /// Inline script source (the object-form equivalent of `{"script": ""}`). + pub source: Option, + /// Named multi-signal inputs: `{name: selector}`. When present the stage caches the latest + /// value of every input and evaluates the script on change with the full snapshot bound as + /// `inputs` (plus the triggering message as `trigger`). Name-ordered, so multi-match trigger + /// resolution is deterministic. + pub inputs: Option>, + /// Explicit output: publish each successful result as a **new** message on `output.topic` + /// instead of mutating the triggering message in place. + pub output: Option, +} + +impl ScriptSpec { + /// The script source: exactly one of `file`/`source` must be set. + pub fn script_source(&self) -> anyhow::Result { + match (&self.file, &self.source) { + (Some(_), Some(_)) => { + anyhow::bail!("script stage: `file` and `source` are mutually exclusive") + } + (Some(f), None) => Ok(ScriptSource::File { file: f.clone() }), + (None, Some(s)) => Ok(ScriptSource::Inline(s.clone())), + (None, None) => anyhow::bail!("script stage needs `file` or `source`"), + } + } + + /// Whether this spec uses the stateful multi-signal stage (`inputs` and/or `output` present). + pub fn is_multi(&self) -> bool { + self.inputs.is_some() || self.output.is_some() + } +} + +/// Selects the signal a named script input binds to. At least one of `signalId`/`signalName`/ +/// `topic` is required; `device`/`component`/`instance` narrow by the source envelope identity. +/// Unknown selector fields are a configuration error (never silently ignored). +#[derive(Debug, Clone, Default, PartialEq, Deserialize)] +#[serde(rename_all = "camelCase", default, deny_unknown_fields)] +pub struct InputSelector { + /// Match the envelope identity's device (the deepest hierarchy value). + pub device: Option, + /// Match the envelope identity's component token. + pub component: Option, + /// Match the envelope identity's instance token. + pub instance: Option, + /// Match `body.signal.id`. + pub signal_id: Option, + /// Match `body.signal.name`. + pub signal_name: Option, + /// Match the arriving topic against this MQTT-style filter (`+`/`#` supported). + pub topic: Option, + /// Whether the script waits for this input before its first evaluation. Default `true`; an + /// optional input (`false`) is simply absent from the `inputs` snapshot until it arrives. + pub required: Option, +} + +/// Explicit `script` output: the derived result is published as a new EdgeCommons envelope on +/// `topic`, produced by the processor (identity instance = route id) and correlated to the +/// triggering message. +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(rename_all = "camelCase", default, deny_unknown_fields)] +pub struct OutputSpec { + /// Output topic (template-substituted). Must not target a reserved UNS class and must not + /// overlap the route's subscribe filters. + pub topic: String, + /// Envelope header name for the derived message. Default `ScriptResult`. + pub name: Option, + /// Envelope header version for the derived message. Default `1.0`. + pub version: Option, +} + /// One route (a `component.instances[]` entry). #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase")] @@ -93,9 +182,10 @@ pub enum StageConfig { Sample(SampleSpec), Aggregate(AggregateSpec), Project(ProjectSpec), - /// A Rhai transform: `{"script": ""}` (inline) or `{"script": {"file": "rules/x.rhai"}}`. - /// The script sees `topic` + the message fields and returns a new body map, or `()` to drop. - Script(ScriptSource), + /// A script transform: `{"script": ""}` (inline), `{"script": {"file": "rules/x.rhai"}}`, + /// or the extended object form with multi-signal `inputs` and/or an explicit `output`. The + /// script sees `topic` + the message fields and returns a new body map, or `()`/`nil` to drop. + Script(ScriptStageSpec), } /// `filter` stage. Exactly one form applies, checked in order: `script` (Rhai predicate) → @@ -232,19 +322,71 @@ mod tests { #[test] fn script_stage_parses_inline_and_file_forms() { - // Inline string → ScriptSource::Inline. + // Inline string → the bare inline form. let inline: StageConfig = serde_json::from_value(json!({ "script": "body" })).unwrap(); - assert!(matches!(inline, StageConfig::Script(ScriptSource::Inline(s)) if s == "body")); - // Object `{"file": "..."}` → ScriptSource::File. + assert!(matches!(inline, StageConfig::Script(ScriptStageSpec::Inline(s)) if s == "body")); + // Object `{"file": "..."}` → the object form; the source resolves to the file. let file: StageConfig = serde_json::from_value(json!({ "script": { "file": "rules/x.rhai" } })).unwrap(); - assert!(matches!(file, StageConfig::Script(ScriptSource::File { file }) if file == "rules/x.rhai")); + let StageConfig::Script(ScriptStageSpec::Spec(spec)) = file else { panic!() }; + assert!(!spec.is_multi()); + assert!( + matches!(spec.script_source().unwrap(), ScriptSource::File { file } if file == "rules/x.rhai") + ); // A `filter` may also take a `{"file": "..."}` predicate. let f: FilterSpec = serde_json::from_value(json!({ "script": { "file": "rules/keep.rhai" } })).unwrap(); assert!(matches!(f.script, Some(ScriptSource::File { file }) if file == "rules/keep.rhai")); } + #[test] + fn script_stage_parses_multi_signal_form() { + let v = json!({ "script": { + "file": "oee.lua", + "inputs": { + "running": { "device": "gw-fill-01", "signalId": "FillerRunning" }, + "totalCount": { "topic": "ecv1/gw-fill-01/opcua-adapter/+/data/TotalBottleCount", + "required": false } + }, + "output": { "topic": "ecv1/gw-fill-01/telemetry-processor/oee/data/current" } + }}); + let s: StageConfig = serde_json::from_value(v).unwrap(); + let StageConfig::Script(ScriptStageSpec::Spec(spec)) = s else { panic!() }; + assert!(spec.is_multi()); + let inputs = spec.inputs.as_ref().unwrap(); + assert_eq!(inputs["running"].device.as_deref(), Some("gw-fill-01")); + assert_eq!(inputs["running"].signal_id.as_deref(), Some("FillerRunning")); + assert_eq!(inputs["running"].required, None); // default: required + assert_eq!(inputs["totalCount"].required, Some(false)); + assert!(inputs["totalCount"].topic.as_deref().unwrap().contains("+/data/")); + assert_eq!( + spec.output.as_ref().unwrap().topic, + "ecv1/gw-fill-01/telemetry-processor/oee/data/current" + ); + } + + #[test] + fn script_source_exactly_one_of_file_or_source() { + let both: ScriptSpec = + serde_json::from_value(json!({ "file": "a.lua", "source": "return 1" })).unwrap(); + assert!(both.script_source().is_err()); + let neither: ScriptSpec = serde_json::from_value(json!({})).unwrap(); + assert!(neither.script_source().is_err()); + let src: ScriptSpec = serde_json::from_value(json!({ "source": "return 1" })).unwrap(); + assert!(matches!(src.script_source().unwrap(), ScriptSource::Inline(s) if s == "return 1")); + } + + #[test] + fn unknown_selector_fields_are_rejected() { + // An unknown selector field must fail configuration, never silently bind the wrong signal. + let v = json!({ "script": { + "file": "x.lua", + "inputs": { "a": { "signalld": "Typo" } }, + "output": { "topic": "t" } + }}); + assert!(serde_json::from_value::(v).is_err()); + } + #[test] fn lenient_numbers_for_greengrass_doubles() { let v = json!({ "id": "r", "pipeline": [ { "sample": { "everyN": 100.0 } } ] }); diff --git a/src/proc/mod.rs b/src/proc/mod.rs index e82f9d3..5a15126 100644 --- a/src/proc/mod.rs +++ b/src/proc/mod.rs @@ -15,6 +15,7 @@ use crate::config::{ScriptEngineKind, StageConfig, Window}; pub mod aggregate; pub mod filter; +pub mod multi; pub mod project; pub mod route; pub mod sample; @@ -86,9 +87,25 @@ impl Pipeline { Box::new(aggregate::AggregateStage::build(spec, route_key)?) } StageConfig::Project(spec) => Box::new(project::ProjectStage::build(spec)), - StageConfig::Script(src) => { - Box::new(script::ScriptStage::build(src, engine_kind, engine, loader, ctx)?) - } + StageConfig::Script(spec) => match spec { + crate::config::ScriptStageSpec::Inline(s) => Box::new(script::ScriptStage::build( + &crate::config::ScriptSource::Inline(s.clone()), + engine_kind, + engine, + loader, + ctx, + )?), + crate::config::ScriptStageSpec::Spec(sp) if sp.is_multi() => Box::new( + multi::MultiScriptStage::build(sp, engine_kind, engine, loader, ctx)?, + ), + crate::config::ScriptStageSpec::Spec(sp) => Box::new(script::ScriptStage::build( + &sp.script_source()?, + engine_kind, + engine, + loader, + ctx, + )?), + }, }; built.push(stage); } diff --git a/src/proc/multi.rs b/src/proc/multi.rs new file mode 100644 index 0000000..5d67805 --- /dev/null +++ b/src/proc/multi.rs @@ -0,0 +1,633 @@ +//! # Multi-signal `script` stage — stateful named inputs + explicit output +//! +//! The extended `script` stage form: the stage declares **named inputs**, each bound to one signal +//! by a selector (`device`/`component`/`instance` against the envelope identity, `signalId`/ +//! `signalName` against `body.signal`, and/or an MQTT-style `topic` filter). The stage caches the +//! latest value/quality/timestamps of every input and evaluates the script when a matched input's +//! **value or quality changes**, binding a consistent snapshot of all inputs as `inputs` and the +//! firing input as `trigger` (alongside the ordinary per-message bindings for the triggering +//! message). The script does not run until every `required` input has been initialized. +//! +//! Cached state is **partitioned by the source device** (the envelope identity's deepest hierarchy +//! value), so two devices publishing the same signal ids can never contaminate each other's +//! snapshot. Identity-based selectors only match messages that carry an envelope identity; +//! identity-less sources are selected by explicit `topic` filters and share one partition. State is +//! in-memory and restart-empty: after a restart the stage deterministically re-awaits every +//! required input before the first evaluation. +//! +//! With an `output` configured, each successful evaluation is published as a **new** EdgeCommons +//! envelope on `output.topic`: the body is the script result, the producer identity is the +//! processor's own (instance = route id), and `correlation_id` carries the triggering message's +//! `uuid` for provenance. The triggering input message itself is consumed, never republished. +//! Without an `output`, the result replaces the triggering message's body in place (the classic +//! single-message `script` behavior). A message that matches no input is consumed silently. + +use std::collections::HashMap; +use std::sync::Arc; + +use edgecommons::messaging::message::MessageBuilder; +use edgecommons::messaging::topic_matches; +use rhai::Engine; +use serde_json::{json, Map, Value}; +use smallvec::smallvec; + +use crate::config::{InputSelector, OutputSpec, ScriptEngineKind, ScriptSpec}; +use crate::proc::script::{build_engine, MultiBindings, ScriptContext, ScriptEngine, ScriptLoader}; +use crate::proc::{Out, ProcMsg, Processor}; + +/// Default envelope header name/version for configured-output result messages. +const DEFAULT_OUTPUT_NAME: &str = "ScriptResult"; +const DEFAULT_OUTPUT_VERSION: &str = "1.0"; + +/// One configured input: its name, selector, and whether it gates the first evaluation. +struct CompiledInput { + name: String, + sel: InputSelector, + required: bool, +} + +/// The cached latest observation of one input within one partition. +#[derive(Clone)] +struct InputEntry { + value: Value, + quality: String, + /// The source timestamp (the first sample's `timestamp`), when the payload carries one. + timestamp: Option, + recv_ms: u64, + topic: String, +} + +impl InputEntry { + fn to_json(&self) -> Value { + json!({ + "value": self.value, + "quality": self.quality, + "timestamp": self.timestamp, + "recvMs": self.recv_ms, + "topic": self.topic, + }) + } +} + +/// The stateful multi-signal `script` stage (also the carrier of the explicit-output behavior for +/// an output-only spec with no `inputs`). +pub struct MultiScriptStage { + eval: Box, + /// Name-ordered (BTreeMap config order), so multi-match trigger resolution is deterministic. + inputs: Vec, + output: Option, + ctx: Arc, + /// partition (source device) → input name → latest entry. + state: HashMap>, +} + +impl MultiScriptStage { + pub fn build( + spec: &ScriptSpec, + kind: ScriptEngineKind, + engine: &Arc, + loader: &ScriptLoader, + ctx: &Arc, + ) -> anyhow::Result { + let text = loader.load(&spec.script_source()?)?; + let mut inputs: Vec = Vec::new(); + if let Some(map) = &spec.inputs { + anyhow::ensure!(!map.is_empty(), "script `inputs` must not be empty when present"); + for (name, sel) in map { + validate_selector(name, sel)?; + inputs.push(CompiledInput { + name: name.clone(), + sel: sel.clone(), + required: sel.required.unwrap_or(true), + }); + } + for i in 0..inputs.len() { + for j in (i + 1)..inputs.len() { + anyhow::ensure!( + !same_selector(&inputs[i].sel, &inputs[j].sel), + "script inputs '{}' and '{}' have identical selectors — each input must \ + select a distinct signal", + inputs[i].name, + inputs[j].name + ); + } + } + } + if let Some(out) = &spec.output { + anyhow::ensure!( + !out.topic.trim().is_empty(), + "script `output.topic` must be non-empty" + ); + } + Ok(Self { + eval: build_engine(kind, &text, engine, ctx)?, + inputs, + output: spec.output.clone(), + ctx: ctx.clone(), + state: HashMap::new(), + }) + } + + /// Wrap a successful result: a new envelope on the configured output topic, or the in-place + /// body replacement on the triggering message when no output is configured. + fn finish(&self, body: Value, mut m: ProcMsg) -> Out { + match &self.output { + Some(out) => { + let mut b = MessageBuilder::new( + out.name.as_deref().unwrap_or(DEFAULT_OUTPUT_NAME), + out.version.as_deref().unwrap_or(DEFAULT_OUTPUT_VERSION), + ) + .payload(body) + // Provenance: correlate the derived output to the triggering message. + .correlation_id(m.msg.header.uuid.clone()); + if let Some(id) = &self.ctx.identity { + b = b.identity(id.clone()); + } + smallvec![ProcMsg { topic: out.topic.clone(), msg: b.build(), recv_ms: m.recv_ms }] + } + None => { + m.msg.body = body; + smallvec![m] + } + } + } +} + +impl Processor for MultiScriptStage { + fn process(&mut self, m: ProcMsg) -> Out { + // Output-only form (no `inputs`): evaluate every message, wrap the result. + if self.inputs.is_empty() { + return match self.eval.eval_body(&m) { + Some(body) => self.finish(body, m), + None => smallvec![], + }; + } + + // 1. Which inputs does this message bind? None → consumed, contributes nothing. + let matched: Vec = self + .inputs + .iter() + .enumerate() + .filter(|(_, ci)| selector_matches(&ci.sel, &m)) + .map(|(i, _)| i) + .collect(); + if matched.is_empty() { + return smallvec![]; + } + + // 2. Update the cached entries; only a changed value/quality (or first init) fires. + let entry = extract_entry(&m); + let part = self.state.entry(partition_key(&m)).or_default(); + let mut changed = false; + for &i in &matched { + let name = &self.inputs[i].name; + if let Some(prev) = part.get_mut(name) { + if prev.value == entry.value && prev.quality == entry.quality { + // Unchanged: refresh the timestamps, do not re-evaluate. + prev.timestamp = entry.timestamp.clone(); + prev.recv_ms = entry.recv_ms; + prev.topic = entry.topic.clone(); + } else { + *prev = entry.clone(); + changed = true; + } + } else { + tracing::debug!(input = %name, "multi-signal input initialized"); + part.insert(name.clone(), entry.clone()); + changed = true; + } + } + if !changed { + return smallvec![]; + } + + // 3. Gate: every required input must be initialized before the first evaluation. + let missing: Vec<&str> = self + .inputs + .iter() + .filter(|ci| ci.required && !part.contains_key(&ci.name)) + .map(|ci| ci.name.as_str()) + .collect(); + if !missing.is_empty() { + tracing::debug!(missing = ?missing, "multi-signal evaluation deferred: awaiting inputs"); + return smallvec![]; + } + + // 4. Evaluate with a consistent snapshot + the trigger view. On a multi-input match the + // trigger is the first matching input in name order (deterministic). + let mut snapshot = Map::new(); + for ci in &self.inputs { + if let Some(e) = part.get(&ci.name) { + snapshot.insert(ci.name.clone(), e.to_json()); + } + } + let trigger_name = &self.inputs[matched[0]].name; + let mut trigger = entry.to_json(); + trigger["name"] = json!(trigger_name); + let bindings = MultiBindings { inputs: Value::Object(snapshot), trigger }; + match self.eval.eval_body_with(&m, Some(&bindings)) { + Some(body) => self.finish(body, m), + None => smallvec![], + } + } +} + +/// A selector must discriminate on at least one signal-identifying field. +fn validate_selector(name: &str, sel: &InputSelector) -> anyhow::Result<()> { + anyhow::ensure!( + sel.signal_id.is_some() || sel.signal_name.is_some() || sel.topic.is_some(), + "script input '{name}': selector needs at least one of `signalId`, `signalName`, `topic`" + ); + if let Some(t) = &sel.topic { + anyhow::ensure!(!t.trim().is_empty(), "script input '{name}': `topic` must be non-empty"); + } + Ok(()) +} + +/// Selector equality for the duplicate-input check (ignores `required`). +fn same_selector(a: &InputSelector, b: &InputSelector) -> bool { + InputSelector { required: None, ..a.clone() } == InputSelector { required: None, ..b.clone() } +} + +/// Does `m` bind to this selector? Identity-based fields match only against a present envelope +/// identity (an identity-less message never matches them). +fn selector_matches(sel: &InputSelector, m: &ProcMsg) -> bool { + if sel.device.is_some() || sel.component.is_some() || sel.instance.is_some() { + let Some(id) = &m.msg.identity else { return false }; + if sel.device.as_deref().is_some_and(|d| id.device() != d) { + return false; + } + if sel.component.as_deref().is_some_and(|c| id.component() != c) { + return false; + } + if sel.instance.as_deref().is_some_and(|i| id.instance() != Some(i)) { + return false; + } + } + let signal = m.msg.body.get("signal"); + if let Some(sid) = sel.signal_id.as_deref() { + if signal.and_then(|s| s.get("id")).and_then(Value::as_str) != Some(sid) { + return false; + } + } + if let Some(sname) = sel.signal_name.as_deref() { + if signal.and_then(|s| s.get("name")).and_then(Value::as_str) != Some(sname) { + return false; + } + } + if let Some(f) = sel.topic.as_deref() { + if !topic_matches(f, &m.topic) { + return false; + } + } + true +} + +/// The state-partition key: the source device from the envelope identity. Identity-less messages +/// share one partition (select them with explicit `topic` filters). +fn partition_key(m: &ProcMsg) -> String { + m.msg.identity.as_ref().map(|id| id.device().to_string()).unwrap_or_default() +} + +/// The observation carried by one message: the first sample's value/quality/timestamp +/// (`SouthboundSignalUpdate` shape), falling back to the whole body for non-sample payloads. +fn extract_entry(m: &ProcMsg) -> InputEntry { + let first = m.msg.body.get("samples").and_then(Value::as_array).and_then(|a| a.first()); + match first { + Some(s) => InputEntry { + value: s.get("value").cloned().unwrap_or(Value::Null), + quality: s.get("quality").and_then(Value::as_str).unwrap_or("").to_string(), + timestamp: s.get("timestamp").cloned(), + recv_ms: m.recv_ms, + topic: m.topic.clone(), + }, + None => InputEntry { + value: m.msg.body.clone(), + quality: String::new(), + timestamp: None, + recv_ms: m.recv_ms, + topic: m.topic.clone(), + }, + } +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use super::*; + use crate::config::ScriptStageSpec; + use crate::proc::now_ms; + use edgecommons::messaging::message::{HierEntry, MessageBuilder, MessageIdentity}; + use serde_json::json; + + fn identity(device: &str, component: &str, instance: Option<&str>) -> MessageIdentity { + MessageIdentity::new( + vec![HierEntry { level: "device".into(), value: device.into() }], + component, + instance.map(String::from), + ) + .unwrap() + } + + /// A `SouthboundSignalUpdate`-shaped message from `device` for `signal` with one sample. + fn signal_msg(device: &str, signal: &str, value: Value, quality: &str) -> ProcMsg { + let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0") + .identity(identity(device, "opcua-adapter", Some("kep1"))) + .payload(json!({ + "signal": { "id": signal, "name": signal }, + "samples": [{ "value": value, "quality": quality, "timestamp": "2026-07-18T12:00:00Z" }] + })) + .build(); + ProcMsg { + topic: format!("ecv1/{device}/opcua-adapter/kep1/data/{signal}"), + msg: m, + recv_ms: now_ms(), + } + } + + fn spec_from(v: Value) -> ScriptSpec { + match serde_json::from_value::(v).unwrap() { + ScriptStageSpec::Spec(s) => s, + _ => panic!("expected the object form"), + } + } + + fn build(spec: &ScriptSpec) -> MultiScriptStage { + try_build(spec).unwrap() + } + + fn try_build(spec: &ScriptSpec) -> anyhow::Result { + let ctx = Arc::new(ScriptContext { + identity: Some(identity("edge-proc", "telemetry-processor", Some("r1"))), + route_id: "r1".into(), + ..Default::default() + }); + MultiScriptStage::build( + spec, + ScriptEngineKind::Rhai, + &Arc::new(Engine::new()), + &ScriptLoader::default(), + &ctx, + ) + } + + /// A two-input spec (a + b) whose script returns both values plus the trigger name. + fn two_input_spec(output: bool) -> ScriptSpec { + let mut v = json!({ + "source": r#"#{ "a": inputs.a.value, "b": inputs.b.value, "by": trigger.name }"#, + "inputs": { + "a": { "device": "gw-1", "signalId": "A" }, + "b": { "device": "gw-1", "signalId": "B" } + } + }); + if output { + v["output"] = json!({ "topic": "ecv1/gw-1/telemetry-processor/r1/data/derived" }); + } + spec_from(v) + } + + #[test] + fn does_not_evaluate_until_all_required_inputs_initialized() { + let mut s = build(&two_input_spec(false)); + // Only `a` has arrived → no evaluation, out-of-order init is fine. + assert!(s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).is_empty()); + // `b` arrives → the snapshot is complete and the script runs once. + let out = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); + assert_eq!(out.len(), 1); + assert_eq!(out[0].msg.body, json!({ "a": 1, "b": 2, "by": "b" })); + } + + #[test] + fn any_input_update_reevaluates_with_the_full_snapshot() { + let mut s = build(&two_input_spec(false)); + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); + // Updating `a` re-fires with b's latest cached value; the trigger is identified. + let out = s.process(signal_msg("gw-1", "A", json!(10), "GOOD")); + assert_eq!(out.len(), 1); + assert_eq!(out[0].msg.body, json!({ "a": 10, "b": 2, "by": "a" })); + } + + #[test] + fn repeated_unchanged_values_do_not_reevaluate() { + let mut s = build(&two_input_spec(false)); + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); + // The same value + quality again → no evaluation. + assert!(s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).is_empty()); + // A quality change alone IS a change (scripts gate on bad quality themselves). + let out = s.process(signal_msg("gw-1", "A", json!(1), "BAD")); + assert_eq!(out.len(), 1); + } + + #[test] + fn quality_and_timestamps_available_to_the_script() { + let spec = spec_from(json!({ + "source": r#"#{ "q": inputs.a.quality, "ts": inputs.a.timestamp, + "recv": inputs.a.recvMs > 0, "tq": trigger.quality }"#, + "inputs": { "a": { "device": "gw-1", "signalId": "A" } } + })); + let mut s = build(&spec); + let out = s.process(signal_msg("gw-1", "A", json!(5), "BAD")); + assert_eq!(out.len(), 1); + assert_eq!( + out[0].msg.body, + json!({ "q": "BAD", "ts": "2026-07-18T12:00:00Z", "recv": true, "tq": "BAD" }) + ); + } + + #[test] + fn configured_output_publishes_a_new_envelope_and_consumes_the_trigger() { + let mut s = build(&two_input_spec(true)); + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + let trigger = signal_msg("gw-1", "B", json!(2), "GOOD"); + let trigger_uuid = trigger.msg.header.uuid.clone(); + let out = s.process(trigger); + // Exactly one message: the derived output — the trigger is not republished. + assert_eq!(out.len(), 1); + let pm = &out[0]; + assert_eq!(pm.topic, "ecv1/gw-1/telemetry-processor/r1/data/derived"); + assert_eq!(pm.msg.body, json!({ "a": 1, "b": 2, "by": "b" })); + // A valid envelope: the processor (instance = route id) is the producer… + let id = pm.msg.identity.as_ref().unwrap(); + assert_eq!(id.device(), "edge-proc"); + assert_eq!(id.component(), "telemetry-processor"); + assert_eq!(id.instance(), Some("r1")); + // …with default header name/version and trigger provenance via the correlation id. + assert_eq!(pm.msg.header.name, "ScriptResult"); + assert_eq!(pm.msg.header.version, "1.0"); + assert_eq!(pm.msg.header.correlation_id, trigger_uuid); + assert!(!pm.msg.header.uuid.is_empty() && pm.msg.header.uuid != trigger_uuid); + } + + #[test] + fn output_envelope_name_and_version_are_configurable() { + let mut spec = two_input_spec(true); + spec.output.as_mut().unwrap().name = Some("OeeSnapshot".into()); + spec.output.as_mut().unwrap().version = Some("2.0".into()); + let mut s = build(&spec); + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + let out = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); + assert_eq!(out[0].msg.header.name, "OeeSnapshot"); + assert_eq!(out[0].msg.header.version, "2.0"); + } + + #[test] + fn in_place_mode_without_output_keeps_the_trigger_message() { + let mut s = build(&two_input_spec(false)); + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + let trigger = signal_msg("gw-1", "B", json!(2), "GOOD"); + let (topic, uuid) = (trigger.topic.clone(), trigger.msg.header.uuid.clone()); + let out = s.process(trigger); + // The classic in-place contract: same message/topic, new body. + assert_eq!(out[0].topic, topic); + assert_eq!(out[0].msg.header.uuid, uuid); + assert_eq!(out[0].msg.body["a"], json!(1)); + } + + #[test] + fn state_is_partitioned_by_source_device() { + let mut spec = two_input_spec(false); + // Fleet-style selectors: by signal id only, no device pin. + spec.inputs = Some(BTreeMap::from([ + ("a".into(), InputSelector { signal_id: Some("A".into()), ..Default::default() }), + ("b".into(), InputSelector { signal_id: Some("B".into()), ..Default::default() }), + ])); + let mut s = build(&spec); + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + // gw-2 publishing the same signal ids must not complete gw-1's snapshot… + assert!(s.process(signal_msg("gw-2", "B", json!(99), "GOOD")).is_empty()); + // …and gw-1's own B completes gw-1's partition with gw-1's values only. + let out = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); + assert_eq!(out.len(), 1); + assert_eq!(out[0].msg.body, json!({ "a": 1, "b": 2, "by": "b" })); + // gw-2's partition still awaits its own A. + assert!(s.process(signal_msg("gw-2", "B", json!(100), "GOOD")).is_empty()); + } + + #[test] + fn non_matching_messages_are_consumed() { + let mut s = build(&two_input_spec(false)); + assert!(s.process(signal_msg("gw-1", "Other", json!(7), "GOOD")).is_empty()); + assert!(s.process(signal_msg("gw-9", "A", json!(7), "GOOD")).is_empty()); + } + + #[test] + fn optional_inputs_do_not_gate_evaluation() { + let spec = spec_from(json!({ + "source": r#"if "c" in inputs { #{ "c": inputs.c.value } } else { #{ "a": inputs.a.value } }"#, + "inputs": { + "a": { "device": "gw-1", "signalId": "A" }, + "c": { "device": "gw-1", "signalId": "C", "required": false } + } + })); + let mut s = build(&spec); + // The optional `c` is absent from the snapshot; the required `a` alone evaluates. + let out = s.process(signal_msg("gw-1", "A", json!(3), "GOOD")); + assert_eq!(out[0].msg.body, json!({ "a": 3 })); + // Once `c` arrives it appears in the snapshot. + let out = s.process(signal_msg("gw-1", "C", json!(4), "GOOD")); + assert_eq!(out[0].msg.body, json!({ "c": 4 })); + } + + #[test] + fn topic_filter_selectors_match_identityless_messages() { + let spec = spec_from(json!({ + "source": r#"#{ "v": inputs.raw.value }"#, + "inputs": { "raw": { "topic": "plant/+/counter" } } + })); + let mut s = build(&spec); + // A foreign message with no envelope identity, selected purely by topic filter. + let m = MessageBuilder::new("X", "1.0").payload(json!({ "n": 41 })).build(); + let pm = ProcMsg { topic: "plant/line1/counter".into(), msg: m, recv_ms: now_ms() }; + let out = s.process(pm); + // No samples array → the whole body is the value. + assert_eq!(out[0].msg.body, json!({ "v": { "n": 41 } })); + } + + #[test] + fn output_only_spec_wraps_every_result() { + let spec = spec_from(json!({ + "source": r#"#{ "doubled": value * 2 }"#, + "output": { "topic": "ecv1/edge-proc/telemetry-processor/r1/data/doubled" } + })); + let mut s = build(&spec); + let out = s.process(signal_msg("gw-1", "A", json!(21), "GOOD")); + assert_eq!(out.len(), 1); + assert_eq!(out[0].topic, "ecv1/edge-proc/telemetry-processor/r1/data/doubled"); + assert_eq!(out[0].msg.body, json!({ "doubled": 42 })); + assert_eq!(out[0].msg.identity.as_ref().unwrap().instance(), Some("r1")); + } + + #[test] + fn build_rejects_bad_specs() { + // A selector without any signal-identifying field. + let s = spec_from(json!({ + "source": "1", "inputs": { "a": { "device": "gw-1" } }, + })); + assert!(try_build(&s).is_err()); + // Two inputs with identical selectors (ambiguous binding). + let s = spec_from(json!({ + "source": "1", + "inputs": { + "a": { "device": "gw-1", "signalId": "X" }, + "b": { "device": "gw-1", "signalId": "X" } + }, + })); + assert!(try_build(&s).is_err()); + // An empty inputs map. + let s = spec_from(json!({ "source": "1", "inputs": {} })); + assert!(try_build(&s).is_err()); + // An empty output topic. + let s = spec_from(json!({ "source": "1", "output": { "topic": " " } })); + assert!(try_build(&s).is_err()); + } + + #[cfg(feature = "scripting-lua")] + mod lua_tests { + use super::*; + + /// The OEE shape from the design: named inputs, a Lua calculation, an explicit output. + #[test] + fn lua_oee_calculation_over_named_inputs() { + let spec = spec_from(json!({ + "source": r#" + if inputs.running.value ~= true then return nil end + local avail = inputs.totalCount.value / inputs.plannedCount.value + return { oee = avail, by = trigger.name } + "#, + "inputs": { + "running": { "device": "gw-fill-01", "signalId": "FillerRunning" }, + "totalCount": { "device": "gw-fill-01", "signalId": "TotalBottleCount" }, + "plannedCount": { "device": "gw-fill-01", "signalId": "PlannedBottleCount" } + }, + "output": { "topic": "ecv1/gw-fill-01/telemetry-processor/oee/data/current" } + })); + let ctx = Arc::new(ScriptContext { + identity: Some(identity("edge-proc", "telemetry-processor", Some("oee"))), + route_id: "oee".into(), + ..Default::default() + }); + let mut s = MultiScriptStage::build( + &spec, + ScriptEngineKind::Lua, + &Arc::new(Engine::new()), + &ScriptLoader::default(), + &ctx, + ) + .unwrap(); + assert!(s.process(signal_msg("gw-fill-01", "FillerRunning", json!(true), "GOOD")).is_empty()); + assert!(s.process(signal_msg("gw-fill-01", "PlannedBottleCount", json!(100), "GOOD")).is_empty()); + let out = s.process(signal_msg("gw-fill-01", "TotalBottleCount", json!(80), "GOOD")); + assert_eq!(out.len(), 1); + assert_eq!(out[0].topic, "ecv1/gw-fill-01/telemetry-processor/oee/data/current"); + assert_eq!(out[0].msg.body["oee"], json!(0.8)); + assert_eq!(out[0].msg.body["by"], json!("totalCount")); + + // `running` flips false → the script returns nil → no output published. + let none = s.process(signal_msg("gw-fill-01", "FillerRunning", json!(false), "GOOD")); + assert!(none.is_empty()); + } + } +} diff --git a/src/proc/script.rs b/src/proc/script.rs index 9d8d751..f31e4f2 100644 --- a/src/proc/script.rs +++ b/src/proc/script.rs @@ -16,6 +16,7 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; use anyhow::Context; +use edgecommons::messaging::message::MessageIdentity; use rhai::{Dynamic, Engine, Scope, AST}; use serde_json::Value; use smallvec::smallvec; @@ -42,6 +43,20 @@ pub struct ScriptContext { pub component_full_name: String, /// The owning route's id — exposed as `routeId`. pub route_id: String, + /// The processor's own message identity for this route (instance = route id) — the producer + /// identity stamped on a multi-signal script stage's configured-output envelopes. Not exposed + /// as a script binding. + pub identity: Option, +} + +/// The extra bindings a multi-signal `script` stage passes into an evaluation: the `inputs` +/// snapshot map and the `trigger` view of the input that fired it. +#[derive(Debug, Clone)] +pub struct MultiBindings { + /// `{name: {value, quality, timestamp, recvMs, topic}}` for every initialized input. + pub inputs: Value, + /// `{name, value, quality, timestamp, recvMs, topic}` of the triggering input. + pub trigger: Value, } /// Resolves [`ScriptSource`]s to Rhai source text. `File` paths resolve against `base` (the @@ -82,7 +97,12 @@ pub trait ScriptEngine: Send { /// Evaluate as a `filter` predicate. Truthy → keep; error → `false` (drop), logged. Fails closed. fn eval_bool(&self, m: &ProcMsg) -> bool; /// Evaluate as a `script` transform: `Some(new_body)` or `None` to drop. Error / non-JSON → `None`. - fn eval_body(&self, m: &ProcMsg) -> Option; + fn eval_body(&self, m: &ProcMsg) -> Option { + self.eval_body_with(m, None) + } + /// [`Self::eval_body`] with the multi-signal `inputs`/`trigger` bindings additionally in scope + /// (`None` leaves them unbound / `nil`). + fn eval_body_with(&self, m: &ProcMsg, multi: Option<&MultiBindings>) -> Option; } /// Compile `src` into the engine selected by `kind`, sharing the Rhai `engine` (Rhai) or building a @@ -133,7 +153,7 @@ impl RhaiEval { Ok(Self { engine: engine.clone(), ast, ctx: ctx.clone() }) } - fn scope_for(&self, m: &ProcMsg) -> Scope<'static> { + fn scope_for(&self, m: &ProcMsg, multi: Option<&MultiBindings>) -> Scope<'static> { let mut scope = Scope::new(); scope.push("topic", m.topic.clone()); // Runtime context — constant per route, so a generic/reused script can branch on identity. @@ -164,6 +184,11 @@ impl RhaiEval { first.and_then(|s| s.get("quality")).and_then(|q| q.as_str()).unwrap_or("").to_string(); scope.push_dynamic("value", to_dyn(&value)); scope.push("quality", quality); + // Multi-signal bindings (only a multi-input stage passes them). + if let Some(mb) = multi { + scope.push_dynamic("inputs", to_dyn(&mb.inputs)); + scope.push_dynamic("trigger", to_dyn(&mb.trigger)); + } scope } @@ -172,7 +197,7 @@ impl RhaiEval { impl ScriptEngine for RhaiEval { /// Errors → `false` (drop), logged. fn eval_bool(&self, m: &ProcMsg) -> bool { - let mut scope = self.scope_for(m); + let mut scope = self.scope_for(m, None); match self.engine.eval_ast_with_scope::(&mut scope, &self.ast) { Ok(d) => d.as_bool().unwrap_or(false), Err(e) => { @@ -183,8 +208,8 @@ impl ScriptEngine for RhaiEval { } /// `()` → drop; non-convertible/error → drop, logged. - fn eval_body(&self, m: &ProcMsg) -> Option { - let mut scope = self.scope_for(m); + fn eval_body_with(&self, m: &ProcMsg, multi: Option<&MultiBindings>) -> Option { + let mut scope = self.scope_for(m, multi); match self.engine.eval_ast_with_scope::(&mut scope, &self.ast) { Ok(d) if d.is_unit() => None, Ok(d) => match rhai::serde::from_dynamic::(&d) { @@ -248,7 +273,7 @@ mod lua { use mlua::{HookTriggers, Lua, LuaOptions, LuaSerdeExt, StdLib, Value as LuaValue, VmState}; use serde_json::Value; - use super::{ProcMsg, ScriptContext, ScriptEngine}; + use super::{MultiBindings, ProcMsg, ScriptContext, ScriptEngine}; /// Per-evaluation instruction budget, mirroring Rhai's `max_operations`. const OP_BUDGET: i64 = 1_000_000; @@ -304,7 +329,7 @@ mod lua { } /// Marshal the per-message data into globals + reset the op budget. - fn bind(&self, m: &ProcMsg) { + fn bind(&self, m: &ProcMsg, multi: Option<&MultiBindings>) { self.budget.store(OP_BUDGET, Ordering::Relaxed); let g = self.lua.globals(); let _ = g.set("topic", m.topic.clone()); @@ -338,12 +363,28 @@ mod lua { let _ = g.set("value", v); } let _ = g.set("quality", quality); + // Multi-signal bindings — always (re)set, so a stale `inputs`/`trigger` from a prior + // evaluation can never leak into a later one (globals persist across calls). + match multi { + Some(mb) => { + if let Ok(v) = self.lua.to_value(&mb.inputs) { + let _ = g.set("inputs", v); + } + if let Ok(v) = self.lua.to_value(&mb.trigger) { + let _ = g.set("trigger", v); + } + } + None => { + let _ = g.set("inputs", LuaValue::Nil); + let _ = g.set("trigger", LuaValue::Nil); + } + } } } impl ScriptEngine for LuaEngine { fn eval_bool(&self, m: &ProcMsg) -> bool { - self.bind(m); + self.bind(m, None); match self.func.call::(()) { // Lua truthiness: only `nil` and `false` drop; everything else keeps. Ok(LuaValue::Nil) | Ok(LuaValue::Boolean(false)) => false, @@ -355,8 +396,8 @@ mod lua { } } - fn eval_body(&self, m: &ProcMsg) -> Option { - self.bind(m); + fn eval_body_with(&self, m: &ProcMsg, multi: Option<&MultiBindings>) -> Option { + self.bind(m, multi); match self.func.call::(()) { Ok(LuaValue::Nil) => None, Ok(v) => match self.lua.from_value::(v) { @@ -490,6 +531,7 @@ mod tests { component_name: comp.into(), component_full_name: full.into(), route_id: route.into(), + identity: None, }) } From 84e47db8f7093a5d8f11b7f746a393f466047be5 Mon Sep 17 00:00:00 2001 From: breis Date: Sat, 18 Jul 2026 11:42:27 -0400 Subject: [PATCH 2/3] Add tests isolating stage-level gating/suppression from script behavior MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The existing gating/unchanged tests used scripts that read their inputs, so a missing or unchanged input could yield no output either because the stage withheld the invocation OR because the script errored/returned nothing — the two layers weren't distinguished. Add two tests driven by a constant-returning script (`#{ "fired": true }`) that never inspects its inputs, so it cannot be the layer deciding to emit. Output therefore appears iff the stage invokes the script, proving the stage itself gates until all required inputs are initialized and suppresses invocation when no value/quality changed. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_019emrYEw39h1jJbsA6C8oBZ --- src/proc/multi.rs | 45 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/src/proc/multi.rs b/src/proc/multi.rs index 5d67805..47e2dc1 100644 --- a/src/proc/multi.rs +++ b/src/proc/multi.rs @@ -421,6 +421,51 @@ mod tests { assert_eq!(out.len(), 1); } + // A script that ignores its inputs and always returns a constant. It CANNOT be the thing + // deciding whether to emit — so if the stage ever invokes it, output appears. These two tests + // therefore isolate the *stage's* gating/change-detection from any fail-closed script + // behavior (a real script that errors or returns nil on a missing/unchanged input would also + // yield no output, which would not distinguish the two layers). + fn constant_two_input_spec() -> ScriptSpec { + spec_from(json!({ + "source": r#"#{ "fired": true }"#, + "inputs": { + "a": { "device": "gw-1", "signalId": "A" }, + "b": { "device": "gw-1", "signalId": "B" } + } + })) + } + + #[test] + fn stage_gates_invocation_the_script_is_not_called_before_all_inputs() { + let mut s = build(&constant_two_input_spec()); + // Only `a` present. A constant script WOULD emit `{fired:true}` if the stage called it — + // it doesn't, proving the stage withholds the invocation itself. + assert!( + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).is_empty(), + "the stage must not invoke the script before all required inputs are initialized" + ); + // `b` completes the set → now the stage invokes it and the constant is emitted. + let out = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); + assert_eq!(out.len(), 1); + assert_eq!(out[0].msg.body, json!({ "fired": true })); + } + + #[test] + fn stage_suppresses_invocation_on_an_unchanged_value() { + let mut s = build(&constant_two_input_spec()); + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + let first = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); + assert_eq!(first.len(), 1, "fires once both inputs are present"); + // Re-send `a` with the same value + quality. A constant script WOULD emit again if the + // stage called it — it doesn't, proving the stage's change detection suppresses the + // invocation (not the script choosing to return nothing). + assert!( + s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).is_empty(), + "the stage must not invoke the script when no input value/quality changed" + ); + } + #[test] fn quality_and_timestamps_available_to_the_script() { let spec = spec_from(json!({ From 8a3031befb5c849546da380bddc0d53dba4af801 Mon Sep 17 00:00:00 2001 From: breis Date: Sat, 18 Jul 2026 12:00:05 -0400 Subject: [PATCH 3/3] Make multi-signal gating the script's responsibility (default off) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per explicit design direction: the core stage must not withhold script invocation for incomplete inputs — completeness is the script writer's call. Flip the `required` default from true to false, so by default the stage runs the script on the first (and every) matched change with whatever inputs are currently present; an unarrived input is simply absent from the `inputs` snapshot and the script guards itself. Keep `required: true` as a per-input opt-in for stage-level gating (the "need them all" convenience), and gate only on those inputs. Tests: replace the gating tests with ones that isolate the mechanism via a constant script — default-off fires on the first input (no core gate), `required:true` withholds until all such inputs arrive. OEE/ratio example scripts now guard presence themselves. Docs: scripting guide, configuration reference, how-to, sample, and explanation updated — completeness is the script's job by default, `required:true` opts into stage gating. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_019emrYEw39h1jJbsA6C8oBZ --- docs/explanation.md | 7 ++- docs/how-to-guides.md | 14 +++-- docs/reference/configuration.md | 20 +++--- docs/sample-configurations.md | 12 ++-- docs/scripting.mdx | 28 ++++++--- src/config.rs | 6 +- src/proc/multi.rs | 108 +++++++++++++++++++++----------- 7 files changed, 128 insertions(+), 67 deletions(-) diff --git a/docs/explanation.md b/docs/explanation.md index 6c57afc..5f022a3 100644 --- a/docs/explanation.md +++ b/docs/explanation.md @@ -101,9 +101,10 @@ Scripting appears in **three roles**, all backed by the same scope: Rhai, `nil` in Lua) to **drop** the message. - a **multi-signal `script` stage** — the same transform contract computed over **several independent signals**: the stage caches the latest value of each named input and re-runs the - script whenever one changes, binding the snapshot as `inputs` and the firing input as `trigger`; - with an `output` topic the result becomes a **new derived signal** rather than an edit of the - triggering message. + script whenever one changes, binding the snapshot as `inputs` and the firing input as `trigger`. + The stage does not gate on missing inputs by default — the script owns completeness (an input + opts into stage-level waiting with `required: true`). With an `output` topic the result becomes a + **new derived signal** rather than an edit of the triggering message. A script sees the **message view** (`topic`, the `header`/`body`/`tags` maps, the source publisher's `identity`, `samples`, and the first-sample conveniences `value`/`quality`) plus the **runtime diff --git a/docs/how-to-guides.md b/docs/how-to-guides.md index 314c6f9..1432c01 100644 --- a/docs/how-to-guides.md +++ b/docs/how-to-guides.md @@ -210,7 +210,8 @@ topic. Give a `script` stage named `inputs` (one selector per operand) and an `output.topic`. The stage caches the latest value of every input, runs the script whenever one of them **changes**, and -publishes each result as a fresh envelope: +publishes each result as a fresh envelope. Here both operands are marked `"required": true`, so the +stage waits until both exist before running — which lets the script stay a clean one-liner: ```jsonc "instances": [ @@ -219,8 +220,8 @@ publishes each result as a fresh envelope: { "script": { "source": "return { ratio = inputs.good.value / inputs.total.value, by = trigger.name }", "inputs": { - "good": { "device": "gw-fill-01", "signalId": "GoodBottleCount" }, - "total": { "device": "gw-fill-01", "signalId": "TotalBottleCount" } + "good": { "device": "gw-fill-01", "signalId": "GoodBottleCount", "required": true }, + "total": { "device": "gw-fill-01", "signalId": "TotalBottleCount", "required": true } }, "output": { "topic": "ecv1/gw-fill-01/telemetry-processor/fill-ratio/data/current" } } } @@ -231,8 +232,11 @@ publishes each result as a fresh envelope: ``` - The script sees `inputs..value` / `.quality` / `.timestamp` / `.recvMs` / `.topic` for every - operand, and `trigger` for the one whose change fired the evaluation. It does not run until every - `required` input (the default) has been observed. + operand, and `trigger` for the one whose change fired the evaluation. +- **Completeness is the script's call.** By default the stage does not wait for missing inputs — it + runs the script on the first change and the script guards itself (`if inputs.total == nil then + return nil end`). Marking an input `"required": true` (as above) opts into stage-level waiting so + the script doesn't have to check; with no `required` input the stage never waits. - Input state is isolated **per source device**, so two lines with the same signal ids never mix. - The published result is a new envelope — producer = the processor (instance = the route id), `correlation_id` = the triggering message's `uuid`. The output topic must not fall under the diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 26a564c..d9684dc 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -149,11 +149,13 @@ cookbook of worked examples in **both engines** — see the dedicated **[Scripti **`inputs` — named multi-signal inputs.** A map of input name → **selector**. The stage caches the latest observation of every input and evaluates the script when a -matched input's **value or quality changes**, binding the full snapshot as -[`inputs` and the firing input as `trigger`](../scripting.mdx#multi-signal-inputs). The script does -not run until every required input has been observed. Each selector needs at least one of -`signalId`/`signalName`/`topic`; unknown selector fields fail the route at build time, as do two -inputs with identical selectors. +matched input's **value or quality changes**, binding the current snapshot as +[`inputs` and the firing input as `trigger`](../scripting.mdx#multi-signal-inputs). **By default the +stage does not gate on missing inputs** — it runs the script on the first (and every) matched change, +an unarrived input is simply absent from the snapshot, and the script decides whether it has enough +to compute (see [completeness](../scripting.mdx#multi-signal-inputs)). Each selector needs at least +one of `signalId`/`signalName`/`topic`; unknown selector fields fail the route at build time, as do +two inputs with identical selectors. | Selector field | Type | Meaning | |----------------|------|---------| @@ -163,12 +165,12 @@ inputs with identical selectors. | `device` | string | Match the source envelope identity's device. Identity-based fields never match a message without an envelope identity. | | `component` | string | Match the source envelope identity's component token. | | `instance` | string | Match the source envelope identity's instance token. | -| `required` | boolean (default `true`) | Whether the script waits for this input before its first evaluation. An optional input is absent from the `inputs` snapshot until it arrives. | +| `required` | boolean (default `false`) | Opt this input into stage-level gating. When `true`, the stage withholds every evaluation until this input has been observed. When `false` (default), the stage never waits on it — the script owns completeness and the input is simply absent from the snapshot until it arrives. | Cached input state is **partitioned by the source device** (the envelope identity), so two devices -publishing the same signal ids never mix into one snapshot. State is in-memory and empty at startup: -the stage re-awaits every required input before the first evaluation. A message that matches no -input is consumed by the stage. +publishing the same signal ids never mix into one snapshot. State is in-memory and empty at startup; +each input (re)initializes on its next message. A message that matches no input is consumed by the +stage. **`output` — an explicit output topic.** Without `output`, the script result replaces the triggering message's body in place (the classic behavior). With `output`, each successful evaluation is diff --git a/docs/sample-configurations.md b/docs/sample-configurations.md index dfd807b..bb15f95 100644 --- a/docs/sample-configurations.md +++ b/docs/sample-configurations.md @@ -951,7 +951,10 @@ into whichever source message happened to arrive last. See ``` ```lua --- scripts/oee.lua +-- scripts/oee.lua — the script owns completeness (the stage does not gate by default) +for _, k in ipairs({ "running", "idealCycleS", "plannedRunS", "totalCount", "goodCount" }) do + if inputs[k] == nil then return nil end -- wait until every operand has arrived +end if inputs.running.value ~= true then return nil end -- line stopped → hold the last output local perf = (inputs.idealCycleS.value * inputs.totalCount.value) / inputs.plannedRunS.value @@ -968,9 +971,10 @@ return { What each lever does here: - **`inputs` selectors** — each operand is pinned by `device` + `signalId`, so an unrelated signal - under the same subscribe filter is consumed without effect. The stage evaluates only after **all - five** inputs have been observed (they are all `required` by default) and then on every - value/quality change of any one of them. Input state is partitioned per source device. + under the same subscribe filter is consumed without effect. The stage runs the script on every + value/quality change of any operand and does **not** gate on missing inputs, so the script waits + for all five itself (the `ipairs` guard). Prefer `"required": true` on each input if you'd rather + the stage do the waiting and drop the guard. Input state is partitioned per source device. - **`output.topic`** — every successful evaluation publishes a fresh `OeeSnapshot` envelope on the `oee` instance's `data/current` topic, produced by the processor (identity instance = `oee-filler`) and correlated (`correlation_id`) to the triggering update. The topic must not fall diff --git a/docs/scripting.mdx b/docs/scripting.mdx index 4fa51b0..a488301 100644 --- a/docs/scripting.mdx +++ b/docs/scripting.mdx @@ -159,10 +159,16 @@ Two bindings join the ordinary scope (which still describes the *triggering* mes The semantics, precisely: -- **Initialization gate.** The script does not run until every input marked `required` (the default) - has been observed at least once — a half-initialized OEE is never computed. An input with - `"required": false` doesn't gate; it is simply absent from `inputs` until it arrives, so the - script checks for it (`inputs.x != ()` in Rhai, `inputs.x ~= nil` in Lua… or Rhai's `"x" in inputs`). +- **Completeness is the script's job, not the stage's.** By default the stage does **not** withhold + evaluation for missing inputs: it runs the script on the first — and every — matched change, and an + input that hasn't arrived yet is simply absent from `inputs`. The script decides whether it has + enough to compute and returns the drop value (`()` in Rhai / `nil` in Lua) otherwise. Check for an + input with `"x" in inputs` (Rhai) or `inputs.x ~= nil` (Lua). This keeps the *policy* — which + operands are essential, and what to do while you wait — in the script, where the domain logic lives. +- **Opt-in stage gating (`required: true`).** If you'd rather the stage hold off until an operand + exists, mark that input `"required": true`; the stage then withholds every evaluation until all + such inputs have been observed. It's a convenience for the common "need them all" case — the script + is still free to guard beyond it (e.g. on quality). With no `required` input, the stage never gates. - **Change detection.** A message that repeats an input's current value *and* quality refreshes its timestamps but does not re-evaluate. A quality flip alone (GOOD → BAD) *is* a change — the script decides what to do with a bad-quality operand (return the drop value to hold the last output). @@ -172,8 +178,8 @@ The semantics, precisely: - **Consumed, not forwarded.** A multi-input stage is a *sink* for its matched messages: the triggering message is consumed (its data lives on in the snapshot), and a subscribed message that matches no input is consumed silently. Put a multi-input `script` last in its pipeline. -- **Restart-empty.** The cache is in-memory; after a restart the stage deterministically re-awaits - its required inputs before the first evaluation (each initialization is logged). +- **Restart-empty.** The cache is in-memory; after a restart each input re-initializes on its next + message (an `inputs`-guarding script simply waits for what it needs, as it does at first startup). **The output side.** Without `output`, the result replaces the triggering message's body in place — which couples the derived value to whichever source topic happened to move last. A derived signal @@ -185,13 +191,18 @@ route's startup validation rejects an output topic on a reserved UNS class or on `subscribe` filters would re-consume (a feedback loop), and the processor's self-echo guard drops any re-consumed copy of its own output as a second line of defense. -A working OEE script over those inputs: +A working OEE script over those inputs. Because the stage doesn't gate, the script waits for its +operands itself — a one-line guard. (Prefer `"required": true` on the five inputs if you'd rather the +stage hold off until they exist; then you can drop the presence check.) ```rhai // Availability × Performance × Quality, refreshed on any operand change. +for k in ["running", "idealCycleS", "plannedRunS", "totalCount", "goodCount"] { + if !(k in inputs) { return (); } // wait until every operand exists +} if inputs.running.value != true { return (); } // line stopped → hold the last value let perf = (inputs.idealCycleS.value * inputs.totalCount.value) / inputs.plannedRunS.value; @@ -210,6 +221,9 @@ let qual = inputs.goodCount.value / inputs.totalCount.value; ```lua -- Availability × Performance × Quality, refreshed on any operand change. +for _, k in ipairs({ "running", "idealCycleS", "plannedRunS", "totalCount", "goodCount" }) do + if inputs[k] == nil then return nil end -- wait until every operand exists +end if inputs.running.value ~= true then return nil end -- line stopped → hold the last value local perf = (inputs.idealCycleS.value * inputs.totalCount.value) / inputs.plannedRunS.value diff --git a/src/config.rs b/src/config.rs index 7ee3c1b..2e5a4e8 100644 --- a/src/config.rs +++ b/src/config.rs @@ -128,8 +128,10 @@ pub struct InputSelector { pub signal_name: Option, /// Match the arriving topic against this MQTT-style filter (`+`/`#` supported). pub topic: Option, - /// Whether the script waits for this input before its first evaluation. Default `true`; an - /// optional input (`false`) is simply absent from the `inputs` snapshot until it arrives. + /// Opt this input into stage-level gating: when `true`, the stage withholds every evaluation + /// until this input has been observed. Default `false` — the stage does not gate and the + /// script decides whether the (possibly partial) `inputs` snapshot is enough to compute; an + /// input that has not arrived is simply absent from the snapshot. pub required: Option, } diff --git a/src/proc/multi.rs b/src/proc/multi.rs index 47e2dc1..e119b43 100644 --- a/src/proc/multi.rs +++ b/src/proc/multi.rs @@ -4,16 +4,23 @@ //! by a selector (`device`/`component`/`instance` against the envelope identity, `signalId`/ //! `signalName` against `body.signal`, and/or an MQTT-style `topic` filter). The stage caches the //! latest value/quality/timestamps of every input and evaluates the script when a matched input's -//! **value or quality changes**, binding a consistent snapshot of all inputs as `inputs` and the -//! firing input as `trigger` (alongside the ordinary per-message bindings for the triggering -//! message). The script does not run until every `required` input has been initialized. +//! **value or quality changes**, binding a consistent snapshot of the currently-known inputs as +//! `inputs` and the firing input as `trigger` (alongside the ordinary per-message bindings for the +//! triggering message). +//! +//! **Completeness is the script's responsibility, not the stage's.** By default the stage does not +//! withhold evaluation for missing inputs: it runs the script on the first (and every) matched +//! change, and an input that has not arrived yet is simply absent from the `inputs` snapshot +//! (`inputs.x == ()` in Rhai / `nil` in Lua). The script decides whether it has enough to compute +//! and returns the drop value (`()` / `nil`) otherwise. An input may **opt in** to stage-level +//! gating with `required: true`; the stage then withholds every evaluation until all such inputs +//! have been observed. With no `required` input configured, the stage never gates. //! //! Cached state is **partitioned by the source device** (the envelope identity's deepest hierarchy //! value), so two devices publishing the same signal ids can never contaminate each other's //! snapshot. Identity-based selectors only match messages that carry an envelope identity; //! identity-less sources are selected by explicit `topic` filters and share one partition. State is -//! in-memory and restart-empty: after a restart the stage deterministically re-awaits every -//! required input before the first evaluation. +//! in-memory and restart-empty. //! //! With an `output` configured, each successful evaluation is published as a **new** EdgeCommons //! envelope on `output.topic`: the body is the script result, the producer identity is the @@ -98,7 +105,9 @@ impl MultiScriptStage { inputs.push(CompiledInput { name: name.clone(), sel: sel.clone(), - required: sel.required.unwrap_or(true), + // Completeness is the script's job by default; `required: true` opts an input + // into stage-level gating. + required: sel.required.unwrap_or(false), }); } for i in 0..inputs.len() { @@ -201,7 +210,9 @@ impl Processor for MultiScriptStage { return smallvec![]; } - // 3. Gate: every required input must be initialized before the first evaluation. + // 3. Optional gate: only inputs that explicitly opt in with `required: true` withhold + // evaluation. By default nothing is required, so the stage never gates here and the + // script itself decides whether the (possibly partial) snapshot is enough to compute. let missing: Vec<&str> = self .inputs .iter() @@ -209,7 +220,7 @@ impl Processor for MultiScriptStage { .map(|ci| ci.name.as_str()) .collect(); if !missing.is_empty() { - tracing::debug!(missing = ?missing, "multi-signal evaluation deferred: awaiting inputs"); + tracing::debug!(missing = ?missing, "multi-signal evaluation deferred: awaiting required inputs"); return smallvec![]; } @@ -372,10 +383,13 @@ mod tests { ) } - /// A two-input spec (a + b) whose script returns both values plus the trigger name. + /// A two-input spec (a + b). The stage does not gate (no `required`), so the **script** guards + /// its own completeness — it computes only once both inputs are present, else drops. fn two_input_spec(output: bool) -> ScriptSpec { let mut v = json!({ - "source": r#"#{ "a": inputs.a.value, "b": inputs.b.value, "by": trigger.name }"#, + "source": r#"if "a" in inputs && "b" in inputs { + #{ "a": inputs.a.value, "b": inputs.b.value, "by": trigger.name } + } else { () }"#, "inputs": { "a": { "device": "gw-1", "signalId": "A" }, "b": { "device": "gw-1", "signalId": "B" } @@ -388,11 +402,12 @@ mod tests { } #[test] - fn does_not_evaluate_until_all_required_inputs_initialized() { + fn script_owns_completeness_and_computes_once_all_inputs_present() { let mut s = build(&two_input_spec(false)); - // Only `a` has arrived → no evaluation, out-of-order init is fine. + // Only `a` has arrived. The stage DOES invoke the script (it does not gate); the script + // itself drops because `b` is not in the snapshot yet. Out-of-order init is fine. assert!(s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).is_empty()); - // `b` arrives → the snapshot is complete and the script runs once. + // `b` arrives → the script sees both and computes. let out = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); assert_eq!(out.len(), 1); assert_eq!(out[0].msg.body, json!({ "a": 1, "b": 2, "by": "b" })); @@ -422,30 +437,44 @@ mod tests { } // A script that ignores its inputs and always returns a constant. It CANNOT be the thing - // deciding whether to emit — so if the stage ever invokes it, output appears. These two tests - // therefore isolate the *stage's* gating/change-detection from any fail-closed script - // behavior (a real script that errors or returns nil on a missing/unchanged input would also - // yield no output, which would not distinguish the two layers). - fn constant_two_input_spec() -> ScriptSpec { + // deciding whether to emit — so output appears exactly when the stage invokes it. These tests + // therefore isolate the *stage's* (non-)gating and change-detection from any fail-closed + // script behavior (a real script that drops on a missing input would also yield no output, + // which would not distinguish the two layers). + fn constant_two_input_spec(required: bool) -> ScriptSpec { + let sel = |id: &str| { + if required { + json!({ "device": "gw-1", "signalId": id, "required": true }) + } else { + json!({ "device": "gw-1", "signalId": id }) + } + }; spec_from(json!({ "source": r#"#{ "fired": true }"#, - "inputs": { - "a": { "device": "gw-1", "signalId": "A" }, - "b": { "device": "gw-1", "signalId": "B" } - } + "inputs": { "a": sel("A"), "b": sel("B") } })) } #[test] - fn stage_gates_invocation_the_script_is_not_called_before_all_inputs() { - let mut s = build(&constant_two_input_spec()); - // Only `a` present. A constant script WOULD emit `{fired:true}` if the stage called it — - // it doesn't, proving the stage withholds the invocation itself. + fn default_does_not_gate_the_script_runs_on_the_first_change() { + // No `required` inputs → the stage must NOT gate. A constant script that ignores its + // inputs fires on the very first matched change, proving the core does not withhold + // invocation for missing inputs — completeness is the script's own responsibility. + let mut s = build(&constant_two_input_spec(false)); + let out = s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); + assert_eq!(out.len(), 1, "stage must invoke the script on the first change, not gate"); + assert_eq!(out[0].msg.body, json!({ "fired": true })); + } + + #[test] + fn required_true_opts_into_stage_gating() { + // With `required: true` on both inputs, the stage DOES withhold evaluation until both are + // present. A constant script would emit if called, so its silence proves the stage gates. + let mut s = build(&constant_two_input_spec(true)); assert!( s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).is_empty(), - "the stage must not invoke the script before all required inputs are initialized" + "with required:true the stage withholds invocation until all required inputs arrive" ); - // `b` completes the set → now the stage invokes it and the constant is emitted. let out = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); assert_eq!(out.len(), 1); assert_eq!(out[0].msg.body, json!({ "fired": true })); @@ -453,13 +482,12 @@ mod tests { #[test] fn stage_suppresses_invocation_on_an_unchanged_value() { - let mut s = build(&constant_two_input_spec()); - s.process(signal_msg("gw-1", "A", json!(1), "GOOD")); - let first = s.process(signal_msg("gw-1", "B", json!(2), "GOOD")); - assert_eq!(first.len(), 1, "fires once both inputs are present"); - // Re-send `a` with the same value + quality. A constant script WOULD emit again if the - // stage called it — it doesn't, proving the stage's change detection suppresses the - // invocation (not the script choosing to return nothing). + // Change detection is independent of gating. A constant script fires on each genuine + // change but not on a repeat, proving the STAGE suppresses the unchanged re-evaluation. + let mut s = build(&constant_two_input_spec(false)); + assert_eq!(s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).len(), 1, "fires on init"); + assert_eq!(s.process(signal_msg("gw-1", "B", json!(2), "GOOD")).len(), 1, "fires on change"); + // Re-send `a` with the same value + quality → the stage does not invoke the script again. assert!( s.process(signal_msg("gw-1", "A", json!(1), "GOOD")).is_empty(), "the stage must not invoke the script when no input value/quality changed" @@ -568,7 +596,7 @@ mod tests { } })); let mut s = build(&spec); - // The optional `c` is absent from the snapshot; the required `a` alone evaluates. + // `c` is absent from the snapshot; the script computes from `a` alone (no gating). let out = s.process(signal_msg("gw-1", "A", json!(3), "GOOD")); assert_eq!(out[0].msg.body, json!({ "a": 3 })); // Once `c` arrives it appears in the snapshot. @@ -634,10 +662,16 @@ mod tests { use super::*; /// The OEE shape from the design: named inputs, a Lua calculation, an explicit output. + /// The stage does not gate (no `required`); the **script** owns completeness by checking + /// each operand is present before computing. #[test] - fn lua_oee_calculation_over_named_inputs() { + fn lua_oee_script_owns_completeness_over_named_inputs() { let spec = spec_from(json!({ "source": r#" + if inputs.running == nil or inputs.totalCount == nil + or inputs.plannedCount == nil then + return nil -- wait for all operands (the script gates, not the stage) + end if inputs.running.value ~= true then return nil end local avail = inputs.totalCount.value / inputs.plannedCount.value return { oee = avail, by = trigger.name }