fix(sensing-server): train the adaptive classifier per node, as it runs

Runtime classification is per node, from that node's own features and
amplitudes. The trainer instead built one sample per recorded line from the
fused frame-level features plus the first node's amplitudes, so a model
trained on a multi-node recording learned a different input distribution
from the one it is later asked to classify.

- load_recording now emits one sample per node per new frame, from that
  node's node_features entry and amplitude vector. A node contributes only
  when its amplitude vector changed since the previous line (every broadcast
  snapshots all nodes), and stale nodes are skipped. Lines without
  node_features (older single-node recordings) fall back to the frame-level
  features.
- Runtime subcarrier statistics use the same 56-subcarrier window that
  sensing_update, and therefore every recording, stores
  (RECORDED_AMPLITUDE_LEN, now shared with the broadcast in main.rs).
- The trainer stamps model version 2 (TRAINED_MODEL_VERSION), so a freshly
  trained model consumes the measured temporal dominant frequency (ADR-356)
  instead of being switched to the legacy subcarrier proxy at runtime.
- The training accuracy log line is labelled in-sample and printed as a
  percentage (it printed the 0..1 fraction with a % sign).

Tests: cargo test -p wifi-densepose-sensing-server --no-default-features:
bin 337 passed (5 new node_sample_tests), lib 539 passed,
integration tests pass. Clippy reports nothing on the changed lines.

Co-Authored-By: claude-flow <ruv@ruv.net>

v2/crates/wifi-densepose-sensing-server/src/adaptive_classifier.rs

@@ -27,6 +27,65 @@ const N_FEATURES: usize = 15;
 /// Default class names for backward compatibility with old saved models.
 const DEFAULT_CLASSES: &[&str] = &["absent", "present_still", "present_moving", "active"];
 
+/// Model version written by the trainer. Version 2 models consume the measured
+/// temporal `dominant_freq_hz` that recordings carry (ADR-356); stamping 1 here
+/// would make the server swap in the legacy subcarrier proxy at runtime.
+pub const TRAINED_MODEL_VERSION: u32 = 2;
+
+/// Subcarrier amplitudes kept per node in the broadcast `sensing_update`, and
+/// therefore in every recording. Runtime classification truncates to the same
+/// length so the subcarrier statistics match what training saw.
+pub const RECORDED_AMPLITUDE_LEN: usize = 56;
+
+/// Extract one feature vector per node from a recorded `sensing_update` line.
+///
+/// Runtime classification is per node, from that node's own features and
+/// amplitudes. Each broadcast carries a snapshot of every node, so a node only
+/// contributes when its amplitude vector changed since the previous line
+/// (`last_amps`), i.e. when it delivered a new frame. Stale nodes are skipped.
+/// Lines without `node_features` (older single-node recordings) fall back to
+/// the frame-level features and the first node.
+pub fn node_samples_from_frame(
+    frame: &serde_json::Value,
+    last_amps: &mut HashMap<u64, Vec<f64>>,
+) -> Vec<[f64; N_FEATURES]> {
+    let Some(per_node) = frame.get("node_features").and_then(|n| n.as_array()) else {
+        return vec![features_from_frame(frame)];
+    };
+    let nodes = frame
+        .get("nodes")
+        .and_then(|n| n.as_array())
+        .map(Vec::as_slice)
+        .unwrap_or(&[]);
+    let mut samples = Vec::new();
+    for entry in per_node {
+        if entry
+            .get("stale")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false)
+        {
+            continue;
+        }
+        let Some(node_id) = entry.get("node_id").and_then(|v| v.as_u64()) else {
+            continue;
+        };
+        let amps: Vec<f64> = nodes
+            .iter()
+            .find(|n| n.get("node_id").and_then(|v| v.as_u64()) == Some(node_id))
+            .and_then(|n| n.get("amplitude"))
+            .and_then(|a| a.as_array())
+            .map(|arr| arr.iter().filter_map(|v| v.as_f64()).collect())
+            .unwrap_or_default();
+        if amps.is_empty() || last_amps.get(&node_id) == Some(&amps) {
+            continue;
+        }
+        let feat = entry.get("features").cloned().unwrap_or_default();
+        samples.push(features_from_runtime(&feat, &amps));
+        last_amps.insert(node_id, amps);
+    }
+    samples
+}
+
 /// Extract extended feature vector from a JSONL frame (features + raw amplitudes).
 pub fn features_from_frame(frame: &serde_json::Value) -> [f64; N_FEATURES] {
     let feat = frame
@@ -118,6 +177,7 @@ pub fn features_from_runtime(feat: &serde_json::Value, amps: &[f64]) -> [f64; N_
         .get("mean_rssi")
         .and_then(|v| v.as_f64())
         .unwrap_or(0.0);
+    let amps = &amps[..amps.len().min(RECORDED_AMPLITUDE_LEN)];
     let (amp_mean, amp_std, amp_skew, amp_kurt, amp_iqr, amp_entropy, amp_max, amp_range) =
         subcarrier_stats(amps);
     [
@@ -349,15 +409,14 @@ fn load_recording(path: &Path, class_idx: usize) -> Vec<Sample> {
         Ok(c) => c,
         Err(_) => return Vec::new(),
     };
+    let mut last_amps = HashMap::new();
     content
         .lines()
-        .filter_map(|line| {
-            let v: serde_json::Value = serde_json::from_str(line).ok()?;
-            // Use extended features (server features + subcarrier stats).
-            Some(Sample {
-                features: features_from_frame(&v),
-                class_idx,
-            })
+        .filter_map(|line| serde_json::from_str::<serde_json::Value>(line).ok())
+        .flat_map(|v| node_samples_from_frame(&v, &mut last_amps))
+        .map(|features| Sample {
+            features,
+            class_idx,
         })
         .collect()
 }
@@ -630,7 +689,11 @@ pub fn train_from_recordings(recordings_dir: &Path) -> Result<AdaptiveModel, Str
         }
     }
     let accuracy = correct as f64 / n as f64;
-    eprintln!("Training accuracy: {correct}/{n} = {accuracy:.1}%");
+    // In-sample only: this is not an evaluation and must not be quoted as one.
+    eprintln!(
+        "Training accuracy (in-sample): {correct}/{n} = {:.1}%",
+        accuracy * 100.0
+    );
 
     // ── Per-class accuracy ──
     let mut class_correct = vec![0usize; n_classes];
@@ -666,7 +729,7 @@ pub fn train_from_recordings(recordings_dir: &Path) -> Result<AdaptiveModel, Str
         global_std,
         trained_frames: n,
         training_accuracy: accuracy,
-        version: 1,
+        version: TRAINED_MODEL_VERSION,
         class_names,
     })
 }
@@ -676,6 +739,90 @@ pub fn model_path() -> PathBuf {
     PathBuf::from("data/adaptive_model.json")
 }
 
+#[cfg(test)]
+mod node_sample_tests {
+    use super::*;
+    use serde_json::json;
+
+    fn node_features(node_id: u64, variance: f64, stale: bool) -> serde_json::Value {
+        json!({
+            "node_id": node_id,
+            "features": { "variance": variance, "mean_rssi": -50.0, "change_points": 3 },
+            "stale": stale,
+        })
+    }
+
+    fn node(node_id: u64, amplitude: &[f64]) -> serde_json::Value {
+        json!({ "node_id": node_id, "amplitude": amplitude })
+    }
+
+    #[test]
+    fn one_sample_per_node_using_that_nodes_features() {
+        let frame = json!({
+            "features": { "variance": 999.0 },
+            "nodes": [node(1, &[1.0, 2.0]), node(2, &[3.0, 4.0])],
+            "node_features": [node_features(1, 10.0, false), node_features(2, 20.0, false)],
+        });
+        let samples = node_samples_from_frame(&frame, &mut HashMap::new());
+        let mut variances: Vec<f64> = samples.iter().map(|s| s[0]).collect();
+        variances.sort_by(|a, b| a.partial_cmp(b).unwrap());
+        assert_eq!(variances, vec![10.0, 20.0], "per-node, never the fused 999");
+    }
+
+    #[test]
+    fn repeated_snapshot_of_an_unchanged_node_is_not_resampled() {
+        let mut last = HashMap::new();
+        let first = json!({
+            "nodes": [node(1, &[1.0, 2.0]), node(2, &[3.0, 4.0])],
+            "node_features": [node_features(1, 10.0, false), node_features(2, 20.0, false)],
+        });
+        // Node 2 delivered a new frame; node 1's entry is the same snapshot.
+        let second = json!({
+            "nodes": [node(1, &[1.0, 2.0]), node(2, &[5.0, 6.0])],
+            "node_features": [node_features(1, 10.0, false), node_features(2, 21.0, false)],
+        });
+        assert_eq!(node_samples_from_frame(&first, &mut last).len(), 2);
+        let again = node_samples_from_frame(&second, &mut last);
+        assert_eq!(again.len(), 1);
+        assert_eq!(again[0][0], 21.0);
+    }
+
+    #[test]
+    fn stale_and_amplitude_less_nodes_are_skipped() {
+        let frame = json!({
+            "nodes": [node(1, &[1.0, 2.0]), node(2, &[])],
+            "node_features": [node_features(1, 10.0, true), node_features(2, 20.0, false)],
+        });
+        assert!(node_samples_from_frame(&frame, &mut HashMap::new()).is_empty());
+    }
+
+    #[test]
+    fn recordings_without_node_features_fall_back_to_frame_level() {
+        let frame = json!({
+            "features": { "variance": 7.0 },
+            "nodes": [node(1, &[1.0, 2.0])],
+        });
+        let samples = node_samples_from_frame(&frame, &mut HashMap::new());
+        assert_eq!(samples.len(), 1);
+        assert_eq!(samples[0][0], 7.0);
+    }
+
+    #[test]
+    fn runtime_uses_the_same_amplitude_window_as_recordings() {
+        let mut long = vec![1.0; RECORDED_AMPLITUDE_LEN];
+        long.extend([100.0; 144]);
+        let short = vec![1.0; RECORDED_AMPLITUDE_LEN];
+        let feat = json!({});
+        assert_eq!(
+            features_from_runtime(&feat, &long),
+            features_from_runtime(&feat, &short)
+        );
+    }
+
+    // The trainer must stamp a version that consumes measured frequency.
+    const _: () = assert!(TRAINED_MODEL_VERSION >= 2);
+}
+
 #[cfg(test)]
 mod compatibility_tests {
     use super::compatible_dominant_frequency;

v2/crates/wifi-densepose-sensing-server/src/main.rs

@@ -10464,7 +10464,12 @@ async fn udp_receiver_task(
                             } else {
                                 n.frame_history
                                     .back()
-                                    .map(|a| a.iter().take(56).cloned().collect())
+                                    .map(|a| {
+                                        a.iter()
+                                            .take(adaptive_classifier::RECORDED_AMPLITUDE_LEN)
+                                            .cloned()
+                                            .collect()
+                                    })
                                     .unwrap_or_default()
                             },
                             subcarrier_count: if suppress_raw {