diff --git a/docs/adr/ADR-357-raw-csi-calibration-sample-integrity.md b/docs/adr/ADR-357-raw-csi-calibration-sample-integrity.md new file mode 100644 index 000000000..e4eef61f3 --- /dev/null +++ b/docs/adr/ADR-357-raw-csi-calibration-sample-integrity.md @@ -0,0 +1,89 @@ +# ADR 357: Raw CSI calibration sample integrity + +Status: Accepted + +Date: 2026-09-06 + +## Context + +The empty room field model requires 1,000 accepted observations over at least +600 seconds. The sensing server previously invoked calibration from both the +raw CSI path and the one second edge vital path. Edge vital packets did not add +a CSI observation, so that path repeatedly fed the existing history tail. A +counter could therefore satisfy the frame gate without representing 1,000 +independent CSI packets. + +Node liveness also combined edge vital and CSI traffic. A source emitting only +edge vital summaries could appear live enough to start calibration even though +it had no fresh raw CSI. Finally, cloning the complete per node history before +each feed copied as much as 204,800 bytes for a 100 frame, 256 subcarrier +window even though calibration consumes only the current observation. + +The earlier single installation result recorded in ADR 355 predates this +admission boundary. Its frame count is not accepted as evidence of independent +CSI coverage. It remains useful only as diagnosis of the failure mode and must +not be promoted as a production bootstrap prior. + +## Decision + +1. Only a grid admitted raw CSI packet may advance field calibration. Edge + vital packets update their own health and presentation state but never feed + the field model. +2. Each calibration session stores the last admitted wrapping sequence for + every contributing node. Duplicate sequences are ignored. Normal forward + progress and the `u32::MAX` to zero wrap are accepted. +3. A backward or ambiguous half range sequence latches that node as faulted for + the remainder of the session. Status exposes `sequence_fault_node_ids`, and + finalization returns `calibration_sequence_discontinuity`. The operator must + cancel and restart the empty room capture. +4. A requested source may start calibration only when it has a grid admitted + raw CSI packet newer than five seconds. General node liveness is not enough. + The node endpoint reports both `csi_status` and `csi_last_seen_ms` so clients + can explain the gate before the operator begins. +5. Start, cancel, reset, and process initialization clear sequence admission + state. A failed feed does not advance the sequence cursor, so the same + packet may be retried after a transient model input failure. +6. The field bridge receives the current amplitude slice rather than a cloned + history. Canonical grid normalization remains unchanged. + +## Consequences + +Calibration frame count now means unique, forward raw CSI observations within +one uninterrupted sequence epoch. A vitals only node cannot produce a +bootstrap model. + +One legitimately reordered UDP packet invalidates a ten minute capture. This +is an intentional fail closed tradeoff because the current wire format has no +cryptographic boot epoch and cannot distinguish reordering from a device +restart. A future firmware protocol may add a signed boot identifier and a +bounded reordering window. + +No new raw CSI is persisted. Sequence cursors and fault markers remain bounded +in memory to active calibration sources. The persisted bootstrap format and +its negative only authority are unchanged. + +## Implementation + +1. `NodeState` records a separate accepted CSI arrival clock. +2. `AppStateInner::maybe_feed_calibration_frame` owns source binding, sequence + ordering, fault latching, model feed, and cursor commit. +3. `field_bridge::maybe_feed_calibration` accepts exactly one current amplitude + slice and rejects an empty observation. +4. Calibration start, status, stop, cancel, reset, and node status expose and + enforce the new admission contract. + +## Acceptance test + +Unit tests must prove first packet acceptance, forward progress, duplicate +rejection, source isolation, maximum counter wrap, empty input retry, backward +sequence latching, half range rejection, session clear, status visibility, and +typed stop refusal. + +On physical hardware, begin a Node 5 empty room capture and observe a fresh +sixty second interval. Calibration frame growth must match unique grid admitted +Node 5 CSI sequences. Edge vital packets alone must produce zero growth. The +source must retain `csi_status=active`, sequence faults must remain empty, and +numeric vital output must remain absent. Complete both the 1,000 frame and 600 +second gates, then run the twelve sample server controlled bootstrap holdout. +Do not claim human detection improvement until a separate consented occupied +holdout demonstrates retained recall. diff --git a/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs b/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs index 21c4ee461..4ba83b0e7 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs @@ -283,31 +283,36 @@ pub fn calibrated_occupancy( }) } -/// Feed the latest frame to the FieldModel during calibration collection. +/// Feed the current CSI frame to the FieldModel during calibration. /// /// Acts while the model is `Uncalibrated` or `Collecting`. The first fed frame /// flips a freshly-started (`Uncalibrated`) model to `Collecting` inside /// `feed_calibration`; without accepting the `Uncalibrated` state here the two /// gates deadlock and the frame count never leaves 0 (calibration/start yields -/// an `Uncalibrated` model that nothing would ever advance). Wraps the latest -/// frame as a single-link observation (n_links=1) and feeds it. -pub fn maybe_feed_calibration(field: &mut FieldModel, frame_history: &VecDeque>) { +/// an `Uncalibrated` model that nothing would ever advance). Edge-vitals +/// packets have no CSI observation and must never call this helper. The caller +/// owns the stateful sequence admission gate. Wraps the current frame as a +/// single-link observation (n_links=1) and feeds it. +pub fn maybe_feed_calibration(field: &mut FieldModel, amplitudes: &[f64]) -> bool { if !matches!( field.status(), CalibrationStatus::Uncalibrated | CalibrationStatus::Collecting - ) { - return; + ) || amplitudes.is_empty() + { + return false; } - if let Some(latest) = frame_history.back() { - // Resample the raw amplitude vector onto the FieldModel's canonical - // 56-tone grid before feeding. Real HT40 nodes stream 128-wide frames; - // feeding those raw made every `feed_calibration` fail DimensionMismatch - // (swallowed at debug level), pinning frame_count at 0 even after the - // status-gate deadlock was fixed. Single-link observation: [1][56]. - let canonical = CALIB_NORMALIZER.resample_to_canonical(latest); - let observations = vec![canonical]; - if let Err(e) = field.feed_calibration(&observations) { + // Resample the raw amplitude vector onto the FieldModel's canonical + // 56-tone grid before feeding. Real HT40 nodes stream 128-wide frames; + // feeding those raw made every `feed_calibration` fail DimensionMismatch + // (swallowed at debug level), pinning frame_count at 0 even after the + // status-gate deadlock was fixed. Single-link observation: [1][56]. + let canonical = CALIB_NORMALIZER.resample_to_canonical(amplitudes); + let observations = vec![canonical]; + match field.feed_calibration(&observations) { + Ok(()) => true, + Err(e) => { tracing::debug!("FieldModel calibration feed: {e}"); + false } } } @@ -418,10 +423,7 @@ mod tests { // n_subcarriers defaults to 56; one single-link frame of that width. let frame = vec![0.5_f64; 56]; - let mut history: VecDeque> = VecDeque::new(); - history.push_back(frame); - - maybe_feed_calibration(&mut field, &history); + assert!(maybe_feed_calibration(&mut field, &frame)); assert_eq!( field.status(), @@ -434,8 +436,8 @@ mod tests { "frame count must advance past 0" ); - // Subsequent frames keep accumulating while Collecting. - maybe_feed_calibration(&mut field, &history); + // A subsequent unique frame keeps accumulating while Collecting. + assert!(maybe_feed_calibration(&mut field, &frame)); assert_eq!(field.calibration_frame_count(), 2); } @@ -450,10 +452,7 @@ mod tests { // 128-wide frame (HT40), NOT the model's 56 — would DimensionMismatch raw. let wide = vec![0.5_f64; 128]; - let mut history: VecDeque> = VecDeque::new(); - history.push_back(wide); - - maybe_feed_calibration(&mut field, &history); + assert!(maybe_feed_calibration(&mut field, &wide)); assert_eq!( field.status(), @@ -467,6 +466,13 @@ mod tests { ); } + #[test] + fn maybe_feed_calibration_rejects_an_empty_observation() { + let mut field = FieldModel::new(single_link_config()).expect("field model"); + assert!(!maybe_feed_calibration(&mut field, &[])); + assert_eq!(field.calibration_frame_count(), 0); + } + #[test] fn calibrated_occupancy_is_fresh_model_only_and_normalizes_runtime_frames() { let field = fresh_test_model(); diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index c2f8790dd..8b2bc921b 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -849,6 +849,9 @@ struct NodeState { latest_sync: Option, /// Last time a sync packet from this node was received (for staleness). latest_sync_at: Option, + /// Arrival time of the newest grid-admitted raw CSI frame. Edge-vitals + /// packets intentionally do not refresh this clock. + latest_accepted_csi_at: Option, /// Sequence number of the newest CSI frame admitted to `frame_history`. /// Kept alongside the history so multistatic fusion can timestamp the /// exact sample it consumes, rather than the host's UDP arrival time. @@ -1161,6 +1164,7 @@ impl NodeState { sync_valid: bool, now: std::time::Instant, ) -> bool { + self.latest_accepted_csi_at = Some(now); self.latest_csi_sequence = Some(sequence); self.latest_csi_sync_valid = sync_valid; self.observe_csi_frame_arrival(now) @@ -1190,6 +1194,7 @@ impl NodeState { edge_vitals: None, latest_sync: None, latest_sync_at: None, + latest_accepted_csi_at: None, latest_csi_sequence: None, latest_csi_sync_valid: false, csi_fps_ema: 20.0, @@ -1627,6 +1632,12 @@ struct AppStateInner { calibration_model_id: Option, /// Nodes that actually contributed frames to the current calibration. calibration_source_node_ids: std::collections::BTreeSet, + /// Last admitted raw CSI sequence per contributing node. Edge-vitals + /// packets do not carry a new CSI observation and cannot advance this map. + calibration_last_sequences: HashMap, + /// Nodes whose sequence moved backward during the current session. Without + /// a wire boot epoch, the session must stay failed closed after regression. + calibration_sequence_fault_node_ids: std::collections::BTreeSet, // ── ADR-044 §5.2: adaptive rolling-p95 normalization ───────────────────── /// Rolling P95 of `FeatureInfo.variance` over the last ~30 s (600 frames @ 20 Hz). pub(crate) p95_variance: RollingP95, @@ -1729,7 +1740,74 @@ mod adr323_pose_physics_http_tests { /// If no ESP32 frame arrives within this duration, source reverts to offline. const ESP32_OFFLINE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CalibrationSequenceOrder { + First, + Forward, + Duplicate, + Regression, +} + +fn calibration_sequence_order(previous: Option, sequence: u32) -> CalibrationSequenceOrder { + let Some(previous) = previous else { + return CalibrationSequenceOrder::First; + }; + let delta = sequence.wrapping_sub(previous); + if delta == 0 { + CalibrationSequenceOrder::Duplicate + } else if delta < (1_u32 << 31) { + CalibrationSequenceOrder::Forward + } else { + CalibrationSequenceOrder::Regression + } +} + impl AppStateInner { + /// Admit at most one raw CSI observation per forward node sequence into + /// the current calibration session. This stateful boundary prevents a + /// caller from advancing the model with a replayed history tail. + fn maybe_feed_calibration_frame( + &mut self, + node_id: u8, + sequence: u32, + amplitudes: &[f64], + ) -> bool { + if !calibration_source_accepts(&self.calibration_source_node_ids, node_id) + || self.calibration_sequence_fault_node_ids.contains(&node_id) + { + return false; + } + let previous_sequence = self.calibration_last_sequences.get(&node_id).copied(); + match calibration_sequence_order(previous_sequence, sequence) { + CalibrationSequenceOrder::Duplicate => return false, + CalibrationSequenceOrder::Regression => { + self.calibration_sequence_fault_node_ids.insert(node_id); + warn!( + node_id, + previous_sequence, + sequence, + "Calibration source sequence regressed; restart the empty-room capture" + ); + return false; + } + CalibrationSequenceOrder::First | CalibrationSequenceOrder::Forward => {} + } + let accepted = self + .field_model + .as_mut() + .is_some_and(|field| field_bridge::maybe_feed_calibration(field, amplitudes)); + if accepted { + self.calibration_source_node_ids.insert(node_id); + self.calibration_last_sequences.insert(node_id, sequence); + } + accepted + } + + fn clear_calibration_sequence_state(&mut self) { + self.calibration_last_sequences.clear(); + self.calibration_sequence_fault_node_ids.clear(); + } + fn field_model_status_at(&self, observed_at_unix_ms: u64) -> Option { self.field_model.as_ref().map(|model| { field_bridge::calibration_status_at(model, observed_at_unix_ms.saturating_mul(1_000)) @@ -1972,6 +2050,8 @@ impl AppStateInner { bootstrap_baseline_active: false, calibration_model_id: None, calibration_source_node_ids: std::collections::BTreeSet::new(), + calibration_last_sequences: HashMap::new(), + calibration_sequence_fault_node_ids: std::collections::BTreeSet::new(), p95_variance: RollingP95::new(600, 60), p95_motion_band_power: RollingP95::new(600, 60), p95_spectral_power: RollingP95::new(600, 60), @@ -6288,6 +6368,11 @@ fn calibration_source_accepts( source_node_ids.is_empty() || source_node_ids.contains(&node_id) } +fn calibration_source_has_fresh_csi(node: &NodeState, now: std::time::Instant) -> bool { + node.latest_accepted_csi_at + .is_some_and(|seen| now.saturating_duration_since(seen) < ESP32_OFFLINE_TIMEOUT) +} + #[cfg(test)] mod bootstrap_vital_publication_tests { use super::*; @@ -6309,6 +6394,148 @@ mod bootstrap_vital_publication_tests { assert!(!calibration_source_accepts(&sources, 7)); } + #[test] + fn edge_vitals_liveness_cannot_authorize_raw_csi_calibration() { + let now = std::time::Instant::now(); + let mut node = NodeState::new(); + node.last_frame_time = Some(now); + assert!(!calibration_source_has_fresh_csi(&node, now)); + + node.observe_accepted_csi_frame(9, false, now); + assert!(calibration_source_has_fresh_csi(&node, now)); + assert!(!calibration_source_has_fresh_csi( + &node, + now + ESP32_OFFLINE_TIMEOUT + )); + } + + #[test] + fn calibration_session_counts_each_unique_bound_csi_sequence_once() { + let mut state = AppStateInner::minimal(); + state.field_model = Some( + FieldModel::new(field_bridge::single_link_config()).expect("field model"), + ); + state.calibration_source_node_ids.insert(5); + + let first = vec![0.25; 56]; + assert!(state.maybe_feed_calibration_frame(5, 41, &first)); + assert!(!state.maybe_feed_calibration_frame(5, 41, &first)); + + let second = vec![0.5; 56]; + assert!(state.maybe_feed_calibration_frame(5, 42, &second)); + assert!(!state.maybe_feed_calibration_frame(3, 43, &second)); + + let model = state.field_model.as_ref().expect("field model remains active"); + assert_eq!(model.calibration_frame_count(), 2); + assert_eq!(state.calibration_last_sequences.get(&5), Some(&42)); + assert!(!state.calibration_last_sequences.contains_key(&3)); + } + + #[test] + fn calibration_sequence_regression_latches_session_failure() { + let mut state = AppStateInner::minimal(); + state.field_model = Some( + FieldModel::new(field_bridge::single_link_config()).expect("field model"), + ); + state.calibration_source_node_ids.insert(5); + let frame = vec![0.25; 56]; + + assert!(state.maybe_feed_calibration_frame(5, 10, &frame)); + assert!(state.maybe_feed_calibration_frame(5, 11, &frame)); + assert!(!state.maybe_feed_calibration_frame(5, 10, &frame)); + assert!(!state.maybe_feed_calibration_frame(5, 12, &frame)); + assert!(state.calibration_sequence_fault_node_ids.contains(&5)); + assert_eq!( + state + .field_model + .as_ref() + .expect("field model") + .calibration_frame_count(), + 2 + ); + + state.clear_calibration_sequence_state(); + assert!(state.calibration_sequence_fault_node_ids.is_empty()); + assert!(state.calibration_last_sequences.is_empty()); + } + + #[test] + fn calibration_sequence_wraps_and_failed_empty_input_can_retry() { + let mut state = AppStateInner::minimal(); + state.field_model = Some( + FieldModel::new(field_bridge::single_link_config()).expect("field model"), + ); + state.calibration_source_node_ids.insert(5); + let frame = vec![0.25; 56]; + + assert!(!state.maybe_feed_calibration_frame(5, u32::MAX, &[])); + assert!(state.calibration_last_sequences.is_empty()); + assert!(state.maybe_feed_calibration_frame(5, u32::MAX, &frame)); + assert!(state.maybe_feed_calibration_frame(5, 0, &frame)); + assert!(!state.maybe_feed_calibration_frame(5, 0, &frame)); + assert!(state.maybe_feed_calibration_frame(5, 1, &frame)); + assert_eq!( + state + .field_model + .as_ref() + .expect("field model") + .calibration_frame_count(), + 3 + ); + } + + #[test] + fn calibration_sequence_order_rejects_backward_and_half_range() { + assert_eq!( + calibration_sequence_order(None, 7), + CalibrationSequenceOrder::First + ); + assert_eq!( + calibration_sequence_order(Some(7), 8), + CalibrationSequenceOrder::Forward + ); + assert_eq!( + calibration_sequence_order(Some(8), 8), + CalibrationSequenceOrder::Duplicate + ); + assert_eq!( + calibration_sequence_order(Some(8), 7), + CalibrationSequenceOrder::Regression + ); + assert_eq!( + calibration_sequence_order(Some(0), 1_u32 << 31), + CalibrationSequenceOrder::Regression + ); + assert_eq!( + calibration_sequence_order(Some(u32::MAX), 0), + CalibrationSequenceOrder::Forward + ); + } + + #[tokio::test] + async fn calibration_status_and_stop_fail_closed_after_sequence_regression() { + let mut inner = AppStateInner::minimal(); + inner.field_model = Some( + FieldModel::new(field_bridge::single_link_config()).expect("field model"), + ); + inner.calibration_model_id = Some("cal-model-test".to_string()); + inner.calibration_source_node_ids.insert(5); + inner.calibration_sequence_fault_node_ids.insert(5); + let state = Arc::new(RwLock::new(inner)); + + let Json(status) = calibration_status(State(state.clone())).await; + assert_eq!(status["sequence_fault_node_ids"], serde_json::json!([5])); + + let Json(stopped) = calibration_stop(State(state)).await; + assert_eq!(stopped["success"], false); + assert_eq!( + stopped["error_code"], + "calibration_sequence_discontinuity" + ); + assert_eq!(stopped["model_id"], "cal-model-test"); + assert_eq!(stopped["source_node_ids"], serde_json::json!([5])); + } + fn strong_candidates() -> VitalSigns { VitalSigns { breathing_rate_bpm: Some(15.0), @@ -6420,15 +6647,15 @@ async fn calibration_start( let mut s = state.write().await; if let Some(source_node_id) = query.source_node_id { let now = std::time::Instant::now(); - let source_is_live = s.node_states.get(&source_node_id).is_some_and(|node| { - node.last_frame_time - .is_some_and(|seen| now.saturating_duration_since(seen) < ESP32_OFFLINE_TIMEOUT) - }); + let source_is_live = s + .node_states + .get(&source_node_id) + .is_some_and(|node| calibration_source_has_fresh_csi(node, now)); if !source_is_live { return Json(serde_json::json!({ "success": false, "error_code": "calibration_source_unavailable", - "error": "The requested calibration source is missing or stale.", + "error": "The requested calibration source is missing or has no fresh raw CSI.", "source_node_id": source_node_id, })); } @@ -6463,6 +6690,7 @@ async fn calibration_start( s.field_model = Some(fm); s.calibration_model_id = Some(opaque_calibration_model_id()); s.calibration_source_node_ids.clear(); + s.clear_calibration_sequence_state(); if let Some(source_node_id) = query.source_node_id { s.calibration_source_node_ids.insert(source_node_id); } @@ -6482,6 +6710,16 @@ async fn calibration_stop(State(state): State) -> Json = s.calibration_source_node_ids.iter().copied().collect(); + if !s.calibration_sequence_fault_node_ids.is_empty() { + return Json(serde_json::json!({ + "success": false, + "error_code": "calibration_sequence_discontinuity", + "error": "Raw CSI sequence moved backward during capture. Cancel and restart the empty-room calibration.", + "model_id": model_id, + "source_node_ids": source_node_ids, + "sequence_fault_node_ids": s.calibration_sequence_fault_node_ids, + })); + } if let Some(ref mut fm) = s.field_model { // Guard: finalizing before enough empty-room frames have accumulated // is a client-side sequencing error, not a server fault. Return a @@ -6585,6 +6823,7 @@ async fn calibration_status(State(state): State) -> Json) -> Json) -> Json) -> Json 5000; let status = if stale { "stale" } else { "active" }; + let csi_status = if csi_elapsed_ms.is_some_and(|age| { + age < ESP32_OFFLINE_TIMEOUT.as_millis() as u64 + }) { + "active" + } else { + "stale" + }; let rssi = ns.rssi_history.back().copied().unwrap_or(-90.0); let bootstrap_empty = s.bootstrap_baseline_active && s.bootstrap_baseline.as_ref().is_some_and(|metadata| { @@ -7458,6 +7709,9 @@ async fn nodes_endpoint(State(state): State) -> Json before - } else { - false - } - } else { - false - }; - if accepted { - s.calibration_source_node_ids.insert(node_id); - } - } - // Build nodes array with all active nodes. let resolved_positions = node_positions_by_active_id(&s.node_positions_config, &s.node_states, now); @@ -8362,29 +8592,13 @@ async fn udp_receiver_task( sref.engine_bridge.observe_cycle(&sref.node_states, now_ms); } - // Feed field model calibration if active (use per-node history for ESP32). - if let Some(frame_history) = s - .node_states - .get(&node_id) - .map(|ns| ns.frame_history.clone()) - { - let source_allowed = - calibration_source_accepts(&s.calibration_source_node_ids, node_id); - let accepted = if source_allowed { - if let Some(ref mut fm) = s.field_model { - let before = fm.calibration_frame_count(); - field_bridge::maybe_feed_calibration(fm, &frame_history); - fm.calibration_frame_count() > before - } else { - false - } - } else { - false - }; - if accepted { - s.calibration_source_node_ids.insert(node_id); - } - } + // Feed only this grid-admitted raw CSI observation. The + // session boundary deduplicates its wrapping sequence. + s.maybe_feed_calibration_frame( + node_id, + frame.sequence, + &frame.amplitudes, + ); // Build nodes array with all active nodes. ADR-141 output // gating (review finding 1c): when the governed engine @@ -10157,6 +10371,8 @@ async fn main() { bootstrap_baseline_active, calibration_model_id: args.calibrate.then(opaque_calibration_model_id), calibration_source_node_ids: std::collections::BTreeSet::new(), + calibration_last_sequences: HashMap::new(), + calibration_sequence_fault_node_ids: std::collections::BTreeSet::new(), // ADR-044 §5.2: rolling-P95 over ~30 s at 20 Hz; warm-up after 60 samples. p95_variance: RollingP95::new(600, 60), p95_motion_band_power: RollingP95::new(600, 60),