Add stateful multi-signal inputs + explicit output to the script stage

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 |

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
 

docs/how-to-guides.md

@@ -201,6 +201,56 @@ if celsius == () { return (); }            // no reading → drop
 
 ---
 
+<a id="compute-a-derived-signal-from-several-inputs"></a>
+## 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.<name>.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.
+
+---
+
 <a id="aggregate-a-non-southbound-payload"></a>
 ## Aggregate a non-southbound payload
 

docs/reference/configuration.md

@@ -140,12 +140,54 @@ follows the engine. The source is given **inline** or from an **external file**:
 |------|---------|
 | `{"script": "<source>"}` | 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)**.
 
+<a id="script-inputs"></a>**`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.
+
 <a id="rhai-scope"></a>**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 |
 
 <a id="key-paths"></a>
 > **Key paths** are dotted paths over the message: roots `body.` (the default when no known root

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:<name>`** 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