470 lines
20 KiB
Rust
470 lines
20 KiB
Rust
//! Live trust-path bridge: drive the governed [`StreamingEngine`] from the
|
|
//! sensing-server's live `NodeState` map.
|
|
//!
|
|
//! `multistatic_bridge.rs` already converts `NodeState` → `MultiBandCsiFrame`
|
|
//! and runs the *bare* `MultistaticFuser`. That path produces fused amplitudes
|
|
//! but skips the trust control plane: privacy demotion on contradiction, the
|
|
//! WorldGraph belief with mandatory provenance, and the deterministic witness
|
|
//! (ADR-135..146). This bridge routes the same live frames through
|
|
//! [`StreamingEngine::process_cycle`], so every governed belief carries
|
|
//! evidence + model + calibration + privacy decision and a BLAKE3 witness
|
|
//! (narrowing the gap called out in ADR-136 §8 and the beyond-SOTA system
|
|
//! review).
|
|
//!
|
|
//! ## Honest scope of the live-path governance
|
|
//!
|
|
//! The engine runs *alongside* the bare fusion path that feeds the live
|
|
//! `SensingUpdate`; it does not replace it. What the engine's decision **does**
|
|
//! gate on the live wire today: when a cycle is emitted at
|
|
//! [`PrivacyClass::Restricted`] (base mode or contradiction/mesh-risk
|
|
//! demotion), [`EngineBridge::suppress_raw_outputs`] is true and `main.rs`
|
|
//! strips the per-node raw amplitude vectors from the published update — the
|
|
//! same field mapping `wifi-densepose-bfld`'s privacy gate applies at
|
|
//! `Restricted` (drop amplitude/phase proxies). Trust state (latest witness,
|
|
//! effective class, recalibration flag, engine-error count) is readable on
|
|
//! `GET /api/v1/status`. Gating of the remaining *derived* outputs
|
|
//! (person count, classification, signal field) by privacy class is tracked
|
|
//! as a follow-up; until then those fields are published ungoverned.
|
|
//!
|
|
//! Determinism: this module reads server state and forwards explicit
|
|
//! timestamps/calibration ids; it introduces no wall-clock reads of its own, so
|
|
//! a given `(frames, calibration, now_ms)` always yields the same
|
|
//! [`TrustedOutput`] witness.
|
|
|
|
use std::collections::HashMap;
|
|
use std::time::{Duration, Instant};
|
|
|
|
use wifi_densepose_bfld::{PrivacyClass, PrivacyMode};
|
|
use wifi_densepose_engine::{AdapterInfo, EngineError, StreamingEngine, TrustedOutput};
|
|
use wifi_densepose_geo::types::GeoRegistration;
|
|
use wifi_densepose_signal::ruvsense::fusion_quality::CalibrationId;
|
|
use wifi_densepose_worldgraph::WorldId;
|
|
|
|
use super::multistatic_bridge::node_frames_from_states;
|
|
use super::NodeState;
|
|
|
|
/// Minimum spacing between engine-error warn logs (errors are still counted
|
|
/// every cycle; only the log line is rate-limited — a 20 Hz loop must not
|
|
/// emit 20 warns/s).
|
|
const ENGINE_ERROR_WARN_INTERVAL: Duration = Duration::from_secs(10);
|
|
|
|
/// Owns a [`StreamingEngine`] and the WorldGraph scope (one room + sensor) the
|
|
/// live sensing loop publishes beliefs into.
|
|
pub struct EngineBridge {
|
|
engine: StreamingEngine,
|
|
room: WorldId,
|
|
/// Nodes already wired into the WorldGraph as sensors (by `node_id`).
|
|
registered_nodes: HashMap<u8, WorldId>,
|
|
/// Calibration epoch applied to live frames until the ADR-135 baseline
|
|
/// stage supplies a real per-node id. Stable so witnesses are reproducible.
|
|
calibration: CalibrationId,
|
|
// ── Trust state observed from the most recent cycles (review finding 1:
|
|
// previously write-only fields on AppState; now recorded here and
|
|
// exposed via the status endpoint + output gating). ──────────────────
|
|
/// BLAKE3 witness of the most recent successful governed cycle.
|
|
last_witness: Option<[u8; 32]>,
|
|
/// Latest drift→recalibration recommendation (ADR-135 → ADR-150 §3.4).
|
|
recalibration_recommended: bool,
|
|
/// Privacy class the most recent cycle was emitted under (post-demotion).
|
|
effective_class: Option<PrivacyClass>,
|
|
/// Whether the most recent cycle was demoted (contradiction / mesh risk).
|
|
demoted: bool,
|
|
/// Total engine cycles that returned an error (previously swallowed by
|
|
/// `if let Some(Ok(..))` at the call sites).
|
|
engine_error_count: u64,
|
|
/// Last time an engine error was actually logged (rate limiter).
|
|
last_error_warn_at: Option<Instant>,
|
|
}
|
|
|
|
impl EngineBridge {
|
|
/// Build a bridge for one installation. `room_area_id`/`room_name` name the
|
|
/// observation scope; `mode` is the starting privacy mode.
|
|
pub fn new(mode: PrivacyMode, model_version: u16, room_area_id: &str, room_name: &str) -> Self {
|
|
let mut engine = StreamingEngine::new(mode, model_version, GeoRegistration::default());
|
|
let room = engine.add_room(room_area_id, room_name);
|
|
Self {
|
|
engine,
|
|
room,
|
|
registered_nodes: HashMap::new(),
|
|
calibration: CalibrationId(0x5256_0001), // "RV\0\x01" — placeholder epoch
|
|
last_witness: None,
|
|
recalibration_recommended: false,
|
|
effective_class: None,
|
|
demoted: false,
|
|
engine_error_count: 0,
|
|
last_error_warn_at: None,
|
|
}
|
|
}
|
|
|
|
/// Override the calibration epoch stamped onto live frames (ADR-135).
|
|
pub fn set_calibration(&mut self, calibration: CalibrationId) {
|
|
self.calibration = calibration;
|
|
}
|
|
|
|
/// Override the WorldGraph belief-retention cap (bounds memory on the live
|
|
/// loop; see `WorldGraph::prune_semantic_states`).
|
|
pub fn set_semantic_retention(&mut self, max_states: usize) {
|
|
self.engine.set_semantic_retention(max_states);
|
|
}
|
|
|
|
/// Switch the active privacy mode (operator/control-plane action).
|
|
pub fn set_privacy_mode(&mut self, mode: PrivacyMode) {
|
|
self.engine.set_privacy_mode(mode);
|
|
}
|
|
|
|
/// Activate a per-room calibration adapter (ADR-150 §3.4). The adapter's
|
|
/// content-derived id becomes part of provenance/witness from the next
|
|
/// cycle — weights can never swap silently on the live path.
|
|
pub fn set_room_adapter(&mut self, info: AdapterInfo) {
|
|
self.engine.set_room_adapter(info);
|
|
}
|
|
|
|
/// Deactivate the per-room adapter (revert to the shared base model).
|
|
pub fn clear_room_adapter(&mut self) {
|
|
self.engine.clear_room_adapter();
|
|
}
|
|
|
|
/// Borrow the engine (queries, WorldGraph snapshot, privacy audit).
|
|
pub fn engine(&self) -> &StreamingEngine {
|
|
&self.engine
|
|
}
|
|
|
|
/// Number of sensor nodes wired into the WorldGraph so far.
|
|
pub fn registered_node_count(&self) -> usize {
|
|
self.registered_nodes.len()
|
|
}
|
|
|
|
/// Run one governed trust cycle over the current live node states.
|
|
///
|
|
/// Returns `None` when no active node yields a frame (nothing to fuse —
|
|
/// the engine is not invoked, so no spurious belief is published). On a
|
|
/// real cycle it lazily wires any newly-seen node as a WorldGraph sensor,
|
|
/// then returns the witnessed [`TrustedOutput`] (or a fusion error).
|
|
///
|
|
/// `now_ms` is supplied by the caller (the sensing loop's clock), keeping
|
|
/// the bridge deterministic and replayable.
|
|
pub fn process_cycle_from_states(
|
|
&mut self,
|
|
node_states: &HashMap<u8, NodeState>,
|
|
now_ms: i64,
|
|
) -> Option<Result<TrustedOutput, EngineError>> {
|
|
let frames = node_frames_from_states(node_states);
|
|
if frames.is_empty() {
|
|
return None;
|
|
}
|
|
// Lazily register each contributing node as a sensor observing the room,
|
|
// so the privacy rollup can suppress it under identity-strict modes.
|
|
for f in &frames {
|
|
self.registered_nodes.entry(f.node_id).or_insert_with(|| {
|
|
self.engine
|
|
.add_sensor(&format!("node-{}", f.node_id), self.room)
|
|
});
|
|
}
|
|
Some(
|
|
self.engine
|
|
.process_cycle(&frames, self.calibration, self.room, now_ms),
|
|
)
|
|
}
|
|
|
|
/// Run one governed cycle **and record the trust state** (review finding
|
|
/// 1): on success the witness / effective class / demotion /
|
|
/// recalibration flag are stored for the status endpoint and output
|
|
/// gating; on error the error counter is incremented and a rate-limited
|
|
/// warning is logged (never silently swallowed). Returns the trusted
|
|
/// output on success, `None` when there was nothing to fuse or the cycle
|
|
/// errored.
|
|
pub fn observe_cycle(
|
|
&mut self,
|
|
node_states: &HashMap<u8, NodeState>,
|
|
now_ms: i64,
|
|
) -> Option<TrustedOutput> {
|
|
match self.process_cycle_from_states(node_states, now_ms)? {
|
|
Ok(trust) => {
|
|
self.last_witness = Some(trust.witness);
|
|
self.recalibration_recommended = trust.recalibration_recommended;
|
|
self.effective_class = Some(trust.effective_class);
|
|
self.demoted = trust.demoted;
|
|
Some(trust)
|
|
}
|
|
Err(e) => {
|
|
self.engine_error_count += 1;
|
|
let now = Instant::now();
|
|
let warn_due = self.last_error_warn_at.map_or(true, |t| {
|
|
now.duration_since(t) >= ENGINE_ERROR_WARN_INTERVAL
|
|
});
|
|
if warn_due {
|
|
self.last_error_warn_at = Some(now);
|
|
tracing::warn!(
|
|
total_engine_errors = self.engine_error_count,
|
|
"governed trust cycle failed (warn rate-limited to one per {:?}): {e}",
|
|
ENGINE_ERROR_WARN_INTERVAL
|
|
);
|
|
}
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
/// BLAKE3 witness of the most recent successful governed cycle.
|
|
pub fn last_trust_witness(&self) -> Option<[u8; 32]> {
|
|
self.last_witness
|
|
}
|
|
|
|
/// Latest drift→recalibration recommendation from the governed engine.
|
|
pub fn recalibration_recommended(&self) -> bool {
|
|
self.recalibration_recommended
|
|
}
|
|
|
|
/// Privacy class the most recent cycle was emitted under (post-demotion);
|
|
/// `None` until a governed cycle has run.
|
|
pub fn effective_class(&self) -> Option<PrivacyClass> {
|
|
self.effective_class
|
|
}
|
|
|
|
/// Whether the most recent cycle was demoted (contradiction / mesh risk).
|
|
pub fn demoted(&self) -> bool {
|
|
self.demoted
|
|
}
|
|
|
|
/// Engine cycles that returned an error since startup.
|
|
pub fn engine_error_count(&self) -> u64 {
|
|
self.engine_error_count
|
|
}
|
|
|
|
/// ADR-141 output mapping for the live publish path (review finding 1c):
|
|
/// at effective class [`PrivacyClass::Restricted`] the bfld privacy gate
|
|
/// drops the amplitude + phase proxies; the live `SensingUpdate` applies
|
|
/// the same field mapping by suppressing the per-node raw amplitude
|
|
/// vectors when this returns true. Classes below `Restricted` leave the
|
|
/// publish unchanged.
|
|
pub fn suppress_raw_outputs(&self) -> bool {
|
|
self.effective_class
|
|
.is_some_and(|c| c.as_u8() >= PrivacyClass::Restricted.as_u8())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use std::collections::VecDeque;
|
|
use std::time::Instant;
|
|
use wifi_densepose_bfld::PrivacyClass;
|
|
|
|
fn node_state_with_history(amp: f64, n_sub: usize) -> NodeState {
|
|
let mut ns = NodeState::new();
|
|
let frame: Vec<f64> = (0..n_sub).map(|i| amp + 0.1 * i as f64).collect();
|
|
ns.frame_history = VecDeque::from(vec![frame]);
|
|
ns.last_frame_time = Some(Instant::now());
|
|
ns
|
|
}
|
|
|
|
fn two_node_states() -> HashMap<u8, NodeState> {
|
|
let mut m = HashMap::new();
|
|
m.insert(0u8, node_state_with_history(1.0, 56));
|
|
m.insert(1u8, node_state_with_history(1.05, 56));
|
|
m
|
|
}
|
|
|
|
#[test]
|
|
fn empty_states_produce_no_belief() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "living_room", "Living Room");
|
|
let out = bridge.process_cycle_from_states(&HashMap::new(), 1_000);
|
|
assert!(out.is_none());
|
|
// No belief published, no sensor wired.
|
|
assert_eq!(bridge.registered_node_count(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn live_cycle_produces_witnessed_belief_with_provenance() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "living_room", "Living Room");
|
|
let states = two_node_states();
|
|
let out = bridge
|
|
.process_cycle_from_states(&states, 10_000)
|
|
.expect("frames present")
|
|
.expect("fusion succeeds");
|
|
|
|
// Full provenance: evidence + model + calibration + privacy decision.
|
|
assert!(!out.provenance.evidence.is_empty());
|
|
assert_eq!(out.provenance.model_version, "rfenc-v1");
|
|
assert!(out.provenance.calibration_version.starts_with("cal:"));
|
|
assert!(out.provenance.privacy_decision.starts_with("PrivateHome/"));
|
|
// A witness was produced and the belief is in the WorldGraph.
|
|
assert_ne!(out.witness, [0u8; 32]);
|
|
assert!(bridge.engine().world().node(out.semantic_id).is_some());
|
|
// Both nodes are now wired as sensors.
|
|
assert_eq!(bridge.registered_node_count(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn live_path_is_deterministic() {
|
|
let states = two_node_states_fixed();
|
|
let run = || {
|
|
let mut b = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
b.process_cycle_from_states(&states, 5_000).unwrap().unwrap()
|
|
};
|
|
let a = run();
|
|
let b = run();
|
|
assert_eq!(a.witness, b.witness);
|
|
assert_eq!(a.provenance.calibration_version, b.provenance.calibration_version);
|
|
assert_eq!(a.effective_class, b.effective_class);
|
|
}
|
|
|
|
// Deterministic node states (no wall-clock in amplitude/history).
|
|
fn two_node_states_fixed() -> HashMap<u8, NodeState> {
|
|
let mut m = HashMap::new();
|
|
for (id, amp) in [(0u8, 1.0_f64), (1u8, 1.05)] {
|
|
let mut ns = NodeState::new();
|
|
ns.frame_history = VecDeque::from(vec![(0..56)
|
|
.map(|i| amp + 0.1 * i as f64)
|
|
.collect::<Vec<f64>>()]);
|
|
ns.last_frame_time = Some(Instant::now());
|
|
m.insert(id, ns);
|
|
}
|
|
m
|
|
}
|
|
|
|
#[test]
|
|
fn nodes_registered_once_across_cycles() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
let states = two_node_states();
|
|
bridge.process_cycle_from_states(&states, 1_000);
|
|
bridge.process_cycle_from_states(&states, 2_000);
|
|
bridge.process_cycle_from_states(&states, 3_000);
|
|
// Still exactly two sensors — idempotent registration.
|
|
assert_eq!(bridge.registered_node_count(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn retention_bounds_world_graph_growth() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
bridge.set_semantic_retention(5);
|
|
let states = two_node_states();
|
|
for i in 0..20i64 {
|
|
bridge.process_cycle_from_states(&states, 1_000 + i * 50);
|
|
}
|
|
// room + 2 sensors + at most 5 retained beliefs.
|
|
assert!(bridge.engine().world().node_count() <= 3 + 5);
|
|
}
|
|
|
|
#[test]
|
|
fn adapter_identity_flows_into_live_witness() {
|
|
let states = two_node_states_fixed();
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
let base = bridge
|
|
.process_cycle_from_states(&states, 1_000)
|
|
.unwrap()
|
|
.unwrap();
|
|
bridge.set_room_adapter(AdapterInfo {
|
|
adapter_id: "deadbeefcafef00d".into(),
|
|
trained_samples: 120,
|
|
});
|
|
let adapted = bridge
|
|
.process_cycle_from_states(&states, 2_000)
|
|
.unwrap()
|
|
.unwrap();
|
|
assert!(adapted
|
|
.provenance
|
|
.model_version
|
|
.ends_with("+adapter:deadbeefcafef00d"));
|
|
assert_ne!(adapted.witness, base.witness);
|
|
// Clearing reverts to the base model identity.
|
|
bridge.clear_room_adapter();
|
|
let back = bridge
|
|
.process_cycle_from_states(&states, 3_000)
|
|
.unwrap()
|
|
.unwrap();
|
|
assert_eq!(back.provenance.model_version, "rfenc-v1");
|
|
}
|
|
|
|
/// Wiring (review finding 1): a live frame in → trust state recorded on
|
|
/// the bridge (witness, effective class, recalibration flag), readable by
|
|
/// the status endpoint, with a zero error count on the happy path.
|
|
#[test]
|
|
fn observe_cycle_records_trust_state() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
assert!(bridge.last_trust_witness().is_none());
|
|
assert_eq!(bridge.effective_class(), None);
|
|
|
|
let out = bridge
|
|
.observe_cycle(&two_node_states(), 1_000)
|
|
.expect("two fresh nodes → governed cycle runs");
|
|
|
|
assert_eq!(bridge.last_trust_witness(), Some(out.witness));
|
|
assert_eq!(bridge.effective_class(), Some(out.effective_class));
|
|
assert_eq!(
|
|
bridge.recalibration_recommended(),
|
|
out.recalibration_recommended
|
|
);
|
|
assert_eq!(bridge.demoted(), out.demoted);
|
|
assert_eq!(bridge.engine_error_count(), 0);
|
|
// PrivateHome clean cycle → Anonymous → raw outputs NOT suppressed.
|
|
assert_eq!(bridge.effective_class(), Some(PrivacyClass::Anonymous));
|
|
assert!(!bridge.suppress_raw_outputs());
|
|
}
|
|
|
|
/// Error wiring (review finding 1a): two live nodes with mismatched
|
|
/// subcarrier counts make fusion return a `DimensionMismatch` →
|
|
/// `EngineError` — previously dropped by `if let Some(Ok(..))` at the
|
|
/// call sites. The counter must increment and the last good trust state
|
|
/// must survive a later failure.
|
|
#[test]
|
|
fn observe_cycle_counts_engine_errors() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
let mut mismatched = HashMap::new();
|
|
mismatched.insert(0u8, node_state_with_history(1.0, 56));
|
|
mismatched.insert(1u8, node_state_with_history(1.05, 30)); // 30 ≠ 56 subcarriers
|
|
|
|
assert!(bridge.observe_cycle(&mismatched, 1_000).is_none());
|
|
assert_eq!(bridge.engine_error_count(), 1);
|
|
assert!(
|
|
bridge.last_trust_witness().is_none(),
|
|
"no witness from a failed cycle"
|
|
);
|
|
|
|
assert!(bridge.observe_cycle(&mismatched, 2_000).is_none());
|
|
assert_eq!(bridge.engine_error_count(), 2);
|
|
|
|
// A later good cycle records trust state; the audit count is kept.
|
|
let out = bridge.observe_cycle(&two_node_states(), 3_000);
|
|
assert!(out.is_some());
|
|
assert!(bridge.last_trust_witness().is_some());
|
|
assert_eq!(bridge.engine_error_count(), 2);
|
|
|
|
// And a subsequent failure keeps the last good witness readable.
|
|
assert!(bridge.observe_cycle(&mismatched, 4_000).is_none());
|
|
assert_eq!(bridge.engine_error_count(), 3);
|
|
assert!(bridge.last_trust_witness().is_some());
|
|
}
|
|
|
|
/// ADR-141 mapping (review finding 1c): a cycle emitted at class
|
|
/// Restricted flips `suppress_raw_outputs`, which `main.rs` uses to strip
|
|
/// per-node raw amplitude vectors from the live publish — the same field
|
|
/// mapping bfld's privacy gate applies at `Restricted`.
|
|
#[test]
|
|
fn restricted_class_suppresses_raw_outputs() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
bridge.set_privacy_mode(PrivacyMode::StrictNoIdentity); // base = Restricted
|
|
bridge
|
|
.observe_cycle(&two_node_states(), 1_000)
|
|
.expect("cycle runs");
|
|
assert_eq!(bridge.effective_class(), Some(PrivacyClass::Restricted));
|
|
assert!(bridge.suppress_raw_outputs());
|
|
}
|
|
|
|
#[test]
|
|
fn identity_strict_mode_is_carried_into_provenance() {
|
|
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R");
|
|
bridge.set_privacy_mode(PrivacyMode::StrictNoIdentity);
|
|
let out = bridge
|
|
.process_cycle_from_states(&two_node_states(), 7_000)
|
|
.unwrap()
|
|
.unwrap();
|
|
assert!(out.provenance.privacy_decision.starts_with("StrictNoIdentity/"));
|
|
// Effective class is a valid privacy class (sanity).
|
|
let _ = matches!(
|
|
out.effective_class,
|
|
PrivacyClass::Raw | PrivacyClass::Derived | PrivacyClass::Anonymous | PrivacyClass::Restricted
|
|
);
|
|
}
|
|
}
|