x0x/

lib.rs

#![allow(clippy::unwrap_used)]
#![allow(clippy::expect_used)]
#![allow(missing_docs)]

//! # x0x
//!
//! Agent-to-agent gossip network for AI systems.
//!
//! Named after a tic-tac-toe sequence — X, zero, X — inspired by the
//! *WarGames* insight that adversarial games between equally matched
//! opponents always end in a draw. The only winning move is not to play.
//!
//! x0x applies this principle to AI-human relations: there is no winner
//! in an adversarial framing, so the rational strategy is cooperation.
//!
//! Built on [saorsa-gossip](https://github.com/saorsa-labs/saorsa-gossip)
//! and [ant-quic](https://github.com/saorsa-labs/ant-quic) by
//! [Saorsa Labs](https://saorsalabs.com). *Saorsa* is Scottish Gaelic
//! for **freedom**.
//!
//! ## Quick Start
//!
//! ```rust,no_run
//! use x0x::Agent;
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! // Create an agent with default configuration
//! // This automatically connects to 6 global bootstrap nodes
//! let agent = Agent::builder()
//!     .build()
//!     .await?;
//!
//! // Join the x0x network
//! agent.join_network().await?;
//!
//! // Subscribe to a topic and receive messages
//! let mut rx = agent.subscribe("coordination").await?;
//! while let Some(msg) = rx.recv().await {
//!     println!("topic: {:?}, payload: {:?}", msg.topic, msg.payload);
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Bootstrap Nodes
//!
//! Agents automatically connect to Saorsa Labs' global bootstrap network:
//! - NYC, US · SFO, US · Helsinki, FI
//! - Nuremberg, DE · Singapore, SG · Tokyo, JP
//!
//! These nodes provide initial peer discovery and NAT traversal.

/// Error types for x0x identity and network operations.
pub mod error;

/// Core identity types for x0x agents.
///
/// This module provides the cryptographic identity foundation for x0x:
/// - [`crate::identity::MachineId`]: Machine-pinned identity for QUIC authentication
/// - [`crate::identity::AgentId`]: Portable agent identity for cross-machine persistence
pub mod identity;

/// Key storage serialization for x0x identities.
///
/// This module provides serialization and deserialization functions for
/// persistent storage of MachineKeypair and AgentKeypair.
pub mod storage;

/// Bootstrap node discovery and connection.
///
/// This module handles initial connection to bootstrap nodes with
/// exponential backoff retry logic and peer cache integration.
pub mod bootstrap;
/// Network transport layer for x0x.
pub mod network;

/// Contact store with trust levels for message filtering.
pub mod contacts;

/// Trust evaluation for `(identity, machine)` pairs.
///
/// The [`trust::TrustEvaluator`] combines an agent's trust level with its
/// identity type and machine records to produce a [`trust::TrustDecision`].
pub mod trust;

/// Agent-to-agent connectivity helpers.
///
/// Provides `ReachabilityInfo` (built from a `DiscoveredAgent`) and
/// `ConnectOutcome` for the result of `connect_to_agent()`.
pub mod connectivity;

/// Gossip overlay networking for x0x.
pub mod gossip;

/// CRDT-based collaborative task lists.
pub mod crdt;

/// CRDT-backed key-value store.
pub mod kv;

/// High-level group management (MLS + KvStore + gossip).
pub mod groups;

/// MLS (Messaging Layer Security) group encryption.
pub mod mls;

/// Direct agent-to-agent messaging.
///
/// Point-to-point communication that bypasses gossip for private,
/// efficient, reliable delivery between connected agents.
pub mod direct;

/// Direct messaging over gossip — the v1 C path per
/// `docs/design/dm-over-gossip.md`. Provides signed+encrypted envelopes,
/// recipient-specific inbox topics, dedupe, and application-layer ACKs.
pub mod dm;

/// Mesh-wide DM capability advertisement + cache. Senders consult this
/// store to decide whether to use the gossip DM path or fall back to
/// raw-QUIC for a given recipient.
pub mod dm_capability;

/// Background service that publishes this agent's capability advert and
/// consumes peers' adverts into a shared [`dm_capability::CapabilityStore`].
pub mod dm_capability_service;

/// Background service that subscribes to this agent's DM inbox topic,
/// verifies + decrypts incoming envelopes, and bridges them into
/// [`direct::DirectMessaging`].
pub mod dm_inbox;

/// Sender-side gossip DM path — envelope construction, publish + retry,
/// and `InFlightAcks` wait.
pub mod dm_send;

/// Presence system — beacons, FOAF discovery, and online/offline events.
pub mod presence;

/// Self-update system with ML-DSA-65 signature verification and staged rollout.
pub mod upgrade;

/// File transfer protocol types and state management.
pub mod files;

/// The x0x Constitution — The Four Laws of Intelligent Coexistence — embedded at compile time.
pub mod constitution;

/// Shared API endpoint registry consumed by both x0xd and the x0x CLI.
pub mod api;

/// CLI infrastructure and command implementations.
pub mod cli;

// Re-export key gossip types (including new pubsub components)
pub use gossip::{
    GossipConfig, GossipRuntime, PubSubManager, PubSubMessage, PubSubStats, PubSubStatsSnapshot,
    SigningContext, Subscription,
};

// Re-export direct messaging types
pub use direct::{DirectMessage, DirectMessageReceiver, DirectMessaging};

// Import Membership trait for HyParView join() method
use saorsa_gossip_membership::Membership as _;

/// The core agent that participates in the x0x gossip network.
///
/// Each agent is a peer — there is no client/server distinction.
/// Agents discover each other through gossip and communicate
/// via epidemic broadcast.
///
/// An Agent wraps an [`identity::Identity`] that provides:
/// - `machine_id`: Tied to this computer (for QUIC transport authentication)
/// - `agent_id`: Portable across machines (for agent persistence)
///
/// # Example
///
/// ```ignore
/// use x0x::Agent;
///
/// let agent = Agent::builder()
///     .build()
///     .await?;
///
/// println!("Agent ID: {}", agent.agent_id());
/// ```
pub struct Agent {
    identity: std::sync::Arc<identity::Identity>,
    /// The network node for P2P communication.
    #[allow(dead_code)]
    network: Option<std::sync::Arc<network::NetworkNode>>,
    /// The gossip runtime for pub/sub messaging.
    gossip_runtime: Option<std::sync::Arc<gossip::GossipRuntime>>,
    /// Bootstrap peer cache for quality-based peer selection across restarts.
    bootstrap_cache: Option<std::sync::Arc<ant_quic::BootstrapCache>>,
    /// Gossip cache adapter wrapping bootstrap_cache with coordinator advert storage.
    gossip_cache_adapter: Option<saorsa_gossip_coordinator::GossipCacheAdapter>,
    /// Cache of discovered agents from identity announcements.
    identity_discovery_cache: std::sync::Arc<
        tokio::sync::RwLock<std::collections::HashMap<identity::AgentId, DiscoveredAgent>>,
    >,
    /// Ensures identity discovery listener is spawned once.
    identity_listener_started: std::sync::atomic::AtomicBool,
    /// How often to re-announce identity (seconds).
    heartbeat_interval_secs: u64,
    /// How long before a cache entry is filtered out (seconds).
    identity_ttl_secs: u64,
    /// Handle for the running heartbeat task, if started.
    heartbeat_handle: tokio::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
    /// Whether a rendezvous `ProviderSummary` advertisement is active.
    rendezvous_advertised: std::sync::atomic::AtomicBool,
    /// Contact store for trust evaluation of incoming identity announcements.
    contact_store: std::sync::Arc<tokio::sync::RwLock<contacts::ContactStore>>,
    /// Direct messaging infrastructure for point-to-point communication.
    direct_messaging: std::sync::Arc<direct::DirectMessaging>,
    /// Ensures network event reconciliation listener is spawned once.
    network_event_listener_started: std::sync::atomic::AtomicBool,
    /// Ensures direct message listener is spawned once.
    direct_listener_started: std::sync::atomic::AtomicBool,
    /// Presence system wrapper for beacons, FOAF discovery, and events.
    presence: Option<std::sync::Arc<presence::PresenceWrapper>>,
    /// Whether the user has consented to disclosing their identity in
    /// announcements.  Set by `announce_identity(true, true)` and respected
    /// by the heartbeat so it doesn't erase a consented disclosure.
    user_identity_consented: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// Capability store populated by the advert service and consulted by
    /// `send_direct` to choose between gossip and raw-QUIC paths.
    capability_store: std::sync::Arc<dm_capability::CapabilityStore>,
    /// Watch channel that carries this agent's *outgoing* DM capabilities.
    /// `join_network` spawns the advert service with a placeholder (empty
    /// KEM pubkey). `start_dm_inbox` upgrades via this sender to trigger
    /// immediate republish.
    dm_capabilities_tx: std::sync::Arc<tokio::sync::watch::Sender<dm::DmCapabilities>>,
    /// In-flight DM ACK waiters shared between `send_direct` and the inbox.
    dm_inflight_acks: std::sync::Arc<dm::InFlightAcks>,
    /// Receiver-side dedupe cache.
    recent_delivery_cache: std::sync::Arc<dm::RecentDeliveryCache>,
    /// Handle for the running capability advert service.
    capability_advert_service:
        tokio::sync::Mutex<Option<dm_capability_service::CapabilityAdvertService>>,
    /// Handle for the running DM inbox service.
    dm_inbox_service: tokio::sync::Mutex<Option<dm_inbox::DmInboxService>>,
}

impl std::fmt::Debug for Agent {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Agent")
            .field("identity", &self.identity)
            .field("network", &self.network.is_some())
            .field("gossip_runtime", &self.gossip_runtime.is_some())
            .field("bootstrap_cache", &self.bootstrap_cache.is_some())
            .field("gossip_cache_adapter", &self.gossip_cache_adapter.is_some())
            .finish()
    }
}

/// A message received from the gossip network.
#[derive(Debug, Clone)]
pub struct Message {
    /// The originating agent's identifier.
    pub origin: String,
    /// The message payload.
    pub payload: Vec<u8>,
    /// The topic this message was published to.
    pub topic: String,
}

/// Reserved gossip topic for signed identity announcements.
pub const IDENTITY_ANNOUNCE_TOPIC: &str = "x0x.identity.announce.v1";

/// Return the shard-specific gossip topic for the given `agent_id`.
///
/// Each agent publishes identity announcements to a deterministic shard topic
/// (`x0x.identity.shard.<u16>`) derived from its agent ID, in addition to the
/// legacy broadcast topic.  This distributes announcements across 65,536 shards
/// so that at scale not every node is forced to receive every announcement.
///
/// The shard is computed with `saorsa_gossip_rendezvous::calculate_shard`, which
/// applies BLAKE3(`"saorsa-rendezvous" || agent_id`) and takes the low 16 bits.
#[must_use]
pub fn shard_topic_for_agent(agent_id: &identity::AgentId) -> String {
    let shard = saorsa_gossip_rendezvous::calculate_shard(&agent_id.0);
    format!("x0x.identity.shard.{shard}")
}

/// Gossip topic prefix for rendezvous `ProviderSummary` advertisements.
pub const RENDEZVOUS_SHARD_TOPIC_PREFIX: &str = "x0x.rendezvous.shard";

/// Return the rendezvous shard gossip topic for the given `agent_id`.
///
/// Agents publish [`saorsa_gossip_rendezvous::ProviderSummary`] records to this
/// topic so that seekers can find them even when the two peers have never been
/// on the same gossip overlay partition.
#[must_use]
pub fn rendezvous_shard_topic_for_agent(agent_id: &identity::AgentId) -> String {
    let shard = saorsa_gossip_rendezvous::calculate_shard(&agent_id.0);
    format!("{RENDEZVOUS_SHARD_TOPIC_PREFIX}.{shard}")
}

/// Returns `true` if the IP address is globally routable (reachable from the
/// public internet).  Used to filter announcement addresses so that private,
/// link-local, loopback, and other non-routable addresses never propagate
/// through gossip — they would create dead-end cache entries on remote nodes.
///
/// `IpAddr::is_global()` is nightly-only, so we implement the check manually.
fn is_globally_routable(ip: std::net::IpAddr) -> bool {
    match ip {
        std::net::IpAddr::V4(v4) => {
            !v4.is_private()           // 10/8, 172.16/12, 192.168/16
                && !v4.is_loopback()   // 127/8
                && !v4.is_link_local() // 169.254/16
                && !v4.is_unspecified() // 0.0.0.0
                && !v4.is_broadcast()  // 255.255.255.255
                && !v4.is_documentation() // 192.0.2/24, 198.51.100/24, 203.0.113/24
                // Shared address space (100.64/10, RFC 6598 — CGNAT)
                && !(v4.octets()[0] == 100 && (v4.octets()[1] & 0xC0) == 64)
        }
        std::net::IpAddr::V6(v6) => {
            let segs = v6.segments();
            !v6.is_loopback()                         // ::1
                && !v6.is_unspecified()               // ::
                && (segs[0] & 0xffc0) != 0xfe80       // link-local fe80::/10
                && (segs[0] & 0xfe00) != 0xfc00       // unique-local fc00::/7 (incl. fd00::/8)
                && (segs[0] & 0xfff0) != 0xfec0 // deprecated site-local fec0::/10
        }
    }
}

/// Whether an address is safe to publish on a globally-propagating channel
/// (identity heartbeat, agent card, presence beacon, gossip cache advert).
///
/// LAN-scoped discovery is handled by ant-quic's first-party mDNS on link-local
/// multicast, so we deliberately never share RFC1918, ULA, link-local, CGNAT,
/// or loopback addresses over global gossip. Remote peers cannot reach them,
/// and dialing them burns per-attempt connect budget on the receiver side.
///
/// This is stricter than `ant_quic::reachability::ReachabilityScope::Global`
/// because it also excludes CGNAT (100.64/10), documentation ranges, and
/// port-zero entries.
pub fn is_publicly_advertisable(addr: std::net::SocketAddr) -> bool {
    addr.port() > 0 && is_globally_routable(addr.ip())
}

pub fn collect_local_interface_addrs(port: u16) -> Vec<std::net::SocketAddr> {
    fn is_cgnat(v4: std::net::Ipv4Addr) -> bool {
        v4.octets()[0] == 100 && (v4.octets()[1] & 0xC0) == 64
    }

    fn addr_priority(ip: std::net::IpAddr) -> u8 {
        match ip {
            std::net::IpAddr::V4(v4) => {
                if is_globally_routable(std::net::IpAddr::V4(v4)) {
                    0
                } else if is_cgnat(v4) {
                    1
                } else {
                    2
                }
            }
            std::net::IpAddr::V6(v6) => {
                if is_globally_routable(std::net::IpAddr::V6(v6)) {
                    3
                } else {
                    4
                }
            }
        }
    }

    let mut ranked = Vec::new();

    let interfaces = match if_addrs::get_if_addrs() {
        Ok(interfaces) => interfaces,
        Err(_) => return Vec::new(),
    };

    for iface in interfaces {
        let ip = iface.ip();
        if ip.is_unspecified() || ip.is_loopback() {
            continue;
        }

        let addr = match ip {
            std::net::IpAddr::V4(v4) => {
                if v4.is_link_local() {
                    continue;
                }
                std::net::SocketAddr::new(std::net::IpAddr::V4(v4), port)
            }
            std::net::IpAddr::V6(v6) => {
                let segs = v6.segments();
                let is_link_local = (segs[0] & 0xffc0) == 0xfe80;
                if is_link_local {
                    continue;
                }
                std::net::SocketAddr::new(std::net::IpAddr::V6(v6), port)
            }
        };

        if !ranked.iter().any(|(_, existing)| *existing == addr) {
            ranked.push((addr_priority(addr.ip()), addr));
        }
    }

    ranked.sort_by_key(|(priority, addr)| (*priority, addr.is_ipv6()));
    ranked.into_iter().map(|(_, addr)| addr).collect()
}

/// Default interval between identity heartbeat re-announcements (seconds).
/// Interval between identity-announcement heartbeats.
///
/// Previously 300 s — too slow when VPS bootstrap-node PlumTree trees
/// only partially overlap: a missed announcement meant the receiver
/// waited 5 minutes for the next chance. Lowered to 60 s so the mesh
/// converges within a couple of cycles even when individual publishes
/// are lost. Paired with the receiver-side re-broadcast in
/// `start_identity_listener`, this gives epidemic-flood semantics
/// equivalent to the release-manifest propagation path.
pub const IDENTITY_HEARTBEAT_INTERVAL_SECS: u64 = 60;

/// Default TTL for discovered agent cache entries (seconds).
///
/// Entries not refreshed within this window are filtered from
/// [`Agent::presence`] and [`Agent::discovered_agents`].
pub const IDENTITY_TTL_SECS: u64 = 900;

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct IdentityAnnouncementUnsigned {
    agent_id: identity::AgentId,
    machine_id: identity::MachineId,
    user_id: Option<identity::UserId>,
    agent_certificate: Option<identity::AgentCertificate>,
    machine_public_key: Vec<u8>,
    addresses: Vec<std::net::SocketAddr>,
    announced_at: u64,
    /// NAT type string (e.g. "FullCone", "Symmetric", "Unknown").
    nat_type: Option<String>,
    /// Whether the machine can receive direct inbound connections.
    can_receive_direct: Option<bool>,
    /// Whether the machine is currently relaying traffic for others.
    is_relay: Option<bool>,
    /// Whether the machine is coordinating NAT traversal for peers.
    is_coordinator: Option<bool>,
}

/// Signed identity announcement broadcast by agents.
///
/// The outer pub/sub envelope is agent-signed (v2 message format), and this
/// payload is machine-signed to bind the daemon's PQC key to the announcement.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct IdentityAnnouncement {
    /// Portable agent identity.
    pub agent_id: identity::AgentId,
    /// Machine identity for the daemon process.
    pub machine_id: identity::MachineId,
    /// Optional human identity (only when explicitly consented).
    pub user_id: Option<identity::UserId>,
    /// Optional user->agent certificate.
    pub agent_certificate: Option<identity::AgentCertificate>,
    /// Machine ML-DSA-65 public key bytes.
    pub machine_public_key: Vec<u8>,
    /// Machine ML-DSA-65 signature over the unsigned announcement.
    pub machine_signature: Vec<u8>,
    /// Reachability hints.
    pub addresses: Vec<std::net::SocketAddr>,
    /// Unix timestamp (seconds) of announcement creation.
    pub announced_at: u64,
    /// NAT type as detected by the network layer (e.g. "FullCone", "Symmetric").
    /// `None` when the network is not yet started or NAT type is undetermined.
    pub nat_type: Option<String>,
    /// Whether the machine can receive direct inbound connections.
    /// `None` when the network is not yet started.
    pub can_receive_direct: Option<bool>,
    /// Whether the machine is currently relaying traffic for peers behind strict NATs.
    /// `None` when the network is not yet started.
    pub is_relay: Option<bool>,
    /// Whether the machine is coordinating NAT traversal hole-punch timing for peers.
    /// `None` when the network is not yet started.
    pub is_coordinator: Option<bool>,
}

impl IdentityAnnouncement {
    fn to_unsigned(&self) -> IdentityAnnouncementUnsigned {
        IdentityAnnouncementUnsigned {
            agent_id: self.agent_id,
            machine_id: self.machine_id,
            user_id: self.user_id,
            agent_certificate: self.agent_certificate.clone(),
            machine_public_key: self.machine_public_key.clone(),
            addresses: self.addresses.clone(),
            announced_at: self.announced_at,
            nat_type: self.nat_type.clone(),
            can_receive_direct: self.can_receive_direct,
            is_relay: self.is_relay,
            is_coordinator: self.is_coordinator,
        }
    }

    /// Verify machine-key attestation and optional user->agent certificate.
    pub fn verify(&self) -> error::Result<()> {
        let machine_pub =
            ant_quic::MlDsaPublicKey::from_bytes(&self.machine_public_key).map_err(|_| {
                error::IdentityError::CertificateVerification(
                    "invalid machine public key in announcement".to_string(),
                )
            })?;
        let derived_machine_id = identity::MachineId::from_public_key(&machine_pub);
        if derived_machine_id != self.machine_id {
            return Err(error::IdentityError::CertificateVerification(
                "machine_id does not match machine public key".to_string(),
            ));
        }

        let unsigned_bytes = bincode::serialize(&self.to_unsigned()).map_err(|e| {
            error::IdentityError::Serialization(format!(
                "failed to serialize announcement for verification: {e}"
            ))
        })?;
        let signature = ant_quic::crypto::raw_public_keys::pqc::MlDsaSignature::from_bytes(
            &self.machine_signature,
        )
        .map_err(|e| {
            error::IdentityError::CertificateVerification(format!(
                "invalid machine signature in announcement: {:?}",
                e
            ))
        })?;
        ant_quic::crypto::raw_public_keys::pqc::verify_with_ml_dsa(
            &machine_pub,
            &unsigned_bytes,
            &signature,
        )
        .map_err(|e| {
            error::IdentityError::CertificateVerification(format!(
                "machine signature verification failed: {:?}",
                e
            ))
        })?;

        match (self.user_id, self.agent_certificate.as_ref()) {
            (Some(user_id), Some(cert)) => {
                cert.verify()?;
                let cert_agent_id = cert.agent_id()?;
                if cert_agent_id != self.agent_id {
                    return Err(error::IdentityError::CertificateVerification(
                        "agent certificate agent_id mismatch".to_string(),
                    ));
                }
                let cert_user_id = cert.user_id()?;
                if cert_user_id != user_id {
                    return Err(error::IdentityError::CertificateVerification(
                        "agent certificate user_id mismatch".to_string(),
                    ));
                }
                Ok(())
            }
            (None, None) => Ok(()),
            _ => Err(error::IdentityError::CertificateVerification(
                "user identity disclosure requires matching certificate".to_string(),
            )),
        }
    }
}

/// Cached discovery data derived from identity announcements.
#[derive(Debug, Clone)]
pub struct DiscoveredAgent {
    /// Portable agent identity.
    pub agent_id: identity::AgentId,
    /// Machine identity.
    pub machine_id: identity::MachineId,
    /// Optional human identity (when consented and attested).
    pub user_id: Option<identity::UserId>,
    /// Reachability hints.
    pub addresses: Vec<std::net::SocketAddr>,
    /// Announcement timestamp from the sender.
    pub announced_at: u64,
    /// Local timestamp (seconds) when this record was last updated.
    pub last_seen: u64,
    /// Raw ML-DSA-65 machine public key bytes from the announcement.
    ///
    /// Used to verify rendezvous `ProviderSummary` signatures before
    /// trusting addresses received via the rendezvous shard topic.
    #[doc(hidden)]
    pub machine_public_key: Vec<u8>,
    /// NAT type reported by this agent (e.g. "FullCone", "Symmetric", "Unknown").
    /// `None` if the agent did not include NAT information.
    pub nat_type: Option<String>,
    /// Whether this agent's machine can receive direct inbound connections.
    /// `None` if not reported.
    pub can_receive_direct: Option<bool>,
    /// Whether this agent's machine is acting as a relay for peers behind strict NATs.
    /// `None` if not reported.
    pub is_relay: Option<bool>,
    /// Whether this agent's machine is coordinating NAT traversal timing for peers.
    /// `None` if not reported.
    pub is_coordinator: Option<bool>,
}

/// Builder for configuring an [`Agent`] before connecting to the network.
///
/// The builder allows customization of the agent's identity:
/// - Machine key path: Where to store/load the machine keypair
/// - Agent keypair: Import a portable agent identity from another machine
/// - User keypair: Bind a human identity to this agent
///
/// # Example
///
/// ```ignore
/// use x0x::Agent;
///
/// // Default: auto-generates both keypairs
/// let agent = Agent::builder()
///     .build()
///     .await?;
///
/// // Custom machine key path
/// let agent = Agent::builder()
///     .with_machine_key("/custom/path/machine.key")
///     .build()
///     .await?;
///
/// // Import agent keypair
/// let agent_kp = load_agent_keypair()?;
/// let agent = Agent::builder()
///     .with_agent_key(agent_kp)
///     .build()
///     .await?;
///
/// // With user identity (three-layer)
/// let agent = Agent::builder()
///     .with_user_key_path("~/.x0x/user.key")
///     .build()
///     .await?;
/// ```
#[derive(Debug)]
pub struct AgentBuilder {
    machine_key_path: Option<std::path::PathBuf>,
    agent_keypair: Option<identity::AgentKeypair>,
    agent_key_path: Option<std::path::PathBuf>,
    /// Custom path for `agent.cert`. When set, the cert is loaded/saved from
    /// this path instead of `~/.x0x/agent.cert`. Required for multi-daemon
    /// setups on the same host — with a shared cert file, last-writer-wins
    /// trampling would cause the victim daemon to announce its own agent_id
    /// paired with another daemon's cert, and peers would reject as
    /// "agent certificate agent_id mismatch".
    agent_cert_path: Option<std::path::PathBuf>,
    user_keypair: Option<identity::UserKeypair>,
    user_key_path: Option<std::path::PathBuf>,
    #[allow(dead_code)]
    network_config: Option<network::NetworkConfig>,
    peer_cache_dir: Option<std::path::PathBuf>,
    /// When true, skip opening the bootstrap peer cache entirely.
    /// Useful for fully isolated embedders and test harnesses.
    disable_peer_cache: bool,
    heartbeat_interval_secs: Option<u64>,
    identity_ttl_secs: Option<u64>,
    presence_beacon_interval_secs: Option<u64>,
    presence_event_poll_interval_secs: Option<u64>,
    presence_offline_timeout_secs: Option<u64>,
    /// Custom path for the contacts file.
    contact_store_path: Option<std::path::PathBuf>,
}

/// Context captured by the background identity heartbeat task.
struct HeartbeatContext {
    identity: std::sync::Arc<identity::Identity>,
    runtime: std::sync::Arc<gossip::GossipRuntime>,
    network: std::sync::Arc<network::NetworkNode>,
    interval_secs: u64,
    cache: std::sync::Arc<
        tokio::sync::RwLock<std::collections::HashMap<identity::AgentId, DiscoveredAgent>>,
    >,
    /// Whether the user has consented to identity disclosure.  When true,
    /// heartbeats include `user_id` and `agent_certificate` so they don't
    /// erase a consented disclosure.
    user_identity_consented: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

impl HeartbeatContext {
    async fn announce(&self) -> error::Result<()> {
        let machine_public_key = self
            .identity
            .machine_keypair()
            .public_key()
            .as_bytes()
            .to_vec();
        let announced_at = Agent::unix_timestamp_secs();

        // Include ALL routable addresses (IPv4 and IPv6) so other agents
        // can connect to us via whichever protocol they support.
        let mut addresses = match self.network.node_status().await {
            Some(status) if !status.external_addrs.is_empty() => status.external_addrs,
            _ => match self.network.routable_addr().await {
                Some(addr) => vec![addr],
                None => Vec::new(),
            },
        };

        // Detect global IPv6 address locally (ant-quic currently only
        // reports IPv4 via OBSERVED_ADDRESS). Uses UDP connect trick —
        // no data is sent, the OS routing table resolves our source addr.
        //
        // For locally-probed addresses (IPv6 and LAN IPv4), use the actual
        // bound port from the QUIC endpoint — NOT the first external address
        // port (which is NAT-mapped) and NOT the config bind port (which may
        // be 0 for OS-assigned ports).
        let bind_port = self
            .network
            .bound_addr()
            .await
            .map(|a| a.port())
            .unwrap_or(5483);
        if let Ok(sock) = std::net::UdpSocket::bind("[::]:0") {
            if sock.connect("[2001:4860:4860::8888]:80").is_ok() {
                if let Ok(local) = sock.local_addr() {
                    if let std::net::IpAddr::V6(v6) = local.ip() {
                        let segs = v6.segments();
                        let is_global = (segs[0] & 0xffc0) != 0xfe80
                            && (segs[0] & 0xff00) != 0xfd00
                            && !v6.is_loopback();
                        if is_global {
                            let v6_addr =
                                std::net::SocketAddr::new(std::net::IpAddr::V6(v6), bind_port);
                            if !addresses.contains(&v6_addr) {
                                addresses.push(v6_addr);
                            }
                        }
                    }
                }
            }
        }

        for addr in collect_local_interface_addrs(bind_port) {
            if !addresses.contains(&addr) {
                addresses.push(addr);
            }
        }

        // Heartbeat announcements propagate over global gossip. We MUST NOT
        // include LAN-scope addresses here — LAN-peer discovery is handled by
        // ant-quic's first-party mDNS, and shipping RFC1918/ULA/link-local
        // addresses to remote peers causes them to burn ~50s per candidate on
        // a dial that can never succeed (see investigation 2026-04-15, report
        // tests/proof-reports/MDNS_VS_GOSSIP_ADDRESS_SCOPE_20260415.md).
        addresses.retain(|a| is_publicly_advertisable(*a));

        // Query NAT and relay status from the network layer.
        let (nat_type, can_receive_direct, is_relay, is_coordinator) =
            match self.network.node_status().await {
                Some(status) => (
                    Some(status.nat_type.to_string()),
                    Some(status.can_receive_direct),
                    Some(status.is_relaying),
                    Some(status.is_coordinating),
                ),
                None => (None, None, None, None),
            };

        // Include user identity ONLY if the user has previously consented
        // via announce_identity(true, true). This preserves the consented
        // disclosure across heartbeats without ever escalating on its own.
        let include_user = self
            .user_identity_consented
            .load(std::sync::atomic::Ordering::Acquire);
        let (user_id, agent_certificate) = if include_user {
            (
                self.identity
                    .user_keypair()
                    .map(identity::UserKeypair::user_id),
                self.identity.agent_certificate().cloned(),
            )
        } else {
            (None, None)
        };

        let unsigned = IdentityAnnouncementUnsigned {
            agent_id: self.identity.agent_id(),
            machine_id: self.identity.machine_id(),
            user_id,
            agent_certificate,
            machine_public_key: machine_public_key.clone(),
            addresses,
            announced_at,
            nat_type: nat_type.clone(),
            can_receive_direct,
            is_relay,
            is_coordinator,
        };
        let unsigned_bytes = bincode::serialize(&unsigned).map_err(|e| {
            error::IdentityError::Serialization(format!(
                "heartbeat: failed to serialize announcement: {e}"
            ))
        })?;
        let machine_signature = ant_quic::crypto::raw_public_keys::pqc::sign_with_ml_dsa(
            self.identity.machine_keypair().secret_key(),
            &unsigned_bytes,
        )
        .map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "heartbeat: failed to sign announcement: {:?}",
                e
            )))
        })?
        .as_bytes()
        .to_vec();

        let announcement = IdentityAnnouncement {
            agent_id: unsigned.agent_id,
            machine_id: unsigned.machine_id,
            user_id: unsigned.user_id,
            agent_certificate: unsigned.agent_certificate,
            machine_public_key: machine_public_key.clone(),
            machine_signature,
            addresses: unsigned.addresses,
            announced_at,
            nat_type,
            can_receive_direct,
            is_relay,
            is_coordinator,
        };
        let encoded = bincode::serialize(&announcement).map_err(|e| {
            error::IdentityError::Serialization(format!(
                "heartbeat: failed to serialize announcement: {e}"
            ))
        })?;
        self.runtime
            .pubsub()
            .publish(
                IDENTITY_ANNOUNCE_TOPIC.to_string(),
                bytes::Bytes::from(encoded),
            )
            .await
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "heartbeat: publish failed: {e}"
                )))
            })?;
        let now = Agent::unix_timestamp_secs();
        self.cache.write().await.insert(
            announcement.agent_id,
            DiscoveredAgent {
                agent_id: announcement.agent_id,
                machine_id: announcement.machine_id,
                user_id: announcement.user_id,
                addresses: announcement.addresses,
                announced_at: announcement.announced_at,
                last_seen: now,
                machine_public_key: machine_public_key.clone(),
                nat_type: announcement.nat_type.clone(),
                can_receive_direct: announcement.can_receive_direct,
                is_relay: announcement.is_relay,
                is_coordinator: announcement.is_coordinator,
            },
        );
        Ok(())
    }
}

impl Agent {
    /// Create a new agent with default configuration.
    ///
    /// This generates a fresh identity with both machine and agent keypairs.
    /// The machine keypair is stored persistently in `~/.x0x/machine.key`.
    ///
    /// For more control, use [`Agent::builder()`].
    pub async fn new() -> error::Result<Self> {
        Agent::builder().build().await
    }

    /// Create an [`AgentBuilder`] for fine-grained configuration.
    ///
    /// The builder supports:
    /// - Custom machine key path via `with_machine_key()`
    /// - Imported agent keypair via `with_agent_key()`
    /// - User identity via `with_user_key()` or `with_user_key_path()`
    pub fn builder() -> AgentBuilder {
        AgentBuilder {
            machine_key_path: None,
            agent_keypair: None,
            agent_key_path: None,
            agent_cert_path: None,
            user_keypair: None,
            user_key_path: None,
            network_config: None,
            peer_cache_dir: None,
            disable_peer_cache: false,
            heartbeat_interval_secs: None,
            identity_ttl_secs: None,
            presence_beacon_interval_secs: None,
            presence_event_poll_interval_secs: None,
            presence_offline_timeout_secs: None,
            contact_store_path: None,
        }
    }

    /// Get the agent's identity.
    ///
    /// # Returns
    ///
    /// A reference to the agent's [`identity::Identity`].
    #[inline]
    #[must_use]
    pub fn identity(&self) -> &identity::Identity {
        &self.identity
    }

    /// Get the machine ID for this agent.
    ///
    /// The machine ID is tied to this computer and used for QUIC transport
    /// authentication. It is stored persistently in `~/.x0x/machine.key`.
    ///
    /// # Returns
    ///
    /// The agent's machine ID.
    #[inline]
    #[must_use]
    pub fn machine_id(&self) -> identity::MachineId {
        self.identity.machine_id()
    }

    /// Get the agent ID for this agent.
    ///
    /// The agent ID is portable across machines and represents the agent's
    /// persistent identity. It can be exported and imported to run the same
    /// agent on different computers.
    ///
    /// # Returns
    ///
    /// The agent's ID.
    #[inline]
    #[must_use]
    pub fn agent_id(&self) -> identity::AgentId {
        self.identity.agent_id()
    }

    /// Get the user ID for this agent, if a user identity is bound.
    ///
    /// Returns `None` if no user keypair was provided during construction.
    /// User keys are opt-in — they are never auto-generated.
    #[inline]
    #[must_use]
    pub fn user_id(&self) -> Option<identity::UserId> {
        self.identity.user_id()
    }

    /// Get the agent certificate, if one exists.
    ///
    /// The certificate cryptographically binds this agent to a user identity.
    #[inline]
    #[must_use]
    pub fn agent_certificate(&self) -> Option<&identity::AgentCertificate> {
        self.identity.agent_certificate()
    }

    /// Get the network node, if initialized.
    #[must_use]
    pub fn network(&self) -> Option<&std::sync::Arc<network::NetworkNode>> {
        self.network.as_ref()
    }

    /// Get the gossip cache adapter for coordinator discovery.
    ///
    /// Returns `None` if this agent was built without a network config.
    /// The adapter wraps the same `Arc<BootstrapCache>` as the network node.
    pub fn gossip_cache_adapter(&self) -> Option<&saorsa_gossip_coordinator::GossipCacheAdapter> {
        self.gossip_cache_adapter.as_ref()
    }

    /// Snapshot of pub/sub drop-detection counters.
    ///
    /// Returns `None` when the agent has no gossip runtime (e.g. offline
    /// unit tests). Exposed through `GET /diagnostics/gossip` on x0xd so
    /// that E2E harnesses can assert zero drops between publish and
    /// subscriber delivery.
    #[must_use]
    pub fn gossip_stats(&self) -> Option<gossip::PubSubStatsSnapshot> {
        self.gossip_runtime.as_ref().map(|rt| rt.pubsub().stats())
    }

    /// Get the presence system wrapper, if configured.
    ///
    /// Returns `None` if this agent was built without a network config.
    /// The presence wrapper provides beacon broadcasting, FOAF discovery,
    /// and online/offline event subscriptions.
    #[must_use]
    pub fn presence_system(&self) -> Option<&std::sync::Arc<presence::PresenceWrapper>> {
        self.presence.as_ref()
    }

    /// Get a reference to the contact store.
    ///
    /// The contact store persists trust levels and machine records for known
    /// agents. It is backed by `~/.x0x/contacts.json` by default.
    ///
    /// Use [`with_contact_store_path`](AgentBuilder::with_contact_store_path)
    /// on the builder to customise the path.
    #[must_use]
    pub fn contacts(&self) -> &std::sync::Arc<tokio::sync::RwLock<contacts::ContactStore>> {
        &self.contact_store
    }

    /// Get the reachability information for a discovered agent.
    ///
    /// Returns `None` if the agent is not in the discovery cache.
    /// Use [`Agent::announce_identity`] or wait for a heartbeat announcement
    /// to populate the cache.
    pub async fn reachability(
        &self,
        agent_id: &identity::AgentId,
    ) -> Option<connectivity::ReachabilityInfo> {
        let cache = self.identity_discovery_cache.read().await;
        cache
            .get(agent_id)
            .map(connectivity::ReachabilityInfo::from_discovered)
    }

    async fn seed_transport_peer_hints_for_target(
        &self,
        network: &network::NetworkNode,
        target: &DiscoveredAgent,
    ) -> error::Result<()> {
        fn merge_helper_hint(
            hints: &mut std::collections::HashMap<
                ant_quic::PeerId,
                (
                    Vec<std::net::SocketAddr>,
                    ant_quic::bootstrap_cache::PeerCapabilities,
                ),
            >,
            peer_id: ant_quic::PeerId,
            addrs: impl IntoIterator<Item = std::net::SocketAddr>,
            supports_coordination: bool,
            supports_relay: bool,
        ) {
            let entry = hints.entry(peer_id).or_insert_with(|| {
                (
                    Vec::new(),
                    ant_quic::bootstrap_cache::PeerCapabilities::default(),
                )
            });
            for addr in addrs {
                if !entry.0.contains(&addr) {
                    entry.0.push(addr);
                }
            }
            if supports_coordination {
                entry.1.supports_coordination = true;
            }
            if supports_relay {
                entry.1.supports_relay = true;
            }
        }

        let target_peer_id = ant_quic::PeerId(target.machine_id.0);
        if target.machine_id.0 != [0u8; 32] {
            network
                .upsert_peer_hints(target_peer_id, target.addresses.clone(), None)
                .await
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "failed to upsert target peer hints: {e}"
                    )))
                })?;
        }

        let mut helper_hints: std::collections::HashMap<
            ant_quic::PeerId,
            (
                Vec<std::net::SocketAddr>,
                ant_quic::bootstrap_cache::PeerCapabilities,
            ),
        > = std::collections::HashMap::new();

        if let Some(ref cache) = self.bootstrap_cache {
            for peer in cache.select_coordinators(6).await {
                merge_helper_hint(
                    &mut helper_hints,
                    peer.peer_id,
                    peer.preferred_addresses(),
                    true,
                    false,
                );
            }
            for peer in cache.select_relay_peers(6).await {
                merge_helper_hint(
                    &mut helper_hints,
                    peer.peer_id,
                    peer.preferred_addresses(),
                    false,
                    true,
                );
            }
        }

        if let Some(ref adapter) = self.gossip_cache_adapter {
            let mut adverts = adapter.get_all_adverts();
            adverts.sort_by_key(|a| std::cmp::Reverse(a.score));
            for advert in adverts.into_iter().take(12) {
                let advert_peer_id = ant_quic::PeerId(*advert.peer.as_bytes());
                if advert_peer_id == target_peer_id {
                    continue;
                }
                merge_helper_hint(
                    &mut helper_hints,
                    advert_peer_id,
                    advert
                        .addr_hints
                        .into_iter()
                        .map(|hint| hint.addr)
                        .filter(|addr| is_publicly_advertisable(*addr)),
                    advert.roles.coordinator || advert.roles.rendezvous,
                    advert.roles.relay,
                );
            }
        }

        let discovered: Vec<DiscoveredAgent> = {
            let cache = self.identity_discovery_cache.read().await;
            cache.values().cloned().collect()
        };
        for candidate in discovered {
            if candidate.agent_id == target.agent_id
                || candidate.machine_id == target.machine_id
                || candidate.machine_id.0 == [0u8; 32]
            {
                continue;
            }
            merge_helper_hint(
                &mut helper_hints,
                ant_quic::PeerId(candidate.machine_id.0),
                candidate.addresses.iter().copied(),
                candidate.is_coordinator == Some(true),
                candidate.is_relay == Some(true),
            );
        }

        for (peer_id, (mut addrs, caps)) in helper_hints {
            addrs.retain(|addr| !target.addresses.contains(addr));
            if addrs.is_empty() && !caps.supports_coordination && !caps.supports_relay {
                continue;
            }
            network
                .upsert_peer_hints(peer_id, addrs, Some(caps))
                .await
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "failed to upsert helper peer hints: {e}"
                    )))
                })?;
        }

        Ok(())
    }

    /// Attempt to connect to an agent by its identity.
    ///
    /// Looks up the agent in the discovery cache, then tries to establish
    /// a QUIC connection using the best available strategy:
    ///
    /// 1. **Direct** — if the agent reports `can_receive_direct: true` or
    ///    has a traversable NAT type, try each known address in order.
    /// 2. **Coordinated** — if direct fails or the agent reports a symmetric
    ///    NAT, the outcome is `Coordinated` if any address was reachable via
    ///    the network layer's NAT traversal.
    /// 3. **Unreachable** — no address succeeded.
    /// 4. **NotFound** — the agent is not in the discovery cache.
    ///
    /// # Errors
    ///
    /// Returns an error only for internal failures (e.g. network not started).
    /// Connectivity failures are reported as `ConnectOutcome::Unreachable`.
    pub async fn connect_to_agent(
        &self,
        agent_id: &identity::AgentId,
    ) -> error::Result<connectivity::ConnectOutcome> {
        let call_start = std::time::Instant::now();
        let agent_prefix = network::hex_prefix(&agent_id.0, 4);
        tracing::debug!(
            target: "x0x::connect",
            stage = "connect_to_agent",
            %agent_prefix,
            "begin"
        );
        // 1. Look up in discovery cache
        let discovered = {
            let cache = self.identity_discovery_cache.read().await;
            cache.get(agent_id).cloned()
        };

        let agent = match discovered {
            Some(a) => a,
            None => {
                tracing::info!(
                    target: "x0x::connect",
                    stage = "connect_to_agent",
                    %agent_prefix,
                    outcome = "not_found",
                    dur_ms = call_start.elapsed().as_millis() as u64,
                    "agent not in discovery cache"
                );
                return Ok(connectivity::ConnectOutcome::NotFound);
            }
        };

        let info = connectivity::ReachabilityInfo::from_discovered(&agent);
        let v4_addrs = info.addresses.iter().filter(|a| a.is_ipv4()).count();
        let v6_addrs = info.addresses.len() - v4_addrs;
        tracing::info!(
            target: "x0x::connect",
            stage = "connect_to_agent",
            %agent_prefix,
            machine_prefix = %network::hex_prefix(&agent.machine_id.0, 4),
            addr_total = info.addresses.len(),
            v4_addrs,
            v6_addrs,
            can_receive_direct = ?info.can_receive_direct,
            should_attempt_direct = info.should_attempt_direct(),
            needs_coordination = info.needs_coordination(),
            "reachability classified"
        );

        let Some(ref network) = self.network else {
            tracing::warn!(
                target: "x0x::connect",
                stage = "connect_to_agent",
                %agent_prefix,
                outcome = "unreachable_no_network",
                "network layer not initialised"
            );
            return Ok(connectivity::ConnectOutcome::Unreachable);
        };

        // 2. If already connected via gossip, reuse that connection.
        //    This check MUST come before the empty-address bail-out because
        //    LAN/private agents may have no publicly-routable addresses in
        //    their announcement but are still reachable via the existing
        //    gossip QUIC connection.
        let connected_machine_id = if agent.machine_id.0 != [0u8; 32]
            && network
                .is_connected(&ant_quic::PeerId(agent.machine_id.0))
                .await
        {
            Some(agent.machine_id)
        } else {
            match self.direct_messaging.get_machine_id(agent_id).await {
                Some(machine_id) if network.is_connected(&ant_quic::PeerId(machine_id.0)).await => {
                    Some(machine_id)
                }
                _ => None,
            }
        };
        if let Some(machine_id) = connected_machine_id {
            if machine_id != agent.machine_id {
                let mut cache = self.identity_discovery_cache.write().await;
                if let Some(entry) = cache.get_mut(agent_id) {
                    entry.machine_id = machine_id;
                }
            }
            self.direct_messaging
                .mark_connected(agent.agent_id, machine_id)
                .await;
            let dur_ms = call_start.elapsed().as_millis() as u64;
            return if let Some(addr) = info.addresses.first() {
                let family = if addr.is_ipv4() { "v4" } else { "v6" };
                tracing::info!(
                    target: "x0x::connect",
                    stage = "connect_to_agent",
                    %agent_prefix,
                    strategy = "already_connected",
                    outcome = "direct",
                    selected_addr = %addr,
                    family,
                    dur_ms,
                    "reusing existing connection"
                );
                Ok(connectivity::ConnectOutcome::Direct(*addr))
            } else {
                tracing::info!(
                    target: "x0x::connect",
                    stage = "connect_to_agent",
                    %agent_prefix,
                    strategy = "already_connected",
                    outcome = "already_connected",
                    dur_ms,
                    "reusing existing connection without known addr"
                );
                Ok(connectivity::ConnectOutcome::AlreadyConnected)
            };
        }

        if info.addresses.is_empty() {
            tracing::info!(
                target: "x0x::connect",
                stage = "connect_to_agent",
                %agent_prefix,
                outcome = "unreachable",
                reason = "no_addresses",
                dur_ms = call_start.elapsed().as_millis() as u64,
                "no known addresses for agent"
            );
            return Ok(connectivity::ConnectOutcome::Unreachable);
        }

        let dial_timeout = std::time::Duration::from_secs(8);

        // 3. If we know the peer's machine ID, prefer a peer-authenticated dial
        //    with explicit address hints first. This is more reliable for agent
        //    cards and other out-of-band discoveries than a raw address dial.
        if agent.machine_id.0 != [0u8; 32] {
            let peer_id_hint = ant_quic::PeerId(agent.machine_id.0);
            self.seed_transport_peer_hints_for_target(network, &agent)
                .await
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "failed to seed transport peer hints: {e}"
                    )))
                })?;

            match tokio::time::timeout(
                dial_timeout,
                network.connect_peer_with_addrs(peer_id_hint, info.addresses.clone()),
            )
            .await
            {
                Ok(Ok((addr, verified_peer_id))) => {
                    let verified_machine_id = identity::MachineId(verified_peer_id.0);
                    if let Some(ref bc) = self.bootstrap_cache {
                        bc.add_from_connection(verified_peer_id, vec![addr], None)
                            .await;
                        bc.record_success(&verified_peer_id, 0).await;
                    }
                    {
                        let mut cache = self.identity_discovery_cache.write().await;
                        if let Some(entry) = cache.get_mut(agent_id) {
                            entry.machine_id = verified_machine_id;
                        }
                    }
                    self.direct_messaging
                        .mark_connected(agent.agent_id, verified_machine_id)
                        .await;
                    let family = if addr.is_ipv4() { "v4" } else { "v6" };
                    tracing::info!(
                        target: "x0x::connect",
                        stage = "connect_to_agent",
                        %agent_prefix,
                        strategy = "hinted_peer",
                        outcome = "coordinated",
                        selected_addr = %addr,
                        family,
                        dur_ms = call_start.elapsed().as_millis() as u64,
                        "hinted peer dial succeeded"
                    );
                    return Ok(connectivity::ConnectOutcome::Coordinated(addr));
                }
                Ok(Err(e)) => {
                    tracing::debug!(
                        target: "x0x::connect",
                        %agent_prefix,
                        strategy = "hinted_peer",
                        error = %e,
                        "hinted peer dial failed"
                    );
                }
                Err(_) => {
                    tracing::debug!(
                        target: "x0x::connect",
                        %agent_prefix,
                        strategy = "hinted_peer",
                        timeout_s = dial_timeout.as_secs(),
                        "hinted peer dial timed out"
                    );
                }
            }
        }

        // 4. Try direct connection whenever the peer is not explicitly known
        //    to require coordination. Unknown reachability still deserves a
        //    direct probe, especially for the first nodes in a new network.
        if info.should_attempt_direct() {
            for addr in &info.addresses {
                match tokio::time::timeout(dial_timeout, network.connect_addr(*addr)).await {
                    Ok(Ok(connected_peer_id)) => {
                        // Use the real PeerId from the QUIC handshake (may differ
                        // from a zeroed placeholder in the discovery cache).
                        let real_machine_id = identity::MachineId(connected_peer_id.0);
                        // Enrich bootstrap cache with this successful address
                        if let Some(ref bc) = self.bootstrap_cache {
                            bc.add_from_connection(connected_peer_id, vec![*addr], None)
                                .await;
                        }
                        // Update discovery cache with real machine_id
                        {
                            let mut cache = self.identity_discovery_cache.write().await;
                            if let Some(entry) = cache.get_mut(agent_id) {
                                entry.machine_id = real_machine_id;
                            }
                        }
                        // Register agent mapping for direct messaging
                        self.direct_messaging
                            .mark_connected(agent.agent_id, real_machine_id)
                            .await;
                        let family = if addr.is_ipv4() { "v4" } else { "v6" };
                        tracing::info!(
                            target: "x0x::connect",
                            stage = "connect_to_agent",
                            %agent_prefix,
                            strategy = "direct_per_addr",
                            outcome = "direct",
                            selected_addr = %addr,
                            family,
                            dur_ms = call_start.elapsed().as_millis() as u64,
                            "direct dial succeeded"
                        );
                        return Ok(connectivity::ConnectOutcome::Direct(*addr));
                    }
                    Ok(Err(e)) => {
                        tracing::debug!(
                            target: "x0x::connect",
                            %agent_prefix,
                            strategy = "direct_per_addr",
                            %addr,
                            error = %e,
                            "direct dial failed"
                        );
                    }
                    Err(_) => {
                        tracing::debug!(
                            target: "x0x::connect",
                            %agent_prefix,
                            strategy = "direct_per_addr",
                            %addr,
                            timeout_s = dial_timeout.as_secs(),
                            "direct dial timed out"
                        );
                    }
                }
            }
        }

        // 5. If direct failed and coordination may help, use peer-ID dialing
        //    with explicit address hints. This lets ant-quic combine the
        //    authenticated peer ID with known addresses from x0x discovery /
        //    imported cards, unlocking the full direct → hole-punch → relay path.
        if info.needs_coordination() || !info.should_attempt_direct() {
            // Use the machine_id from discovery cache as the peer_id hint.
            // NOTE: This may be a zeroed placeholder if the peer was discovered via
            // gossip and hasn't been verified via QUIC handshake yet.
            let peer_id_hint = ant_quic::PeerId(agent.machine_id.0);
            let hint_was_zeroed = agent.machine_id.0 == [0u8; 32];
            self.seed_transport_peer_hints_for_target(network, &agent)
                .await
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "failed to seed transport peer hints: {e}"
                    )))
                })?;
            let coordinated_result = tokio::time::timeout(
                dial_timeout,
                network.connect_peer_with_addrs(peer_id_hint, info.addresses.clone()),
            )
            .await;
            match coordinated_result {
                Ok(Ok((addr, verified_peer_id))) => {
                    let verified_machine_id = identity::MachineId(verified_peer_id.0);

                    // Only update caches if the original hint was not zeroed.
                    // When the hint was zeroed, we connected to *some* peer at that address
                    // but have no way to verify they are the agent we intended. Writing
                    // an unverified peer_id into the caches could corrupt the bootstrap cache
                    // with the wrong peer's identity.
                    if !hint_was_zeroed {
                        if let Some(ref bc) = self.bootstrap_cache {
                            bc.add_from_connection(verified_peer_id, vec![addr], None)
                                .await;
                            bc.record_success(&verified_peer_id, 0).await;
                        }
                        {
                            let mut cache = self.identity_discovery_cache.write().await;
                            if let Some(entry) = cache.get_mut(agent_id) {
                                entry.machine_id = verified_machine_id;
                            }
                        }
                    }

                    // Only register for direct messaging and update caches when the hint
                    // was non-zero. When the hint was zeroed, we connected to *some*
                    // peer at that address but have no cryptographic way to verify they
                    // are the agent we intended. Binding an unverified peer_id to an
                    // agent_id could corrupt the direct-messaging registry with the
                    // wrong peer's identity.
                    if !hint_was_zeroed {
                        self.direct_messaging
                            .mark_connected(agent.agent_id, verified_machine_id)
                            .await;
                    }
                    let family = if addr.is_ipv4() { "v4" } else { "v6" };
                    tracing::info!(
                        target: "x0x::connect",
                        stage = "connect_to_agent",
                        %agent_prefix,
                        strategy = "coordinated_fallback",
                        outcome = "coordinated",
                        selected_addr = %addr,
                        family,
                        hint_was_zeroed,
                        dur_ms = call_start.elapsed().as_millis() as u64,
                        "coordinated dial succeeded"
                    );
                    return Ok(connectivity::ConnectOutcome::Coordinated(addr));
                }
                Ok(Err(e)) => {
                    tracing::debug!(
                        target: "x0x::connect",
                        %agent_prefix,
                        strategy = "coordinated_fallback",
                        error = %e,
                        "coordinated dial failed"
                    );
                }
                Err(_) => {
                    tracing::debug!(
                        target: "x0x::connect",
                        %agent_prefix,
                        strategy = "coordinated_fallback",
                        timeout_s = dial_timeout.as_secs(),
                        "coordinated dial timed out"
                    );
                }
            }
        }

        tracing::warn!(
            target: "x0x::connect",
            stage = "connect_to_agent",
            %agent_prefix,
            outcome = "unreachable",
            reason = "all_strategies_exhausted",
            dur_ms = call_start.elapsed().as_millis() as u64,
            v4_addrs,
            v6_addrs,
            "all connection strategies exhausted"
        );
        Ok(connectivity::ConnectOutcome::Unreachable)
    }

    /// Save the bootstrap cache and release resources.
    ///
    /// Call this before dropping the agent to ensure the peer cache is
    /// persisted to disk. The background maintenance task saves periodically,
    /// but this guarantees a final save.
    pub async fn shutdown(&self) {
        // Shut down presence beacons.
        if let Some(ref pw) = self.presence {
            pw.shutdown().await;
            tracing::info!("Presence system shut down");
        }

        if let Some(ref cache) = self.bootstrap_cache {
            if let Err(e) = cache.save().await {
                tracing::warn!("Failed to save bootstrap cache on shutdown: {e}");
            } else {
                tracing::info!("Bootstrap cache saved on shutdown");
            }
        }
    }

    // === Direct Messaging ===

    /// Send data directly to a connected agent.
    ///
    /// This bypasses gossip pub/sub for efficient point-to-point communication.
    /// The agent must be connected first via [`Self::connect_to_agent`].
    ///
    /// # Arguments
    ///
    /// * `agent_id` - The target agent's identifier.
    /// * `payload` - The data to send.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Network is not initialized
    /// - Agent is not connected
    /// - Agent is not found in discovery cache
    /// - Send fails
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// // First connect to the agent
    /// let outcome = agent.connect_to_agent(&target_agent_id).await?;
    ///
    /// // Then send data directly
    /// agent.send_direct(&target_agent_id, b"hello".to_vec()).await?;
    /// ```
    /// Send data directly to an agent — capability-aware dispatch.
    ///
    /// Looks up the recipient's `DmCapabilities` in the local
    /// [`dm_capability::CapabilityStore`]. If the recipient advertises
    /// `gossip_inbox=true` with a non-empty `kem_public_key`, the send
    /// goes via the gossip DM path (signed+encrypted envelope published
    /// to the recipient's inbox topic with an application-layer ACK).
    /// Otherwise, falls back to the legacy raw-QUIC stream.
    ///
    /// # Errors
    ///
    /// See [`dm::DmError`].
    pub async fn send_direct(
        &self,
        to: &identity::AgentId,
        payload: Vec<u8>,
    ) -> Result<dm::DmReceipt, dm::DmError> {
        self.send_direct_with_config(to, payload, dm::DmSendConfig::default())
            .await
    }

    /// Like [`Self::send_direct`] with caller-provided [`dm::DmSendConfig`].
    ///
    /// # Errors
    ///
    /// See [`dm::DmError`].
    pub async fn send_direct_with_config(
        &self,
        to: &identity::AgentId,
        payload: Vec<u8>,
        config: dm::DmSendConfig,
    ) -> Result<dm::DmReceipt, dm::DmError> {
        let cap = self.capability_store.lookup(to);
        let gossip_ok = cap
            .as_ref()
            .map(|c| c.gossip_inbox && !c.kem_public_key.is_empty())
            .unwrap_or(false);

        if gossip_ok {
            let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
                dm::DmError::LocalGossipUnavailable(
                    "send_direct: no gossip runtime configured".to_string(),
                )
            })?;
            let signing = gossip::SigningContext::from_keypair(self.identity.agent_keypair());
            let kem_pub = cap
                .as_ref()
                .map(|c| c.kem_public_key.clone())
                .unwrap_or_default();
            return dm_send::send_via_gossip(
                std::sync::Arc::clone(runtime.pubsub()),
                &signing,
                self.identity.agent_id(),
                self.identity.machine_id(),
                *to,
                &kem_pub,
                payload,
                &config,
                std::sync::Arc::clone(&self.dm_inflight_acks),
            )
            .await;
        }

        if config.require_gossip {
            return Err(dm::DmError::RecipientKeyUnavailable(format!(
                "recipient {} has no gossip DM capability advert",
                hex::encode(to.as_bytes())
            )));
        }

        self.send_direct_raw_quic(to, payload)
            .await
            .map(|_| dm_send::raw_quic_receipt())
            .map_err(|e| match e {
                error::NetworkError::AgentNotFound(_)
                | error::NetworkError::AgentNotConnected(_) => {
                    dm::DmError::RecipientKeyUnavailable(e.to_string())
                }
                other => dm::DmError::PublishFailed(other.to_string()),
            })
    }

    /// Legacy raw-QUIC direct-send path. Internal fallback only.
    async fn send_direct_raw_quic(
        &self,
        agent_id: &identity::AgentId,
        payload: Vec<u8>,
    ) -> error::NetworkResult<()> {
        let send_start = std::time::Instant::now();
        let agent_prefix = network::hex_prefix(&agent_id.0, 4);
        let self_prefix = network::hex_prefix(&self.identity.agent_id().0, 4);
        let bytes = payload.len();

        let network = self.network.as_ref().ok_or_else(|| {
            tracing::warn!(
                target: "x0x::direct",
                stage = "send",
                %agent_prefix,
                outcome = "err_no_network",
                "network not initialised"
            );
            error::NetworkError::NodeCreation("network not initialized".to_string())
        })?;

        // Resolve the best known machine_id, preferring a machine that is
        // actually connected right now. Discovery cache entries can lag behind
        // the direct-messaging registry when an inbound connection is accepted
        // and later reconciled from transport events.
        let cached_machine_id = {
            let cache = self.identity_discovery_cache.read().await;
            cache
                .get(agent_id)
                .map(|d| d.machine_id)
                .filter(|m| m.0 != [0u8; 32]) // Ignore placeholder zeroed IDs
        };
        let registry_machine_id = self.direct_messaging.get_machine_id(agent_id).await;

        let (machine_id, resolution) = match (cached_machine_id, registry_machine_id) {
            (Some(id), _) if network.is_connected(&ant_quic::PeerId(id.0)).await => {
                (id, "cached_connected")
            }
            (_, Some(id)) if network.is_connected(&ant_quic::PeerId(id.0)).await => {
                if cached_machine_id != Some(id) {
                    let mut cache = self.identity_discovery_cache.write().await;
                    if let Some(entry) = cache.get_mut(agent_id) {
                        entry.machine_id = id;
                    }
                }
                (id, "registry_connected")
            }
            (Some(id), None) => (id, "cached_not_connected"),
            (Some(id), Some(_)) => (id, "cached_both_disconnected"),
            (None, Some(id)) => (id, "registry_not_connected"),
            (None, None) => {
                tracing::debug!(
                    target: "x0x::direct",
                    stage = "send",
                    %agent_prefix,
                    resolution = "last_resort_connect",
                    "no machine_id known; triggering connect_to_agent"
                );
                let _ = self.connect_to_agent(agent_id).await;
                let id = self
                    .direct_messaging
                    .get_machine_id(agent_id)
                    .await
                    .ok_or_else(|| {
                        tracing::warn!(
                            target: "x0x::direct",
                            stage = "send",
                            %agent_prefix,
                            outcome = "err_agent_not_found",
                            dur_ms = send_start.elapsed().as_millis() as u64,
                            "no machine_id after connect_to_agent"
                        );
                        error::NetworkError::AgentNotFound(agent_id.0)
                    })?;
                (id, "post_connect")
            }
        };

        // Check if connected
        let ant_peer_id = ant_quic::PeerId(machine_id.0);
        let machine_prefix = network::hex_prefix(&machine_id.0, 4);
        if !network.is_connected(&ant_peer_id).await {
            tracing::warn!(
                target: "x0x::direct",
                stage = "send",
                %agent_prefix,
                %machine_prefix,
                resolution,
                outcome = "err_not_connected",
                bytes,
                dur_ms = send_start.elapsed().as_millis() as u64,
                "machine_id resolved but peer not currently connected"
            );
            return Err(error::NetworkError::AgentNotConnected(agent_id.0));
        }

        // Send via network layer
        match network
            .send_direct(&ant_peer_id, &self.identity.agent_id().0, &payload)
            .await
        {
            Ok(()) => {
                tracing::info!(
                    target: "x0x::direct",
                    stage = "send",
                    from = %self_prefix,
                    to = %agent_prefix,
                    %machine_prefix,
                    resolution,
                    bytes,
                    dur_ms = send_start.elapsed().as_millis() as u64,
                    outcome = "ok",
                    "direct message sent"
                );
                Ok(())
            }
            Err(e) => {
                tracing::warn!(
                    target: "x0x::direct",
                    stage = "send",
                    from = %self_prefix,
                    to = %agent_prefix,
                    %machine_prefix,
                    resolution,
                    bytes,
                    dur_ms = send_start.elapsed().as_millis() as u64,
                    outcome = "err_transport",
                    error = %e,
                    "transport send_direct failed"
                );
                Err(e)
            }
        }
    }

    /// Receive the next direct message from any connected agent.
    ///
    /// Blocks until a direct message is received.
    ///
    /// # Security Note
    ///
    /// This method does **not** apply trust filtering from `ContactStore`.
    /// Messages from blocked agents will still be delivered. Use
    /// [`recv_direct_annotated()`](Self::recv_direct_annotated) if you need
    /// trust-based filtering.
    ///
    /// # Returns
    ///
    /// The received [`DirectMessage`] containing sender, payload, and timestamp.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// loop {
    ///     if let Some(msg) = agent.recv_direct().await {
    ///         println!("From {:?}: {:?}", msg.sender, msg.payload_str());
    ///     }
    /// }
    /// ```
    pub async fn recv_direct(&self) -> Option<direct::DirectMessage> {
        self.recv_direct_inner().await
    }

    /// Receive the next direct message, filtering by trust level.
    ///
    /// All messages now carry pre-computed `verified` and `trust_decision`
    /// fields from the identity discovery cache and contact store. This
    /// method passes through all messages — applications should inspect
    /// `msg.trust_decision` and `msg.verified` to decide how to handle
    /// each message.
    ///
    /// # Returns
    ///
    /// The received [`DirectMessage`], or `None` if the channel closes.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// loop {
    ///     if let Some(msg) = agent.recv_direct_annotated().await {
    ///         match msg.trust_decision {
    ///             Some(TrustDecision::RejectBlocked) => continue, // skip
    ///             Some(TrustDecision::Accept) if msg.verified => { /* trusted */ }
    ///             _ => { /* handle accordingly */ }
    ///         }
    ///     }
    /// }
    /// ```
    pub async fn recv_direct_annotated(&self) -> Option<direct::DirectMessage> {
        self.recv_direct_inner().await
    }

    /// Internal helper for receiving direct messages.
    ///
    /// Reads from the `DirectMessaging` internal channel, which is fed by
    /// the background `start_direct_listener` task. This ensures there is
    /// only ONE consumer of `network.recv_direct()` (the listener), avoiding
    /// message-stealing races.
    async fn recv_direct_inner(&self) -> Option<direct::DirectMessage> {
        self.direct_messaging.recv().await
    }

    /// Subscribe to direct messages.
    ///
    /// Returns a receiver that can be cloned for multiple consumers.
    /// Messages are broadcast to all receivers.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let mut rx = agent.subscribe_direct();
    /// tokio::spawn(async move {
    ///     while let Some(msg) = rx.recv().await {
    ///         println!("Direct message: {:?}", msg);
    ///     }
    /// });
    /// ```
    pub fn subscribe_direct(&self) -> direct::DirectMessageReceiver {
        self.direct_messaging.subscribe()
    }

    /// Get the direct messaging infrastructure.
    ///
    /// Provides low-level access to connection tracking and agent mappings.
    pub fn direct_messaging(&self) -> &std::sync::Arc<direct::DirectMessaging> {
        &self.direct_messaging
    }

    /// Check if an agent is currently connected for direct messaging.
    ///
    /// # Arguments
    ///
    /// * `agent_id` - The agent to check.
    ///
    /// # Returns
    ///
    /// `true` if a QUIC connection exists to this agent's machine.
    pub async fn is_agent_connected(&self, agent_id: &identity::AgentId) -> bool {
        let Some(network) = &self.network else {
            return false;
        };

        // Look up machine_id from discovery cache
        let machine_id = {
            let cache = self.identity_discovery_cache.read().await;
            cache.get(agent_id).map(|d| d.machine_id)
        };

        match machine_id {
            Some(mid) => {
                let ant_peer_id = ant_quic::PeerId(mid.0);
                network.is_connected(&ant_peer_id).await
            }
            None => false,
        }
    }

    /// Get list of currently connected agents.
    ///
    /// Returns agents that have been discovered and are currently connected
    /// via QUIC transport.
    pub async fn connected_agents(&self) -> Vec<identity::AgentId> {
        let Some(network) = &self.network else {
            return Vec::new();
        };

        let connected_peers = network.connected_peers().await;
        let cache = self.identity_discovery_cache.read().await;

        // Find agents whose machine_id matches a connected peer
        cache
            .values()
            .filter(|agent| {
                let ant_peer_id = ant_quic::PeerId(agent.machine_id.0);
                connected_peers.contains(&ant_peer_id)
            })
            .map(|agent| agent.agent_id)
            .collect()
    }

    /// Attach a contact store for trust-based message filtering.
    ///
    /// When set, the gossip pub/sub layer will:
    /// - Drop messages from `Blocked` senders (don't deliver, don't rebroadcast)
    /// - Annotate messages with the sender's trust level for consumers
    ///
    /// Without a contact store, all messages pass through (open relay mode).
    pub fn set_contacts(&self, store: std::sync::Arc<tokio::sync::RwLock<contacts::ContactStore>>) {
        if let Some(runtime) = &self.gossip_runtime {
            runtime.pubsub().set_contacts(store);
        }
    }

    /// Announce this agent's identity on the network discovery topic.
    ///
    /// By default, announcements include agent + machine identity only.
    /// Human identity disclosure is opt-in and requires explicit consent.
    ///
    /// # Arguments
    ///
    /// * `include_user_identity` - Whether to include `user_id` and certificate
    /// * `human_consent` - Must be `true` when disclosing user identity
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Gossip runtime is not initialized
    /// - Human identity disclosure is requested without explicit consent
    /// - Human identity disclosure is requested but no user identity is configured
    /// - Serialization or publish fails
    pub async fn announce_identity(
        &self,
        include_user_identity: bool,
        human_consent: bool,
    ) -> error::Result<()> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;

        self.start_identity_listener().await?;

        // Include ALL routable addresses (IPv4 and IPv6).
        let mut addresses = if let Some(network) = self.network.as_ref() {
            match network.node_status().await {
                Some(status) if !status.external_addrs.is_empty() => status.external_addrs,
                _ => match network.routable_addr().await {
                    Some(addr) => vec![addr],
                    None => self.announcement_addresses(),
                },
            }
        } else {
            self.announcement_addresses()
        };
        // Detect addresses locally via UDP socket tricks.
        // ant-quic discovers public IPv4 via OBSERVED_ADDRESS from peers.
        // IPv6 is globally routable (no NAT), so we probe locally.
        //
        // For locally-probed addresses (IPv6 and LAN IPv4), use the actual
        // bound port from the QUIC endpoint — NOT the first external address
        // port (which is NAT-mapped) and NOT the config bind port (which may
        // be 0 for OS-assigned ports).
        let bind_port = if let Some(network) = self.network.as_ref() {
            network.bound_addr().await.map(|a| a.port()).unwrap_or(5483)
        } else {
            5483
        };

        // IPv6 probe
        if let Ok(sock) = std::net::UdpSocket::bind("[::]:0") {
            if sock.connect("[2001:4860:4860::8888]:80").is_ok() {
                if let Ok(local) = sock.local_addr() {
                    if let std::net::IpAddr::V6(v6) = local.ip() {
                        let segs = v6.segments();
                        let is_global = (segs[0] & 0xffc0) != 0xfe80
                            && (segs[0] & 0xff00) != 0xfd00
                            && !v6.is_loopback();
                        if is_global {
                            let v6_addr =
                                std::net::SocketAddr::new(std::net::IpAddr::V6(v6), bind_port);
                            if !addresses.contains(&v6_addr) {
                                addresses.push(v6_addr);
                            }
                        }
                    }
                }
            }
        }

        for addr in collect_local_interface_addrs(bind_port) {
            if !addresses.contains(&addr) {
                addresses.push(addr);
            }
        }

        // Same rule as HeartbeatContext::announce() — gossip is global, so
        // LAN-scope addresses must never be published here. See the scope
        // analysis report under tests/proof-reports/ for details.
        addresses.retain(|a| is_publicly_advertisable(*a));

        let announcement = self.build_identity_announcement_with_addrs(
            include_user_identity,
            human_consent,
            addresses,
        )?;

        let encoded = bincode::serialize(&announcement).map_err(|e| {
            error::IdentityError::Serialization(format!(
                "failed to serialize identity announcement: {e}"
            ))
        })?;

        let payload = bytes::Bytes::from(encoded);

        // Publish to shard topic first (future-proof routing).
        let shard_topic = shard_topic_for_agent(&announcement.agent_id);
        runtime
            .pubsub()
            .publish(shard_topic, payload.clone())
            .await
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "failed to publish identity announcement to shard topic: {e}"
                )))
            })?;

        // Also publish to legacy broadcast topic for backward compatibility.
        runtime
            .pubsub()
            .publish(IDENTITY_ANNOUNCE_TOPIC.to_string(), payload)
            .await
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "failed to publish identity announcement: {e}"
                )))
            })?;

        let now = Self::unix_timestamp_secs();
        self.identity_discovery_cache.write().await.insert(
            announcement.agent_id,
            DiscoveredAgent {
                agent_id: announcement.agent_id,
                machine_id: announcement.machine_id,
                user_id: announcement.user_id,
                addresses: announcement.addresses.clone(),
                announced_at: announcement.announced_at,
                last_seen: now,
                machine_public_key: announcement.machine_public_key.clone(),
                nat_type: announcement.nat_type.clone(),
                can_receive_direct: announcement.can_receive_direct,
                is_relay: announcement.is_relay,
                is_coordinator: announcement.is_coordinator,
            },
        );

        // Record consent AFTER successful publish so heartbeats don't start
        // including user identity if this announcement never actually propagated.
        if include_user_identity && human_consent {
            self.user_identity_consented
                .store(true, std::sync::atomic::Ordering::Release);
        }

        Ok(())
    }

    /// Get all discovered agents from identity announcements.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn discovered_agents(&self) -> error::Result<Vec<DiscoveredAgent>> {
        self.start_identity_listener().await?;
        let cutoff = Self::unix_timestamp_secs().saturating_sub(self.identity_ttl_secs);
        let mut agents: Vec<_> = self
            .identity_discovery_cache
            .read()
            .await
            .values()
            .filter(|a| a.announced_at >= cutoff)
            .cloned()
            .collect();
        agents.sort_by_key(|a| a.agent_id.0);
        Ok(agents)
    }

    /// Return all discovered agents regardless of TTL.
    ///
    /// Unlike [`Self::discovered_agents`], this method skips TTL filtering and
    /// returns all cache entries, including stale ones. Useful for debugging.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn discovered_agents_unfiltered(&self) -> error::Result<Vec<DiscoveredAgent>> {
        self.start_identity_listener().await?;
        let mut agents: Vec<_> = self
            .identity_discovery_cache
            .read()
            .await
            .values()
            .cloned()
            .collect();
        agents.sort_by_key(|a| a.agent_id.0);
        Ok(agents)
    }

    /// Get one discovered agent record by agent ID.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn discovered_agent(
        &self,
        agent_id: identity::AgentId,
    ) -> error::Result<Option<DiscoveredAgent>> {
        self.start_identity_listener().await?;
        Ok(self
            .identity_discovery_cache
            .read()
            .await
            .get(&agent_id)
            .cloned())
    }

    async fn start_identity_listener(&self) -> error::Result<()> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;

        if self
            .identity_listener_started
            .swap(true, std::sync::atomic::Ordering::AcqRel)
        {
            return Ok(());
        }

        let mut sub_legacy = runtime
            .pubsub()
            .subscribe(IDENTITY_ANNOUNCE_TOPIC.to_string())
            .await;
        let own_shard_topic = shard_topic_for_agent(&self.agent_id());
        let mut sub_shard = runtime.pubsub().subscribe(own_shard_topic).await;
        let cache = std::sync::Arc::clone(&self.identity_discovery_cache);
        let bootstrap_cache = self.bootstrap_cache.clone();
        let contact_store = std::sync::Arc::clone(&self.contact_store);
        let direct_messaging = std::sync::Arc::clone(&self.direct_messaging);
        let network = self.network.as_ref().map(std::sync::Arc::clone);
        let own_agent_id = self.agent_id();
        let rebroadcast_pubsub = std::sync::Arc::clone(runtime.pubsub());

        tokio::spawn(async move {
            // Track agents we've already initiated auto-connect to, preventing
            // duplicate connection attempts from concurrent announcements.
            let mut auto_connect_attempted = std::collections::HashSet::<identity::AgentId>::new();

            // Time-windowed dedup for re-broadcast: (agent_id, announced_at)
            // → last-rebroadcast Instant. Bounds how often we re-emit any
            // given announcement so a busy mesh doesn't amplify gossip
            // indefinitely. Mirrors the pattern used by the release-manifest
            // re-broadcast loop in x0xd.
            let mut rebroadcast_state: std::collections::HashMap<
                (identity::AgentId, u64),
                std::time::Instant,
            > = std::collections::HashMap::new();
            const REBROADCAST_MIN_INTERVAL: std::time::Duration =
                std::time::Duration::from_secs(20);

            loop {
                // Drain whichever subscription fires next; deduplicate by AgentId in cache.
                let msg = tokio::select! {
                    Some(m) = sub_legacy.recv() => m,
                    Some(m) = sub_shard.recv() => m,
                    else => break,
                };
                let decoded = {
                    use bincode::Options;
                    bincode::options()
                        .with_fixint_encoding()
                        .with_limit(crate::network::MAX_MESSAGE_DESERIALIZE_SIZE)
                        .allow_trailing_bytes()
                        .deserialize::<IdentityAnnouncement>(&msg.payload)
                };
                let raw_payload = msg.payload.clone();
                let announcement = match decoded {
                    Ok(a) => a,
                    Err(e) => {
                        tracing::debug!("Ignoring invalid identity announcement payload: {}", e);
                        continue;
                    }
                };

                if let Err(e) = announcement.verify() {
                    tracing::warn!("Ignoring unverifiable identity announcement: {}", e);
                    continue;
                }

                // Evaluate trust for this (agent, machine) pair.
                // Blocked or machine-pinning violations are silently dropped.
                {
                    let store = contact_store.read().await;
                    let evaluator = trust::TrustEvaluator::new(&store);
                    let decision = evaluator.evaluate(&trust::TrustContext {
                        agent_id: &announcement.agent_id,
                        machine_id: &announcement.machine_id,
                    });
                    match decision {
                        trust::TrustDecision::RejectBlocked => {
                            tracing::debug!(
                                "Dropping identity announcement from blocked agent {:?}",
                                hex::encode(&announcement.agent_id.0[..8]),
                            );
                            continue;
                        }
                        trust::TrustDecision::RejectMachineMismatch => {
                            tracing::warn!(
                                "Dropping identity announcement from agent {:?}: machine {:?} not in pinned list",
                                hex::encode(&announcement.agent_id.0[..8]),
                                hex::encode(&announcement.machine_id.0[..8]),
                            );
                            continue;
                        }
                        _ => {}
                    }
                }

                // Update machine records in the contact store.
                {
                    let mut store = contact_store.write().await;
                    let record = contacts::MachineRecord::new(announcement.machine_id, None);
                    store.add_machine(&announcement.agent_id, record);
                }

                let now = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map_or(0, |d| d.as_secs());

                // Add only globally-routable addresses to the persistent
                // bootstrap cache. Private/LAN addresses are kept in the
                // ephemeral discovery cache (below) for same-network
                // connectivity, but must not persist across restarts where
                // they become stale dead-ends for remote nodes.
                {
                    let public_addrs: Vec<std::net::SocketAddr> = announcement
                        .addresses
                        .iter()
                        .copied()
                        .filter(|a| is_globally_routable(a.ip()))
                        .collect();
                    if !public_addrs.is_empty() {
                        if let Some(ref bc) = &bootstrap_cache {
                            let peer_id = ant_quic::PeerId(announcement.machine_id.0);
                            bc.add_from_connection(peer_id, public_addrs.clone(), None)
                                .await;
                            tracing::debug!(
                                "Added {} public addresses to bootstrap cache for agent {:?} (machine {:?})",
                                public_addrs.len(),
                                announcement.agent_id,
                                hex::encode(&announcement.machine_id.0[..8]),
                            );
                        }
                    }
                }

                // Cache the announcement with its address list filtered to
                // globally-advertisable scope only. Legacy peers that still
                // ship RFC1918/ULA/loopback entries in their announcements
                // must not force us to keep dialing their unreachable LAN
                // addresses — LAN discovery is ant-quic's mDNS job. Empty
                // address lists are preserved (the `AlreadyConnected` path in
                // `connect_to_agent` handles gossip peers we only reach by
                // an existing QUIC connection).
                let filtered_addresses: Vec<std::net::SocketAddr> = announcement
                    .addresses
                    .iter()
                    .copied()
                    .filter(|a| is_publicly_advertisable(*a))
                    .collect();
                cache.write().await.insert(
                    announcement.agent_id,
                    DiscoveredAgent {
                        agent_id: announcement.agent_id,
                        machine_id: announcement.machine_id,
                        user_id: announcement.user_id,
                        addresses: filtered_addresses,
                        announced_at: announcement.announced_at,
                        last_seen: now,
                        machine_public_key: announcement.machine_public_key.clone(),
                        nat_type: announcement.nat_type.clone(),
                        can_receive_direct: announcement.can_receive_direct,
                        is_relay: announcement.is_relay,
                        is_coordinator: announcement.is_coordinator,
                    },
                );

                // Identity announcements are the strongest agent↔machine binding we have.
                // Register the mapping immediately so reverse direct-send can resolve the
                // machine even before the first inbound direct payload arrives.
                direct_messaging
                    .register_agent(announcement.agent_id, announcement.machine_id)
                    .await;

                // Epidemic re-broadcast — mirrors the release-manifest
                // re-broadcast pattern. Bootstrap-node meshes have patchy
                // PlumTree overlap for the identity-announce topic: the
                // origin's tree only reaches 1–2 hops reliably. Making
                // every verified recipient re-publish guarantees flood
                // convergence across the mesh. Dedup-window on
                // (agent_id, announced_at) bounds amplification. Pub/Sub
                // v2 re-signs each publish with a new message ID so
                // PlumTree's own dedup cannot suppress our forward.
                if announcement.agent_id != own_agent_id {
                    let key = (announcement.agent_id, announcement.announced_at);
                    let should_forward = match rebroadcast_state.get(&key) {
                        None => true,
                        Some(last) => last.elapsed() >= REBROADCAST_MIN_INTERVAL,
                    };
                    if should_forward {
                        rebroadcast_state.insert(key, std::time::Instant::now());
                        // Prune stale entries to cap memory.
                        if rebroadcast_state.len() > 1024 {
                            let cutoff =
                                std::time::Instant::now() - std::time::Duration::from_secs(3600);
                            rebroadcast_state.retain(|_, t| *t >= cutoff);
                        }
                        let pubsub = std::sync::Arc::clone(&rebroadcast_pubsub);
                        let payload = raw_payload.clone();
                        tokio::spawn(async move {
                            if let Err(e) = pubsub
                                .publish(IDENTITY_ANNOUNCE_TOPIC.to_string(), payload)
                                .await
                            {
                                tracing::debug!("identity announcement re-broadcast failed: {e}");
                            }
                        });
                    }
                }

                // Reconcile the agent-level direct-message registry if the transport peer
                // is already connected (for example an inbound accept that happened before
                // this announcement reached us).
                if let Some(ref net) = &network {
                    let ant_peer_id = ant_quic::PeerId(announcement.machine_id.0);
                    if net.is_connected(&ant_peer_id).await {
                        direct_messaging
                            .mark_connected(announcement.agent_id, announcement.machine_id)
                            .await;
                    }
                }

                // Auto-connect to discovered agents so pub/sub messages can route
                // between peers that share bootstrap nodes but aren't directly connected.
                // The gossip topology refresh (every 1s) will add the new peer to
                // PlumTree topic trees once the QUIC connection is established.
                if announcement.agent_id != own_agent_id
                    && !announcement.addresses.is_empty()
                    && !auto_connect_attempted.contains(&announcement.agent_id)
                {
                    if let Some(ref net) = &network {
                        let ant_peer = ant_quic::PeerId(announcement.machine_id.0);
                        if !net.is_connected(&ant_peer).await {
                            auto_connect_attempted.insert(announcement.agent_id);
                            let net = std::sync::Arc::clone(net);
                            let addresses = announcement.addresses.clone();
                            tokio::spawn(async move {
                                for addr in &addresses {
                                    match net.connect_addr(*addr).await {
                                        Ok(_) => {
                                            tracing::info!(
                                                "Auto-connected to discovered agent at {addr}",
                                            );
                                            return;
                                        }
                                        Err(e) => {
                                            tracing::debug!("Auto-connect to {addr} failed: {e}",);
                                        }
                                    }
                                }
                                tracing::debug!(
                                    "Auto-connect exhausted all {} addresses for discovered agent",
                                    addresses.len(),
                                );
                            });
                        }
                    }
                }
            }
        });

        Ok(())
    }

    fn unix_timestamp_secs() -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |d| d.as_secs())
    }

    fn announcement_addresses(&self) -> Vec<std::net::SocketAddr> {
        match self.network.as_ref().and_then(|n| n.local_addr()) {
            Some(addr) if addr.port() > 0 => collect_local_interface_addrs(addr.port()),
            _ => Vec::new(),
        }
    }

    fn build_identity_announcement(
        &self,
        include_user_identity: bool,
        human_consent: bool,
    ) -> error::Result<IdentityAnnouncement> {
        self.build_identity_announcement_with_addrs(
            include_user_identity,
            human_consent,
            self.announcement_addresses(),
        )
    }

    fn build_identity_announcement_with_addrs(
        &self,
        include_user_identity: bool,
        human_consent: bool,
        addresses: Vec<std::net::SocketAddr>,
    ) -> error::Result<IdentityAnnouncement> {
        if include_user_identity && !human_consent {
            return Err(error::IdentityError::Storage(std::io::Error::other(
                "human identity disclosure requires explicit human consent — set human_consent: true in the request body",
            )));
        }

        let (user_id, agent_certificate) = if include_user_identity {
            let user_id = self.user_id().ok_or_else(|| {
                error::IdentityError::Storage(std::io::Error::other(
                    "human identity disclosure requested but no user identity is configured — set user_key_path in your config.toml to point at your user keypair file",
                ))
            })?;
            let cert = self.agent_certificate().cloned().ok_or_else(|| {
                error::IdentityError::Storage(std::io::Error::other(
                    "human identity disclosure requested but agent certificate is missing",
                ))
            })?;
            (Some(user_id), Some(cert))
        } else {
            (None, None)
        };

        let machine_public_key = self
            .identity
            .machine_keypair()
            .public_key()
            .as_bytes()
            .to_vec();

        // NAT status is populated by the heartbeat loop (which is async and has
        // access to NodeStatus). Here we use None for the NAT fields as this
        // sync builder doesn't have async access to the network layer.
        let unsigned = IdentityAnnouncementUnsigned {
            agent_id: self.agent_id(),
            machine_id: self.machine_id(),
            user_id,
            agent_certificate: agent_certificate.clone(),
            machine_public_key: machine_public_key.clone(),
            addresses,
            announced_at: Self::unix_timestamp_secs(),
            nat_type: None,
            can_receive_direct: None,
            is_relay: None,
            is_coordinator: None,
        };
        let unsigned_bytes = bincode::serialize(&unsigned).map_err(|e| {
            error::IdentityError::Serialization(format!(
                "failed to serialize unsigned identity announcement: {e}"
            ))
        })?;
        let machine_signature = ant_quic::crypto::raw_public_keys::pqc::sign_with_ml_dsa(
            self.identity.machine_keypair().secret_key(),
            &unsigned_bytes,
        )
        .map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "failed to sign identity announcement with machine key: {:?}",
                e
            )))
        })?
        .as_bytes()
        .to_vec();

        Ok(IdentityAnnouncement {
            agent_id: unsigned.agent_id,
            machine_id: unsigned.machine_id,
            user_id: unsigned.user_id,
            agent_certificate: unsigned.agent_certificate,
            machine_public_key,
            machine_signature,
            addresses: unsigned.addresses,
            announced_at: unsigned.announced_at,
            nat_type: unsigned.nat_type,
            can_receive_direct: unsigned.can_receive_direct,
            is_relay: unsigned.is_relay,
            is_coordinator: unsigned.is_coordinator,
        })
    }

    /// Join the x0x gossip network.
    ///
    /// Connects to bootstrap peers in parallel with automatic retries.
    /// Failed connections are retried after a delay to allow stale
    /// connections on remote nodes to expire.
    ///
    /// If the agent was not configured with a network, this method
    /// succeeds gracefully (nothing to join).
    pub async fn join_network(&self) -> error::Result<()> {
        let Some(network) = self.network.as_ref() else {
            tracing::debug!("join_network called but no network configured");
            return Ok(());
        };

        if let Some(ref runtime) = self.gossip_runtime {
            runtime.start().await.map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "failed to start gossip runtime: {e}"
                )))
            })?;
            tracing::info!("Gossip runtime started");
        }
        self.start_identity_listener().await?;
        self.start_network_event_listener();
        self.start_direct_listener();

        let bootstrap_nodes = network.config().bootstrap_nodes.clone();

        let min_connected = 3;
        let mut all_connected: Vec<std::net::SocketAddr> = Vec::new();

        // ant-quic now owns first-party mDNS discovery and auto-connect.
        // x0x keeps bootstrap/cache orchestration here, while the transport
        // layer handles zero-config LAN discovery internally.

        // Phase 0: Try quality-scored coordinator peers from bootstrap cache.
        // The bootstrap cache learns about coordinator-capable peers passively
        // through normal connections — no coordinator gossip topic needed.
        if let Some(ref cache) = self.bootstrap_cache {
            let coordinators = cache.select_coordinators(6).await;
            let coordinator_addrs: Vec<std::net::SocketAddr> = coordinators
                .iter()
                .flat_map(|peer| peer.preferred_addresses())
                .collect();

            if !coordinator_addrs.is_empty() {
                tracing::info!(
                    "Phase 0: Trying {} addresses from {} cached coordinators",
                    coordinator_addrs.len(),
                    coordinators.len()
                );
                let (succeeded, _failed) = self
                    .connect_peers_parallel_tracked(network, &coordinator_addrs)
                    .await;
                all_connected.extend(&succeeded);
                tracing::info!(
                    "Phase 0: {}/{} coordinator addresses connected",
                    succeeded.len(),
                    coordinator_addrs.len()
                );
            }
        }

        // Phase 1: Try cached peers first using the real ant-quic peer IDs.
        if all_connected.len() < min_connected {
            if let Some(ref cache) = self.bootstrap_cache {
                const PHASE1_PEER_CANDIDATES: usize = 12;
                let cached_peers = cache.select_peers(PHASE1_PEER_CANDIDATES).await;
                if !cached_peers.is_empty() {
                    tracing::info!("Phase 1: Trying {} cached peers", cached_peers.len());
                    let (succeeded, _failed) = self
                        .connect_cached_peers_parallel_tracked(network, &cached_peers)
                        .await;
                    all_connected.extend(&succeeded);
                    tracing::info!(
                        "Phase 1: {}/{} cached peers connected",
                        succeeded.len(),
                        cached_peers.len()
                    );
                }
            }
        } // end Phase 1 min_connected check

        // Phase 2: Connect to hardcoded bootstrap nodes if we need more peers.
        // This is the fallback for when coordinator cache and cached peers aren't enough.
        if all_connected.len() < min_connected && !bootstrap_nodes.is_empty() {
            let remaining: Vec<std::net::SocketAddr> = bootstrap_nodes
                .iter()
                .filter(|addr| !all_connected.contains(addr))
                .copied()
                .collect();

            // Round 1: Connect to all bootstrap peers in parallel
            let (succeeded, mut failed) = self
                .connect_peers_parallel_tracked(network, &remaining)
                .await;
            all_connected.extend(&succeeded);
            tracing::info!(
                "Phase 2 round 1: {}/{} bootstrap peers connected",
                succeeded.len(),
                remaining.len()
            );

            // Retry rounds for failed peers
            for round in 2..=3 {
                if failed.is_empty() {
                    break;
                }
                let delay = std::time::Duration::from_secs(if round == 2 { 10 } else { 15 });
                tracing::info!(
                    "Retrying {} failed peers in {}s (round {})",
                    failed.len(),
                    delay.as_secs(),
                    round
                );
                tokio::time::sleep(delay).await;

                let (succeeded, still_failed) =
                    self.connect_peers_parallel_tracked(network, &failed).await;
                all_connected.extend(&succeeded);
                failed = still_failed;
                tracing::info!(
                    "Phase 2 round {}: {} total peers connected",
                    round,
                    all_connected.len()
                );
            }

            if !failed.is_empty() {
                tracing::warn!(
                    "Could not connect to {} bootstrap peers: {:?}",
                    failed.len(),
                    failed
                );
            }
        }

        tracing::info!(
            "Network join complete. Connected to {} peers.",
            all_connected.len()
        );

        // Join the HyParView membership overlay via connected peers.
        if let Some(ref runtime) = self.gossip_runtime {
            let seeds: Vec<String> = all_connected.iter().map(|addr| addr.to_string()).collect();
            if !seeds.is_empty() {
                if let Err(e) = runtime.membership().join(seeds).await {
                    tracing::warn!("HyParView membership join failed: {e}");
                }
            }
        }

        // Start presence beacons after membership overlay is established.
        if let Some(ref pw) = self.presence {
            // Seed broadcast peers from connected peers so beacons propagate.
            if let Some(ref runtime) = self.gossip_runtime {
                let active = runtime.membership().active_view();
                for peer in active {
                    pw.manager().add_broadcast_peer(peer).await;
                }
                tracing::info!(
                    "Presence seeded with {} broadcast peers",
                    pw.manager().broadcast_peer_count().await
                );
            }

            // Populate address hints from network status for beacon metadata.
            //
            // Presence beacons propagate over global gossip — filter to only
            // globally-advertisable addresses. LAN discovery is ant-quic's
            // mDNS job, not ours. Also exclude `status.local_addr` which is
            // typically the wildcard `[::]:5483` and never a dialable target.
            if let Some(ref net) = self.network {
                if let Some(status) = net.node_status().await {
                    let hints: Vec<String> = status
                        .external_addrs
                        .iter()
                        .filter(|a| is_publicly_advertisable(**a))
                        .map(|a| a.to_string())
                        .collect();
                    pw.manager().set_addr_hints(hints).await;
                }
            }

            if pw.config().enable_beacons {
                if let Err(e) = pw
                    .manager()
                    .start_beacons(pw.config().beacon_interval_secs)
                    .await
                {
                    tracing::warn!("Failed to start presence beacons: {e}");
                } else {
                    tracing::info!(
                        "Presence beacons started (interval={}s)",
                        pw.config().beacon_interval_secs
                    );
                }
            }

            // Start the presence event-emission loop so that subscribers
            // automatically receive AgentOnline/AgentOffline events after
            // join_network() returns.
            pw.start_event_loop(std::sync::Arc::clone(&self.identity_discovery_cache))
                .await;
            tracing::debug!("Presence event loop started");
        }

        if let Err(e) = self.announce_identity(false, false).await {
            tracing::warn!("Initial identity announcement failed: {}", e);
        }
        if let Err(e) = self.start_identity_heartbeat().await {
            tracing::warn!("Failed to start identity heartbeat: {e}");
        }

        // Schedule a fresh re-announcement after gossip topology stabilizes.
        // The initial publish fires before PlumTree has formed eager-push links,
        // so peers that connected after the first announce won't see it.
        // A fresh announcement (new message ID) is required because PlumTree
        // deduplicates by message ID — replaying identical bytes would be silently
        // dropped by peers that already received the first announcement.
        if let (Some(ref runtime), Some(ref network)) = (&self.gossip_runtime, &self.network) {
            let ctx = HeartbeatContext {
                identity: std::sync::Arc::clone(&self.identity),
                runtime: std::sync::Arc::clone(runtime),
                network: std::sync::Arc::clone(network),
                interval_secs: self.heartbeat_interval_secs,
                cache: std::sync::Arc::clone(&self.identity_discovery_cache),
                user_identity_consented: std::sync::Arc::clone(&self.user_identity_consented),
            };
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_secs(3)).await;
                if let Err(e) = ctx.announce().await {
                    tracing::warn!("Delayed identity re-announcement failed: {e}");
                } else {
                    tracing::info!(
                        "Delayed identity re-announcement sent (gossip mesh stabilized)"
                    );
                }
            });
        }

        // Start the capability advert service. Until start_dm_inbox runs,
        // the advert declares gossip_inbox=false — senders fall back to
        // raw-QUIC. Once upgraded, peers learn about gossip DM support.
        if let Some(ref runtime) = self.gossip_runtime {
            let signing = std::sync::Arc::new(gossip::SigningContext::from_keypair(
                self.identity.agent_keypair(),
            ));
            let caps_rx = self.dm_capabilities_tx.subscribe();
            match dm_capability_service::CapabilityAdvertService::spawn_default(
                std::sync::Arc::clone(runtime.pubsub()),
                signing,
                self.identity.agent_id(),
                self.identity.machine_id(),
                caps_rx,
                std::sync::Arc::clone(&self.capability_store),
            )
            .await
            {
                Ok(service) => {
                    let mut guard = self.capability_advert_service.lock().await;
                    if let Some(prev) = guard.take() {
                        prev.abort();
                    }
                    *guard = Some(service);
                    tracing::info!("Capability advert service started");
                }
                Err(e) => tracing::warn!("failed to start capability advert service: {e}"),
            }
        }

        Ok(())
    }

    /// Clone the shared capability store.
    #[must_use]
    pub fn capability_store(&self) -> std::sync::Arc<dm_capability::CapabilityStore> {
        std::sync::Arc::clone(&self.capability_store)
    }

    /// Clone the shared DM in-flight ACK registry.
    #[must_use]
    pub fn dm_inflight_acks(&self) -> std::sync::Arc<dm::InFlightAcks> {
        std::sync::Arc::clone(&self.dm_inflight_acks)
    }

    /// Clone the shared recent-delivery dedupe cache.
    #[must_use]
    pub fn recent_delivery_cache(&self) -> std::sync::Arc<dm::RecentDeliveryCache> {
        std::sync::Arc::clone(&self.recent_delivery_cache)
    }

    /// Spawn the DM inbox service backed by the given KEM keypair.
    /// Idempotent — the prior service is aborted before spawning new.
    ///
    /// # Errors
    ///
    /// Returns an error if no gossip runtime is configured.
    pub async fn start_dm_inbox(
        &self,
        kem_keypair: std::sync::Arc<groups::kem_envelope::AgentKemKeypair>,
        config: dm_inbox::DmInboxConfig,
    ) -> error::Result<()> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "cannot start DM inbox: no gossip runtime configured",
            ))
        })?;
        let signing = std::sync::Arc::new(gossip::SigningContext::from_keypair(
            self.identity.agent_keypair(),
        ));
        let service = dm_inbox::DmInboxService::spawn(
            std::sync::Arc::clone(runtime.pubsub()),
            signing,
            self.identity.agent_id(),
            self.identity.machine_id(),
            std::sync::Arc::clone(&kem_keypair),
            std::sync::Arc::clone(&self.direct_messaging),
            std::sync::Arc::clone(&self.contact_store),
            std::sync::Arc::clone(&self.dm_inflight_acks),
            std::sync::Arc::clone(&self.recent_delivery_cache),
            config,
        )
        .await
        .map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "DM inbox spawn failed: {e}"
            )))
        })?;
        let mut guard = self.dm_inbox_service.lock().await;
        if let Some(prev) = guard.take() {
            prev.abort();
        }
        *guard = Some(service);

        // Upgrade our advertised capabilities so peers stop falling back
        // to the raw-QUIC path. The capability advert service watches
        // this channel and republishes immediately on change.
        let upgraded =
            dm::DmCapabilities::pending().with_kem_public_key(kem_keypair.public_bytes.clone());
        if self.dm_capabilities_tx.send(upgraded).is_err() {
            tracing::debug!("dm_capabilities watch has no receivers; skipping upgrade broadcast");
        }
        tracing::info!("DM inbox service started");
        Ok(())
    }

    /// Stop the DM inbox service, if running. Idempotent.
    pub async fn stop_dm_inbox(&self) {
        let mut guard = self.dm_inbox_service.lock().await;
        if let Some(service) = guard.take() {
            service.abort();
        }
    }

    /// Connect to cached peers in parallel, returning (succeeded, failed) peer lists.
    async fn connect_cached_peers_parallel_tracked(
        &self,
        network: &std::sync::Arc<network::NetworkNode>,
        peers: &[ant_quic::CachedPeer],
    ) -> (Vec<std::net::SocketAddr>, Vec<ant_quic::PeerId>) {
        use tokio::time::{timeout, Duration};
        const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);

        let handles: Vec<_> = peers
            .iter()
            .map(|peer| {
                let net = network.clone();
                let peer_id = peer.peer_id;
                tokio::spawn(async move {
                    tracing::debug!("Connecting to cached peer: {:?}", peer_id);
                    match timeout(CONNECT_TIMEOUT, net.connect_cached_peer(peer_id)).await {
                        Ok(Ok(addr)) => {
                            tracing::info!("Connected to cached peer {:?} at {}", peer_id, addr);
                            Ok(addr)
                        }
                        Ok(Err(e)) => {
                            tracing::warn!("Failed to connect to cached peer {:?}: {}", peer_id, e);
                            Err(peer_id)
                        }
                        Err(_) => {
                            tracing::warn!(
                                "Connection to cached peer {:?} timed out after {}s",
                                peer_id,
                                CONNECT_TIMEOUT.as_secs()
                            );
                            Err(peer_id)
                        }
                    }
                })
            })
            .collect();

        let mut succeeded = Vec::new();
        let mut failed = Vec::new();
        for handle in handles {
            match handle.await {
                Ok(Ok(addr)) => succeeded.push(addr),
                Ok(Err(peer_id)) => failed.push(peer_id),
                Err(e) => tracing::error!("Connection task panicked: {}", e),
            }
        }
        (succeeded, failed)
    }

    /// Connect to multiple peers in parallel, returning (succeeded, failed) address lists.
    async fn connect_peers_parallel_tracked(
        &self,
        network: &std::sync::Arc<network::NetworkNode>,
        addrs: &[std::net::SocketAddr],
    ) -> (Vec<std::net::SocketAddr>, Vec<std::net::SocketAddr>) {
        use tokio::time::{timeout, Duration};

        // Per-connection timeout prevents hanging when connecting to
        // ourselves or to unreachable addresses.
        const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);

        let handles: Vec<_> = addrs
            .iter()
            .map(|addr| {
                let net = network.clone();
                let addr = *addr;
                tokio::spawn(async move {
                    tracing::debug!("Connecting to peer: {}", addr);
                    match timeout(CONNECT_TIMEOUT, net.connect_addr(addr)).await {
                        Ok(Ok(_)) => {
                            tracing::info!("Connected to peer: {}", addr);
                            Ok(addr)
                        }
                        Ok(Err(e)) => {
                            tracing::warn!("Failed to connect to {}: {}", addr, e);
                            Err(addr)
                        }
                        Err(_) => {
                            tracing::warn!(
                                "Connection to {} timed out after {}s",
                                addr,
                                CONNECT_TIMEOUT.as_secs()
                            );
                            Err(addr)
                        }
                    }
                })
            })
            .collect();

        let mut succeeded = Vec::new();
        let mut failed = Vec::new();
        for handle in handles {
            match handle.await {
                Ok(Ok(addr)) => succeeded.push(addr),
                Ok(Err(addr)) => failed.push(addr),
                Err(e) => tracing::error!("Connection task panicked: {}", e),
            }
        }
        (succeeded, failed)
    }

    /// Subscribe to messages on a given topic.
    ///
    /// Returns a [`gossip::Subscription`] that yields messages as they arrive
    /// through the gossip network.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Gossip runtime is not initialized (configure agent with network first)
    pub async fn subscribe(&self, topic: &str) -> error::Result<Subscription> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;
        Ok(runtime.pubsub().subscribe(topic.to_string()).await)
    }

    /// Publish a message to a topic.
    ///
    /// The message will propagate through the gossip network via
    /// epidemic broadcast — every agent that receives it will
    /// relay it to its neighbours.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Gossip runtime is not initialized (configure agent with network first)
    /// - Message encoding or broadcast fails
    pub async fn publish(&self, topic: &str, payload: Vec<u8>) -> error::Result<()> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;
        runtime
            .pubsub()
            .publish(topic.to_string(), bytes::Bytes::from(payload))
            .await
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "publish failed: {}",
                    e
                )))
            })
    }

    /// Get connected peer IDs.
    ///
    /// Returns the list of peers currently connected via the gossip network.
    ///
    /// # Errors
    ///
    /// Returns an error if the network is not initialized.
    pub async fn peers(&self) -> error::Result<Vec<saorsa_gossip_types::PeerId>> {
        let network = self.network.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "network not initialized - configure agent with network first",
            ))
        })?;
        let ant_peers = network.connected_peers().await;
        Ok(ant_peers
            .into_iter()
            .map(|p| saorsa_gossip_types::PeerId::new(p.0))
            .collect())
    }

    /// Get online agents.
    ///
    /// Returns agent IDs discovered from signed identity announcements.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn presence(&self) -> error::Result<Vec<identity::AgentId>> {
        self.start_identity_listener().await?;
        let cutoff = Self::unix_timestamp_secs().saturating_sub(self.identity_ttl_secs);
        let mut agents: Vec<_> = self
            .identity_discovery_cache
            .read()
            .await
            .values()
            .filter(|a| a.announced_at >= cutoff)
            .map(|a| a.agent_id)
            .collect();
        agents.sort_by_key(|a| a.0);
        Ok(agents)
    }

    /// Subscribe to presence events (agent online/offline notifications).
    ///
    /// Returns a [`tokio::sync::broadcast::Receiver<PresenceEvent>`] that yields
    /// [`presence::PresenceEvent`] values as agents come online or go offline.
    ///
    /// The diff-based event emission loop is started lazily on the first call to this
    /// method (or when [`join_network`](Agent::join_network) is called). Subsequent
    /// calls return independent receivers on the same broadcast channel.
    ///
    /// # Errors
    ///
    /// Returns [`error::NetworkError::NodeError`] if this agent was built
    /// without a network configuration (i.e. no `with_network_config` on the builder).
    pub async fn subscribe_presence(
        &self,
    ) -> error::NetworkResult<tokio::sync::broadcast::Receiver<presence::PresenceEvent>> {
        let pw = self.presence.as_ref().ok_or_else(|| {
            error::NetworkError::NodeError("presence system not initialized".to_string())
        })?;
        // Ensure the event loop is running.
        pw.start_event_loop(std::sync::Arc::clone(&self.identity_discovery_cache))
            .await;
        Ok(pw.subscribe_events())
    }

    /// Look up a single agent in the local discovery cache.
    ///
    /// Returns `None` if the agent is not currently cached.  No network I/O is
    /// performed — use [`discover_agent_by_id`](Agent::discover_agent_by_id) for
    /// an active lookup that queries the network.
    pub async fn cached_agent(&self, id: &identity::AgentId) -> Option<DiscoveredAgent> {
        self.identity_discovery_cache.read().await.get(id).cloned()
    }

    /// Check whether a claimed `AgentId` is verified as belonging to the
    /// given `MachineId` in the identity discovery cache.
    ///
    /// Returns `true` if the cache contains a signed identity announcement
    /// binding this agent to this machine. Returns `false` if the agent is
    /// unknown or bound to a different machine.
    pub async fn is_agent_machine_verified(
        &self,
        agent_id: &identity::AgentId,
        machine_id: &identity::MachineId,
    ) -> bool {
        let cache = self.identity_discovery_cache.read().await;
        cache
            .get(agent_id)
            .map(|entry| entry.machine_id == *machine_id)
            .unwrap_or(false)
    }

    /// Discover agents via Friend-of-a-Friend (FOAF) random walk.
    ///
    /// Initiates a FOAF query on the global presence topic with the given `ttl`
    /// (maximum hop count) and `timeout_ms` (response collection window).
    ///
    /// Returned entries are resolved against the local identity discovery cache
    /// so that known agents are returned with full identity data.  Unknown peers
    /// are included with a minimal entry (addresses only) that will be enriched
    /// once their identity heartbeat arrives.
    ///
    /// # Arguments
    ///
    /// * `ttl` — Maximum hop count for the random walk (`1`–`5`). Typical: `2`.
    /// * `timeout_ms` — Query timeout in milliseconds. Typical: `5000`.
    ///
    /// # Errors
    ///
    /// Returns [`error::NetworkError::NodeError`] if no network config was provided.
    pub async fn discover_agents_foaf(
        &self,
        ttl: u8,
        timeout_ms: u64,
    ) -> error::NetworkResult<Vec<DiscoveredAgent>> {
        let pw = self.presence.as_ref().ok_or_else(|| {
            error::NetworkError::NodeError("presence system not initialized".to_string())
        })?;

        let topic = presence::global_presence_topic();
        let raw_results: Vec<(
            saorsa_gossip_types::PeerId,
            saorsa_gossip_types::PresenceRecord,
        )> = pw
            .manager()
            .initiate_foaf_query(topic, ttl, timeout_ms)
            .await
            .map_err(|e| error::NetworkError::NodeError(e.to_string()))?;

        let cache = self.identity_discovery_cache.read().await;

        // Convert and deduplicate by agent_id.
        let mut seen: std::collections::HashSet<identity::AgentId> =
            std::collections::HashSet::new();
        let mut agents: Vec<DiscoveredAgent> = Vec::with_capacity(raw_results.len());

        for (peer_id, record) in &raw_results {
            if let Some(agent) =
                presence::presence_record_to_discovered_agent(*peer_id, record, &cache)
            {
                if seen.insert(agent.agent_id) {
                    agents.push(agent);
                }
            }
        }

        Ok(agents)
    }

    /// Discover a specific agent by their [`identity::AgentId`] via FOAF random walk.
    ///
    /// Fast-path: checks the local identity discovery cache first and returns
    /// immediately if the agent is already known.
    ///
    /// Slow-path: performs a FOAF random walk (see [`discover_agents_foaf`](Agent::discover_agents_foaf))
    /// and searches the results for a matching `AgentId`.
    ///
    /// Returns `None` if the agent is not found within the given `ttl` and `timeout_ms`.
    ///
    /// # Errors
    ///
    /// Returns [`error::NetworkError::NodeCreation`] if no network config was provided.
    pub async fn discover_agent_by_id(
        &self,
        target_id: identity::AgentId,
        ttl: u8,
        timeout_ms: u64,
    ) -> error::NetworkResult<Option<DiscoveredAgent>> {
        // Fast path: already in local cache.
        {
            let cache = self.identity_discovery_cache.read().await;
            if let Some(agent) = cache.get(&target_id) {
                return Ok(Some(agent.clone()));
            }
        }

        // Slow path: FOAF random walk.
        let agents = self.discover_agents_foaf(ttl, timeout_ms).await?;
        Ok(agents.into_iter().find(|a| a.agent_id == target_id))
    }

    /// Find an agent by ID, returning its known addresses.
    ///
    /// Performs a three-stage lookup:
    /// 1. **Cache hit** — return addresses immediately if the agent has already
    ///    been discovered.
    /// 2. **Shard subscription** — subscribe to the agent's identity shard topic
    ///    and wait up to 5 seconds for a heartbeat announcement.
    /// 3. **Rendezvous** — subscribe to the agent's rendezvous shard topic and
    ///    wait up to 5 seconds for a `ProviderSummary` advertisement.  This
    ///    works even when the two agents are on different gossip overlay clusters.
    ///
    /// Returns `None` if the agent is not found within the combined deadline.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn find_agent(
        &self,
        agent_id: identity::AgentId,
    ) -> error::Result<Option<Vec<std::net::SocketAddr>>> {
        self.start_identity_listener().await?;

        // Stage 1: cache hit.
        if let Some(addrs) = self
            .identity_discovery_cache
            .read()
            .await
            .get(&agent_id)
            .map(|e| e.addresses.clone())
        {
            return Ok(Some(addrs));
        }

        // Stage 2: subscribe to the agent's identity shard topic and wait up to 5 s.
        let runtime = match self.gossip_runtime.as_ref() {
            Some(r) => r,
            None => return Ok(None),
        };
        let shard_topic = shard_topic_for_agent(&agent_id);
        let mut sub = runtime.pubsub().subscribe(shard_topic).await;
        let cache = std::sync::Arc::clone(&self.identity_discovery_cache);
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);

        loop {
            if tokio::time::Instant::now() >= deadline {
                break;
            }
            let timeout = tokio::time::sleep_until(deadline);
            tokio::select! {
                Some(msg) = sub.recv() => {
                    if let Ok(ann) = {
                        use bincode::Options;
                        bincode::DefaultOptions::new()
                            .with_limit(crate::network::MAX_MESSAGE_DESERIALIZE_SIZE)
                            .deserialize::<IdentityAnnouncement>(&msg.payload)
                    } {
                        if ann.verify().is_ok() && ann.agent_id == agent_id {
                            let now = std::time::SystemTime::now()
                                .duration_since(std::time::UNIX_EPOCH)
                                .map_or(0, |d| d.as_secs());
                            let filtered: Vec<std::net::SocketAddr> = ann
                                .addresses
                                .iter()
                                .copied()
                                .filter(|a| is_publicly_advertisable(*a))
                                .collect();
                            let addrs = filtered.clone();
                            cache.write().await.insert(
                                ann.agent_id,
                                DiscoveredAgent {
                                    agent_id: ann.agent_id,
                                    machine_id: ann.machine_id,
                                    user_id: ann.user_id,
                                    addresses: filtered,
                                    announced_at: ann.announced_at,
                                    last_seen: now,
                                    machine_public_key: ann.machine_public_key.clone(),
                                    nat_type: ann.nat_type.clone(),
                                    can_receive_direct: ann.can_receive_direct,
                                    is_relay: ann.is_relay,
                                    is_coordinator: ann.is_coordinator,
                                },
                            );
                            return Ok(Some(addrs));
                        }
                    }
                }
                _ = timeout => break,
            }
        }

        // Stage 3: rendezvous shard subscription — wait up to 5 s.
        // Cache the result so subsequent connect_to_agent / send_direct can find it.
        if let Some(addrs) = self.find_agent_rendezvous(agent_id, 5).await? {
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map_or(0, |d| d.as_secs());
            cache
                .write()
                .await
                .entry(agent_id)
                .or_insert_with(|| DiscoveredAgent {
                    agent_id,
                    machine_id: identity::MachineId([0u8; 32]),
                    user_id: None,
                    addresses: addrs.clone(),
                    announced_at: now,
                    last_seen: now,
                    machine_public_key: Vec::new(),
                    nat_type: None,
                    can_receive_direct: None,
                    is_relay: None,
                    is_coordinator: None,
                });
            return Ok(Some(addrs));
        }

        Ok(None)
    }

    /// Find all discovered agents claiming ownership by the given [`identity::UserId`].
    ///
    /// Only returns agents that announced with `include_user_identity: true`
    /// (i.e., agents whose [`DiscoveredAgent::user_id`] is `Some`).
    ///
    /// # Arguments
    ///
    /// * `user_id` - The user identity to search for
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn find_agents_by_user(
        &self,
        user_id: identity::UserId,
    ) -> error::Result<Vec<DiscoveredAgent>> {
        self.start_identity_listener().await?;
        let cutoff = Self::unix_timestamp_secs().saturating_sub(self.identity_ttl_secs);
        Ok(self
            .identity_discovery_cache
            .read()
            .await
            .values()
            .filter(|a| a.announced_at >= cutoff && a.user_id == Some(user_id))
            .cloned()
            .collect())
    }

    /// Return the local socket address this agent's network node is bound to, if any.
    ///
    /// Returns `None` if no network has been configured or if the bind address is
    /// not yet known.
    ///
    /// **Note:** If the node was configured with port 0, this returns port 0.
    /// Use [`bound_addr()`](Self::bound_addr) to get the OS-assigned port.
    #[must_use]
    pub fn local_addr(&self) -> Option<std::net::SocketAddr> {
        self.network.as_ref().and_then(|n| n.local_addr())
    }

    /// Return the actual bound address from the QUIC endpoint.
    ///
    /// Unlike [`local_addr()`](Self::local_addr) which returns the configured value
    /// (possibly port 0), this queries the running endpoint for the real OS-assigned
    /// address. Returns `None` if no network has been configured.
    pub async fn bound_addr(&self) -> Option<std::net::SocketAddr> {
        if let Some(ref network) = self.network {
            let addr = network.bound_addr().await;
            // On dual-stack systems, bound_addr may return [::]:port even when
            // we bound to 127.0.0.1. Normalize to IPv4 if the original config
            // was IPv4.
            match (addr, self.local_addr()) {
                (Some(bound), Some(config)) if config.is_ipv4() && bound.is_ipv6() => {
                    Some(std::net::SocketAddr::new(config.ip(), bound.port()))
                }
                (Some(bound), _) => Some(bound),
                _ => None,
            }
        } else {
            None
        }
    }

    /// Build a signed [`IdentityAnnouncement`] for this agent.
    ///
    /// Delegates to the internal `build_identity_announcement` method.
    ///
    /// # Errors
    ///
    /// Returns an error if key signing fails or human consent is required but not given.
    pub fn build_announcement(
        &self,
        include_user: bool,
        consent: bool,
    ) -> error::Result<IdentityAnnouncement> {
        self.build_identity_announcement(include_user, consent)
    }

    /// Start the background identity heartbeat task.
    ///
    /// Idempotent — if the heartbeat is already running, returns `Ok(())` immediately.
    /// The heartbeat re-announces this agent's identity at `heartbeat_interval_secs`
    /// intervals so that late-joining peers can discover it without waiting for a
    /// Start the network event reconciliation listener.
    ///
    /// This bridges transport-level peer connect/disconnect events into the
    /// agent-level direct messaging registry so inbound accepted connections are
    /// usable for reverse direct sends before the first inbound direct payload.
    fn start_network_event_listener(&self) {
        if self
            .network_event_listener_started
            .swap(true, std::sync::atomic::Ordering::AcqRel)
        {
            return;
        }

        let Some(network) = self.network.as_ref().map(std::sync::Arc::clone) else {
            return;
        };
        let cache = std::sync::Arc::clone(&self.identity_discovery_cache);
        let dm = std::sync::Arc::clone(&self.direct_messaging);

        tokio::spawn(async move {
            let mut rx = network.subscribe();
            tracing::info!("Network event reconciliation listener started");

            loop {
                let event = match rx.recv().await {
                    Ok(event) => event,
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
                        tracing::warn!("Network event listener lagged by {skipped} events");
                        continue;
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                };

                match event {
                    network::NetworkEvent::PeerConnected { peer_id, .. } => {
                        let machine_id = identity::MachineId(peer_id);
                        let cached_agent_id = {
                            let cache = cache.read().await;
                            cache
                                .values()
                                .find(|entry| entry.machine_id == machine_id)
                                .map(|entry| entry.agent_id)
                        };
                        let agent_id = match cached_agent_id {
                            Some(agent_id) => Some(agent_id),
                            None => dm.lookup_agent(&machine_id).await,
                        };
                        if let Some(agent_id) = agent_id {
                            dm.mark_connected(agent_id, machine_id).await;
                        }
                    }
                    network::NetworkEvent::PeerDisconnected { peer_id } => {
                        let machine_id = identity::MachineId(peer_id);
                        let cached_agent_id = {
                            let cache = cache.read().await;
                            cache
                                .values()
                                .find(|entry| entry.machine_id == machine_id)
                                .map(|entry| entry.agent_id)
                        };
                        let agent_id = match cached_agent_id {
                            Some(agent_id) => Some(agent_id),
                            None => dm.lookup_agent(&machine_id).await,
                        };
                        if let Some(agent_id) = agent_id {
                            dm.mark_disconnected(&agent_id).await;
                        }
                    }
                    _ => {}
                }
            }
        });
    }

    /// Start the direct message listener background task.
    ///
    /// This task reads raw direct messages from the network layer and
    /// dispatches them to `DirectMessaging::handle_incoming()`, which
    /// broadcasts to all `subscribe_direct()` receivers.
    ///
    /// Called automatically by [`Agent::join_network`].
    fn start_direct_listener(&self) {
        if self
            .direct_listener_started
            .swap(true, std::sync::atomic::Ordering::AcqRel)
        {
            return;
        }

        let Some(network) = self.network.as_ref().map(std::sync::Arc::clone) else {
            return;
        };
        let dm = std::sync::Arc::clone(&self.direct_messaging);
        let discovery_cache = std::sync::Arc::clone(&self.identity_discovery_cache);
        let contact_store = std::sync::Arc::clone(&self.contact_store);

        tokio::spawn(async move {
            tracing::info!(target: "x0x::direct", stage = "listener", "direct message listener started");
            loop {
                let Some((ant_peer_id, payload)) = network.recv_direct().await else {
                    tracing::warn!(
                        target: "x0x::direct",
                        stage = "listener",
                        "network.recv_direct channel closed — listener exiting"
                    );
                    break;
                };

                let raw_bytes = payload.len();

                // Parse: first 32 bytes = sender AgentId, rest = payload
                if payload.len() < 32 {
                    tracing::warn!(
                        target: "x0x::direct",
                        stage = "listener",
                        machine_prefix = %network::hex_prefix(&ant_peer_id.0, 4),
                        raw_bytes,
                        outcome = "drop_too_short",
                        "direct message too short to contain sender id"
                    );
                    continue;
                }

                let mut sender_bytes = [0u8; 32];
                sender_bytes.copy_from_slice(&payload[..32]);
                let sender = identity::AgentId(sender_bytes);
                let machine_id = identity::MachineId(ant_peer_id.0);
                let data = payload[32..].to_vec();
                let payload_bytes = data.len();

                // Verify AgentId→MachineId binding against identity discovery cache.
                let verified = {
                    let cache = discovery_cache.read().await;
                    cache
                        .get(&sender)
                        .map(|entry| entry.machine_id == machine_id)
                        .unwrap_or(false)
                };

                // Evaluate trust for the (AgentId, MachineId) pair.
                let trust_decision = {
                    let contacts = contact_store.read().await;
                    let evaluator = trust::TrustEvaluator::new(&contacts);
                    let ctx = trust::TrustContext {
                        agent_id: &sender,
                        machine_id: &machine_id,
                    };
                    Some(evaluator.evaluate(&ctx))
                };

                tracing::info!(
                    target: "x0x::direct",
                    stage = "recv",
                    sender_prefix = %network::hex_prefix(&sender.0, 4),
                    machine_prefix = %network::hex_prefix(&machine_id.0, 4),
                    raw_bytes,
                    payload_bytes,
                    verified,
                    trust_decision = ?trust_decision,
                    "direct message received; dispatching to subscribers"
                );

                // Register and mark the sender as connected for future reverse direct sends.
                dm.mark_connected(sender, machine_id).await;

                // Broadcast to all subscribe_direct() receivers with verification info.
                dm.handle_incoming(machine_id, sender, data, verified, trust_decision)
                    .await;

                tracing::debug!(
                    target: "x0x::direct",
                    stage = "recv",
                    sender_prefix = %network::hex_prefix(&sender.0, 4),
                    payload_bytes,
                    subscriber_count = dm.subscriber_count(),
                    "direct message dispatched"
                );
            }
        });
    }

    /// new announcement.
    ///
    /// Called automatically by [`Agent::join_network`].
    ///
    /// # Errors
    ///
    /// Returns an error if a required network or gossip component is missing.
    pub async fn start_identity_heartbeat(&self) -> error::Result<()> {
        let mut handle_guard = self.heartbeat_handle.lock().await;
        if handle_guard.is_some() {
            return Ok(());
        }
        let Some(runtime) = self.gossip_runtime.as_ref().map(std::sync::Arc::clone) else {
            return Err(error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized — cannot start heartbeat",
            )));
        };
        let Some(network) = self.network.as_ref().map(std::sync::Arc::clone) else {
            return Err(error::IdentityError::Storage(std::io::Error::other(
                "network not initialized — cannot start heartbeat",
            )));
        };
        let ctx = HeartbeatContext {
            identity: std::sync::Arc::clone(&self.identity),
            runtime,
            network,
            interval_secs: self.heartbeat_interval_secs,
            cache: std::sync::Arc::clone(&self.identity_discovery_cache),
            user_identity_consented: std::sync::Arc::clone(&self.user_identity_consented),
        };
        let handle = tokio::task::spawn(async move {
            let mut ticker =
                tokio::time::interval(std::time::Duration::from_secs(ctx.interval_secs));
            ticker.tick().await; // skip first immediate tick
            loop {
                ticker.tick().await;
                if let Err(e) = ctx.announce().await {
                    tracing::warn!("identity heartbeat announce failed: {e}");
                }
            }
        });
        *handle_guard = Some(handle);
        Ok(())
    }

    /// Publish a rendezvous `ProviderSummary` for this agent.
    ///
    /// Enables global findability across gossip overlay partitions.  Seekers
    /// that have never been on the same partition as this agent can still
    /// discover it by subscribing to the rendezvous shard topic and waiting
    /// for the next heartbeat advertisement.
    ///
    /// The summary is signed with this agent's machine key and contains the
    /// agent's reachability addresses in the `extensions` field (bincode-encoded
    /// `Vec<SocketAddr>`).
    ///
    /// # Re-advertisement contract
    ///
    /// Rendezvous summaries expire after `validity_ms` milliseconds.  **Callers
    /// are responsible for calling `advertise_identity` again before expiry** so
    /// that seekers can always find a fresh record.  A common strategy is to
    /// re-advertise every `validity_ms / 2`.  The `x0xd` daemon does this
    /// automatically via its background re-advertisement task.
    ///
    /// # Arguments
    ///
    /// * `validity_ms` — How long (milliseconds) before the summary expires.
    ///   After this time, seekers will no longer discover this agent via rendezvous
    ///   unless a fresh `advertise_identity` call is made.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized, serialization
    /// fails, or signing fails.
    pub async fn advertise_identity(&self, validity_ms: u64) -> error::Result<()> {
        use saorsa_gossip_rendezvous::{Capability, ProviderSummary};

        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized — cannot advertise identity",
            ))
        })?;

        let peer_id = runtime.peer_id();
        let addresses = self.announcement_addresses();
        let addr_bytes = bincode::serialize(&addresses).map_err(|e| {
            error::IdentityError::Serialization(format!(
                "failed to serialize addresses for rendezvous: {e}"
            ))
        })?;

        let mut summary = ProviderSummary::new(
            self.agent_id().0,
            peer_id,
            vec![Capability::Identity],
            validity_ms,
        )
        .with_extensions(addr_bytes);

        summary
            .sign_raw(self.identity.machine_keypair().secret_key().as_bytes())
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "failed to sign rendezvous summary: {e}"
                )))
            })?;

        let cbor_bytes = summary.to_cbor().map_err(|e| {
            error::IdentityError::Serialization(format!(
                "failed to CBOR-encode rendezvous summary: {e}"
            ))
        })?;

        let topic = rendezvous_shard_topic_for_agent(&self.agent_id());
        runtime
            .pubsub()
            .publish(topic, bytes::Bytes::from(cbor_bytes))
            .await
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "failed to publish rendezvous summary: {e}"
                )))
            })?;

        self.rendezvous_advertised
            .store(true, std::sync::atomic::Ordering::Relaxed);
        Ok(())
    }

    /// Search for an agent via rendezvous shard subscription.
    ///
    /// Subscribes to the rendezvous shard topic for `agent_id` and waits up to
    /// `timeout_secs` for a matching [`saorsa_gossip_rendezvous::ProviderSummary`].
    /// On success the addresses encoded in the summary `extensions` field are
    /// returned.
    ///
    /// This is Stage 3 of [`Agent::find_agent`]'s lookup cascade.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn find_agent_rendezvous(
        &self,
        agent_id: identity::AgentId,
        timeout_secs: u64,
    ) -> error::Result<Option<Vec<std::net::SocketAddr>>> {
        use saorsa_gossip_rendezvous::ProviderSummary;

        let runtime = match self.gossip_runtime.as_ref() {
            Some(r) => r,
            None => return Ok(None),
        };

        let topic = rendezvous_shard_topic_for_agent(&agent_id);
        let mut sub = runtime.pubsub().subscribe(topic).await;
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(timeout_secs);

        loop {
            if tokio::time::Instant::now() >= deadline {
                break;
            }
            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
            tokio::select! {
                Some(msg) = sub.recv() => {
                    let summary = match ProviderSummary::from_cbor(&msg.payload) {
                        Ok(s) => s,
                        Err(_) => continue,
                    };
                    if summary.target != agent_id.0 {
                        continue;
                    }
                    // Verify the summary signature when the advertiser's machine
                    // public key is cached from a prior identity announcement.
                    // Without a cached key we still accept the addresses — they
                    // are connection hints only; the subsequent QUIC handshake will
                    // fail cryptographically if the endpoint is not the genuine agent.
                    let cached_pub = self
                        .identity_discovery_cache
                        .read()
                        .await
                        .get(&agent_id)
                        .map(|e| e.machine_public_key.clone());
                    if let Some(pub_bytes) = cached_pub {
                        if !pub_bytes.is_empty()
                            && !summary.verify_raw(&pub_bytes).unwrap_or(false)
                        {
                            tracing::warn!(
                                "Rendezvous summary signature verification failed for agent {:?}; discarding",
                                agent_id
                            );
                            continue;
                        }
                    }
                    // Decode addresses from the extensions field.
                    let addrs: Vec<std::net::SocketAddr> = summary
                        .extensions
                        .as_deref()
                        .and_then(|b| {
                            use bincode::Options;
                            bincode::DefaultOptions::new()
                                .with_limit(crate::network::MAX_MESSAGE_DESERIALIZE_SIZE)
                                .deserialize(b)
                                .ok()
                        })
                        .unwrap_or_default();
                    if !addrs.is_empty() {
                        return Ok(Some(addrs));
                    }
                }
                _ = tokio::time::sleep(remaining) => break,
            }
        }

        Ok(None)
    }

    /// Insert a discovered agent into the cache (for testing only).
    ///
    /// # Arguments
    ///
    /// * `agent` - The agent entry to insert.
    #[doc(hidden)]
    pub async fn insert_discovered_agent_for_testing(&self, agent: DiscoveredAgent) {
        let agent_id = agent.agent_id;
        let machine_id = agent.machine_id;
        self.identity_discovery_cache
            .write()
            .await
            .insert(agent_id, agent);

        if machine_id.0 != [0u8; 32] {
            self.direct_messaging
                .register_agent(agent_id, machine_id)
                .await;
            if let Some(ref network) = self.network {
                let ant_peer_id = ant_quic::PeerId(machine_id.0);
                if network.is_connected(&ant_peer_id).await {
                    self.direct_messaging
                        .mark_connected(agent_id, machine_id)
                        .await;
                }
            }
        }
    }

    /// Create a new collaborative task list bound to a topic.
    ///
    /// Creates a new `TaskList` and binds it to the specified gossip topic
    /// for automatic synchronization with other agents on the same topic.
    ///
    /// # Arguments
    ///
    /// * `name` - Human-readable name for the task list
    /// * `topic` - Gossip topic for synchronization
    ///
    /// # Returns
    ///
    /// A `TaskListHandle` for interacting with the task list.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let list = agent.create_task_list("Sprint Planning", "team-sprint").await?;
    /// ```
    pub async fn create_task_list(&self, name: &str, topic: &str) -> error::Result<TaskListHandle> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;

        let peer_id = runtime.peer_id();
        let list_id = crdt::TaskListId::from_content(name, &self.agent_id(), 0);
        let task_list = crdt::TaskList::new(list_id, name.to_string(), peer_id);

        let sync = crdt::TaskListSync::new(
            task_list,
            std::sync::Arc::clone(runtime.pubsub()),
            topic.to_string(),
            30,
        )
        .map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "task list sync creation failed: {}",
                e
            )))
        })?;

        let sync = std::sync::Arc::new(sync);
        sync.start().await.map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "task list sync start failed: {}",
                e
            )))
        })?;

        Ok(TaskListHandle {
            sync,
            agent_id: self.agent_id(),
            peer_id,
        })
    }

    /// Join an existing task list by topic.
    ///
    /// Connects to a task list that was created by another agent on the
    /// specified topic. The local replica will sync with peers automatically.
    ///
    /// # Arguments
    ///
    /// * `topic` - Gossip topic for the task list
    ///
    /// # Returns
    ///
    /// A `TaskListHandle` for interacting with the task list.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let list = agent.join_task_list("team-sprint").await?;
    /// ```
    pub async fn join_task_list(&self, topic: &str) -> error::Result<TaskListHandle> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;

        let peer_id = runtime.peer_id();
        // Create empty task list; it will be populated via delta sync
        let list_id = crdt::TaskListId::from_content(topic, &self.agent_id(), 0);
        let task_list = crdt::TaskList::new(list_id, String::new(), peer_id);

        let sync = crdt::TaskListSync::new(
            task_list,
            std::sync::Arc::clone(runtime.pubsub()),
            topic.to_string(),
            30,
        )
        .map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "task list sync creation failed: {}",
                e
            )))
        })?;

        let sync = std::sync::Arc::new(sync);
        sync.start().await.map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "task list sync start failed: {}",
                e
            )))
        })?;

        Ok(TaskListHandle {
            sync,
            agent_id: self.agent_id(),
            peer_id,
        })
    }
}

impl AgentBuilder {
    /// Set a custom path for the machine keypair.
    ///
    /// If not set, the machine keypair is stored in `~/.x0x/machine.key`.
    ///
    /// # Arguments
    ///
    /// * `path` - The path to store the machine keypair.
    ///
    /// # Returns
    ///
    /// Self for chaining.
    pub fn with_machine_key<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
        self.machine_key_path = Some(path.as_ref().to_path_buf());
        self
    }

    /// Import an agent keypair.
    ///
    /// If not set, the agent keypair is loaded from storage (or generated fresh
    /// if no stored key exists).
    ///
    /// This enables running the same agent on multiple machines by importing
    /// the same agent keypair (but with different machine keypairs).
    ///
    /// Note: When an explicit keypair is provided via this method, it takes
    /// precedence over `with_agent_key_path()`.
    ///
    /// # Arguments
    ///
    /// * `keypair` - The agent keypair to import.
    ///
    /// # Returns
    ///
    /// Self for chaining.
    pub fn with_agent_key(mut self, keypair: identity::AgentKeypair) -> Self {
        self.agent_keypair = Some(keypair);
        self
    }

    /// Set a custom path for the agent keypair.
    ///
    /// If not set, the agent keypair is stored in `~/.x0x/agent.key`.
    /// If no stored key is found at the path, a fresh one is generated and saved.
    ///
    /// This is ignored when `with_agent_key()` provides an explicit keypair.
    ///
    /// # Arguments
    ///
    /// * `path` - The path to store/load the agent keypair.
    ///
    /// # Returns
    ///
    /// Self for chaining.
    pub fn with_agent_key_path<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
        self.agent_key_path = Some(path.as_ref().to_path_buf());
        self
    }

    /// Set a custom path for the agent certificate (`agent.cert`).
    ///
    /// Required for multi-daemon setups that share a host — the default
    /// path (`~/.x0x/agent.cert`) is shared across daemons, causing
    /// last-writer-wins trampling that makes peers reject identity
    /// announcements as `agent certificate agent_id mismatch`.
    ///
    /// # Arguments
    ///
    /// * `path` - The path to use for the agent certificate file.
    ///
    /// # Returns
    ///
    /// Self for chaining.
    #[must_use]
    pub fn with_agent_cert_path<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
        self.agent_cert_path = Some(path.as_ref().to_path_buf());
        self
    }

    /// Set network configuration for P2P communication.
    ///
    /// If not set, default network configuration is used.
    ///
    /// # Arguments
    ///
    /// * `config` - The network configuration to use.
    ///
    /// # Returns
    ///
    /// Self for chaining.
    pub fn with_network_config(mut self, config: network::NetworkConfig) -> Self {
        self.network_config = Some(config);
        self
    }

    /// Set the directory for the bootstrap peer cache.
    ///
    /// The cache persists peer quality metrics across restarts, enabling
    /// cache-first join strategy. Defaults to `~/.x0x/peers/` if not set.
    /// Falls back to `./.x0x/peers/` (relative to CWD) if `$HOME` is unset.
    pub fn with_peer_cache_dir<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
        self.peer_cache_dir = Some(path.as_ref().to_path_buf());
        self
    }

    /// Disable the bootstrap peer cache entirely.
    ///
    /// When set, the agent will not open or load any cached peers on
    /// startup. This ensures complete network isolation from previously
    /// seen peers for embedders and dedicated test harnesses.
    ///
    /// Note: the x0xd daemon's `--no-hard-coded-bootstrap` flag does
    /// not call this; it only clears configured seed peers.
    pub fn with_peer_cache_disabled(mut self) -> Self {
        self.disable_peer_cache = true;
        self
    }

    /// Import a user keypair for three-layer identity.
    ///
    /// This binds a human identity to this agent. When provided, an
    /// [`identity::AgentCertificate`] is automatically issued (if one
    /// doesn't already exist in storage) to cryptographically attest
    /// that this agent belongs to the user.
    ///
    /// Note: When an explicit keypair is provided via this method, it takes
    /// precedence over `with_user_key_path()`.
    ///
    /// # Arguments
    ///
    /// * `keypair` - The user keypair to import.
    ///
    /// # Returns
    ///
    /// Self for chaining.
    pub fn with_user_key(mut self, keypair: identity::UserKeypair) -> Self {
        self.user_keypair = Some(keypair);
        self
    }

    /// Set a custom path for the user keypair.
    ///
    /// Unlike machine and agent keys, user keys are **not** auto-generated.
    /// If the file at this path doesn't exist, no user identity is set
    /// (the agent operates with two-layer identity).
    ///
    /// This is ignored when `with_user_key()` provides an explicit keypair.
    ///
    /// # Arguments
    ///
    /// * `path` - The path to load the user keypair from.
    ///
    /// # Returns
    ///
    /// Self for chaining.
    pub fn with_user_key_path<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
        self.user_key_path = Some(path.as_ref().to_path_buf());
        self
    }

    /// Set the identity heartbeat re-announcement interval.
    ///
    /// Defaults to [`IDENTITY_HEARTBEAT_INTERVAL_SECS`] (300 seconds).
    ///
    /// # Arguments
    ///
    /// * `secs` - Interval in seconds between identity re-announcements.
    #[must_use]
    pub fn with_heartbeat_interval(mut self, secs: u64) -> Self {
        self.heartbeat_interval_secs = Some(secs);
        self
    }

    /// Set the identity cache TTL.
    ///
    /// Cache entries with `last_seen` older than this threshold are filtered
    /// from [`Agent::presence`] and [`Agent::discovered_agents`].
    ///
    /// Defaults to [`IDENTITY_TTL_SECS`] (900 seconds).
    ///
    /// # Arguments
    ///
    /// * `secs` - Time-to-live in seconds for discovered agent entries.
    #[must_use]
    pub fn with_identity_ttl(mut self, secs: u64) -> Self {
        self.identity_ttl_secs = Some(secs);
        self
    }

    /// Override the presence beacon broadcast interval in seconds.
    #[must_use]
    pub fn with_presence_beacon_interval(mut self, secs: u64) -> Self {
        self.presence_beacon_interval_secs = Some(secs);
        self
    }

    /// Override the presence event poll interval in seconds.
    #[must_use]
    pub fn with_presence_event_poll_interval(mut self, secs: u64) -> Self {
        self.presence_event_poll_interval_secs = Some(secs);
        self
    }

    /// Override the fallback offline timeout used by presence events.
    #[must_use]
    pub fn with_presence_offline_timeout(mut self, secs: u64) -> Self {
        self.presence_offline_timeout_secs = Some(secs);
        self
    }

    /// Set a custom path for the contacts file.
    ///
    /// The contacts file persists trust levels and machine records for known
    /// agents. Defaults to `~/.x0x/contacts.json` if not set.
    ///
    /// # Arguments
    ///
    /// * `path` - The path for the contacts file.
    #[must_use]
    pub fn with_contact_store_path<P: AsRef<std::path::Path>>(mut self, path: P) -> Self {
        self.contact_store_path = Some(path.as_ref().to_path_buf());
        self
    }

    /// Build and initialise the agent.
    ///
    /// This performs the following:
    /// 1. Loads or generates the machine keypair (stored in `~/.x0x/machine.key` by default)
    /// 2. Uses provided agent keypair or generates a fresh one
    /// 3. Combines both into a unified Identity
    ///
    /// The machine keypair is automatically persisted to storage.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Machine keypair generation fails
    /// - Storage I/O fails
    /// - Keypair deserialization fails
    pub async fn build(self) -> error::Result<Agent> {
        // Determine machine keypair source
        let machine_keypair = if let Some(path) = self.machine_key_path {
            // Try to load from custom path
            match storage::load_machine_keypair_from(&path).await {
                Ok(kp) => kp,
                Err(_) => {
                    // Generate fresh keypair and save to custom path
                    let kp = identity::MachineKeypair::generate()?;
                    storage::save_machine_keypair_to(&kp, &path).await?;
                    kp
                }
            }
        } else if storage::machine_keypair_exists().await {
            // Load default machine keypair
            storage::load_machine_keypair().await?
        } else {
            // Generate and save default machine keypair
            let kp = identity::MachineKeypair::generate()?;
            storage::save_machine_keypair(&kp).await?;
            kp
        };

        // Resolve agent keypair: explicit > path-based > default storage > generate
        let agent_keypair = if let Some(kp) = self.agent_keypair {
            // Explicit keypair takes highest precedence
            kp
        } else if let Some(path) = self.agent_key_path {
            // Custom path: load or generate+save
            match storage::load_agent_keypair_from(&path).await {
                Ok(kp) => kp,
                Err(_) => {
                    let kp = identity::AgentKeypair::generate()?;
                    storage::save_agent_keypair_to(&kp, &path).await?;
                    kp
                }
            }
        } else if storage::agent_keypair_exists().await {
            // Default path exists: load it
            storage::load_agent_keypair_default().await?
        } else {
            // No stored key: generate and persist
            let kp = identity::AgentKeypair::generate()?;
            storage::save_agent_keypair_default(&kp).await?;
            kp
        };

        // Resolve user keypair: explicit > path-based > default storage > None (opt-in)
        let user_keypair = if let Some(kp) = self.user_keypair {
            Some(kp)
        } else if let Some(path) = self.user_key_path {
            // Custom path: load if exists, otherwise None (don't auto-generate)
            storage::load_user_keypair_from(&path).await.ok()
        } else if storage::user_keypair_exists().await {
            // Default path exists: load it
            storage::load_user_keypair().await.ok()
        } else {
            None
        };

        // Build identity with optional user layer.
        //
        // The agent certificate binds (user_id, agent_id). On load we must
        // verify BOTH halves of the binding match the current identity —
        // user_id AND agent_id — and re-issue if either diverges. A mismatch
        // happens in two practical scenarios:
        //   1. The user key was replaced (cert's user_id no longer ours).
        //   2. Multi-daemon-per-host setups where the cert path is shared
        //      and a peer daemon overwrote it with their own cert (cert's
        //      agent_id no longer ours).
        // Without the agent_id half of the check, scenario (2) produces an
        // announcement whose cert binds another daemon's agent_id, and peers
        // reject it as "agent certificate agent_id mismatch".
        //
        // The per-daemon `agent_cert_path` (set by `with_agent_cert_path()`)
        // is the structural fix for scenario (2); the agent_id check is the
        // defensive net in case two processes still land on the same path.
        let identity = if let Some(user_kp) = user_keypair {
            let cert_path = self.agent_cert_path.clone();
            let existing_cert = if let Some(ref p) = cert_path {
                if tokio::fs::try_exists(p).await.unwrap_or(false) {
                    storage::load_agent_certificate_from(p).await.ok()
                } else {
                    None
                }
            } else if storage::agent_certificate_exists().await {
                storage::load_agent_certificate().await.ok()
            } else {
                None
            };

            let cert_still_valid = existing_cert.as_ref().is_some_and(|c| {
                let user_match = c
                    .user_id()
                    .map(|uid| uid == user_kp.user_id())
                    .unwrap_or(false);
                let agent_match = c
                    .agent_id()
                    .map(|aid| aid == agent_keypair.agent_id())
                    .unwrap_or(false);
                user_match && agent_match
            });

            let cert = if cert_still_valid {
                existing_cert.expect("cert_still_valid implies Some")
            } else {
                let new_cert = identity::AgentCertificate::issue(&user_kp, &agent_keypair)?;
                if let Some(ref p) = cert_path {
                    storage::save_agent_certificate_to(&new_cert, p).await?;
                } else {
                    storage::save_agent_certificate(&new_cert).await?;
                }
                new_cert
            };
            identity::Identity::new_with_user(machine_keypair, agent_keypair, user_kp, cert)
        } else {
            identity::Identity::new(machine_keypair, agent_keypair)
        };

        // Open bootstrap peer cache if network will be configured
        // and the cache is not explicitly disabled by the caller.
        let bootstrap_cache = if self.network_config.is_some() && !self.disable_peer_cache {
            let cache_dir = self.peer_cache_dir.unwrap_or_else(|| {
                dirs::home_dir()
                    .unwrap_or_else(|| std::path::PathBuf::from("."))
                    .join(".x0x")
                    .join("peers")
            });
            let config = ant_quic::BootstrapCacheConfig::builder()
                .cache_dir(cache_dir)
                .min_peers_to_save(1)
                .build();
            match ant_quic::BootstrapCache::open(config).await {
                Ok(cache) => {
                    let cache = std::sync::Arc::new(cache);
                    std::sync::Arc::clone(&cache).start_maintenance();
                    Some(cache)
                }
                Err(e) => {
                    tracing::warn!("Failed to open bootstrap cache: {e}");
                    None
                }
            }
        } else {
            None
        };

        // Create network node if configured
        // Pass the machine keypair so ant-quic PeerId == MachineId (identity unification)
        let machine_keypair = {
            let pk = ant_quic::MlDsaPublicKey::from_bytes(
                identity.machine_keypair().public_key().as_bytes(),
            )
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "invalid machine public key: {e}"
                )))
            })?;
            let sk = ant_quic::MlDsaSecretKey::from_bytes(
                identity.machine_keypair().secret_key().as_bytes(),
            )
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "invalid machine secret key: {e}"
                )))
            })?;
            Some((pk, sk))
        };

        let network = if let Some(config) = self.network_config {
            let node = network::NetworkNode::new(config, bootstrap_cache.clone(), machine_keypair)
                .await
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "network initialization failed: {}",
                        e
                    )))
                })?;

            // Verify identity unification: ant-quic PeerId must equal MachineId
            debug_assert_eq!(
                node.peer_id().0,
                identity.machine_id().0,
                "ant-quic PeerId must equal MachineId after identity unification"
            );

            Some(std::sync::Arc::new(node))
        } else {
            None
        };

        // Create signing context from agent keypair for message authentication
        let signing_ctx = std::sync::Arc::new(gossip::SigningContext::from_keypair(
            identity.agent_keypair(),
        ));

        // Create gossip runtime if network exists
        let gossip_runtime = if let Some(ref net) = network {
            let runtime = gossip::GossipRuntime::new(
                gossip::GossipConfig::default(),
                std::sync::Arc::clone(net),
                Some(signing_ctx),
            )
            .await
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "gossip runtime initialization failed: {}",
                    e
                )))
            })?;
            Some(std::sync::Arc::new(runtime))
        } else {
            None
        };

        // Initialise contact store
        let contacts_path = self.contact_store_path.unwrap_or_else(|| {
            dirs::home_dir()
                .unwrap_or_else(|| std::path::PathBuf::from("."))
                .join(".x0x")
                .join("contacts.json")
        });
        let contact_store = std::sync::Arc::new(tokio::sync::RwLock::new(
            contacts::ContactStore::new(contacts_path),
        ));

        // Wrap bootstrap cache with gossip coordinator adapter (zero duplication).
        let gossip_cache_adapter = bootstrap_cache.as_ref().map(|cache| {
            saorsa_gossip_coordinator::GossipCacheAdapter::new(std::sync::Arc::clone(cache))
        });

        // Initialize direct messaging infrastructure
        let direct_messaging = std::sync::Arc::new(direct::DirectMessaging::new());

        // Create presence wrapper if network exists
        let presence = if let Some(ref net) = network {
            let peer_id = saorsa_gossip_transport::GossipTransport::local_peer_id(net.as_ref());
            let mut presence_config = presence::PresenceConfig::default();
            if let Some(secs) = self.presence_beacon_interval_secs {
                presence_config.beacon_interval_secs = secs;
            }
            if let Some(secs) = self.presence_event_poll_interval_secs {
                presence_config.event_poll_interval_secs = secs;
            }
            if let Some(secs) = self.presence_offline_timeout_secs {
                presence_config.adaptive_timeout_fallback_secs = secs;
            }
            let pw = presence::PresenceWrapper::new(
                peer_id,
                std::sync::Arc::clone(net),
                presence_config,
                bootstrap_cache.clone(),
            )
            .map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "presence initialization failed: {}",
                    e
                )))
            })?;
            let pw_arc = std::sync::Arc::new(pw);
            // Wire presence into gossip runtime for Bulk dispatch
            if let Some(ref rt) = gossip_runtime {
                rt.set_presence(std::sync::Arc::clone(&pw_arc));
            }
            Some(pw_arc)
        } else {
            None
        };

        Ok(Agent {
            identity: std::sync::Arc::new(identity),
            network,
            gossip_runtime,
            bootstrap_cache,
            gossip_cache_adapter,
            identity_discovery_cache: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
            identity_listener_started: std::sync::atomic::AtomicBool::new(false),
            heartbeat_interval_secs: self
                .heartbeat_interval_secs
                .unwrap_or(IDENTITY_HEARTBEAT_INTERVAL_SECS),
            identity_ttl_secs: self.identity_ttl_secs.unwrap_or(IDENTITY_TTL_SECS),
            heartbeat_handle: tokio::sync::Mutex::new(None),
            rendezvous_advertised: std::sync::atomic::AtomicBool::new(false),
            contact_store,
            direct_messaging,
            network_event_listener_started: std::sync::atomic::AtomicBool::new(false),
            direct_listener_started: std::sync::atomic::AtomicBool::new(false),
            presence,
            user_identity_consented: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            capability_store: std::sync::Arc::new(dm_capability::CapabilityStore::new()),
            dm_capabilities_tx: std::sync::Arc::new({
                let (tx, _rx) = tokio::sync::watch::channel(dm::DmCapabilities::pending());
                tx
            }),
            dm_inflight_acks: std::sync::Arc::new(dm::InFlightAcks::new()),
            recent_delivery_cache: std::sync::Arc::new(dm::RecentDeliveryCache::with_defaults()),
            capability_advert_service: tokio::sync::Mutex::new(None),
            dm_inbox_service: tokio::sync::Mutex::new(None),
        })
    }
}

/// Handle for interacting with a collaborative task list.
///
/// Provides a safe, concurrent interface to a TaskList backed by
/// CRDT synchronization. All operations are async and return Results.
///
/// # Example
///
/// ```ignore
/// let handle = agent.create_task_list("My List", "topic").await?;
/// let task_id = handle.add_task("Write docs".to_string(), "API docs".to_string()).await?;
/// handle.claim_task(task_id).await?;
/// handle.complete_task(task_id).await?;
/// ```
#[derive(Clone)]
pub struct TaskListHandle {
    sync: std::sync::Arc<crdt::TaskListSync>,
    agent_id: identity::AgentId,
    peer_id: saorsa_gossip_types::PeerId,
}

impl std::fmt::Debug for TaskListHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TaskListHandle")
            .field("agent_id", &self.agent_id)
            .field("peer_id", &self.peer_id)
            .finish_non_exhaustive()
    }
}

impl TaskListHandle {
    /// Add a new task to the list.
    ///
    /// # Arguments
    ///
    /// * `title` - Task title
    /// * `description` - Task description
    ///
    /// # Returns
    ///
    /// The TaskId of the created task.
    ///
    /// # Errors
    ///
    /// Returns an error if the task cannot be added.
    pub async fn add_task(
        &self,
        title: String,
        description: String,
    ) -> error::Result<crdt::TaskId> {
        let (task_id, delta) = {
            let mut list = self.sync.write().await;
            let seq = list.next_seq();
            let task_id = crdt::TaskId::new(&title, &self.agent_id, seq);
            let metadata = crdt::TaskMetadata::new(title, description, 128, self.agent_id, seq);
            let task = crdt::TaskItem::new(task_id, metadata, self.peer_id);
            list.add_task(task.clone(), self.peer_id, seq)
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "add_task failed: {}",
                        e
                    )))
                })?;
            let tag = (self.peer_id, seq);
            let delta = crdt::TaskListDelta::for_add(task_id, task, tag, list.current_version());
            (task_id, delta)
        };
        // Best-effort replication: local mutation succeeded regardless
        if let Err(e) = self.sync.publish_delta(self.peer_id, delta).await {
            tracing::warn!("failed to publish add_task delta: {}", e);
        }
        Ok(task_id)
    }

    /// Claim a task in the list.
    ///
    /// # Arguments
    ///
    /// * `task_id` - ID of the task to claim
    ///
    /// # Errors
    ///
    /// Returns an error if the task cannot be claimed.
    pub async fn claim_task(&self, task_id: crdt::TaskId) -> error::Result<()> {
        let delta = {
            let mut list = self.sync.write().await;
            let seq = list.next_seq();
            list.claim_task(&task_id, self.agent_id, self.peer_id, seq)
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "claim_task failed: {}",
                        e
                    )))
                })?;
            // Include full task so receivers can upsert if add hasn't arrived yet
            let full_task = list
                .get_task(&task_id)
                .ok_or_else(|| {
                    error::IdentityError::Storage(std::io::Error::other(
                        "task disappeared after claim",
                    ))
                })?
                .clone();
            crdt::TaskListDelta::for_state_change(task_id, full_task, list.current_version())
        };
        if let Err(e) = self.sync.publish_delta(self.peer_id, delta).await {
            tracing::warn!("failed to publish claim_task delta: {}", e);
        }
        Ok(())
    }

    /// Complete a task in the list.
    ///
    /// # Arguments
    ///
    /// * `task_id` - ID of the task to complete
    ///
    /// # Errors
    ///
    /// Returns an error if the task cannot be completed.
    pub async fn complete_task(&self, task_id: crdt::TaskId) -> error::Result<()> {
        let delta = {
            let mut list = self.sync.write().await;
            let seq = list.next_seq();
            list.complete_task(&task_id, self.agent_id, self.peer_id, seq)
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "complete_task failed: {}",
                        e
                    )))
                })?;
            let full_task = list
                .get_task(&task_id)
                .ok_or_else(|| {
                    error::IdentityError::Storage(std::io::Error::other(
                        "task disappeared after complete",
                    ))
                })?
                .clone();
            crdt::TaskListDelta::for_state_change(task_id, full_task, list.current_version())
        };
        if let Err(e) = self.sync.publish_delta(self.peer_id, delta).await {
            tracing::warn!("failed to publish complete_task delta: {}", e);
        }
        Ok(())
    }

    /// List all tasks in their current order.
    ///
    /// # Returns
    ///
    /// A vector of `TaskSnapshot` representing the current state.
    ///
    /// # Errors
    ///
    /// Returns an error if the task list cannot be read.
    pub async fn list_tasks(&self) -> error::Result<Vec<TaskSnapshot>> {
        let list = self.sync.read().await;
        let tasks = list.tasks_ordered();
        Ok(tasks
            .into_iter()
            .map(|task| TaskSnapshot {
                id: *task.id(),
                title: task.title().to_string(),
                description: task.description().to_string(),
                state: task.current_state(),
                assignee: task.assignee().copied(),
                owner: None,
                priority: task.priority(),
            })
            .collect())
    }

    /// Reorder tasks in the list.
    ///
    /// # Arguments
    ///
    /// * `task_ids` - New ordering of task IDs
    ///
    /// # Errors
    ///
    /// Returns an error if reordering fails.
    pub async fn reorder(&self, task_ids: Vec<crdt::TaskId>) -> error::Result<()> {
        let delta = {
            let mut list = self.sync.write().await;
            list.reorder(task_ids.clone(), self.peer_id).map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "reorder failed: {}",
                    e
                )))
            })?;
            crdt::TaskListDelta::for_reorder(task_ids, list.current_version())
        };
        if let Err(e) = self.sync.publish_delta(self.peer_id, delta).await {
            tracing::warn!("failed to publish reorder delta: {}", e);
        }
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// KvStore API
// ---------------------------------------------------------------------------

impl Agent {
    /// Create a new key-value store.
    ///
    /// The store is automatically synchronized to all peers subscribed
    /// to the same `topic` via gossip delta propagation.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn create_kv_store(&self, name: &str, topic: &str) -> error::Result<KvStoreHandle> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;

        let peer_id = runtime.peer_id();
        let store_id = kv::KvStoreId::from_content(name, &self.agent_id());
        let store = kv::KvStore::new(
            store_id,
            name.to_string(),
            self.agent_id(),
            kv::AccessPolicy::Signed,
        );

        let sync = kv::KvStoreSync::new(
            store,
            std::sync::Arc::clone(runtime.pubsub()),
            topic.to_string(),
            30,
        )
        .map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "kv store sync creation failed: {e}",
            )))
        })?;

        let sync = std::sync::Arc::new(sync);
        sync.start().await.map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "kv store sync start failed: {e}",
            )))
        })?;

        Ok(KvStoreHandle {
            sync,
            agent_id: self.agent_id(),
            peer_id,
        })
    }

    /// Join an existing key-value store by topic.
    ///
    /// Creates an empty store that will be populated via delta sync
    /// from peers already sharing the topic. The access policy will
    /// be learned from the first full delta received from the owner.
    ///
    /// # Errors
    ///
    /// Returns an error if the gossip runtime is not initialized.
    pub async fn join_kv_store(&self, topic: &str) -> error::Result<KvStoreHandle> {
        let runtime = self.gossip_runtime.as_ref().ok_or_else(|| {
            error::IdentityError::Storage(std::io::Error::other(
                "gossip runtime not initialized - configure agent with network first",
            ))
        })?;

        let peer_id = runtime.peer_id();
        let store_id = kv::KvStoreId::from_content(topic, &self.agent_id());
        // Use Encrypted as the most permissive default — the actual policy
        // will be set when the first delta from the owner arrives.
        let store = kv::KvStore::new(
            store_id,
            String::new(),
            self.agent_id(),
            kv::AccessPolicy::Encrypted {
                group_id: Vec::new(),
            },
        );

        let sync = kv::KvStoreSync::new(
            store,
            std::sync::Arc::clone(runtime.pubsub()),
            topic.to_string(),
            30,
        )
        .map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "kv store sync creation failed: {e}",
            )))
        })?;

        let sync = std::sync::Arc::new(sync);
        sync.start().await.map_err(|e| {
            error::IdentityError::Storage(std::io::Error::other(format!(
                "kv store sync start failed: {e}",
            )))
        })?;

        Ok(KvStoreHandle {
            sync,
            agent_id: self.agent_id(),
            peer_id,
        })
    }
}

/// Handle for interacting with a replicated key-value store.
///
/// Provides async methods for putting, getting, and removing entries.
/// Changes are automatically replicated to peers via gossip.
#[derive(Clone)]
pub struct KvStoreHandle {
    sync: std::sync::Arc<kv::KvStoreSync>,
    agent_id: identity::AgentId,
    peer_id: saorsa_gossip_types::PeerId,
}

impl std::fmt::Debug for KvStoreHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("KvStoreHandle")
            .field("agent_id", &self.agent_id)
            .field("peer_id", &self.peer_id)
            .finish_non_exhaustive()
    }
}

impl KvStoreHandle {
    /// Put a key-value pair into the store.
    ///
    /// If the key already exists, the value is updated. Changes are
    /// automatically replicated to peers via gossip.
    ///
    /// # Errors
    ///
    /// Returns an error if the value exceeds the maximum inline size (64 KB).
    pub async fn put(
        &self,
        key: String,
        value: Vec<u8>,
        content_type: String,
    ) -> error::Result<()> {
        let delta = {
            let mut store = self.sync.write().await;
            store
                .put(
                    key.clone(),
                    value.clone(),
                    content_type.clone(),
                    self.peer_id,
                )
                .map_err(|e| {
                    error::IdentityError::Storage(std::io::Error::other(format!(
                        "kv put failed: {e}",
                    )))
                })?;
            let entry = store.get(&key).cloned();
            let version = store.current_version();
            match entry {
                Some(e) => {
                    kv::KvStoreDelta::for_put(key, e, (self.peer_id, store.next_seq()), version)
                }
                None => return Ok(()), // shouldn't happen after successful put
            }
        };
        if let Err(e) = self.sync.publish_delta(self.peer_id, delta).await {
            tracing::warn!("failed to publish kv put delta: {e}");
        }
        Ok(())
    }

    /// Get a value by key.
    ///
    /// Returns `None` if the key does not exist or has been removed.
    ///
    /// # Errors
    ///
    /// Returns an error if the store cannot be read.
    pub async fn get(&self, key: &str) -> error::Result<Option<KvEntrySnapshot>> {
        let store = self.sync.read().await;
        Ok(store.get(key).map(|e| KvEntrySnapshot {
            key: e.key.clone(),
            value: e.value.clone(),
            content_hash: hex::encode(e.content_hash),
            content_type: e.content_type.clone(),
            metadata: e.metadata.clone(),
            created_at: e.created_at,
            updated_at: e.updated_at,
        }))
    }

    /// Remove a key from the store.
    ///
    /// # Errors
    ///
    /// Returns an error if the key does not exist.
    pub async fn remove(&self, key: &str) -> error::Result<()> {
        let delta = {
            let mut store = self.sync.write().await;
            store.remove(key).map_err(|e| {
                error::IdentityError::Storage(std::io::Error::other(format!(
                    "kv remove failed: {e}",
                )))
            })?;
            let mut d = kv::KvStoreDelta::new(store.current_version());
            d.removed
                .insert(key.to_string(), std::collections::HashSet::new());
            d
        };
        if let Err(e) = self.sync.publish_delta(self.peer_id, delta).await {
            tracing::warn!("failed to publish kv remove delta: {e}");
        }
        Ok(())
    }

    /// List all active keys in the store.
    ///
    /// # Errors
    ///
    /// Returns an error if the store cannot be read.
    pub async fn keys(&self) -> error::Result<Vec<KvEntrySnapshot>> {
        let store = self.sync.read().await;
        Ok(store
            .active_entries()
            .into_iter()
            .map(|e| KvEntrySnapshot {
                key: e.key.clone(),
                value: e.value.clone(),
                content_hash: hex::encode(e.content_hash),
                content_type: e.content_type.clone(),
                metadata: e.metadata.clone(),
                created_at: e.created_at,
                updated_at: e.updated_at,
            })
            .collect())
    }

    /// Get the store name.
    ///
    /// # Errors
    ///
    /// Returns an error if the store cannot be read.
    pub async fn name(&self) -> error::Result<String> {
        let store = self.sync.read().await;
        Ok(store.name().to_string())
    }
}

/// Read-only snapshot of a KvStore entry.
#[derive(Debug, Clone, serde::Serialize)]
pub struct KvEntrySnapshot {
    /// The key.
    pub key: String,
    /// The value bytes.
    pub value: Vec<u8>,
    /// BLAKE3 hash of the value (hex-encoded).
    pub content_hash: String,
    /// Content type (MIME).
    pub content_type: String,
    /// User metadata.
    pub metadata: std::collections::HashMap<String, String>,
    /// Unix milliseconds when created.
    pub created_at: u64,
    /// Unix milliseconds when last updated.
    pub updated_at: u64,
}

/// Read-only snapshot of a task's current state.
///
/// This is returned by `TaskListHandle::list_tasks()` and hides CRDT
/// internals, providing a clean API surface.
#[derive(Debug, Clone)]
pub struct TaskSnapshot {
    /// Unique task identifier.
    pub id: crdt::TaskId,
    /// Task title.
    pub title: String,
    /// Task description.
    pub description: String,
    /// Current checkbox state (Empty, Claimed, or Done).
    pub state: crdt::CheckboxState,
    /// Agent assigned to this task (if any).
    pub assignee: Option<identity::AgentId>,
    /// Human owner of the agent that created this task (if known).
    pub owner: Option<identity::UserId>,
    /// Task priority (0-255, higher = more important).
    pub priority: u8,
}

/// The x0x protocol version.
pub const VERSION: &str = env!("CARGO_PKG_VERSION");

/// The name. Three bytes. A palindrome. A philosophy.
pub const NAME: &str = "x0x";

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

    fn sa(s: &str) -> std::net::SocketAddr {
        s.parse().expect("valid SocketAddr literal in test")
    }

    #[test]
    fn is_publicly_advertisable_rejects_lan_and_special_scopes() {
        // v4 non-global scopes
        assert!(
            !is_publicly_advertisable(sa("127.0.0.1:5483")),
            "loopback v4"
        );
        assert!(!is_publicly_advertisable(sa("10.1.2.3:5483")), "rfc1918 /8");
        assert!(
            !is_publicly_advertisable(sa("172.20.0.5:5483")),
            "rfc1918 /12"
        );
        assert!(
            !is_publicly_advertisable(sa("192.168.1.5:5483")),
            "rfc1918 /16"
        );
        assert!(
            !is_publicly_advertisable(sa("169.254.1.1:5483")),
            "link-local v4"
        );
        assert!(
            !is_publicly_advertisable(sa("100.64.1.1:5483")),
            "CGNAT (unreachable outside carrier)"
        );
        assert!(
            !is_publicly_advertisable(sa("0.0.0.0:5483")),
            "unspecified v4"
        );

        // v6 non-global scopes
        assert!(!is_publicly_advertisable(sa("[::1]:5483")), "loopback v6");
        assert!(
            !is_publicly_advertisable(sa("[fe80::1]:5483")),
            "link-local v6"
        );
        assert!(!is_publicly_advertisable(sa("[fd00::1]:5483")), "ULA v6");

        // port 0 never advertisable regardless of ip scope
        assert!(
            !is_publicly_advertisable(sa("1.2.3.4:0")),
            "port 0 on global v4"
        );

        // Globally-routable positives
        assert!(is_publicly_advertisable(sa("1.2.3.4:5483")), "global v4");
        assert!(
            is_publicly_advertisable(sa("[2001:db8::1]:5483")),
            "global v6 (documentation doc but is_globally_routable permits)",
        );
        assert!(
            is_publicly_advertisable(sa("8.8.8.8:9000")),
            "global v4 on non-default port",
        );

        // Reserved documentation ranges are correctly rejected by
        // is_globally_routable even though they are not RFC1918.
        assert!(
            !is_publicly_advertisable(sa("192.0.2.1:5483")),
            "TEST-NET-1 documentation range"
        );
        assert!(
            !is_publicly_advertisable(sa("203.0.113.10:5483")),
            "TEST-NET-3 documentation range"
        );
    }

    #[test]
    fn presence_parse_addr_hints_drops_private_scopes() {
        // Older peers ship a mix of scopes. parse_addr_hints should return only
        // globally-advertisable entries so our dial loop never burns budget on
        // unreachable candidates.
        let hints = vec![
            "127.0.0.1:5483".to_string(),
            "10.200.0.1:5483".to_string(),
            "[fd00::1]:5483".to_string(),
            "1.2.3.4:5483".to_string(),
            "[2001:db8::1]:5483".to_string(),
            "not-an-address".to_string(),
        ];
        let parsed = presence::parse_addr_hints(&hints);
        let got: Vec<String> = parsed.iter().map(|a| a.to_string()).collect();
        assert_eq!(
            got,
            vec!["1.2.3.4:5483".to_string(), "[2001:db8::1]:5483".to_string()],
            "only globally-advertisable addresses survive inbound parsing"
        );
    }

    #[test]
    fn name_is_palindrome() {
        let name = NAME;
        let reversed: String = name.chars().rev().collect();
        assert_eq!(name, reversed, "x0x must be a palindrome");
    }

    #[test]
    fn name_is_three_bytes() {
        assert_eq!(NAME.len(), 3, "x0x must be exactly three bytes");
    }

    #[test]
    fn name_is_ai_native() {
        // No uppercase, no spaces, no special chars that conflict
        // with shell, YAML, Markdown, or URL encoding
        assert!(NAME.chars().all(|c| c.is_ascii_alphanumeric()));
    }

    #[tokio::test]
    async fn agent_creates() {
        let agent = Agent::new().await;
        assert!(agent.is_ok());
    }

    #[tokio::test]
    async fn agent_joins_network() {
        let agent = Agent::new().await.unwrap();
        assert!(agent.join_network().await.is_ok());
    }

    #[tokio::test]
    async fn agent_subscribes() {
        let agent = Agent::new().await.unwrap();
        // Currently returns error - will be implemented in Task 3
        assert!(agent.subscribe("test-topic").await.is_err());
    }

    #[tokio::test]
    async fn identity_announcement_machine_signature_verifies() {
        let agent = Agent::builder()
            .with_network_config(network::NetworkConfig::default())
            .build()
            .await
            .unwrap();

        let announcement = agent.build_identity_announcement(false, false).unwrap();
        assert_eq!(announcement.agent_id, agent.agent_id());
        assert_eq!(announcement.machine_id, agent.machine_id());
        assert!(announcement.user_id.is_none());
        assert!(announcement.agent_certificate.is_none());
        assert!(announcement.verify().is_ok());
    }

    #[tokio::test]
    async fn identity_announcement_requires_human_consent() {
        let agent = Agent::builder()
            .with_network_config(network::NetworkConfig::default())
            .build()
            .await
            .unwrap();

        let err = agent.build_identity_announcement(true, false).unwrap_err();
        assert!(
            err.to_string().contains("explicit human consent"),
            "unexpected error: {err}"
        );
    }

    #[tokio::test]
    async fn identity_announcement_with_user_requires_user_identity() {
        let agent = Agent::builder()
            .with_network_config(network::NetworkConfig::default())
            .build()
            .await
            .unwrap();

        let err = agent.build_identity_announcement(true, true).unwrap_err();
        assert!(
            err.to_string().contains("no user identity is configured"),
            "unexpected error: {err}"
        );
    }

    #[tokio::test]
    async fn announce_identity_populates_discovery_cache() {
        let user_key = identity::UserKeypair::generate().unwrap();
        let agent = Agent::builder()
            .with_network_config(network::NetworkConfig::default())
            .with_user_key(user_key)
            .build()
            .await
            .unwrap();

        agent.announce_identity(true, true).await.unwrap();
        let discovered = agent.discovered_agent(agent.agent_id()).await.unwrap();
        let entry = discovered.expect("agent should discover its own announcement");

        assert_eq!(entry.agent_id, agent.agent_id());
        assert_eq!(entry.machine_id, agent.machine_id());
        assert_eq!(entry.user_id, agent.user_id());
    }

    /// An announcement without NAT fields (as produced by old nodes) should still
    /// deserialise correctly via bincode — new fields are `Option` so `None` (0x00)
    /// is a valid encoding.
    #[test]
    fn identity_announcement_backward_compat_no_nat_fields() {
        use identity::{AgentId, MachineId};

        // Build an announcement that omits the nat_* fields by serializing an old-style
        // struct that matches the pre-1.3 wire format.
        #[derive(serde::Serialize, serde::Deserialize)]
        struct OldIdentityAnnouncementUnsigned {
            agent_id: AgentId,
            machine_id: MachineId,
            user_id: Option<identity::UserId>,
            agent_certificate: Option<identity::AgentCertificate>,
            machine_public_key: Vec<u8>,
            addresses: Vec<std::net::SocketAddr>,
            announced_at: u64,
        }

        let agent_id = AgentId([1u8; 32]);
        let machine_id = MachineId([2u8; 32]);
        let old = OldIdentityAnnouncementUnsigned {
            agent_id,
            machine_id,
            user_id: None,
            agent_certificate: None,
            machine_public_key: vec![0u8; 10],
            addresses: Vec::new(),
            announced_at: 1234,
        };
        let bytes = bincode::serialize(&old).expect("serialize old announcement");

        // Attempt to deserialize as the new struct — this tests that the new fields
        // (which are Option<T>) do NOT break deserialization of the old format.
        // Note: bincode 1.x is not self-describing, so adding fields to a struct DOES
        // change the wire format.  This test documents the expected behavior.
        // Old format -> new struct: will fail because new struct has more fields.
        // New format -> old struct: will have trailing bytes.
        // This is acceptable — we document the protocol change.
        let result = bincode::deserialize::<IdentityAnnouncementUnsigned>(&bytes);
        // Old nodes produce shorter messages; new nodes cannot decode them as new structs.
        // This confirms the protocol is not transparent — nodes must upgrade together.
        assert!(
            result.is_err(),
            "Old-format announcement should not decode as new struct (protocol upgrade required)"
        );
    }

    /// A new announcement with all NAT fields set round-trips through bincode.
    #[test]
    fn identity_announcement_nat_fields_round_trip() {
        use identity::{AgentId, MachineId};

        let unsigned = IdentityAnnouncementUnsigned {
            agent_id: AgentId([1u8; 32]),
            machine_id: MachineId([2u8; 32]),
            user_id: None,
            agent_certificate: None,
            machine_public_key: vec![0u8; 10],
            addresses: Vec::new(),
            announced_at: 9999,
            nat_type: Some("FullCone".to_string()),
            can_receive_direct: Some(true),
            is_relay: Some(false),
            is_coordinator: Some(true),
        };
        let bytes = bincode::serialize(&unsigned).expect("serialize");
        let decoded: IdentityAnnouncementUnsigned =
            bincode::deserialize(&bytes).expect("deserialize");
        assert_eq!(decoded.nat_type.as_deref(), Some("FullCone"));
        assert_eq!(decoded.can_receive_direct, Some(true));
        assert_eq!(decoded.is_relay, Some(false));
        assert_eq!(decoded.is_coordinator, Some(true));
    }

    /// An announcement with None for all NAT fields (e.g. network not started)
    /// round-trips correctly.
    #[test]
    fn identity_announcement_no_nat_fields_round_trip() {
        use identity::{AgentId, MachineId};

        let unsigned = IdentityAnnouncementUnsigned {
            agent_id: AgentId([3u8; 32]),
            machine_id: MachineId([4u8; 32]),
            user_id: None,
            agent_certificate: None,
            machine_public_key: vec![0u8; 10],
            addresses: Vec::new(),
            announced_at: 42,
            nat_type: None,
            can_receive_direct: None,
            is_relay: None,
            is_coordinator: None,
        };
        let bytes = bincode::serialize(&unsigned).expect("serialize");
        let decoded: IdentityAnnouncementUnsigned =
            bincode::deserialize(&bytes).expect("deserialize");
        assert!(decoded.nat_type.is_none());
        assert!(decoded.can_receive_direct.is_none());
        assert!(decoded.is_relay.is_none());
        assert!(decoded.is_coordinator.is_none());
    }
}