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..5f022a3 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,27 @@ 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`.
+ 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
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 +231,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..1432c01 100644
--- a/docs/how-to-guides.md
+++ b/docs/how-to-guides.md
@@ -201,6 +201,56 @@ 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. 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": [
+ { "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", "required": true },
+ "total": { "device": "gw-fill-01", "signalId": "TotalBottleCount", "required": true }
+ },
+ "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.
+- **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
+ 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..d9684dc 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -140,12 +140,54 @@ 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 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 |
+|----------------|------|---------|
+| `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 `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;
+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
+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 +206,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..bb15f95 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,81 @@ 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 — 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
+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 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
+ 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..a488301 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,133 @@ 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:
+
+- **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).
+- **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 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
+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. 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;
+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.
+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
+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 +606,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..2e5a4e8 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -59,6 +59,97 @@ 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,
+ /// 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,
+}
+
+/// 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 +184,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 +324,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..e119b43
--- /dev/null
+++ b/src/proc/multi.rs
@@ -0,0 +1,712 @@
+//! # 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 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.
+//!
+//! 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(),
+ // 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() {
+ 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. 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()
+ .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 required 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). 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#"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" }
+ }
+ });
+ if output {
+ v["output"] = json!({ "topic": "ecv1/gw-1/telemetry-processor/r1/data/derived" });
+ }
+ spec_from(v)
+ }
+
+ #[test]
+ fn script_owns_completeness_and_computes_once_all_inputs_present() {
+ let mut s = build(&two_input_spec(false));
+ // 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 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" }));
+ }
+
+ #[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);
+ }
+
+ // A script that ignores its inputs and always returns a constant. It CANNOT be the thing
+ // 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": sel("A"), "b": sel("B") }
+ }))
+ }
+
+ #[test]
+ 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(),
+ "with required:true the stage withholds invocation until all required inputs arrive"
+ );
+ 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() {
+ // 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"
+ );
+ }
+
+ #[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);
+ // `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.
+ 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.
+ /// The stage does not gate (no `required`); the **script** owns completeness by checking
+ /// each operand is present before computing.
+ #[test]
+ 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 }
+ "#,
+ "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,
})
}