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

[mbreissi]

Implements #5: stateful, named multi-signal inputs for script pipeline stages, with an optional explicit output topic for the derived result (the OEE use case).

What changed

• Config — the script stage's object form gains inputs (name → selector: signalId/signalName/topic filter, narrowed by envelope-identity device/component/instance, per-input required) and output (topic, optional envelope name/version). Both legacy forms ("script": "<expr>", {"file": …}) parse and behave exactly as before. Unknown selector fields and duplicate selectors fail the route at build time.
• MultiScriptStage (src/proc/multi.rs) — caches the latest value/quality/source-timestamp/receive-time per input, partitioned by source device, and evaluates when a matched input's value or quality changes, only after every required input is initialized. The script gets inputs.<name>.{value,quality,timestamp,recvMs,topic} and trigger.{name,value,quality,…} in both Rhai and Lua, alongside the ordinary per-message scope.
• Output — with output.topic, each successful evaluation publishes a new envelope: producer identity = the processor with instance = route id, correlation_id = the triggering message's uuid; the trigger is consumed, never republished. Without output, the in-place result contract is unchanged.
• Validation — startup errors for reserved-class output topics, output topics overlapping the route's own subscribe filters (feedback loop; the self-echo guard remains the runtime backstop), empty topics, and publish.topic + stage output.topic together.
• Docs — scripting-guide section (semantics + OEE in both engines), configuration-reference selector/output tables, ScriptResult envelope reference, sample §12 (Lua OEE route), and a how-to.

Acceptance criteria → tests (src/proc/multi.rs, src/config.rs, src/app.rs)

• no evaluation before all required inputs initialize (out-of-order init) ✔
• any input update re-evaluates with the full snapshot ✔ · trigger identification ✔
• quality + source/receive timestamps in scope ✔ · unchanged repeats don't re-fire, quality flips do ✔
• configured output: one new valid envelope per evaluation, producer/correlation provenance, trigger not republished ✔
• no output: in-place behavior preserved (legacy suites untouched and green) ✔
• reserved/looping/ambiguous configs rejected ✔ · per-device state isolation ✔ · restart-empty determinism ✔
• Lua OEE end-to-end stage test (feature-gated) ✔

Validation

• cargo test: 65 passed (default features), 78 passed (--features scripting-lua)
• cargo clippy --all-targets: clean on both feature sets

🤖 Generated with Claude Code

https://claude.ai/code/session_019emrYEw39h1jJbsA6C8oBZ

[mbreissi]

Live-broker E2E — PASS (24/24 checks)

Ran the full HOST/dual-MQTT path from the validation matrix: the real telemetry-processor binary (built --features standalone,streaming,streaming-file-parquet,scripting-lua) against EMQX on localhost:1883, driving genuine SouthboundSignalUpdate protobuf envelopes (built with the Python edgecommons library — wire-identical to the Rust component, the same lib the cross-language interop suite uses) and decoding the derived output off the broker.

Two multi-signal routes exercised end to end:

• oee-filler — device-pinned 5-input Lua OEE, output.topic …/telemetry-processor/oee/data/current
• ratio-fleet — fleet selectors (signalId-only, no device pin) to prove runtime per-device partitioning

Verified over the wire:

Behavior	Result
No output until all 5 required inputs initialized (out-of-order arrival)	✅
First compute correct (OEE 0.81, perf/qual 0.9)	✅
Recompute on any input change; trigger.name correct (totalCount → goodCount, OEE 0.855)	✅
Output is a new OeeSnapshot envelope; producer identity component=telemetry-processor, instance=oee-filler	✅
correlation_id == the triggering message's uuid (tracks across recomputes)	✅
Repeated identical value does not re-evaluate	✅
Bad-quality operand re-triggers but the script suppresses output (holds last)	✅
running=false → script returns nil → no output (hold last)	✅
Per-device isolation: gw-fill-02's half-init partition produces no cross-device leak; completes only from its own values (ratio 0.5 from good=50/total=100)	✅
Device-pinned route ignores the second device entirely	✅

Processor log clean (no WARN/ERROR); graceful shutdown. Broker + processor torn down after.

[mbreissi]

Correction: gating is the script's responsibility (default off) — re-validated live

An earlier revision had the core stage withhold script invocation until all required inputs were initialized (following issue #5's original wording). That was wrong per explicit design direction: completeness is the script writer's responsibility, not the core's. Fixed.

Change: required now defaults to false. By default the stage invokes the script on the first (and every) matched change with whatever inputs are present; an unarrived input is absent from the inputs snapshot and the script guards itself. required: true is a per-input opt-in for stage-level gating. Issue #5's acceptance criterion corrected in a comment there.

Tests rewritten to isolate the mechanism with a constant (input-ignoring) script:

• default_does_not_gate_the_script_runs_on_the_first_change — fires on the first input → no core gate.
• required_true_opts_into_stage_gating — withholds until all required:true inputs arrive.
• stage_suppresses_invocation_on_an_unchanged_value — change detection independent of gating.

68 default / 81 scripting-lua tests pass; clippy clean.

Live E2E re-run — PASS (27/27), log clean

Added Phase 0 as direct proof the core does not gate: an unguarded probe.lua (emits on every change) fired on the first of its two inputs with presentCount=1, aPresent=true, bPresent=false — the stage invoked the script on a partial snapshot. OEE is now gated by the script (a presence guard), and all prior checks (recompute+trigger, output-envelope provenance, unchanged/bad-quality/stopped suppression, per-device partitioning) still pass over real EMQX + protobuf envelopes.