diff --git a/v2/crates/wifi-densepose-cli/src/calibrate_api.rs b/v2/crates/wifi-densepose-cli/src/calibrate_api.rs index 8be6fb034..27a936a05 100644 --- a/v2/crates/wifi-densepose-cli/src/calibrate_api.rs +++ b/v2/crates/wifi-densepose-cli/src/calibrate_api.rs @@ -536,7 +536,7 @@ async fn ingest_loop( if let Some(sess) = active.as_ref() { if Instant::now() >= sess.deadline { let frames = sess.recorder.frames_recorded() as usize; - if frames >= 10 { + if frames >= sess.target_frames { if let Some(done) = active.take() { let _ = finalize(done, &output_dir, &status).await; } @@ -544,8 +544,9 @@ async fn ingest_loop( // not enough frames — abort honestly rather than emit a bad baseline done.motion_flagged = false; let note = format!( - "aborted: only {frames} frames in the time window (need >=10) — \ + "aborted: only {frames} frames in the time window (need >= {}) — \ is the ESP32 streaming to udp:{}? ", + done.target_frames, status.read().await.udp_port ); let snap = session_snapshot(&done, "aborted", Some(note.clone())); @@ -615,19 +616,34 @@ async fn finalize( status.write().await.session = Some(snap); } - let baseline: BaselineCalibration = sess - .recorder - .finalize() - .map_err(|e| format!("finalize failed: {e}"))?; + // `CalibrationRecorder::finalize` consumes the recorder, so retain a + // terminal snapshot before attempting the fallible operation. + let mut aborted_snapshot = session_snapshot(&sess, "aborted", None); + let baseline: BaselineCalibration = match sess.recorder.finalize() { + Ok(baseline) => baseline, + Err(e) => { + let message = format!("finalize failed: {e}"); + let note = format!("aborted: {message}"); + aborted_snapshot.note = Some(note.clone()); + status.write().await.session = Some(aborted_snapshot); + eprintln!("[calibrate-serve] {note}"); + return Err(message); + } + }; let (amp_mean_avg, amp_var_avg, disp_avg) = baseline_averages(&baseline); let uuid = baseline.calibration_uuid().to_string(); let path = format!("{output_dir}/{room_id}-{uuid}.bin"); let bytes = baseline.to_bytes(); // Async write — never block the ingest task's UDP/command path. - tokio::fs::write(&path, &bytes) - .await - .map_err(|e| format!("cannot write {path}: {e}"))?; + if let Err(error) = tokio::fs::write(&path, &bytes).await { + let message = format!("cannot write {path}: {error}"); + let note = format!("aborted: {message}"); + aborted_snapshot.note = Some(note.clone()); + status.write().await.session = Some(aborted_snapshot); + eprintln!("[calibrate-serve] {note}"); + return Err(message); + } let summary = ResultSummary { calibration_id: uuid, @@ -1055,6 +1071,97 @@ mod tests { assert_eq!(a.udp_port, 5005); } + #[tokio::test] + async fn finalize_insufficient_frames_marks_session_aborted() { + let dir = tempfile::tempdir().unwrap(); + let status = Arc::new(RwLock::new(SharedStatus::default())); + let sess = ActiveSession { + recorder: CalibrationRecorder::new(tier_config("ht20")), + room_id: "test".into(), + tier: "ht20".into(), + started: Instant::now(), + deadline: Instant::now(), + target_frames: 1, + z_median: 0.0, + z_max: 0.0, + motion_flagged: false, + }; + + let err = finalize(sess, dir.path().to_str().unwrap(), &status) + .await + .expect_err("an empty recorder cannot finalize"); + assert!(err.contains("finalize failed")); + + let guard = status.read().await; + let session = guard.session.as_ref().expect("terminal session status"); + assert_eq!(session.state, "aborted"); + assert!(session.note.as_deref().unwrap_or_default().contains("finalize failed")); + } + + fn complete_test_session() -> ActiveSession { + let mut config = tier_config("ht20"); + config.min_frames = 2; + let mut recorder = CalibrationRecorder::new(config); + let mut packet = vec![0u8; 20 + 64 * 2]; + packet[..4].copy_from_slice(&0xC511_0001u32.to_le_bytes()); + packet[4] = 1; + packet[5] = 1; + packet[6..8].copy_from_slice(&64u16.to_le_bytes()); + packet[8..12].copy_from_slice(&2437u32.to_le_bytes()); + packet[20..].fill(3); + let frame = parse_csi_packet(&packet, "ht20").unwrap(); + recorder.record(&frame).unwrap(); + recorder.record(&frame).unwrap(); + ActiveSession { + recorder, + room_id: "test".into(), + tier: "ht20".into(), + started: Instant::now(), + deadline: Instant::now(), + target_frames: 2, + z_median: 0.0, + z_max: 0.0, + motion_flagged: false, + } + } + + #[tokio::test] + async fn finalize_write_failure_marks_session_aborted_without_result() { + let dir = tempfile::tempdir().unwrap(); + let blocked = dir.path().join("not-a-directory"); + std::fs::write(&blocked, b"owned test fixture").unwrap(); + let status = Arc::new(RwLock::new(SharedStatus::default())); + let error = finalize(complete_test_session(), blocked.to_str().unwrap(), &status) + .await + .expect_err("a regular file cannot hold the baseline"); + assert!(error.starts_with("cannot write ")); + let guard = status.read().await; + let session = guard.session.as_ref().unwrap(); + assert_eq!(session.state, "aborted"); + assert!(session.note.as_deref().unwrap().contains("cannot write ")); + assert!(guard.last_result.is_none()); + assert_eq!(std::fs::read(blocked).unwrap(), b"owned test fixture"); + } + + #[tokio::test] + async fn finalize_success_publishes_a_readable_baseline_and_complete_status() { + let dir = tempfile::tempdir().unwrap(); + let status = Arc::new(RwLock::new(SharedStatus::default())); + let summary = finalize(complete_test_session(), dir.path().to_str().unwrap(), &status) + .await + .unwrap(); + let bytes = std::fs::read(&summary.output_path).unwrap(); + let baseline = BaselineCalibration::from_bytes(&bytes).unwrap(); + assert_eq!(baseline.frame_count, 2); + assert_eq!(summary.saved_bytes, bytes.len()); + let guard = status.read().await; + let session = guard.session.as_ref().unwrap(); + assert_eq!(session.state, "complete"); + assert_eq!(session.progress, 1.0); + assert!(session.note.is_none()); + assert_eq!(guard.last_result.as_ref().unwrap().calibration_id, summary.calibration_id); + } + #[test] fn sanitize_blocks_path_traversal() { assert_eq!(sanitize_room_id("../../etc/passwd"), "etcpasswd"); 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 2e3f5d5bb..6793cc26f 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/field_bridge.rs @@ -24,6 +24,9 @@ pub const CALIBRATION_MODEL_RECEIPT_SCHEMA: &str = pub const CALIBRATED_PRESENCE_EVIDENCE_SCHEMA: &str = "ruview.calibration.calibrated-presence-evidence.v2"; +/// Schema for bounded explanations, independent from calibration authority. +pub const DETECTOR_DIAGNOSTICS_SCHEMA: &str = "ruview.calibration.detector-diagnostics.v1"; + /// Length-only canonicalizer for calibration frames (issue #1170 pattern, /// shared with `multistatic_bridge`). Raw ESP32 amplitudes arrive at the /// hardware's native width (HT20 ≈ 64, HT40 ≈ 128/192); the FieldModel is @@ -90,7 +93,7 @@ fn mean_frame(frames: &[Vec]) -> Option> { Some(mean) } -fn perturbation_occupancy(field: &FieldModel, frames: &[Vec]) -> Option { +fn perturbation_occupancy_with_diagnostics(field: &FieldModel, frames: &[Vec]) -> Option<(usize, OccupancyDiagnostics)> { let adaptive_threshold = field.empty_room_residual_energy_threshold(); let frame = match adaptive_threshold { Some(_) => mean_frame(frames)?, @@ -122,7 +125,19 @@ fn perturbation_occupancy(field: &FieldModel, frames: &[Vec]) -> Option]) -> Option { + perturbation_occupancy_with_diagnostics(field, frames).map(|(count, _)| count) } /// Provenance for a count produced by the calibrated field model. Callers @@ -143,10 +158,21 @@ impl CalibratedOccupancyMethod { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +/// Explanation of the very same scoring operation; it grants no authority. +#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize)] +pub struct OccupancyDiagnostics { + pub residual_energy: Option, + pub decision_threshold: Option, + pub signed_margin: Option, + pub window_frames: usize, + pub fallback_reason: Option<&'static str>, +} + +#[derive(Debug, Clone, Copy, PartialEq)] pub struct CalibratedOccupancy { pub person_count: usize, pub method: CalibratedOccupancyMethod, + pub diagnostics: OccupancyDiagnostics, } /// Create a FieldModelConfig for single-link mode (one ESP32 node = one link). @@ -284,11 +310,15 @@ pub fn calibrated_occupancy( return Some(CalibratedOccupancy { person_count: person_count.min(MAX_SINGLE_LINK_OCCUPANCY), method: CalibratedOccupancyMethod::Eigenvalue, + diagnostics: OccupancyDiagnostics { + residual_energy: None, decision_threshold: None, signed_margin: None, + window_frames: frames.len(), fallback_reason: None, + }, }); } + let (person_count, diagnostics) = perturbation_occupancy_with_diagnostics(field, &frames)?; Some(CalibratedOccupancy { - person_count: perturbation_occupancy(field, &frames)?, - method: CalibratedOccupancyMethod::PerturbationEnergy, + person_count, method: CalibratedOccupancyMethod::PerturbationEnergy, diagnostics, }) } @@ -799,6 +829,27 @@ mod tests { ); } + #[test] + fn diagnostic_margin_explains_the_same_perturbation_decision() { + let field = fresh_test_model(); + let baseline = field.modes().unwrap().baseline[0].clone(); + for offset in [0.0, 0.3, 5.0, 20.0] { + let history: VecDeque> = (0..50) + .map(|_| baseline.iter().map(|value| value + offset).collect()).collect(); + let result = calibrated_occupancy(&field, &history, 1_500_000).unwrap(); + assert_eq!(result.person_count, occupancy_or_fallback(&field, &history, 1_500_000, 0.0, 0)); + if result.method == CalibratedOccupancyMethod::PerturbationEnergy { + let energy = result.diagnostics.residual_energy.unwrap(); + let threshold = result.diagnostics.decision_threshold.unwrap(); + assert_eq!(result.diagnostics.signed_margin, Some(energy - threshold)); + assert_eq!(result.person_count > 0, energy > threshold); + assert_eq!(result.diagnostics.window_frames, 50); + } else { + assert_eq!(result.diagnostics.signed_margin, None); + } + } + } + #[test] fn occupancy_falls_back_after_expiry_without_a_process_restart() { 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 62b0eecf0..d35f8830f 100644 --- a/v2/crates/wifi-densepose-sensing-server/src/main.rs +++ b/v2/crates/wifi-densepose-sensing-server/src/main.rs @@ -474,6 +474,19 @@ struct CalibratedPresenceEvidence { inference_method: String, presence: bool, person_count: usize, + #[serde(skip)] + diagnostics: Option, +} + +/// Bounded decision explanation, separate from the strict authority envelope. +#[derive(Debug, Clone, Serialize)] +struct DetectorDiagnostics { + schema: &'static str, + scored_node_id: u8, + source_age_ms: u64, + source_sequence: u32, + required_window_frames: usize, + runtime: field_bridge::OccupancyDiagnostics, } /// Sensing update broadcast to WebSocket clients @@ -495,6 +508,8 @@ struct SensingUpdate { /// active model receipt. Omitted when no calibrated result is available. #[serde(default, skip_serializing_if = "Option::is_none")] calibrated_presence_evidence: Option, + #[serde(skip_deserializing, skip_serializing_if = "Option::is_none")] + detector_diagnostics: Option, // ── ADR-022 Phase 3: Enhanced multi-BSSID pipeline fields ── /// Enhanced motion estimate from multi-BSSID pipeline. #[serde(skip_serializing_if = "Option::is_none")] @@ -2793,6 +2808,17 @@ impl AppStateInner { inference_method: occupancy.method.wire_name().to_string(), presence: occupancy.person_count > 0, person_count: occupancy.person_count, + diagnostics: self.calibration_grid_binding.and_then(|binding| { + let node = self.node_states.get(&binding.source_node_id)?; + Some(DetectorDiagnostics { + schema: field_bridge::DETECTOR_DIAGNOSTICS_SCHEMA, + scored_node_id: binding.source_node_id, + source_age_ms: node.field_model_latest_seen?.elapsed().as_millis().min(u64::MAX as u128) as u64, + source_sequence: node.field_model_latest_sequence?, + required_window_frames: self.field_model.as_ref()?.modes()?.baseline_runtime_window_size.unwrap_or(50), + runtime: occupancy.diagnostics, + }) + }), }) } @@ -3235,6 +3261,21 @@ mod calibration_expiry_tests { state } + #[test] + fn diagnostics_stay_outside_the_strict_calibration_authority_envelope() { + let state = state_with_runtime_window(); + let evidence = state.calibrated_presence_evidence(5, 77, 1_500).unwrap(); + let diagnostic = evidence.diagnostics.as_ref().expect("bound source diagnostics"); + assert_eq!(diagnostic.scored_node_id, 5); + assert_eq!(diagnostic.runtime.window_frames, 50); + let encoded = serde_json::to_value(&evidence).unwrap(); + assert!(encoded.get("diagnostics").is_none()); + assert_eq!(encoded.as_object().unwrap().len(), 13); + assert_eq!(encoded["person_count"], 0); + assert_eq!(serde_json::to_value(diagnostic).unwrap()["schema"], + field_bridge::DETECTOR_DIAGNOSTICS_SCHEMA); + } + #[test] fn calibrated_empty_blocks_vitals_even_when_room_heuristic_reports_one() { let state = state_with_runtime_window(); @@ -5058,6 +5099,7 @@ async fn wifi_task(state: SharedState, tick_ms: u64) { ), vital_signs: None, calibrated_presence_evidence: None, + detector_diagnostics: None, enhanced_motion, enhanced_breathing, posture: posture_str, @@ -5227,6 +5269,7 @@ async fn windows_wifi_fallback_tick(state: &SharedState, seq: u32) { ), vital_signs: None, calibrated_presence_evidence: None, + detector_diagnostics: None, enhanced_motion: None, enhanced_breathing: None, posture: None, @@ -6117,6 +6160,7 @@ mod latest_update_staleness_tests { classifier: None, activity: None, calibrated_presence_evidence: None, + detector_diagnostics: None, enhanced_motion: None, enhanced_breathing: None, posture: None, @@ -11720,6 +11764,7 @@ fn mediatek_room_update( }, vital_signs: None, calibrated_presence_evidence: None, + detector_diagnostics: None, enhanced_motion: None, enhanced_breathing: None, posture: None, @@ -12640,6 +12685,7 @@ async fn udp_receiver_task( classification, signal_field, vital_signs: published_vitals, + detector_diagnostics: calibrated_presence_evidence.as_ref().and_then(|e| e.diagnostics.clone()), calibrated_presence_evidence, enhanced_motion: None, enhanced_breathing: None, @@ -13135,6 +13181,7 @@ async fn udp_receiver_task( &sub_variances, ), vital_signs: published_vitals, + detector_diagnostics: calibrated_presence_evidence.as_ref().and_then(|e| e.diagnostics.clone()), calibrated_presence_evidence, enhanced_motion: None, enhanced_breathing: None, @@ -13375,6 +13422,7 @@ async fn simulated_data_task(state: SharedState, tick_ms: u64) { ), vital_signs: None, calibrated_presence_evidence: None, + detector_diagnostics: None, enhanced_motion: None, enhanced_breathing: None, posture: None, @@ -16079,6 +16127,7 @@ mod observatory_persons_field_position_tests { signal_field, vital_signs: None, calibrated_presence_evidence: None, + detector_diagnostics: None, enhanced_motion: None, enhanced_breathing: None, posture: None,