feat(sensing-server): loopback UDP tee for second consumers (ADR-380)

Add --udp-tee / RUVIEW_UDP_TEE: every datagram the source allowlist admits
is copied unchanged to up to four loopback targets, so a fusion host or
Cognitum cog can read full I/Q, sequence numbers and sync packets without
binding the CSI port. Targets must be loopback and may not reuse the
listener port or port 0. Sends are non-blocking and drops are counted. Off
by default.

Tests: cargo test -p wifi-densepose-sensing-server --no-default-features
--lib udp_tee (6 passed) and --bin sensing-server (332 passed). Loopback
smoke test of the debug binary: a 288-byte datagram sent to the UDP port
arrived byte for byte at the tee target, and a non-loopback --udp-tee
target made the server exit 1 with "must be a loopback address".

Co-Authored-By: claude-flow <ruv@ruv.net>

docs/adr/ADR-380-loopback-udp-tee.md

@@ -0,0 +1,58 @@
+# ADR 380: Loopback UDP tee for second consumers of the CSI stream
+
+## Status
+
+Proposed. Unit tested, and smoke tested on loopback: a datagram sent to the
+server arrived byte for byte at the tee target. Not yet exercised against a
+live multi node fleet.
+
+Date: 2026-09-28
+
+Numbering: On 2026-10-02 `main` had up to ADR 378 and open PRs claimed nothing
+higher; ADR 379 is the OTA first-boot health check. If another branch also
+takes 380, renumber whichever merges second.
+
+## Context
+
+A five node deployment (four ESP32 S3, one ESP32 C6) streams ADR 018 frames to
+one sensing server. A second consumer cannot see the raw stream.
+
+Each node sends to a single `target_ip` (`firmware/esp32-csi-node/main/stream_sender.c`).
+The server's WebSocket and REST outputs carry per node amplitude only: no
+phase, no I/Q, no sequence number, and no `0xC511A110` sync packets. A fusion
+host that aligns CSI with camera, radar, or lidar needs all of those. Binding a
+second listener to the CSI port is not an option, and an external relay in
+front of the port would make every source loopback, which the ADR 296
+allowlist always admits.
+
+## Decision
+
+`--udp-tee` (env `RUVIEW_UDP_TEE`) copies every datagram the source allowlist
+admits, byte for byte, to up to four `ip:port` targets.
+
+- Targets must be loopback, so raw CSI never leaves the host through the tee.
+- A target may not use port 0 or the listener's own port, which would loop.
+- Sends are non blocking. A copy that cannot be queued is dropped and counted,
+  so a slow consumer never stalls the receive loop.
+- The copy happens after the ADR 296 allowlist check and before any parsing,
+  so consumers see exactly what the server admitted, in every wire format.
+- The tee is off unless configured.
+
+## Consequences
+
+A fusion host or Cognitum cog can subscribe to the full data plane by reading
+a loopback port. A consumer that today binds the CSI port itself can keep its
+code and move to the tee port instead.
+
+The tee is byte transparent, so it needs no change when the ADR 018 header or
+a later wire version changes.
+
+Each target costs one `send_to` per admitted datagram. Four targets is the
+limit.
+
+## Validation
+
+```bash
+cd v2
+cargo test -p wifi-densepose-sensing-server --no-default-features --lib udp_tee
+```

docs/adr/README.md

@@ -2,6 +2,7 @@
 
 Latest proposed decisions:
 
+- [ADR-380: Loopback UDP tee for second consumers of the CSI stream](ADR-380-loopback-udp-tee.md)
 - [ADR-378: `ruview-live` showcase views — CSI waterfall, radar fan, animated cells](ADR-378-ruview-live-showcase-views.md)
 - [ADR-377: `ruview-live` — a Claude Code mod shipped inside `@ruvnet/ruview`](ADR-377-ruview-live-claude-code-mod.md)
 - [ADR-376: `ruview` — one npm install for every RuView component](ADR-376-ruview-umbrella-npm-package.md)
@@ -181,6 +182,7 @@ Statuses: **Proposed** (under discussion), **Accepted** (approved and/or impleme
 | [ADR-294](ADR-294-wifi-veil-integration.md) | WiFi Veil integration — emission-shaping countermeasure as advisory BFLD dependency | Accepted (initial implementation) |
 | [ADR-295](ADR-295-source-provenance-state-machine.md) | Source provenance state machine — synthetic can never present as live | Accepted (initial implementation) |
 | [ADR-296](ADR-296-sensor-data-plane-bind-hardening.md) | Sensor data-plane hardening — UDP bind control and source allowlist (step one) | Accepted (initial implementation) |
+| [ADR-380](ADR-380-loopback-udp-tee.md) | Loopback UDP tee for second consumers of the CSI stream (`--udp-tee`) | Proposed |
 | [ADR-297](ADR-297-multi-node-semantic-correctness.md) | Multi-node semantic correctness — per-node inference, node-keyed rate limiting, stale state | Accepted (initial implementation) |
 | [ADR-298](ADR-298-model-release-sanity-gates.md) | Model release sanity gates — block degenerate and mislabeled model artifacts | Accepted (initial implementation) |
 | [ADR-299](ADR-299-csi-data-incident-repo-controls.md) | Repository CSI data-incident controls — ignore rules and pre-commit/CI policy check | Accepted (controls implemented; tree remediation gated) |

v2/crates/wifi-densepose-sensing-server/src/lib.rs

@@ -43,6 +43,8 @@ pub mod telemetry;
 pub mod trainer;
 /// ADR-296: UDP data-plane bind scope decision + source IP/CIDR allowlist.
 pub mod udp_bind;
+/// ADR-380: loopback raw-datagram tee for second consumers of the CSI stream.
+pub mod udp_tee;
 pub mod vital_signs;
 /// ADR-270 Mist and NETGEAR telemetry providers.
 pub mod vendor_mist_netgear;

v2/crates/wifi-densepose-sensing-server/src/main.rs

@@ -129,6 +129,14 @@ struct Args {
     #[arg(long, env = "RUVIEW_UDP_INSECURE_LAN")]
     udp_insecure_lan: bool,
 
+    /// Copy every admitted UDP datagram, unchanged, to these loopback
+    /// `ip:port` targets (comma-separated, repeatable; env `RUVIEW_UDP_TEE`;
+    /// ADR-380). Lets a second consumer read full I/Q and sync packets without
+    /// binding the CSI port. Off by default.
+    /// Example: `--udp-tee 127.0.0.1:5007`.
+    #[arg(long = "udp-tee", value_name = "IP:PORT", env = "RUVIEW_UDP_TEE")]
+    udp_tee: Vec<String>,
+
     /// Path to UI static files (repo `ui/`; from `v2/` use `../ui` or rely on auto-detect)
     #[arg(long, default_value = "../ui")]
     ui_path: PathBuf,
@@ -9709,6 +9717,7 @@ async fn udp_receiver_task(
     bind_ip: std::net::IpAddr,
     udp_port: u16,
     allowlist: std::sync::Arc<wifi_densepose_sensing_server::udp_bind::UdpSourceAllowlist>,
+    tee: Option<std::sync::Arc<wifi_densepose_sensing_server::udp_tee::UdpTee>>,
 ) {
     let addr = format!("{bind_ip}:{udp_port}");
     let socket = match UdpSocket::bind(&addr).await {
@@ -9735,6 +9744,10 @@ async fn udp_receiver_task(
                     );
                     continue;
                 }
+                // ADR-380: second consumers see exactly what was admitted.
+                if let Some(tee) = &tee {
+                    tee.forward(&buf[..len]);
+                }
                 if len > 0 && buf[0] == b'{' {
                     match serde_json::from_slice::<wifi_densepose_hardware::vendor_rf::VendorRfEvent>(&buf[..len])
                         .map_err(|error| error.to_string())
@@ -12289,7 +12302,7 @@ async fn main() {
     if plan.bind_udp {
         // ADR-296: resolve the UDP bind scope + source allowlist and fail closed
         // on an unguarded routable bind, mirroring the OAuth boot refusal below.
-        use wifi_densepose_sensing_server::udp_bind;
+        use wifi_densepose_sensing_server::{udp_bind, udp_tee};
         let udp_bind_ip: std::net::IpAddr = match args.udp_bind.parse() {
             Ok(ip) => ip,
             Err(_) => {
@@ -12323,11 +12336,29 @@ async fn main() {
                 std::process::exit(1);
             }
         }
+        let udp_tee = match udp_tee::parse_targets(args.udp_tee.iter(), args.udp_port) {
+            Ok(targets) if targets.is_empty() => None,
+            Ok(targets) => match udp_tee::UdpTee::bind(targets) {
+                Ok(tee) => {
+                    info!("UDP tee: copying admitted datagrams to {:?}", tee.targets());
+                    Some(std::sync::Arc::new(tee))
+                }
+                Err(e) => {
+                    error!("Failed to open UDP tee socket: {e}");
+                    std::process::exit(1);
+                }
+            },
+            Err(e) => {
+                error!("Invalid --udp-tee: {e}");
+                std::process::exit(1);
+            }
+        };
         tokio::spawn(udp_receiver_task(
             state.clone(),
             udp_bind_ip,
             args.udp_port,
             udp_allowlist,
+            udp_tee,
         ));
         tokio::spawn(broadcast_tick_task(state.clone(), args.tick_ms));
     }

v2/crates/wifi-densepose-sensing-server/src/udp_tee.rs

@@ -0,0 +1,178 @@
+//! ADR-380: raw-datagram tee for the UDP data plane.
+//!
+//! The sensing server owns the CSI UDP port, and its WebSocket/REST outputs
+//! carry amplitude only. A second consumer (a fusion host, a Cognitum cog, a
+//! recorder) that needs full I/Q, sequence numbers and the `0xC511A110` sync
+//! packets gets them from this tee: every datagram the source allowlist admits
+//! is copied unchanged to each configured target.
+//!
+//! Targets are loopback-only, so the tee can never send raw CSI off the host,
+//! and may not use the listener's own port, which would loop datagrams back
+//! into the receiver (loopback is always admitted by the allowlist). Sends are
+//! non-blocking; a datagram that cannot be queued is dropped and counted rather
+//! than stalling the receive loop.
+
+use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+/// Upper bound on tee targets; each admitted datagram costs one send per target.
+pub const MAX_TEE_TARGETS: usize = 4;
+
+/// Parse tee targets from CLI/env specs. Each item may be a comma-separated
+/// list of `ip:port`. Returns an error on the first unusable entry rather than
+/// silently dropping it.
+pub fn parse_targets<I, S>(specs: I, listen_port: u16) -> Result<Vec<SocketAddr>, String>
+where
+    I: IntoIterator<Item = S>,
+    S: AsRef<str>,
+{
+    let mut targets: Vec<SocketAddr> = Vec::new();
+    for spec in specs {
+        for part in spec.as_ref().split(',') {
+            let part = part.trim();
+            if part.is_empty() {
+                continue;
+            }
+            let addr: SocketAddr = part
+                .parse()
+                .map_err(|_| format!("tee target '{part}' is not ip:port"))?;
+            if !addr.ip().is_loopback() {
+                return Err(format!("tee target '{part}' must be a loopback address"));
+            }
+            if addr.port() == 0 || addr.port() == listen_port {
+                return Err(format!(
+                    "tee target '{part}' may not use port 0 or the listener port {listen_port}"
+                ));
+            }
+            if targets.contains(&addr) {
+                continue;
+            }
+            if targets.len() >= MAX_TEE_TARGETS {
+                return Err(format!("too many tee targets (limit {MAX_TEE_TARGETS})"));
+            }
+            targets.push(addr);
+        }
+    }
+    Ok(targets)
+}
+
+/// Non-blocking forwarder that copies admitted datagrams to loopback targets.
+#[derive(Debug)]
+pub struct UdpTee {
+    v4: Option<UdpSocket>,
+    v6: Option<UdpSocket>,
+    targets: Vec<SocketAddr>,
+    forwarded: AtomicU64,
+    dropped: AtomicU64,
+}
+
+impl UdpTee {
+    /// Bind one ephemeral loopback socket per address family in use.
+    pub fn bind(targets: Vec<SocketAddr>) -> std::io::Result<Self> {
+        let open = |ip: std::net::IpAddr| -> std::io::Result<UdpSocket> {
+            let socket = UdpSocket::bind(SocketAddr::new(ip, 0))?;
+            socket.set_nonblocking(true)?;
+            Ok(socket)
+        };
+        let v4 = if targets.iter().any(SocketAddr::is_ipv4) {
+            Some(open(Ipv4Addr::LOCALHOST.into())?)
+        } else {
+            None
+        };
+        let v6 = if targets.iter().any(SocketAddr::is_ipv6) {
+            Some(open(Ipv6Addr::LOCALHOST.into())?)
+        } else {
+            None
+        };
+        Ok(Self {
+            v4,
+            v6,
+            targets,
+            forwarded: AtomicU64::new(0),
+            dropped: AtomicU64::new(0),
+        })
+    }
+
+    /// Copy one datagram to every target. Never blocks.
+    pub fn forward(&self, datagram: &[u8]) {
+        for target in &self.targets {
+            let socket = if target.is_ipv4() { &self.v4 } else { &self.v6 };
+            let sent = socket
+                .as_ref()
+                .is_some_and(|s| s.send_to(datagram, target).is_ok());
+            let counter = if sent { &self.forwarded } else { &self.dropped };
+            counter.fetch_add(1, Ordering::Relaxed);
+        }
+    }
+
+    /// Configured targets, for the startup summary.
+    pub fn targets(&self) -> &[SocketAddr] {
+        &self.targets
+    }
+
+    /// Datagram copies successfully handed to the OS.
+    pub fn forwarded(&self) -> u64 {
+        self.forwarded.load(Ordering::Relaxed)
+    }
+
+    /// Datagram copies that could not be sent (buffer full, no listener error).
+    pub fn dropped(&self) -> u64 {
+        self.dropped.load(Ordering::Relaxed)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+
+    #[test]
+    fn parses_loopback_targets_and_ignores_blanks_and_duplicates() {
+        let t = parse_targets(["127.0.0.1:5007, ,127.0.0.1:5007", "[::1]:5008"], 5006).unwrap();
+        assert_eq!(t.len(), 2);
+    }
+
+    #[test]
+    fn rejects_non_loopback_targets() {
+        let err = parse_targets(["192.0.2.10:5007"], 5006).unwrap_err();
+        assert!(err.contains("loopback"), "{err}");
+    }
+
+    #[test]
+    fn rejects_the_listener_port_to_prevent_a_loop() {
+        assert!(parse_targets(["127.0.0.1:5006"], 5006).is_err());
+        assert!(parse_targets(["127.0.0.1:0"], 5006).is_err());
+    }
+
+    #[test]
+    fn rejects_malformed_and_excess_targets() {
+        assert!(parse_targets(["localhost:5007"], 5006).is_err());
+        let many: Vec<String> = (0..=MAX_TEE_TARGETS)
+            .map(|i| format!("127.0.0.1:{}", 6000 + i))
+            .collect();
+        assert!(parse_targets(many, 5006).is_err());
+    }
+
+    #[test]
+    fn empty_spec_means_no_tee() {
+        assert!(parse_targets(Vec::<String>::new(), 5006)
+            .unwrap()
+            .is_empty());
+    }
+
+    #[test]
+    fn forwards_datagrams_byte_for_byte() {
+        let sink = UdpSocket::bind("127.0.0.1:0").unwrap();
+        sink.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+        let tee = UdpTee::bind(vec![sink.local_addr().unwrap()]).unwrap();
+
+        let datagram = [0x01, 0x00, 0x11, 0xC5, 7, 1, 64, 0, 0xFF];
+        tee.forward(&datagram);
+
+        let mut buf = [0u8; 64];
+        let (len, _) = sink.recv_from(&mut buf).unwrap();
+        assert_eq!(&buf[..len], &datagram);
+        assert_eq!(tee.forwarded(), 1);
+        assert_eq!(tee.dropped(), 0);
+    }
+}