diff --git a/Cargo.lock b/Cargo.lock index 39c62fa..6d4c0c5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -985,6 +985,9 @@ dependencies = [ "rucelium-store", "rucelium-transport", "rucelium-worldgraph", + "rufield-core", + "rufield-fusion", + "rufield-provenance", "rustls", "serde", "serde_json", diff --git a/crates/rucelium-gateway/Cargo.toml b/crates/rucelium-gateway/Cargo.toml index ad792ad..5a508e1 100644 --- a/crates/rucelium-gateway/Cargo.toml +++ b/crates/rucelium-gateway/Cargo.toml @@ -30,6 +30,9 @@ rucelium-worldgraph = { workspace = true } rucelium-federation = { workspace = true } rucelium-policy = { workspace = true } rucelium-store = { workspace = true } +rufield-core = { workspace = true } +rufield-fusion = { workspace = true } +rufield-provenance = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/rucelium-gateway/src/api.rs b/crates/rucelium-gateway/src/api.rs index ba2b3d7..17af65e 100644 --- a/crates/rucelium-gateway/src/api.rs +++ b/crates/rucelium-gateway/src/api.rs @@ -78,6 +78,7 @@ pub fn router(state: GatewayState) -> Router { .route("/api/federation/announce", post(fed_announce)) .route("/api/admin/revoke/:node_id", post(admin_revoke)) .route("/api/admin/command", post(admin_command)) + .route("/api/rf-context", get(rf_context)) .with_state(state) } @@ -124,6 +125,32 @@ async fn stats(State(state): State) -> Json { })) } +/// `GET /api/rf-context` — the RuView RF-context bridge's state (ADR-264 §8, +/// see `rf_bridge`): the configured upstream (if any), ingest counters, the +/// bounded recent ring, the current within-RF-domain fused inferences +/// (`rf_bridge::current_inferences` — RuField's own room-state rules over +/// this upstream's own verified events; needs no external pairing), and a +/// bounded history of past inference snapshots so a transient inference is +/// still visible after its own TTL expires (in-memory only — does not +/// survive a restart; see [`crate::rf_bridge::InferenceSnapshot`] docs for +/// why not durable yet). Each ring entry is honestly tagged `verified` — an unverified event +/// is visible here but excluded from fusion, never silently dropped or +/// upgraded. `fused_into_worldgraph` is always `false`: this bridge does not +/// (yet) correlate RF context against any biome environmental sample — see +/// `rf_bridge`'s module docs for why that's a separate, still-unbuilt +/// decision. Empty and idle when `--rf-upstream` was not set. +async fn rf_context(State(state): State) -> Json { + let inner = state.inner.lock().await; + Json(json!({ + "upstream": inner.rf_upstream, + "fused_into_worldgraph": false, + "stats": inner.rf_stats, + "recent": inner.rf_recent.iter().collect::>(), + "room_state_inferences": crate::rf_bridge::current_inferences(&inner.rf_fusion), + "inference_history": inner.rf_inference_history.iter().collect::>(), + })) +} + /// `GET /api/observations/recent?limit=50` — most recent stored samples in /// append order. async fn observations_recent( diff --git a/crates/rucelium-gateway/src/config.rs b/crates/rucelium-gateway/src/config.rs index faf2b44..21d3465 100644 --- a/crates/rucelium-gateway/src/config.rs +++ b/crates/rucelium-gateway/src/config.rs @@ -24,6 +24,9 @@ pub const DEFAULT_ACTUATOR_ID: &str = "sluice-gate-1"; /// Default durability mode for the durable stores: the daemon fsyncs every /// accepted append, so an accepted record survives power loss. pub const DEFAULT_FSYNC: bool = true; +/// Default poll interval for the RuView RF-context bridge (ADR-264 §8), +/// matching `rufield-viewer`'s own default `/api/field` poll cadence. +pub const DEFAULT_RF_POLL_MS: u64 = 500; /// Runtime configuration of one gateway daemon instance. #[derive(Debug, Clone, PartialEq, Eq)] @@ -61,6 +64,13 @@ pub struct GatewayConfig { /// Durability mode for `ObservationStore` / `EventStore`: `true` fsyncs /// every accepted append before it is acknowledged. pub fsync: bool, + /// Base URL of a RuView `wifi-densepose-sensing-server` to poll for RF + /// context (ADR-264 §8). `None` (default) disables the bridge entirely — + /// this is opt-in, not a dependency the gateway assumes exists. + pub rf_upstream: Option, + /// Poll interval for the RF-context bridge, in milliseconds. Unused when + /// `rf_upstream` is `None`. + pub rf_poll_ms: u64, } impl Default for GatewayConfig { @@ -79,6 +89,8 @@ impl Default for GatewayConfig { federation_backfill_ms: None, actuator_id: DEFAULT_ACTUATOR_ID.to_string(), fsync: DEFAULT_FSYNC, + rf_upstream: None, + rf_poll_ms: DEFAULT_RF_POLL_MS, } } } @@ -129,6 +141,10 @@ impl GatewayConfig { } "--actuator" => config.actuator_id = value("--actuator")?, "--fsync" => config.fsync = parse_num(&value("--fsync")?, "--fsync")?, + "--rf-upstream" => config.rf_upstream = Some(value("--rf-upstream")?), + "--rf-poll-ms" => { + config.rf_poll_ms = parse_num(&value("--rf-poll-ms")?, "--rf-poll-ms")?; + } unknown => return Err(format!("unknown flag {unknown}")), } } @@ -169,6 +185,8 @@ mod tests { assert_eq!(c.federation_backfill_ms(), 30_000); assert_eq!(c.actuator_id, "sluice-gate-1"); assert!(c.fsync, "the daemon fsyncs accepted appends by default"); + assert_eq!(c.rf_upstream, None, "the RF bridge is opt-in, off by default"); + assert_eq!(c.rf_poll_ms, 500); } #[test] @@ -210,6 +228,10 @@ mod tests { "weir-3", "--fsync", "false", + "--rf-upstream", + "http://127.0.0.1:8080", + "--rf-poll-ms", + "250", ])) .unwrap(); assert_eq!(c.biome_id, "biome/x"); @@ -223,6 +245,8 @@ mod tests { assert_eq!(c.federation_poll_ms, 200); assert_eq!(c.actuator_id, "weir-3"); assert!(!c.fsync); + assert_eq!(c.rf_upstream, Some("http://127.0.0.1:8080".to_string())); + assert_eq!(c.rf_poll_ms, 250); } #[test] diff --git a/crates/rucelium-gateway/src/lib.rs b/crates/rucelium-gateway/src/lib.rs index 04d4285..e3cea9f 100644 --- a/crates/rucelium-gateway/src/lib.rs +++ b/crates/rucelium-gateway/src/lib.rs @@ -14,8 +14,19 @@ //! ──► /api/federation/{pubkey,summary,revocations,peers} //! ──► /api/federation/announce (push inbox, ADR-269 §3) //! ──► /api/admin/{revoke/:node_id,command} +//! ──► /api/rf-context (ADR-264 §8, see `rf_bridge`) //! ``` //! +//! ## RuView RF context (ADR-264 §8) +//! +//! `--rf-upstream ` starts [`rf_bridge::run_rf_ingest`]: it polls a +//! RuView `wifi-densepose-sensing-server`'s `/api/field` (ADR-262 P3), +//! verifies each event's provenance receipt, and keeps a bounded, honestly +//! verified/unverified-tagged ring at `GET /api/rf-context`. This is storage +//! only — it does **not** fuse into the WorldGraph yet; see [`rf_bridge`]'s +//! module docs for why that pairing decision is deliberately not guessed at +//! here. +//! //! ## Push federation (ADR-269 §3) //! //! Federation is push-first over a swappable [`transport::FederationTransport`]: @@ -61,6 +72,7 @@ pub mod federation; pub mod journal; pub mod net; pub mod pipeline; +pub mod rf_bridge; pub mod simulate; pub mod state; pub mod transport; @@ -185,6 +197,14 @@ pub async fn spawn_gateway_with_transport( ))); } + if let Some(upstream) = config.rf_upstream.clone() { + tasks.push(tokio::spawn(rf_bridge::run_rf_ingest( + state.clone(), + upstream, + config.rf_poll_ms, + ))); + } + Ok(GatewayHandle { state, udp_port, diff --git a/crates/rucelium-gateway/src/main.rs b/crates/rucelium-gateway/src/main.rs index 8e8370c..204306e 100644 --- a/crates/rucelium-gateway/src/main.rs +++ b/crates/rucelium-gateway/src/main.rs @@ -14,7 +14,8 @@ async fn main() { "usage: rucelium-gateway [--biome-id ] [--udp ] [--http ] \ [--data-dir ] [--peer ]... [--simulate ] [--seed ] \ [--sim-interval-ms ] [--retention-check-secs ] \ - [--federation-poll-ms ] [--actuator ] [--fsync ]" + [--federation-poll-ms ] [--actuator ] [--fsync ] \ + [--rf-upstream ] [--rf-poll-ms ]" ); std::process::exit(2); } @@ -35,6 +36,10 @@ async fn main() { println!(" peer: {peer}"); } } + match &config.rf_upstream { + Some(url) => println!(" rf-context: polling {url} every {}ms (ADR-264 §8)", config.rf_poll_ms), + None => println!(" rf-context: disabled (set --rf-upstream to enable)"), + } println!(" WARNING: admin endpoints are UNAUTHENTICATED in v0.1 — bind"); println!(" the http port to localhost or firewall it."); diff --git a/crates/rucelium-gateway/src/rf_bridge.rs b/crates/rucelium-gateway/src/rf_bridge.rs new file mode 100644 index 0000000..c6069df --- /dev/null +++ b/crates/rucelium-gateway/src/rf_bridge.rs @@ -0,0 +1,553 @@ +//! RuView RF-context ingest (ADR-264 §8) — two *different* kinds of "fusion", +//! only one of which this module does. +//! +//! Polls an external RuView `wifi-densepose-sensing-server`'s `GET +//! /api/field` (ADR-262 P3), verifies each event's provenance receipt, keeps +//! a bounded ring of the resulting [`RfContext`]s, and runs verified events +//! through [`rufield_fusion::RuFieldFusion`] to produce real room-state +//! inferences (`person_present`, `motion`, …) — all exposed at +//! `GET /api/rf-context`. +//! +//! ## Two fusions, not one +//! +//! * **`rufield_fusion::RuFieldFusion` (this module, real, wired in below)** +//! — a temporal-window fusion *within* the RF domain: it combines this +//! upstream's own `FieldEvent`s over a short window into inferences using +//! RuField's own rule set. It needs nothing external — no environmental +//! sample, no biome, no zone mapping — so there is nothing to fabricate. +//! This is the same engine `rufield-viewer` already runs on the exact same +//! event stream; wiring it here just makes the gateway able to answer +//! "what does this RF upstream currently think is happening" on its own. +//! * **`rucelium_worldgraph::fuse_rf_context` (still NOT called here)** — a +//! *cross-domain* fusion: pairing an `RfContext` against a *specific* +//! existing environmental sample's WorldGraph node to compute a +//! [`rucelium_worldgraph::Plausibility`] verdict (does the RF context +//! support or contradict *that* biome sensor's reading). Nothing in this +//! codebase yet defines which biome zone/sensor a given RuView device is +//! supposed to corroborate, and inventing that pairing would be exactly +//! the fabricated correlation ADR-264 §8 exists to prevent. That +//! zone-pairing design is still deliberate follow-up work, not guessed at +//! — this module does not touch the WorldGraph. +//! +//! ## Honesty +//! +//! Every event is receipt-checked via [`rufield_provenance::is_fusable`] and +//! tagged `verified: bool` in the stored ring — an unverified event is kept +//! visible (like a forged-data ✗ badge) but is excluded from the fusion +//! engine entirely (mirrors `rufield-viewer`'s own rule: forged data is never +//! rendered as trusted). `/api/rf-context` reports +//! `"fused_into_worldgraph": false` explicitly, so a consumer never has to +//! infer the WorldGraph boundary from absence. + +use rucelium_worldgraph::RfContext; +use rufield_core::{FieldEvent, FusionEngine, InferenceQuery}; +use rufield_fusion::RuFieldFusion; +use serde::{Deserialize, Serialize}; +use std::collections::VecDeque; +use std::time::Duration; + +use crate::state::{now_ns, GatewayState}; + +/// Bounded ring size for recently ingested RF context (mirrors +/// `wifi-densepose-sensing-server`'s own `FIELD_RING_CAPACITY` convention — +/// a live tap, not a store). +pub const RF_RECENT_CAPACITY: usize = 64; + +/// Bounded ring size for retained inference snapshots. Smaller than +/// [`RF_RECENT_CAPACITY`] because each snapshot carries every currently-live +/// inference (typically a handful), not one event. +pub const INFERENCE_HISTORY_CAPACITY: usize = 32; + +/// Default poll interval for `GET /api/field`, matching +/// `rufield-viewer`'s own default. +pub const DEFAULT_RF_POLL_MS: u64 = 500; + +/// One ingested RF context plus its verification outcome and local receipt +/// time. +#[derive(Debug, Clone, Serialize)] +pub struct RfEntry { + /// The distilled RF context (never ground truth — ADR-264 §8). + pub context: RfContext, + /// Whether the source `FieldEvent`'s provenance receipt verified + /// (`rufield_provenance::is_fusable`). `false` means the bytes were + /// reached but the signature/hash did not check out — stored so a forged + /// or corrupted upstream is *visible*, never silently dropped. + pub verified: bool, + /// When this gateway received the event, ns since Unix epoch. + pub received_ns: u64, +} + +/// Ingest counters, exposed at `GET /api/rf-context`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +pub struct RfIngestStats { + /// Successful polls of the upstream `/api/field` ring. + pub polls: u64, + /// Polls that failed (unreachable, non-2xx, or a malformed payload). + pub poll_failures: u64, + /// Total `FieldEvent`s received across all polls. + pub events_received: u64, + /// Received events whose provenance receipt verified. + pub events_verified: u64, + /// Received events whose provenance receipt did NOT verify (stored, + /// flagged, never trusted). + pub events_unverified: u64, + /// Received events that were not `Modality::WifiCsi` and so are not RF + /// context this bridge understands (`RfContext::from_field_event` + /// returns `None`). Not an error — RuView's ring may carry other + /// modalities over time. + pub events_other_modality: u64, +} + +/// One captured moment of the fusion engine's current inferences, retained +/// so a `person_present` (or any other) inference that fired between two +/// polls is not simply lost the moment its 2-second TTL expires. This is the +/// gap between how RF inferences behaved before this struct existed (only +/// ever computed live, on request, from whatever is in the engine's window +/// *right now*) and how every other durable thing in this gateway behaves +/// (`ObservationStore`, `EventStore`) — RF inferences are still NOT written +/// to either of those (they are not `EnvSample`/`EnvironmentalEvent`s and +/// have no biome `node_id`/`sequence` to cite — the same structural gap +/// `fuse_rf_context` ran into), so this is in-memory history, not disk +/// durability. It survives across polls; it does not survive a restart. +#[derive(Debug, Clone, Serialize)] +pub struct InferenceSnapshot { + /// When this snapshot was captured, ns since Unix epoch (the poll's + /// `received_ns`, not any individual inference's own `produced_ns`). + pub captured_ns: u64, + /// Every inference the engine reported live at capture time. + pub inferences: Vec, +} + +/// Shape of RuView's `GET /api/field` response that this bridge actually +/// needs. Deliberately declares **only** `events`: the response also carries +/// `signer_pubkey_hex` and `dev_signing_key`, but their exact shape is +/// RuView's to evolve (see `ruvnet/RuCelium#1`, where a *different* consumer +/// of this same endpoint broke because it over-specified those two +/// informational fields). Serde drops unknown fields by default — no +/// `deny_unknown_fields` — so this struct is inert to that class of drift. +#[derive(Debug, Deserialize)] +struct ApiFieldPayload { + events: Vec, +} + +/// Pure ingest step (no I/O — unit-testable): fold one batch of freshly +/// polled `FieldEvent`s into the bounded ring, the counters, and the +/// within-RF-domain fusion engine (module docs: this is the fusion that +/// needs no external pairing, so it runs unconditionally on every verified +/// event) — and, only when this batch actually fed the engine something new, +/// capture a snapshot of its current inferences into `history` so a +/// transient inference outlives its own TTL instead of vanishing the moment +/// nobody happens to be polling `/api/rf-context`. +pub fn ingest_events( + recent: &mut VecDeque, + stats: &mut RfIngestStats, + fusion: &mut RuFieldFusion, + history: &mut VecDeque, + events: &[FieldEvent], + received_ns: u64, +) { + stats.polls += 1; + let mut fed_engine = false; + for ev in events { + stats.events_received += 1; + let verified = rufield_provenance::is_fusable(ev); + if verified { + stats.events_verified += 1; + // Mirrors rufield-viewer's own rule: only verified events reach + // the fusion engine. `ingest` itself re-checks §11 fusability + // (belt-and-suspenders), so forged data cannot reach it either way. + let _ = fusion.ingest(ev.clone()); + fed_engine = true; + } else { + stats.events_unverified += 1; + } + match RfContext::from_field_event(ev) { + Some(context) => { + if recent.len() == RF_RECENT_CAPACITY { + recent.pop_front(); + } + recent.push_back(RfEntry { + context, + verified, + received_ns, + }); + } + None => stats.events_other_modality += 1, + } + } + // Only capture a snapshot when this batch actually changed the engine's + // state — an empty or all-unverified poll would just re-record the same + // stale inferences (or an empty vec) under a new timestamp, which is + // noise, not history. + if fed_engine { + if history.len() == INFERENCE_HISTORY_CAPACITY { + history.pop_front(); + } + history.push_back(InferenceSnapshot { + captured_ns: received_ns, + inferences: current_inferences(fusion), + }); + } +} + +/// Record a failed poll attempt (unreachable upstream, non-2xx, or a +/// malformed payload). Never fatal — the next tick tries again. +pub fn record_poll_failure(stats: &mut RfIngestStats) { + stats.poll_failures += 1; +} + +/// Current room-state inferences from the within-RF-domain fusion engine +/// (module docs: the fusion that needs no external pairing). Empty on a +/// query error — `infer` errors only on the engine's own internal +/// consistency checks, not on caller input, so this stays a best-effort read +/// rather than a `Result` the HTTP handler has to plumb through. +#[must_use] +pub fn current_inferences(fusion: &RuFieldFusion) -> Vec { + fusion.infer(&InferenceQuery::all()).unwrap_or_default() +} + +/// Fetch and decode one `GET /api/field` response. The only I/O in +/// this module — everything else is pure and covered by unit tests. +async fn poll_once(client: &reqwest::Client, url: &str) -> Result, String> { + let resp = client + .get(url) + .send() + .await + .map_err(|e| format!("request failed: {e}"))?; + if !resp.status().is_success() { + return Err(format!("http {}", resp.status())); + } + let body = resp + .text() + .await + .map_err(|e| format!("read body: {e}"))?; + let payload: ApiFieldPayload = + serde_json::from_str(&body).map_err(|e| format!("decode /api/field: {e}"))?; + Ok(payload.events) +} + +/// Run the RF ingest task forever: poll `/api/field` every +/// `poll_ms`, verify each event, and fold accepted RF context into +/// [`crate::state::Inner::rf_recent`]. Never fatal — an unreachable upstream +/// or a malformed payload just counts a failure and retries on the next +/// tick. +pub async fn run_rf_ingest(state: GatewayState, upstream: String, poll_ms: u64) { + let client = match reqwest::Client::builder() + .timeout(Duration::from_secs(10)) + .build() + { + Ok(c) => c, + Err(_) => return, + }; + let url = format!("{}/api/field", upstream.trim_end_matches('/')); + let mut tick = tokio::time::interval(Duration::from_millis(poll_ms.max(50))); + loop { + tick.tick().await; + match poll_once(&client, &url).await { + Ok(events) => { + let received_ns = now_ns(); + let mut guard = state.inner.lock().await; + let inner = &mut *guard; + ingest_events( + &mut inner.rf_recent, + &mut inner.rf_stats, + &mut inner.rf_fusion, + &mut inner.rf_inference_history, + &events, + received_ns, + ); + } + Err(e) => { + eprintln!("gateway: rf-context poll of {url} failed: {e}"); + let mut inner = state.inner.lock().await; + record_poll_failure(&mut inner.rf_stats); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rufield_core::{ + FieldAxis, FieldTensor, Modality, Observation, PrivacyClass, ProvenanceRef, + SensorDescriptor, + }; + use rufield_provenance::Signer; + + fn wifi_csi_event(device_id: &str, signed: bool) -> FieldEvent { + let tensor = FieldTensor::new( + 1_000, + Modality::WifiCsi, + vec![FieldAxis::Frequency], + vec![2], + vec![0.1, 0.2], + 0.9, + 0.01, + Some("cal".into()), + PrivacyClass::P2, + ) + .unwrap(); + let mut observation = Observation::occupancy(0.8, PrivacyClass::P2); + observation + .features + .insert("motion_energy".to_string(), 0.7); + // "presence" is what `rufield-fusion`'s default `person_present` rule + // (rules/room_state.toml) actually keys on — without it the fusion + // regression tests below would pass trivially (no rule ever fires). + observation.features.insert("presence".to_string(), 0.8); + let mut ev = FieldEvent::new( + format!("ev-{device_id}"), + 1_000, + SensorDescriptor { + modality: "wifi_csi".into(), + vendor: "simulated".into(), + device_id: device_id.to_string(), + placement: "unknown".into(), + clock_domain: "local".into(), + }, + tensor, + observation, + ProvenanceRef { + raw_hash: "sha256:raw".into(), + firmware_hash: "sha256:fw".into(), + model_id: "m".into(), + calibration_id: "cal".into(), + // Deliberately non-synthetic: `is_fusable` short-circuits to + // `true` for synthetic events regardless of signature, which + // would make the "unsigned → unverified" test meaningless. + // Real verification only actually runs when this is `false`. + synthetic: false, + signature_hex: None, + signer_pubkey_hex: None, + }, + ); + if signed { + Signer::from_seed(b"rf-bridge-test-seed-32-bytes-ok!") + .sign_event(&mut ev) + .expect("test event signs"); + } + ev + } + + fn other_modality_event() -> FieldEvent { + let tensor = FieldTensor::new( + 1_000, + Modality::MmwaveRadar, + vec![FieldAxis::Frequency], + vec![1], + vec![0.5], + 0.5, + 0.01, + None, + PrivacyClass::P2, + ) + .unwrap(); + FieldEvent::new( + "ev-radar", + 1_000, + SensorDescriptor { + modality: "mmwave_radar".into(), + vendor: "sim".into(), + device_id: "radar-1".into(), + placement: "unknown".into(), + clock_domain: "local".into(), + }, + tensor, + Observation::occupancy(0.5, PrivacyClass::P2), + ProvenanceRef { + raw_hash: "sha256:raw".into(), + firmware_hash: "sha256:fw".into(), + model_id: "m".into(), + calibration_id: "cal".into(), + synthetic: true, + signature_hex: None, + signer_pubkey_hex: None, + }, + ) + } + + #[test] + fn verified_wifi_csi_event_is_stored_and_counted() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-1", true); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 5_000); + + assert_eq!(recent.len(), 1); + assert!(recent[0].verified); + assert_eq!(recent[0].context.device_id, "node-1"); + assert_eq!(recent[0].received_ns, 5_000); + assert_eq!(stats.polls, 1); + assert_eq!(stats.events_received, 1); + assert_eq!(stats.events_verified, 1); + assert_eq!(stats.events_unverified, 0); + assert_eq!(stats.events_other_modality, 0); + } + + #[test] + fn unsigned_event_is_stored_but_flagged_unverified() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-2", false); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + assert_eq!(recent.len(), 1, "unverified events are still visible, not dropped"); + assert!(!recent[0].verified); + assert_eq!(stats.events_unverified, 1); + assert_eq!(stats.events_verified, 0); + } + + #[test] + fn non_wifi_csi_event_is_not_stored_as_rf_context() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = other_modality_event(); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + assert!(recent.is_empty()); + assert_eq!(stats.events_other_modality, 1); + assert_eq!(stats.events_received, 1, "still counted as received"); + } + + #[test] + fn ring_is_bounded_and_drops_oldest() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + for i in 0..(RF_RECENT_CAPACITY + 10) { + let ev = wifi_csi_event(&format!("node-{i}"), true); + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), i as u64); + } + assert_eq!(recent.len(), RF_RECENT_CAPACITY); + // Oldest entries evicted first: the front should be from late in the + // sequence, not node-0. + assert_ne!(recent.front().unwrap().context.device_id, "node-0"); + assert_eq!( + recent.back().unwrap().context.device_id, + format!("node-{}", RF_RECENT_CAPACITY + 9) + ); + } + + #[test] + fn verified_events_actually_reach_the_fusion_engine() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-fused", true); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + let inferences = current_inferences(&fusion); + assert!( + inferences.iter().any(|i| i.label == "person_present"), + "a verified presence-bearing event should produce a person_present \ + inference from RuField's own default room-state rules, got: {inferences:?}" + ); + } + + #[test] + fn unverified_events_never_reach_the_fusion_engine() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-forged", false); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 1); + + let inferences = current_inferences(&fusion); + assert!( + inferences.is_empty(), + "an unverified event must never influence a fused inference, got: {inferences:?}" + ); + } + + #[test] + fn a_verified_batch_captures_an_inference_snapshot() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + let ev = wifi_csi_event("node-hist", true); + + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, std::slice::from_ref(&ev), 42); + + assert_eq!(history.len(), 1, "a batch that fed the engine must be captured"); + let snap = &history[0]; + assert_eq!(snap.captured_ns, 42); + assert!( + snap.inferences.iter().any(|i| i.label == "person_present"), + "the snapshot must carry the inference the batch actually produced" + ); + } + + #[test] + fn an_empty_or_all_unverified_batch_is_not_recorded_as_history_noise() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + + // Empty batch (e.g. RuView's ring had nothing new this poll). + ingest_events(&mut recent, &mut stats, &mut fusion, &mut history, &[], 1); + assert!(history.is_empty(), "an empty poll must not add a snapshot"); + + // All-unverified batch never fed the engine either. + let unsigned = wifi_csi_event("node-x", false); + ingest_events( + &mut recent, + &mut stats, + &mut fusion, + &mut history, + std::slice::from_ref(&unsigned), + 2, + ); + assert!( + history.is_empty(), + "a batch that never fed the engine must not add a snapshot either" + ); + } + + #[test] + fn inference_history_is_bounded_and_drops_oldest() { + let mut recent = VecDeque::new(); + let mut stats = RfIngestStats::default(); + let mut fusion = RuFieldFusion::new(); + let mut history: VecDeque = VecDeque::new(); + for i in 0..(INFERENCE_HISTORY_CAPACITY + 5) { + let ev = wifi_csi_event(&format!("node-{i}"), true); + ingest_events( + &mut recent, + &mut stats, + &mut fusion, + &mut history, + std::slice::from_ref(&ev), + i as u64, + ); + } + assert_eq!(history.len(), INFERENCE_HISTORY_CAPACITY); + assert_eq!(history.back().unwrap().captured_ns, (INFERENCE_HISTORY_CAPACITY + 4) as u64); + } + + #[test] + fn poll_failure_is_counted_and_never_touches_the_ring() { + let recent: VecDeque = VecDeque::new(); + let mut stats = RfIngestStats::default(); + record_poll_failure(&mut stats); + record_poll_failure(&mut stats); + assert_eq!(stats.poll_failures, 2); + assert!(recent.is_empty()); + } +} diff --git a/crates/rucelium-gateway/src/state.rs b/crates/rucelium-gateway/src/state.rs index bad2f3c..684144f 100644 --- a/crates/rucelium-gateway/src/state.rs +++ b/crates/rucelium-gateway/src/state.rs @@ -9,6 +9,8 @@ use crate::config::GatewayConfig; use crate::journal; +use crate::rf_bridge::{InferenceSnapshot, RfEntry, RfIngestStats}; +use rufield_fusion::RuFieldFusion; use rucelium_calibration::{ AuthorityRegistry as CalibrationAuthorities, CalibrationAuthority, CalibrationError, CalibrationSigner, CalibrationStore, Calibrator, DriftDetector, @@ -24,7 +26,7 @@ use rucelium_store::{EventStore, ObservationStore}; use rucelium_transport::Reassembler; use rucelium_worldgraph::WorldGraph; use serde::Serialize; -use std::collections::{BTreeMap, BTreeSet}; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::path::PathBuf; use std::sync::Arc; use std::time::{Instant, SystemTime, UNIX_EPOCH}; @@ -180,6 +182,26 @@ pub struct Inner { pub control: ControlStats, /// Path of the durable command-phase journal (`commands.jsonl`). pub command_journal: PathBuf, + + // --- RuView RF-context bridge (ADR-264 §8, `rf_bridge`) --- + /// Bounded ring of recently ingested, verification-tagged RF context. + /// Empty and never populated unless `--rf-upstream` is configured. + pub rf_recent: VecDeque, + /// RF-context ingest counters. + pub rf_stats: RfIngestStats, + /// The configured RF upstream, if any — retained here (not just in + /// `GatewayConfig`, which the running daemon doesn't otherwise keep) so + /// `GET /api/rf-context` can report what it's actually polling. + pub rf_upstream: Option, + /// Within-RF-domain fusion engine (`rf_bridge` module docs): combines + /// this upstream's own verified `FieldEvent`s into room-state inferences. + /// Needs no biome/WorldGraph pairing — see `rf_bridge` for why that's a + /// deliberately different, still-unbuilt kind of fusion. + pub rf_fusion: RuFieldFusion, + /// Bounded in-memory history of inference snapshots (`rf_bridge` module + /// docs) — survives across polls, not across a restart; RF inferences + /// have no biome `node_id`/`sequence` to durably cite yet. + pub rf_inference_history: VecDeque, } impl Inner { @@ -284,6 +306,11 @@ impl Inner { receipts: Vec::new(), control: ControlStats::default(), command_journal, + rf_recent: VecDeque::new(), + rf_stats: RfIngestStats::default(), + rf_upstream: config.rf_upstream.clone(), + rf_fusion: RuFieldFusion::new(), + rf_inference_history: VecDeque::new(), }) }