Merge pull request #2246 from ruvnet/codex/sensing-release-20261012

fix(calibration): finish failed sessions and reconcile decision diagnostics

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");

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<f64>]) -> Option<Vec<f64>> {
     Some(mean)
 }
 
-fn perturbation_occupancy(field: &FieldModel, frames: &[Vec<f64>]) -> Option<usize> {
+fn perturbation_occupancy_with_diagnostics(field: &FieldModel, frames: &[Vec<f64>]) -> 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<f64>]) -> Option<usi
             }
         }
     };
-    Some(count)
+    let threshold = adaptive_threshold.unwrap_or(1.0).max(1.0);
+    let energy = perturbation.total_energy;
+    Some((count, OccupancyDiagnostics {
+        residual_energy: energy.is_finite().then_some(energy),
+        decision_threshold: Some(threshold),
+        signed_margin: (energy - threshold).is_finite().then_some(energy - threshold),
+        window_frames: frames.len(),
+        fallback_reason: Some("eigenvalue_estimator_unavailable"),
+    }))
+}
+
+fn perturbation_occupancy(field: &FieldModel, frames: &[Vec<f64>]) -> Option<usize> {
+    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<f64>,
+    pub decision_threshold: Option<f64>,
+    pub signed_margin: Option<f64>,
+    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<Vec<f64>> = (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();

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<DetectorDiagnostics>,
+}
+
+/// 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<CalibratedPresenceEvidence>,
+    #[serde(skip_deserializing, skip_serializing_if = "Option::is_none")]
+    detector_diagnostics: Option<DetectorDiagnostics>,
     // ── 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,