Pin tolerance for additive sample fields and explicit-null samples

src/dispatch.rs

@@ -429,6 +429,49 @@ mod tests {
         assert_eq!(s2.messages_in.load(Ordering::Relaxed), 1);
     }
 
+    #[tokio::test]
+    async fn forwards_additive_sample_fields_and_explicit_null_value_untouched() {
+        // SOUTHBOUND §2 ("Explicit null values" + "Additive sample fields"): a SouthboundSignalUpdate
+        // whose samples carry unknown additive fields (valueType/valueEncoding) and an explicit
+        // null value with quality GOOD must flow through ingestion without error, without the
+        // sample being dropped, and without the additive fields being stripped.
+        let body = json!({
+            "signal": { "id": "bacnet-av-3", "name": "AV3" },
+            "samples": [{
+                "value": null, "quality": "GOOD", "qualityRaw": "relinquished",
+                "sourceTs": "2026-07-26T00:00:00Z",
+                "valueType": "REAL", "valueEncoding": "scalar"
+            }]
+        });
+        let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0")
+            .southbound_signal_update(body.clone())
+            .build();
+
+        let fake = FakeMessaging::new();
+        let messaging: Arc<dyn MessagingService> = fake.clone();
+        let (_rec, evt) = EvtEmitter::recording();
+        let (tx, mut rx) = mpsc::channel(4);
+        let stats = RouteStats::new("r1");
+
+        self_subscribe(
+            &messaging,
+            "ecv1/+/+/+/data/#",
+            vec![(tx, stats.clone())],
+            "my-device".into(),
+            "telemetry-processor".into(),
+            evt,
+        )
+        .await
+        .unwrap();
+
+        fake.deliver("ecv1/+/+/+/data/#", "ecv1/other/bacnet-adapter/i1/data/av3", m).await;
+
+        let pm = rx.try_recv().expect("the explicit-null sample must be forwarded, not dropped");
+        assert_eq!(pm.msg.body, body, "additive sample fields and the null value must survive ingestion verbatim");
+        assert_eq!(stats.messages_in.load(Ordering::Relaxed), 1);
+        assert_eq!(stats.messages_dropped.load(Ordering::Relaxed), 0);
+    }
+
     #[tokio::test]
     async fn drops_a_re_consumed_message_carrying_its_own_identity() {
         let fake = FakeMessaging::new();

src/proc/aggregate.rs

@@ -315,6 +315,61 @@ mod tests {
         assert_eq!(agg["count"], json!(5));
     }
 
+    #[test]
+    fn explicit_null_samples_fold_per_documented_null_semantics() {
+        // SOUTHBOUND §2: explicit null values (quality GOOD) must not crash aggregation. This
+        // repo's documented null semantics (docs/reference/data-types.md "Aggregate `agg` value
+        // types"): `count` counts ALL folded values (null included), `first`/`last` carry the raw
+        // value (null included), and numeric reducers are null when the window had no numeric
+        // sample. Additive sample fields (valueType/...) beside the value must not disturb any of it.
+        let spec = AggregateSpec {
+            window: "2".into(),
+            by: None,
+            fns: vec!["avg".into(), "sum".into(), "min".into(), "max".into(), "count".into(), "first".into(), "last".into()],
+            value: None,
+        };
+        let mk = |value: Value, recv: u64| {
+            let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0")
+                .payload(json!({
+                    "signal": { "id": "a" },
+                    "samples": [ { "value": value, "quality": "GOOD", "qualityRaw": "relinquished",
+                                   "valueType": "REAL", "valueEncoding": "scalar" } ]
+                }))
+                .build();
+            ProcMsg { topic: "t".into(), msg: m, recv_ms: recv }
+        };
+
+        // Window 1: all-null. Numeric reducers are null; count still counts both samples.
+        let mut s = AggregateStage::build(&spec, "body.signal.id").unwrap();
+        assert!(s.process(mk(Value::Null, 1)).is_empty());
+        let out = s.process(mk(Value::Null, 2));
+        assert_eq!(out.len(), 1);
+        let agg = &out[0].msg.body["agg"];
+        assert_eq!(agg["avg"], Value::Null);
+        assert_eq!(agg["sum"], Value::Null);
+        assert_eq!(agg["min"], Value::Null);
+        assert_eq!(agg["max"], Value::Null);
+        assert_eq!(agg["count"], json!(2));
+        assert_eq!(agg["first"], Value::Null);
+        assert_eq!(agg["last"], Value::Null);
+        // The primary (first fn = avg) lands as samples[0].value = null without error.
+        assert_eq!(out[0].msg.body["samples"][0]["value"], Value::Null);
+
+        // Window 2: mixed null + numeric. Numeric reducers fold the numeric values only; count
+        // includes the null.
+        assert!(s.process(mk(json!(10.0), 3)).is_empty());
+        let out = s.process(mk(Value::Null, 4));
+        assert_eq!(out.len(), 1);
+        let agg = &out[0].msg.body["agg"];
+        assert_eq!(agg["avg"], json!(10.0));
+        assert_eq!(agg["sum"], json!(10.0));
+        assert_eq!(agg["min"], json!(10.0));
+        assert_eq!(agg["max"], json!(10.0));
+        assert_eq!(agg["count"], json!(2));
+        assert_eq!(agg["first"], json!(10.0));
+        assert_eq!(agg["last"], Value::Null);
+    }
+
     #[test]
     fn aggregate_custom_value_path_non_southbound() {
         // A non-SouthboundSignalUpdate payload: aggregate `body.temp`, keyed by `body.id`.

src/proc/filter.rs

@@ -233,6 +233,25 @@ mod tests {
         assert_eq!(mk("body.signal.id", "contains", Some(json!("zzz"))).process(m()).len(), 0);
     }
 
+    #[test]
+    fn quality_filter_keeps_explicit_null_value_with_good_quality_and_additive_fields() {
+        // SOUTHBOUND §2: an explicit null value with quality GOOD is a legitimate sample (e.g. a
+        // BACnet relinquished slot), and samples MAY carry additive fields (valueType/valueEncoding)
+        // consumers MUST ignore. A quality filter keys on `quality` only — it must keep the sample.
+        let spec = FilterSpec { quality: Some("GOOD".into()), ..Default::default() };
+        let mut s = FilterStage::build(&spec, ScriptEngineKind::Rhai, &engine(), &ScriptLoader::default(), &ctx()).unwrap();
+        let m = msg(json!([{
+            "value": null, "quality": "GOOD", "qualityRaw": "relinquished",
+            "valueType": "REAL", "valueEncoding": "scalar"
+        }]));
+        assert_eq!(s.process(m).len(), 1, "explicit-null GOOD sample must not be dropped");
+
+        // A failed read (null + BAD) is still droppable by the same filter — the two null cases
+        // stay distinguishable by quality alone.
+        let bad = msg(json!([{ "value": null, "quality": "BAD", "qualityRaw": "timeout" }]));
+        assert_eq!(s.process(bad).len(), 0);
+    }
+
     #[test]
     fn build_and_op_parse_errors() {
         // No predicate form configured.

src/proc/mod.rs

@@ -187,6 +187,50 @@ mod tests {
         assert_eq!(out[0].msg.body["agg"]["avg"], json!(25.0));
     }
 
+    #[test]
+    fn southbound_s2_null_and_additive_sample_fields_flow_through_a_pass_through_pipeline() {
+        // SOUTHBOUND §2 pin: a representative pass-through pipeline (quality filter → sample →
+        // project keep) must carry an explicit-null GOOD sample with additive fields
+        // (valueType/valueEncoding) end to end — no error, no dropped sample, no stripped fields.
+        use crate::config::{ProjectSpec, StageConfig};
+
+        let sample = json!({
+            "value": null, "quality": "GOOD", "qualityRaw": "relinquished",
+            "sourceTs": "2026-07-26T00:00:00Z",
+            "valueType": "REAL", "valueEncoding": "scalar"
+        });
+        let stages = vec![
+            StageConfig::Filter(FilterSpec { quality: Some("GOOD".into()), ..Default::default() }),
+            StageConfig::Sample(SampleSpec { every_n: Some(1), ..Default::default() }),
+            StageConfig::Project(ProjectSpec {
+                keep: Some(vec!["signal".into(), "samples".into()]),
+                set: None,
+            }),
+        ];
+        let mut p = Pipeline::build(
+            &stages,
+            "body.signal.id",
+            ScriptEngineKind::Rhai,
+            &engine(),
+            &script::ScriptLoader::default(),
+            &Arc::new(script::ScriptContext::default()),
+        )
+        .unwrap();
+
+        let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0")
+            .payload(json!({ "signal": { "id": "bacnet-av-3" }, "samples": [sample.clone()] }))
+            .build();
+        let mut input: Out = SmallVec::new();
+        input.push(ProcMsg { topic: "t".into(), msg: m, recv_ms: 1 });
+
+        let out = p.run(input, None);
+        assert_eq!(out.len(), 1, "the explicit-null GOOD sample must survive the pipeline");
+        assert_eq!(
+            out[0].msg.body["samples"][0], sample,
+            "additive fields and the null value must pass through verbatim"
+        );
+    }
+
     #[test]
     fn flush_via_max_tick_closes_open_time_window() {
         // The `flush` command verb force-closes open TIME windows by running a `u64::MAX` tick

src/proc/project.rs

@@ -80,6 +80,31 @@ mod tests {
         assert!(body.get("noise").is_none());
     }
 
+    #[test]
+    fn keep_preserves_additive_sample_fields_and_explicit_null_value() {
+        // SOUTHBOUND §2: samples MAY carry additive fields beside the canonical five, and a
+        // deliberate null value rides with quality GOOD. `keep` whitelists top-level body keys and
+        // copies them verbatim — the samples entries (extras and null included) must survive intact.
+        let sample = json!({
+            "value": null, "quality": "GOOD", "qualityRaw": "relinquished",
+            "sourceTs": "2026-07-26T00:00:00Z",
+            "valueType": "REAL", "valueEncoding": "scalar"
+        });
+        let m = MessageBuilder::new("SouthboundSignalUpdate", "1.0")
+            .payload(json!({ "signal": { "id": "t1" }, "samples": [sample.clone()], "noise": 1 }))
+            .build();
+        let pm = ProcMsg { topic: "t".into(), msg: m, recv_ms: now_ms() };
+
+        let mut s = ProjectStage::build(&ProjectSpec {
+            keep: Some(vec!["signal".into(), "samples".into()]),
+            set: None,
+        });
+        let out = s.process(pm);
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].msg.body["samples"][0], sample, "sample must pass through byte-identical");
+        assert!(out[0].msg.body.get("noise").is_none());
+    }
+
     #[test]
     fn set_overlays_literals() {
         let mut set = Map::new();