saorsa_core/

network.rs

// Copyright 2024 Saorsa Labs Limited
//
// This software is licensed under the MIT license <LICENSE-MIT or
// https://opensource.org/licenses/MIT> or the Apache License, Version 2.0
// <LICENSE-APACHE or https://www.apache.org/licenses/LICENSE-2.0>, at your
// option. This file may not be copied, modified, or distributed except
// according to those terms.
//
// Unless required by applicable law or agreed to in writing, software
// distributed under these licenses is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

//! Network module
//!
//! This module provides core networking functionality for the P2P Foundation.
//! It handles peer connections, network events, and node lifecycle management.

use crate::PeerId;
use crate::adaptive::trust::{TrustRecord, TrustSnapshot};
use crate::adaptive::{AdaptiveDHT, AdaptiveDhtConfig, TrustEngine, TrustEvent};
use crate::bootstrap::cache::{CachedCloseGroupPeer, CloseGroupCache};
use crate::bootstrap::routing_snapshot::{RoutingSnapshot, SnapshotPeer};
use crate::dht::core_engine::AddressType;
use crate::dht_network_manager::{DhtNetworkConfig, DhtNetworkEvent, DhtNetworkManager};
use crate::error::{NetworkError, P2PError, P2pResult as Result};
use crate::reachability::spawn_acquisition_driver;

use crate::MultiAddr;
use crate::identity::node_identity::{NodeIdentity, peer_id_from_public_key};
use crate::quantum_crypto::saorsa_transport_integration::{MlDsaPublicKey, MlDsaSignature};
use bytes::Bytes;
use dashmap::DashMap;
use futures::StreamExt;
use parking_lot::Mutex as ParkingMutex;
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::{Mutex as TokioMutex, RwLock, broadcast};
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, trace, warn};

fn bootstrap_peer_identity_matches(expected: Option<PeerId>, actual: PeerId) -> bool {
    expected.is_none_or(|expected| expected == actual)
}

/// Wire protocol message format for P2P communication.
///
/// Serialized with postcard for compact binary encoding.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct WireMessage {
    /// Protocol/topic identifier
    pub(crate) protocol: String,
    /// Raw payload bytes
    pub(crate) data: Vec<u8>,
    /// Sender's peer ID (verified against transport-level identity)
    pub(crate) from: PeerId,
    /// Unix timestamp in seconds
    pub(crate) timestamp: u64,
    /// User agent string identifying the sender's software.
    ///
    /// Convention: `"node/<version>"` for full DHT participants,
    /// `"client/<version>"` or `"<app>/<version>"` for ephemeral clients.
    /// Included in the signed bytes — tamper-proof.
    #[serde(default)]
    pub(crate) user_agent: String,
    /// Sender's ML-DSA-65 public key (1952 bytes). Empty if unsigned.
    #[serde(default)]
    pub(crate) public_key: Vec<u8>,
    /// ML-DSA-65 signature over the signable bytes. Empty if unsigned.
    #[serde(default)]
    pub(crate) signature: Vec<u8>,
}

/// Borrowing view of a [`WireMessage`] for serialization.
///
/// Serializes to exactly the bytes [`WireMessage`] would produce for the
/// same content (same field order and postcard encoding), so a sender can
/// frame a payload it only borrows without first copying it into an owned
/// message. Receivers keep decoding into [`WireMessage`].
#[derive(Serialize)]
pub(crate) struct WireMessageRef<'a> {
    pub(crate) protocol: &'a str,
    pub(crate) data: &'a [u8],
    pub(crate) from: &'a PeerId,
    pub(crate) timestamp: u64,
    pub(crate) user_agent: &'a str,
    pub(crate) public_key: &'a [u8],
    pub(crate) signature: &'a [u8],
}

/// Operating mode of a P2P node.
///
/// Determines the default user agent and DHT participation behavior.
/// `Node` peers participate in the DHT routing table; `Client` peers
/// are treated as ephemeral and excluded from routing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum NodeMode {
    /// Full DHT-participant node that maintains routing state and routes messages.
    #[default]
    Node,
    /// Ephemeral client that connects to perform operations without joining the DHT.
    Client,
}

/// Internal listen mode controlling which network interfaces the node binds to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ListenMode {
    /// Bind to all interfaces (`0.0.0.0` / `::`).
    Public,
    /// Bind to loopback only (`127.0.0.1` / `::1`).
    Local,
}

/// Returns the default software identifier for the given mode.
///
/// - `Node` → `"node/<saorsa-core-version>"`
/// - `Client` → `"client/<saorsa-core-version>"`
pub fn user_agent_for_mode(mode: NodeMode) -> String {
    let prefix = match mode {
        NodeMode::Node => "node",
        NodeMode::Client => "client",
    };
    format!("{prefix}/{}", env!("CARGO_PKG_VERSION"))
}

/// Returns `true` if the user agent identifies a full DHT participant (prefix `"node/"`).
pub fn is_dht_participant(user_agent: &str) -> bool {
    user_agent.starts_with("node/")
}

/// Capacity of the internal channel used by the message receiving system.
pub(crate) const MESSAGE_RECV_CHANNEL_CAPACITY: usize = 256;

/// Maximum number of concurrent in-flight request/response operations.
pub(crate) const MAX_ACTIVE_REQUESTS: usize = 256;

/// Maximum allowed timeout for a single request (5 minutes).
pub(crate) const MAX_REQUEST_TIMEOUT: Duration = Duration::from_secs(300);

/// Default listen port for the P2P node.
const DEFAULT_LISTEN_PORT: u16 = 9000;

/// Default maximum number of concurrent connections.
const DEFAULT_MAX_CONNECTIONS: usize = 10_000;

/// Default connection timeout in seconds.
///
/// The transport adapter keeps each direct Happy Eyeballs attempt short so
/// DHT lookups can move past offline peers quickly. 25s leaves room for
/// multi-stage connection strategies and identity exchange while preserving
/// the historical API default.
const DEFAULT_CONNECTION_TIMEOUT_SECS: u64 = 25;

/// Default maximum age of a close-group cache snapshot before it is skipped
/// as Priority-0 bootstrap material.
const DEFAULT_CLOSE_GROUP_CACHE_MAX_AGE_SECS: u64 = 60 * 60;

/// Lower bound for periodic close-group-cache saves. Prevents a very short DHT
/// refresh interval from turning cache persistence into a hot write loop.
const MIN_CLOSE_GROUP_CACHE_SAVE_INTERVAL: Duration = Duration::from_secs(60);

/// Timeout in seconds for waiting on a bootstrap peer's identity exchange.
///
/// Tighter than the post-bootstrap budget
/// ([`crate::dht_network_manager::IDENTITY_EXCHANGE_TIMEOUT`],
/// 5 s) on purpose: bootstrap candidates are unverified and a stuck one
/// must not be allowed to head-of-line block convergence. 3 s covers
/// loopback (<100 ms) and direct WAN paths (~1–2 s with one handshake
/// retry); a relay-tunnelled path with congested ML-DSA verification
/// can exceed this and will fail identity exchange, but bootstrap simply
/// moves on to other candidates rather than retrying the same one.
///
/// `wait_for_peer_identity` short-circuits on channel close, so most dead
/// channels surface in microseconds regardless of this budget.
const BOOTSTRAP_IDENTITY_TIMEOUT_SECS: u64 = 3;

/// Maximum number of bootstrap peers dialed concurrently in Phase B.
///
/// Bounds the fan-out of configured bootstrap dials so simultaneous QUIC+PQC
/// handshakes don't spike CPU or saturate the UDP socket. Chosen
/// low on purpose: each dial runs a full ML-KEM key exchange and ML-DSA
/// verification, and a cold-start node has no spare compute budget.
const MAX_CONCURRENT_BOOTSTRAP_DIALS: usize = 4;

/// Number of successful bootstrap connections after which a client-mode
/// node stops dialing further candidates.
///
/// Clients only need enough peers to route their own lookups (α=3 parallel
/// queries → 6 gives ~2× redundancy) and don't serve the DHT, so a fully
/// populated close-group buys them nothing. Stopping early cuts cold-start
/// latency by skipping the tail of slow / dead candidates. Nodes always
/// dial every candidate so their routing table converges fully.
const CLIENT_BOOTSTRAP_TARGET: usize = 6;

/// Maximum routing-snapshot peers dialled concurrently.
///
/// Higher than [`MAX_CONCURRENT_BOOTSTRAP_DIALS`] because this set is an order
/// of magnitude larger — a full table rather than a handful of seeds — and the
/// whole value of restoring it is that the table is usable in seconds. Still
/// bounded, so a restart cannot open an unbounded number of simultaneous
/// QUIC+PQC handshakes.
const MAX_CONCURRENT_SNAPSHOT_DIALS: usize = 16;

/// How long the restored peer count holds authority over saving.
///
/// It exists to stop a node that is stopped mid-restore from writing the
/// fragment it has re-dialled so far over a complete snapshot. Once the table
/// has had this long to converge it is the node's best knowledge, and a smaller
/// table is a smaller network rather than an unfinished restore, so the floor
/// stops applying and the file keeps being refreshed.
const SNAPSHOT_FLOOR_ENFORCED_FOR: Duration = Duration::from_secs(60 * 60);

/// The floor has not been decided yet: bootstrap has not reached the restore
/// step, or this is a client, which never restores and never writes one.
const SNAPSHOT_FLOOR_UNRESOLVED: usize = usize::MAX;

/// Addresses tried per snapshot peer.
///
/// The budget below stops new peers, so the phase's real bound is the budget
/// plus the last peer's attempts. Two keeps that tail short while still
/// covering a peer whose first address has gone stale.
const MAX_SNAPSHOT_ADDRESSES_DIALLED: usize = 2;

/// Wall-clock budget for the routing-snapshot dial phase.
///
/// Startup must not be held hostage to a snapshot full of departed peers. When
/// the budget expires the remaining candidates are abandoned and the table
/// refills through ordinary discovery, which is exactly the behaviour of a node
/// that had no snapshot at all.
const SNAPSHOT_RESTORE_BUDGET: Duration = Duration::from_secs(20);

/// Serde helper — returns `true`.
const fn default_true() -> bool {
    true
}

/// Configuration for a P2P node
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeConfig {
    /// Bind to loopback only (`127.0.0.1` / `::1`).
    ///
    /// When `true`, the node listens on loopback addresses suitable for
    /// local development and testing. When `false` (the default), the node
    /// listens on all interfaces (`0.0.0.0` / `::`).
    #[serde(default)]
    pub local: bool,

    /// Listen port. `0` means OS-assigned ephemeral port.
    #[serde(default)]
    pub port: u16,

    /// Enable IPv6 dual-stack binding.
    ///
    /// When `true` (the default), both an IPv4 and an IPv6 address are
    /// bound. When `false`, only IPv4 is used.
    #[serde(default = "default_true")]
    pub ipv6: bool,

    /// Bootstrap peers to connect to on startup.
    pub bootstrap_peers: Vec<crate::MultiAddr>,

    // MCP removed; will be redesigned later
    /// Connection timeout duration
    pub connection_timeout: Duration,

    /// Maximum number of concurrent connections
    pub max_connections: usize,

    /// DHT configuration
    pub dht_config: DHTConfig,

    /// Optional IP diversity configuration for Sybil protection tuning.
    ///
    /// When set, this configuration is used by diversity-enforcing subsystems.
    /// If `None`, defaults are used.
    pub diversity_config: Option<crate::security::IPDiversityConfig>,

    /// Optional override for the maximum application-layer message size.
    ///
    /// When `None`, the underlying saorsa-transport default is used.
    #[serde(default)]
    pub max_message_size: Option<usize>,

    /// Optional node identity for app-level message signing.
    ///
    /// When set, outgoing messages are signed with the node's ML-DSA-65 key
    /// and incoming signed messages are verified at the transport layer.
    #[serde(skip)]
    pub node_identity: Option<Arc<NodeIdentity>>,

    /// Operating mode of this node.
    ///
    /// Determines the default user agent and DHT participation:
    /// - `Node` → user agent `"node/<version>"`, added to DHT routing tables.
    /// - `Client` → user agent `"client/<version>"`, treated as ephemeral.
    #[serde(default)]
    pub mode: NodeMode,

    /// Optional custom user agent override.
    ///
    /// When `Some`, this value is used instead of the mode-derived default.
    /// When `None`, the user agent is derived from [`NodeConfig::mode`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub custom_user_agent: Option<String>,

    /// Allow loopback addresses (127.0.0.1, ::1) in the transport layer.
    ///
    /// In production, loopback addresses are rejected because they are not
    /// routable. Enable this for local devnets and testnets where all nodes
    /// run on the same machine.
    ///
    /// Default: `false`
    #[serde(default)]
    pub allow_loopback: bool,

    /// Adaptive DHT configuration (trust-based swap-out).
    ///
    /// Controls whether peers with low trust scores are eligible for
    /// swap-out from the routing table when better candidates arrive. Use
    /// `NodeConfigBuilder::trust_enforcement` for a simple on/off toggle.
    ///
    /// Default: enabled with a swap threshold of 0.35.
    #[serde(default)]
    pub adaptive_dht_config: AdaptiveDhtConfig,

    /// Optional path for persisting the close group cache.
    ///
    /// Directory for persisting the close group cache.
    ///
    /// When set, the node saves its close group peers and their trust
    /// scores to `{dir}/close_group_cache.json` periodically, on shutdown,
    /// and after bootstrap. On startup, fresh cached peers are loaded and
    /// contacted first, preserving close group consistency across restarts.
    ///
    /// When `None`, no close group cache is used.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub close_group_cache_dir: Option<PathBuf>,

    /// Maximum age for using a close-group cache snapshot as Priority-0
    /// bootstrap material. Older snapshots are logged and skipped so the node
    /// falls through to configured bootstrap peers. `None` disables the age
    /// check. Default: one hour.
    #[serde(default = "default_close_group_cache_max_age")]
    pub close_group_cache_max_age: Option<Duration>,
}

fn default_close_group_cache_max_age() -> Option<Duration> {
    Some(Duration::from_secs(DEFAULT_CLOSE_GROUP_CACHE_MAX_AGE_SECS))
}

/// DHT-specific configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DHTConfig {
    /// Kademlia K parameter (bucket size)
    pub k_value: usize,

    /// Kademlia alpha parameter (parallelism)
    pub alpha_value: usize,

    /// DHT refresh interval
    pub refresh_interval: Duration,
}

// ============================================================================
// Address Construction Helpers
// ============================================================================

/// Build QUIC listen addresses based on port, IPv6 preference, and listen mode.
///
/// All returned addresses use the QUIC transport — the only transport
/// currently supported for dialing. When additional transports are added,
/// extend this function to produce addresses for those transports as well.
///
/// `ListenMode::Public` uses unspecified (all-interface) addresses;
/// `ListenMode::Local` uses loopback addresses.
#[inline]
fn build_listen_addrs(port: u16, ipv6_enabled: bool, mode: ListenMode) -> Vec<MultiAddr> {
    let mut addrs = Vec::with_capacity(if ipv6_enabled { 2 } else { 1 });

    let (v4, v6) = match mode {
        ListenMode::Public => (
            std::net::Ipv4Addr::UNSPECIFIED,
            std::net::Ipv6Addr::UNSPECIFIED,
        ),
        ListenMode::Local => (std::net::Ipv4Addr::LOCALHOST, std::net::Ipv6Addr::LOCALHOST),
    };

    if ipv6_enabled {
        addrs.push(MultiAddr::quic(std::net::SocketAddr::new(
            std::net::IpAddr::V6(v6),
            port,
        )));
    }

    addrs.push(MultiAddr::quic(std::net::SocketAddr::new(
        std::net::IpAddr::V4(v4),
        port,
    )));

    addrs
}

impl NodeConfig {
    /// Returns the effective user agent string.
    ///
    /// If a custom user agent was set, returns that. Otherwise, derives
    /// the user agent from the node's [`NodeMode`].
    pub fn user_agent(&self) -> String {
        self.custom_user_agent
            .clone()
            .unwrap_or_else(|| user_agent_for_mode(self.mode))
    }

    /// Compute the listen addresses from the configuration fields.
    ///
    /// The returned addresses are derived from [`local`](Self::local),
    /// [`port`](Self::port), and [`ipv6`](Self::ipv6).
    pub fn listen_addrs(&self) -> Vec<MultiAddr> {
        let mode = if self.local {
            ListenMode::Local
        } else {
            ListenMode::Public
        };
        build_listen_addrs(self.port, self.ipv6, mode)
    }

    /// Create a new NodeConfig with default values
    ///
    /// # Errors
    ///
    /// Returns an error if default addresses cannot be parsed
    pub fn new() -> Result<Self> {
        Ok(Self::default())
    }

    /// Create a builder for customized NodeConfig construction
    pub fn builder() -> NodeConfigBuilder {
        NodeConfigBuilder::default()
    }
}

// ============================================================================
// NodeConfig Builder Pattern
// ============================================================================

/// Builder for constructing [`NodeConfig`] with a transport-aware fluent API.
///
/// Defaults are chosen for quick local development:
/// - QUIC on a random free port (`0`)
/// - IPv6 enabled (dual-stack)
/// - All interfaces (not local-only)
///
/// # Examples
///
/// ```rust,ignore
/// // Simplest — QUIC on random port, IPv6 on, all interfaces
/// let config = NodeConfig::builder().build()?;
///
/// // Local dev/test mode (loopback, auto-enables allow_loopback)
/// let config = NodeConfig::builder()
///     .local(true)
///     .build()?;
/// ```
#[derive(Debug, Clone)]
pub struct NodeConfigBuilder {
    port: u16,
    ipv6: bool,
    local: bool,
    bootstrap_peers: Vec<crate::MultiAddr>,
    max_connections: Option<usize>,
    connection_timeout: Option<Duration>,
    dht_config: Option<DHTConfig>,
    max_message_size: Option<usize>,
    mode: NodeMode,
    custom_user_agent: Option<String>,
    allow_loopback: Option<bool>,
    adaptive_dht_config: Option<AdaptiveDhtConfig>,
    close_group_cache_dir: Option<PathBuf>,
    /// Outer `None` means the builder setter was not called; inner `None`
    /// explicitly disables age enforcement.
    close_group_cache_max_age: Option<Option<Duration>>,
}

impl Default for NodeConfigBuilder {
    fn default() -> Self {
        Self {
            port: 0,
            ipv6: true,
            local: false,
            bootstrap_peers: Vec::new(),
            max_connections: None,
            connection_timeout: None,
            dht_config: None,
            max_message_size: None,
            mode: NodeMode::default(),
            custom_user_agent: None,
            allow_loopback: None,
            adaptive_dht_config: None,
            close_group_cache_dir: None,
            close_group_cache_max_age: None,
        }
    }
}

impl NodeConfigBuilder {
    /// Set the listen port. Default: `0` (random free port).
    pub fn port(mut self, port: u16) -> Self {
        self.port = port;
        self
    }

    /// Enable or disable IPv6 dual-stack. Default: `true`.
    pub fn ipv6(mut self, enabled: bool) -> Self {
        self.ipv6 = enabled;
        self
    }

    /// Bind to loopback only (`true`) or all interfaces (`false`).
    ///
    /// When `true`, automatically enables `allow_loopback` unless explicitly
    /// overridden via [`Self::allow_loopback`].
    ///
    /// Default: `false` (all interfaces).
    pub fn local(mut self, local: bool) -> Self {
        self.local = local;
        self
    }

    /// Add a bootstrap peer.
    pub fn bootstrap_peer(mut self, addr: crate::MultiAddr) -> Self {
        self.bootstrap_peers.push(addr);
        self
    }

    /// Set maximum connections.
    pub fn max_connections(mut self, max: usize) -> Self {
        self.max_connections = Some(max);
        self
    }

    /// Set connection timeout.
    pub fn connection_timeout(mut self, timeout: Duration) -> Self {
        self.connection_timeout = Some(timeout);
        self
    }

    /// Set DHT configuration.
    pub fn dht_config(mut self, config: DHTConfig) -> Self {
        self.dht_config = Some(config);
        self
    }

    /// Set maximum application-layer message size in bytes.
    ///
    /// If this method is not called, saorsa-transport's built-in default is used.
    pub fn max_message_size(mut self, max_message_size: usize) -> Self {
        self.max_message_size = Some(max_message_size);
        self
    }

    /// Set the operating mode (Node or Client).
    pub fn mode(mut self, mode: NodeMode) -> Self {
        self.mode = mode;
        self
    }

    /// Set a custom user agent string, overriding the mode-derived default.
    pub fn custom_user_agent(mut self, user_agent: impl Into<String>) -> Self {
        self.custom_user_agent = Some(user_agent.into());
        self
    }

    /// Explicitly control whether loopback addresses are allowed in the
    /// transport layer. When not called, `local(true)` auto-enables this;
    /// `local(false)` defaults to `false`.
    pub fn allow_loopback(mut self, allow: bool) -> Self {
        self.allow_loopback = Some(allow);
        self
    }

    /// Enable or disable trust-based peer swap-out.
    ///
    /// When `false`, peers are never swapped out of the routing table
    /// based on trust scores. Trust scores are still tracked but have
    /// no enforcement effect.
    ///
    /// When `true` (the default), peers whose trust score falls below the
    /// swap threshold (0.35) become eligible for replacement when a
    /// better candidate arrives.
    ///
    /// For fine-grained control over the threshold, use
    /// [`adaptive_dht_config`](Self::adaptive_dht_config) instead.
    pub fn trust_enforcement(mut self, enabled: bool) -> Self {
        let threshold = if enabled {
            AdaptiveDhtConfig::default().swap_threshold
        } else {
            0.0
        };
        self.adaptive_dht_config = Some(AdaptiveDhtConfig {
            swap_threshold: threshold,
        });
        self
    }

    /// Set the full adaptive DHT configuration.
    ///
    /// Overrides any previous call to [`trust_enforcement`](Self::trust_enforcement).
    pub fn adaptive_dht_config(mut self, config: AdaptiveDhtConfig) -> Self {
        self.adaptive_dht_config = Some(config);
        self
    }

    /// Set the directory for persisting the close group cache.
    ///
    /// The node writes `close_group_cache.json` inside this directory on
    /// shutdown and after bootstrap, and loads it on startup.
    pub fn close_group_cache_dir(mut self, path: impl Into<PathBuf>) -> Self {
        self.close_group_cache_dir = Some(path.into());
        self
    }

    /// Set the maximum age for using a close-group cache as Priority-0
    /// bootstrap material. `None` disables the age check.
    pub fn close_group_cache_max_age(mut self, max_age: Option<Duration>) -> Self {
        self.close_group_cache_max_age = Some(max_age);
        self
    }

    /// Build the [`NodeConfig`].
    ///
    /// # Errors
    ///
    /// Returns an error if address construction fails.
    pub fn build(self) -> Result<NodeConfig> {
        // local mode auto-enables allow_loopback unless explicitly overridden
        let allow_loopback = self.allow_loopback.unwrap_or(self.local);

        Ok(NodeConfig {
            local: self.local,
            port: self.port,
            ipv6: self.ipv6,
            bootstrap_peers: self.bootstrap_peers,
            connection_timeout: self
                .connection_timeout
                .unwrap_or(Duration::from_secs(DEFAULT_CONNECTION_TIMEOUT_SECS)),
            max_connections: self.max_connections.unwrap_or(DEFAULT_MAX_CONNECTIONS),
            dht_config: self.dht_config.unwrap_or_default(),
            diversity_config: None,
            max_message_size: self.max_message_size,
            node_identity: None,
            mode: self.mode,
            custom_user_agent: self.custom_user_agent,
            allow_loopback,
            adaptive_dht_config: self.adaptive_dht_config.unwrap_or_default(),
            close_group_cache_dir: self.close_group_cache_dir,
            close_group_cache_max_age: self
                .close_group_cache_max_age
                .unwrap_or_else(default_close_group_cache_max_age),
        })
    }
}

impl Default for NodeConfig {
    fn default() -> Self {
        Self {
            local: false,
            port: DEFAULT_LISTEN_PORT,
            ipv6: true,
            bootstrap_peers: Vec::new(),
            connection_timeout: Duration::from_secs(DEFAULT_CONNECTION_TIMEOUT_SECS),
            max_connections: DEFAULT_MAX_CONNECTIONS,
            dht_config: DHTConfig::default(),
            diversity_config: None,
            max_message_size: None,
            node_identity: None,
            mode: NodeMode::default(),
            custom_user_agent: None,
            allow_loopback: false,
            adaptive_dht_config: AdaptiveDhtConfig::default(),
            close_group_cache_dir: None,
            close_group_cache_max_age: default_close_group_cache_max_age(),
        }
    }
}

impl DHTConfig {
    /// Default K value (bucket size) for Kademlia routing.
    pub const DEFAULT_K_VALUE: usize = crate::dht_lookup::DEFAULT_K_VALUE;
    const DEFAULT_ALPHA_VALUE: usize = crate::dht_lookup::DEFAULT_ALPHA_VALUE;
    const DEFAULT_REFRESH_INTERVAL_SECS: u64 = 600;
    /// Minimum k_value — values below this produce degenerate routing behavior.
    const MIN_K_VALUE: usize = 4;

    /// Validate parameter safety constraints (Section 4 points 1-13).
    ///
    /// Returns `Err` if any constraint is violated.
    pub fn validate(&self) -> Result<()> {
        if self.k_value < Self::MIN_K_VALUE {
            return Err(P2PError::Validation(
                format!(
                    "k_value must be >= {} (got {}), values below {} produce degenerate behavior",
                    Self::MIN_K_VALUE,
                    self.k_value,
                    Self::MIN_K_VALUE,
                )
                .into(),
            ));
        }
        if self.alpha_value < 1 {
            return Err(P2PError::Validation(
                format!("alpha_value must be >= 1 (got {})", self.alpha_value).into(),
            ));
        }
        if self.refresh_interval.is_zero() {
            return Err(P2PError::Validation("refresh_interval must be > 0".into()));
        }
        Ok(())
    }
}

impl Default for DHTConfig {
    fn default() -> Self {
        Self {
            k_value: Self::DEFAULT_K_VALUE,
            alpha_value: Self::DEFAULT_ALPHA_VALUE,
            refresh_interval: Duration::from_secs(Self::DEFAULT_REFRESH_INTERVAL_SECS),
        }
    }
}

/// Information about a connected peer
#[derive(Debug, Clone)]
pub struct PeerInfo {
    /// Transport-level channel identifier (internal use only).
    #[allow(dead_code)]
    pub(crate) channel_id: String,

    /// Peer's addresses
    pub addresses: Vec<MultiAddr>,

    /// Connection timestamp
    pub connected_at: Instant,

    /// Last seen timestamp
    pub last_seen: Instant,

    /// Connection status
    pub status: ConnectionStatus,

    /// Supported protocols
    pub protocols: Vec<String>,

    /// Number of heartbeats received
    pub heartbeat_count: u64,
}

/// Connection status for a peer
#[derive(Debug, Clone, PartialEq)]
pub enum ConnectionStatus {
    /// Connection is being established
    Connecting,
    /// Connection is established and active
    Connected,
    /// Connection is being closed
    Disconnecting,
    /// Connection is closed
    Disconnected,
    /// Connection failed
    Failed(String),
}

/// Coarse classification of the route by which a peer's traffic reached us.
///
/// Derived by matching the transport source address of an incoming message
/// against the typed dial addresses the DHT phonebook holds for that peer.
/// See [`P2PNode::classify_peer_transport_route`].
///
/// This is a **provenance label** for logging and metrics. It is not an
/// identity signal and must not be used for trust scoring or routing
/// decisions.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PeerRouteKind {
    /// Traffic arrived via the peer's MASQUE relay address.
    Relay,
    /// Traffic arrived via the peer's direct public IP address.
    Direct,
    /// Traffic arrived via a LAN or other local-scope address.
    Lan,
    /// Traffic arrived via an address whose reachability is unverified.
    Unverified,
    /// The transport source is absent, non-IP, or does not match any address
    /// the local phonebook holds for the peer.
    Unknown,
}

impl PeerRouteKind {
    /// Canonical lowercase label for the variant, used by the [`Display`]
    /// implementation and suitable for log fields / metric tags.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Relay => "relay",
            Self::Direct => "direct",
            Self::Lan => "lan",
            Self::Unverified => "unverified",
            Self::Unknown => "unknown",
        }
    }
}

impl std::fmt::Display for PeerRouteKind {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

impl From<AddressType> for PeerRouteKind {
    fn from(ty: AddressType) -> Self {
        match ty {
            AddressType::Relay => Self::Relay,
            AddressType::Direct => Self::Direct,
            AddressType::Unverified => Self::Unverified,
            AddressType::Lan => Self::Lan,
        }
    }
}

/// Pure classification of a transport source against a typed address slice.
///
/// This is the testable, I/O-free core of
/// [`P2PNode::classify_peer_transport_route`]; it touches no node state and
/// can be exercised without constructing a node.
///
/// # Classification rules
/// - `transport_source` `None` or not an IP socket address →
///   [`PeerRouteKind::Unknown`].
/// - The transport source's socket address is compared by exact equality
///   against each `(MultiAddr, AddressType)` entry's socket address
///   ([`MultiAddr::socket_addr`]).
/// - A single matching entry yields that entry's [`AddressType`] mapped to
///   [`PeerRouteKind`].
/// - Multiple matching entries that **agree** on the [`AddressType`] yield
///   that type (deduplicated duplicates).
/// - Multiple matching entries with **conflicting** [`AddressType`]s yield
///   [`PeerRouteKind::Unknown`]. We deliberately do **not** use
///   [`AddressType::priority`] to pick a winner here: that ordering governs
///   *dial preference*, not the provenance of an already-arrived message,
///   and silently promoting one type over another would mask a phonebook
///   inconsistency. `Unknown` is the conservative, honest label.
/// - No match → [`PeerRouteKind::Unknown`].
///
/// Address-list order is never used as a classification signal.
#[must_use]
pub(crate) fn classify_peer_route_from_typed(
    transport_source: Option<&MultiAddr>,
    typed_addrs: &[(MultiAddr, AddressType)],
) -> PeerRouteKind {
    let Some(src) = transport_source else {
        return PeerRouteKind::Unknown;
    };
    let Some(src_sa) = src.socket_addr() else {
        return PeerRouteKind::Unknown;
    };

    // Walk all entries; remember the first matched type and flag a conflict
    // if a later match disagrees. We do not short-circuit on first match
    // because duplicates with conflicting types must surface as Unknown.
    let mut matched: Option<AddressType> = None;
    let mut conflict = false;
    for (addr, ty) in typed_addrs {
        if addr.socket_addr().is_some_and(|sa| sa == src_sa) {
            match matched {
                None => matched = Some(*ty),
                Some(prev) if prev == *ty => {}
                Some(_) => {
                    conflict = true;
                    break;
                }
            }
        }
    }

    match (matched, conflict) {
        (Some(ty), false) => ty.into(),
        _ => PeerRouteKind::Unknown,
    }
}

/// Network events that can occur in the P2P system
///
/// Events are broadcast to all listeners and provide real-time
/// notifications of network state changes and message arrivals.
#[derive(Debug, Clone)]
pub enum P2PEvent {
    /// Message received from a peer on a specific topic
    Message {
        /// Topic or channel the message was sent on
        topic: String,
        /// For signed messages this is the authenticated app-level [`PeerId`];
        /// `None` for unsigned messages.
        source: Option<PeerId>,
        /// IP transport address that delivered this message, when known.
        ///
        /// This is provenance metadata, not an identity signal.
        transport_source: Option<MultiAddr>,
        /// Sender-supplied Unix timestamp in seconds.
        ///
        /// For signed messages this value is covered by the ML-DSA-65 signature
        /// alongside the payload, so handlers can use it for application-level
        /// freshness or replay defense. Wire-level acceptance no longer gates
        /// on this value; subscribers MUST do their own age/dedup checks when
        /// the protocol requires them.
        timestamp: u64,
        /// Raw message data payload
        data: Vec<u8>,
    },
    /// An authenticated peer has connected (first signed message verified on any channel).
    /// The `user_agent` identifies the remote software (e.g. `"node/0.12.1"`, `"client/1.0"`).
    PeerConnected(PeerId, String),
    /// An authenticated peer has fully disconnected (all channels closed).
    PeerDisconnected(PeerId),
}

/// Response from a peer to a request sent via [`P2PNode::send_request`].
///
/// Contains the response payload along with metadata about the responder
/// and round-trip latency.
#[derive(Debug, Clone)]
pub struct PeerResponse {
    /// The peer that sent the response.
    pub peer_id: PeerId,
    /// Raw response payload bytes.
    pub data: Vec<u8>,
    /// Round-trip latency from request to response.
    pub latency: Duration,
}

/// Wire format for request/response correlation.
///
/// Wraps application payloads with a message ID and direction flag
/// so the receive loop can route responses back to waiting callers.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct RequestResponseEnvelope {
    /// Unique identifier to correlate request ↔ response.
    pub(crate) message_id: String,
    /// `false` for requests, `true` for responses.
    pub(crate) is_response: bool,
    /// Application payload.
    pub(crate) payload: Vec<u8>,
}

/// An in-flight request awaiting a response from a specific peer.
pub(crate) struct PendingRequest {
    /// Oneshot sender for delivering the response payload.
    pub(crate) response_tx: tokio::sync::oneshot::Sender<Vec<u8>>,
    /// The peer we expect the response from (for origin validation).
    pub(crate) expected_peer: PeerId,
}

/// Short grace period after closing stale QUIC connections before re-dialing.
///
/// `disconnect_channel` is async and waits for the QUIC close, but the
/// transport endpoint may need a moment to fully release internal state.
/// Only applied when stale channels were actually disconnected.
const QUIC_TEARDOWN_GRACE: Duration = Duration::from_millis(100);

/// Main P2P network node that manages connections, routing, and communication
///
/// This struct represents a complete P2P network participant that can:
/// - Connect to other peers via QUIC transport
/// - Participate in distributed hash table (DHT) operations
/// - Send and receive messages through various protocols
/// - Handle network events and peer lifecycle
///
/// Transport concerns (connections, messaging, events) are delegated to
/// `TransportHandle`.
pub struct P2PNode {
    /// Node configuration
    config: NodeConfig,

    /// Our peer ID
    peer_id: PeerId,

    /// Transport handle owning all QUIC / peer / event state
    transport: Arc<crate::transport_handle::TransportHandle>,

    /// Node start time
    start_time: Instant,

    /// Shutdown token — cancelled when the node should stop
    shutdown: CancellationToken,

    /// Dedicated cancellation token for periodic close-group-cache saves.
    /// Cancelled and joined before the authoritative shutdown snapshot.
    close_group_cache_save_shutdown: CancellationToken,
    /// Whether the routing snapshot has already seeded a bootstrap in this
    /// process, so a re-bootstrap does not replay it.
    routing_snapshot_restored: AtomicBool,
    /// How many peers the routing snapshot restored at startup, zero when there
    /// was none to restore, and [`SNAPSHOT_FLOOR_UNRESOLVED`] until bootstrap
    /// has decided. A save must not write a table smaller than this while the
    /// floor still applies. Shared with the periodic save task.
    routing_snapshot_floor: Arc<AtomicUsize>,
    /// When that floor was decided, in epoch seconds. Zero while unresolved.
    routing_snapshot_floor_at: Arc<AtomicU64>,

    /// Periodic close-group-cache task, retained so shutdown can prevent a
    /// late periodic write from replacing the final snapshot.
    close_group_cache_save_handle: TokioMutex<Option<tokio::task::JoinHandle<()>>>,

    /// Adaptive DHT layer — owns both the DHT manager and the trust engine.
    /// All DHT operations and trust signals go through this component.
    adaptive_dht: AdaptiveDHT,

    /// Bootstrap state tracking - indicates whether peer discovery has completed
    is_bootstrapped: Arc<AtomicBool>,

    /// Whether `start()` has been called (and `stop()` has not yet completed)
    is_started: Arc<AtomicBool>,

    /// Per-peer locks that serialise reconnect attempts so concurrent sends
    /// to the same stale peer don't race to dial.  Entries accumulate over
    /// the node's lifetime; each is a lightweight `Arc<TokioMutex<()>>`.
    reconnect_locks: ParkingMutex<HashMap<PeerId, Arc<TokioMutex<()>>>>,

    /// The peer ID of the node currently relaying traffic for us (ADR-014).
    ///
    /// Set after the reachability classifier acquires a relay in `start()`.
    /// The relayer monitor watches this against the K-closest set: if the
    /// relayer drops out, it triggers rebinding.
    ///
    /// `None` when the node is publicly reachable (no relay needed) or
    /// before classification has run.
    relayer_peer_id: Arc<RwLock<Option<PeerId>>>,

    /// The relay-allocated public address (ADR-014).
    ///
    /// Set after a proactive MASQUE relay is acquired in `start()`. This is
    /// the address that external peers must dial to reach this node through
    /// the relay. `None` when the node is publicly reachable (no relay) or
    /// before classification has run.
    relay_address: Arc<RwLock<Option<SocketAddr>>>,
}

/// Normalize wildcard bind addresses to localhost loopback addresses
///
/// saorsa-transport correctly rejects "unspecified" addresses (0.0.0.0 and [::]) for remote connections
/// because you cannot connect TO an unspecified address - these are only valid for BINDING.
///
/// This function converts wildcard addresses to appropriate loopback addresses for local connections:
/// - IPv6 [::]:port → ::1:port (IPv6 loopback)
/// - IPv4 0.0.0.0:port → 127.0.0.1:port (IPv4 loopback)
/// - All other addresses pass through unchanged
///
/// # Arguments
/// * `addr` - The SocketAddr to normalize
///
/// # Returns
/// * Normalized SocketAddr suitable for remote connections
pub(crate) fn normalize_wildcard_to_loopback(addr: std::net::SocketAddr) -> std::net::SocketAddr {
    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};

    if addr.ip().is_unspecified() {
        // Convert unspecified addresses to loopback
        let loopback_ip = match addr {
            std::net::SocketAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST), // ::1
            std::net::SocketAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST), // 127.0.0.1
        };
        std::net::SocketAddr::new(loopback_ip, addr.port())
    } else {
        // Not a wildcard address, pass through unchanged
        addr
    }
}

impl P2PNode {
    /// Create a new P2P node with the given configuration
    pub async fn new(config: NodeConfig) -> Result<Self> {
        // Ensure a cryptographic identity exists — generate one if not provided.
        let node_identity = match config.node_identity.clone() {
            Some(identity) => identity,
            None => Arc::new(NodeIdentity::generate()?),
        };

        // Derive the canonical peer ID from the cryptographic identity.
        let peer_id = *node_identity.peer_id();

        // Validate parameter safety constraints (Section 4 points 1-13).
        // Reject invalid config early, before any resources are allocated.
        config.dht_config.validate()?;
        if let Some(ref diversity) = config.diversity_config {
            diversity
                .validate()
                .map_err(|e| P2PError::Validation(format!("IP diversity config: {e}").into()))?;
        }

        // Build transport handle with all transport-level concerns
        let transport_config = crate::transport_handle::TransportConfig::from_node_config(
            &config,
            crate::DEFAULT_EVENT_CHANNEL_CAPACITY,
            node_identity.clone(),
        );
        let transport =
            Arc::new(crate::transport_handle::TransportHandle::new(transport_config).await?);

        // Initialize AdaptiveDHT — creates the trust engine and DHT manager
        let dht_manager_config = DhtNetworkConfig {
            peer_id,
            node_config: config.clone(),
            request_timeout: config.connection_timeout,
            max_concurrent_operations: MAX_ACTIVE_REQUESTS,
            enable_security: true,
            swap_threshold: 0.0, // Set by AdaptiveDHT::new() from AdaptiveDhtConfig
        };
        let adaptive_dht = AdaptiveDHT::new(
            transport.clone(),
            dht_manager_config,
            config.adaptive_dht_config.clone(),
        )
        .await?;

        let node = Self {
            config,
            peer_id,
            transport,
            start_time: Instant::now(),
            shutdown: CancellationToken::new(),
            close_group_cache_save_shutdown: CancellationToken::new(),
            routing_snapshot_restored: AtomicBool::new(false),
            routing_snapshot_floor: Arc::new(AtomicUsize::new(SNAPSHOT_FLOOR_UNRESOLVED)),
            routing_snapshot_floor_at: Arc::new(AtomicU64::new(0)),
            close_group_cache_save_handle: TokioMutex::new(None),
            adaptive_dht,
            is_bootstrapped: Arc::new(AtomicBool::new(false)),
            is_started: Arc::new(AtomicBool::new(false)),
            reconnect_locks: ParkingMutex::new(HashMap::new()),
            relayer_peer_id: Arc::new(RwLock::new(None)),
            relay_address: Arc::new(RwLock::new(None)),
        };
        info!(
            "Created P2P node with peer ID: {} (call start() to begin networking)",
            node.peer_id
        );

        Ok(node)
    }

    /// Get the peer ID of this node.
    pub fn peer_id(&self) -> &PeerId {
        &self.peer_id
    }

    /// Get the transport handle for sharing with other components.
    pub fn transport(&self) -> &Arc<crate::transport_handle::TransportHandle> {
        &self.transport
    }

    /// The relay-allocated public address, if this node acquired a MASQUE relay.
    ///
    /// Returns `Some(addr)` when the node is behind NAT and successfully
    /// acquired a proactive relay during `start()`. External peers must dial
    /// this address to reach the node through the relay. Returns `None` when
    /// the node is publicly reachable or no relay was established.
    pub async fn relay_address(&self) -> Option<SocketAddr> {
        *self.relay_address.read().await
    }

    pub fn local_addr(&self) -> Option<MultiAddr> {
        self.transport.local_addr()
    }

    /// Check if the node has completed the initial bootstrap process
    ///
    /// Returns `true` if the node has successfully connected to at least one
    /// bootstrap peer and performed peer discovery (FIND_NODE).
    pub fn is_bootstrapped(&self) -> bool {
        self.is_bootstrapped.load(Ordering::SeqCst)
    }

    /// Manually trigger re-bootstrap (useful for recovery or network rejoin)
    ///
    /// This clears the bootstrapped state and attempts to reconnect to
    /// bootstrap peers and discover new peers.
    pub async fn re_bootstrap(&self) -> Result<()> {
        self.is_bootstrapped.store(false, Ordering::SeqCst);
        self.connect_bootstrap_peers(None).await
    }

    // =========================================================================
    // Trust API — delegates to AdaptiveDHT
    // =========================================================================

    /// Get the trust engine for advanced use cases
    pub fn trust_engine(&self) -> Arc<TrustEngine> {
        self.adaptive_dht.trust_engine().clone()
    }

    /// Report a trust event for a peer.
    ///
    /// Core only records penalties (connection failures). Positive trust
    /// signals are the consumer's responsibility via [`TrustEvent::ApplicationSuccess`].
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// use saorsa_core::adaptive::TrustEvent;
    ///
    /// node.report_trust_event(&peer_id, TrustEvent::ApplicationSuccess(1.0)).await;
    /// node.report_trust_event(&peer_id, TrustEvent::ConnectionFailed).await;
    /// ```
    pub async fn report_trust_event(&self, peer_id: &PeerId, event: TrustEvent) {
        self.adaptive_dht.report_trust_event(peer_id, event).await;
    }

    /// Get the current trust score for a peer (0.0 to 1.0).
    ///
    /// Returns 0.5 (neutral) for unknown peers.
    pub fn peer_trust(&self, peer_id: &PeerId) -> f64 {
        self.adaptive_dht.peer_trust(peer_id)
    }

    /// Get the AdaptiveDHT component for direct access
    pub fn adaptive_dht(&self) -> &AdaptiveDHT {
        &self.adaptive_dht
    }

    /// Classify the route by which a peer's traffic reached this node.
    ///
    /// Looks up the typed dial addresses the DHT phonebook holds for `peer_id`
    /// and matches `transport_source` against them using the internal pure
    /// classifier. `transport_source` must be the address observed by the
    /// authenticated transport while delivering the message (currently a
    /// QUIC address), not an address asserted by the remote peer. Returns
    /// [`PeerRouteKind::Unknown`] when the source is absent, non-IP, or
    /// matches no known address.
    ///
    /// This is a **provenance label** for logging and metrics — not an
    /// identity or trust signal. It never infers a route from address-list
    /// ordering, and a conflicting phonebook (same socket address tagged
    /// with two different [`AddressType`]s) is reported as
    /// [`PeerRouteKind::Unknown`] rather than silently picking a winner.
    pub async fn classify_peer_transport_route(
        &self,
        peer_id: &PeerId,
        transport_source: Option<&MultiAddr>,
    ) -> PeerRouteKind {
        let typed = self
            .adaptive_dht
            .peer_addresses_for_dial_typed(peer_id)
            .await;
        classify_peer_route_from_typed(transport_source, &typed)
    }

    /// Return this process's local last-successful-DHT-interaction age for a
    /// peer. This is a diagnostic observation, not proof of remote uptime.
    pub async fn peer_last_seen_elapsed(&self, peer_id: &PeerId) -> Option<Duration> {
        self.adaptive_dht
            .dht_manager()
            .peer_last_seen_elapsed(peer_id)
            .await
    }

    // =========================================================================
    // Request/Response API — Automatic Trust Feedback
    // =========================================================================

    /// Send a request to a peer and wait for a response with automatic trust penalty reporting.
    ///
    /// Unlike fire-and-forget `send_message()`, this method:
    /// 1. Wraps the payload in a `RequestResponseEnvelope` with a unique message ID
    /// 2. Sends it on the `/rr/<protocol>` protocol prefix
    /// 3. Waits for a matching response (or timeout)
    /// 4. Automatically reports failure to the trust engine (success is the expected baseline)
    ///
    /// The remote peer's handler should call `send_response()` with the
    /// incoming message ID to route the response back.
    ///
    /// # Arguments
    ///
    /// * `peer_id` - Target peer
    /// * `protocol` - Application protocol name (e.g. `"peer_info"`)
    /// * `data` - Request payload bytes
    /// * `timeout` - Maximum time to wait for a response
    ///
    /// # Returns
    ///
    /// A `PeerResponse` on success, or an error on timeout / connection failure.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let response = node.send_request(&peer_id, "peer_info", request_data, Duration::from_secs(10)).await?;
    /// println!("Got {} bytes from {}", response.data.len(), response.peer_id);
    /// ```
    pub async fn send_request(
        &self,
        peer_id: &PeerId,
        protocol: &str,
        data: Vec<u8>,
        timeout: Duration,
    ) -> Result<PeerResponse> {
        let result = self
            .send_request_reconnecting(peer_id, protocol, data, timeout)
            .await;
        if let Err(ref e) = result {
            let event = if matches!(e, P2PError::Timeout(_)) {
                TrustEvent::ConnectionTimeout
            } else {
                TrustEvent::ConnectionFailed
            };
            self.report_trust_event(peer_id, event).await;
        }
        result
    }

    /// Request/response send with reconnect-on-demand.
    ///
    /// Mirrors [`Self::send_message`]: when there is no live channel to
    /// `peer_id` it dials one (serialised per peer via
    /// [`Self::reconnect_lock_for`]) before sending, and when an existing
    /// channel turns out to be stale it tears it down, reconnects, and retries
    /// the request exactly once. The plain transport `send_request` only sends
    /// over a pre-existing channel and fails fast with `PeerNotFound`
    /// otherwise; routing request/response through this reconnecting path means
    /// a request to a peer whose QUIC connection has dropped (e.g. a periodic
    /// audit of a close peer that idled out) re-establishes the connection
    /// instead of surfacing as a spurious timeout.
    ///
    /// `timeout` bounds only the response wait inside the transport; the dial
    /// is independently bounded by `connect_peer_typed` plus
    /// [`crate::dht_network_manager::IDENTITY_EXCHANGE_TIMEOUT`].
    async fn send_request_reconnecting(
        &self,
        peer_id: &PeerId,
        protocol: &str,
        data: Vec<u8>,
        timeout: Duration,
    ) -> Result<PeerResponse> {
        // Snapshot channel IDs before the send attempt — transport.send_request
        // prunes dead channels from bookkeeping but does NOT close the
        // underlying QUIC connection. We need the original IDs for
        // disconnect_channel later.
        let existing_channels = self.transport.channels_for_peer(peer_id).await;

        // No live channel — serialise dials so concurrent requests to the same
        // unconnected peer don't each open their own QUIC connection.
        if existing_channels.is_empty() {
            // Hold the per-peer reconnect lock only across the dial so
            // concurrent requests to the same cold peer collapse onto one dial —
            // not across the response wait, which would serialise every such
            // request for the full `timeout`.
            {
                let lock = self.reconnect_lock_for(peer_id);
                let _guard = lock.lock().await;
                // Another caller may have connected while we waited for the lock.
                if !self.transport.is_peer_connected(peer_id).await {
                    self.ensure_channel(peer_id, &[], &[], &[]).await?;
                }
            }
            return self
                .transport
                .send_request(peer_id, protocol, data, timeout)
                .await;
        }

        // Snapshot addresses before the attempt — transport.send_request prunes
        // stale channels, which removes peer_info.
        let saved_addrs: Vec<MultiAddr> = self
            .transport
            .peer_info(peer_id)
            .await
            .map(|info| info.addresses)
            .unwrap_or_default();

        // Clone the payload for a possible retry — transport.send_request
        // consumes the Vec, and only stale-channel failures are retried.
        let retry_data = data.clone();

        // Fast path: try the existing connection.
        match self
            .transport
            .send_request(peer_id, protocol, data, timeout)
            .await
        {
            Ok(resp) => return Ok(resp),
            Err(e) => {
                // A response-deadline timeout means the request WAS delivered
                // but went unanswered — reconnecting would not help, so do not
                // retry. Only a stale-channel send failure warrants a redial.
                if !e.is_stale_channel_send_failure() {
                    return Err(e);
                }
                debug!(
                    peer = %peer_id.to_hex(),
                    error = %e,
                    "stale channel request failed, attempting reconnect",
                );
            }
        }

        // Serialise the reconnect (stale-channel teardown + dial) so concurrent
        // requests to the same stale peer don't race to dial, but release the
        // lock before the response wait so they don't serialise for the full
        // `timeout`.
        {
            let lock = self.reconnect_lock_for(peer_id);
            let _guard = lock.lock().await;

            // Another caller may have reconnected while we waited for the lock.
            if self.transport.is_peer_connected(peer_id).await {
                // Close stale QUIC connections that transport.send_request's
                // bookkeeping cleanup didn't tear down (it only drops the mapping).
                for channel_id in &existing_channels {
                    self.transport.disconnect_channel(channel_id).await;
                }
            } else {
                self.ensure_channel(peer_id, &[], &saved_addrs, &existing_channels)
                    .await?;
            }
        }
        self.transport
            .send_request(peer_id, protocol, retry_data, timeout)
            .await
    }

    pub async fn send_response(
        &self,
        peer_id: &PeerId,
        protocol: &str,
        message_id: &str,
        data: Vec<u8>,
    ) -> Result<()> {
        self.transport
            .send_response(peer_id, protocol, message_id, data)
            .await
    }

    pub fn parse_request_envelope(data: &[u8]) -> Option<(String, bool, Vec<u8>)> {
        crate::transport_handle::TransportHandle::parse_request_envelope(data)
    }

    pub async fn subscribe(&self, topic: &str) -> Result<()> {
        self.transport.subscribe(topic).await
    }

    pub async fn publish(&self, topic: &str, data: &[u8]) -> Result<()> {
        self.transport.publish(topic, data).await
    }

    /// Get the node configuration
    pub fn config(&self) -> &NodeConfig {
        &self.config
    }

    /// Start the P2P node
    pub async fn start(&self) -> Result<()> {
        info!("Starting P2P node...");

        // Start transport listeners and message receiving
        self.transport.start_network_listeners().await?;

        // Start the adaptive DHT layer (DHT manager + trust engine)
        self.adaptive_dht.start().await?;

        // Log current listen addresses
        let listen_addrs = self.transport.listen_addrs().await;
        info!("P2P node started on addresses: {:?}", listen_addrs);

        // NOTE: Message receiving is now integrated into the accept loop in start_network_listeners()
        // The old start_message_receiving_system() is no longer needed as it competed with the accept
        // loop for incoming connections, causing messages to be lost.

        // Load close group cache and import trust scores before connecting to peers.
        // This ensures trust scores are available when peers are added to the routing table.
        let close_group_cache = if let Some(ref dir) = self.config.close_group_cache_dir {
            match CloseGroupCache::load_from_dir(dir).await {
                Ok(Some(cache)) => {
                    let now_epoch = SystemTime::now()
                        .duration_since(UNIX_EPOCH)
                        .map_or(0, |duration| duration.as_secs());
                    if cache.is_stale(now_epoch, self.config.close_group_cache_max_age) {
                        warn!(
                            cache_age_secs = now_epoch.saturating_sub(cache.saved_at_epoch_secs),
                            max_age_secs = self
                                .config
                                .close_group_cache_max_age
                                .map(|age| age.as_secs()),
                            "Close group cache is stale; skipping cached trust and Priority-0 peers"
                        );
                        None
                    } else {
                        // Filter out peers with non-finite trust scores (NaN/Inf)
                        // that could corrupt trust engine state or sort ordering.
                        let original_count = cache.peers.len();
                        let cache = CloseGroupCache {
                            peers: cache
                                .peers
                                .into_iter()
                                .filter(|p| p.trust.score.is_finite())
                                .collect(),
                            ..cache
                        };
                        let filtered_count = original_count - cache.peers.len();
                        if filtered_count > 0 {
                            warn!(
                                "Filtered {filtered_count} peers with non-finite trust scores from close group cache"
                            );
                        }

                        let trust_snapshot = TrustSnapshot {
                            peers: cache
                                .peers
                                .iter()
                                .map(|p| (p.peer_id, p.trust.clone()))
                                .collect(),
                        };
                        self.adaptive_dht
                            .trust_engine()
                            .import_snapshot(&trust_snapshot);
                        info!(
                            cache_age_secs = now_epoch.saturating_sub(cache.saved_at_epoch_secs),
                            saved_at_epoch_secs = cache.saved_at_epoch_secs,
                            "Loaded {} peers from close group cache (trust scores imported)",
                            cache.peers.len()
                        );
                        Some(cache)
                    }
                }
                Ok(None) => {
                    debug!(
                        "No close group cache found in {}, fresh start",
                        dir.display()
                    );
                    None
                }
                Err(e) => {
                    warn!(
                        "Failed to load close group cache from {}: {e}",
                        dir.display()
                    );
                    None
                }
            }
        } else {
            None
        };

        // Connect to bootstrap peers
        self.connect_bootstrap_peers(close_group_cache.as_ref())
            .await?;

        // Emit BootstrapComplete — the node is connected to the network and
        // the DHT routing table is populated; consumers waiting on this
        // event can start issuing queries. The relay-acquisition driver
        // runs asynchronously after this point, so the node's published
        // self-record may be direct-only for a brief window until the
        // driver's first acquisition attempt finishes.
        {
            let dht = self.adaptive_dht.dht_manager();
            let rt_size = dht.get_routing_table_size().await;
            dht.emit_event(DhtNetworkEvent::BootstrapComplete { num_peers: rt_size });
        }

        // Spawn the relay-acquisition driver for Node mode.
        //
        // The driver unconditionally tries to acquire a MASQUE relay from
        // an XOR-closest peer right after bootstrap — there is no public/
        // private classification. Private candidates are filtered out
        // ambiently: their Direct addresses are unreachable from outside
        // their NAT, so the QUIC dial fails and the walker advances to
        // the next-closest peer.
        //
        // The driver also owns the relay-lost → republish → reacquire
        // state machine (see `reachability::driver` for the full flow).
        // Clients (`NodeMode::Client`) do not run the driver at all: they
        // are outbound-only and do not need to be reachable.
        if self.config.mode != NodeMode::Client {
            spawn_acquisition_driver(
                self.adaptive_dht.dht_manager().clone(),
                Arc::clone(&self.transport),
                Arc::clone(&self.relayer_peer_id),
                Arc::clone(&self.relay_address),
                self.shutdown.clone(),
            );
        } else {
            info!("client mode — skipping relay acquisition driver");
        }

        if let Some(dir) = self.config.close_group_cache_dir.clone() {
            let interval = self
                .config
                .dht_config
                .refresh_interval
                .max(MIN_CLOSE_GROUP_CACHE_SAVE_INTERVAL);
            let mut task = self.close_group_cache_save_handle.lock().await;
            if task.is_none() {
                let dht_manager = Arc::clone(self.adaptive_dht.dht_manager());
                let trust_engine = Arc::clone(self.adaptive_dht.trust_engine());
                let peer_id = self.peer_id;
                let k_value = self.config.dht_config.k_value;
                let shutdown = self.close_group_cache_save_shutdown.clone();
                let snapshot_floor = Arc::clone(&self.routing_snapshot_floor);
                let snapshot_floor_at = Arc::clone(&self.routing_snapshot_floor_at);
                *task = Some(tokio::spawn(periodic_close_group_cache_save(
                    dht_manager,
                    trust_engine,
                    peer_id,
                    k_value,
                    dir,
                    interval,
                    snapshot_floor,
                    snapshot_floor_at,
                    shutdown,
                )));
                info!(
                    interval_secs = interval.as_secs(),
                    "Started periodic close group cache persistence"
                );
            }
        }

        self.is_started
            .store(true, std::sync::atomic::Ordering::Release);

        Ok(())
    }

    // start_network_listeners and start_message_receiving_system
    // are now implemented in TransportHandle

    /// Run the P2P node (blocks until shutdown)
    pub async fn run(&self) -> Result<()> {
        if !self.is_running() {
            self.start().await?;
        }

        info!("P2P node running...");

        // Block until shutdown is signalled. All background work (connection
        // lifecycle, DHT maintenance, EigenTrust) runs in dedicated tasks.
        self.shutdown.cancelled().await;

        info!("P2P node stopped");
        Ok(())
    }

    /// Stop the P2P node
    pub async fn stop(&self) -> Result<()> {
        info!("Stopping P2P node...");

        // Stop periodic cache persistence and wait for any in-flight write.
        // The final save below is then the authoritative shutdown snapshot.
        self.close_group_cache_save_shutdown.cancel();
        let cache_task = self.close_group_cache_save_handle.lock().await.take();
        if let Some(cache_task) = cache_task
            && let Err(error) = cache_task.await
        {
            warn!("Periodic close group cache task failed during shutdown: {error}");
        }

        // Save close group cache and routing snapshot before tearing down the
        // DHT and transport layers.
        if let Some(ref dir) = self.config.close_group_cache_dir {
            if let Err(e) = self.save_close_group_cache(dir, "shutdown").await {
                warn!("Failed to save close group cache on shutdown: {e}");
            }
            let now_epoch = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map_or(0, |duration| duration.as_secs());
            save_routing_snapshot(
                self.dht_manager(),
                self.peer_id,
                now_epoch,
                dir,
                &self.routing_snapshot_floor,
                &self.routing_snapshot_floor_at,
                "shutdown",
            )
            .await;
        }

        // Signal the run loop to exit
        self.shutdown.cancel();

        // Stop DHT layer first so leave messages can be sent while transport is still active.
        self.adaptive_dht.stop().await?;

        // Stop the transport layer (shutdown endpoints, join tasks, disconnect peers)
        self.transport.stop().await?;

        self.is_started
            .store(false, std::sync::atomic::Ordering::Release);

        info!("P2P node stopped");
        Ok(())
    }

    /// Graceful shutdown alias for tests
    pub async fn shutdown(&self) -> Result<()> {
        self.stop().await
    }

    /// Check if the node is running
    pub fn is_running(&self) -> bool {
        self.is_started.load(std::sync::atomic::Ordering::Acquire) && !self.shutdown.is_cancelled()
    }

    /// Get the current listen addresses
    pub async fn listen_addrs(&self) -> Vec<MultiAddr> {
        self.transport.listen_addrs().await
    }

    /// Get connected peers
    pub async fn connected_peers(&self) -> Vec<PeerId> {
        self.transport.connected_peers().await
    }

    /// Get peer count
    pub async fn peer_count(&self) -> usize {
        self.transport.peer_count().await
    }

    /// Get peer info
    pub async fn peer_info(&self, peer_id: &PeerId) -> Option<PeerInfo> {
        self.transport.peer_info(peer_id).await
    }

    /// Get the channel ID for a given address, if connected (internal only).
    #[allow(dead_code)]
    pub(crate) async fn get_channel_id_by_address(&self, addr: &MultiAddr) -> Option<String> {
        self.transport.get_channel_id_by_address(addr).await
    }

    /// List all active transport-level connections (internal only).
    #[allow(dead_code)]
    pub(crate) async fn list_active_connections(&self) -> Vec<(String, Vec<MultiAddr>)> {
        self.transport.list_active_connections().await
    }

    /// Remove a channel from the peers map (internal only).
    #[allow(dead_code)]
    pub(crate) async fn remove_channel(&self, channel_id: &str) -> bool {
        self.transport.remove_channel(channel_id).await
    }

    /// Close a channel's QUIC connection and remove it from all tracking maps.
    ///
    /// Use when a transport-level connection was established but identity
    /// exchange failed, so no [`PeerId`] is available for [`disconnect_peer`].
    pub(crate) async fn disconnect_channel(&self, channel_id: &str) {
        self.transport.disconnect_channel(channel_id).await;
    }

    /// Check if an authenticated peer is connected (has at least one active channel).
    pub async fn is_peer_connected(&self, peer_id: &PeerId) -> bool {
        self.transport.is_peer_connected(peer_id).await
    }

    /// Connect to a peer, returning the transport-level channel ID.
    ///
    /// The returned channel ID is **not** the app-level [`PeerId`]. To obtain
    /// the authenticated peer identity, call
    /// [`wait_for_peer_identity`](Self::wait_for_peer_identity) with the
    /// returned channel ID.
    ///
    /// Callers that already know how the address was classified should
    /// prefer [`Self::connect_peer_typed`] so the resulting log line
    /// carries an accurate `kind` field instead of `unknown`.
    pub async fn connect_peer(&self, address: &MultiAddr) -> Result<String> {
        self.transport.connect_peer(address).await
    }

    /// Connect to a peer at the given typed address.
    ///
    /// Same as [`Self::connect_peer`] but threads the [`AddressType`]
    /// through to the transport-level dial log so an operator can tell,
    /// after the fact, whether a failed dial was against a `Direct`,
    /// `Relay`, `Unverified`, or `Lan` address.
    pub async fn connect_peer_typed(
        &self,
        address: &MultiAddr,
        kind: AddressType,
    ) -> Result<String> {
        self.transport.connect_peer_typed(address, kind).await
    }

    /// Wait for the identity exchange on `channel_id` to complete, returning
    /// the authenticated [`PeerId`].
    ///
    /// Use this after [`connect_peer`](Self::connect_peer) to bridge the gap
    /// between the transport-level channel ID and the app-level peer identity
    /// required by [`send_message`](Self::send_message).
    pub async fn wait_for_peer_identity(
        &self,
        channel_id: &str,
        timeout: Duration,
    ) -> Result<PeerId> {
        self.transport
            .wait_for_peer_identity(channel_id, timeout)
            .await
    }

    /// Disconnect from a peer
    pub async fn disconnect_peer(&self, peer_id: &PeerId) -> Result<()> {
        self.transport.disconnect_peer(peer_id).await
    }

    /// Check if a connection to a peer is active (internal only).
    #[allow(dead_code)]
    pub(crate) async fn is_connection_active(&self, channel_id: &str) -> bool {
        self.transport.is_connection_active(channel_id).await
    }

    /// Send a message to an authenticated peer, reconnecting on demand.
    ///
    /// Tries the existing connection first. If the send fails (stale QUIC
    /// session, peer not found, etc.), resolves a dial address from:
    ///
    /// 1. Caller-provided `addrs` (highest priority)
    /// 2. Latest owner-published QUIC addresses in the DHT routing table
    /// 3. Addresses cached in the transport layer (snapshotted before the
    ///    send attempt, since stale-channel cleanup removes them)
    /// 4. Other DHT routing-table or live transport addresses
    ///
    /// Then dials, waits for identity exchange, and retries the send exactly
    /// once on the fresh connection.  Concurrent reconnects to the same peer
    /// are serialised so only one dial is attempted at a time.
    pub async fn send_message(
        &self,
        peer_id: &PeerId,
        protocol: &str,
        data: impl Into<Bytes>,
        addrs: &[MultiAddr],
    ) -> Result<()> {
        // `Bytes` lets retries and the transport share one buffer; a `Vec<u8>`
        // converts without copying.
        let data: Bytes = data.into();
        // Snapshot channel IDs before the send attempt — transport.send_message
        // prunes dead channels from bookkeeping but does NOT close the
        // underlying QUIC connection.  We need the original IDs for
        // disconnect_channel later.
        let existing_channels = self.transport.channels_for_peer(peer_id).await;

        // No existing connection — serialise so concurrent sends to the same
        // unconnected peer don't each open their own QUIC connection.
        if existing_channels.is_empty() {
            let lock = self.reconnect_lock_for(peer_id);
            let _guard = lock.lock().await;

            // Another sender may have connected while we waited for the lock.
            if self.transport.is_peer_connected(peer_id).await {
                return self.transport.send_message(peer_id, protocol, data).await;
            }

            return self
                .reconnect_and_send(peer_id, protocol, data, addrs, &[], &[])
                .await;
        }

        // Snapshot addresses before the send attempt — transport.send_message
        // prunes stale channels, which removes peer_info.
        let saved_addrs: Vec<MultiAddr> = self
            .transport
            .peer_info(peer_id)
            .await
            .map(|info| info.addresses)
            .unwrap_or_default();

        // Keep a handle for the retry — only stale-channel failures are
        // retried; cloning `Bytes` shares the buffer.
        let retry_data = data.clone();

        // Fast path: try existing connection.
        let send_result = self.transport.send_message(peer_id, protocol, data).await;
        match send_result {
            Ok(()) => return Ok(()),
            Err(e) => {
                if !e.is_stale_channel_send_failure() {
                    debug!(
                        peer = %peer_id.to_hex(),
                        error = %e,
                        "send failed during active channel use, not reconnecting",
                    );
                    return Err(e);
                }

                debug!(
                    peer = %peer_id.to_hex(),
                    error = %e,
                    "stale channel send failed, attempting reconnect",
                );
            }
        }

        // Serialise reconnect attempts so concurrent sends to the same
        // stale peer don't race to dial.
        let lock = self.reconnect_lock_for(peer_id);
        let _guard = lock.lock().await;

        // Another sender may have reconnected while we waited for the lock.
        if self.transport.is_peer_connected(peer_id).await {
            // Close stale QUIC connections that remove_channel (called inside
            // transport.send_message on failure) didn't tear down — it only
            // removes bookkeeping, not the underlying QUIC session.
            for channel_id in &existing_channels {
                self.transport.disconnect_channel(channel_id).await;
            }
            return self
                .transport
                .send_message(peer_id, protocol, retry_data)
                .await;
        }

        self.reconnect_and_send(
            peer_id,
            protocol,
            retry_data,
            addrs,
            &saved_addrs,
            &existing_channels,
        )
        .await
    }

    /// Ensure an identity-authenticated channel to `peer_id` exists, dialing a
    /// fresh connection when necessary.
    ///
    /// Resolves dial addresses from explicit caller input, the latest owner
    /// publication, or saved connection and discovery addresses in that order,
    /// tears down any stale channels, dials, waits for the identity exchange,
    /// and verifies the authenticated peer matches `peer_id`. On success the
    /// transport's `peer_to_channel` map is populated, so a subsequent
    /// `send_message` / `send_request` finds the channel instead of failing
    /// with `PeerNotFound`. Returns `PeerNotFound` when no dialable address is
    /// available.
    ///
    /// Shared by [`Self::reconnect_and_send`] and
    /// [`Self::send_request_reconnecting`] so both gain identical dial
    /// behaviour.
    async fn ensure_channel(
        &self,
        peer_id: &PeerId,
        addrs: &[MultiAddr],
        saved_addrs: &[MultiAddr],
        stale_channels: &[String],
    ) -> Result<()> {
        // Tear down stale QUIC connections using their actual channel IDs.
        // transport.send_message only removes bookkeeping (peer_to_channel,
        // peers, active_connections) — it does NOT close the underlying QUIC
        // connection.  We must use the real channel IDs, not the resolved
        // dial address, because NAT / port migration can make them differ.
        if !stale_channels.is_empty() {
            for channel_id in stale_channels {
                self.transport.disconnect_channel(channel_id).await;
            }
            tokio::time::sleep(QUIC_TEARDOWN_GRACE).await;
        }

        let candidates = self
            .resolve_dial_candidates(peer_id, addrs, saved_addrs)
            .await;
        if candidates.is_empty() {
            return Err(P2PError::Network(NetworkError::PeerNotFound(
                peer_id.to_hex().into(),
            )));
        }
        self.adaptive_dht
            .ensure_peer_channel(peer_id, &candidates)
            .await
    }

    /// Tear down stale channels, reconnect to a peer, and send a message.
    async fn reconnect_and_send(
        &self,
        peer_id: &PeerId,
        protocol: &str,
        data: Bytes,
        addrs: &[MultiAddr],
        saved_addrs: &[MultiAddr],
        stale_channels: &[String],
    ) -> Result<()> {
        self.ensure_channel(peer_id, addrs, saved_addrs, stale_channels)
            .await?;
        // Send on the fresh connection.
        self.transport.send_message(peer_id, protocol, data).await
    }

    /// Resolve typed dial candidates for `peer_id`. Explicit caller input wins;
    /// otherwise an owner-published QUIC set supersedes saved connection addresses.
    ///
    /// Returns every dialable (QUIC, non-unspecified) address from the chosen
    /// source. An authoritative set does not fall back to superseded
    /// cached addresses if none of its endpoints are dialable.
    /// Caller-provided / saved addresses inherit the
    /// [`AddressType`] from the DHT when possible and otherwise fall back to
    /// [`AddressType::Unverified`] — the same default the routing table
    /// applies to legacy peers that never asserted reachability.
    async fn resolve_dial_candidates(
        &self,
        peer_id: &PeerId,
        caller_addrs: &[MultiAddr],
        saved_addrs: &[MultiAddr],
    ) -> Vec<(MultiAddr, AddressType)> {
        if caller_addrs.is_empty()
            && let Some(published) = self
                .dht_manager()
                .published_peer_addresses_for_dial_typed(peer_id)
                .await
        {
            return published;
        }
        let dht_candidates = self
            .adaptive_dht
            .peer_addresses_for_dial_typed(peer_id)
            .await;
        let preferred = if !caller_addrs.is_empty() {
            caller_addrs
        } else if !saved_addrs.is_empty() {
            saved_addrs
        } else {
            return dht_candidates;
        };

        preferred
            .iter()
            .filter(|a| {
                let dialable = a
                    .dialable_socket_addr()
                    .is_some_and(|sa| !sa.ip().is_unspecified());
                if !dialable {
                    trace!(address = %a, "skipping non-dialable address");
                }
                dialable
            })
            .map(|addr| {
                let kind = dht_candidates
                    .iter()
                    .find_map(|(candidate, kind)| (candidate == addr).then_some(*kind))
                    .unwrap_or(AddressType::Unverified);
                (addr.clone(), kind)
            })
            .collect()
    }

    /// Get or create a per-peer reconnect lock.
    fn reconnect_lock_for(&self, peer_id: &PeerId) -> Arc<TokioMutex<()>> {
        self.reconnect_locks
            .lock()
            .entry(*peer_id)
            .or_insert_with(|| Arc::new(TokioMutex::new(())))
            .clone()
    }
}

/// Convenience constructor for `P2PError::Network(NetworkError::ProtocolError(...))`.
fn protocol_error(msg: impl std::fmt::Display) -> P2PError {
    P2PError::Network(NetworkError::ProtocolError(msg.to_string().into()))
}

/// Helper to send an event via a broadcast sender, logging at trace level if no receivers.
pub(crate) fn broadcast_event(tx: &broadcast::Sender<P2PEvent>, event: P2PEvent) {
    if let Err(e) = tx.send(event) {
        tracing::trace!("Event broadcast has no receivers: {e}");
    }
}

/// Result of parsing a protocol message, including optional authenticated identity.
///
/// The signed wire timestamp is carried on the inner [`P2PEvent::Message`]
/// (see its `timestamp` field) so subscribers can apply their own freshness
/// or replay policy now that the wire-level skew gate is gone.
pub(crate) struct ParsedMessage {
    /// The P2P event to broadcast.
    pub(crate) event: P2PEvent,
    /// If the message was signed and verified, the authenticated app-level [`PeerId`].
    pub(crate) authenticated_node_id: Option<PeerId>,
    /// The sender's user agent string from the wire message.
    pub(crate) user_agent: String,
    /// Decoded payload length (bytes). Lets the rx choke point compute wire
    /// envelope overhead (wire − payload) for V2-623 traffic accounting.
    pub(crate) payload_len: usize,
}

/// Parse a postcard-encoded protocol message into a `P2PEvent::Message`.
///
/// Returns `None` if the bytes cannot be deserialized as a valid `WireMessage`.
///
/// The `from` field is a required part of the wire protocol but is **not**
/// used as the event source. Instead, `source` — the transport-level peer ID
/// derived from the authenticated QUIC connection — is used so that consumers
/// can pass it directly to `send_message()`. This eliminates a spoofing
/// vector where a peer could claim an arbitrary identity via the payload.
pub(crate) fn parse_protocol_message(bytes: &[u8], source: &str) -> Option<ParsedMessage> {
    let message: WireMessage = postcard::from_bytes(bytes).ok()?;
    let transport_source = source.parse::<SocketAddr>().ok().map(MultiAddr::quic);

    // Verify app-level signature if present
    let authenticated_node_id = if !message.signature.is_empty() {
        match verify_message_signature(&message) {
            Ok(peer_id) => {
                debug!(
                    "Message from {} authenticated as app-level NodeId {}",
                    source, peer_id
                );
                Some(peer_id)
            }
            Err(e) => {
                // `source` (the channel's socket address) is left out: the
                // error names the sender's peer ID.
                warn!("Rejecting message: signature verification failed: {}", e);
                return None;
            }
        }
    } else {
        None
    };

    debug!(
        "Parsed P2PEvent::Message - topic: {}, source: {:?} (transport: {}, logical: {}), payload_len: {}",
        message.protocol,
        authenticated_node_id,
        source,
        message.from,
        message.data.len()
    );

    let payload_len = message.data.len();
    Some(ParsedMessage {
        event: P2PEvent::Message {
            topic: message.protocol,
            source: authenticated_node_id,
            transport_source,
            timestamp: message.timestamp,
            data: message.data,
        },
        authenticated_node_id,
        payload_len,
        user_agent: message.user_agent,
    })
}

/// Verify the ML-DSA-65 signature on a WireMessage and return the authenticated [`PeerId`].
///
/// Besides verifying the cryptographic signature, this also checks that the
/// self-asserted `from` field matches the [`PeerId`] derived from the public
/// key. This prevents a sender from signing with their real key while
/// claiming a different identity in the `from` field.
fn verify_message_signature(message: &WireMessage) -> std::result::Result<PeerId, String> {
    let pubkey = MlDsaPublicKey::from_bytes(&message.public_key)
        .map_err(|e| format!("invalid public key: {e:?}"))?;

    let peer_id = peer_id_from_public_key(&pubkey);

    // Validate that the self-asserted `from` field matches the public key.
    if message.from != peer_id {
        return Err(format!(
            "from field mismatch: message claims '{}' but public key derives '{}'",
            message.from, peer_id
        ));
    }

    let signable = postcard::to_stdvec(&(
        &message.protocol,
        &message.data as &[u8],
        &message.from,
        message.timestamp,
        &message.user_agent,
    ))
    .map_err(|e| format!("failed to serialize signable bytes: {e}"))?;

    let sig = MlDsaSignature::from_bytes(&message.signature)
        .map_err(|e| format!("invalid signature: {e:?}"))?;

    let valid = crate::quantum_crypto::ml_dsa_verify(&pubkey, &signable, &sig)
        .map_err(|e| format!("verification error: {e}"))?;

    if valid {
        Ok(peer_id)
    } else {
        Err("signature is invalid".to_string())
    }
}

impl P2PNode {
    /// Subscribe to network events
    pub fn subscribe_events(&self) -> broadcast::Receiver<P2PEvent> {
        self.transport.subscribe_events()
    }

    /// Backwards-compat event stream accessor for tests
    pub fn events(&self) -> broadcast::Receiver<P2PEvent> {
        self.subscribe_events()
    }

    /// Get node uptime
    pub fn uptime(&self) -> Duration {
        self.start_time.elapsed()
    }

    // MCP removed: all MCP tool/service methods removed

    // /// Handle MCP remote tool call with network integration

    // /// List tools available on a specific remote peer

    // /// Get MCP server statistics

    // Background tasks (connection_lifecycle_monitor, keepalive, periodic_maintenance)
    // are now implemented in TransportHandle.

    /// Check system health
    pub async fn health_check(&self) -> Result<()> {
        let peer_count = self.peer_count().await;
        if peer_count > self.config.max_connections {
            Err(protocol_error(format!(
                "Too many connections: {peer_count}"
            )))
        } else {
            Ok(())
        }
    }

    /// Get the attached DHT manager.
    pub fn dht_manager(&self) -> &Arc<DhtNetworkManager> {
        self.adaptive_dht.dht_manager()
    }

    /// Backwards-compatible alias for `dht_manager()`.
    pub fn dht(&self) -> &Arc<DhtNetworkManager> {
        self.dht_manager()
    }

    /// Connect to bootstrap peers and perform initial peer discovery.
    ///
    /// If a `close_group_cache` was loaded on startup, its peers are injected
    /// as the highest-priority addresses before configured bootstrap peers.
    /// Their trust scores were already imported into the `TrustEngine` before
    /// this method is called.
    async fn connect_bootstrap_peers(
        &self,
        close_group_cache: Option<&CloseGroupCache>,
    ) -> Result<()> {
        // Each entry is a list of addresses for a single peer. Close-group
        // peers are dialed serially to preserve trust-priority ordering;
        // configured bootstrap peers are dialed concurrently to cut cold-start
        // latency when some peers are slow or dead.
        let mut serial_addr_sets: Vec<(PeerId, Vec<MultiAddr>)> = Vec::new();
        let mut parallel_addr_sets: Vec<Vec<MultiAddr>> = Vec::new();
        let mut seen_addresses = std::collections::HashSet::new();

        // Priority 0: Cached close group peers (pre-trusted, highest priority).
        // These peers had trust scores loaded into the TrustEngine earlier in start(),
        // so they are already known-good when added to the routing table.
        // Sorted by trust score (highest first), then XOR distance (closest first)
        // as tiebreaker so we reconnect to the most trusted, closest peers first.
        if let Some(cache) = close_group_cache {
            let mut sorted_peers: Vec<&CachedCloseGroupPeer> = cache.peers.iter().collect();
            sorted_peers.sort_by(|a, b| {
                // NaN-safe comparison: push NaN scores to the back instead
                // of treating them as equal (which would silently promote
                // corrupted entries to the front of the reconnection queue).
                let score_ord = match b.trust.score.partial_cmp(&a.trust.score) {
                    Some(ord) => ord,
                    None => {
                        if a.trust.score.is_nan() {
                            std::cmp::Ordering::Greater // a is NaN, push to back
                        } else {
                            std::cmp::Ordering::Less // b is NaN, push b to back
                        }
                    }
                };
                score_ord.then_with(|| {
                    let da = self.peer_id.xor_distance(&a.peer_id);
                    let db = self.peer_id.xor_distance(&b.peer_id);
                    da.cmp(&db)
                })
            });

            let mut added_from_close_group = 0usize;
            for peer in &sorted_peers {
                let new_addresses: Vec<MultiAddr> = peer
                    .addresses
                    .iter()
                    .filter(|a| {
                        a.dialable_socket_addr()
                            .is_some_and(|sa| !seen_addresses.contains(&sa))
                    })
                    .cloned()
                    .collect();

                if !new_addresses.is_empty() {
                    for addr in &new_addresses {
                        if let Some(sa) = addr.socket_addr() {
                            seen_addresses.insert(sa);
                        }
                    }
                    serial_addr_sets.push((peer.peer_id, new_addresses));
                    added_from_close_group += 1;
                }
            }
            if added_from_close_group > 0 {
                info!(
                    "Added {} close group cache peers (highest trust first)",
                    added_from_close_group
                );
            }
        }

        // Priority 1: Configured bootstrap peers.
        if !self.config.bootstrap_peers.is_empty() {
            info!(
                "Using {} configured bootstrap peers (priority)",
                self.config.bootstrap_peers.len()
            );
            for multiaddr in &self.config.bootstrap_peers {
                let Some(socket_addr) = multiaddr.dialable_socket_addr() else {
                    warn!(
                        "Skipping non-QUIC bootstrap peer: {}",
                        multiaddr.transport()
                    );
                    continue;
                };
                seen_addresses.insert(socket_addr);
                parallel_addr_sets.push(vec![multiaddr.clone()]);
            }
        }

        // Priority 2: the routing snapshot, if one validated. Addresses already
        // queued as close-group or configured candidates are skipped so a peer
        // is never dialled twice.
        //
        // Restored once per process. A client keeps its existing six-peer
        // startup bound: it does not serve the DHT, so it never asks whether it
        // is responsible for a key, which is the only question this repairs.
        // A re-bootstrap of a running node skips it too — the table it would
        // restore is the table the node already has.
        let snapshot_addr_sets = if self.config.mode == NodeMode::Client
            || self.routing_snapshot_restored.swap(true, Ordering::Relaxed)
        {
            None
        } else {
            self.routing_snapshot_dial_sets(&mut seen_addresses).await
        };

        if serial_addr_sets.is_empty()
            && parallel_addr_sets.is_empty()
            && snapshot_addr_sets.is_none()
        {
            info!("No bootstrap peers configured");
            return Ok(());
        }

        // Connect to bootstrap peers, wait for identity exchange, then
        // perform DHT peer discovery using the real cryptographic PeerIds.
        let identity_timeout = Duration::from_secs(BOOTSTRAP_IDENTITY_TIMEOUT_SECS);
        let mut successful_connections = 0;
        let cache_dial_candidates = serial_addr_sets.len();
        let configured_dial_candidates = parallel_addr_sets.len();
        let mut cache_dial_successes = 0usize;
        let mut configured_dial_successes = 0usize;
        let mut connected_peer_ids: Vec<PeerId> = Vec::new();

        // Phase A: serial close-group dials to preserve trust-priority ordering.
        let client_mode = matches!(self.config.mode, NodeMode::Client);
        for (expected_peer_id, addrs) in &serial_addr_sets {
            if let Some(peer_id) = self
                .dial_bootstrap_addr_set(addrs, identity_timeout, "cache", Some(*expected_peer_id))
                .await
            {
                successful_connections += 1;
                cache_dial_successes += 1;
                connected_peer_ids.push(peer_id);
                if client_mode && successful_connections >= CLIENT_BOOTSTRAP_TARGET {
                    debug!(
                        "Client bootstrap target reached ({successful_connections} peers) — skipping remaining serial dials"
                    );
                    break;
                }
            }
        }

        // Phase B: concurrent dials of configured bootstrap peers, bounded by
        // `MAX_CONCURRENT_BOOTSTRAP_DIALS` to cap simultaneous QUIC+PQC
        // handshakes. Skipped entirely when a client has already hit its
        // target during Phase A.
        if !client_mode || successful_connections < CLIENT_BOOTSTRAP_TARGET {
            let mut parallel_stream =
                futures::stream::iter(parallel_addr_sets.into_iter().map(|addrs| async move {
                    self.dial_bootstrap_addr_set(&addrs, identity_timeout, "configured", None)
                        .await
                }))
                .buffer_unordered(MAX_CONCURRENT_BOOTSTRAP_DIALS);
            while let Some(result) = parallel_stream.next().await {
                if let Some(peer_id) = result {
                    successful_connections += 1;
                    configured_dial_successes += 1;
                    connected_peer_ids.push(peer_id);
                    if client_mode && successful_connections >= CLIENT_BOOTSTRAP_TARGET {
                        debug!(
                            "Client bootstrap target reached ({successful_connections} peers) — cancelling pending dials"
                        );
                        break;
                    }
                }
            }
            // `parallel_stream` is dropped here when the `if` block exits,
            // cancelling any in-flight futures inside `buffer_unordered`
            // before we proceed to the DHT discovery phase below.
        }

        // Phase C: the routing snapshot.
        //
        // The close group reconnects a neighbourhood; this restores the rest of
        // the table, which is what a node consults to decide whether anyone is
        // closer to a given key than it is. Dialled last, because the close
        // group and the configured peers are what connectivity depends on, and
        // concurrently, because the set is an order of magnitude larger.
        //
        // Two bounds, both deliberate:
        //
        // - New dials stop once the budget expires; dials already in flight are
        //   allowed to finish, each bounded by `identity_timeout`. Cancelling a
        //   handshake mid-flight would leave the far side holding a half-open
        //   connection, which is a worse outcome than waiting out one timeout.
        // - Successes are NOT added to `connected_peer_ids`. A dialled,
        //   identity-verified peer is already admitted to the routing table by
        //   the connection path, so seeding DHT discovery with all of them would
        //   issue a serial FIND_NODE per peer to rediscover the table just
        //   restored, and then serially dial everything those queries returned.
        //
        // Every peer here is dialled and identity-verified through the same path
        // as any other candidate. Nothing from the file enters the routing table
        // on the file's authority alone.
        let mut snapshot_dial_candidates = 0usize;
        let mut snapshot_dial_successes = 0usize;
        if let Some(snapshot_sets) = snapshot_addr_sets {
            snapshot_dial_candidates = snapshot_sets.len();
            let deadline = tokio::time::Instant::now() + SNAPSHOT_RESTORE_BUDGET;
            let mut snapshot_stream = futures::stream::iter(snapshot_sets.into_iter().map(
                |(expected_peer_id, addrs)| async move {
                    if tokio::time::Instant::now() >= deadline {
                        return None;
                    }
                    self.dial_bootstrap_addr_set(
                        &addrs,
                        identity_timeout,
                        "routing_snapshot",
                        Some(expected_peer_id),
                    )
                    .await
                },
            ))
            .buffer_unordered(MAX_CONCURRENT_SNAPSHOT_DIALS);

            while let Some(outcome) = snapshot_stream.next().await {
                if outcome.is_some() {
                    snapshot_dial_successes += 1;
                    successful_connections += 1;
                }
            }

            if snapshot_dial_successes < snapshot_dial_candidates {
                debug!(
                    snapshot_dial_successes,
                    snapshot_dial_candidates,
                    "Some routing-snapshot peers were not restored; the rest is left to \
                     ordinary discovery"
                );
            }
        }

        info!(
            cache_dial_candidates,
            cache_dial_successes,
            configured_dial_candidates,
            configured_dial_successes,
            snapshot_dial_candidates,
            snapshot_dial_successes,
            outbound_bootstrap_successes = successful_connections,
            outbound_reachable = successful_connections > 0,
            "Bootstrap reachability summary"
        );

        if successful_connections == 0 {
            // Outbound connections failed — but for nodes behind symmetric NAT,
            // the bootstrap peer may have already connected INBOUND to us.
            // Wait briefly and check if we have any transport-level connections.
            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
            let transport_peers = self.transport.connected_peers().await;
            if !transport_peers.is_empty() {
                info!(
                    "No outbound bootstrap succeeded, but {} inbound peer(s) connected — proceeding with DHT bootstrap",
                    transport_peers.len()
                );
                connected_peer_ids = transport_peers;
                successful_connections = connected_peer_ids.len();
            } else {
                warn!("Failed to connect to any bootstrap peers");
                // Starting a node should not be gated on immediate bootstrap connectivity.
                // Keep running and allow background discovery / retries to populate peers later.
                return Ok(());
            }
        }

        info!(
            "Successfully connected to {} bootstrap peers",
            successful_connections
        );

        // Perform DHT peer discovery from connected bootstrap peers.
        match self
            .dht_manager()
            .bootstrap_from_peers(&connected_peer_ids)
            .await
        {
            Ok(count) => info!("DHT peer discovery found {} peers", count),
            Err(e) => warn!("DHT peer discovery failed: {}", e),
        }

        // Perform two consecutive self-lookups to fully refresh the close
        // neighborhood. The second lookup may discover peers that joined or
        // became reachable during the first lookup (Section 11.2 step 5).
        //
        // Client-mode nodes don't serve the DHT, so they don't need an
        // accurate close neighborhood — they just need enough peers to route
        // lookups for their own requests, which `bootstrap_from_peers` above
        // already provides. Skipping the self-lookups here cuts cold-start
        // latency by tens of seconds when α-sized batches include dead peers.
        if matches!(self.config.mode, NodeMode::Node) {
            const SELF_LOOKUP_ROUNDS: u8 = 2;
            for i in 1..=SELF_LOOKUP_ROUNDS {
                if let Err(e) = self.dht_manager().trigger_self_lookup().await {
                    warn!("Post-bootstrap self-lookup {i}/{SELF_LOOKUP_ROUNDS} failed: {e}");
                } else {
                    debug!("Post-bootstrap self-lookup {i}/{SELF_LOOKUP_ROUNDS} completed");
                }
            }
        } else {
            debug!("Skipping post-bootstrap self-lookups (client mode)");
        }

        // Mark node as bootstrapped - we have connected to bootstrap peers
        // and initiated peer discovery
        self.is_bootstrapped.store(true, Ordering::SeqCst);
        info!(
            "Bootstrap complete: connected to {} peers, initiated {} discovery requests",
            successful_connections,
            connected_peer_ids.len()
        );

        // Save close group cache after initial bootstrap so a crash before
        // graceful shutdown still preserves the newly-discovered close group.
        if let Some(ref dir) = self.config.close_group_cache_dir
            && let Err(e) = self.save_close_group_cache(dir, "post_bootstrap").await
        {
            warn!("Failed to save close group cache after bootstrap: {e}");
        }

        Ok(())
    }

    /// Dial a single bootstrap peer's address set, stopping at the first
    /// address that completes the identity handshake. Returns the remote peer's
    /// cryptographic PeerId on success. Safe to call concurrently for different
    /// peers.
    async fn dial_bootstrap_addr_set(
        &self,
        addrs: &[MultiAddr],
        identity_timeout: Duration,
        source: &'static str,
        expected_peer_id: Option<PeerId>,
    ) -> Option<PeerId> {
        for addr in addrs {
            // Log the transport part only: a MultiAddr's Display appends
            // `/p2p/<peer id>`, and no line here may tie a peer ID to its IP.
            let log_addr = addr.transport();
            // Bootstrap addresses come from operator-supplied seeds (CLI
            // flags or config file). The local reachability classifier hasn't
            // proven them yet, so log them as `Unverified` rather than
            // `unknown`.
            match self
                .transport
                .connect_peer_typed(addr, AddressType::Unverified)
                .await
            {
                Ok(channel_id) => match self
                    .transport
                    .wait_for_peer_identity(&channel_id, identity_timeout)
                    .await
                {
                    Ok(real_peer_id) => {
                        if !bootstrap_peer_identity_matches(expected_peer_id, real_peer_id) {
                            warn!(
                                bootstrap_source = source,
                                address = %log_addr,
                                "Bootstrap cache identity mismatch; rejecting connection"
                            );
                            self.disconnect_channel(&channel_id).await;
                            continue;
                        }
                        info!(
                            bootstrap_source = source,
                            outcome = "ok",
                            address = %log_addr,
                            "Bootstrap dial completed"
                        );
                        return Some(real_peer_id);
                    }
                    Err(e) => {
                        info!(
                            bootstrap_source = source,
                            outcome = "identity_timeout",
                            address = %log_addr,
                            error = %e,
                            "Bootstrap dial failed"
                        );
                        warn!(
                            "Timeout waiting for identity from bootstrap peer {}: {}, \
                             closing channel {}",
                            log_addr, e, channel_id
                        );
                        self.disconnect_channel(&channel_id).await;
                    }
                },
                Err(e) => {
                    info!(
                        bootstrap_source = source,
                        outcome = "connect_error",
                        address = %log_addr,
                        error = %e,
                        "Bootstrap dial failed"
                    );
                    warn!("Failed to connect to bootstrap peer {}: {}", log_addr, e);
                }
            }
        }
        None
    }

    /// Record how many peers the restore recovered from, and when.
    ///
    /// Until this is called a snapshot is never written, so a node stopped
    /// before the restore step, and a client that never restores at all, cannot
    /// overwrite a good file with an empty or partial table.
    fn resolve_snapshot_floor(&self, restored_peers: usize) {
        let now_epoch = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0, |duration| duration.as_secs());
        // Publish the timestamp first and the count second with `Release`, so a
        // save that sees a resolved floor cannot also see the zero timestamp it
        // was published with and conclude the floor has already expired.
        self.routing_snapshot_floor_at
            .store(now_epoch, Ordering::Relaxed);
        self.routing_snapshot_floor
            .store(restored_peers, Ordering::Release);
    }

    /// Load the routing snapshot and turn it into dial candidates.
    ///
    /// Returns `None` when there is nothing usable — no configured directory,
    /// no file, or a rejected one. A rejection is logged with its reason: a node
    /// that quietly cold-starts on every restart looks exactly like one that
    /// never had a snapshot, and an operator needs to tell those apart.
    async fn routing_snapshot_dial_sets(
        &self,
        seen_addresses: &mut HashSet<SocketAddr>,
    ) -> Option<Vec<(PeerId, Vec<MultiAddr>)>> {
        let dir = self.config.close_group_cache_dir.as_ref()?;

        let snapshot = match RoutingSnapshot::load_from_dir(dir).await {
            Ok(Some(snapshot)) => snapshot,
            Ok(None) => {
                self.resolve_snapshot_floor(0);
                return None;
            }
            Err(rejection) => {
                warn!(%rejection, "Discarding routing snapshot");
                self.resolve_snapshot_floor(0);
                return None;
            }
        };

        let now_epoch = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_or(0, |duration| duration.as_secs());
        let peers = match snapshot.peers_for(&self.peer_id, now_epoch) {
            Ok(peers) => peers,
            Err(rejection) => {
                warn!(%rejection, "Discarding routing snapshot");
                self.resolve_snapshot_floor(0);
                return None;
            }
        };

        // A later save must not write a table smaller than this one, for as
        // long as the floor applies.
        self.resolve_snapshot_floor(peers.len());

        let sets = snapshot_dial_sets(peers, &self.peer_id, seen_addresses);
        if sets.is_empty() {
            return None;
        }

        info!(
            snapshot_peers = peers.len(),
            new_candidates = sets.len(),
            age_secs = now_epoch.saturating_sub(snapshot.saved_at_epoch_secs),
            "Restoring routing table from snapshot"
        );
        Some(sets)
    }

    /// Persist the current close group peers and their trust scores to disk.
    async fn save_close_group_cache(
        &self,
        dir: &Path,
        save_reason: &'static str,
    ) -> anyhow::Result<()> {
        save_close_group_cache_snapshot(
            self.dht_manager(),
            self.adaptive_dht.trust_engine(),
            self.peer_id,
            self.config.dht_config.k_value,
            dir,
            save_reason,
        )
        .await
    }

    // disconnect_all_peers and periodic_tasks are now in TransportHandle
}

/// Turn snapshot peers into dial candidates, skipping self and anything an
/// earlier bootstrap priority already queued.
///
/// Extends `seen_addresses` with what it returns, so a peer reachable through
/// the close-group cache and the snapshot is dialled once, not twice. Only
/// dialable (QUIC) addresses survive.
fn snapshot_dial_sets(
    peers: &[SnapshotPeer],
    self_id: &PeerId,
    seen_addresses: &mut HashSet<SocketAddr>,
) -> Vec<(PeerId, Vec<MultiAddr>)> {
    let mut sets: Vec<(PeerId, Vec<MultiAddr>)> = Vec::new();
    let mut seen_peers: HashSet<PeerId> = HashSet::new();
    for peer in peers {
        if peer.peer_id == *self_id || !seen_peers.insert(peer.peer_id) {
            continue;
        }
        let new_addresses: Vec<MultiAddr> = peer
            .addresses
            .iter()
            .filter(|addr| {
                addr.dialable_socket_addr()
                    .is_some_and(|socket| !seen_addresses.contains(&socket))
            })
            .take(MAX_SNAPSHOT_ADDRESSES_DIALLED)
            .cloned()
            .collect();
        if new_addresses.is_empty() {
            continue;
        }
        seen_addresses.extend(
            new_addresses
                .iter()
                .filter_map(MultiAddr::dialable_socket_addr),
        );
        sets.push((peer.peer_id, new_addresses));
    }
    sets
}

/// Persist a close-group snapshot using owned subsystem handles.
///
/// Keeping this separate from `P2PNode` allows the periodic task to own every
/// dependency it needs without borrowing the node across a spawned task.
async fn save_close_group_cache_snapshot(
    dht_manager: &DhtNetworkManager,
    trust_engine: &TrustEngine,
    peer_id: PeerId,
    k_value: usize,
    dir: &Path,
    save_reason: &'static str,
) -> anyhow::Result<()> {
    let key: crate::dht::Key = *peer_id.as_bytes();
    let close_group = dht_manager.find_closest_nodes_local(&key, k_value).await;

    let now_epoch = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |duration| duration.as_secs());
    let peers: Vec<CachedCloseGroupPeer> = close_group
        .into_iter()
        .filter_map(|dht_node| {
            let score = trust_engine.score(&dht_node.peer_id);
            // Guard against NaN/Infinity — serde_json cannot round-trip
            // non-finite f64 values, which would corrupt the cache file.
            if !score.is_finite() {
                return None;
            }
            Some(CachedCloseGroupPeer {
                peer_id: dht_node.peer_id,
                addresses: dht_node.addresses,
                trust: TrustRecord {
                    score,
                    last_updated_epoch_secs: now_epoch,
                },
            })
        })
        .collect();

    let peer_count = peers.len();
    let cache = CloseGroupCache {
        peers,
        saved_at_epoch_secs: now_epoch,
    };

    cache.save_to_dir(dir).await?;
    info!(
        save_reason,
        "Saved {} close group peers to cache in {}",
        peer_count,
        dir.display()
    );

    Ok(())
}

/// Persist the whole routing table as a [`RoutingSnapshot`].
///
/// Best-effort and non-fatal: the close-group cache is the established path and
/// must not start failing because a newer, unread file could not be written.
///
/// Writes nothing until the restore step has decided what the floor is, and
/// then nothing smaller than the table it restored while that floor still
/// applies. Between opening and finishing its restore a node holds only what it
/// has re-dialled so far, and one stopped in that window would otherwise
/// replace a complete snapshot with a fragment, shrinking its own file a little
/// further on every cycle. A client never resolves a floor and so never writes
/// a snapshot it would not read back.
async fn save_routing_snapshot(
    dht_manager: &DhtNetworkManager,
    peer_id: PeerId,
    now_epoch: u64,
    dir: &Path,
    floor: &AtomicUsize,
    floor_at: &AtomicU64,
    save_reason: &'static str,
) {
    let floor_peers = floor.load(Ordering::Acquire);
    if floor_peers == SNAPSHOT_FLOOR_UNRESOLVED {
        debug!(
            save_reason,
            "Skipping routing snapshot save: the restore step has not run"
        );
        return;
    }

    let peers: Vec<SnapshotPeer> = dht_manager
        .routing_table_peers()
        .await
        .into_iter()
        .map(|node| SnapshotPeer {
            peer_id: node.peer_id,
            addresses: node.addresses,
        })
        .collect();
    let peer_count = peers.len();

    if peer_count == 0 {
        debug!(
            save_reason,
            "Skipping routing snapshot save: the table is empty"
        );
        return;
    }
    let floor_applies = now_epoch.saturating_sub(floor_at.load(Ordering::Relaxed))
        < SNAPSHOT_FLOOR_ENFORCED_FOR.as_secs();
    if floor_applies && peer_count < floor_peers {
        debug!(
            save_reason,
            peer_count,
            floor_peers,
            "Skipping routing snapshot save: it would shrink the snapshot while the table is \
             still restoring"
        );
        return;
    }

    let snapshot = RoutingSnapshot::new(peer_id, now_epoch, peers);
    match snapshot.save_to_dir(dir).await {
        Ok(()) => debug!(save_reason, peer_count, "Saved routing snapshot"),
        Err(error) => warn!(save_reason, %error, "Failed to save routing snapshot"),
    }
}

/// Periodically persist the close group until cancelled.
async fn periodic_close_group_cache_save(
    dht_manager: Arc<DhtNetworkManager>,
    trust_engine: Arc<TrustEngine>,
    peer_id: PeerId,
    k_value: usize,
    dir: PathBuf,
    interval: Duration,
    snapshot_floor: Arc<AtomicUsize>,
    snapshot_floor_at: Arc<AtomicU64>,
    shutdown: CancellationToken,
) {
    let start = tokio::time::Instant::now() + interval;
    let mut ticker = tokio::time::interval_at(start, interval);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        tokio::select! {
            biased;
            () = shutdown.cancelled() => break,
            _ = ticker.tick() => {
                if let Err(error) = save_close_group_cache_snapshot(
                    &dht_manager,
                    &trust_engine,
                    peer_id,
                    k_value,
                    &dir,
                    "periodic",
                ).await {
                    warn!("Periodic close group cache save failed: {error}");
                }
                let now_epoch = SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .map_or(0, |duration| duration.as_secs());
                save_routing_snapshot(
                    &dht_manager,
                    peer_id,
                    now_epoch,
                    &dir,
                    &snapshot_floor,
                    &snapshot_floor_at,
                    "periodic",
                ).await;
            }
        }
    }
}

/// Network sender trait for sending messages
#[async_trait::async_trait]
#[allow(dead_code)]
pub trait NetworkSender: Send + Sync {
    /// Send a message to an authenticated peer.
    async fn send_message(&self, peer_id: &PeerId, protocol: &str, data: Vec<u8>) -> Result<()>;

    /// Get our local peer ID (cryptographic identity).
    fn local_peer_id(&self) -> PeerId;
}

// P2PNetworkSender removed — NetworkSender is now implemented directly on TransportHandle.
// NodeBuilder removed — use NodeConfigBuilder + P2PNode::new() instead.

/// Helper function to register a new channel.
///
/// Sync because the underlying map is a sharded `DashMap` — no `.await` is
/// needed to take a write lock. Keeping this sync is what lets the inbound
/// accept loop in `TransportHandle` insert without yielding, so it cannot
/// stall and back-pressure the upstream handshake channel.
pub(crate) fn register_new_channel(
    peers: &DashMap<String, PeerInfo>,
    channel_id: &str,
    remote_addr: &MultiAddr,
) {
    let peer_info = PeerInfo {
        channel_id: channel_id.to_owned(),
        addresses: vec![remote_addr.clone()],
        connected_at: tokio::time::Instant::now(),
        last_seen: tokio::time::Instant::now(),
        status: ConnectionStatus::Connected,
        protocols: vec!["p2p-core/1.0.0".to_string()],
        heartbeat_count: 0,
    };
    peers.insert(channel_id.to_owned(), peer_info);
}

#[cfg(test)]
mod tests {
    use super::*;
    // MCP removed from tests
    use std::time::Duration;
    use tokio::time::timeout;

    /// 2 MiB — used in builder tests to verify max_message_size configuration.
    const TEST_MAX_MESSAGE_SIZE: usize = 2 * 1024 * 1024;
    /// A loopback channel ID with no QUIC connection behind it (discard port).
    const UNCONNECTED_CHANNEL_ID: &str = "127.0.0.1:9";

    /// The borrowing frame writer must stay byte-for-byte compatible with the
    /// owned `WireMessage` that receivers decode.
    #[test]
    fn wire_message_ref_serializes_like_wire_message() {
        let owned = WireMessage {
            protocol: "/ant/replication/1".to_string(),
            data: (0..=255u8).cycle().take(70_000).collect(),
            from: PeerId::from_bytes([7; 32]),
            timestamp: 1_790_000_000,
            user_agent: "node/0.27.4".to_string(),
            public_key: vec![1; 1952],
            signature: vec![2; 3309],
        };
        let borrowed = WireMessageRef {
            protocol: &owned.protocol,
            data: &owned.data,
            from: &owned.from,
            timestamp: owned.timestamp,
            user_agent: &owned.user_agent,
            public_key: &owned.public_key,
            signature: &owned.signature,
        };

        let expected = postcard::to_stdvec(&owned).unwrap();
        let actual = postcard::to_stdvec(&borrowed).unwrap();
        assert_eq!(actual, expected);
        assert_eq!(
            postcard::experimental::serialized_size(&borrowed).unwrap(),
            actual.len()
        );

        let decoded: WireMessage = postcard::from_bytes(&actual).unwrap();
        assert_eq!(decoded.protocol, owned.protocol);
        assert_eq!(decoded.data, owned.data);
        assert_eq!(decoded.from, owned.from);
        assert_eq!(decoded.timestamp, owned.timestamp);
        assert_eq!(decoded.user_agent, owned.user_agent);
        assert_eq!(decoded.public_key, owned.public_key);
        assert_eq!(decoded.signature, owned.signature);
    }

    #[test]
    fn cached_bootstrap_identity_must_match_handshake_peer() {
        let expected = PeerId::from_bytes([1; 32]);
        let other = PeerId::from_bytes([2; 32]);

        assert!(bootstrap_peer_identity_matches(Some(expected), expected));
        assert!(!bootstrap_peer_identity_matches(Some(expected), other));
        assert!(bootstrap_peer_identity_matches(None, other));
    }

    #[test]
    fn user_agents_identify_software_without_capability_tokens() {
        assert_eq!(
            user_agent_for_mode(NodeMode::Node),
            format!("node/{}", env!("CARGO_PKG_VERSION"))
        );
        assert_eq!(
            user_agent_for_mode(NodeMode::Client),
            format!("client/{}", env!("CARGO_PKG_VERSION"))
        );
        let mut config = NodeConfig::default();
        for custom in ["my-app/1.0", "", "node/custom;custom-feature"] {
            config.custom_user_agent = Some(custom.to_owned());
            assert_eq!(config.user_agent(), custom);
        }
    }

    // Test tool handler for network tests

    // MCP removed

    /// Helper function to create a test node configuration
    fn create_test_node_config() -> NodeConfig {
        NodeConfig {
            local: true,
            port: 0,
            ipv6: true,
            bootstrap_peers: vec![],
            connection_timeout: Duration::from_secs(2),
            max_connections: 100,
            dht_config: DHTConfig::default(),
            diversity_config: None,
            max_message_size: None,
            node_identity: None,
            mode: NodeMode::default(),
            custom_user_agent: None,
            allow_loopback: true,
            adaptive_dht_config: AdaptiveDhtConfig::default(),
            close_group_cache_dir: None,
            close_group_cache_max_age: default_close_group_cache_max_age(),
        }
    }

    /// Helper function to create a test tool
    // MCP removed: test tool helper deleted

    // =========================================================================
    // PeerRouteKind / route classification — pure-helper unit tests.
    //
    // These exercise `classify_peer_route_from_typed` directly so no live
    // node/DHT is required. `P2PNode::classify_peer_transport_route` is a
    // thin async wrapper over this helper plus `peer_addresses_for_dial_typed`,
    // so the classification logic is fully covered here.
    // =========================================================================

    fn quic_addr(s: &str) -> MultiAddr {
        MultiAddr::quic(s.parse().expect("valid socket addr"))
    }

    #[test]
    fn test_peer_route_kind_display_labels() {
        assert_eq!(PeerRouteKind::Relay.to_string(), "relay");
        assert_eq!(PeerRouteKind::Direct.to_string(), "direct");
        assert_eq!(PeerRouteKind::Lan.to_string(), "lan");
        assert_eq!(PeerRouteKind::Unverified.to_string(), "unverified");
        assert_eq!(PeerRouteKind::Unknown.to_string(), "unknown");
        // as_str agrees with Display
        assert_eq!(PeerRouteKind::Relay.as_str(), "relay");
    }

    #[test]
    fn test_classify_none_source_is_unknown() {
        let typed = vec![(quic_addr("203.0.113.7:9000"), AddressType::Relay)];
        assert_eq!(
            classify_peer_route_from_typed(None, &typed),
            PeerRouteKind::Unknown
        );
    }

    #[test]
    fn test_classify_non_ip_source_is_unknown() {
        // Bluetooth is a non-IP transport: socket_addr() returns None.
        let ble = MultiAddr::new(crate::address::TransportAddr::Ble {
            mac: [0x01, 0x02, 0x03, 0x04, 0x05, 0x06],
            psm: 128,
        });
        let typed = vec![(quic_addr("203.0.113.7:9000"), AddressType::Relay)];
        assert_eq!(
            classify_peer_route_from_typed(Some(&ble), &typed),
            PeerRouteKind::Unknown
        );
    }

    #[test]
    fn test_classify_each_address_type() {
        let src = quic_addr("203.0.113.7:9000");
        for (ty, kind) in [
            (AddressType::Relay, PeerRouteKind::Relay),
            (AddressType::Direct, PeerRouteKind::Direct),
            (AddressType::Unverified, PeerRouteKind::Unverified),
            (AddressType::Lan, PeerRouteKind::Lan),
        ] {
            let typed = vec![(src.clone(), ty)];
            assert_eq!(
                classify_peer_route_from_typed(Some(&src), &typed),
                kind,
                "mismatch for {ty:?}"
            );
        }
    }

    #[test]
    fn test_classify_no_match_is_unknown() {
        let src = quic_addr("198.51.100.42:7000");
        let typed = vec![
            (quic_addr("203.0.113.7:9000"), AddressType::Relay),
            (quic_addr("203.0.113.8:9000"), AddressType::Direct),
        ];
        assert_eq!(
            classify_peer_route_from_typed(Some(&src), &typed),
            PeerRouteKind::Unknown
        );
    }

    #[test]
    fn test_classify_empty_typed_is_unknown() {
        let src = quic_addr("203.0.113.7:9000");
        let typed: Vec<(MultiAddr, AddressType)> = vec![];
        assert_eq!(
            classify_peer_route_from_typed(Some(&src), &typed),
            PeerRouteKind::Unknown
        );
    }

    #[test]
    fn test_classify_duplicate_agreeing_type_dedups() {
        // Same socket address listed twice with the same type must yield that
        // type, not Unknown — duplicates are benign.
        let src = quic_addr("203.0.113.7:9000");
        let typed = vec![
            (quic_addr("203.0.113.7:9000"), AddressType::Relay),
            (quic_addr("203.0.113.7:9000"), AddressType::Relay),
        ];
        assert_eq!(
            classify_peer_route_from_typed(Some(&src), &typed),
            PeerRouteKind::Relay
        );
    }

    #[test]
    fn test_classify_duplicate_conflicting_types_is_unknown() {
        // Same socket address tagged with two different AddressTypes is a
        // phonebook inconsistency. The classifier refuses to guess and
        // reports Unknown — it does NOT use AddressType::priority to pick a
        // winner, since that ordering governs dial preference, not the
        // provenance of an arrived message.
        let src = quic_addr("203.0.113.7:9000");
        let typed = vec![
            (quic_addr("203.0.113.7:9000"), AddressType::Relay),
            (quic_addr("203.0.113.7:9000"), AddressType::Direct),
        ];
        assert_eq!(
            classify_peer_route_from_typed(Some(&src), &typed),
            PeerRouteKind::Unknown
        );
    }

    #[test]
    fn test_classify_matches_in_any_position() {
        // The matching entry need not be first; order is not a signal.
        let src = quic_addr("203.0.113.9:9000");
        let typed = vec![
            (quic_addr("203.0.113.7:9000"), AddressType::Relay),
            (quic_addr("203.0.113.8:9000"), AddressType::Direct),
            (quic_addr("203.0.113.9:9000"), AddressType::Lan),
            (quic_addr("203.0.113.10:9000"), AddressType::Unverified),
        ];
        assert_eq!(
            classify_peer_route_from_typed(Some(&src), &typed),
            PeerRouteKind::Lan
        );
    }

    #[test]
    fn test_classify_tcp_source_matches_tcp_entry() {
        // socket_addr() is shared by Quic/Tcp/Udp, so a TCP source matches a
        // TCP phonebook entry with the same socket address.
        let src = MultiAddr::tcp("203.0.113.7:9000".parse().unwrap());
        let typed = vec![(
            MultiAddr::tcp("203.0.113.7:9000".parse().unwrap()),
            AddressType::Direct,
        )];
        assert_eq!(
            classify_peer_route_from_typed(Some(&src), &typed),
            PeerRouteKind::Direct
        );
    }

    #[tokio::test]
    async fn reconnect_uses_latest_v1_and_v2_publications_over_saved_addresses() {
        use crate::dht_network_manager::{DhtMessageType, DhtNetworkMessage, DhtNetworkOperation};
        use crate::signed_address::SignedAddressRecord;
        use crate::{KnownReachability, TransportAddressRecord};

        let config = || {
            NodeConfig::builder()
                .local(true)
                .port(0)
                .ipv6(false)
                .build()
                .unwrap()
        };
        let receiver = P2PNode::new(config()).await.unwrap();
        let publisher = P2PNode::new(config()).await.unwrap();
        // Keep the publication sequence controlled by this test. Starting the
        // publisher's DHT would allow an automatic V2 publication to correctly
        // suppress the initial V1 fixture before it can be observed.
        receiver.transport.start_network_listeners().await.unwrap();
        receiver.dht_manager().start().await.unwrap();
        publisher.transport.start_network_listeners().await.unwrap();
        let old_address = publisher
            .listen_addrs()
            .await
            .into_iter()
            .find(MultiAddr::is_ipv4)
            .unwrap();
        let channel = receiver.connect_peer(&old_address).await.unwrap();
        receiver
            .wait_for_peer_identity(&channel, Duration::from_secs(2))
            .await
            .unwrap();
        let owner = *publisher.peer_id();
        timeout(Duration::from_secs(2), async {
            while !publisher.is_peer_connected(receiver.peer_id()).await
                || !receiver.dht_manager().is_in_routing_table(&owner).await
            {
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap();
        let saved = receiver
            .transport
            .peer_info(&owner)
            .await
            .unwrap()
            .addresses;
        let first: MultiAddr = "/ip4/9.9.9.9/udp/9001/quic".parse().unwrap();
        let second: MultiAddr = "/ip4/1.1.1.1/udp/9002/quic".parse().unwrap();
        let sequence = u64::MAX - 2;
        let operations = [
            (
                "/dht/1.0.0",
                first.clone(),
                DhtNetworkOperation::PublishAddressSet {
                    seq: sequence,
                    addresses: vec![(first.clone(), AddressType::Direct)],
                },
            ),
            (
                "/dht/address/2.0.0",
                second.clone(),
                DhtNetworkOperation::PublishAddressSetV2 {
                    record: SignedAddressRecord::sign(
                        publisher.transport.node_identity(),
                        sequence - 1,
                        vec![
                            TransportAddressRecord::from_multiaddr(
                                &second,
                                KnownReachability::Direct,
                            )
                            .unwrap()
                            .unwrap(),
                        ],
                    )
                    .unwrap(),
                },
            ),
            (
                "/dht/1.0.0",
                second.clone(),
                DhtNetworkOperation::PublishAddressSet {
                    seq: u64::MAX,
                    addresses: vec![(first.clone(), AddressType::Direct)],
                },
            ),
        ];
        for (topic, expected, operation) in operations {
            let message = DhtNetworkMessage {
                message_id: "reconnect-publication".into(),
                source: owner,
                target: None,
                message_type: DhtMessageType::Request,
                payload: operation,
                result: None,
                timestamp: 1,
                ttl: 10,
                hop_count: 0,
                signed_records: Vec::new(),
            };
            publisher
                .transport
                .send_message(
                    receiver.peer_id(),
                    topic,
                    postcard::to_stdvec(&message).unwrap(),
                )
                .await
                .unwrap();
            timeout(Duration::from_secs(2), async {
                loop {
                    let candidates = receiver.resolve_dial_candidates(&owner, &[], &saved).await;
                    if candidates == vec![(expected.clone(), AddressType::Direct)] {
                        break;
                    }
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
            })
            .await
            .expect("reconnect must use the published endpoint without the old cached endpoint");
        }
        // Explicit caller overrides still take priority.
        let explicit = receiver
            .resolve_dial_candidates(&owner, std::slice::from_ref(&old_address), &saved)
            .await;
        assert_eq!(explicit, vec![(old_address, AddressType::Unverified)]);
        receiver.stop().await.unwrap();
        publisher.stop().await.unwrap();
    }

    #[tokio::test]
    async fn reconnect_uses_saved_addresses_without_an_owner_publication() {
        let node = P2PNode::new(create_test_node_config()).await.unwrap();
        let peer = PeerId::random();
        let saved: MultiAddr = "/ip4/9.9.9.9/udp/9001/quic".parse().unwrap();
        assert_eq!(
            node.dht_manager()
                .published_peer_addresses_for_dial_typed(&peer)
                .await,
            None
        );
        assert_eq!(
            node.resolve_dial_candidates(&peer, &[], std::slice::from_ref(&saved))
                .await,
            vec![(saved, AddressType::Unverified)]
        );
    }

    #[tokio::test]
    async fn test_node_config_default() {
        let config = NodeConfig::default();

        assert_eq!(config.listen_addrs().len(), 2); // IPv4 + IPv6
        assert_eq!(config.max_connections, 10000);
        assert_eq!(config.connection_timeout, Duration::from_secs(25));
        assert_eq!(
            config.close_group_cache_max_age,
            Some(Duration::from_secs(DEFAULT_CLOSE_GROUP_CACHE_MAX_AGE_SECS))
        );
    }

    #[test]
    fn close_group_cache_builder_default_and_explicit_disable() {
        let defaulted = NodeConfig::builder().build().unwrap();
        assert_eq!(
            defaulted.close_group_cache_max_age,
            Some(Duration::from_secs(DEFAULT_CLOSE_GROUP_CACHE_MAX_AGE_SECS))
        );

        let disabled = NodeConfig::builder()
            .close_group_cache_max_age(None)
            .build()
            .unwrap();
        assert_eq!(disabled.close_group_cache_max_age, None);
        let disabled_json = serde_json::to_string(&disabled).unwrap();
        let disabled_roundtrip: NodeConfig = serde_json::from_str(&disabled_json).unwrap();
        assert_eq!(disabled_roundtrip.close_group_cache_max_age, None);

        let custom = NodeConfig::builder()
            .close_group_cache_max_age(Some(Duration::from_secs(120)))
            .build()
            .unwrap();
        assert_eq!(
            custom.close_group_cache_max_age,
            Some(Duration::from_secs(120))
        );
    }

    #[test]
    fn old_serialized_config_gets_default_cache_max_age() {
        let mut value = serde_json::to_value(NodeConfig::default()).unwrap();
        value
            .as_object_mut()
            .unwrap()
            .remove("close_group_cache_max_age");

        let decoded: NodeConfig = serde_json::from_value(value).unwrap();
        assert_eq!(
            decoded.close_group_cache_max_age,
            Some(Duration::from_secs(DEFAULT_CLOSE_GROUP_CACHE_MAX_AGE_SECS))
        );
    }

    #[tokio::test]
    async fn test_dht_config_default() {
        let config = DHTConfig::default();

        assert_eq!(config.k_value, 20);
        assert_eq!(config.alpha_value, 3);
        assert_eq!(config.refresh_interval, Duration::from_secs(600));
    }

    #[test]
    fn test_connection_status_variants() {
        let connecting = ConnectionStatus::Connecting;
        let connected = ConnectionStatus::Connected;
        let disconnecting = ConnectionStatus::Disconnecting;
        let disconnected = ConnectionStatus::Disconnected;
        let failed = ConnectionStatus::Failed("test error".to_string());

        assert_eq!(connecting, ConnectionStatus::Connecting);
        assert_eq!(connected, ConnectionStatus::Connected);
        assert_eq!(disconnecting, ConnectionStatus::Disconnecting);
        assert_eq!(disconnected, ConnectionStatus::Disconnected);
        assert_ne!(connecting, connected);

        if let ConnectionStatus::Failed(msg) = failed {
            assert_eq!(msg, "test error");
        } else {
            panic!("Expected Failed status");
        }
    }

    #[tokio::test]
    async fn test_node_creation() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // PeerId is derived from the cryptographic identity (32-byte BLAKE3 hash)
        assert_eq!(node.peer_id().to_hex().len(), 64);
        assert!(!node.is_running());
        assert_eq!(node.peer_count().await, 0);
        assert!(node.connected_peers().await.is_empty());

        Ok(())
    }

    #[tokio::test]
    async fn test_node_lifecycle() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Initially not running
        assert!(!node.is_running());

        // Start the node
        node.start().await?;
        assert!(node.is_running());

        // Check listen addresses were set (at least one)
        let listen_addrs = node.listen_addrs().await;
        assert!(
            !listen_addrs.is_empty(),
            "Expected at least one listening address"
        );

        // Stop the node
        node.stop().await?;
        assert!(!node.is_running());

        Ok(())
    }

    #[tokio::test]
    async fn close_group_cache_task_is_joined_on_stop() -> Result<()> {
        let cache_dir = tempfile::tempdir().unwrap();
        let mut config = create_test_node_config();
        config.close_group_cache_dir = Some(cache_dir.path().to_path_buf());
        let node = P2PNode::new(config).await?;

        node.start().await?;
        assert!(node.close_group_cache_save_handle.lock().await.is_some());

        node.stop().await?;
        assert!(node.close_group_cache_save_handle.lock().await.is_none());

        Ok(())
    }

    #[tokio::test]
    async fn test_peer_connection() -> Result<()> {
        let config1 = create_test_node_config();
        let config2 = create_test_node_config();

        let node1 = P2PNode::new(config1).await?;
        let node2 = P2PNode::new(config2).await?;

        node1.start().await?;
        node2.start().await?;

        let node2_addr = node2
            .listen_addrs()
            .await
            .into_iter()
            .find(|a| a.is_ipv4())
            .ok_or_else(|| {
                P2PError::Network(crate::error::NetworkError::InvalidAddress(
                    "Node 2 did not expose an IPv4 listen address".into(),
                ))
            })?;

        // Connect to a real peer (unsigned — no node_identity configured).
        // connect_peer returns a transport-level channel ID (String), not a PeerId.
        let channel_id = node1.connect_peer(&node2_addr).await?;

        // Unauthenticated connections don't appear in the app-level peer maps.
        // Verify transport-level tracking via is_connection_active / peers map.
        assert!(node1.is_connection_active(&channel_id).await);

        // Get peer info from the transport-level peers map (keyed by channel ID)
        let peer_info = node1.transport.peer_info_by_channel(&channel_id).await;
        assert!(peer_info.is_some());
        let info = peer_info.expect("Peer info should exist after connect");
        assert_eq!(info.channel_id, channel_id);
        assert_eq!(info.status, ConnectionStatus::Connected);
        assert!(info.protocols.contains(&"p2p-foundation/1.0".to_string()));

        // Disconnect the channel
        node1.remove_channel(&channel_id).await;
        assert!(!node1.is_connection_active(&channel_id).await);

        node1.stop().await?;
        node2.stop().await?;

        Ok(())
    }

    /// A send failure confined to one stream must not drop a channel whose
    /// QUIC connection is still open, or the next send redials a live peer.
    /// A stale-channel failure, or a channel with no open connection behind
    /// it, still removes the channel.
    #[tokio::test]
    async fn send_failure_keeps_channel_only_while_connection_is_open() -> Result<()> {
        let node1 = P2PNode::new(create_test_node_config()).await?;
        let node2 = P2PNode::new(create_test_node_config()).await?;
        node1.start().await?;
        node2.start().await?;

        let node2_addr = node2
            .listen_addrs()
            .await
            .into_iter()
            .find(|a| a.is_ipv4())
            .ok_or_else(|| {
                P2PError::Network(crate::error::NetworkError::InvalidAddress(
                    "Node 2 did not expose an IPv4 listen address".into(),
                ))
            })?;
        let channel_id = node1.connect_peer(&node2_addr).await?;

        let stalled_write = P2PError::Transport(crate::error::TransportError::SendFailed {
            kind: crate::error::SendFailureKind::WriteProgressTimeout,
            reason: "stream write made no progress".into(),
        });
        assert!(
            !node1
                .transport
                .remove_channel_after_send_failure(&channel_id, &stalled_write)
                .await,
            "a stalled stream on an open connection must keep the channel"
        );
        assert!(
            node1
                .transport
                .peer_info_by_channel(&channel_id)
                .await
                .is_some()
        );

        let stale = P2PError::Network(crate::error::NetworkError::PeerNotFound(
            channel_id.clone().into(),
        ));
        assert!(
            node1
                .transport
                .remove_channel_after_send_failure(&channel_id, &stale)
                .await,
            "a stale-channel failure must remove the channel"
        );
        assert!(
            node1
                .transport
                .peer_info_by_channel(&channel_id)
                .await
                .is_none()
        );

        assert!(
            node1
                .transport
                .remove_channel_after_send_failure(UNCONNECTED_CHANNEL_ID, &stalled_write)
                .await,
            "a channel with no open connection must be removed whatever the failure"
        );

        node1.stop().await?;
        node2.stop().await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_connect_peer_rejects_tcp_multiaddr() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        let tcp_addr: MultiAddr = "/ip4/127.0.0.1/tcp/1".parse().unwrap();
        let result = node.connect_peer(&tcp_addr).await;

        assert!(
            matches!(
                result,
                Err(P2PError::Network(
                    crate::error::NetworkError::InvalidAddress(_)
                ))
            ),
            "TCP multiaddrs should be rejected before a QUIC dial is attempted, got: {:?}",
            result
        );

        Ok(())
    }

    // TODO(windows): Investigate QUIC connection issues on Windows CI
    // This test consistently fails on Windows GitHub Actions runners with
    // "All connect attempts failed" even with IPv4-only config, long delays,
    // and multiple retry attempts. The underlying saorsa-transport library may have
    // issues on Windows that need investigation.
    // See: https://github.com/WithAutonomi/saorsa-core/issues/TBD
    #[cfg_attr(target_os = "windows", ignore)]
    #[tokio::test]
    async fn test_event_subscription() -> Result<()> {
        // PeerConnected/PeerDisconnected only fire for authenticated peers
        // (nodes with node_identity that send signed messages).
        // Configure both nodes with identities so the event subscription test works.
        let identity1 =
            Arc::new(NodeIdentity::generate().expect("should generate identity for test node1"));
        let identity2 =
            Arc::new(NodeIdentity::generate().expect("should generate identity for test node2"));

        let mut config1 = create_test_node_config();
        config1.ipv6 = false;
        config1.node_identity = Some(identity1);

        let node2_peer_id = *identity2.peer_id();
        let mut config2 = create_test_node_config();
        config2.ipv6 = false;
        config2.node_identity = Some(identity2);

        let node1 = P2PNode::new(config1).await?;
        let node2 = P2PNode::new(config2).await?;

        node1.start().await?;
        node2.start().await?;

        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;

        // Subscribe to node2's events (node2 will receive the signed message)
        let mut events = node2.subscribe_events();

        let node2_addr = node2.local_addr().ok_or_else(|| {
            P2PError::Network(crate::error::NetworkError::ProtocolError(
                "No listening address".to_string().into(),
            ))
        })?;

        // Connect node1 → node2
        let mut channel_id = None;
        for attempt in 0..3 {
            if attempt > 0 {
                tokio::time::sleep(tokio::time::Duration::from_millis(200)).await;
            }
            match timeout(Duration::from_secs(2), node1.connect_peer(&node2_addr)).await {
                Ok(Ok(id)) => {
                    channel_id = Some(id);
                    break;
                }
                Ok(Err(_)) | Err(_) => continue,
            }
        }
        let channel_id = channel_id.expect("Failed to connect after 3 attempts");

        // Wait for identity exchange to complete via wait_for_peer_identity.
        let target_peer_id = node1
            .wait_for_peer_identity(&channel_id, Duration::from_secs(2))
            .await?;
        assert_eq!(target_peer_id, node2_peer_id);

        // node1 sends a signed message → node2 authenticates → PeerConnected fires on node2
        node1
            .send_message(&target_peer_id, "test-topic", b"hello".to_vec(), &[])
            .await?;

        // Check for PeerConnected event on node2
        let event = timeout(Duration::from_secs(2), async {
            loop {
                match events.recv().await {
                    Ok(P2PEvent::PeerConnected(id, _)) => return Ok(id),
                    Ok(P2PEvent::Message { .. }) => continue, // skip messages
                    Ok(_) => continue,
                    Err(e) => return Err(e),
                }
            }
        })
        .await;
        assert!(event.is_ok(), "Should receive PeerConnected event");
        let connected_peer_id = event.expect("Timed out").expect("Channel error");
        // The connected peer ID should be node1's app-level ID (a valid PeerId)
        assert!(
            connected_peer_id.0.iter().any(|&b| b != 0),
            "PeerConnected should carry a non-zero peer ID"
        );

        node1.stop().await?;
        node2.stop().await?;

        Ok(())
    }

    // TODO(windows): Same QUIC connection issues as test_event_subscription
    #[cfg_attr(target_os = "windows", ignore)]
    #[tokio::test]
    async fn test_message_sending() -> Result<()> {
        // Create two nodes (IPv4-only loopback)
        let mut config1 = create_test_node_config();
        config1.ipv6 = false;
        let node1 = P2PNode::new(config1).await?;
        node1.start().await?;

        let mut config2 = create_test_node_config();
        config2.ipv6 = false;
        let node2 = P2PNode::new(config2).await?;
        node2.start().await?;

        // Wait a bit for nodes to start listening
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

        // Get actual listening address of node2
        let node2_addr = node2.local_addr().ok_or_else(|| {
            P2PError::Network(crate::error::NetworkError::ProtocolError(
                "No listening address".to_string().into(),
            ))
        })?;

        // Connect node1 to node2
        let channel_id =
            match timeout(Duration::from_millis(500), node1.connect_peer(&node2_addr)).await {
                Ok(res) => res?,
                Err(_) => return Err(P2PError::Network(NetworkError::Timeout)),
            };

        // Wait for identity exchange via wait_for_peer_identity.
        let target_peer_id = node1
            .wait_for_peer_identity(&channel_id, Duration::from_secs(2))
            .await?;
        assert_eq!(target_peer_id, node2.peer_id().clone());

        // Send a message
        let message_data = b"Hello, peer!".to_vec();
        let result = match timeout(
            Duration::from_millis(500),
            node1.send_message(&target_peer_id, "test-protocol", message_data, &[]),
        )
        .await
        {
            Ok(res) => res,
            Err(_) => return Err(P2PError::Network(NetworkError::Timeout)),
        };
        // For now, we'll just check that we don't get a "not connected" error
        // The actual send might fail due to no handler on the other side
        if let Err(e) = &result {
            assert!(!e.to_string().contains("not connected"), "Got error: {}", e);
        }

        // Try to send to non-existent peer
        let non_existent_peer = PeerId::from_bytes([0xFFu8; 32]);
        let result = node1
            .send_message(&non_existent_peer, "test-protocol", vec![], &[])
            .await;
        assert!(result.is_err(), "Sending to non-existent peer should fail");

        node1.stop().await?;
        node2.stop().await?;

        Ok(())
    }

    #[tokio::test]
    async fn test_remote_mcp_operations() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // MCP removed; test reduced to simple start/stop
        node.start().await?;
        node.stop().await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_health_check() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Health check should pass with no connections
        let result = node.health_check().await;
        assert!(result.is_ok());

        // Note: We're not actually connecting to real peers here
        // since that would require running bootstrap nodes.
        // The health check should still pass with no connections.

        Ok(())
    }

    #[tokio::test]
    async fn test_node_uptime() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        let uptime1 = node.uptime();
        assert!(uptime1 >= Duration::from_secs(0));

        // Wait a bit
        tokio::time::sleep(Duration::from_millis(10)).await;

        let uptime2 = node.uptime();
        assert!(uptime2 > uptime1);

        Ok(())
    }

    #[tokio::test]
    async fn test_node_config_access() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        let node_config = node.config();
        assert_eq!(node_config.max_connections, 100);
        // MCP removed

        Ok(())
    }

    #[tokio::test]
    async fn test_mcp_server_access() -> Result<()> {
        let config = create_test_node_config();
        let _node = P2PNode::new(config).await?;

        // MCP removed
        Ok(())
    }

    #[tokio::test]
    async fn test_dht_access() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // DHT is always available
        let _dht = node.dht();

        Ok(())
    }

    #[tokio::test]
    async fn test_node_config_builder() -> Result<()> {
        let bootstrap: MultiAddr = "/ip4/127.0.0.1/udp/9000/quic".parse().unwrap();

        let config = NodeConfig::builder()
            .local(true)
            .ipv6(true)
            .bootstrap_peer(bootstrap)
            .connection_timeout(Duration::from_secs(15))
            .max_connections(200)
            .max_message_size(TEST_MAX_MESSAGE_SIZE)
            .build()?;

        assert_eq!(config.listen_addrs().len(), 2); // IPv4 + IPv6
        assert!(config.local);
        assert!(config.ipv6);
        assert_eq!(config.bootstrap_peers.len(), 1);
        assert_eq!(config.connection_timeout, Duration::from_secs(15));
        assert_eq!(config.max_connections, 200);
        assert_eq!(config.max_message_size, Some(TEST_MAX_MESSAGE_SIZE));
        assert!(config.allow_loopback); // auto-enabled by local(true)

        Ok(())
    }

    #[tokio::test]
    async fn test_bootstrap_peers() -> Result<()> {
        let mut config = create_test_node_config();
        config.bootstrap_peers = vec![
            crate::MultiAddr::from_ipv4(std::net::Ipv4Addr::LOCALHOST, 9200),
            crate::MultiAddr::from_ipv4(std::net::Ipv4Addr::LOCALHOST, 9201),
        ];

        let node = P2PNode::new(config).await?;

        // Start node (which attempts to connect to bootstrap peers)
        node.start().await?;

        // In a test environment, bootstrap peers may not be available
        // The test verifies the node starts correctly with bootstrap configuration
        // Peer count may include local/internal tracking, so we just verify it's reasonable
        let _peer_count = node.peer_count().await;

        node.stop().await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_peer_info_structure() {
        let peer_info = PeerInfo {
            channel_id: "test_peer".to_string(),
            addresses: vec!["/ip4/127.0.0.1/tcp/9000".parse::<MultiAddr>().unwrap()],
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        assert_eq!(peer_info.channel_id, "test_peer");
        assert_eq!(peer_info.addresses.len(), 1);
        assert_eq!(peer_info.status, ConnectionStatus::Connected);
        assert_eq!(peer_info.protocols.len(), 1);
    }

    #[tokio::test]
    async fn test_serialization() -> Result<()> {
        // Test that configs can be serialized/deserialized
        let config = create_test_node_config();
        let serialized = serde_json::to_string(&config)?;
        let deserialized: NodeConfig = serde_json::from_str(&serialized)?;

        assert_eq!(config.local, deserialized.local);
        assert_eq!(config.port, deserialized.port);
        assert_eq!(config.ipv6, deserialized.ipv6);
        assert_eq!(config.bootstrap_peers, deserialized.bootstrap_peers);

        Ok(())
    }

    #[tokio::test]
    async fn test_get_channel_id_by_address_found() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Manually insert a peer for testing
        let test_channel_id = "peer_test_123".to_string();
        let test_address = "192.168.1.100:9000";
        let test_multiaddr = MultiAddr::quic(test_address.parse().unwrap());

        let peer_info = PeerInfo {
            channel_id: test_channel_id.clone(),
            addresses: vec![test_multiaddr],
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        node.transport
            .inject_peer(test_channel_id.clone(), peer_info)
            .await;

        // Test: Find channel by address
        let lookup_addr = MultiAddr::quic(test_address.parse().unwrap());
        let found_channel_id = node.get_channel_id_by_address(&lookup_addr).await;
        assert_eq!(found_channel_id, Some(test_channel_id));

        Ok(())
    }

    #[tokio::test]
    async fn test_get_channel_id_by_address_not_found() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Test: Try to find a channel that doesn't exist
        let unknown_addr = MultiAddr::quic("192.168.1.200:9000".parse().unwrap());
        let result = node.get_channel_id_by_address(&unknown_addr).await;
        assert_eq!(result, None);

        Ok(())
    }

    #[tokio::test]
    async fn test_get_channel_id_by_address_invalid_format() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Test: Non-IP address should return None (no matching socket addr)
        let ble_addr = MultiAddr::new(crate::address::TransportAddr::Ble {
            mac: [0x02, 0x00, 0x00, 0x00, 0x00, 0x01],
            psm: 0x0025,
        });
        let result = node.get_channel_id_by_address(&ble_addr).await;
        assert_eq!(result, None);

        Ok(())
    }

    #[tokio::test]
    async fn test_get_channel_id_by_address_multiple_peers() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Add multiple peers with different addresses
        let peer1_id = "peer_1".to_string();
        let peer1_addr_str = "192.168.1.101:9001";
        let peer1_multiaddr = MultiAddr::quic(peer1_addr_str.parse().unwrap());

        let peer2_id = "peer_2".to_string();
        let peer2_addr_str = "192.168.1.102:9002";
        let peer2_multiaddr = MultiAddr::quic(peer2_addr_str.parse().unwrap());

        let peer1_info = PeerInfo {
            channel_id: peer1_id.clone(),
            addresses: vec![peer1_multiaddr],
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        let peer2_info = PeerInfo {
            channel_id: peer2_id.clone(),
            addresses: vec![peer2_multiaddr],
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        node.transport
            .inject_peer(peer1_id.clone(), peer1_info)
            .await;
        node.transport
            .inject_peer(peer2_id.clone(), peer2_info)
            .await;

        // Test: Find each channel by their unique address
        let found_peer1 = node
            .get_channel_id_by_address(&MultiAddr::quic(peer1_addr_str.parse().unwrap()))
            .await;
        let found_peer2 = node
            .get_channel_id_by_address(&MultiAddr::quic(peer2_addr_str.parse().unwrap()))
            .await;

        assert_eq!(found_peer1, Some(peer1_id));
        assert_eq!(found_peer2, Some(peer2_id));

        Ok(())
    }

    #[tokio::test]
    async fn test_list_active_connections_empty() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Test: No connections initially
        let connections = node.list_active_connections().await;
        assert!(connections.is_empty());

        Ok(())
    }

    #[tokio::test]
    async fn test_list_active_connections_with_peers() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Add multiple peers
        let peer1_id = "peer_1".to_string();
        let peer1_addrs = vec![
            MultiAddr::quic("192.168.1.101:9001".parse().unwrap()),
            MultiAddr::quic("192.168.1.101:9002".parse().unwrap()),
        ];

        let peer2_id = "peer_2".to_string();
        let peer2_addrs = vec![MultiAddr::quic("192.168.1.102:9003".parse().unwrap())];

        let peer1_info = PeerInfo {
            channel_id: peer1_id.clone(),
            addresses: peer1_addrs.clone(),
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        let peer2_info = PeerInfo {
            channel_id: peer2_id.clone(),
            addresses: peer2_addrs.clone(),
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        node.transport
            .inject_peer(peer1_id.clone(), peer1_info)
            .await;
        node.transport
            .inject_peer(peer2_id.clone(), peer2_info)
            .await;

        // Also add to active_connections (list_active_connections iterates over this)
        node.transport
            .inject_active_connection(peer1_id.clone())
            .await;
        node.transport
            .inject_active_connection(peer2_id.clone())
            .await;

        // Test: List all active connections
        let connections = node.list_active_connections().await;
        assert_eq!(connections.len(), 2);

        // Verify peer1 and peer2 are in the list
        let peer1_conn = connections.iter().find(|(id, _)| id == &peer1_id);
        let peer2_conn = connections.iter().find(|(id, _)| id == &peer2_id);

        assert!(peer1_conn.is_some());
        assert!(peer2_conn.is_some());

        // Verify addresses match
        assert_eq!(peer1_conn.unwrap().1, peer1_addrs);
        assert_eq!(peer2_conn.unwrap().1, peer2_addrs);

        Ok(())
    }

    #[tokio::test]
    async fn test_remove_channel_success() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Add a peer
        let channel_id = "peer_to_remove".to_string();
        let channel_peer_id = PeerId::from_name(&channel_id);
        let peer_info = PeerInfo {
            channel_id: channel_id.clone(),
            addresses: vec![MultiAddr::quic("192.168.1.100:9000".parse().unwrap())],
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        node.transport
            .inject_peer(channel_id.clone(), peer_info)
            .await;
        node.transport
            .inject_peer_to_channel(channel_peer_id, channel_id.clone())
            .await;

        // Verify peer exists
        assert!(node.is_peer_connected(&channel_peer_id).await);

        // Remove the channel
        let removed = node.remove_channel(&channel_id).await;
        assert!(removed);

        // Verify peer no longer exists
        assert!(!node.is_peer_connected(&channel_peer_id).await);

        Ok(())
    }

    #[tokio::test]
    async fn test_remove_channel_nonexistent() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        // Try to remove a channel that doesn't exist
        let removed = node.remove_channel("nonexistent_peer").await;
        assert!(!removed);

        Ok(())
    }

    #[tokio::test]
    async fn test_is_peer_connected() -> Result<()> {
        let config = create_test_node_config();
        let node = P2PNode::new(config).await?;

        let channel_id = "test_peer".to_string();
        let channel_peer_id = PeerId::from_name(&channel_id);

        // Initially not connected
        assert!(!node.is_peer_connected(&channel_peer_id).await);

        // Add peer
        let peer_info = PeerInfo {
            channel_id: channel_id.clone(),
            addresses: vec![MultiAddr::quic("192.168.1.100:9000".parse().unwrap())],
            connected_at: Instant::now(),
            last_seen: Instant::now(),
            status: ConnectionStatus::Connected,
            protocols: vec!["test-protocol".to_string()],
            heartbeat_count: 0,
        };

        node.transport
            .inject_peer(channel_id.clone(), peer_info)
            .await;
        node.transport
            .inject_peer_to_channel(channel_peer_id, channel_id.clone())
            .await;

        // Now connected
        assert!(node.is_peer_connected(&channel_peer_id).await);

        // Remove channel
        node.remove_channel(&channel_id).await;

        // No longer connected
        assert!(!node.is_peer_connected(&channel_peer_id).await);

        Ok(())
    }

    #[test]
    fn test_normalize_ipv6_wildcard() {
        use std::net::{IpAddr, Ipv6Addr, SocketAddr};

        let wildcard = SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 8080);
        let normalized = normalize_wildcard_to_loopback(wildcard);

        assert_eq!(normalized.ip(), IpAddr::V6(Ipv6Addr::LOCALHOST));
        assert_eq!(normalized.port(), 8080);
    }

    #[test]
    fn test_normalize_ipv4_wildcard() {
        use std::net::{IpAddr, Ipv4Addr, SocketAddr};

        let wildcard = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 9000);
        let normalized = normalize_wildcard_to_loopback(wildcard);

        assert_eq!(normalized.ip(), IpAddr::V4(Ipv4Addr::LOCALHOST));
        assert_eq!(normalized.port(), 9000);
    }

    #[test]
    fn test_normalize_specific_address_unchanged() {
        let specific: std::net::SocketAddr = "192.168.1.100:3000".parse().unwrap();
        let normalized = normalize_wildcard_to_loopback(specific);

        assert_eq!(normalized, specific);
    }

    #[test]
    fn test_normalize_loopback_unchanged() {
        use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};

        let loopback_v6 = SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 5000);
        let normalized_v6 = normalize_wildcard_to_loopback(loopback_v6);
        assert_eq!(normalized_v6, loopback_v6);

        let loopback_v4 = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 5000);
        let normalized_v4 = normalize_wildcard_to_loopback(loopback_v4);
        assert_eq!(normalized_v4, loopback_v4);
    }

    // ---- parse_protocol_message regression tests ----

    /// Get current Unix timestamp for tests
    fn current_timestamp() -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0)
    }

    /// Helper to create a postcard-serialized unsigned WireMessage for tests
    fn make_wire_bytes(protocol: &str, data: Vec<u8>, from: &str, timestamp: u64) -> Vec<u8> {
        let msg = WireMessage {
            protocol: protocol.to_string(),
            data,
            from: PeerId::from_name(from),
            timestamp,
            user_agent: String::new(),
            public_key: Vec::new(),
            signature: Vec::new(),
        };
        postcard::to_stdvec(&msg).unwrap()
    }

    /// Helper to create a postcard-serialized signed WireMessage for tests.
    fn make_signed_wire_bytes(
        identity: &NodeIdentity,
        protocol: &str,
        data: Vec<u8>,
        timestamp: u64,
    ) -> Vec<u8> {
        let from = *identity.peer_id();
        let user_agent = "test/1.0";
        let signable =
            postcard::to_stdvec(&(protocol, data.as_slice(), &from, timestamp, user_agent))
                .unwrap();
        let sig = identity.sign(&signable).expect("signing should succeed");
        let msg = WireMessage {
            protocol: protocol.to_string(),
            data,
            from,
            timestamp,
            user_agent: user_agent.to_string(),
            public_key: identity.public_key().as_bytes().to_vec(),
            signature: sig.as_bytes().to_vec(),
        };
        postcard::to_stdvec(&msg).unwrap()
    }

    #[test]
    fn test_parse_protocol_message_uses_transport_peer_id_as_source() {
        // Regression: For unsigned messages, P2PEvent::Message.source must be the
        // transport peer ID, NOT the "from" field from the wire message.
        let transport_id = "abcdef0123456789";
        let logical_id = "spoofed-logical-id";
        let bytes = make_wire_bytes("test/v1", vec![1, 2, 3], logical_id, current_timestamp());

        let parsed =
            parse_protocol_message(&bytes, transport_id).expect("valid message should parse");

        // Unsigned message: no authenticated node ID
        assert!(parsed.authenticated_node_id.is_none());

        match parsed.event {
            P2PEvent::Message {
                topic,
                source,
                transport_source,
                timestamp: _,
                data,
            } => {
                assert!(source.is_none(), "unsigned message source must be None");
                assert!(
                    transport_source.is_none(),
                    "non-socket transport source should not produce an IP transport address"
                );
                assert_eq!(topic, "test/v1");
                assert_eq!(data, vec![1u8, 2, 3]);
            }
            other => panic!("expected P2PEvent::Message, got {:?}", other),
        }
    }

    #[test]
    fn test_parse_protocol_message_rejects_invalid_bytes() {
        // Random bytes that are not valid bincode should be rejected
        assert!(parse_protocol_message(b"not valid bincode", "peer-id").is_none());
    }

    #[test]
    fn test_parse_protocol_message_rejects_truncated_message() {
        // A truncated bincode message should fail to deserialize
        let full_bytes = make_wire_bytes("test/v1", vec![1, 2, 3], "sender", current_timestamp());
        let truncated = &full_bytes[..full_bytes.len() / 2];
        assert!(parse_protocol_message(truncated, "peer-id").is_none());
    }

    #[test]
    fn test_parse_protocol_message_empty_payload() {
        let bytes = make_wire_bytes("ping", vec![], "sender", current_timestamp());

        let parsed = parse_protocol_message(&bytes, "transport-peer")
            .expect("valid message with empty data should parse");

        match parsed.event {
            P2PEvent::Message { data, .. } => assert!(data.is_empty()),
            other => panic!("expected P2PEvent::Message, got {:?}", other),
        }
    }

    #[test]
    fn test_parse_protocol_message_records_ip_transport_source() {
        let bytes = make_wire_bytes("ping", vec![1], "sender", current_timestamp());

        let parsed =
            parse_protocol_message(&bytes, "192.168.1.2:4567").expect("valid message should parse");

        match parsed.event {
            P2PEvent::Message {
                transport_source, ..
            } => {
                assert_eq!(
                    transport_source,
                    Some(MultiAddr::quic("192.168.1.2:4567".parse().unwrap()))
                );
            }
            other => panic!("expected P2PEvent::Message, got {:?}", other),
        }
    }

    #[test]
    fn test_parse_protocol_message_preserves_binary_payload() {
        // Verify that arbitrary byte values (including 0xFF, 0x00) survive round-trip
        let payload: Vec<u8> = (0..=255).collect();
        let bytes = make_wire_bytes("binary/v1", payload.clone(), "sender", current_timestamp());

        let parsed = parse_protocol_message(&bytes, "peer-id")
            .expect("valid message with full byte range should parse");

        match parsed.event {
            P2PEvent::Message { data, topic, .. } => {
                assert_eq!(topic, "binary/v1");
                assert_eq!(
                    data, payload,
                    "payload must survive bincode round-trip exactly"
                );
            }
            other => panic!("expected P2PEvent::Message, got {:?}", other),
        }
    }

    #[test]
    fn test_parse_signed_message_verifies_and_uses_node_id() {
        let identity = NodeIdentity::generate().expect("should generate identity");
        let protocol = "test/signed";
        let data: Vec<u8> = vec![10, 20, 30];
        // The `from` field must match the PeerId derived from the public key.
        let from = *identity.peer_id();
        let timestamp = current_timestamp();
        let user_agent = "test/1.0";

        // Compute signable bytes the same way create_protocol_message does
        let signable =
            postcard::to_stdvec(&(protocol, data.as_slice(), &from, timestamp, user_agent))
                .unwrap();
        let sig = identity.sign(&signable).expect("signing should succeed");

        let msg = WireMessage {
            protocol: protocol.to_string(),
            data: data.clone(),
            from,
            timestamp,
            user_agent: user_agent.to_string(),
            public_key: identity.public_key().as_bytes().to_vec(),
            signature: sig.as_bytes().to_vec(),
        };
        let bytes = postcard::to_stdvec(&msg).unwrap();

        let parsed =
            parse_protocol_message(&bytes, "transport-xyz").expect("signed message should parse");

        let expected_peer_id = *identity.peer_id();
        assert_eq!(
            parsed.authenticated_node_id.as_ref(),
            Some(&expected_peer_id)
        );

        match parsed.event {
            P2PEvent::Message { source, .. } => {
                assert_eq!(
                    source.as_ref(),
                    Some(&expected_peer_id),
                    "source should be the verified PeerId"
                );
            }
            other => panic!("expected P2PEvent::Message, got {:?}", other),
        }
    }

    #[test]
    fn test_parse_message_with_bad_signature_is_rejected() {
        let identity = NodeIdentity::generate().expect("should generate identity");
        let protocol = "test/bad-sig";
        let data: Vec<u8> = vec![1, 2, 3];
        let from = *identity.peer_id();
        let timestamp = current_timestamp();
        let user_agent = "test/1.0";

        // Sign correct signable bytes
        let signable =
            postcard::to_stdvec(&(protocol, data.as_slice(), &from, timestamp, user_agent))
                .unwrap();
        let sig = identity.sign(&signable).expect("signing should succeed");

        // Tamper with the data (signature was over [1,2,3], not [99,99,99])
        let msg = WireMessage {
            protocol: protocol.to_string(),
            data: vec![99, 99, 99],
            from,
            timestamp,
            user_agent: user_agent.to_string(),
            public_key: identity.public_key().as_bytes().to_vec(),
            signature: sig.as_bytes().to_vec(),
        };
        let bytes = postcard::to_stdvec(&msg).unwrap();

        assert!(
            parse_protocol_message(&bytes, "transport-xyz").is_none(),
            "message with bad signature should be rejected"
        );
    }

    #[test]
    fn test_parse_message_with_mismatched_from_is_rejected() {
        let identity = NodeIdentity::generate().expect("should generate identity");
        let protocol = "test/from-mismatch";
        let data: Vec<u8> = vec![1, 2, 3];
        // Use a `from` field that does NOT match the public key's PeerId.
        let fake_from = PeerId::from_bytes([0xDE; 32]);
        let timestamp = current_timestamp();
        let user_agent = "test/1.0";

        let signable =
            postcard::to_stdvec(&(protocol, data.as_slice(), &fake_from, timestamp, user_agent))
                .unwrap();
        let sig = identity.sign(&signable).expect("signing should succeed");

        let msg = WireMessage {
            protocol: protocol.to_string(),
            data,
            from: fake_from,
            timestamp,
            user_agent: user_agent.to_string(),
            public_key: identity.public_key().as_bytes().to_vec(),
            signature: sig.as_bytes().to_vec(),
        };
        let bytes = postcard::to_stdvec(&msg).unwrap();

        assert!(
            parse_protocol_message(&bytes, "transport-xyz").is_none(),
            "message with mismatched from field should be rejected"
        );
    }

    #[test]
    fn test_parse_protocol_message_accepts_arbitrary_timestamps() {
        // Clock skew between peers must not drop messages.
        // Regression: previously ±5 min tolerance silently rejected all
        // traffic when client and node clocks differed.
        let payload = vec![1, 2, 3];

        // 10 hours in the past
        let old_ts = current_timestamp().saturating_sub(36_000);
        let old_bytes = make_wire_bytes("test/old", payload.clone(), "sender", old_ts);
        assert!(
            parse_protocol_message(&old_bytes, "peer-id").is_some(),
            "should accept unsigned message with timestamp 10h in the past"
        );

        // 10 hours in the future
        let future_ts = current_timestamp().saturating_add(36_000);
        let future_bytes = make_wire_bytes("test/future", payload.clone(), "sender", future_ts);
        assert!(
            parse_protocol_message(&future_bytes, "peer-id").is_some(),
            "should accept unsigned message with timestamp 10h in the future"
        );

        // Signed messages must take the same path: timestamp remains part of the
        // signed bytes for integrity, but is not used for wall-clock rejection.
        let identity = NodeIdentity::generate().expect("should generate identity");
        let signed_old =
            make_signed_wire_bytes(&identity, "test/signed-old", payload.clone(), old_ts);
        assert!(
            parse_protocol_message(&signed_old, "transport-xyz").is_some(),
            "should accept signed message with timestamp 10h in the past"
        );

        let signed_future =
            make_signed_wire_bytes(&identity, "test/signed-future", payload, future_ts);
        assert!(
            parse_protocol_message(&signed_future, "transport-xyz").is_some(),
            "should accept signed message with timestamp 10h in the future"
        );
    }

    #[test]
    fn test_parse_protocol_message_exposes_timestamp_on_event() {
        // After removing the wall-clock skew gate, the signed timestamp must
        // remain reachable on `P2PEvent::Message` so application-layer handlers
        // can implement freshness / replay defense.
        let ts: u64 = 1_234_567_890;
        let bytes = make_wire_bytes("test/ts", vec![9, 9, 9], "sender", ts);
        let parsed = parse_protocol_message(&bytes, "peer-id").expect("valid message should parse");
        match parsed.event {
            P2PEvent::Message { timestamp, .. } => {
                assert_eq!(timestamp, ts, "P2PEvent::Message.timestamp must round-trip");
            }
            other => panic!("expected P2PEvent::Message, got {:?}", other),
        }
    }

    #[test]
    fn test_signed_message_timestamp_is_signature_covered() {
        // Sign once, mutate only the timestamp, assert rejection. This is the
        // only timestamp property still enforced by `parse_protocol_message`
        // after the wall-clock gate was removed: signature integrity.
        let identity = NodeIdentity::generate().expect("should generate identity");
        let ts: u64 = 1_700_000_000;
        let signed = make_signed_wire_bytes(&identity, "test/sig", vec![1, 2, 3], ts);

        // Sanity: unmodified bytes parse and authenticate.
        let parsed = parse_protocol_message(&signed, "transport-xyz")
            .expect("unmodified signed message should parse");
        assert!(parsed.authenticated_node_id.is_some());

        // Now tamper with just the timestamp on the wire and re-serialize.
        let mut tampered: WireMessage =
            postcard::from_bytes(&signed).expect("signed bytes must deserialize");
        tampered.timestamp = ts.wrapping_add(1);
        let tampered_bytes = postcard::to_stdvec(&tampered).expect("re-serialize");

        assert!(
            parse_protocol_message(&tampered_bytes, "transport-xyz").is_none(),
            "timestamp-only mutation on a signed message must fail signature verification"
        );
    }

    fn snapshot_peer(addr: &str) -> SnapshotPeer {
        SnapshotPeer {
            peer_id: PeerId::random(),
            addresses: vec![addr.parse().expect("valid multiaddr")],
        }
    }

    #[test]
    fn snapshot_dial_sets_skips_self() {
        let self_id = PeerId::random();
        let mut peers = vec![snapshot_peer("/ip4/10.0.0.1/udp/9000/quic")];
        peers[0].peer_id = self_id;
        let mut seen = HashSet::new();

        assert!(snapshot_dial_sets(&peers, &self_id, &mut seen).is_empty());
        assert!(seen.is_empty(), "self must not reserve an address");
    }

    #[test]
    fn snapshot_dial_sets_skips_addresses_already_queued() {
        let self_id = PeerId::random();
        let queued = snapshot_peer("/ip4/10.0.0.2/udp/9000/quic");
        let fresh = snapshot_peer("/ip4/10.0.0.3/udp/9000/quic");
        let mut seen: HashSet<SocketAddr> = queued
            .addresses
            .iter()
            .filter_map(MultiAddr::dialable_socket_addr)
            .collect();

        let sets = snapshot_dial_sets(&[queued, fresh.clone()], &self_id, &mut seen);

        assert_eq!(
            sets.len(),
            1,
            "the already-queued peer must not be redialled"
        );
        assert_eq!(sets[0].0, fresh.peer_id);
        assert_eq!(
            seen.len(),
            2,
            "the new address is reserved for later phases"
        );
    }

    #[test]
    fn snapshot_dial_sets_drops_undialable_addresses() {
        let self_id = PeerId::random();
        let peers = vec![snapshot_peer("/ip4/10.0.0.4/tcp/9000")];
        let mut seen = HashSet::new();

        assert!(
            snapshot_dial_sets(&peers, &self_id, &mut seen).is_empty(),
            "only QUIC addresses are dialable today"
        );
    }

    #[test]
    fn snapshot_dial_sets_dedupes_repeated_peers_and_bounds_addresses() {
        let self_id = PeerId::random();
        let mut peer = snapshot_peer("/ip4/10.0.0.5/udp/9000/quic");
        peer.addresses = (0..6)
            .map(|i| {
                format!("/ip4/10.0.0.5/udp/900{i}/quic")
                    .parse()
                    .expect("valid multiaddr")
            })
            .collect();
        let mut seen = HashSet::new();

        let sets = snapshot_dial_sets(&[peer.clone(), peer], &self_id, &mut seen);

        assert_eq!(sets.len(), 1, "the same peer twice is one candidate");
        assert_eq!(
            sets[0].1.len(),
            MAX_SNAPSHOT_ADDRESSES_DIALLED,
            "the dial tail past the budget stays bounded"
        );
    }

    #[test]
    fn snapshot_dial_sets_returns_every_other_peer_once() {
        let self_id = PeerId::random();
        let peers: Vec<SnapshotPeer> = (0..130)
            .map(|i| snapshot_peer(&format!("/ip4/10.1.{}.{}/udp/9000/quic", i / 256, i % 256)))
            .collect();
        let mut seen = HashSet::new();

        let sets = snapshot_dial_sets(&peers, &self_id, &mut seen);

        assert_eq!(sets.len(), 130);
        assert_eq!(seen.len(), 130);
    }
}