Merge pull request #1588 from ruvnet/fix/issue-triage-batch-1

fix: issue triage batch (#1521-1557) + wire the certificate spine together

.github/workflows/model-release-gate.yml

@@ -0,0 +1,67 @@
+name: Model release gate (ADR-298)
+
+# ADR-298 model-release sanity gates (issue #1521): structural checks that
+# block a degenerate/mislabeled classifier head (unreachable decision
+# boundary, near-constant output, degenerate class balance, a metric
+# surfaced under a task name it wasn't computed as) before it ships.
+#
+# Checker: v2/crates/wifi-densepose-train/src/model_gates.rs
+#
+# IMPORTANT — the honest scope of this job: it protects the *checker itself*
+# from regressing (the gate logic + its issue-1521 regression fixture are
+# exercised on every push/PR that touches this crate), and running it is
+# required before ADR-298 can be called "wired in" at all. It does NOT gate
+# an actual model publish — this repository does not automate uploading to
+# the HuggingFace model repo (`ruvnet/wifi-densepose-pretrained`); that
+# remains a manual, human-run step. Before publishing or replacing a model
+# artifact there, run this gate against the real head weights locally:
+#
+#   cargo test -p wifi-densepose-train model_gates
+#
+# and, until a CLI entry point exists to run `evaluate_linear_head` against an
+# arbitrary `.safetensors`/`.rvf` file, load the head's `weight`/`bias` in a
+# short script and call `wifi_densepose_train::evaluate_linear_head` directly.
+
+on:
+  push:
+    branches:
+      - main
+      - master
+    paths:
+      - "v2/crates/wifi-densepose-train/**"
+  pull_request:
+    paths:
+      - "v2/crates/wifi-densepose-train/**"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  model-release-gate:
+    name: Model release gate check
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262
+        with:
+          persist-credentials: false
+          submodules: recursive
+
+      - name: Install Rust toolchain
+        run: rustup toolchain install stable --profile minimal
+
+      - name: Run the model-release gate's own test suite
+        working-directory: v2
+        run: cargo test -p wifi-densepose-train --no-default-features model_gates -- --nocapture
+
+      - name: Summarize result
+        if: always()
+        run: |
+          {
+            echo '### Model release gate (ADR-298)'
+            echo ''
+            echo 'This job protects `model_gates.rs` from regressing. It does not itself'
+            echo 'gate a real HuggingFace model publish — that upload is a manual step'
+            echo 'outside this repository; run `cargo test -p wifi-densepose-train model_gates`'
+            echo 'against real head weights before publishing one.'
+          } >> "$GITHUB_STEP_SUMMARY"

README.md

@@ -223,7 +223,7 @@ huggingface-cli download ruvnet/wifi-densepose-pretrained --local-dir models/wif
 
 | Consumer | Format used | Status |
 |----------|-------------|--------|
-| Python training / evaluation / embedding extraction | `model.safetensors` | ✅ Works — load with `safetensors.torch.load_file` |
+| Python training / evaluation / embedding extraction | `model.safetensors` | ⚠️ The published file's header is NUL-padded, which the reference `safetensors.torch.load_file` rejects (issue [#1522](https://github.com/ruvnet/RuView/issues/1522)) — pending a corrected re-upload. `csi-embed-v2.safetensors` in the same repo is unaffected and loads normally. |
 | Inspect / re-export the bundle | `model.rvf.jsonl` (line-by-line JSON) | ✅ Works — plain JSONL |
 | Sensing-server `--model <PATH>` flag | native RVF, `model.safetensors`, or `model.rvf.jsonl` | ✅ Native RVF loads directly; safetensors and JSONL auto-convert in memory |
 

docs/user-guide.md

@@ -1502,12 +1502,24 @@ action` pipeline, where a downstream consumer either gets a calibrated, provenan
 answer or an explicit `UNKNOWN` — never a confident-looking guess outside the sensor's
 proven operating envelope.
 
-**Status: developer preview.** Phase 1 shipped nine new crates with their own test
-suites, and each one works correctly in isolation. **They are not yet wired together or
-into the live `sensing-server` request path** — there is currently no code path where a
-real drift signal from a running sensor flows through calibration → certificate
-invalidation → policy denial. Treat everything below as a library you can compose
-yourself today, not a safety guarantee the server enforces for you yet.
+**Status: developer preview, now wired at the crate level.** Phase 1 shipped nine new
+crates. `ruview-certify` and `ruview-policy` now depend on `ruview-ood` and provide a
+real adapter (`impl From<ruview_ood::DomainState> for _`) plus a composed entry point,
+`ruview_policy::authorize_from_certificate`, that takes a real signed
+`CapabilityCertificate` and a real `ruview_ood::DomainState` and drives them through
+`authorize()` — not a hand-built `AssuranceInputs`. A cross-crate integration test
+(`ruview-policy`'s `acceptance_test_b_real_integration` module) mints an actual signed
+certificate and proves a real post-drift `Unknown` denies a `SafetyCritical` action
+through that one composed pipeline.
+
+**What's still not done:** none of this runs automatically inside the live
+`sensing-server` request path yet — there is no continuous calibration/OOD-monitoring
+loop wired into the running server that calls this pipeline on live sensor data. Treat
+`authorize_from_certificate` as a real, tested library entry point you can call from your
+own integration today, not something the server invokes for you on every request yet.
+That remaining step is a genuinely separate, larger effort (deciding polling cadence,
+where calibration state lives, what triggers re-certification) — see ADR-300 for the
+phased plan.
 
 ### The crates
 
@@ -1539,30 +1551,41 @@ assert!(!cert.is_valid(now_ms, DomainState::Degraded));
 assert!(!cert.is_valid(now_ms, DomainState::Unknown));
 ```
 
-### Gating an action
+### Gating an action from a real certificate + a real OOD reading
 
 ```rust
-use ruview_policy::{authorize, ActionClass, DomainState};
+use ruview_policy::authorize_from_certificate;
 
-let decision = authorize(ActionClass::SafetyCritical, &inputs);
-// Deny with a named FailedCondition (e.g. `domain_not_known`) rather than a
-// silent false-positive, whenever the domain isn't KNOWN.
+// `cert` (ruview_certify::CapabilityCertificate) and `domain`
+// (ruview_ood::DomainState) come from your own certify/OOD calls.
+let decision = authorize_from_certificate(
+    ActionClass::SafetyCritical,
+    &cert, &verifier, now_unix_s, domain,
+    certificate_class, uncertainty, evidence_level,
+);
+// Deny with a named FailedCondition (e.g. DomainNotKnown) — not a silent
+// false-positive — the moment `domain` degrades, even though `cert` itself
+// is still validly signed and unexpired.
 ```
 
-**Important:** `ruview_certify::DomainState` and `ruview_policy::DomainState` (and
-`ruview_ood`'s) are currently three separate enum types — `ruview-ood`'s `Degraded`
-variant even carries different data. There is no automatic conversion between them.
-If you compose these crates yourself today, you own writing that bridge; don't assume
-one crate's domain read automatically reaches another's gate.
+`ruview_certify::DomainState` and `ruview_policy::DomainState` are still each their own
+type (`ruview-ood`'s `Degraded`/`Unknown` additionally carry a `DomainCause`), but the
+conversion between them is no longer something you have to write yourself —
+`authorize_from_certificate` does it via the crates' own `From<ruview_ood::DomainState>`
+impls.
 
 ### What's genuinely enforced today, for comparison
 
-Not every ADR-295–296 remediation item is preview-only. Two are live now:
+Not every ADR-295–296 remediation item is preview-only. Three are live now:
 
 - **UDP data-plane bind hardening (ADR-296)** — `sensing-server`'s `UdpSourceAllowlist`
   is checked on every incoming packet (`main.rs`), not just defined.
 - **CSI data-incident repo controls (ADR-299)** — `scripts/csi-data-policy-check.sh`
   runs in CI on every push/PR and fails the build on a policy violation.
+- **Synthetic-export watermarking (ADR-295)** — `start_recording` stamps a `SYNTHETIC`
+  watermark on a recording's metadata (`GET /api/v1/recordings`, the start-recording
+  response) whenever it captures while the live source is synthetic — an operator
+  browsing or scripting against recordings can't mistake generated data for a capture.
 
 ---
 

ui/pose-fusion/js/main.js

@@ -140,13 +140,23 @@ function init() {
   csiCnn.tryLoadWasm(wasmBase);
 
   // Auto-connect to local sensing server WebSocket if available.
-  // Served from the Docker image the sensing stream lives on :3001 (same
-  // port mapping sensing.service.js uses); the standalone dev server stays
-  // on :8765.
-  const wsPortMap = { '3000': '3001' };
-  const mappedPort = wsPortMap[window.location.port];
-  const defaultWsUrl = mappedPort
-    ? `ws://${window.location.hostname}:${mappedPort}/ws/sensing`
+  // Served from the Docker image, the sensing stream lives one port above the
+  // HTTP port (3000 -> 3001 is the documented default mapping); the
+  // standalone dev server stays on :8765.
+  //
+  // Issue #1557: a hardcoded `{'3000': '3001'}` map only worked when the UI
+  // was served on exactly port 3000 — remapping the host port (e.g.
+  // `-p 3010:3000 -p 3011:3001`, needed whenever 3000 is already taken) fell
+  // through to `localhost:8765`, which for a remote viewer is nothing on
+  // *their* machine. `connectLive()` then failed silently. Derive the WS port
+  // from the actual HTTP port instead of a fixed lookup table, so it follows
+  // any host-port remapping automatically. This is a best-effort default, not
+  // a safety mechanism — `onVerifiedFrame` below (issue #1557 fix) is what
+  // actually prevents an unconfirmed/failed connection from ever being shown
+  // as LIVE, regardless of whether this guess is right.
+  const httpPort = Number(window.location.port);
+  const defaultWsUrl = Number.isFinite(httpPort) && httpPort > 0
+    ? `ws://${window.location.hostname}:${httpPort + 1}/ws/sensing`
     : 'ws://localhost:8765/ws/sensing';
   if (wsUrlInput) wsUrlInput.value = defaultWsUrl;
   // ADR-272: exchange the stored bearer for a single-use ?ticket= before the

ui/services/sensing.service.js

@@ -1,4 +1,5 @@
 import { withWsTicket } from './ws-ticket.js';
+import { apiService } from './api.service.js';
 /**
  * Sensing WebSocket Service
  *
@@ -308,8 +309,13 @@ class SensingService {
     // an *unknown* state — it must NOT collapse to "live". Prefer the canonical
     // `source_state` the server now returns; on any error stay conservative
     // (server-simulated) until a real frame's `source` field promotes us.
+    //
+    // Send the bearer token via `apiService.getHeaders()` so the probe can
+    // actually succeed under the documented secure posture (API auth
+    // enabled) instead of always 401ing and relying solely on the
+    // conservative fallback below (issue #1526, suggested fix #1).
     try {
-      const resp = await fetch('/api/v1/status');
+      const resp = await fetch('/api/v1/status', { headers: apiService.getHeaders() });
       if (resp.ok) {
         const json = await resp.json();
         this._applyServerSource(json.source, json.source_state);

v2/Cargo.lock

@@ -7918,6 +7918,7 @@ dependencies = [
  "ruview-attest",
  "ruview-evidence",
  "ruview-ontology",
+ "ruview-ood",
  "serde",
  "serde_json",
  "thiserror 2.0.18",
@@ -8034,7 +8035,11 @@ dependencies = [
 name = "ruview-policy"
 version = "0.3.1"
 dependencies = [
+ "ruview-attest",
+ "ruview-certify",
  "ruview-evidence",
+ "ruview-ontology",
+ "ruview-ood",
  "serde",
  "serde_json",
  "thiserror 2.0.18",
@@ -11788,6 +11793,10 @@ dependencies = [
  "tower-http",
 ]
 
+[[package]]
+name = "wifi-densepose-privshield"
+version = "0.1.0"
+
 [[package]]
 name = "wifi-densepose-rufield"
 version = "0.3.0"

v2/crates/ruview-certify/Cargo.toml

@@ -13,6 +13,7 @@ blake3 = { version = "1.5", default-features = false }
 ruview-ontology = { path = "../ruview-ontology" }
 ruview-attest = { path = "../ruview-attest" }
 ruview-evidence = { path = "../ruview-evidence" }
+ruview-ood = { path = "../ruview-ood" }
 wifi-densepose-calibration = { path = "../wifi-densepose-calibration", default-features = false }
 
 [dev-dependencies]

v2/crates/ruview-certify/src/lib.rs

@@ -106,6 +106,59 @@ impl DomainState {
     }
 }
 
+/// The adapter boundary from the real `ruview-ood` gate result to this
+/// crate's three-way domain signature (ADR-297/302). `ruview_ood::DomainState`
+/// carries a `DomainCause` on `Degraded`/`Unknown`; `is_valid` only needs the
+/// three-way outcome, so the cause is dropped here — this is the conversion
+/// this crate's own `DomainState` doc comment already described but that
+/// previously did not exist anywhere, leaving `ruview-ood`'s live drift result
+/// with no path into a certificate check.
+impl From<ruview_ood::DomainState> for DomainState {
+    fn from(state: ruview_ood::DomainState) -> Self {
+        match state {
+            ruview_ood::DomainState::Known => DomainState::Known,
+            ruview_ood::DomainState::Degraded(_) => DomainState::Degraded,
+            ruview_ood::DomainState::Unknown(_) => DomainState::Unknown,
+        }
+    }
+}
+
+#[cfg(test)]
+mod domain_state_adapter_tests {
+    //! Pins the `ruview-ood` -> `ruview-certify` adapter boundary: a real OOD
+    //! gate result must map to the matching certify-side outcome, and — the
+    //! load-bearing case — a post-drift `Unknown` from `ood` must actually
+    //! invalidate an otherwise-valid certificate through this conversion,
+    //! not just when a test hand-constructs `certify::DomainState::Unknown`
+    //! directly.
+    use super::DomainState;
+
+    #[test]
+    fn known_maps_to_known() {
+        assert_eq!(DomainState::from(ruview_ood::DomainState::Known), DomainState::Known);
+    }
+
+    #[test]
+    fn degraded_drops_the_cause_and_maps_to_degraded() {
+        assert_eq!(
+            DomainState::from(ruview_ood::DomainState::Degraded(
+                ruview_ood::DomainCause::ModerateDrift
+            )),
+            DomainState::Degraded
+        );
+    }
+
+    #[test]
+    fn unknown_drops_the_cause_and_maps_to_unknown() {
+        assert_eq!(
+            DomainState::from(ruview_ood::DomainState::Unknown(
+                ruview_ood::DomainCause::DriftBeyondEnvelope
+            )),
+            DomainState::Unknown
+        );
+    }
+}
+
 /// The operating metrics frozen onto a certificate, sliced from the ADR-304
 /// ledger for one exact context (never a global average).
 ///

v2/crates/ruview-policy/Cargo.toml

@@ -10,6 +10,10 @@ repository.workspace = true
 thiserror.workspace = true
 serde = { workspace = true, features = ["derive"] }
 ruview-evidence = { path = "../ruview-evidence" }
+ruview-ood = { path = "../ruview-ood" }
+ruview-certify = { path = "../ruview-certify" }
+ruview-attest = { path = "../ruview-attest" }
 
 [dev-dependencies]
 serde_json.workspace = true
+ruview-ontology = { path = "../ruview-ontology" }

v2/crates/ruview-policy/src/lib.rs

@@ -433,6 +433,70 @@ pub fn authorize_with(
     }
 }
 
+// ---------------------------------------------------------------------------
+// The real adapter (ADR-297/318/321) — ruview-ood + ruview-certify -> here.
+// ---------------------------------------------------------------------------
+
+/// The adapter boundary from a real `ruview-ood` gate result to this crate's
+/// three-way domain signature, exactly as the crate-level doc comment
+/// prescribes: the cause carried by `Degraded`/`Unknown` is dropped, since
+/// [`authorize`] only needs the three-way outcome.
+impl From<ruview_ood::DomainState> for DomainState {
+    fn from(state: ruview_ood::DomainState) -> Self {
+        match state {
+            ruview_ood::DomainState::Known => DomainState::Known,
+            ruview_ood::DomainState::Degraded(_) => DomainState::Degraded,
+            ruview_ood::DomainState::Unknown(_) => DomainState::Unknown,
+        }
+    }
+}
+
+/// Authorize an action from a **real** [`ruview_certify::CapabilityCertificate`]
+/// and a **real** [`ruview_ood::DomainState`], instead of a hand-built
+/// [`AssuranceInputs`].
+///
+/// This is the composition the crate-level "Adapter note" describes but that,
+/// before this function existed, no code in the workspace actually performed:
+/// `ruview-policy` never depended on `ruview-certify`/`ruview-ood`, so a real
+/// OOD drift result had no path into a real `authorize()` call — only two
+/// disconnected unit tests, each hand-setting the same enum value on its own
+/// crate's local type, exercised the *shape* of the story.
+///
+/// Deliberately **not** `cert.is_valid(now, domain)`: that folds the domain
+/// check into `certificate_valid`, which would attribute a domain-caused deny
+/// to a generic "certificate invalid" instead of
+/// [`FailedCondition::DomainNotKnown`]/[`FailedCondition::DomainDegraded`].
+/// Time+signature and domain are validated separately here, per the crate's
+/// own documented contract, and `authorize` (pure, clock-free) does the rest.
+#[must_use]
+pub fn authorize_from_certificate<V: ruview_attest::Verifier + ?Sized>(
+    class: ActionClass,
+    cert: &ruview_certify::CapabilityCertificate,
+    verifier: &V,
+    now_unix_s: i64,
+    domain: ruview_ood::DomainState,
+    certificate_class: CertificateClass,
+    uncertainty: f64,
+    evidence_level: EvidenceLevel,
+) -> Authorization {
+    let certificate_valid =
+        cert.verify(verifier) && now_unix_s < cert.content.valid_until_unix_s;
+    let certificate_age_secs =
+        (now_unix_s - cert.content.calibrated_date_unix_s).max(0) as u64;
+
+    authorize(
+        class,
+        &AssuranceInputs {
+            certificate_class,
+            certificate_valid,
+            certificate_age_secs,
+            domain_state: DomainState::from(domain),
+            uncertainty,
+            evidence_level,
+        },
+    )
+}
+
 // ---------------------------------------------------------------------------
 // Tests — all fixtures are SYNTHETIC / L0 (CLAUDE.md honesty rule).
 // ---------------------------------------------------------------------------
@@ -751,3 +815,146 @@ mod tests {
         assert_eq!(decision, back);
     }
 }
+
+// ---------------------------------------------------------------------------
+// Real cross-crate integration test (ADR-297/300/318/321 acceptance test B).
+//
+// The PR that introduced ruview-ood/-certify/-policy claimed this exact
+// scenario — "a post-drift UNKNOWN domain invalidates the capability
+// certificate and denies a SafetyCritical action" — as a load-bearing
+// acceptance test. What existed were two disconnected unit tests in two
+// crates that had no dependency on each other, each hand-setting the same
+// enum value on its own crate's *local* type. Neither exercised
+// `authorize_from_certificate`, because it didn't exist, and `ruview-policy`
+// didn't depend on `ruview-certify`/`ruview-ood` at all.
+//
+// This test mints a REAL signed `CapabilityCertificate` (ruview-certify) and
+// drives `authorize_from_certificate` (this crate) with a REAL
+// `ruview_ood::DomainState`, so the drift-invalidates-capability claim is
+// backed by one composed pipeline, not two same-shaped unit tests.
+// ---------------------------------------------------------------------------
+#[cfg(test)]
+mod acceptance_test_b_real_integration {
+    use super::*;
+    use ruview_attest::{Blake3MacSigner, DeviceId};
+    use ruview_certify::{Capability, CertificateContent, OperatingMetrics};
+    use ruview_evidence::EvidenceLevel;
+    use ruview_ontology::SpaceId;
+
+    const CALIBRATED_AT: i64 = 1_000;
+    const VALID_UNTIL: i64 = 5_000;
+
+    /// A real, signed `CapabilityCertificate` — hand-assembled from public
+    /// types rather than going through `ruview_certify::mint`'s full
+    /// calibration/evidence-ledger pipeline (already covered by that crate's
+    /// own test suite). What this test is pinning is the *adapter*, not
+    /// certify's minting validation.
+    fn real_signed_certificate() -> (ruview_certify::CapabilityCertificate, Blake3MacSigner) {
+        let signer = Blake3MacSigner::new([9u8; 32]);
+        let content = CertificateContent {
+            capability: Capability::Presence,
+            room: SpaceId::new("kitchen").unwrap(),
+            calibration_version: 1,
+            calibration_expires_at_unix_s: VALID_UNTIL + 1,
+            hardware: DeviceId::new("dev-1").unwrap(),
+            model_version: "m-1".to_string(),
+            calibrated_date_unix_s: CALIBRATED_AT,
+            metrics: OperatingMetrics {
+                moving_recall: 0.9,
+                stationary_recall: 0.85,
+                false_presence_per_24h: 0.1,
+            },
+            valid_until_unix_s: VALID_UNTIL,
+            evidence_level: EvidenceLevel::L0,
+        };
+        let signature = ruview_attest::Signer::sign(&signer, &content.canonical_bytes());
+        let cert = ruview_certify::CapabilityCertificate {
+            content,
+            signature: Some(signature),
+        };
+        (cert, signer)
+    }
+
+    #[test]
+    fn known_domain_authorizes_safety_critical() {
+        let (cert, signer) = real_signed_certificate();
+        let decision = authorize_from_certificate(
+            ActionClass::SafetyCritical,
+            &cert,
+            &signer,
+            CALIBRATED_AT + 10, // well inside validity
+            ruview_ood::DomainState::Known,
+            CertificateClass::High,
+            0.05,
+            EvidenceLevel::L3,
+        );
+        assert!(decision.is_allowed(), "{decision:?}");
+    }
+
+    /// The acceptance-test-B claim, for real: the SAME certificate that just
+    /// authorized a SafetyCritical action, re-evaluated with nothing changed
+    /// except a real post-drift `ruview_ood::DomainState::Unknown`, denies —
+    /// through the actual ood -> certify-adapter -> policy composition, not a
+    /// hand-set field on a local type.
+    #[test]
+    fn post_drift_unknown_domain_denies_safety_critical() {
+        let (cert, signer) = real_signed_certificate();
+        let now = CALIBRATED_AT + 10;
+
+        let pre_drift = authorize_from_certificate(
+            ActionClass::SafetyCritical,
+            &cert,
+            &signer,
+            now,
+            ruview_ood::DomainState::Known,
+            CertificateClass::High,
+            0.05,
+            EvidenceLevel::L3,
+        );
+        assert!(pre_drift.is_allowed(), "pre-drift: {pre_drift:?}");
+
+        let post_drift = authorize_from_certificate(
+            ActionClass::SafetyCritical,
+            &cert,
+            &signer,
+            now,
+            ruview_ood::DomainState::Unknown(ruview_ood::DomainCause::DriftBeyondEnvelope),
+            CertificateClass::High,
+            0.05,
+            EvidenceLevel::L3,
+        );
+        assert_eq!(
+            post_drift,
+            Authorization::Deny {
+                failed_condition: FailedCondition::DomainNotKnown,
+            },
+            "a real post-drift UNKNOWN domain must deny the same certificate \
+             that was just valid, before a false-confident inference reaches \
+             an actuator",
+        );
+    }
+
+    #[test]
+    fn tampered_certificate_is_rejected_regardless_of_domain() {
+        let (cert, _signer) = real_signed_certificate();
+        // A different signer's key cannot verify this certificate's tag —
+        // exercises the "real verify(), not is_valid()" half of the adapter.
+        let wrong_signer = Blake3MacSigner::new([0xAAu8; 32]);
+        let decision = authorize_from_certificate(
+            ActionClass::Convenience,
+            &cert,
+            &wrong_signer,
+            CALIBRATED_AT + 10,
+            ruview_ood::DomainState::Known,
+            CertificateClass::Basic,
+            0.05,
+            EvidenceLevel::L0,
+        );
+        assert_eq!(
+            decision,
+            Authorization::Deny {
+                failed_condition: FailedCondition::CertificateInvalid,
+            },
+        );
+    }
+}

v2/crates/wifi-densepose-sensing-server/src/main.rs

@@ -422,7 +422,7 @@ struct FeatureInfo {
     spectral_power: f64,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 struct ClassificationInfo {
     motion_level: String,
     presence: bool,
@@ -451,6 +451,25 @@ fn classify_vitals(motion: bool, presence: bool, presence_score: f32) -> Classif
     }
 }
 
+/// Derive the top-level (room) [`ClassificationInfo`] from the fused
+/// [`RoomInference`] aggregate (ADR-297, issue #1554), rather than from
+/// whichever node's packet happened to arrive last. `RoomInference`'s
+/// `"unavailable"` (no fresh node contributing) maps to `"absent"` with zero
+/// confidence — no evidence of presence, since `ClassificationInfo` has no
+/// separate "unknown" state — never a frozen last-known value.
+fn classification_from_room(room: &RoomInference) -> ClassificationInfo {
+    let motion_level = if room.classification == "unavailable" {
+        "absent"
+    } else {
+        room.classification.as_str()
+    };
+    ClassificationInfo {
+        motion_level: motion_level.to_string(),
+        presence: motion_level != "absent",
+        confidence: room.confidence,
+    }
+}
+
 /// ADR-297 — the window a node may be silent before it stops contributing to
 /// the fused room aggregate (its entities go stale/unavailable rather than
 /// holding a frozen online value). Mirrors the 10 s active-node filter used to
@@ -505,6 +524,73 @@ mod classify_vitals_tests {
     }
 }
 
+#[cfg(test)]
+mod issue_1554_room_classification_tests {
+    //! Issue #1554 — the top-level `classification` in `SensingUpdate` (served
+    //! by `GET /api/v1/sensing/latest`) used to be `classify_vitals(...)` on
+    //! *this packet's* single node, overwritten on every UDP packet. With 2+
+    //! disagreeing nodes it flapped at packet rate (~40/s in the field report)
+    //! because whichever node's packet arrived last won.
+    //!
+    //! `classification_from_room` instead derives the top-level classification
+    //! from the deterministic, freshness-weighted `RoomInference` aggregate
+    //! (ADR-297) — the same aggregate `fuse_room` already computes for the
+    //! `room_inference` field — so it no longer depends on packet arrival
+    //! order.
+    use super::{classification_from_room, RoomInference};
+
+    #[test]
+    fn derives_presence_and_motion_from_room_classification() {
+        let room = RoomInference {
+            classification: "present_moving".to_string(),
+            confidence: 0.82,
+            contributing_nodes: 3,
+        };
+        let c = classification_from_room(&room);
+        assert_eq!(c.motion_level, "present_moving");
+        assert!(c.presence);
+        assert!((c.confidence - 0.82).abs() < 1e-9);
+    }
+
+    #[test]
+    fn absent_room_classification_has_no_presence() {
+        let room = RoomInference {
+            classification: "absent".to_string(),
+            confidence: 0.4,
+            contributing_nodes: 1,
+        };
+        let c = classification_from_room(&room);
+        assert_eq!(c.motion_level, "absent");
+        assert!(!c.presence);
+    }
+
+    #[test]
+    fn unavailable_room_maps_to_absent_not_a_frozen_value() {
+        // No fresh node contributing -> RoomInference::unavailable(). Must not
+        // surface as a stale "present" from whichever node reported last.
+        let room = RoomInference::unavailable();
+        let c = classification_from_room(&room);
+        assert_eq!(c.motion_level, "absent");
+        assert!(!c.presence);
+        assert_eq!(c.confidence, 0.0);
+    }
+
+    #[test]
+    fn does_not_depend_on_which_node_is_named_last() {
+        // The old bug was order/arrival dependent. The room-derived value must
+        // be a pure function of the aggregate, so two RoomInferences with the
+        // same fields (regardless of which node contributed them) classify
+        // identically.
+        let a = RoomInference {
+            classification: "present_still".to_string(),
+            confidence: 0.6,
+            contributing_nodes: 2,
+        };
+        let b = a.clone();
+        assert_eq!(classification_from_room(&a), classification_from_room(&b));
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 struct SignalField {
     grid_size: [usize; 3],
@@ -4389,7 +4475,15 @@ fn derive_single_person_pose(
                 x: final_x,
                 y: final_y,
                 z: lean_x * 0.02,
-                confidence: kp_conf.clamp(0.1, 1.0),
+                // Issue #1525: the UI's default `keypointConfidenceThreshold`
+                // is 0.1 (`ui/utils/pose-renderer.js`), and every draw gate
+                // there rejects at `<=`/requires `>` that value — so a floor
+                // of exactly 0.1 still renders nothing on the default,
+                // no-model Docker path (low `base_confidence` hits this floor
+                // often). Clamp strictly above the client's threshold so a
+                // signal-derived keypoint is always visible, never fully
+                // transparent/undrawn by construction.
+                confidence: kp_conf.clamp(0.15, 1.0),
             }
         })
         .collect();
@@ -5023,6 +5117,17 @@ async fn start_recording(
         .map(|s| s.to_string())
         .unwrap_or_else(|| format!("rec_{}", chrono_timestamp()));
 
+    // ADR-295: a recording captured while the live source is `Synthetic` is an
+    // export, and every export of synthetic data must be watermarked so it can
+    // never later be mistaken for a real capture (`export_watermark`). Stamped
+    // onto the recording's own metadata entry — not the filename or the
+    // per-line JSON — so the on-disk `.jsonl` schema and the `{id}.jsonl` path
+    // convention `delete_recording`/`scan_recording_files` both rely on stay
+    // exactly as every existing consumer (including `wifi-densepose-train`'s
+    // dataset loader) already expects. Still unmissable: every caller of
+    // `GET /api/v1/recordings` (and the success response below) sees it.
+    let watermark = s.source_state().export_watermark();
+
     // Create the recording file
     let rec_path = PathBuf::from("data/recordings").join(format!("{}.jsonl", id));
     let file = match std::fs::File::create(&rec_path) {
@@ -5051,6 +5156,7 @@ async fn start_recording(
         "status": "recording",
         "started_at": chrono_timestamp(),
         "frames": 0,
+        "watermark": watermark,
     }));
 
     let rec_id = id.clone();
@@ -5096,8 +5202,11 @@ async fn start_recording(
         info!("Recording {rec_id} finished: {frame_count} frames written");
     });
 
-    info!("Recording started: {id}");
-    Json(serde_json::json!({ "success": true, "recording_id": id }))
+    match watermark {
+        Some(mark) => info!("Recording started: {id} (source watermarked {mark})"),
+        None => info!("Recording started: {id}"),
+    }
+    Json(serde_json::json!({ "success": true, "recording_id": id, "watermark": watermark }))
 }
 
 /// POST /api/v1/recording/stop — stop recording CSI data.
@@ -6072,23 +6181,12 @@ async fn udp_receiver_task(
                     // Cross-node fusion: combine features from all active nodes.
                     let fused_features = fuse_multi_node_features(&features, &s.node_states);
 
-                    let mut classification =
-                        classify_vitals(vitals.motion, vitals.presence, vitals.presence_score);
-
-                    // Boost classification confidence with multi-node coverage.
-                    let n_active = s
-                        .node_states
-                        .values()
-                        .filter(|ns| {
-                            ns.last_frame_time
-                                .is_some_and(|t| now.duration_since(t).as_secs() < 10)
-                        })
-                        .count();
-                    if n_active > 1 {
-                        classification.confidence = (classification.confidence
-                            * (1.0 + 0.15 * (n_active as f64 - 1.0)))
-                            .clamp(0.0, 1.0);
-                    }
+                    // ADR-297 (issue #1554): the top-level classification is the
+                    // fused room aggregate, not this packet's single node — a
+                    // node's own reading no longer overwrites the room's. The
+                    // old ad-hoc "boost confidence by node count" is replaced by
+                    // `room_inference`'s freshness-weighted multi-node confidence.
+                    let classification = classification_from_room(&room_inference);
 
                     let signal_field = generate_signal_field(
                         fused_features.mean_rssi,
@@ -6560,7 +6658,12 @@ async fn udp_receiver_task(
                         tick,
                         nodes: active_nodes,
                         features: fused_features.clone(),
-                        classification,
+                        // ADR-297 (issue #1554): top-level classification is the
+                        // fused room aggregate, not this frame's single node.
+                        // `classification` (this node's own smoothed reading)
+                        // still drives `motion_score`/`total_persons` above,
+                        // which are legitimately this-packet-local.
+                        classification: classification_from_room(&room_inference),
                         signal_field: generate_signal_field(
                             fused_features.mean_rssi,
                             motion_score,
@@ -6977,12 +7080,20 @@ fn vitals_snapshots_from_sensing_json(
             .iter()
             .map(|node| {
                 let n = node["node_id"].as_u64().unwrap_or(0);
-                // Each node carries its OWN classification — use it, deferring to
-                // the room aggregate only for fields the node omits.
-                let ncls = &node["classification"];
-                let presence = ncls["presence"].as_bool().unwrap_or(agg_presence);
-                let motion = motion_of(ncls["motion_level"].as_str(), agg_motion);
-                let conf = ncls["confidence"].as_f64().unwrap_or(agg_conf);
+                // Each node carries its OWN classification under `node_inference`
+                // (ADR-297) — use it, deferring to the room aggregate only for
+                // fields the node omits. Issue #1541: this previously read a
+                // `"classification"` key that does not exist on `NodeInfo`'s
+                // serialized JSON (the field is `node_inference`), so every
+                // per-node lookup silently fell through to the room aggregate —
+                // per-node MQTT topics carried array-global values.
+                let ninf = &node["node_inference"];
+                let presence = ninf["classification"]
+                    .as_str()
+                    .map(|c| c != "absent")
+                    .unwrap_or(agg_presence);
+                let motion = motion_of(ninf["classification"].as_str(), agg_motion);
+                let conf = ninf["confidence"].as_f64().unwrap_or(agg_conf);
                 mk(
                     format!("{base_id}-node{n}"),
                     presence,
@@ -9162,10 +9273,21 @@ mod mqtt_bridge_tests {
     use super::vitals_snapshots_from_sensing_json;
     use serde_json::json;
 
-    /// Regression for the per-node presence bug (#872/#898): each node must
-    /// surface its OWN classification, not the room-level aggregate. Node 1 is
-    /// present+moving; node 2 is absent — node 2 must NOT inherit node 1's
-    /// "present".
+    /// Regression for the per-node presence bug (#872/#898, and its
+    /// resurgence as #1541): each node must surface its OWN classification,
+    /// not the room-level aggregate. Node 1 is present+moving; node 2 is
+    /// absent — node 2 must NOT inherit node 1's "present".
+    ///
+    /// The fixture below uses `nodes[].node_inference.classification` — the
+    /// field `NodeInfo` actually serializes (ADR-297) — not a bare
+    /// `nodes[].classification`. Issue #1541: an earlier version of this exact
+    /// test used the latter, non-existent shape, which the reader silently
+    /// treated as "field omitted" and fell back to the room aggregate for
+    /// every node. The test therefore passed while the real per-node MQTT
+    /// output was array-global — 100% line coverage of
+    /// `vitals_snapshots_from_sensing_json` with a fixture that didn't match
+    /// what `NodeInfo` actually serializes. Keep this fixture in the real
+    /// shape so this can't recur silently.
     #[test]
     fn per_node_presence_uses_each_nodes_own_classification() {
         let v = json!({
@@ -9175,9 +9297,9 @@ mod mqtt_bridge_tests {
             "persons": [{}, {}],
             "nodes": [
                 { "node_id": 1, "rssi_dbm": -40.0,
-                  "classification": { "presence": true, "motion_level": "walking", "confidence": 0.8 } },
+                  "node_inference": { "classification": "present_moving", "confidence": 0.8 } },
                 { "node_id": 2, "rssi_dbm": -70.0,
-                  "classification": { "presence": false, "motion_level": "absent", "confidence": 0.1 } }
+                  "node_inference": { "classification": "absent", "confidence": 0.1 } }
             ]
         });
         let snaps = vitals_snapshots_from_sensing_json(&v, "ruview");

v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs

@@ -30,7 +30,7 @@
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
-use rumqttc::{AsyncClient, ClientError, EventLoop, MqttOptions, QoS, Transport};
+use rumqttc::{AsyncClient, ClientError, EventLoop, MqttOptions, QoS, Transport, TlsConfiguration};
 use tokio::sync::broadcast;
 use tokio::task::JoinHandle;
 use tracing::{error, info, warn};
@@ -62,6 +62,12 @@ use super::state::{RateLimiter, StateEncoder, StateMessage, VitalsSnapshot};
 /// Heartbeat cadence for availability re-publication (per §3.6).
 const AVAILABILITY_HEARTBEAT: Duration = Duration::from_secs(30);
 
+/// A node whose broadcast snapshot hasn't arrived within this window is
+/// treated as stale for the availability heartbeat, not just "quiet" (issue
+/// #1555). Matches `NODE_STALE_AFTER_MS` in `main.rs`'s room-fusion staleness
+/// window, so "stale" means the same thing on the MQTT and WebSocket paths.
+const NODE_SNAPSHOT_STALE_AFTER: Duration = Duration::from_secs(10);
+
 /// Build a `rumqttc::MqttOptions` from validated [`MqttConfig`].
 fn build_mqtt_options(cfg: &MqttConfig) -> MqttOptions {
     let mut opts = MqttOptions::new(&cfg.client_id, &cfg.host, cfg.port);
@@ -74,16 +80,69 @@ fn build_mqtt_options(cfg: &MqttConfig) -> MqttOptions {
         opts.set_credentials(u, "");
     }
 
-    if !matches!(cfg.tls, TlsConfig::Off) {
-        // We always use rustls (matches `ureq` in this crate). The
-        // specific cert / CA wiring is done by the runtime constructor;
-        // here we just flip the transport.
-        opts.set_transport(Transport::tls_with_default_config());
-    }
+    opts.set_transport(build_transport(&cfg.tls));
 
     opts
 }
 
+/// Build the `rumqttc::Transport` for the configured [`TlsConfig`].
+///
+/// Issue #1556: `PinnedCa`/`MutualTls` were parsed from the CLI and stored,
+/// but this function used to ignore them entirely and always call
+/// `Transport::tls_with_default_config()` — the system trust store only, so
+/// a self-signed broker (the common case for a home-LAN Mosquitto add-on)
+/// always failed with `UnknownIssuer`. `TlsConfiguration::Simple` (rumqttc's
+/// own pinned-CA / mTLS variant — see `rumqttc::tls::rustls_connector`) takes
+/// raw PEM bytes directly, so no new TLS dependency is needed here.
+///
+/// A CA/cert/key file that can't be read falls back to the system trust
+/// store (same behavior as today, i.e. still fails `UnknownIssuer` against a
+/// self-signed broker) rather than panicking this background task — but now
+/// logs *why*, which issue #1556 also asked for.
+fn build_transport(tls: &TlsConfig) -> Transport {
+    let read_pem = |label: &str, path: &std::path::Path| -> Option<Vec<u8>> {
+        match std::fs::read(path) {
+            Ok(bytes) => Some(bytes),
+            Err(e) => {
+                let path = path.display();
+                otel_error!(
+                    "[mqtt] tls: could not read {label} file {path}: {e} — falling back to \
+                     the system trust store (issue #1556)"
+                );
+                None
+            }
+        }
+    };
+
+    match tls {
+        TlsConfig::Off => return Transport::Tcp,
+        TlsConfig::SystemTrust => {}
+        TlsConfig::PinnedCa { ca_file } => {
+            if let Some(ca) = read_pem("CA", ca_file) {
+                return Transport::tls_with_config(TlsConfiguration::Simple {
+                    ca,
+                    alpn: None,
+                    client_auth: None,
+                });
+            }
+        }
+        TlsConfig::MutualTls { ca_file, client_cert, client_key } => {
+            if let (Some(ca), Some(cert), Some(key)) = (
+                read_pem("CA", ca_file),
+                read_pem("client cert", client_cert),
+                read_pem("client key", client_key),
+            ) {
+                return Transport::tls_with_config(TlsConfiguration::Simple {
+                    ca,
+                    alpn: None,
+                    client_auth: Some((cert, key)),
+                });
+            }
+        }
+    }
+    Transport::tls_with_default_config()
+}
+
 /// One node's per-entity availability topics, pre-computed at startup so
 /// the heartbeat loop doesn't allocate per tick.
 struct NodeAvailability {
@@ -177,8 +236,14 @@ async fn run(
     // each node's builder + availability are retained here for heartbeats and
     // the offline LWT. (Previously a single hard-coded builder collapsed every
     // node into one device.)
-    let mut nodes: std::collections::HashMap<String, (OwnedDiscoveryBuilder, NodeAvailability)> =
-        std::collections::HashMap::new();
+    // Issue #1555: the third tuple element is the Instant this node's last
+    // broadcast snapshot arrived, so the heartbeat below can tell a node that
+    // has genuinely gone quiet from one that's just between publish-rate
+    // ticks, and stop asserting "online" for it.
+    let mut nodes: std::collections::HashMap<
+        String,
+        (OwnedDiscoveryBuilder, NodeAvailability, Instant),
+    > = std::collections::HashMap::new();
 
     let mut rate_limiter = RateLimiter::new();
     let mut last_heartbeat = Instant::now();
@@ -215,15 +280,26 @@ async fn run(
             // Periodic heartbeat / discovery refresh.
             _ = tokio::time::sleep(Duration::from_secs(1)) => {
                 if last_heartbeat.elapsed() >= AVAILABILITY_HEARTBEAT {
-                    for (_, na) in nodes.values() {
-                        if let Err(e) = publish_availability(&client, na, "online").await {
-                            otel_warn!("[mqtt] heartbeat publish failed: {e}");
+                    for (node_id, (_, na, last_seen)) in &nodes {
+                        // Issue #1555: a node whose snapshots have actually
+                        // stopped arriving must go `offline`, not keep
+                        // reporting `online` on a fixed timer regardless of
+                        // whether its data is still flowing — a frozen HA
+                        // entity that still shows "available" is worse than
+                        // one correctly marked unavailable.
+                        let state = if last_seen.elapsed() < NODE_SNAPSHOT_STALE_AFTER {
+                            "online"
+                        } else {
+                            "offline"
+                        };
+                        if let Err(e) = publish_availability(&client, na, state).await {
+                            otel_warn!("[mqtt] heartbeat publish failed for node {node_id}: {e}");
                         }
                     }
                     last_heartbeat = Instant::now();
                 }
                 if last_refresh.elapsed() >= Duration::from_secs(cfg.refresh_secs) {
-                    for (nb, _) in nodes.values() {
+                    for (nb, _, _) in nodes.values() {
                         if let Err(e) =
                             publish_all_discovery(&client, &nb.as_borrowed(), &entities).await
                         {
@@ -239,6 +315,7 @@ async fn run(
                 match recv {
                     Ok(snap) => {
                         let elapsed = start_instant.elapsed();
+                        let now = Instant::now();
                         // #898: on first sight of a node_id, publish that
                         // node's discovery + availability; then route its
                         // state to per-node topics.
@@ -254,7 +331,12 @@ async fn run(
                             if let Err(e) = publish_availability(&client, &na, "online").await {
                                 otel_warn!("[mqtt] node {} availability failed: {e}", snap.node_id);
                             }
-                            nodes.insert(snap.node_id.clone(), (nb, na));
+                            nodes.insert(snap.node_id.clone(), (nb, na, now));
+                        } else if let Some(entry) = nodes.get_mut(&snap.node_id) {
+                            // Issue #1555: record that this node is still
+                            // alive so the heartbeat above doesn't have to
+                            // guess from a fixed timer.
+                            entry.2 = now;
                         }
                         let borrowed = nodes[&snap.node_id].0.as_borrowed();
                         publish_snapshot(&client, &borrowed, &snap, &cfg, &mut rate_limiter, elapsed).await;
@@ -265,7 +347,7 @@ async fn run(
                     Err(broadcast::error::RecvError::Closed) => {
                         info!("[mqtt] broadcast channel closed, draining");
                         // Publish offline for every known node before exit.
-                        for (_, na) in nodes.values() {
+                        for (_, na, _) in nodes.values() {
                             let _ = publish_availability(&client, na, "offline").await;
                         }
                         let _ = client.disconnect().await;