From d3c1082c03ca1498f15b12983e21082d8f066481 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 10 Jun 2026 03:58:00 +0000 Subject: [PATCH] fix(worldgraph): bound SemanticState growth with deterministic retention MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StreamingEngine::process_cycle appended one SemanticState belief per cycle with no eviction — ~1.7M nodes/day at 20 Hz (beyond-SOTA roadmap finding #6). Add WorldGraph::prune_semantic_states(max): deterministic eviction of the oldest beliefs by (valid_from_unix_ms, id); structural nodes (rooms, zones, sensors, anchors, tracks, events) are never eligible. Wire it into the engine after each belief append (DEFAULT_SEMANTIC_RETENTION = 7,200, ~6 min at 20 Hz; set_semantic_retention to tune). The WorldGraph holds current beliefs; durable history is the recorder's job, so no audit data is lost. 3 new tests: end-to-end bounded growth, oldest-only eviction, deterministic equal-timestamp tie-break. Workspace gate: 2,865 passed, 0 failed. https://claude.ai/code/session_01MjBucx95K4BuUxZi8NWwRH --- CHANGELOG.md | 1 + v2/crates/wifi-densepose-engine/src/lib.rs | 45 +++++++++ .../wifi-densepose-worldgraph/src/graph.rs | 91 +++++++++++++++++++ 3 files changed, 137 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d678006f..305ec72a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **RF tomography solver hoisting** — ISTA gradient buffer no longer allocated inside the 100-iteration loop, and the Frobenius Lipschitz bound moved from per-`reconstruct` to construction (`ruvsense/tomography.rs`). Bit-identical results. ### Fixed +- **WorldGraph no longer grows unboundedly under the live loop.** `StreamingEngine::process_cycle` appended one `SemanticState` belief per cycle with no eviction — ~1.7M nodes/day at 20 Hz (identified in `docs/research/ruview-beyond-sota/04-optimization-roadmap.md`). Added `WorldGraph::prune_semantic_states(max)` — deterministic eviction of the oldest beliefs by `(valid_from_unix_ms, id)`, structural nodes (rooms/zones/sensors/anchors/tracks/events) never eligible — and wired it into the engine after each belief append (`StreamingEngine::DEFAULT_SEMANTIC_RETENTION` = 7,200 ≈ 6 min at 20 Hz; tunable via `set_semantic_retention`). The WorldGraph holds *current* beliefs; durable history is the recorder's job, so no audit data is lost. 3 new tests (bounded growth end-to-end, oldest-only eviction, deterministic tie-break). - **ESP32 edge heart rate no longer stuck at ~45 BPM / dropping wildly — #987.** The on-device HR estimator (`edge_processing.c`, `0xC5110002`) reported ~45 BPM regardless of true heart rate (Apple-Watch ground truth 87 BPM read as ~45) and swung frame-to-frame. Two root causes: (1) a hardcoded `sample_rate = 10.0f` that became wrong after #985's self-ping raised the CSI callback rate to a variable ~13–19 Hz — BPM scales as `assumed/actual × true`, so 87 read ~45 and the reading swung as CSI yield fluctuated; (2) the zero-crossing estimator locked onto a breathing harmonic (a 0.25 Hz breathing fundamental puts its 3rd harmonic at ~0.74 Hz ≈ 44 BPM inside the HR band). Fix: measure the real sample rate from inter-frame timestamps (used for BPM conversion + biquad re-tuning on >15% drift); replace the HR zero-crossing with an autocorrelation estimator that rejects breathing harmonics (driven by a robust autocorr breathing period); median-13 smooth the output. Hardware A/B (fixed vs unmodified control board, both `edge_tier=2`): control pegged 40–49 BPM; fixed reaches the true 88–91 BPM (vs 87 GT) and holds a stable physiological value (spread 59→0 for a steady subject). Known limitation: heavy subject motion still degrades the estimate (motion gating is a follow-up). - **Person count no longer leaks up to 10 in heuristic mode — addresses #894.** `field_bridge::occupancy_or_fallback` returned the eigenvalue-based `FieldModel::estimate_occupancy` count **unbounded** (its internal ceiling is 10), while the sibling estimators on the same single-link data — the perturbation-energy fallback right below it and `score_to_person_count` — both cap at 3 ("1-3 for single ESP32"). On noisy / under-calibrated CSI the eigenvalue count inflated, producing the "10 persons reported when 1 present" symptom (seen when `--model` fails to load and the server runs on heuristics). Bounded the eigenvalue path to the shared `MAX_SINGLE_LINK_OCCUPANCY` (3) so every estimator on one link agrees; genuine higher counts come from the multistatic fusion path, not a single-link covariance estimate. - **MQTT multi-node deployments now create one Home-Assistant device per node — closes #898.** After the #872 MQTT wiring landed, the JSON→`VitalsSnapshot` bridge hard-coded a single `node_id` (the MQTT client id) and the publisher used a single `OwnedDiscoveryBuilder`, so every physical node collapsed into one device (`identifiers:["wifi_densepose_wifi-densepose-1"]`), contradicting the "one device per node" docs. The bridge now emits one snapshot per node in the sensing update's `nodes[]` (each with its own `node_id` + RSSI, falling back to a single aggregate snapshot for wifi/simulate sources), and the publisher derives a per-node builder (`OwnedDiscoveryBuilder::for_node`) that publishes discovery + availability lazily on first sight of each `node_id` and routes state to per-node topics — yielding N distinct HA devices with per-node availability/LWT. Unit-tested (distinct nodes → distinct `wifi_densepose_` identifiers); 71 MQTT tests pass. diff --git a/v2/crates/wifi-densepose-engine/src/lib.rs b/v2/crates/wifi-densepose-engine/src/lib.rs index 10d31ea0..994b8b35 100644 --- a/v2/crates/wifi-densepose-engine/src/lib.rs +++ b/v2/crates/wifi-densepose-engine/src/lib.rs @@ -116,6 +116,10 @@ pub struct StreamingEngine { slam: RfSlam, // ADR-139 live loop: stable track_id -> PersonTrack WorldId. person_tracks: BTreeMap, + // WorldGraph belief retention: max live SemanticState nodes. The live loop + // appends one belief per cycle (1.7M/day at 20 Hz); durable history is the + // recorder's job, so old beliefs are evicted deterministically past this cap. + semantic_retention: usize, } impl StreamingEngine { @@ -135,9 +139,20 @@ impl StreamingEngine { evolution: None, slam: RfSlam::with_discovery(0.5, 5, 0.6), person_tracks: BTreeMap::new(), + semantic_retention: Self::DEFAULT_SEMANTIC_RETENTION, } } + /// Default cap on live `SemanticState` beliefs in the WorldGraph + /// (~6 minutes of full-rate history at 20 Hz; older beliefs are evicted — + /// durable history belongs to the recorder). + pub const DEFAULT_SEMANTIC_RETENTION: usize = 7_200; + + /// Override the `SemanticState` retention cap (minimum 1). + pub fn set_semantic_retention(&mut self, max_states: usize) { + self.semantic_retention = max_states.max(1); + } + /// ADR-139 live loop: create or update a `PersonTrack` node by stable /// `track_id`, locate it in `room`, and wire an `Observes` edge from /// `sensor` (so the privacy rollup can suppress it under identity-strict @@ -350,6 +365,10 @@ impl StreamingEngine { provenance.clone(), &[room], ); + // Retention: bound the live belief set (one node is appended per cycle; + // without this the graph grows ~1.7M nodes/day at 20 Hz). Deterministic + // eviction; the just-added belief is always newest and survives. + self.world.prune_semantic_states(self.semantic_retention); // 7. Deterministic witness over the trust decision (ADR-137 §2.7). let witness = witness_of(&provenance, effective_class); @@ -547,6 +566,32 @@ mod tests { assert_eq!(o1.quality.per_node_weights, o2.quality.per_node_weights); } + /// WorldGraph belief retention: the live loop appends one SemanticState per + /// cycle; past the cap the oldest beliefs are evicted so graph memory is + /// bounded, while structural nodes and the newest belief always survive. + #[test] + fn semantic_state_growth_is_bounded() { + let (mut e, room) = engine(); + e.set_semantic_retention(5); + let cal = CalibrationId(1); + let mut last_id = None; + let baseline_nodes = 2; // room + sensor + for i in 0..20u64 { + let frames = [ + node_frame(0, 1000 + i * 50_000, 56), + node_frame(1, 1001 + i * 50_000, 56), + ]; + let out = e.process_cycle(&frames, cal, room, 5_000 + i as i64).unwrap(); + last_id = Some(out.semantic_id); + assert!(e.world().node_count() <= baseline_nodes + 5); + } + // 20 cycles ran, only 5 beliefs remain, newest is still present. + assert_eq!(e.world().node_count(), baseline_nodes + 5); + assert!(e.world().node(last_id.unwrap()).is_some()); + // Structural nodes survive eviction. + assert!(e.world().node(room).is_some()); + } + fn node_frame_scaled(node_id: u8, ts_us: u64, n_sub: usize, scale: f32) -> MultiBandCsiFrame { MultiBandCsiFrame { node_id, diff --git a/v2/crates/wifi-densepose-worldgraph/src/graph.rs b/v2/crates/wifi-densepose-worldgraph/src/graph.rs index b6a29921..6ffad2e2 100644 --- a/v2/crates/wifi-densepose-worldgraph/src/graph.rs +++ b/v2/crates/wifi-densepose-worldgraph/src/graph.rs @@ -201,6 +201,47 @@ impl WorldGraph { id } + /// Retention: evict the oldest `SemanticState` nodes (with their incident + /// edges) until at most `max_states` remain. Returns the evicted ids, + /// oldest first. + /// + /// The live loop appends one belief per cycle (`StreamingEngine:: + /// process_cycle`), which at 20 Hz is ~1.7M nodes/day — unbounded without + /// this. The WorldGraph holds *current* beliefs; durable history belongs to + /// the recorder (`homecore-recorder`), so evicting old beliefs loses no + /// audit data. + /// + /// Deterministic: eviction order is ascending `(valid_from_unix_ms, id)`, + /// so replaying the same cycle sequence prunes identically. Only + /// `SemanticState` nodes are eligible — rooms, zones, sensors, anchors, + /// person tracks, and events are never evicted by this method. + pub fn prune_semantic_states(&mut self, max_states: usize) -> Vec { + let mut states: Vec<(i64, u64)> = self + .inner + .node_weights() + .filter_map(|n| match n { + WorldNode::SemanticState { id, valid_from_unix_ms, .. } => { + Some((*valid_from_unix_ms, id.0)) + } + _ => None, + }) + .collect(); + if states.len() <= max_states { + return Vec::new(); + } + states.sort_unstable(); + let n_evict = states.len() - max_states; + states.truncate(n_evict); + states + .into_iter() + .map(|(_, raw)| { + let id = WorldId(raw); + self.remove_node(id); + id + }) + .collect() + } + /// Record a contradiction between two still-live beliefs (ADR-139 §2.3). /// Neither node is deleted — the disagreement stays queryable. /// @@ -424,6 +465,56 @@ mod tests { assert!(g.neighbors(s1).iter().any(|(_, e)| matches!(e, WorldEdge::Contradicts { .. }))); } + #[test] + fn prune_semantic_states_evicts_oldest_only() { + let mut g = WorldGraph::new(GeoRegistration::default()); + let room = g.upsert_node(living_room()); + let prov = SemanticProvenance { + evidence: vec!["ev:abc".into()], + model_version: "rfenc-1.0".into(), + calibration_version: "cal:uuid".into(), + privacy_decision: "PrivateHome/Allow".into(), + }; + let ids: Vec = (0..10) + .map(|t| g.add_semantic_state(format!("s{t}"), 0.9, t, prov.clone(), &[room])) + .collect(); + assert_eq!(g.node_count(), 11); // room + 10 beliefs + + let evicted = g.prune_semantic_states(3); + // Oldest 7 evicted, in ascending timestamp order. + assert_eq!(evicted, ids[..7].to_vec()); + assert_eq!(g.node_count(), 4); // room + 3 newest beliefs + for kept in &ids[7..] { + assert!(g.node(*kept).is_some()); + } + // The room (structural node) is never eligible for eviction. + assert!(g.node(room).is_some()); + // Below the cap, pruning is a no-op. + assert!(g.prune_semantic_states(3).is_empty()); + } + + #[test] + fn prune_is_deterministic_for_equal_timestamps() { + let prov = SemanticProvenance { + evidence: vec![], + model_version: "m".into(), + calibration_version: "c".into(), + privacy_decision: "p".into(), + }; + let build = || { + let mut g = WorldGraph::new(GeoRegistration::default()); + let room = g.upsert_node(living_room()); + for _ in 0..6 { + // Identical timestamps: tie-break must fall back to id order. + g.add_semantic_state("s".into(), 0.5, 100, prov.clone(), &[room]); + } + g + }; + let mut g1 = build(); + let mut g2 = build(); + assert_eq!(g1.prune_semantic_states(2), g2.prune_semantic_states(2)); + } + #[test] fn privacy_rollup_suppresses_person_tracks() { let mut g = WorldGraph::new(GeoRegistration::default());