feat(sensing-server): consume ADR-110 §A0.12 sync packets, store per-node

Iter 15 — converts the iter 14 SyncPacket decoder from "shipped" to
"consumed" by wiring it into the sensing-server UDP receive loop.

Wiring:
- Cargo.toml gains wifi-densepose-hardware = path = "../wifi-densepose-hardware"
  to pull in the SyncPacket decoder + SYNC_PACKET_MAGIC dispatch constant.
- NodeState gains two new fields:
    latest_sync:    Option<SyncPacket>           — the parsed packet
    latest_sync_at: Option<std::time::Instant>   — staleness clock
- udp_receiver_task now magic-dispatches every incoming datagram against
  SYNC_PACKET_MAGIC (0xC511A110) before falling through to the existing
  ADR-039 vitals / ADR-040 WASM / ADR-018 CSI parsers. Same Option-returning
  pattern as the other parsers, so future packet types slot in cleanly.

When a sync packet arrives:
  * write-lock state, lookup-or-create NodeState by node_id
  * stash the SyncPacket + Instant::now() on the node
  * debug-log node, leader/valid/smoothed flags, sequence, offset_us
  * continue (don't fall through — we know it's not a CSI frame)

Downstream multistatic CSI fusion now has a documented landing pad: any
CSI frame with byte 19 bit 4 set looks up the matching NodeState, applies
ns.latest_sync.epoch_us + (now_local - ns.latest_sync.local_us) to get a
mesh-aligned timestamp. Implementation of that fusion math is the next
ADR-029/030 layer (wifi-densepose-signal).

Verification:
- cargo check -p wifi-densepose-sensing-server --no-default-features → green
- cargo test -p wifi-densepose-hardware sync_packet → 7/7 pass, 122 filtered
- Zero behavioral change for nodes that don't emit sync packets — the
  dispatch only fires on magic match.

Co-Authored-By: claude-flow <ruv@ruv.net>
This commit is contained in:
ruv 2026-05-23 13:11:35 -04:00
parent d72944f887
commit 23fd8ac371
2 changed files with 43 additions and 0 deletions

View File

@ -50,6 +50,9 @@ wifi-densepose-wifiscan = { version = "0.3.0", path = "../wifi-densepose-wifisca
# build without vcpkg/openblas (issue #366, #415).
wifi-densepose-signal = { version = "0.3.0", path = "../wifi-densepose-signal", default-features = false }
# Hardware crate — SyncPacket decoder for ADR-110 §A0.12 mesh-aligned timestamps.
wifi-densepose-hardware = { version = "0.3.0", path = "../wifi-densepose-hardware" }
# midstream — real-time introspection / low-latency tap (ADR-099 D1).
# Two crates only, on purpose: scheduler / neural-solver / strange-loop are
# explicitly out of scope of ADR-099 (D5).

View File

@ -371,6 +371,13 @@ struct NodeState {
latest_vitals: VitalSigns,
pub(crate) last_frame_time: Option<std::time::Instant>,
edge_vitals: Option<Esp32VitalsPacket>,
/// ADR-110 §A0.12: Latest sync packet received from this node. When a
/// CSI frame arrives with byte 19 bit 4 set (`adr018_flags.ieee802154_sync_valid`),
/// the host can recover a mesh-aligned timestamp via
/// `latest_sync.epoch_us + (now_local - latest_sync.local_us)`.
latest_sync: Option<wifi_densepose_hardware::SyncPacket>,
/// Last time a sync packet from this node was received (for staleness).
latest_sync_at: Option<std::time::Instant>,
/// Latest extracted features for cross-node fusion.
latest_features: Option<FeatureInfo>,
// ── RuVector Phase 2: Temporal smoothing & coherence gating ──
@ -434,6 +441,8 @@ impl NodeState {
latest_vitals: VitalSigns::default(),
last_frame_time: None,
edge_vitals: None,
latest_sync: None,
latest_sync_at: None,
latest_features: None,
prev_keypoints: None,
motion_energy_history: VecDeque::with_capacity(COHERENCE_WINDOW),
@ -4146,6 +4155,37 @@ async fn udp_receiver_task(state: SharedState, udp_port: u16) {
continue;
}
// ADR-110 §A0.12: Try sync packet (magic 0xC511_A110).
// A 32-byte UDP datagram carrying mesh-aligned epoch + sequence
// high-water from the node's c6_sync_espnow EMA-smoothed offset.
// Stored per-node so subsequent CSI frames with byte 19 bit 4
// set can have an aligned timestamp recovered downstream.
if len >= wifi_densepose_hardware::SYNC_PACKET_SIZE {
let magic = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
if magic == wifi_densepose_hardware::SYNC_PACKET_MAGIC {
match wifi_densepose_hardware::SyncPacket::from_bytes(&buf[..len]) {
Ok(sync) => {
debug!("ESP32 sync from {src}: node={} leader={} valid={} smoothed={} \
seq={} offset_us={}",
sync.node_id, sync.flags.is_leader, sync.flags.is_valid,
sync.flags.smoothed_used, sync.sequence,
sync.local_minus_epoch_us());
let mut s = state.write().await;
let ns = s.node_states.entry(sync.node_id)
.or_insert_with(NodeState::new);
ns.latest_sync = Some(sync);
ns.latest_sync_at = Some(std::time::Instant::now());
continue;
}
Err(e) => {
debug!("Sync packet decode error from {src}: {e}");
// Fall through — magic matched but decode failed; not a CSI frame.
continue;
}
}
}
}
// ADR-040: Try WASM output packet (magic 0xC511_0004).
if let Some(wasm_output) = parse_wasm_output(&buf[..len]) {
debug!("WASM output from {src}: node={} module={} events={}",