From 82e05d06640bdc2ed1cedb3b7f3323a2e5679b77 Mon Sep 17 00:00:00 2001 From: ruv Date: Sun, 6 Sep 2026 09:37:03 -0400 Subject: [PATCH] fix(calibration): expire live field models safely --- .../src/field_bridge.rs | 45 +++- .../wifi-densepose-sensing-server/src/main.rs | 228 ++++++++++++++---- .../src/types.rs | 5 + 3 files changed, 228 insertions(+), 50 deletions(-) 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 980f04cfb..21c4ee461 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs @@ -149,6 +149,18 @@ pub fn single_link_config() -> FieldModelConfig { } } +/// Resolve the model status at the observation wall clock. `FieldModel::status` +/// records the state at collection or restore time and does not advance as a +/// long-running server crosses the stale and expiry boundaries. +pub fn calibration_status_at(field: &FieldModel, observed_at_us: u64) -> CalibrationStatus { + match field.status() { + CalibrationStatus::Uncalibrated | CalibrationStatus::Collecting => field.status(), + CalibrationStatus::Fresh | CalibrationStatus::Stale | CalibrationStatus::Expired => { + field.check_freshness(observed_at_us) + } + } +} + /// Estimate occupancy using the FieldModel when calibrated, falling back /// to the score-based heuristic otherwise. /// @@ -158,10 +170,11 @@ pub fn single_link_config() -> FieldModelConfig { pub fn occupancy_or_fallback( field: &FieldModel, frame_history: &VecDeque>, + observed_at_us: u64, smoothed_score: f64, prev_count: usize, ) -> usize { - match field.status() { + match calibration_status_at(field, observed_at_us) { CalibrationStatus::Fresh | CalibrationStatus::Stale => { let frames: Vec> = frame_history .iter() @@ -473,6 +486,36 @@ mod tests { ); } + #[test] + fn occupancy_falls_back_after_expiry_without_a_process_restart() { + let field = fresh_test_model(); + let baseline = field.modes().unwrap().baseline[0].clone(); + let history: VecDeque> = + (0..OCCUPANCY_WINDOW).map(|_| baseline.clone()).collect(); + + assert_eq!(field.status(), CalibrationStatus::Fresh); + assert_eq!( + calibration_status_at(&field, 1_500_000), + CalibrationStatus::Fresh + ); + assert_eq!( + occupancy_or_fallback(&field, &history, 1_500_000, 0.0, 0), + 0, + "a fresh field model may resolve its learned baseline as empty" + ); + + assert_eq!(field.status(), CalibrationStatus::Fresh); + assert_eq!( + calibration_status_at(&field, 4_000_000), + CalibrationStatus::Expired + ); + assert_eq!( + occupancy_or_fallback(&field, &history, 4_000_000, 0.0, 0), + score_to_person_count(0.0, 0), + "an expired in-memory model must fall back even though its cached status is fresh" + ); + } + #[test] fn adaptive_runtime_reference_scores_empty_and_rejects_shift() { let config = FieldModelConfig { diff --git a/v2/crates/wifi-densepose-sensing-server/src/main.rs b/v2/crates/wifi-densepose-sensing-server/src/main.rs index 5c159c417..c2f8790dd 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -1730,12 +1730,38 @@ mod adr323_pose_physics_http_tests { const ESP32_OFFLINE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); impl AppStateInner { + 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)) + }) + } + + fn field_model_active_at(&self, observed_at_unix_ms: u64) -> bool { + self.field_model_status_at(observed_at_unix_ms) + .is_some_and(|status| status != CalibrationStatus::Expired) + } + + fn bootstrap_baseline_active_at(&self, observed_at_unix_ms: u64) -> bool { + self.bootstrap_baseline_active && self.field_model_active_at(observed_at_unix_ms) + } + + fn explicit_calibration_fresh_at(&self, observed_at_unix_ms: u64) -> bool { + !self.bootstrap_baseline_active + && self.field_model_status_at(observed_at_unix_ms) == Some(CalibrationStatus::Fresh) + } + /// Return the effective data source, accounting for ESP32 frame timeout. /// If the source is "esp32" but no frame has arrived in 5 seconds, returns /// "esp32:offline" so the UI can distinguish active vs stale connections. /// Person count: eigenvalue-based if field model is calibrated, else heuristic. /// Uses global frame_history if populated, otherwise the freshest per-node history. - fn person_count(&self) -> usize { + fn person_count_at(&self, observed_at_unix_ms: u64) -> usize { + // A persisted bootstrap model has negative-only authority. Its only + // allowed occupancy effect is the explicit empty-background suppression + // in `bootstrap_empty_prior_applies`; it must never infer positive count. + if self.bootstrap_baseline_active { + return score_to_person_count(self.smoothed_person_score, self.prev_person_count); + } match self.field_model.as_ref() { Some(fm) => { // A single-link model may score only the source node it was @@ -1769,6 +1795,7 @@ impl AppStateInner { field_bridge::occupancy_or_fallback( fm, history, + observed_at_unix_ms.saturating_mul(1_000), self.smoothed_person_score, self.prev_person_count, ) @@ -1777,6 +1804,10 @@ impl AppStateInner { } } + fn person_count(&self) -> usize { + self.person_count_at(chrono::Utc::now().timestamp_millis().max(0) as u64) + } + /// A restored startup image has negative-only authority. It may suppress /// an empty-room false positive after one complete runtime window, but it /// cannot establish positive presence or publish vital signs. @@ -1955,6 +1986,95 @@ impl AppStateInner { } } +#[cfg(test)] +mod calibration_expiry_tests { + use super::*; + use wifi_densepose_signal::ruvsense::field_model::FieldModelConfig; + + fn finalized_model(calibrated_at_unix_ms: u64) -> FieldModel { + let mut model = FieldModel::new(FieldModelConfig { + n_links: 1, + n_subcarriers: 56, + n_modes: 3, + min_calibration_frames: 10, + min_calibration_duration_s: 0.0, + baseline_expiry_s: 2.0, + }) + .unwrap(); + for frame_index in 0..20 { + let frame = (0..56) + .map(|subcarrier| { + 10.0 + subcarrier as f64 * 0.01 + + (frame_index as f64 * 0.1 + subcarrier as f64 * 0.03).sin() * 0.01 + }) + .collect(); + model.feed_calibration(&[frame]).unwrap(); + } + model + .finalize_calibration(calibrated_at_unix_ms.saturating_mul(1_000), 7) + .unwrap(); + model + } + + fn state_with_model(bootstrap: bool) -> AppStateInner { + let model = finalized_model(1_000); + let baseline = model.modes().unwrap().baseline[0].clone(); + let mut state = AppStateInner::minimal(); + state.frame_history = (0..50).map(|_| baseline.clone()).collect(); + state.field_model = Some(model); + state.bootstrap_baseline_active = bootstrap; + if bootstrap { + state.bootstrap_baseline = Some(BootstrapBaselineMetadata { + authority: bootstrap_baseline::BOOTSTRAP_BASELINE_AUTHORITY, + source_node_ids: vec![5], + source_model_id: "test-model".to_string(), + created_at_unix_ms: 1_000, + expires_at_unix_ms: 3_000, + content_sha256: "00".repeat(32), + }); + } + state + } + + #[test] + fn bootstrap_field_model_remains_negative_only_for_person_count() { + let runtime = state_with_model(false); + assert_eq!(runtime.person_count_at(1_500), 0); + + let bootstrap = state_with_model(true); + assert_eq!( + bootstrap.person_count_at(1_500), + score_to_person_count(0.0, 0), + "a bootstrap prior must not derive a positive or negative count through the runtime model path" + ); + } + + #[tokio::test] + async fn long_running_expiry_is_not_reported_active_for_runtime_or_bootstrap() { + let runtime = Arc::new(RwLock::new(state_with_model(false))); + let Json(runtime_status) = calibration_status(State(runtime)).await; + assert_eq!(runtime_status["active"], false); + assert_eq!(runtime_status["status"], "expired"); + assert_eq!(runtime_status["binding_mode"], "none"); + + let bootstrap = Arc::new(RwLock::new(state_with_model(true))); + let Json(bootstrap_status) = calibration_status(State(bootstrap)).await; + assert_eq!(bootstrap_status["active"], false); + assert_eq!(bootstrap_status["status"], "expired"); + assert_eq!(bootstrap_status["binding_mode"], "none"); + assert_eq!(bootstrap_status["bootstrap_baseline"]["stored"], true); + assert_eq!(bootstrap_status["bootstrap_baseline"]["active"], false); + assert_eq!( + bootstrap_status["bootstrap_baseline"]["calibrated_evidence_authorized"], + false + ); + assert_eq!( + bootstrap_status["bootstrap_baseline"]["numeric_vitals_authorized"], + false + ); + } +} + // ── ESP32 Edge Vitals Packet (ADR-039, magic 0xC511_0002) ──────────────────── /// Decoded vitals packet from ESP32 edge processing pipeline. @@ -6319,7 +6439,9 @@ async fn calibration_start( } // Guard: don't discard an in-progress or fresh calibration if let Some(ref fm) = s.field_model { - match fm.status() { + let observed_at_us = + (chrono::Utc::now().timestamp_millis().max(0) as u64).saturating_mul(1_000); + match field_bridge::calibration_status_at(fm, observed_at_us) { CalibrationStatus::Collecting => { return Json(serde_json::json!({ "success": false, @@ -6423,6 +6545,9 @@ async fn calibration_stop(State(state): State) -> Json) -> Json { let s = state.read().await; let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; + let effective_status = s.field_model_status_at(observed_at_unix_ms); + let active = s.field_model_active_at(observed_at_unix_ms); + let bootstrap_active = s.bootstrap_baseline_active_at(observed_at_unix_ms); let bootstrap_background_match = s.bootstrap_background_match(observed_at_unix_ms); let runtime_reference = s.field_model.as_ref().and_then(|model| { model.modes().map(|modes| serde_json::json!({ @@ -6434,21 +6559,22 @@ async fn calibration_status(State(state): State) -> Json) -> Json Json { let expected_model_id = { let s = state.read().await; - if s.bootstrap_baseline_active - || !s.field_model.as_ref().is_some_and(|model| { - matches!(model.status(), CalibrationStatus::Fresh) - }) - { + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; + if s.bootstrap_baseline_active || !s.explicit_calibration_fresh_at(observed_at_unix_ms) { return Json(serde_json::json!({ "success": false, "error_code": "calibration_not_finalized", @@ -6540,8 +6663,9 @@ async fn calibration_promote_bootstrap( while tokio::time::Instant::now() < interval_deadline { let observed = { let s = state.read().await; + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; if s.calibration_model_id.as_deref() != Some(expected_model_id.as_str()) - || s.bootstrap_baseline_active + || !s.explicit_calibration_fresh_at(observed_at_unix_ms) { None } else { @@ -6581,8 +6705,9 @@ async fn calibration_promote_bootstrap( } let mut s = state.write().await; + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; if s.calibration_model_id.as_deref() != Some(expected_model_id.as_str()) - || s.bootstrap_baseline_active + || !s.explicit_calibration_fresh_at(observed_at_unix_ms) { return Json(serde_json::json!({ "success": false, @@ -6661,7 +6786,11 @@ async fn calibration_refine_bootstrap( let (expected_model_id, expected_digest, source_node_id) = { let s = state.read().await; - let Some(metadata) = s.bootstrap_baseline.as_ref().filter(|_| s.bootstrap_baseline_active) + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; + let Some(metadata) = s + .bootstrap_baseline + .as_ref() + .filter(|_| s.bootstrap_baseline_active_at(observed_at_unix_ms)) else { return Json(serde_json::json!({ "success": false, @@ -6706,7 +6835,8 @@ async fn calibration_refine_bootstrap( while tokio::time::Instant::now() < interval_deadline { observed = { let s = state.read().await; - let identity_matches = s.bootstrap_baseline_active + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; + let identity_matches = s.bootstrap_baseline_active_at(observed_at_unix_ms) && s.bootstrap_baseline.as_ref().is_some_and(|metadata| { metadata.source_model_id == expected_model_id && metadata.content_sha256 == expected_digest @@ -6771,7 +6901,8 @@ async fn calibration_refine_bootstrap( } let mut s = state.write().await; - let identity_matches = s.bootstrap_baseline_active + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; + let identity_matches = s.bootstrap_baseline_active_at(observed_at_unix_ms) && s.bootstrap_baseline.as_ref().is_some_and(|metadata| { metadata.source_model_id == expected_model_id && metadata.content_sha256 == expected_digest @@ -6951,11 +7082,10 @@ fn chrono_timestamp() -> u64 { async fn vital_signs_endpoint(State(state): State) -> Json { let s = state.read().await; - let person_count = s.person_count(); - let explicit_calibration_fresh = !s.bootstrap_baseline_active - && s.field_model - .as_ref() - .is_some_and(|model| matches!(model.status(), CalibrationStatus::Fresh)); + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; + let person_count = s.person_count_at(observed_at_unix_ms); + let explicit_calibration_fresh = + s.explicit_calibration_fresh_at(observed_at_unix_ms); let published = vitals_for_publication( &s.latest_vitals, explicit_calibration_fresh, @@ -7036,11 +7166,9 @@ async fn edge_registry_endpoint( /// GET /api/v1/edge-vitals — latest edge vitals from ESP32 (ADR-039). async fn edge_vitals_endpoint(State(state): State) -> Json { let s = state.read().await; - let explicit_single_occupant = !s.bootstrap_baseline_active - && s.field_model - .as_ref() - .is_some_and(|model| matches!(model.status(), CalibrationStatus::Fresh)) - && s.person_count() == 1; + let observed_at_unix_ms = chrono::Utc::now().timestamp_millis().max(0) as u64; + let explicit_single_occupant = s.explicit_calibration_fresh_at(observed_at_unix_ms) + && s.person_count_at(observed_at_unix_ms) == 1; if !explicit_single_occupant { return Json(serde_json::json!({ "status": "abstained", @@ -7654,7 +7782,10 @@ async fn udp_receiver_task( // #803: don't let the saturating activity score // discard count-aware per-node estimates. let count = - aggregate_person_count(s.person_count(), &s.node_states); + aggregate_person_count( + s.person_count_at(observed_at_unix_ms), + &s.node_states, + ); s.prev_person_count = count; count.max(1) // presence=true => at least 1 } @@ -7791,10 +7922,8 @@ async fn udp_receiver_task( heartbeat_confidence: if vitals.presence { 0.7 } else { 0.0 }, signal_quality: vitals.presence_score as f64, }; - let explicit_calibration_fresh = !s.bootstrap_baseline_active - && s.field_model.as_ref().is_some_and(|model| { - matches!(model.status(), CalibrationStatus::Fresh) - }); + let explicit_calibration_fresh = + s.explicit_calibration_fresh_at(observed_at_unix_ms); let published_vitals = vitals_for_publication( &vital_candidates, explicit_calibration_fresh, @@ -8201,7 +8330,10 @@ async fn udp_receiver_task( // #803: don't let the saturating activity score // discard count-aware per-node estimates. let count = - aggregate_person_count(s.person_count(), &s.node_states); + aggregate_person_count( + s.person_count_at(observed_at_unix_ms), + &s.node_states, + ); s.prev_person_count = count; count.max(1) } @@ -8314,10 +8446,8 @@ async fn udp_receiver_task( } else { debounce_room_classification(&mut s, &room_inference) }; - let explicit_calibration_fresh = !s.bootstrap_baseline_active - && s.field_model.as_ref().is_some_and(|model| { - matches!(model.status(), CalibrationStatus::Fresh) - }); + let explicit_calibration_fresh = + s.explicit_calibration_fresh_at(observed_at_unix_ms); let published_vitals = vitals_for_publication( &vitals, explicit_calibration_fresh, diff --git a/v2/crates/wifi-densepose-sensing-server/src/types.rs b/v2/crates/wifi-densepose-sensing-server/src/types.rs index da2630497..e86ceaba9 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/types.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/types.rs @@ -544,6 +544,10 @@ impl AppStateInner { pub fn person_count(&self) -> usize { use crate::csi::score_to_person_count; use crate::field_bridge; + let observed_at_us = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_micros() as u64) + .unwrap_or(0); match self.field_model.as_ref() { Some(fm) => { let history = if !self.frame_history.is_empty() { @@ -559,6 +563,7 @@ impl AppStateInner { field_bridge::occupancy_or_fallback( fm, history, + observed_at_us, self.smoothed_person_score, self.prev_person_count, )