smolkik_adm/RuView
1
0
Fork 0
mirror of https://github.com/ruvnet/RuView.git synced 2026-08-28 02:55:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sensing): fuse only coherent frame cohorts (#1726)

Browse Source
This commit is contained in:
rUv
2026-08-27 10:10:31 -04:00
committed by GitHub
parent d42c5581f3
commit b742eae7d6
3 changed files with 281 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

95
v2/crates/wifi-densepose-sensing-server/src/engine_bridge.rs
View File

@@ -41,7 +41,7 @@ use wifi_densepose_signal::ruvsense::fusion_quality::CalibrationId;
use wifi_densepose_signal::ruvsense::multistatic::MultistaticConfig;
use wifi_densepose_worldgraph::WorldId;
use super::multistatic_bridge::node_frames_from_states;
use super::multistatic_bridge::node_frames_from_states_with_guard;
use super::NodeState;
/// Minimum spacing between engine-error warn logs (errors are still counted
@@ -53,6 +53,8 @@ const ENGINE_ERROR_WARN_INTERVAL: Duration = Duration::from_secs(10);
/// live sensing loop publishes beliefs into.
pub struct EngineBridge {
engine: StreamingEngine,
/// Hard timestamp guard shared by cohort selection and engine validation.
guard_interval_us: u64,
room: WorldId,
/// Nodes already wired into the WorldGraph as sensors (by `node_id`).
registered_nodes: HashMap<u8, WorldId>,
@@ -94,6 +96,11 @@ impl EngineBridge {
room_name: &str,
multistatic_cfg: Option<MultistaticConfig>,
) -> Self {
let guard_interval_us = multistatic_cfg
.as_ref()
.map_or_else(|| MultistaticConfig::default().guard_interval_us, |cfg| {
cfg.guard_interval_us
});
let mut engine = StreamingEngine::new(mode, model_version, GeoRegistration::default());
if let Some(cfg) = multistatic_cfg {
engine.set_multistatic_config(cfg);
@@ -101,6 +108,7 @@ impl EngineBridge {
let room = engine.add_room(room_area_id, room_name);
Self {
engine,
guard_interval_us,
room,
registered_nodes: HashMap::new(),
calibration: CalibrationId(0x5256_0001), // "RV\0\x01" — placeholder epoch
@@ -165,7 +173,8 @@ impl EngineBridge {
node_states: &HashMap<u8, NodeState>,
now_ms: i64,
) -> Option<Result<TrustedOutput, EngineError>> {
let frames = node_frames_from_states(node_states);
let frames =
node_frames_from_states_with_guard(node_states, self.guard_interval_us);
if frames.is_empty() {
return None;
}
@@ -195,7 +204,15 @@ impl EngineBridge {
node_states: &HashMap<u8, NodeState>,
now_ms: i64,
) -> Option<TrustedOutput> {
match self.process_cycle_from_states(node_states, now_ms)? {
let result = self.process_cycle_from_states(node_states, now_ms)?;
self.record_cycle_result(result)
}
fn record_cycle_result(
&mut self,
result: Result<TrustedOutput, EngineError>,
) -> Option<TrustedOutput> {
match result {
Ok(trust) => {
self.last_witness = Some(trust.witness);
self.recalibration_recommended = trust.recalibration_recommended;
@@ -419,45 +436,29 @@ mod tests {
assert!(!bridge.suppress_raw_outputs());
}
/// Error wiring (review finding 1a): a live cycle that fails fusion yields
/// an `EngineError` — previously dropped by `if let Some(Ok(..))` at the
/// call sites. The counter must increment and the last good trust state
/// must survive a later failure.
///
/// Originally this forced the failure with a 56-vs-30 subcarrier mismatch
/// (`DimensionMismatch`). Since #1170 the live bridge canonicalizes every
/// node onto the 56-tone grid, so heterogeneous counts now fuse cleanly —
/// a frame-timestamp spread wider than the fuser's 60 ms guard interval is
/// the remaining deterministic way to provoke a fusion error here.
/// Error-result accounting is independent from live cohort selection, so
/// a future engine error remains auditable without deliberately forwarding
/// temporally incoherent frames through the production bridge.
#[test]
fn observe_cycle_counts_engine_errors() {
// Both nodes are 56-subcarrier (canonicalization-clean), but their
// frame timestamps are 500 ms apart — far beyond the 60 ms guard —
// so the fuser rejects the cycle with TimestampMismatch. Future
// offsets keep both instants safely after the bridge's lazy EPOCH.
fn mismatched_states() -> HashMap<u8, NodeState> {
let now = Instant::now();
let mut a = node_state_with_history(1.0, 56);
a.last_frame_time = Some(now + std::time::Duration::from_millis(600));
let mut b = node_state_with_history(1.05, 56);
b.last_frame_time = Some(now + std::time::Duration::from_millis(100));
let mut m = HashMap::new();
m.insert(0u8, a);
m.insert(1u8, b);
m
}
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R", None);
let mismatched = mismatched_states();
let error = || {
EngineError::Fusion(
wifi_densepose_signal::ruvsense::multistatic::MultistaticError::TimestampMismatch {
spread_us: 500_000,
guard_us: 60_000,
},
)
};
assert!(bridge.observe_cycle(&mismatched, 1_000).is_none());
assert!(bridge.record_cycle_result(Err(error())).is_none());
assert_eq!(bridge.engine_error_count(), 1);
assert!(
bridge.last_trust_witness().is_none(),
"no witness from a failed cycle"
);
assert!(bridge.observe_cycle(&mismatched, 2_000).is_none());
assert!(bridge.record_cycle_result(Err(error())).is_none());
assert_eq!(bridge.engine_error_count(), 2);
// A later good cycle records trust state; the audit count is kept.
@@ -467,11 +468,39 @@ mod tests {
assert_eq!(bridge.engine_error_count(), 2);
// And a subsequent failure keeps the last good witness readable.
assert!(bridge.observe_cycle(&mismatched, 4_000).is_none());
assert!(bridge.record_cycle_result(Err(error())).is_none());
assert_eq!(bridge.engine_error_count(), 3);
assert!(bridge.last_trust_witness().is_some());
}
#[test]
fn governed_cycles_prune_slow_mixed_width_nodes_without_errors() {
let mut bridge = EngineBridge::new(PrivacyMode::PrivateHome, 1, "r", "R", None);
for tick in 0..50_i64 {
let now = Instant::now();
let mut states = HashMap::new();
for (node_id, age_ms, n_sub) in [
(1, 0, 64),
(3, 10, 256),
(4, 50, 64),
(7, 1_000, 256),
] {
let mut node = node_state_with_history(1.0 + node_id as f64 * 0.01, n_sub);
node.last_frame_time = Some(now - Duration::from_millis(age_ms));
states.insert(node_id, node);
}
assert!(
bridge.observe_cycle(&states, 1_000 + tick * 50).is_some(),
"governed cycle {tick}"
);
}
assert_eq!(bridge.engine_error_count(), 0);
assert!(bridge.last_trust_witness().is_some());
}
/// ADR-141 mapping (review finding 1c): a cycle emitted at class
/// Restricted flips `suppress_raw_outputs`, which `main.rs` uses to strip
/// per-node raw amplitude vectors from the live publish — the same field

241
v2/crates/wifi-densepose-sensing-server/src/multistatic_bridge.rs
View File

@@ -12,7 +12,9 @@ use std::time::{Duration, Instant};
use wifi_densepose_signal::hardware_norm::{CanonicalCsiFrame, HardwareNormalizer, HardwareType};
use wifi_densepose_signal::ruvsense::multiband::MultiBandCsiFrame;
use wifi_densepose_signal::ruvsense::multistatic::{FusedSensingFrame, MultistaticFuser};
use wifi_densepose_signal::ruvsense::multistatic::{
FusedSensingFrame, MultistaticConfig, MultistaticFuser,
};
use super::NodeState;
@@ -44,6 +46,24 @@ static NORMALIZER: LazyLock<HardwareNormalizer> = LazyLock::new(HardwareNormaliz
/// `last_frame_time`.
pub fn node_frame_from_state(node_id: u8, ns: &NodeState) -> Option<MultiBandCsiFrame> {
let last_time = ns.last_frame_time.as_ref()?;
let timestamp_us = ns
.mesh_aligned_us_for_latest_csi_frame()
.unwrap_or_else(|| host_arrival_timestamp_us(last_time));
node_frame_from_state_at(node_id, ns, timestamp_us)
}
fn host_arrival_timestamp_us(last_time: &Instant) -> u64 {
last_time
.checked_duration_since(*EPOCH)
.unwrap_or_default()
.as_micros() as u64
}
fn node_frame_from_state_at(
node_id: u8,
ns: &NodeState,
timestamp_us: u64,
) -> Option<MultiBandCsiFrame> {
let latest = ns.frame_history.back()?;
if latest.is_empty() {
return None;
@@ -60,19 +80,6 @@ pub fn node_frame_from_state(node_id: u8, ns: &NodeState) -> Option<MultiBandCsi
let n_sub = amplitude.len();
let phase = vec![0.0_f32; n_sub];
// Prefer the capture timestamp recovered from the node's mesh sync. This
// keeps UDP scheduling jitter out of the fuser's cross-node guard. Older
// firmware, stale sync state, and frames without the sync-valid bit retain
// the process-local host-arrival fallback.
let timestamp_us = ns
.mesh_aligned_us_for_latest_csi_frame()
.unwrap_or_else(|| {
last_time
.checked_duration_since(*EPOCH)
.unwrap_or_default()
.as_micros() as u64
});
let canonical = CanonicalCsiFrame {
amplitude,
phase,
@@ -88,25 +95,91 @@ pub fn node_frame_from_state(node_id: u8, ns: &NodeState) -> Option<MultiBandCsi
})
}
/// Collect `MultiBandCsiFrame`s from all active nodes.
/// Collect the default-guard coherent `MultiBandCsiFrame` cohort.
///
/// A node is considered active if its `last_frame_time` is within
/// [`STALE_THRESHOLD`] of `now`.
pub fn node_frames_from_states(node_states: &HashMap<u8, NodeState>) -> Vec<MultiBandCsiFrame> {
let now = Instant::now();
let mut frames = Vec::with_capacity(node_states.len());
node_frames_from_states_with_guard(
node_states,
MultistaticConfig::default().guard_interval_us,
)
}
for (&node_id, ns) in node_states {
// Skip stale nodes
if let Some(ref t) = ns.last_frame_time {
if now.duration_since(*t) > STALE_THRESHOLD {
continue;
}
} else {
/// Collect the freshest temporally coherent cohort of active node frames.
///
/// Nodes can publish at very different rates (for example, a mixed S3/C6
/// fleet). `STALE_THRESHOLD` determines whether a node is alive; it does not
/// mean its latest frame belongs to the current sensing cycle. After choosing
/// one timestamp domain for the whole cycle, retain only frames within the
/// fuser's hard guard of the freshest frame. This prevents a slow-but-live node
/// from turning every governed cycle into `TimestampMismatch`, while always
/// preserving at least the freshest node for the supported single-node path.
pub fn node_frames_from_states_with_guard(
node_states: &HashMap<u8, NodeState>,
guard_interval_us: u64,
) -> Vec<MultiBandCsiFrame> {
let now = Instant::now();
let mut active: Vec<(u8, &NodeState)> = node_states
.iter()
.filter_map(|(&node_id, ns)| {
let last_time = ns.last_frame_time.as_ref()?;
(now.duration_since(*last_time) <= STALE_THRESHOLD).then_some((node_id, ns))
})
.collect();
active.sort_unstable_by_key(|(node_id, _)| *node_id);
if active.is_empty() {
return Vec::new();
}
let guard_interval_us = guard_interval_us.max(1);
// Timestamp domains must be selected for the cycle as a whole. A CSI
// frame can legitimately arrive between periodic sync-marked frames. If
// that one node fell back to process-local host time while a peer retained
// mesh epoch time, the resulting hundreds-of-seconds spread made every
// governed fusion cycle fail. Use mesh time only when every active node
// can provide it; otherwise use host-arrival time consistently for all.
let mesh_times: Option<Vec<u64>> = active
.iter()
.map(|(_, ns)| ns.mesh_aligned_us_for_latest_csi_frame())
.collect::<Option<Vec<_>>>()
.filter(|times| {
let Some(min) = times.iter().min() else {
return false;
};
let Some(max) = times.iter().max() else {
return false;
};
max.saturating_sub(*min) <= guard_interval_us
});
let mut timed = Vec::with_capacity(active.len());
for (index, (node_id, ns)) in active.into_iter().enumerate() {
let timestamp_us = mesh_times.as_ref().map_or_else(
|| {
host_arrival_timestamp_us(
ns.last_frame_time.as_ref().expect("active node has time"),
)
},
|times| times[index],
);
timed.push((node_id, ns, timestamp_us));
}
let freshest_timestamp_us = timed
.iter()
.map(|(_, _, timestamp_us)| *timestamp_us)
.max()
.expect("non-empty active cohort");
let mut frames = Vec::with_capacity(timed.len());
for (node_id, ns, timestamp_us) in timed {
if freshest_timestamp_us.saturating_sub(timestamp_us) > guard_interval_us {
continue;
}
if let Some(frame) = node_frame_from_state(node_id, ns) {
if let Some(frame) = node_frame_from_state_at(node_id, ns, timestamp_us) {
frames.push(frame);
}
}
@@ -125,7 +198,8 @@ pub fn fuse_or_fallback(
node_states: &HashMap<u8, NodeState>,
dedup_factor: f64,
) -> (Option<FusedSensingFrame>, Option<usize>) {
let frames = node_frames_from_states(node_states);
let frames =
node_frames_from_states_with_guard(node_states, fuser.guard_interval_us());
if frames.is_empty() {
return (None, Some(0));
}
@@ -313,26 +387,127 @@ mod tests {
}
#[test]
fn unsynchronized_frames_keep_host_arrival_guard() {
fn partial_sync_uses_one_host_timestamp_domain_for_the_cycle() {
let base = Instant::now() - Duration::from_millis(500);
let mut states = HashMap::new();
for (node_id, arrival) in [
(1, base),
(2, base + Duration::from_millis(200)),
let mut synced_history = VecDeque::new();
synced_history.push_back(vec![1.0; 64]);
let mut synced = make_node_state(synced_history, None, 0);
mark_mesh_timed_frame(&mut synced, 1, 100, 101, 500_000_000, base);
states.insert(1, synced);
let mut unsynced_history = VecDeque::new();
unsynced_history.push_back(vec![1.1; 64]);
states.insert(
2,
make_node_state(unsynced_history, Some(base + Duration::from_millis(5)), 0),
);
let frames = node_frames_from_states(&states);
let spread = frames.iter().map(|f| f.timestamp_us).max().unwrap()
- frames.iter().map(|f| f.timestamp_us).min().unwrap();
assert_eq!(spread, 5_000, "partial sync must fall back as one cycle");
assert!(
MultistaticFuser::new().fuse(&frames).is_ok(),
"mixed sync validity must not mix mesh and host timestamp domains"
);
}
#[test]
fn incoherent_mesh_timestamps_fall_back_to_host_arrival_for_the_cycle() {
let base = Instant::now() - Duration::from_millis(500);
let mut states = HashMap::new();
for (node_id, mesh_epoch_us, arrival) in [
(1, 1_000_000, base),
(2, 500_000_000, base + Duration::from_millis(5)),
] {
let mut history = VecDeque::new();
history.push_back(vec![1.0; 64]);
let mut state = make_node_state(history, None, 0);
mark_mesh_timed_frame(&mut state, node_id, 100, 101, mesh_epoch_us, arrival);
states.insert(node_id, state);
}
let frames = node_frames_from_states(&states);
let spread = frames.iter().map(|f| f.timestamp_us).max().unwrap()
- frames.iter().map(|f| f.timestamp_us).min().unwrap();
assert_eq!(spread, 5_000, "incoherent mesh time must not reach fusion");
assert!(MultistaticFuser::new().fuse(&frames).is_ok());
}
#[test]
fn unsynchronized_frames_prune_to_freshest_host_cohort() {
let base = Instant::now() - Duration::from_millis(500);
let mut states = HashMap::new();
for (node_id, arrival) in [(1, base), (2, base + Duration::from_millis(200))] {
let mut history = VecDeque::new();
history.push_back(vec![1.0; 64]);
states.insert(node_id, make_node_state(history, Some(arrival), 0));
}
let frames = node_frames_from_states(&states);
assert_eq!(frames.len(), 1, "only the freshest host frame is coherent");
assert_eq!(frames[0].node_id, 2);
assert!(
MultistaticFuser::new().fuse(&frames).is_err(),
"without valid mesh time, 200 ms arrival skew must still trip the 60 ms guard"
MultistaticFuser::new().fuse(&frames).is_ok(),
"an asynchronous slow node must not fail the live cycle"
);
}
#[test]
fn slow_live_node_is_excluded_from_fresh_cohort() {
let now = Instant::now();
let mut states = HashMap::new();
for (node_id, age_ms, n_sub) in [
(1, 0, 64),
(3, 10, 256),
(4, 50, 64),
(7, 1_000, 256),
] {
let mut history = VecDeque::new();
history.push_back(vec![1.0 + node_id as f64 * 0.01; n_sub]);
states.insert(
node_id,
make_node_state(
history,
Some(now - Duration::from_millis(age_ms)),
1,
),
);
}
let frames = node_frames_from_states_with_guard(&states, 60_000);
let ids: Vec<u8> = frames.iter().map(|frame| frame.node_id).collect();
assert_eq!(ids, vec![1, 3, 4]);
assert!(MultistaticFuser::new().fuse(&frames).is_ok());
}
#[test]
fn configured_guard_is_shared_with_cohort_selection() {
let base = Instant::now() - Duration::from_millis(500);
let mut states = HashMap::new();
for (node_id, arrival) in [
(1, base),
(2, base + Duration::from_millis(150)),
] {
let mut history = VecDeque::new();
history.push_back(vec![1.0; 64]);
states.insert(node_id, make_node_state(history, Some(arrival), 0));
}
let cfg = MultistaticConfig {
guard_interval_us: 200_000,
..MultistaticConfig::default()
};
let fuser = MultistaticFuser::with_config(cfg);
let (fused, fallback) = fuse_or_fallback(&fuser, &states, 3.0);
assert_eq!(fused.as_ref().map(|frame| frame.active_nodes), Some(2));
assert!(fallback.is_none());
}
#[test]
fn heterogeneous_node_counts_canonicalize_and_fuse() {
// Issue #1170 regression: a mixed mesh with HT20 (64-bin) and HT40

11
v2/crates/wifi-densepose-signal/src/ruvsense/multistatic.rs
View File

@@ -272,6 +272,17 @@ impl MultistaticFuser {
self.node_positions = positions;
}
/// Return the configured hard timestamp guard in microseconds.
///
/// Callers that assemble frames before invoking [`Self::fuse`] use this
/// to select one temporally coherent sensing cohort. Keeping selection and
/// validation on the same guard prevents stale low-rate nodes from making
/// every otherwise-live fusion cycle fail.
#[must_use]
pub fn guard_interval_us(&self) -> u64 {
self.config.guard_interval_us
}
/// Return the current node positions.
pub fn node_positions(&self) -> &[[f32; 3]] {
&self.node_positions