mirror of
https://github.com/ruvnet/RuView
synced 2026-08-08 20:11:43 +00:00
fix(worldgraph): bound SemanticState growth with deterministic retention
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
This commit is contained in:
@@ -116,6 +116,10 @@ pub struct StreamingEngine {
|
||||
slam: RfSlam,
|
||||
// ADR-139 live loop: stable track_id -> PersonTrack WorldId.
|
||||
person_tracks: BTreeMap<u64, WorldId>,
|
||||
// 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,
|
||||
|
||||
@@ -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<WorldId> {
|
||||
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<WorldId> = (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());
|
||||
|
||||
Reference in New Issue
Block a user