net-mesh 0.36.0

//! Phase 4H: Proximity Graph Integration (PINGWAVE++)
//!
//! This module integrates pingwave discovery with the behavior plane:
//! - Enhanced pingwaves carrying capability summaries
//! - Proximity-aware capability routing
//! - Latency-weighted graph for routing decisions
//! - Integration with load balancer for locality-aware selection
//! - Automatic capability index updates from pingwave data

use dashmap::DashMap;
use parking_lot::RwLock;
use std::collections::{HashMap, HashSet, VecDeque};
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::time::{Duration, Instant};

use super::capability::{CapabilityFilter, CapabilitySet};
use super::loadbalance::{Endpoint, HealthStatus, LoadBalancer, LoadMetrics};
use super::metadata::NodeId;

/// Enhanced pingwave with capability summary
#[derive(Debug, Clone)]
pub struct EnhancedPingwave {
    /// Originating node ID
    pub origin_id: NodeId,
    /// Sequence number (monotonic per origin)
    pub seq: u64,
    /// Time-to-live (hop count remaining)
    pub ttl: u8,
    /// Hops traversed so far
    pub hop_count: u8,
    /// Origin timestamp (microseconds since epoch)
    pub origin_timestamp_us: u64,
    /// Capability summary hash (for quick change detection)
    pub capability_hash: u64,
    /// Capability version
    pub capability_version: u32,
    /// Load summary (0-255, 0=idle, 255=overloaded)
    pub load_level: u8,
    /// Health status
    pub health: HealthStatus,
    /// Primary capabilities (compact representation)
    pub primary_caps: PrimaryCapabilities,
}

/// Compact primary capabilities (fits in 8 bytes)
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct PrimaryCapabilities {
    /// Has GPU
    pub gpu: bool,
    /// Number of model slots
    pub model_slots: u8,
    /// Memory tier (0-7, 0=<1GB, 7=>256GB)
    pub memory_tier: u8,
    /// Available tools bitmap (first 8 common tools)
    pub tools_bitmap: u8,
    /// Custom flags
    pub flags: u32,
}

impl PrimaryCapabilities {
    /// Encode to 8 bytes
    pub fn to_bytes(&self) -> [u8; 8] {
        let mut buf = [0u8; 8];
        buf[0] = if self.gpu { 1 } else { 0 };
        buf[1] = self.model_slots;
        buf[2] = self.memory_tier;
        buf[3] = self.tools_bitmap;
        buf[4..8].copy_from_slice(&self.flags.to_le_bytes());
        buf
    }

    /// Decode from 8 bytes
    pub fn from_bytes(buf: &[u8; 8]) -> Self {
        Self {
            gpu: buf[0] != 0,
            model_slots: buf[1],
            memory_tier: buf[2],
            tools_bitmap: buf[3],
            flags: u32::from_le_bytes([buf[4], buf[5], buf[6], buf[7]]),
        }
    }

    /// Create from full capability set
    pub fn from_capability_set(caps: &CapabilitySet) -> Self {
        // Phase A.5.3: read through views() so this code keeps
        // working post-Phase-A.5.N (when the typed-struct fields
        // are removed and the projection becomes a tag-set scan).
        let views = caps.views();
        let hw = views.hardware();
        let memory_tier = match hw.memory_gb {
            0..=1 => 0,
            2..=4 => 1,
            5..=8 => 2,
            9..=16 => 3,
            17..=32 => 4,
            33..=64 => 5,
            65..=128 => 6,
            _ => 7,
        };

        Self {
            gpu: hw.gpu.is_some(),
            model_slots: views.models().len() as u8,
            memory_tier,
            tools_bitmap: 0, // Could map common tools to bits
            flags: 0,
        }
    }

    /// Quick check if this matches a filter
    pub fn matches_basic(&self, filter: &CapabilityFilter) -> bool {
        if filter.require_gpu && !self.gpu {
            return false;
        }
        true
    }
}

impl EnhancedPingwave {
    /// Wire size in bytes (64 base + 8 primary_caps)
    pub const SIZE: usize = 72;

    /// Create a new enhanced pingwave
    pub fn new(origin_id: NodeId, seq: u64, ttl: u8) -> Self {
        Self {
            origin_id,
            seq,
            ttl,
            hop_count: 0,
            origin_timestamp_us: current_time_us(),
            capability_hash: 0,
            capability_version: 0,
            load_level: 0,
            health: HealthStatus::Healthy,
            primary_caps: PrimaryCapabilities::default(),
        }
    }

    /// Set capability info
    pub fn with_capabilities(
        mut self,
        hash: u64,
        version: u32,
        primary: PrimaryCapabilities,
    ) -> Self {
        self.capability_hash = hash;
        self.capability_version = version;
        self.primary_caps = primary;
        self
    }

    /// Set load info
    pub fn with_load(mut self, load_level: u8, health: HealthStatus) -> Self {
        self.load_level = load_level;
        self.health = health;
        self
    }

    /// Serialize to bytes
    pub fn to_bytes(&self) -> [u8; Self::SIZE] {
        let mut buf = [0u8; Self::SIZE];
        buf[0..32].copy_from_slice(&self.origin_id);
        buf[32..40].copy_from_slice(&self.seq.to_le_bytes());
        buf[40] = self.ttl;
        buf[41] = self.hop_count;
        buf[42..50].copy_from_slice(&self.origin_timestamp_us.to_le_bytes());
        buf[50..58].copy_from_slice(&self.capability_hash.to_le_bytes());
        buf[58..62].copy_from_slice(&self.capability_version.to_le_bytes());
        buf[62] = self.load_level;
        buf[63] = self.health as u8;
        buf[64..72].copy_from_slice(&self.primary_caps.to_bytes());
        buf
    }

    /// Deserialize from bytes
    pub fn from_bytes(buf: &[u8]) -> Option<Self> {
        if buf.len() < Self::SIZE {
            return None;
        }
        let mut origin_id = [0u8; 32];
        origin_id.copy_from_slice(&buf[0..32]);

        let mut caps_buf = [0u8; 8];
        caps_buf.copy_from_slice(&buf[64..72]);

        Some(Self {
            origin_id,
            seq: u64::from_le_bytes(buf[32..40].try_into().ok()?),
            ttl: buf[40],
            hop_count: buf[41],
            origin_timestamp_us: u64::from_le_bytes(buf[42..50].try_into().ok()?),
            capability_hash: u64::from_le_bytes(buf[50..58].try_into().ok()?),
            capability_version: u32::from_le_bytes(buf[58..62].try_into().ok()?),
            load_level: buf[62],
            // Previously coerced any unknown byte to
            // `HealthStatus::Unknown`. A flipped byte downgrades the
            // peer to `Unknown`, which `can_receive_traffic()` treats
            // as unroutable — silent peer eviction on a single
            // bit-flip. `from_bytes` callers already handle `None`
            // (this returns `Option<Self>`), so refuse the parse on
            // unknown discriminants instead of guessing.
            health: match buf[63] {
                0 => HealthStatus::Healthy,
                1 => HealthStatus::Degraded,
                2 => HealthStatus::Unhealthy,
                3 => HealthStatus::Unknown,
                _ => return None,
            },
            primary_caps: PrimaryCapabilities::from_bytes(&caps_buf),
        })
    }

    /// Check if expired
    pub fn is_expired(&self) -> bool {
        self.ttl == 0
    }

    /// Forward (decrement TTL, increment hop count)
    pub fn forward(&mut self) -> bool {
        if self.ttl == 0 {
            return false;
        }
        self.ttl -= 1;
        self.hop_count = self.hop_count.saturating_add(1);
        true
    }

    /// Calculate one-way latency estimate (microseconds)
    pub fn latency_estimate_us(&self) -> u64 {
        let now = current_time_us();
        now.saturating_sub(self.origin_timestamp_us)
    }
}

/// Outcome of admitting an inbound pingwave into the proximity graph
/// ([`ProximityGraph::admit_pingwave_from`]).
///
/// The variant is what lets the receive path order dedup BEFORE any
/// routing-table mutation (RT-5 review: duplicate pingwave resurrects a
/// withdrawn route). A `Option<EnhancedPingwave>` return conflated
/// "rejected duplicate" with "accepted but expired" — both mapped to
/// `None` — so a caller could not tell whether it was safe to install
/// or refresh a route. It never is for a duplicate.
#[derive(Debug)]
pub enum PingwaveAdmission {
    /// The `(origin, seq)` pair was already seen — rejected before any
    /// node/edge mutation. The receive path MUST NOT install or refresh
    /// a route for it, or a byte-identical replay would undo a
    /// withdrawal that just removed the route + edge.
    RejectedDuplicate,
    /// A new pingwave whose local node/edge state was updated, but which
    /// is TTL-expired and so must NOT be re-broadcast. A route to the
    /// origin may still be installed/refreshed from it.
    AcceptedNoForward,
    /// A new, non-expired pingwave: update local state, install/refresh
    /// the route, AND re-broadcast the returned (hop-advanced) pingwave.
    AcceptedAndForward(EnhancedPingwave),
    /// The graph is at a configured cap (`max_nodes` /
    /// `max_seen_pingwaves`), so this novel pingwave was refused
    /// before it could grow attacker-influenced state. No route may be
    /// installed and nothing is re-broadcast.
    ///
    /// Distinct from [`RejectedDuplicate`](Self::RejectedDuplicate) so
    /// the two are distinguishable in metrics: duplicates are normal
    /// mesh chatter, whereas a sustained stream of these means the
    /// graph is saturated — either the mesh outgrew its configured
    /// caps, or something is flooding it.
    RejectedCapacity,
}

/// Proximity node info combining discovery and capability data
#[derive(Debug)]
pub struct ProximityNode {
    /// Node ID
    pub node_id: NodeId,
    /// Network address
    pub addr: SocketAddr,
    /// Hop distance
    pub hops: u8,
    /// Estimated latency in microseconds
    pub latency_us: u64,
    /// Last seen timestamp
    pub last_seen: Instant,
    /// Latest pingwave sequence
    pub last_seq: u64,
    /// Capability hash (for change detection)
    pub capability_hash: u64,
    /// Capability version
    pub capability_version: u32,
    /// Primary capabilities (quick filter)
    pub primary_caps: PrimaryCapabilities,
    /// Current load level
    pub load_level: u8,
    /// Health status
    pub health: HealthStatus,
    /// Full capabilities (lazy loaded)
    full_capabilities: RwLock<Option<CapabilitySet>>,
}

impl Clone for ProximityNode {
    fn clone(&self) -> Self {
        Self {
            node_id: self.node_id,
            addr: self.addr,
            hops: self.hops,
            latency_us: self.latency_us,
            last_seen: self.last_seen,
            last_seq: self.last_seq,
            capability_hash: self.capability_hash,
            capability_version: self.capability_version,
            primary_caps: self.primary_caps,
            load_level: self.load_level,
            health: self.health,
            full_capabilities: RwLock::new(self.full_capabilities.read().clone()),
        }
    }
}

impl ProximityNode {
    /// Create new proximity node from pingwave
    pub fn from_pingwave(pw: &EnhancedPingwave, addr: SocketAddr) -> Self {
        // `pw.hop_count + 1` would panic in debug at u8::MAX and
        // silently wrap to 0 in release. A buggy or malicious peer
        // can advertise `hop_count == 255`, after which:
        //   - Debug builds would panic the receive loop.
        //   - Release builds would record `hops=0`, falsely
        //     promoting the node to "directly connected" status —
        //     a proximity-routing poisoning vector.
        // `saturating_add(1)` keeps hops at 255 in the overflow
        // case; combined with `MAX_HOPS` cap on routing
        // installation, this is a non-poisoning floor.
        Self {
            node_id: pw.origin_id,
            addr,
            hops: pw.hop_count.saturating_add(1),
            latency_us: pw.latency_estimate_us(),
            last_seen: Instant::now(),
            last_seq: pw.seq,
            capability_hash: pw.capability_hash,
            capability_version: pw.capability_version,
            primary_caps: pw.primary_caps,
            load_level: pw.load_level,
            health: pw.health,
            full_capabilities: RwLock::new(None),
        }
    }

    /// Update from new pingwave
    pub fn update_from_pingwave(&mut self, pw: &EnhancedPingwave, addr: SocketAddr) {
        // Same `+ 1` overflow concern as `from_pingwave`. Use
        // `saturating_add` here too. The "better path" comparison
        // also uses the saturated value so a 255-hop pingwave
        // can never falsely beat a real path.
        let new_hops = pw.hop_count.saturating_add(1);

        // Separate freshness from path quality. Pre-fix the OR
        // (`seq > last_seq || new_hops < self.hops`) let a flooded
        // high-seq pingwave delivered through a long route demote
        // a previously-cached direct route — the freshness arm
        // accepted the new (worse) path purely because of seq
        // monotonicity. Now: track `last_seq` as the freshness
        // signal even on long routes, but only adopt the new
        // `addr` / `hops` / `latency_us` when the path is
        // genuinely no worse than what we have.
        if pw.seq > self.last_seq {
            self.last_seq = pw.seq;
        }
        if new_hops <= self.hops {
            self.addr = addr;
            self.hops = new_hops;
            self.latency_us = pw.latency_estimate_us();
        }

        // Always update load/health from latest
        self.load_level = pw.load_level;
        self.health = pw.health;
        self.last_seen = Instant::now();

        // Check capability change
        if pw.capability_version > self.capability_version {
            self.capability_hash = pw.capability_hash;
            self.capability_version = pw.capability_version;
            self.primary_caps = pw.primary_caps;
            // Clear cached full capabilities
            *self.full_capabilities.write() = None;
        }
    }

    /// Check if node is stale
    pub fn is_stale(&self, timeout: Duration) -> bool {
        self.last_seen.elapsed() > timeout
    }

    /// Check if node is available for routing
    pub fn is_available(&self) -> bool {
        self.health.can_receive_traffic()
    }

    /// Get or fetch full capabilities
    pub fn get_capabilities(&self) -> Option<CapabilitySet> {
        self.full_capabilities.read().clone()
    }

    /// Set full capabilities (after fetching)
    pub fn set_capabilities(&self, caps: CapabilitySet) {
        *self.full_capabilities.write() = Some(caps);
    }

    /// Calculate routing score (lower is better)
    pub fn routing_score(&self, prefer_low_latency: bool) -> f64 {
        let latency_factor = if prefer_low_latency {
            (self.latency_us as f64) / 1000.0 // Convert to ms
        } else {
            self.hops as f64 * 10.0 // 10ms per hop estimate
        };

        let load_factor = (self.load_level as f64) / 255.0 * 50.0; // 0-50 penalty

        let health_factor = match self.health {
            HealthStatus::Healthy => 0.0,
            HealthStatus::Degraded => 25.0,
            HealthStatus::Unhealthy => 1000.0,
            HealthStatus::Unknown => 50.0,
        };

        latency_factor + load_factor + health_factor
    }
}

/// Edge in the proximity graph
#[derive(Debug, Clone)]
pub struct ProximityEdge {
    /// Source node
    pub from: NodeId,
    /// Destination node
    pub to: NodeId,
    /// Latency in microseconds
    pub latency_us: u64,
    /// Last updated
    pub last_updated: Instant,
    /// Reliability (0.0-1.0, based on packet loss)
    pub reliability: f32,
}

/// Configuration for the proximity graph
#[derive(Debug, Clone)]
pub struct ProximityConfig {
    /// Maximum hops to track
    pub radius: u8,
    /// Node timeout
    pub node_timeout: Duration,
    /// Pingwave dedup cache timeout
    pub dedup_timeout: Duration,
    /// Pingwave interval
    pub pingwave_interval: Duration,
    /// Whether to prefer low latency over hop count
    pub prefer_low_latency: bool,
    /// Maximum nodes to track.
    ///
    /// A soft cap, enforced in [`ProximityGraph::admit_pingwave_from`]:
    /// at or above it, a *novel* `origin_id` is not inserted, while
    /// known nodes keep updating. Periodic
    /// [`cleanup`](ProximityGraph::cleanup) reclaims slots as entries
    /// idle out.
    pub max_nodes: usize,
    /// Maximum `(origin_id, seq)` dedup entries to retain.
    ///
    /// Sized as 4× `max_nodes` by default, matching the ratio
    /// [`MAX_SEEN_PINGWAVES`](crate::adapter::net::swarm::MAX_SEEN_PINGWAVES)
    /// uses, so a multi-second pingwave burst per node fits before the
    /// cap bites.
    ///
    /// A quarter of it is reserved for origins the graph already knows,
    /// so a flood of novel origins cannot suppress the pingwaves of the
    /// topology it is competing with — see
    /// `ProximityGraph::unreserved_dedup_capacity`.
    pub max_seen_pingwaves: usize,
    /// Maximum directed edges to retain.
    ///
    /// Edges are the third attacker-influenced map: each accepted
    /// pingwave from peer Z carrying origin Y inserts `(Z, Y)`, so a
    /// flood of distinct origins grows this alongside `nodes`. Capped
    /// at 4× `max_nodes` because a node legitimately has several
    /// in-edges.
    pub max_edges: usize,
    /// Whether to auto-update capability index
    pub auto_index_update: bool,
}

impl Default for ProximityConfig {
    fn default() -> Self {
        Self {
            radius: 3,
            node_timeout: Duration::from_secs(30),
            dedup_timeout: Duration::from_secs(10),
            pingwave_interval: Duration::from_secs(5),
            prefer_low_latency: true,
            max_nodes: 10000,
            max_seen_pingwaves: 40_000,
            max_edges: 40_000,
            auto_index_update: true,
        }
    }
}

/// Proximity graph integrating discovery with behavior plane
pub struct ProximityGraph {
    /// Local node ID
    my_id: NodeId,
    /// Configuration
    config: ProximityConfig,
    /// Known nodes
    nodes: DashMap<NodeId, ProximityNode>,
    /// Edges (from, to) -> edge info
    edges: DashMap<(NodeId, NodeId), ProximityEdge>,
    /// Seen pingwaves for deduplication
    seen_pingwaves: DashMap<(NodeId, u64), Instant>,
    /// O(1) entry counts for `nodes` / `edges` / `seen_pingwaves`. Avoids the
    /// per-shard walk of `DashMap::len()` (~1us) in `node_count()` / `stats()`.
    /// Maintained exactly on every insert and decremented on eviction. See
    /// docs/internal/misc/PERF_AUDIT_2026_06_08_BENCHMARK_WINS.md §2.
    num_nodes: AtomicUsize,
    num_edges: AtomicUsize,
    num_seen: AtomicUsize,
    /// Next pingwave sequence
    next_seq: AtomicU64,
    /// Local capability hash
    local_capability_hash: AtomicU64,
    /// Local capability version
    local_capability_version: AtomicU64,
    /// Local capabilities
    local_capabilities: RwLock<Option<CapabilitySet>>,
    /// Local load level
    local_load_level: AtomicU64,
    /// Statistics
    stats: ProximityStats,
}

/// Proximity graph statistics
#[derive(Debug, Default)]
pub struct ProximityStats {
    /// Number of ping waves initiated by this node
    pub pingwaves_sent: AtomicU64,
    /// Number of ping waves received from other nodes
    pub pingwaves_received: AtomicU64,
    /// Number of ping waves forwarded to neighbors
    pub pingwaves_forwarded: AtomicU64,
    /// Number of ping waves dropped due to deduplication or TTL expiry
    pub pingwaves_dropped: AtomicU64,
    /// Number of new nodes discovered through ping waves
    pub nodes_discovered: AtomicU64,
    /// Number of nodes removed after failing liveness checks
    pub nodes_expired: AtomicU64,
    /// Number of capability set updates processed
    pub capability_updates: AtomicU64,
}

impl ProximityGraph {
    /// Create a new proximity graph
    pub fn new(my_id: NodeId, config: ProximityConfig) -> Self {
        Self {
            my_id,
            config,
            nodes: DashMap::new(),
            edges: DashMap::new(),
            seen_pingwaves: DashMap::new(),
            num_nodes: AtomicUsize::new(0),
            num_edges: AtomicUsize::new(0),
            num_seen: AtomicUsize::new(0),
            next_seq: AtomicU64::new(1),
            local_capability_hash: AtomicU64::new(0),
            local_capability_version: AtomicU64::new(0),
            local_capabilities: RwLock::new(None),
            local_load_level: AtomicU64::new(0),
            stats: ProximityStats::default(),
        }
    }

    /// Get local node ID
    pub fn my_id(&self) -> NodeId {
        self.my_id
    }

    /// Set local capabilities
    pub fn set_local_capabilities(&self, caps: CapabilitySet) {
        // Take the caps write lock FIRST so the three updates
        // (hash, version, the `Option<CapabilitySet>`) cannot
        // tear from a concurrent reader's perspective. Pre-fix:
        //   1. `fetch_add` then `store(version)` was a classic
        //      lost-update race - two callers A and B doing
        //      `fetch_add -> 1` / `fetch_add -> 2` could
        //      interleave their stores so the final atomic
        //      reads back the older value, regressing the
        //      version counter and breaking the "newer caps
        //      always have a strictly-higher version" contract
        //      pingwave consumers rely on.
        //   2. Three independent unordered writes (hash, version,
        //      RwLock) let a reader inside `create_pingwave`
        //      sample a hash that didn't match the version it
        //      read, or a CapabilitySet that didn't match either.
        // Holding the write lock across all three serialises
        // concurrent set_local_capabilities callers, and uses
        // the fetch_add return value directly (no second store).
        let hash = hash_capabilities(&caps);
        let mut guard = self.local_capabilities.write();
        // fetch_add returns the prior value; the atomic is now
        // at `prior + 1`. Use that value directly - storing it
        // back is what introduced the lost-update race pre-fix.
        let _new_version = self
            .local_capability_version
            .fetch_add(1, Ordering::AcqRel)
            .wrapping_add(1);
        self.local_capability_hash.store(hash, Ordering::Release);
        *guard = Some(caps);
    }

    /// Set local load level (0-255)
    pub fn set_local_load(&self, load_level: u8) {
        self.local_load_level
            .store(load_level as u64, Ordering::Relaxed);
    }

    /// Create a pingwave to broadcast
    pub fn create_pingwave(&self, health: HealthStatus) -> EnhancedPingwave {
        let seq = self.next_seq.fetch_add(1, Ordering::Relaxed);
        let caps = self.local_capabilities.read();
        let primary = caps
            .as_ref()
            .map(PrimaryCapabilities::from_capability_set)
            .unwrap_or_default();

        self.stats.pingwaves_sent.fetch_add(1, Ordering::Relaxed);

        EnhancedPingwave::new(self.my_id, seq, self.config.radius)
            .with_capabilities(
                self.local_capability_hash.load(Ordering::Relaxed),
                self.local_capability_version.load(Ordering::Relaxed) as u32,
                primary,
            )
            .with_load(self.local_load_level.load(Ordering::Relaxed) as u8, health)
    }

    /// Back-compat shim: attribute the pingwave as if it arrived
    /// directly from its origin (i.e. `from_node = pw.origin_id`).
    /// Tests and benchmarks that don't model a separate forwarding
    /// hop call this shape; production dispatch should use the full
    /// [`Self::on_pingwave_from`] so multi-hop edge attribution is
    /// correct.
    pub fn on_pingwave(
        &self,
        pw: EnhancedPingwave,
        from_addr: SocketAddr,
    ) -> Option<EnhancedPingwave> {
        let from_node = pw.origin_id;
        self.on_pingwave_from(pw, from_node, from_addr)
    }

    /// Process incoming pingwave.
    ///
    /// `from_node` is the graph-id of the **direct peer** that just
    /// forwarded this pingwave to us (i.e. the sender on the wire), not
    /// the pingwave's origin. On multi-hop paths `from_node` differs
    /// from `pw.origin_id`; on the first-hop case (a pingwave direct
    /// from its origin) they match.
    ///
    /// Returns `Some(pingwave)` if it should be forwarded, `None`
    /// otherwise. Thin back-compat wrapper over
    /// [`Self::admit_pingwave_from`] for callers (the shim above, tests,
    /// benches) that only care about the forward decision; the receive
    /// path uses the richer [`PingwaveAdmission`] so it can gate route
    /// installation on admission (RT-5 review: a duplicate must not
    /// resurrect a withdrawn route).
    pub fn on_pingwave_from(
        &self,
        pw: EnhancedPingwave,
        from_node: NodeId,
        from_addr: SocketAddr,
    ) -> Option<EnhancedPingwave> {
        match self.admit_pingwave_from(pw, from_node, from_addr) {
            PingwaveAdmission::AcceptedAndForward(fwd) => Some(fwd),
            PingwaveAdmission::RejectedDuplicate
            | PingwaveAdmission::RejectedCapacity
            | PingwaveAdmission::AcceptedNoForward => None,
        }
    }

    /// How much of the dedup cache a *novel* origin may fill.
    ///
    /// The remainder — a quarter of
    /// [`max_seen_pingwaves`](ProximityConfig::max_seen_pingwaves) — is
    /// headroom that only origins already in the graph can reach.
    ///
    /// A dedup entry is a `(origin_id, seq)` pair, and a known peer's
    /// next pingwave always carries a fresh `seq`, so without this a
    /// flood of novel origins could fill the cache and every legitimate
    /// pingwave would then be refused until the next
    /// [`cleanup`](Self::cleanup) — with `dedup_timeout` at 10 s and the
    /// sweep on a 60 s cadence, roughly a minute of suppressed
    /// discovery, refillable at will for the cost of a few MB of UDP.
    /// The known peers would then idle past `node_timeout` and be
    /// evicted, so the flood would not merely fail to add topology: it
    /// would destroy the topology already there.
    ///
    /// Reserving is enough because the reserve only has to absorb what
    /// known peers produce between sweeps. It is not a second ceiling:
    /// `max_seen_pingwaves` still binds every origin, known or not, so
    /// the memory bound this cap exists for is unchanged.
    fn unreserved_dedup_capacity(&self) -> usize {
        let max = self.config.max_seen_pingwaves;
        // `max - max/4`, never below 1 for a nonzero cap: a
        // pathologically small configuration should still admit
        // something rather than reserving the entire cache.
        max.saturating_sub(max / 4).max(usize::from(max > 0))
    }

    /// Admit an inbound pingwave: dedup, then (only for a NEW pingwave)
    /// update node + edge state and report whether the caller may
    /// install a route and/or re-broadcast.
    ///
    /// Dedup precedes every mutation: a [`PingwaveAdmission::RejectedDuplicate`]
    /// is returned before any node/edge is touched, so the receive path
    /// that gates route installation on the result cannot reinstall a
    /// route (or re-add an edge) that a withdrawal just removed when a
    /// byte-identical pingwave is replayed or duplicated (RT-5 review).
    pub fn admit_pingwave_from(
        &self,
        mut pw: EnhancedPingwave,
        from_node: NodeId,
        from_addr: SocketAddr,
    ) -> PingwaveAdmission {
        self.stats
            .pingwaves_received
            .fetch_add(1, Ordering::Relaxed);

        // Ignore our own pingwaves (origin self-check — also defends
        // against a buffered pingwave we emitted earlier being replayed
        // back at us by a partitioned-then-healed peer). Treated as a
        // duplicate for admission purposes: no mutation, no forward.
        if pw.origin_id == self.my_id {
            return PingwaveAdmission::RejectedDuplicate;
        }

        // Check dedup cache
        let key = (pw.origin_id, pw.seq);
        if self.seen_pingwaves.contains_key(&key) {
            self.stats.pingwaves_dropped.fetch_add(1, Ordering::Relaxed);
            return PingwaveAdmission::RejectedDuplicate;
        }

        // SEC-02. `origin_id` and `seq` are attacker-chosen: the frame
        // is a fixed 72-byte unsigned format admitted on nothing more
        // than a registered source address, and `origin_id` is 256-bit,
        // so accidental dedup collisions are negligible. Every novel
        // tuple used to grow `seen_pingwaves`, `nodes` and `edges` for
        // the process lifetime, and the only thing that reclaimed the
        // first two — `cleanup()` — had no production caller at all.
        // A peer could grow them at line rate until the node died of
        // memory exhaustion.
        //
        // Same soft-cap policy the older `LocalGraph` already carries
        // for this exact threat (see `MAX_GRAPH_NODES` /
        // `MAX_SEEN_PINGWAVES` in `swarm.rs`): at the cap, novel keys
        // are refused while known ones keep updating, so a flood
        // cannot evict the legitimate topology it is competing with —
        // it can only fail to add to it. Periodic cleanup reclaims
        // slots as entries idle out.
        //
        // Refuse the whole pingwave rather than admitting it
        // untracked: without a dedup entry we would re-admit and
        // re-forward every copy of it, turning a memory cap into a
        // rebroadcast amplifier.
        //
        // The cap is applied the same way the node cap below is: to
        // origins the graph does not already know. A flat check here
        // would have read on *every* pingwave, and a known peer's next
        // one always carries a fresh `seq` — so a saturated cache would
        // refuse the legitimate topology along with the flood, which is
        // the opposite of the policy stated above. `reserved_dedup_slots`
        // is the headroom that keeps that from happening; the absolute
        // ceiling still binds everyone, so the memory bound is unchanged.
        let seen = self.num_seen.load(Ordering::Relaxed);
        let known_origin = self.nodes.contains_key(&pw.origin_id);
        if seen >= self.config.max_seen_pingwaves
            || (!known_origin && seen >= self.unreserved_dedup_capacity())
        {
            self.stats.pingwaves_dropped.fetch_add(1, Ordering::Relaxed);
            return PingwaveAdmission::RejectedCapacity;
        }

        // Key the dedup-count bump on the insert result (None == new key) so
        // it stays exact even under a concurrent insert of the same key.
        if self.seen_pingwaves.insert(key, Instant::now()).is_none() {
            self.num_seen.fetch_add(1, Ordering::Relaxed);
        }

        // Node cap. Checked after the dedup insert so a refused novel
        // origin is still deduplicated — otherwise every retransmission
        // of it would take the full admission path again.
        if !known_origin && self.num_nodes.load(Ordering::Relaxed) >= self.config.max_nodes {
            self.stats.pingwaves_dropped.fetch_add(1, Ordering::Relaxed);
            return PingwaveAdmission::RejectedCapacity;
        }

        // Update or create node
        self.nodes
            .entry(pw.origin_id)
            .and_modify(|node| node.update_from_pingwave(&pw, from_addr))
            .or_insert_with(|| {
                self.stats.nodes_discovered.fetch_add(1, Ordering::Relaxed);
                self.num_nodes.fetch_add(1, Ordering::Relaxed);
                ProximityNode::from_pingwave(&pw, from_addr)
            });

        // Topology edges: a pingwave carrying origin Y that we just
        // received via direct peer Z tells us two facts:
        //   * we have a direct edge to Z (already true by
        //     construction — Z is our direct peer),
        //   * Z has a route to Y (otherwise Z wouldn't be forwarding).
        //
        // The first is redundant after the initial insert; the
        // `last_updated` refresh on re-insert is the cheap liveness
        // signal. The second is what makes `path_to(Y)` able to
        // return multi-hop paths.
        //
        // Latency estimate: `now_us − pw.origin_timestamp_us` is a
        // noisy one-way delay; clock-skew-sensitive, but good enough
        // as an equal-hop tiebreaker. EWMA (α = 1/8) smooths
        // successive samples per `(from, to)` pair.
        //
        // Throttle the self-edge `(my_id → Z)` update: a hot
        // pingwave-receive path (one per peer per heartbeat
        // interval, scaled across N peers) hit the DashMap
        // entry lock + `Instant::now()` on every receive even
        // though the liveness signal only needs second-level
        // freshness. Skip the update when the existing edge is
        // less than a second old; the multi-hop edge below still
        // refreshes unconditionally because it carries a fresh
        // latency sample.
        let now_us = current_time_us();
        let sample_us = now_us.saturating_sub(pw.origin_timestamp_us);
        let needs_self_edge_refresh = self
            .edges
            .get(&(self.my_id, from_node))
            .map(|e| e.last_updated.elapsed() >= Duration::from_secs(1))
            .unwrap_or(true);
        if needs_self_edge_refresh {
            self.insert_or_update_edge(self.my_id, from_node, 0);
        }
        if from_node != pw.origin_id {
            self.insert_or_update_edge(from_node, pw.origin_id, sample_us);
        }

        // Check if should forward. Expired = accepted (local state was
        // updated above) but not re-broadcast.
        if pw.is_expired() {
            return PingwaveAdmission::AcceptedNoForward;
        }

        // Forward
        pw.forward();
        self.stats
            .pingwaves_forwarded
            .fetch_add(1, Ordering::Relaxed);
        PingwaveAdmission::AcceptedAndForward(pw)
    }

    /// Latency of the directed edge `from → to`, if the graph holds
    /// one (pingwave-learned, EWMA-smoothed). `None` when the pair
    /// has never been observed. The sensing rendezvous reads these
    /// as its shared centrality inputs
    /// (SENSING_INTEREST_COALESCING_PLAN §4.1); callers wanting an
    /// undirected view should try both orientations.
    pub fn edge_latency(&self, from: NodeId, to: NodeId) -> Option<Duration> {
        self.edges
            .get(&(from, to))
            .map(|edge| Duration::from_micros(edge.value().latency_us))
    }

    /// Test-only helper — install or refresh one directed edge with
    /// an explicit latency sample, bypassing the pingwave path.
    /// Fixtures inject topology this way so the SI-2b candidate
    /// resolver's reads ([`Self::edge_latency`], [`Self::path_to`])
    /// can be exercised without timing-dependent pingwaves.
    #[doc(hidden)]
    pub fn test_insert_edge(&self, from: NodeId, to: NodeId, latency_us: u64) {
        self.insert_or_update_edge(from, to, latency_us);
    }

    /// Remove one directed edge (RT-5,
    /// REALTIME_ROUTING_AND_DISCOVERY_PLAN). Called when a peer
    /// withdraws its route toward `to`: the `(peer, to)` edge is what
    /// `path_to` would otherwise keep using to synthesize alternates
    /// through the withdrawn hop. Removing an absent edge is a no-op.
    pub fn remove_edge(&self, from: NodeId, to: NodeId) -> bool {
        let removed = self.edges.remove(&(from, to)).is_some();
        if removed {
            self.num_edges.fetch_sub(1, Ordering::Relaxed);
        }
        removed
    }

    /// Insert or refresh an edge. If the edge already exists, EWMA the
    /// latency sample into `latency_us` (α = 1/8) and bump
    /// `last_updated`. `sample_us == 0` means "no latency info" (e.g.
    /// the self → peer edge added at session setup); leave the
    /// existing latency alone in that case.
    fn insert_or_update_edge(&self, from: NodeId, to: NodeId, sample_us: u64) {
        // SEC-02 soft cap. Novel edges are refused at the cap; existing
        // ones keep taking latency samples, so a flood degrades
        // discovery of new topology without disturbing what is already
        // known. `sweep_stale_edges` (driven from the heartbeat tick)
        // reclaims slots — unlike `nodes` and `seen_pingwaves`, edges
        // always had a production sweep; what they lacked was a bound
        // on growth between sweeps.
        if !self.edges.contains_key(&(from, to))
            && self.num_edges.load(Ordering::Relaxed) >= self.config.max_edges
        {
            return;
        }
        let mut edge_inserted = false;
        self.edges
            .entry((from, to))
            .and_modify(|edge| {
                if sample_us > 0 {
                    // α = 1/8 EWMA on integer microseconds.
                    let prev = edge.latency_us;
                    edge.latency_us = prev - prev / 8 + sample_us / 8;
                }
                edge.last_updated = Instant::now();
            })
            .or_insert_with(|| {
                edge_inserted = true;
                ProximityEdge {
                    from,
                    to,
                    latency_us: sample_us,
                    last_updated: Instant::now(),
                    reliability: 1.0,
                }
            });
        if edge_inserted {
            self.num_edges.fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Drop edges whose `last_updated` is older than `max_age`. Called
    /// from the heartbeat-loop tick alongside `RoutingTable::sweep_stale`
    /// so the graph and the routing table age out in lockstep. Returns
    /// the number of edges removed.
    ///
    /// Uses `DashMap::retain` so the staleness check + remove is
    /// atomic per entry. A collect-stale-keys-then-remove shape would
    /// race with concurrent pingwave receipt: another thread could
    /// refresh an edge's `last_updated` between the collect and
    /// remove phases, and we'd delete a freshly-alive edge.
    pub fn sweep_stale_edges(&self, max_age: Duration) -> usize {
        let cutoff = match Instant::now().checked_sub(max_age) {
            Some(c) => c,
            None => return 0,
        };
        let mut removed = 0usize;
        self.edges.retain(|_, edge| {
            let is_stale = edge.last_updated < cutoff;
            if is_stale {
                removed += 1;
            }
            !is_stale
        });
        self.num_edges.fetch_sub(removed, Ordering::Relaxed);
        removed
    }

    /// Update capabilities for a node (from full capability fetch)
    pub fn update_node_capabilities(&self, node_id: &NodeId, caps: CapabilitySet) {
        if let Some(node) = self.nodes.get(node_id) {
            node.set_capabilities(caps);
            self.stats
                .capability_updates
                .fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Get node info
    pub fn get_node(&self, node_id: &NodeId) -> Option<ProximityNode> {
        self.nodes.get(node_id).map(|r| r.clone())
    }

    /// Get all nodes
    pub fn all_nodes(&self) -> Vec<ProximityNode> {
        self.nodes.iter().map(|r| r.value().clone()).collect()
    }

    /// Get nodes within hop distance
    pub fn nodes_within_hops(&self, max_hops: u8) -> Vec<ProximityNode> {
        self.nodes
            .iter()
            .filter(|r| r.hops <= max_hops)
            .map(|r| r.value().clone())
            .collect()
    }

    /// Find nodes matching a capability filter (quick check using primary caps)
    pub fn find_matching(&self, filter: &CapabilityFilter) -> Vec<ProximityNode> {
        self.nodes
            .iter()
            .filter(|r| r.is_available() && r.primary_caps.matches_basic(filter))
            .map(|r| r.value().clone())
            .collect()
    }

    /// Find best node for a capability filter
    pub fn find_best(&self, filter: &CapabilityFilter) -> Option<ProximityNode> {
        self.find_matching(filter).into_iter().min_by(|a, b| {
            a.routing_score(self.config.prefer_low_latency)
                .total_cmp(&b.routing_score(self.config.prefer_low_latency))
        })
    }

    /// Lowest-RTT node that satisfies `predicate`, returned as a
    /// [`Duration`]. Phase F slice 4 of `CAPABILITY_SYSTEM_PLAN.md`
    /// §7a — used by `StandardPlacement`'s scope-attraction scoring
    /// (slice 5) and by Phase E of `REDEX_DISTRIBUTED_PLAN.md`.
    ///
    /// Scans every node, picks the one with the lowest
    /// `latency_us` field where `predicate(node)` returns true.
    /// Returns `None` when no node matches the predicate.
    /// Available + non-available nodes are both considered — the
    /// caller's predicate decides which subset matters.
    ///
    /// Direct lookup variant: pass `|n| n.node_id == target` to
    /// fetch the RTT to a specific candidate (used by the
    /// placement tie-breaker).
    pub fn nearest_rtt(&self, predicate: impl Fn(&ProximityNode) -> bool) -> Option<Duration> {
        self.nodes
            .iter()
            .filter(|r| predicate(r.value()))
            .map(|r| r.latency_us)
            .min()
            .map(Duration::from_micros)
    }

    /// Find k best nodes for a capability filter
    pub fn find_k_best(&self, filter: &CapabilityFilter, k: usize) -> Vec<ProximityNode> {
        let mut matching = self.find_matching(filter);
        matching.sort_by(|a, b| {
            a.routing_score(self.config.prefer_low_latency)
                .total_cmp(&b.routing_score(self.config.prefer_low_latency))
        });
        matching.truncate(k);
        matching
    }

    /// Get shortest path to node (BFS).
    ///
    /// Uses a parent map to reconstruct the path once on arrival,
    /// avoiding the quadratic `path.clone()`-per-neighbor cost of the
    /// naive "queue of paths" BFS.
    pub fn path_to(&self, dest: &NodeId) -> Option<Vec<NodeId>> {
        self.bfs_path_to(dest, None)
    }

    /// Shortest path from self to `dest` that does NOT traverse the
    /// direct `(self, dest)` edge — a genuinely INDIRECT route through
    /// another neighbor. `None` means `dest` is reachable only
    /// directly (or not at all).
    ///
    /// When a direct link fails, its `(self → dest)` edge lingers in
    /// the graph until `sweep_stale_edges`, so an unrestricted
    /// [`Self::path_to`] returns `[self, dest]` — the now-dead direct
    /// hop — and masks any indirect alternate. Callers deciding
    /// whether a failed peer is still reachable through someone else
    /// must exclude that edge so the search is forced to route around
    /// it (RT-5 review: withdrawal suppression, cubic P1).
    pub fn path_to_excluding_direct(&self, dest: &NodeId) -> Option<Vec<NodeId>> {
        self.bfs_path_to(dest, Some((self.my_id, *dest)))
    }

    /// Shortest path from self to `dest` that does NOT start with the
    /// `self → excluded_first_hop` edge — an alternate through a
    /// DIFFERENT direct neighbor, even a longer one.
    ///
    /// When a peer withdraws its route toward `dest`, the withdrawing
    /// first hop is no longer usable, but the UNRESTRICTED shortest path
    /// may still start with it — and a caller that bails the moment
    /// `path[1] == withdrawing_peer` would ignore a perfectly good
    /// longer route through another neighbor and cascade a needless
    /// withdrawal (RT-5 review: alternate search considers only one
    /// shortest path). Excluding just the first-hop edge forces the BFS
    /// to leave through some other neighbor. `dest` reachable only
    /// through `excluded_first_hop` yields `None`.
    pub fn path_to_excluding_first_hop(
        &self,
        dest: &NodeId,
        excluded_first_hop: &NodeId,
    ) -> Option<Vec<NodeId>> {
        self.bfs_path_to(dest, Some((self.my_id, *excluded_first_hop)))
    }

    /// BFS shortest path from self to `dest`, optionally dropping one
    /// directed `(from, to)` edge from the adjacency so the search
    /// routes around it. Shared by [`Self::path_to`] (no exclusion)
    /// and [`Self::path_to_excluding_direct`].
    fn bfs_path_to(
        &self,
        dest: &NodeId,
        exclude_edge: Option<(NodeId, NodeId)>,
    ) -> Option<Vec<NodeId>> {
        if *dest == self.my_id {
            return Some(vec![self.my_id]);
        }

        // Build adjacency from edges, skipping the excluded edge.
        let mut adjacency: HashMap<NodeId, Vec<NodeId>> = HashMap::new();
        for edge in self.edges.iter() {
            if exclude_edge == Some((edge.from, edge.to)) {
                continue;
            }
            adjacency.entry(edge.from).or_default().push(edge.to);
        }

        // BFS with parent pointers
        let mut parent: HashMap<NodeId, NodeId> = HashMap::new();
        let mut visited: HashSet<NodeId> = HashSet::new();
        let mut queue: VecDeque<NodeId> = VecDeque::new();

        queue.push_back(self.my_id);
        visited.insert(self.my_id);

        while let Some(current) = queue.pop_front() {
            if current == *dest {
                // Walk back through the parent map to recover the path.
                let mut path = vec![current];
                let mut node = current;
                while node != self.my_id {
                    node = *parent.get(&node)?;
                    path.push(node);
                }
                path.reverse();
                return Some(path);
            }

            if let Some(neighbors) = adjacency.get(&current) {
                for &neighbor in neighbors {
                    if visited.insert(neighbor) {
                        parent.insert(neighbor, current);
                        queue.push_back(neighbor);
                    }
                }
            }
        }

        None
    }

    /// Create load balancer endpoints from proximity nodes
    pub fn to_endpoints(&self, filter: Option<&CapabilityFilter>) -> Vec<Endpoint> {
        self.nodes
            .iter()
            .filter(|r| {
                r.is_available()
                    && filter
                        .map(|f| r.primary_caps.matches_basic(f))
                        .unwrap_or(true)
            })
            .map(|r| {
                let node = r.value();
                // Weight inversely proportional to latency/hops
                let base_weight = 1000u32;
                let latency_penalty = (node.latency_us / 100) as u32; // 1 weight per 100us
                let weight = base_weight.saturating_sub(latency_penalty).max(1);

                Endpoint::new(node.node_id)
                    .with_weight(weight)
                    .with_priority(node.hops as u32)
            })
            .collect()
    }

    /// Update load balancer from proximity data
    pub fn update_load_balancer(&self, lb: &LoadBalancer, filter: Option<&CapabilityFilter>) {
        for entry in self.nodes.iter() {
            let node = entry.value();

            if !filter
                .map(|f| node.primary_caps.matches_basic(f))
                .unwrap_or(true)
            {
                continue;
            }

            // Update health
            lb.update_health(&node.node_id, node.health);

            // Update metrics
            let metrics = LoadMetrics {
                cpu_usage: (node.load_level as f64) / 255.0,
                avg_response_time_ms: (node.latency_us as f64) / 1000.0,
                ..Default::default()
            };
            lb.update_metrics(&node.node_id, metrics);
        }
    }

    /// Sync discovered nodes to capability index
    ///
    /// Note: This requires the caller to handle index updates appropriately.
    /// The CapabilityIndex uses announcements, so this returns nodes with capabilities
    /// that need to be announced.
    pub fn nodes_with_capabilities(&self) -> Vec<(NodeId, CapabilitySet)> {
        self.nodes
            .iter()
            .filter_map(|entry| {
                let node = entry.value();
                node.get_capabilities().map(|caps| (node.node_id, caps))
            })
            .collect()
    }

    /// Clean up stale entries
    pub fn cleanup(&self) -> CleanupStats {
        let mut removed_nodes = 0;
        let mut removed_pingwaves = 0;

        // Remove stale nodes
        self.nodes.retain(|_, node| {
            if node.is_stale(self.config.node_timeout) {
                removed_nodes += 1;
                self.stats.nodes_expired.fetch_add(1, Ordering::Relaxed);
                false
            } else {
                true
            }
        });

        // Remove old dedup entries
        self.seen_pingwaves.retain(|_, instant| {
            if instant.elapsed() > self.config.dedup_timeout {
                removed_pingwaves += 1;
                false
            } else {
                true
            }
        });

        // Keep the O(1) counters exact after eviction.
        self.num_nodes.fetch_sub(removed_nodes, Ordering::Relaxed);
        self.num_seen
            .fetch_sub(removed_pingwaves, Ordering::Relaxed);

        CleanupStats {
            removed_nodes,
            removed_pingwaves,
        }
    }

    /// Get statistics snapshot
    pub fn stats(&self) -> ProximityStatsSnapshot {
        ProximityStatsSnapshot {
            node_count: self.num_nodes.load(Ordering::Relaxed),
            edge_count: self.num_edges.load(Ordering::Relaxed),
            dedup_cache_size: self.num_seen.load(Ordering::Relaxed),
            pingwaves_sent: self.stats.pingwaves_sent.load(Ordering::Relaxed),
            pingwaves_received: self.stats.pingwaves_received.load(Ordering::Relaxed),
            pingwaves_forwarded: self.stats.pingwaves_forwarded.load(Ordering::Relaxed),
            pingwaves_dropped: self.stats.pingwaves_dropped.load(Ordering::Relaxed),
            nodes_discovered: self.stats.nodes_discovered.load(Ordering::Relaxed),
            nodes_expired: self.stats.nodes_expired.load(Ordering::Relaxed),
            capability_updates: self.stats.capability_updates.load(Ordering::Relaxed),
        }
    }

    /// Get node count
    pub fn node_count(&self) -> usize {
        self.num_nodes.load(Ordering::Relaxed)
    }
}

/// Cleanup statistics
#[derive(Debug, Clone, Default)]
pub struct CleanupStats {
    /// Number of expired nodes removed from the graph
    pub removed_nodes: usize,
    /// Number of stale ping wave deduplication entries removed
    pub removed_pingwaves: usize,
}

/// Statistics snapshot
#[derive(Debug, Clone, Default)]
pub struct ProximityStatsSnapshot {
    /// Number of nodes currently tracked in the proximity graph
    pub node_count: usize,
    /// Number of edges currently in the proximity graph
    pub edge_count: usize,
    /// Number of entries in the ping wave deduplication cache
    pub dedup_cache_size: usize,
    /// Total ping waves sent since startup
    pub pingwaves_sent: u64,
    /// Total ping waves received since startup
    pub pingwaves_received: u64,
    /// Total ping waves forwarded since startup
    pub pingwaves_forwarded: u64,
    /// Total ping waves dropped since startup
    pub pingwaves_dropped: u64,
    /// Total nodes discovered since startup
    pub nodes_discovered: u64,
    /// Total nodes expired since startup
    pub nodes_expired: u64,
    /// Total capability updates processed since startup
    pub capability_updates: u64,
}

/// Get current time in microseconds
fn current_time_us() -> u64 {
    use std::time::SystemTime;
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .map(|d| d.as_micros() as u64)
        .unwrap_or(0)
}

/// Hash capabilities for quick comparison
fn hash_capabilities(caps: &CapabilitySet) -> u64 {
    // Phase A.5.3: read through views() so the hash function
    // keeps working unchanged when typed-struct fields go away
    // (Phase A.5.N). Hash semantics intentionally unchanged —
    // capability hashes are persistent state in the proximity
    // graph; an accidental hash-shape change would invalidate
    // every node's cached neighbor metadata.
    let views = caps.views();
    let hw = views.hardware();

    // FNV-1a hash of key capability fields (shared helper; byte shape
    // intentionally unchanged — see the module note above).
    use super::hash::{fnv1a_step, FNV1A_OFFSET};
    let mut hash = FNV1A_OFFSET;
    hash = fnv1a_step(hash, hw.memory_gb as u64); // hardware memory
    hash = fnv1a_step(hash, if hw.gpu.is_some() { 1 } else { 0 }); // GPU presence
    hash = fnv1a_step(hash, hw.accelerators.len() as u64); // accelerator count
    hash = fnv1a_step(hash, views.tools().len() as u64); // tool count
    hash = fnv1a_step(hash, views.models().len() as u64); // model count
    hash = fnv1a_step(hash, caps.tags.len() as u64); // tag count
    hash
}

#[cfg(test)]
mod tests {
    use super::*;

    fn make_node_id(n: u8) -> NodeId {
        let mut id = [0u8; 32];
        id[0] = n;
        id
    }

    #[test]
    fn test_primary_capabilities_roundtrip() {
        let caps = PrimaryCapabilities {
            gpu: true,
            model_slots: 4,
            memory_tier: 5,
            tools_bitmap: 0b10101010,
            flags: 0x12345678,
        };

        let bytes = caps.to_bytes();
        let parsed = PrimaryCapabilities::from_bytes(&bytes);

        assert_eq!(caps, parsed);
    }

    #[test]
    fn test_enhanced_pingwave_roundtrip() {
        let pw = EnhancedPingwave::new(make_node_id(1), 42, 3)
            .with_capabilities(0xDEADBEEF, 5, PrimaryCapabilities::default())
            .with_load(128, HealthStatus::Healthy);

        let bytes = pw.to_bytes();
        let parsed = EnhancedPingwave::from_bytes(&bytes).unwrap();

        assert_eq!(pw.origin_id, parsed.origin_id);
        assert_eq!(pw.seq, parsed.seq);
        assert_eq!(pw.ttl, parsed.ttl);
        assert_eq!(pw.capability_hash, parsed.capability_hash);
        assert_eq!(pw.load_level, parsed.load_level);
    }

    /// Regression: BUG_REPORT.md #38 — `from_bytes` previously
    /// coerced any unknown discriminant on the `health` byte (63)
    /// into `HealthStatus::Unknown`. A single bit-flip in transit
    /// could downgrade a peer to `Unknown`, which
    /// `can_receive_traffic()` treats as unroutable — silent peer
    /// eviction. The fix returns `None` on unknown discriminants
    /// so the caller drops the malformed pingwave entirely.
    #[test]
    fn from_bytes_rejects_unknown_health_discriminant() {
        let pw = EnhancedPingwave::new(make_node_id(1), 1, 3).with_load(64, HealthStatus::Healthy);
        let mut bytes = pw.to_bytes().to_vec();

        // Sanity: round-trip works at the legitimate value.
        assert!(EnhancedPingwave::from_bytes(&bytes).is_some());

        // Mutate the health byte to an out-of-range discriminant.
        // 4..=255 are all unknown; sample a few across the range.
        for bad in [4u8, 99, 200, 255] {
            bytes[63] = bad;
            assert!(
                EnhancedPingwave::from_bytes(&bytes).is_none(),
                "health discriminant {} should be rejected, not coerced (#38)",
                bad
            );
        }

        // The four legitimate values still round-trip.
        for ok in 0u8..=3 {
            bytes[63] = ok;
            assert!(
                EnhancedPingwave::from_bytes(&bytes).is_some(),
                "health discriminant {} must still parse",
                ok
            );
        }
    }

    #[test]
    fn test_pingwave_forward() {
        let mut pw = EnhancedPingwave::new(make_node_id(1), 1, 3);
        assert_eq!(pw.ttl, 3);
        assert_eq!(pw.hop_count, 0);

        assert!(pw.forward());
        assert_eq!(pw.ttl, 2);
        assert_eq!(pw.hop_count, 1);

        assert!(pw.forward());
        assert!(pw.forward());
        assert_eq!(pw.ttl, 0);
        assert!(!pw.forward()); // Can't forward when expired
    }

    #[test]
    fn test_proximity_graph_pingwave_processing() {
        let my_id = make_node_id(1);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());

        let pw = EnhancedPingwave::new(make_node_id(2), 1, 3);
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        // Process pingwave
        let forwarded = graph.on_pingwave(pw, from);
        assert!(forwarded.is_some());

        // Node should be added
        let node = graph.get_node(&make_node_id(2)).unwrap();
        assert_eq!(node.hops, 1);

        // Duplicate should be dropped
        let pw2 = EnhancedPingwave::new(make_node_id(2), 1, 3);
        assert!(graph.on_pingwave(pw2, from).is_none());

        // New sequence should work
        let pw3 = EnhancedPingwave::new(make_node_id(2), 2, 3);
        assert!(graph.on_pingwave(pw3, from).is_some());
    }

    /// Regression for BUG_AUDIT_2026_04_30_CORE.md #108: pre-fix
    /// `from_pingwave` and `update_from_pingwave` used raw
    /// `pw.hop_count + 1`, which panics in debug at `u8::MAX`
    /// and silently wraps to 0 in release. A peer advertising
    /// `hop_count == 255` could either crash the receive loop
    /// or falsely promote itself to "0 hops" (directly
    /// connected) — a proximity-routing poisoning vector.
    /// Post-fix uses `saturating_add(1)` so 255 stays at 255.
    #[test]
    fn proximity_node_from_pingwave_saturates_at_max_hop_count() {
        let mut pw = EnhancedPingwave::new(make_node_id(2), 1, 3);
        pw.hop_count = u8::MAX;
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        // Pre-fix this would panic in debug builds and wrap to
        // 0 in release builds. Post-fix it saturates at 255.
        let node = ProximityNode::from_pingwave(&pw, from);
        assert_eq!(
            node.hops,
            u8::MAX,
            "saturating_add must clamp at u8::MAX, NOT wrap to 0"
        );
        assert_ne!(
            node.hops, 0,
            "a 255-hop peer must NOT be reported as 0 hops"
        );
    }

    #[test]
    fn proximity_node_update_from_pingwave_saturates_at_max_hop_count() {
        let mut pw_initial = EnhancedPingwave::new(make_node_id(2), 1, 3);
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();
        let mut node = ProximityNode::from_pingwave(&pw_initial, from);
        let initial_hops = node.hops;

        // Update with hop_count = 255. The saturating bump
        // (`pw.hop_count.saturating_add(1) = 255`) must not panic
        // in debug or wrap to 0 in release.
        pw_initial.hop_count = u8::MAX;
        pw_initial.seq = 2;
        node.update_from_pingwave(&pw_initial, from);
        // The path-quality arm rejects `new_hops=255 > self.hops`,
        // so the better cached hop count survives. Freshness still
        // advances `last_seq`. Sanity: no panic, no wrap.
        assert_eq!(node.hops, initial_hops);
        assert_eq!(node.last_seq, 2);
    }

    /// Regression for the "worse path overwrites better" hazard.
    /// Pre-fix `update_from_pingwave` used an OR predicate
    /// (`seq > last_seq || new_hops < hops`), so a flooded
    /// high-seq pingwave reaching us through a long route demoted
    /// a previously-cached direct route purely on freshness.
    /// Post-fix `last_seq` always advances on a newer pingwave,
    /// but `addr` / `hops` / `latency_us` only update when the
    /// new path is no worse.
    #[test]
    fn update_from_pingwave_keeps_better_path_when_newer_seq_arrives_via_longer_route() {
        // Direct path: 1 hop after the +1 bump.
        let mut pw_direct = EnhancedPingwave::new(make_node_id(2), 5, 0);
        let direct_addr: SocketAddr = "127.0.0.1:9000".parse().unwrap();
        let mut node = ProximityNode::from_pingwave(&pw_direct, direct_addr);
        let direct_hops = node.hops;
        let direct_last_seq = node.last_seq;
        assert_eq!(direct_hops, 1, "test setup: direct route is 1 hop");

        // A later, higher-seq pingwave for the same node arrives via
        // a 7-hop indirect path from a different source address.
        let indirect_addr: SocketAddr = "10.0.0.5:9000".parse().unwrap();
        pw_direct.seq = 9;
        pw_direct.hop_count = 7;
        node.update_from_pingwave(&pw_direct, indirect_addr);

        // Path-quality arm: the longer route MUST NOT overwrite the
        // direct route's address or hop count.
        assert_eq!(
            node.hops, direct_hops,
            "longer-route pingwave must not demote a better cached path",
        );
        assert_eq!(
            node.addr, direct_addr,
            "longer-route pingwave must not redirect to the indirect source",
        );
        // Freshness arm: `last_seq` still advances on the newer
        // pingwave so subsequent staleness / restart checks see the
        // current sequence number.
        assert!(
            node.last_seq > direct_last_seq,
            "freshness must still advance"
        );
        assert_eq!(node.last_seq, 9);
    }

    #[test]
    fn test_proximity_graph_find_matching() {
        let my_id = make_node_id(1);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());

        // Add some nodes via pingwaves
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        let mut pw1 = EnhancedPingwave::new(make_node_id(2), 1, 3);
        pw1.primary_caps = PrimaryCapabilities {
            gpu: true,
            model_slots: 4,
            ..Default::default()
        };
        graph.on_pingwave(pw1, from);

        let mut pw2 = EnhancedPingwave::new(make_node_id(3), 1, 3);
        pw2.primary_caps = PrimaryCapabilities {
            gpu: false,
            model_slots: 2,
            ..Default::default()
        };
        graph.on_pingwave(pw2, from);

        // Find GPU nodes
        let filter = CapabilityFilter {
            require_gpu: true,
            ..Default::default()
        };
        let gpu_nodes = graph.find_matching(&filter);
        assert_eq!(gpu_nodes.len(), 1);
        assert_eq!(gpu_nodes[0].node_id, make_node_id(2));
    }

    #[test]
    fn test_proximity_graph_to_endpoints() {
        let my_id = make_node_id(1);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());

        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        // Add nodes
        graph.on_pingwave(EnhancedPingwave::new(make_node_id(2), 1, 3), from);
        graph.on_pingwave(EnhancedPingwave::new(make_node_id(3), 1, 3), from);

        // Get endpoints
        let endpoints = graph.to_endpoints(None);
        assert_eq!(endpoints.len(), 2);
    }

    #[test]
    fn test_routing_score() {
        let pw = EnhancedPingwave::new(make_node_id(1), 1, 3);
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        let mut node = ProximityNode::from_pingwave(&pw, from);
        node.latency_us = 1000; // 1ms
        node.load_level = 128; // 50% load
        node.health = HealthStatus::Healthy;

        let score = node.routing_score(true);
        assert!(score > 0.0);

        // Degraded health should increase score
        node.health = HealthStatus::Degraded;
        let degraded_score = node.routing_score(true);
        assert!(degraded_score > score);
    }

    /// SEC-02 RED. A single registered sender floods novel
    /// `(origin_id, seq)` tuples; every configured cap must hold.
    ///
    /// Pre-fix `max_nodes` was declared and never read, and neither
    /// the dedup cache nor the edge map had a bound at all — so this
    /// loop grew all three without limit, for the process lifetime,
    /// on unauthenticated UDP admitted from a registered source
    /// address.
    #[test]
    fn pingwave_flood_cannot_grow_the_graph_past_its_caps() {
        let my_id = make_node_id(1);
        let config = ProximityConfig {
            max_nodes: 4,
            max_seen_pingwaves: 16,
            max_edges: 8,
            // Long timeouts: this test is about admission refusing to
            // grow, not about eviction reclaiming. A short timeout
            // would let cleanup mask a missing cap.
            node_timeout: Duration::from_secs(600),
            dedup_timeout: Duration::from_secs(600),
            ..Default::default()
        };
        let graph = ProximityGraph::new(my_id, config);
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();
        let sender = make_node_id(2);

        // 500 distinct origins, each with a distinct seq — the attack
        // shape: 256-bit random origins make accidental dedup
        // negligible, so every frame is novel. `make_node_id` only
        // varies one byte, so spread across two here to get 500
        // genuinely distinct ids.
        let wide_id = |n: u64| {
            let mut id = [0u8; 32];
            id[0] = (n & 0xff) as u8;
            id[1] = ((n >> 8) & 0xff) as u8;
            id[2] = 0xAA;
            id
        };
        let mut accepted = 0usize;
        let mut refused = 0usize;
        for i in 0..500u64 {
            let pw = EnhancedPingwave::new(wide_id(i), i, 3);
            match graph.admit_pingwave_from(pw, sender, from) {
                PingwaveAdmission::RejectedCapacity => refused += 1,
                _ => accepted += 1,
            }
        }

        let stats = graph.stats();
        assert!(
            stats.node_count <= 4,
            "node_count {} exceeded max_nodes 4 — `max_nodes` is declared but not enforced",
            stats.node_count
        );
        assert!(
            stats.dedup_cache_size <= 16,
            "seen_pingwaves {} exceeded max_seen_pingwaves 16",
            stats.dedup_cache_size
        );
        assert!(
            stats.edge_count <= 8,
            "edges {} exceeded max_edges 8",
            stats.edge_count
        );
        assert!(
            refused > 0,
            "a 500-origin flood against caps of 4/16/8 refused nothing"
        );
        assert!(
            accepted > 0,
            "the caps refused everything, including the frames that fit"
        );
    }

    /// The caps must not cost liveness for peers already known: a
    /// saturated graph still tracks the nodes it holds. A fix that
    /// froze the graph entirely would satisfy the RED above while
    /// breaking proximity routing under load.
    ///
    /// This one saturates the *node* cap; `max_seen_pingwaves` is left
    /// deliberately slack so the assertion is about that cap alone. The
    /// dedup cache has its own version of this property and its own
    /// test below — it needed one, because the flat check it originally
    /// carried failed it.
    #[test]
    fn a_saturated_graph_still_updates_the_nodes_it_already_knows() {
        let my_id = make_node_id(1);
        let config = ProximityConfig {
            max_nodes: 2,
            max_seen_pingwaves: 1_000,
            node_timeout: Duration::from_secs(600),
            dedup_timeout: Duration::from_secs(600),
            ..Default::default()
        };
        let graph = ProximityGraph::new(my_id, config);
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();
        let sender = make_node_id(2);
        let known = make_node_id(42);

        // Learn `known`, then saturate with junk.
        graph.admit_pingwave_from(EnhancedPingwave::new(known, 1, 3), sender, from);
        for i in 0..100u64 {
            let mut junk = [0u8; 32];
            junk[0] = (i & 0xff) as u8;
            junk[1] = 0xBB;
            graph.admit_pingwave_from(EnhancedPingwave::new(junk, i, 3), sender, from);
        }
        assert!(graph.stats().node_count <= 2);

        // The known peer's next pingwave must still be admitted and
        // still refresh its entry — it is competing with the flood for
        // slots it already occupies, and must not lose.
        let admitted = graph.admit_pingwave_from(EnhancedPingwave::new(known, 2, 3), sender, from);
        assert!(
            !matches!(admitted, PingwaveAdmission::RejectedCapacity),
            "a known peer was refused because attacker traffic filled the graph: {admitted:?}"
        );
        assert!(
            graph.get_node(&known).is_some(),
            "the known peer was evicted by the flood"
        );
    }

    /// The same property for the *dedup* cache, which is where it was
    /// actually missing.
    ///
    /// The `max_seen_pingwaves` check was a flat one taken ahead of
    /// everything else, so a full cache refused every novel
    /// `(origin_id, seq)` — and a known peer's next pingwave always
    /// carries a fresh `seq`. Filling the cache therefore suppressed
    /// the legitimate topology as well as the flood, until the next
    /// `cleanup()`. With `dedup_timeout` at 10 s against a 60 s sweep
    /// cadence that is most of a minute of dead discovery for a few MB
    /// of UDP, repeatable after every sweep — and long enough for the
    /// real peers to idle past `node_timeout` and be evicted, so the
    /// flood destroys topology rather than merely failing to add any.
    ///
    /// `max_nodes` is left slack here for the mirror-image reason the
    /// test above leaves `max_seen_pingwaves` slack.
    #[test]
    fn a_saturated_dedup_cache_still_admits_a_known_peers_next_pingwave() {
        let my_id = make_node_id(1);
        let config = ProximityConfig {
            max_nodes: 1_000,
            max_seen_pingwaves: 64,
            max_edges: 1_000,
            // Long timeouts: this is about admission reserving room,
            // not about eviction reclaiming it. A short dedup_timeout
            // would let expiry mask a missing reserve.
            node_timeout: Duration::from_secs(600),
            dedup_timeout: Duration::from_secs(600),
            ..Default::default()
        };
        let graph = ProximityGraph::new(my_id, config);
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();
        let sender = make_node_id(2);
        let known = make_node_id(42);

        // The known peer is in the graph first — the shape of a mesh
        // that was healthy before the flood arrived.
        graph.admit_pingwave_from(EnhancedPingwave::new(known, 1, 3), sender, from);
        assert!(graph.get_node(&known).is_some(), "setup: peer not learned");

        // Flood novel origins until the cache stops taking them.
        for i in 0..500u64 {
            let mut junk = [0u8; 32];
            junk[0] = (i & 0xff) as u8;
            junk[1] = ((i >> 8) & 0xff) as u8;
            junk[2] = 0xBB;
            graph.admit_pingwave_from(EnhancedPingwave::new(junk, i, 3), sender, from);
        }
        let saturated = graph.stats().dedup_cache_size;
        assert!(
            saturated <= 64,
            "dedup cache {saturated} exceeded max_seen_pingwaves 64"
        );
        assert!(
            saturated >= 16,
            "the flood barely filled the cache ({saturated}); this test needs a \
             genuinely saturated one to prove anything"
        );

        // Every subsequent pingwave from the known peer is novel — a
        // fresh seq each time — and every one must still be admitted.
        for seq in 2..12u64 {
            let admitted =
                graph.admit_pingwave_from(EnhancedPingwave::new(known, seq, 3), sender, from);
            assert!(
                !matches!(admitted, PingwaveAdmission::RejectedCapacity),
                "a known peer's pingwave (seq {seq}) was refused because a flood of \
                 novel origins filled the dedup cache: {admitted:?}"
            );
        }
        assert!(
            graph.get_node(&known).is_some(),
            "the known peer went missing while the cache was saturated"
        );

        // The reserve is headroom, not a second ceiling: the absolute
        // cap still binds, including the peers it is reserved for.
        assert!(
            graph.stats().dedup_cache_size <= 64,
            "the reserve let the dedup cache past max_seen_pingwaves — the memory \
             bound the cap exists for is gone"
        );

        // And novel origins are still refused, or the reserve would
        // just be a bigger cap.
        let mut fresh = [0u8; 32];
        fresh[0] = 0xEE;
        fresh[1] = 0xCC;
        assert!(
            matches!(
                graph.admit_pingwave_from(EnhancedPingwave::new(fresh, 1, 3), sender, from),
                PingwaveAdmission::RejectedCapacity
            ),
            "a novel origin reached the reserved headroom"
        );
    }

    /// The reserve's arithmetic, including the degenerate configs a
    /// `max / 4` split gets wrong if written carelessly.
    #[test]
    fn the_dedup_reserve_leaves_room_for_both_sides() {
        let graph_with = |max_seen: usize| {
            ProximityGraph::new(
                make_node_id(1),
                ProximityConfig {
                    max_seen_pingwaves: max_seen,
                    ..Default::default()
                },
            )
        };

        // Ordinary sizing: a quarter held back, three quarters open.
        assert_eq!(graph_with(40_000).unreserved_dedup_capacity(), 30_000);
        assert_eq!(graph_with(64).unreserved_dedup_capacity(), 48);

        // A cap of zero admits nothing, and must not underflow.
        assert_eq!(graph_with(0).unreserved_dedup_capacity(), 0);

        // Tiny caps must still admit a novel origin. `1 - 1/4 == 1`
        // already, but 2 and 3 are where an unclamped `max * 3 / 4`
        // would be fine and a `max - max/4` on a rounding-down
        // division would not — pin the whole small range.
        for max in 1..=8usize {
            let unreserved = graph_with(max).unreserved_dedup_capacity();
            assert!(
                unreserved >= 1,
                "max_seen_pingwaves {max} reserved the entire cache, so no novel \
                 origin is ever admitted"
            );
            assert!(
                unreserved <= max,
                "max_seen_pingwaves {max} gave novel origins {unreserved} slots, \
                 more than the cache holds"
            );
        }
    }

    #[test]
    fn test_cleanup() {
        let my_id = make_node_id(1);
        let config = ProximityConfig {
            node_timeout: Duration::from_millis(10),
            dedup_timeout: Duration::from_millis(10),
            ..Default::default()
        };
        let graph = ProximityGraph::new(my_id, config);

        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();
        graph.on_pingwave(EnhancedPingwave::new(make_node_id(2), 1, 3), from);

        assert_eq!(graph.node_count(), 1);

        // Wait for timeout
        std::thread::sleep(Duration::from_millis(20));

        let stats = graph.cleanup();
        assert_eq!(stats.removed_nodes, 1);
        assert_eq!(graph.node_count(), 0);
    }

    /// The O(1) `num_nodes` / `num_edges` / `num_seen` counters backing
    /// `node_count()` / `stats()` must stay exactly in step with the
    /// underlying `DashMap` lengths across inserts, duplicates, edge
    /// sweeps, and node/dedup cleanup.
    #[test]
    fn proximity_entry_counters_track_map_lengths() {
        let config = ProximityConfig {
            node_timeout: Duration::from_millis(5),
            dedup_timeout: Duration::from_millis(5),
            ..Default::default()
        };
        let graph = ProximityGraph::new(make_node_id(1), config);
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        for origin in 2u8..6 {
            graph.on_pingwave(EnhancedPingwave::new(make_node_id(origin), 1, 3), from);
        }
        let s = graph.stats();
        assert_eq!(s.node_count, graph.nodes.len());
        assert_eq!(s.edge_count, graph.edges.len());
        assert_eq!(s.dedup_cache_size, graph.seen_pingwaves.len());

        // Duplicate (origin, seq) must not double-count.
        graph.on_pingwave(EnhancedPingwave::new(make_node_id(2), 1, 3), from);
        assert_eq!(graph.stats().dedup_cache_size, graph.seen_pingwaves.len());
        assert_eq!(graph.node_count(), graph.nodes.len());

        // Edge sweep must keep edge_count in step.
        std::thread::sleep(Duration::from_millis(10));
        graph.sweep_stale_edges(Duration::from_millis(5));
        assert_eq!(graph.stats().edge_count, graph.edges.len());

        // Node/dedup cleanup must keep their counters in step.
        graph.cleanup();
        let s = graph.stats();
        assert_eq!(s.node_count, graph.nodes.len());
        assert_eq!(s.dedup_cache_size, graph.seen_pingwaves.len());
    }

    #[test]
    fn test_regression_pingwave_primary_caps_survive_roundtrip() {
        // Regression: EnhancedPingwave::to_bytes/from_bytes did not
        // serialize primary_caps (gpu, model_slots, etc.), so after
        // crossing the wire all capabilities were reset to defaults.
        // This made capability-based routing silently fail for remote
        // nodes — e.g., `require_gpu: true` never matched anyone.
        let caps = PrimaryCapabilities {
            gpu: true,
            model_slots: 4,
            memory_tier: 5,
            tools_bitmap: 0b10101010,
            flags: 0x12345678,
        };
        let pw = EnhancedPingwave::new(make_node_id(1), 42, 3).with_capabilities(0xDEAD, 7, caps);

        let bytes = pw.to_bytes();
        let parsed = EnhancedPingwave::from_bytes(&bytes).unwrap();

        assert!(
            parsed.primary_caps.gpu,
            "gpu capability must survive serialization"
        );
        assert_eq!(parsed.primary_caps.model_slots, 4);
        assert_eq!(parsed.primary_caps.memory_tier, 5);
        assert_eq!(parsed.primary_caps.tools_bitmap, 0b10101010);
        assert_eq!(parsed.primary_caps.flags, 0x12345678);
    }

    #[test]
    fn test_regression_find_best_no_panic_on_nan() {
        // Regression: find_best() used partial_cmp().unwrap() which
        // panics on NaN routing scores. Now uses total_cmp().
        let my_id = make_node_id(1);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());

        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        // Add nodes with very high latency (edge case for routing_score)
        let pw = EnhancedPingwave::new(make_node_id(2), 1, 3).with_load(0, HealthStatus::Healthy);
        graph.on_pingwave(pw, from);

        let filter = CapabilityFilter::default();
        // This should not panic
        let _best = graph.find_best(&filter);
        let _k_best = graph.find_k_best(&filter, 5);
    }

    #[test]
    fn test_regression_hop_count_saturates() {
        // Regression: forward() used `hop_count += 1` which wraps at
        // u8::MAX (255 → 0), making a distant node appear 1 hop away.
        // Now uses saturating_add.
        let mut pw = EnhancedPingwave::new(make_node_id(1), 1, 255);
        pw.hop_count = 254;

        assert!(pw.forward());
        assert_eq!(pw.hop_count, 255);

        // At 255, saturating_add should keep it at 255
        assert!(pw.forward());
        assert_eq!(
            pw.hop_count, 255,
            "hop_count should saturate at 255, not wrap to 0"
        );
    }

    #[test]
    fn test_edge_insert_on_pingwave_receipt() {
        // On pingwave receipt for origin Y via peer Z, two edges
        // materialize: (self → Z) and (Z → Y). `path_to(Y)` then
        // returns the 3-step path [self, Z, Y].
        let my_id = make_node_id(1);
        let z = make_node_id(2);
        let y = make_node_id(3);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        // Pingwave carries origin Y, arrived via Z (hop_count=1).
        let pw = EnhancedPingwave::new(y, 1, 3).with_load(0, HealthStatus::Healthy);
        let mut pw = pw;
        pw.hop_count = 1;
        graph.on_pingwave_from(pw, z, from);

        let path = graph.path_to(&y).expect("path_to(y) should return Some");
        assert_eq!(path, vec![my_id, z, y]);
    }

    /// RT-5 review witness: a byte-identical replay of an already-seen
    /// pingwave must be rejected by dedup BEFORE it can re-add the edge
    /// (or, in the receive path, reinstall the route) that a withdrawal
    /// removed. Admission is checked ahead of every node/edge mutation.
    #[test]
    fn duplicate_pingwave_is_rejected_before_resurrecting_an_edge() {
        let my_id = make_node_id(1);
        let z = make_node_id(2); // forwarding direct peer
        let y = make_node_id(3); // origin
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        let from: SocketAddr = "127.0.0.1:9100".parse().unwrap();

        // Accept pingwave P (origin Y via Z). Byte-identical replay
        // means identical (origin, seq) — capture the exact bytes and
        // re-parse so the replay is provably the same frame.
        let mut p = EnhancedPingwave::new(y, 7, 3).with_load(0, HealthStatus::Healthy);
        p.hop_count = 1;
        let p_bytes = p.to_bytes();

        match graph.admit_pingwave_from(p, z, from) {
            PingwaveAdmission::AcceptedAndForward(_) => {}
            other => panic!("first receipt should be accepted+forwarded, got {other:?}"),
        }
        assert!(
            graph.edge_latency(z, y).is_some(),
            "accepting P installs the Z→Y edge",
        );

        // Withdraw Y: the withdrawal path removes the Z→Y edge (and, in
        // the mesh, the route through Z).
        assert!(graph.remove_edge(z, y), "edge should have existed");
        assert!(graph.edge_latency(z, y).is_none());

        // Replay the byte-identical P.
        let replay = EnhancedPingwave::from_bytes(&p_bytes).expect("re-parse P");
        match graph.admit_pingwave_from(replay, z, from) {
            PingwaveAdmission::RejectedDuplicate => {}
            other => panic!("byte-identical replay must be a duplicate, got {other:?}"),
        }
        assert!(
            graph.edge_latency(z, y).is_none(),
            "the removed edge must stay absent — a duplicate must not resurrect it",
        );
    }

    #[test]
    fn path_to_excluding_direct_routes_around_the_direct_edge() {
        // A dest reachable BOTH directly (self→Y) and indirectly
        // (self→Z→Y). The unrestricted search takes the direct edge;
        // excluding it forces the indirect route (cubic P1: the failed
        // link's direct edge must not mask a live alternate).
        let my_id = make_node_id(1);
        let z = make_node_id(2);
        let y = make_node_id(3);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        let from_y: SocketAddr = "127.0.0.1:9001".parse().unwrap();
        let from_z: SocketAddr = "127.0.0.1:9002".parse().unwrap();

        // Direct edge self→Y (pingwave straight from Y).
        graph.on_pingwave_from(EnhancedPingwave::new(y, 1, 3), y, from_y);
        // Indirect self→Z→Y (pingwave for Y forwarded via Z).
        let mut pw = EnhancedPingwave::new(y, 2, 3);
        pw.hop_count = 1;
        graph.on_pingwave_from(pw, z, from_z);

        assert_eq!(
            graph.path_to(&y),
            Some(vec![my_id, y]),
            "unrestricted search takes the shortest (direct) edge",
        );
        assert_eq!(
            graph.path_to_excluding_direct(&y),
            Some(vec![my_id, z, y]),
            "excluding the direct edge forces the indirect route",
        );
    }

    /// RT-5 review P2: when the SHORTEST path to a dest starts with the
    /// withdrawing peer, the alternate search must still find a LONGER
    /// valid route through a different first hop instead of giving up.
    #[test]
    fn path_to_excluding_first_hop_finds_the_longer_route_through_another_peer() {
        // Short:  self→B→X→D   (first hop B, the withdrawing peer)
        // Long:   self→C→Y→Z→D (first hop C)
        let my_id = make_node_id(1);
        let b = make_node_id(2);
        let x = make_node_id(3);
        let c = make_node_id(4);
        let y = make_node_id(5);
        let z = make_node_id(6);
        let d = make_node_id(7);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());

        // Build both paths by hand via the edge test-seam.
        graph.test_insert_edge(my_id, b, 100);
        graph.test_insert_edge(b, x, 100);
        graph.test_insert_edge(x, d, 100);
        graph.test_insert_edge(my_id, c, 100);
        graph.test_insert_edge(c, y, 100);
        graph.test_insert_edge(y, z, 100);
        graph.test_insert_edge(z, d, 100);

        assert_eq!(
            graph.path_to(&d),
            Some(vec![my_id, b, x, d]),
            "unrestricted search takes the shorter path via B",
        );
        assert_eq!(
            graph.path_to_excluding_first_hop(&d, &b),
            Some(vec![my_id, c, y, z, d]),
            "excluding B as a first hop must find the longer route via C, not cascade",
        );
    }

    #[test]
    fn path_to_excluding_first_hop_none_when_only_via_excluded_peer() {
        let my_id = make_node_id(1);
        let b = make_node_id(2);
        let d = make_node_id(7);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        graph.test_insert_edge(my_id, b, 100);
        graph.test_insert_edge(b, d, 100);

        assert_eq!(graph.path_to(&d), Some(vec![my_id, b, d]));
        assert_eq!(
            graph.path_to_excluding_first_hop(&d, &b),
            None,
            "a dest reachable only through the excluded first hop has no alternate",
        );
    }

    #[test]
    fn path_to_excluding_direct_none_when_only_direct() {
        let my_id = make_node_id(1);
        let y = make_node_id(3);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        let from_y: SocketAddr = "127.0.0.1:9001".parse().unwrap();
        graph.on_pingwave_from(EnhancedPingwave::new(y, 1, 3), y, from_y);

        assert_eq!(graph.path_to(&y), Some(vec![my_id, y]));
        assert_eq!(
            graph.path_to_excluding_direct(&y),
            None,
            "a dest reachable only directly has no indirect alternate",
        );
    }

    #[test]
    fn test_edge_sweep_removes_stale() {
        use std::time::Duration;
        let my_id = make_node_id(1);
        let z = make_node_id(2);
        let y = make_node_id(3);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        let mut pw = EnhancedPingwave::new(y, 1, 3).with_load(0, HealthStatus::Healthy);
        pw.hop_count = 1;
        graph.on_pingwave_from(pw, z, from);
        assert!(graph.path_to(&y).is_some());

        // Backdate both edges so the sweep finds them stale.
        // `checked_sub` avoids the overflow panic that fires on
        // hosts with system uptime < the subtracted duration
        // (Windows Instant is bounded by boot). Pair a short
        // backdate (200ms) with a tighter sweep threshold (50ms)
        // so the same invariant — "stale edges get swept" —
        // holds without depending on hour-scale uptime.
        let stale_ts = Instant::now()
            .checked_sub(Duration::from_millis(200))
            .expect("test host uptime should exceed 200ms");
        for mut entry in graph.edges.iter_mut() {
            entry.last_updated = stale_ts;
        }
        let removed = graph.sweep_stale_edges(Duration::from_millis(50));
        assert_eq!(removed, 2, "both synthetic edges should be swept");
        assert!(graph.path_to(&y).is_none());
    }

    #[test]
    fn test_origin_self_check_drops_pingwave() {
        // Pingwave claiming `origin == self_id` must be dropped.
        // The graph's `on_pingwave` already has this check; the same
        // rule is enforced in `mesh.rs` dispatch earlier.
        let my_id = make_node_id(1);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        let pw = EnhancedPingwave::new(my_id, 1, 3);
        let forwarded = graph.on_pingwave(pw, from);
        assert!(forwarded.is_none(), "self-origin pingwave must be dropped");
        assert!(graph.get_node(&my_id).is_none());
    }

    #[test]
    fn test_latency_ewma_smooths_successive_samples() {
        let my_id = make_node_id(1);
        let z = make_node_id(2);
        let y = make_node_id(3);
        let graph = ProximityGraph::new(my_id, ProximityConfig::default());
        let from: SocketAddr = "127.0.0.1:9000".parse().unwrap();

        // Two pingwaves with known timestamps → two latency samples.
        let now = current_time_us();
        let mut pw1 = EnhancedPingwave::new(y, 1, 3).with_load(0, HealthStatus::Healthy);
        pw1.hop_count = 1;
        pw1.origin_timestamp_us = now.saturating_sub(10_000); // 10 ms ago
        graph.on_pingwave_from(pw1, z, from);

        // Edge should have latency ≈ 10_000 us after first insert.
        let edge1 = graph.edges.get(&(z, y)).expect("z→y edge");
        assert!(edge1.latency_us > 0);
        let first = edge1.latency_us;
        drop(edge1);

        // Second sample with a different latency — EWMA drags it.
        let mut pw2 = EnhancedPingwave::new(y, 2, 3).with_load(0, HealthStatus::Healthy);
        pw2.hop_count = 1;
        pw2.origin_timestamp_us = current_time_us().saturating_sub(50_000); // 50 ms ago
        graph.on_pingwave_from(pw2, z, from);

        let edge2 = graph.edges.get(&(z, y)).unwrap();
        // EWMA α=1/8 → edge.latency_us moved toward 50_000 from
        // `first`, but not all the way.
        assert_ne!(edge2.latency_us, first, "EWMA should shift latency");
        assert!(
            edge2.latency_us < 50_000,
            "EWMA should not snap to the new sample"
        );
    }
}