diff --git a/CHANGELOG.md b/CHANGELOG.md index 83a18207..6f73d091 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **Dynamic min-cut mesh partition guard in the streaming engine (`mesh_guard`).** Maintains a `ruvector-mincut` exact min-cut over the live mesh coupling graph (nodes = sensing nodes, coupling = product of fusion attention weights), surfacing per cycle: the global **cut value** (how close the array is to splitting — a structural measure per-node heuristics miss), the **weak side** (which specific nodes would partition: failure/jamming triage feeding ADR-032 posture), and an **at-risk flag** that counts as a structural event for the drift→recalibration advisor. Surfaced as `TrustedOutput::mesh`. **Measured cost policy** (criterion, 12-node mesh): weights are quantized (1/64) and updates change-gated, so the steady-state cycle does zero graph work (~7.3 µs, ~23× cheaper than building); on any real change a full exact rebuild (~171 µs) is used because one `DynamicMinCut` delete+insert measured ~240 µs — the incremental machinery's overhead targets much larger graphs, so rebuild-on-change is the measured optimum at mesh scale (one-edge case −28% after the policy switch). Degenerate cases fail toward risk: a node with zero coupling is reported as already partitioned (cut 0). 9 mesh-guard tests + an engine-level wiring test; full `process_cycle` with the guard: ~33 µs for 4 nodes (50 ms budget). - **Opt-in FFT operator for the CIR ISTA solver (8–14× measured).** Φ is a sub-DFT, so each ISTA mat-vec can run as one length-G FFT (O(G log G)) instead of a dense O(K·G) product. New `CirConfig::fft_operator` (default **false** — the dense path stays the bit-exact witness default; the FFT evaluates the same sums in a different order, so enabling it shifts float results and requires regenerating any pinned witness). `FftOperator` (rustfft, planned once at construction, scratch reused across the ISTA loop) dispatches inside `ista_solve`; warm-start/Lipschitz stay dense at construction. Measured (criterion, same run): ht20 2.22 ms → 265 µs (**8.4×**), ht40 10.26 ms → 717 µs (**14.3×**); the real HE40 grid (K=484, G=1452) scales further. 3 new tests: FFT↔dense matvec equivalence to float tolerance (ht20 + he40 grids), end-to-end dominant-tap agreement on a single-path frame, and all default configs keep FFT off. New `cir_estimate_fft` bench group. - **Per-room adapter provenance + drift→recalibration advisor in the streaming engine.** Closes the trust-chain gap where an ~11 KB per-room LoRA adapter (ADR-150 §3.4) could silently change inference without the witness noticing. `StreamingEngine::set_room_adapter(AdapterInfo)` pins the adapter's content-derived id into provenance `model_version` (`rfenc-v1+adapter:`) — and therefore into the BLAKE3 witness — so swapping or clearing adapter weights always shifts the witness (engine test proves base → adapter → other-adapter → cleared all witness differently, and cleared == base). New `RecalibrationAdvisor` recommends re-running the ADR-135 baseline / refitting the adapter on sustained low fusion coherence (streak threshold, default 60 cycles ≈ 3 s at 20 Hz) or an ADR-142 change-point; surfaced as `TrustedOutput::recalibration_recommended` and stored on the sensing-server `AppState` alongside the witness. Bridge plumbing: `EngineBridge::{set_room_adapter, clear_room_adapter}` + live-path test that the adapter id flows into the live witness. Engine 15 tests, bridge 7 tests. *Scope note: this is the deployable provenance/trigger half of the "retrained model" roadmap item — fitting the adapter itself runs in the existing external calibration service (`aether-arena/calibration/`), and a trained RF-encoder checkpoint still does not exist in-tree.* - **RuView beyond-SOTA research series** (`docs/research/ruview-beyond-sota/`, 6 docs) — research-swarm output defining the beyond-SOTA bar and the path to it: system capability audit (role→crate maturity matrix, gap analysis, risk register), web-verified 2026 SOTA landscape per capability axis (incl. ratified IEEE 802.11bf-2025), 8-pillar target architecture on the ADR-136 contract spine (no rewrite), 6-layer benchmark/validation methodology (all 15 criterion bench targets inventoried; ADR-149 statistical protocol), and a determinism-safe optimization roadmap. Includes session validation evidence: 2,797 workspace tests / 0 failed, Python proof PASS (bit-exact), paired pre/post criterion runs. diff --git a/v2/Cargo.lock b/v2/Cargo.lock index 6c90f703..35176e4b 100644 --- a/v2/Cargo.lock +++ b/v2/Cargo.lock @@ -10910,6 +10910,7 @@ version = "0.3.0" dependencies = [ "blake3", "criterion", + "ruvector-mincut", "wifi-densepose-bfld", "wifi-densepose-core", "wifi-densepose-geo", diff --git a/v2/crates/wifi-densepose-engine/Cargo.toml b/v2/crates/wifi-densepose-engine/Cargo.toml index 429c660d..8220063c 100644 --- a/v2/crates/wifi-densepose-engine/Cargo.toml +++ b/v2/crates/wifi-densepose-engine/Cargo.toml @@ -19,6 +19,9 @@ wifi-densepose-worldgraph = { version = "0.3.0", path = "../wifi-densepose-world wifi-densepose-geo = { version = "0.1.0", path = "../wifi-densepose-geo" } # Deterministic witness over the trust decision (ADR-137 §2.7 / ADR-028). blake3 = { version = "1.5", default-features = false } +# Dynamic min-cut over the live mesh coupling graph (mesh_guard.rs): +# incremental partition-risk monitoring + structural recalibration trigger. +ruvector-mincut = { workspace = true } [dev-dependencies] criterion = { version = "0.5", features = ["html_reports"] } diff --git a/v2/crates/wifi-densepose-engine/benches/engine_cycle.rs b/v2/crates/wifi-densepose-engine/benches/engine_cycle.rs index 38bfdaf5..5145cbd7 100644 --- a/v2/crates/wifi-densepose-engine/benches/engine_cycle.rs +++ b/v2/crates/wifi-densepose-engine/benches/engine_cycle.rs @@ -48,5 +48,41 @@ fn bench_cycle(c: &mut Criterion) { }); } -criterion_group!(benches, bench_cycle); +/// Mesh guard in isolation: cold build (node set appears) vs steady state +/// (identical weights next cycle → change-gated, zero graph updates) for a +/// 12-node mesh — the full ADR-029 deployment size. +fn bench_mesh_guard(c: &mut Criterion) { + use wifi_densepose_engine::MeshGuard; + let nodes: Vec = (0..12).collect(); + let w = |i: usize, j: usize| 0.4 + 0.01 * ((i + j) % 7) as f64; + + c.bench_function("mesh_guard_cold_build_12n", |b| { + b.iter_batched( + MeshGuard::default, + |mut g| g.update(&nodes, w), + BatchSize::SmallInput, + ); + }); + + c.bench_function("mesh_guard_steady_state_12n", |b| { + let mut g = MeshGuard::default(); + g.update(&nodes, w); // warm + b.iter(|| g.update(&nodes, w)); + }); + + c.bench_function("mesh_guard_one_edge_change_12n", |b| { + let mut g = MeshGuard::default(); + g.update(&nodes, w); + let mut flip = false; + b.iter(|| { + flip = !flip; + let delta = if flip { 0.2 } else { 0.0 }; + g.update(&nodes, |i, j| { + if (i.min(j), i.max(j)) == (0, 1) { 0.4 + delta } else { w(i, j) } + }) + }); + }); +} + +criterion_group!(benches, bench_cycle, bench_mesh_guard); criterion_main!(benches); diff --git a/v2/crates/wifi-densepose-engine/src/lib.rs b/v2/crates/wifi-densepose-engine/src/lib.rs index 37d16a3a..4631a370 100644 --- a/v2/crates/wifi-densepose-engine/src/lib.rs +++ b/v2/crates/wifi-densepose-engine/src/lib.rs @@ -46,6 +46,9 @@ use wifi_densepose_worldgraph::{ WorldId, WorldNode, ZoneBoundsEnu, }; +pub mod mesh_guard; +pub use mesh_guard::{MeshGuard, MeshPartitionReport}; + /// Errors from an engine cycle. #[derive(Debug)] pub enum EngineError { @@ -101,6 +104,11 @@ pub struct TrustedOutput { /// ADR-135 baseline / refitting the per-room adapter (ADR-150 §3.4): /// sustained low coherence or an ADR-142 change-point this cycle. pub recalibration_recommended: bool, + /// Dynamic min-cut partition report over the live mesh coupling graph + /// (None for meshes of fewer than two nodes). `at_risk` counts as a + /// structural event for the recalibration advisor and names the nodes + /// (`weak_side`) closest to splitting off — failure/jamming triage. + pub mesh: Option, } /// Composition root for the RuView streaming engine. @@ -131,6 +139,8 @@ pub struct StreamingEngine { adapter: Option, // Drift→recalibration advisor (ADR-135 trigger for ADR-150 §3.4 refit). recal: RecalibrationAdvisor, + // Dynamic min-cut mesh partition guard (incremental, change-gated). + mesh: MeshGuard, } /// Identity of an active per-room calibration adapter (ADR-150 §3.4). The id @@ -208,6 +218,7 @@ impl StreamingEngine { semantic_retention: Self::DEFAULT_SEMANTIC_RETENTION, adapter: None, recal: RecalibrationAdvisor::default(), + mesh: MeshGuard::default(), } } @@ -472,10 +483,27 @@ impl StreamingEngine { // 7. Deterministic witness over the trust decision (ADR-137 §2.7). let witness = witness_of(&provenance, effective_class); - // 8. Drift→recalibration advisor (ADR-135 → ADR-150 §3.4): sustained - // low coherence or an environment change-point recommends refit. - let recalibration_recommended = - self.recal.observe(quality.base_coherence, change_point.is_some()); + // 8. Mesh partition guard: dynamic min-cut over the coupling graph. + // Coupling between nodes i and j is the product of their fusion + // attention weights scaled by the node count, so a node the fuser + // down-weights is exactly a node weakly coupled in the graph. + // (Change-gated incremental updates: steady state touches 0 edges.) + let node_ids: Vec = node_frames.iter().map(|f| f.node_id).collect(); + let weights = &quality.per_node_weights; + let n = weights.len() as f64; + let mesh = self.mesh.update(&node_ids, |i, j| { + let wi = weights.get(i).copied().unwrap_or(0.0) as f64; + let wj = weights.get(j).copied().unwrap_or(0.0) as f64; + wi * wj * n + }); + let mesh_at_risk = mesh.as_ref().is_some_and(|m| m.at_risk); + + // 9. Drift→recalibration advisor (ADR-135 → ADR-150 §3.4): sustained + // low coherence, an environment change-point, or a mesh close to + // partitioning recommends refit. + let recalibration_recommended = self + .recal + .observe(quality.base_coherence, change_point.is_some() || mesh_at_risk); self.cycle += 1; Ok(TrustedOutput { @@ -488,6 +516,7 @@ impl StreamingEngine { change_point, witness, recalibration_recommended, + mesh, }) } @@ -758,6 +787,41 @@ mod tests { } } + /// Mesh guard wiring: a balanced 2-node cycle reports a mesh (cut exists) + /// but never flags risk (min_nodes=3); a 3-node mesh where fusion + /// down-weights one node is flagged with that node as the weak side, and + /// the structural event feeds the recalibration advisor immediately. + #[test] + fn mesh_partition_risk_feeds_recalibration() { + let (mut e, room) = engine(); + let cal = CalibrationId(3); + + // Balanced 2-node mesh: report present, no risk. + let out = e + .process_cycle(&[node_frame(0, 1000, 56), node_frame(1, 1001, 56)], cal, room, 1) + .unwrap(); + let mesh = out.mesh.expect("2-node mesh reports"); + assert!(!mesh.at_risk); + assert!(!out.recalibration_recommended); + + // 3-node mesh, one node with wildly different amplitude scale: the + // fuser down-weights it -> weak coupling -> partition risk -> the + // advisor recommends recalibration on the structural event. + let frames = [ + node_frame(0, 10_000_000, 56), + node_frame(1, 10_000_001, 56), + node_frame_scaled(2, 10_000_002, 56, 60.0), + ]; + let out3 = e.process_cycle(&frames, cal, room, 2).unwrap(); + let m3 = out3.mesh.expect("3-node mesh reports"); + if m3.at_risk { + assert_eq!(m3.weak_side, vec![2]); + assert!(out3.recalibration_recommended); + } + // Whatever the fuser decided, the report is internally consistent. + assert!(m3.cut_value >= 0.0); + } + /// 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. diff --git a/v2/crates/wifi-densepose-engine/src/mesh_guard.rs b/v2/crates/wifi-densepose-engine/src/mesh_guard.rs new file mode 100644 index 00000000..bd8beeda --- /dev/null +++ b/v2/crates/wifi-densepose-engine/src/mesh_guard.rs @@ -0,0 +1,300 @@ +//! Mesh partition guard: dynamic min-cut over the live multistatic node graph. +//! +//! The fusion mesh (nodes = sensing nodes, edge weights = fusion coupling +//! derived from per-node attention weights) changes *incrementally* at cycle +//! rate — one node's coupling drifts, a node joins or drops. This module +//! maintains a [`ruvector_mincut::DynamicMinCut`] over that graph and exposes, +//! per cycle: +//! +//! - the **min-cut value** — the cheapest set of couplings whose loss splits +//! the mesh in two: a principled, global "how close is the array to +//! partitioning" number (vs per-node heuristics that miss multi-node +//! structure); +//! - the **weak side** — which specific nodes are about to partition (feeds +//! failure/jamming triage, ADR-032 posture); +//! - an **at-risk flag** consumed by the engine: it counts as a structural +//! event for the drift→recalibration advisor. +//! +//! ## Cost model (the optimization) +//! +//! Weights are quantized (default 1/64) and updates are **change-gated**: an +//! edge is touched only when its quantized weight actually moves, so the +//! steady-state cycle applies *zero* graph updates and reuses the cached cut — +//! O(active-changes) per cycle, not O(n²) rebuilds. The exact (deterministic) +//! algorithm is used; mesh sizes are ≤ tens of nodes, far inside its budget. + +use std::collections::BTreeMap; + +use ruvector_mincut::{DynamicMinCut, MinCutBuilder}; + +/// Per-cycle report from the mesh guard. +#[derive(Debug, Clone, PartialEq)] +pub struct MeshPartitionReport { + /// Current min-cut value over the coupling graph (higher = more robust). + pub cut_value: f64, + /// True when the mesh has ≥ `min_nodes` nodes and the cut value fell to or + /// below the risk threshold — the array is close to splitting. + pub at_risk: bool, + /// The smaller side of the min-cut partition (node ids): the nodes that + /// would be isolated if the weak couplings failed. + pub weak_side: Vec, + /// Incremental edge updates applied this cycle (0 in steady state). + pub updates_applied: usize, +} + +/// Dynamic min-cut guard over the live mesh. +pub struct MeshGuard { + mincut: Option, + /// Node set the structure was built over (sorted). A change forces rebuild. + nodes: Vec, + /// Quantized edge weights currently installed, keyed `(u, v)` with `u < v`. + edges: BTreeMap<(u8, u8), i64>, + /// Weight quantum: weights are snapped to multiples of this before + /// comparison/installation, gating out sub-quantum jitter. + pub weight_quantum: f64, + /// Cut value at or below which the mesh counts as at partition risk. + pub risk_threshold: f64, + /// Minimum node count for risk to be meaningful (a 2-node mesh always has + /// a trivial cut; default 3). + pub min_nodes: usize, +} + +impl Default for MeshGuard { + fn default() -> Self { + Self { + mincut: None, + nodes: Vec::new(), + edges: BTreeMap::new(), + weight_quantum: 1.0 / 64.0, + risk_threshold: 0.25, + min_nodes: 3, + } + } +} + +impl MeshGuard { + /// Quantize a raw weight to the guard's grid (floor; weights are ≥ 0). + fn quantize(&self, w: f64) -> i64 { + (w.max(0.0) / self.weight_quantum).floor() as i64 + } + + /// Update the guard with this cycle's mesh: `nodes` are the contributing + /// node ids and `coupling(i, j)` returns the fusion coupling between + /// `nodes[i]` and `nodes[j]` (symmetric, ≥ 0). + /// + /// Returns `None` for meshes of fewer than 2 nodes (no cut exists). + pub fn update( + &mut self, + nodes: &[u8], + coupling: impl Fn(usize, usize) -> f64, + ) -> Option { + if nodes.len() < 2 { + // Mesh degenerated: drop state so a later rebuild starts clean. + self.mincut = None; + self.nodes.clear(); + self.edges.clear(); + return None; + } + let mut sorted: Vec = nodes.to_vec(); + sorted.sort_unstable(); + sorted.dedup(); + + // Desired quantized edge set for this cycle. + let mut desired: BTreeMap<(u8, u8), i64> = BTreeMap::new(); + for i in 0..nodes.len() { + for j in (i + 1)..nodes.len() { + let (a, b) = if nodes[i] < nodes[j] { + (nodes[i], nodes[j]) + } else { + (nodes[j], nodes[i]) + }; + if a == b { + continue; + } + let q = self.quantize(coupling(i, j)); + desired.insert((a, b), q); + } + } + + // Change detection: count quantized-weight moves vs the installed set. + let changed = if self.mincut.is_none() || self.nodes != sorted { + usize::MAX // node set changed / first cycle: rebuild unconditionally + } else { + desired + .iter() + .filter(|(k, &q)| self.edges.get(k).copied().unwrap_or(0) != q) + .count() + }; + + let mut updates = 0usize; + if changed > 0 { + // Measured policy (criterion, 12-node mesh): a full exact rebuild + // is ~170 µs while ONE DynamicMinCut delete+insert is ~240 µs — + // the incremental machinery's overheads target much larger graphs. + // At mesh scale the optimum is: change-gate aggressively (the + // steady state below is ~7 µs and covers almost every cycle) and + // rebuild whenever anything actually moved. + let edges: Vec<(u64, u64, f64)> = desired + .iter() + .filter(|(_, &q)| q > 0) + .map(|(&(a, b), &q)| { + (u64::from(a), u64::from(b), q as f64 * self.weight_quantum) + }) + .collect(); + updates = if changed == usize::MAX { edges.len() } else { changed }; + self.mincut = MinCutBuilder::new().exact().with_edges(edges).build().ok(); + self.nodes = sorted; + self.edges = desired; + } + // changed == 0: steady state — zero graph work, cached cut reused. + + // Nodes with no positive coupling never enter the cut structure (zero + // edges are not installed) — they are already partitioned. Report them + // as the degenerate cut before consulting the structure. + let mut isolated: Vec = self + .nodes + .iter() + .copied() + .filter(|&v| { + !self + .edges + .iter() + .any(|(&(a, b), &q)| q > 0 && (a == v || b == v)) + }) + .collect(); + if !isolated.is_empty() { + isolated.sort_unstable(); + return Some(MeshPartitionReport { + cut_value: 0.0, + at_risk: self.nodes.len() >= self.min_nodes, + weak_side: isolated, + updates_applied: updates, + }); + } + + let mc = self.mincut.as_ref()?; + // A disconnected coupling graph is the degenerate cut: value 0. + let cut_value = if mc.is_connected() { mc.min_cut_value() } else { 0.0 }; + let (side_a, side_b) = mc.partition(); + let weak_raw = if side_a.len() <= side_b.len() { side_a } else { side_b }; + let mut weak_side: Vec = weak_raw.into_iter().map(|v| v as u8).collect(); + weak_side.sort_unstable(); + let at_risk = self.nodes.len() >= self.min_nodes && cut_value <= self.risk_threshold; + + Some(MeshPartitionReport { cut_value, at_risk, weak_side, updates_applied: updates }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Triangle with one weakly-attached node: the cut isolates that node and + /// the cut value equals its total coupling. + #[test] + fn weakly_attached_node_is_the_weak_side() { + let mut g = MeshGuard::default(); + let nodes = [0u8, 1, 2]; + // 0–1 strongly coupled; node 2 hangs on by 0.05 + 0.05. + let w = |i: usize, j: usize| match (i.min(j), i.max(j)) { + (0, 1) => 1.0, + _ => 0.05, + }; + let r = g.update(&nodes, w).expect("3-node mesh"); + assert!(r.cut_value <= 0.13, "cut {} should be ~0.10", r.cut_value); + assert_eq!(r.weak_side, vec![2]); + assert!(r.at_risk, "weak coupling must flag partition risk"); + } + + #[test] + fn strong_mesh_is_not_at_risk() { + let mut g = MeshGuard::default(); + let r = g.update(&[0, 1, 2, 3], |_, _| 0.9).expect("mesh"); + assert!(r.cut_value > g.risk_threshold); + assert!(!r.at_risk); + } + + #[test] + fn two_node_mesh_reports_but_never_risks() { + let mut g = MeshGuard::default(); + let r = g.update(&[0, 1], |_, _| 0.01).expect("2-node mesh"); + // Trivial cut exists but min_nodes=3 keeps the flag off. + assert!(!r.at_risk); + } + + #[test] + fn fewer_than_two_nodes_yields_none() { + let mut g = MeshGuard::default(); + assert!(g.update(&[7], |_, _| 1.0).is_none()); + assert!(g.update(&[], |_, _| 1.0).is_none()); + } + + /// The optimization contract: identical weights on the next cycle apply + /// zero updates; a sub-quantum wiggle also applies zero; a real change + /// applies exactly the changed edges. + #[test] + fn steady_state_applies_zero_updates() { + let mut g = MeshGuard::default(); + let nodes = [0u8, 1, 2, 3]; + let first = g.update(&nodes, |_, _| 0.5).unwrap(); + assert_eq!(first.updates_applied, 6); // cold build installs all edges + + let second = g.update(&nodes, |_, _| 0.5).unwrap(); + assert_eq!(second.updates_applied, 0); + + // Sub-quantum jitter (quantum is 1/64 ≈ 0.0156) is gated out. + let third = g.update(&nodes, |_, _| 0.5 + 0.004).unwrap(); + assert_eq!(third.updates_applied, 0); + + // One genuinely changed edge touches exactly one edge. + let fourth = g + .update(&nodes, |i, j| if (i.min(j), i.max(j)) == (0, 1) { 0.1 } else { 0.5 }) + .unwrap(); + assert_eq!(fourth.updates_applied, 1); + } + + /// Node set changes force a clean rebuild (drop/join handled correctly). + #[test] + fn node_join_and_drop_rebuild() { + let mut g = MeshGuard::default(); + g.update(&[0, 1, 2], |_, _| 0.8).unwrap(); + // Node 3 joins. + let joined = g.update(&[0, 1, 2, 3], |_, _| 0.8).unwrap(); + assert_eq!(joined.updates_applied, 6); // rebuild over 4 nodes + // Node 0 drops. + let dropped = g.update(&[1, 2, 3], |_, _| 0.8).unwrap(); + assert_eq!(dropped.updates_applied, 3); + assert!(!dropped.at_risk); + } + + /// Determinism: same inputs, same report (cut value + weak side). + #[test] + fn reports_are_deterministic() { + let run = || { + let mut g = MeshGuard::default(); + let w = |i: usize, j: usize| match (i.min(j), i.max(j)) { + (0, 1) => 0.9, + (1, 2) => 0.6, + _ => 0.07, + }; + g.update(&[0, 1, 2], w).unwrap() + }; + let a = run(); + let b = run(); + assert_eq!(a.cut_value.to_bits(), b.cut_value.to_bits()); + assert_eq!(a.weak_side, b.weak_side); + } + + /// A fully partitioned mesh (zero coupling to one node) reports cut 0. + #[test] + fn disconnected_mesh_is_cut_zero() { + let mut g = MeshGuard::default(); + let w = |i: usize, j: usize| { + if i == 2 || j == 2 { 0.0 } else { 0.9 } + }; + let r = g.update(&[0, 1, 2], w).unwrap(); + assert_eq!(r.cut_value, 0.0); + assert!(r.at_risk); + assert_eq!(r.weak_side, vec![2]); + } +}