diff --git a/.github/workflows/model-release-gate.yml b/.github/workflows/model-release-gate.yml new file mode 100644 index 00000000..f0ee4f81 --- /dev/null +++ b/.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" diff --git a/README.md b/README.md index c3ac6f0d..3a2cc6b4 100644 --- a/README.md +++ b/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 ` flag | native RVF, `model.safetensors`, or `model.rvf.jsonl` | ✅ Native RVF loads directly; safetensors and JSONL auto-convert in memory | diff --git a/docs/user-guide.md b/docs/user-guide.md index de364df2..9b1a51ce 100644 --- a/docs/user-guide.md +++ b/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 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` +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. --- diff --git a/ui/pose-fusion/js/main.js b/ui/pose-fusion/js/main.js index f9dface3..e79d1366 100644 --- a/ui/pose-fusion/js/main.js +++ b/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 diff --git a/ui/services/sensing.service.js b/ui/services/sensing.service.js index 5eca2d93..57e37cdc 100644 --- a/ui/services/sensing.service.js +++ b/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); diff --git a/v2/Cargo.lock b/v2/Cargo.lock index 71e668d1..d80053d0 100644 --- a/v2/Cargo.lock +++ b/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" diff --git a/v2/crates/ruview-certify/Cargo.toml b/v2/crates/ruview-certify/Cargo.toml index c539e9c1..650efb74 100644 --- a/v2/crates/ruview-certify/Cargo.toml +++ b/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] diff --git a/v2/crates/ruview-certify/src/lib.rs b/v2/crates/ruview-certify/src/lib.rs index 1f4d387b..881b5e18 100644 --- a/v2/crates/ruview-certify/src/lib.rs +++ b/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 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). /// diff --git a/v2/crates/ruview-policy/Cargo.toml b/v2/crates/ruview-policy/Cargo.toml index a93c4d97..54abe5a9 100644 --- a/v2/crates/ruview-policy/Cargo.toml +++ b/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" } diff --git a/v2/crates/ruview-policy/src/lib.rs b/v2/crates/ruview-policy/src/lib.rs index 1d000b7c..290b4aac 100644 --- a/v2/crates/ruview-policy/src/lib.rs +++ b/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 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( + 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, + }, + ); + } +} diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 5893df17..7911a70c 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/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"); diff --git a/v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs b/v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs index 769cc0e4..fd670e05 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/mqtt/publisher.rs +++ b/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> { + 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 = - 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;