//! Transport and protocol handler for peat-mesh-node Automerge sync.
//!
//! Provides two main types:
//!
//! - [`MeshSyncTransport`]: implements [`SyncTransport`] by tracking active
//!   QUIC connections in a concurrent map.
//! - [`SyncProtocolHandler`]: implements `iroh::protocol::ProtocolHandler` so
//!   the Iroh Router dispatches incoming `CAP_AUTOMERGE_ALPN` connections to
//!   the sync coordinator.

use std::collections::HashMap;
use std::fmt;
use std::sync::{Arc, OnceLock, RwLock, Weak};
use std::time::Duration;

use async_trait::async_trait;
use iroh::endpoint::{Connection, PathId, PathStats};
use iroh::protocol::AcceptError;
use iroh::{Endpoint, EndpointAddr, EndpointId};
use tracing::{debug, info, warn};

use super::automerge_sync::AutomergeSyncCoordinator;
use super::iroh_blob_store::NetworkedIrohBlobStore;
use super::sync_transport::{SyncTransport, CAP_AUTOMERGE_ALPN};
use crate::security::formation_key::{
    FormationAuthResult, FormationChallenge, FormationChallengeResponse, FormationKey,
};

/// Owned snapshot of a single iroh QUIC path's state, as returned by
/// [`MeshSyncTransport::peer_paths`].
///
/// This is the iroh-1.0-era replacement for the pre-1.0 `PathInfo` type
/// that peat-mesh's API previously exposed. iroh 1.0.0-rc.0 made `Path<'_>`
/// borrow from a connection-scoped `PathList<'_>` snapshot, which forces
/// callers to thread the snapshot's lifetime through their own code.
/// `PathSnapshot` captures the same per-path bits in owned form so callers
/// can hold a `Vec<PathSnapshot>` indefinitely.
#[derive(Debug, Clone)]
pub struct PathSnapshot {
    /// Stable iroh-side identifier for this path.
    pub id: PathId,
    /// True iff this is the currently-active transmission path.
    pub is_selected: bool,
    /// True iff this path is a direct IP transport (not a relay).
    pub is_ip: bool,
    /// True iff this path is reached via a relay server.
    pub is_relay: bool,
    /// Round-trip-time estimate for this path. iroh 1.0.0-rc.0's
    /// `Path::rtt()` returns `Duration` directly (no Option).
    pub rtt: Duration,
    /// Per-path noq-proto counters (bytes sent / received, current MTU,
    /// congestion-window, loss, etc.).
    pub stats: PathStats,
}

/// Timeout for formation key authentication handshake reads and writes.
const FORMATION_AUTH_TIMEOUT: Duration = Duration::from_secs(30);

/// Version byte sent by the initiator to unblock `accept_bi()`.
///
/// iroh 0.97's `accept_bi()` will not return until the opener writes at
/// least one byte to the send stream.  This constant is the first byte the
/// initiator sends so the acceptor can detect the auth stream and,
/// in the future, negotiate protocol versions.
const FORMATION_AUTH_VERSION: u8 = 1;

// ────────────────────────────────────────────────────────────────────────────
// MeshSyncTransport
// ────────────────────────────────────────────────────────────────────────────

/// Lightweight transport for peat-mesh-node that tracks QUIC connections
/// to peers and hands them to the sync coordinator on demand.
///
/// Holds a mandatory [`FormationKey`] used to authenticate outgoing
/// connections via HMAC challenge-response.
pub struct MeshSyncTransport {
    endpoint: Endpoint,
    connections: RwLock<HashMap<EndpointId, Connection>>,
    formation_key: FormationKey,
}

impl fmt::Debug for MeshSyncTransport {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let peer_count = self.connections.read().map(|c| c.len()).unwrap_or(0);
        f.debug_struct("MeshSyncTransport")
            .field(
                "endpoint_id",
                &format_args!("{}", self.endpoint.id().fmt_short()),
            )
            .field("connected_peers", &peer_count)
            .finish()
    }
}

impl MeshSyncTransport {
    /// Create a new transport sharing the given Iroh endpoint.
    ///
    /// The `formation_key` is used to authenticate outgoing connections
    /// via HMAC challenge-response before sync streams are opened.
    pub fn new(endpoint: Endpoint, formation_key: FormationKey) -> Self {
        Self {
            endpoint,
            connections: RwLock::new(HashMap::new()),
            formation_key,
        }
    }

    /// Store (or replace) the connection for a peer.
    pub fn insert_connection(&self, peer_id: EndpointId, conn: Connection) {
        let mut conns = self.connections.write().unwrap_or_else(|e| e.into_inner());
        conns.insert(peer_id, conn);
    }

    /// Remove the connection for a peer.
    pub fn remove_connection(&self, peer_id: &EndpointId) {
        let mut conns = self.connections.write().unwrap_or_else(|e| e.into_inner());
        conns.remove(peer_id);
    }

    /// Start full-duplex sync on a connection.
    ///
    /// Stores the connection and spawns a background task that accepts
    /// incoming bidirectional streams, forwarding each to the sync
    /// coordinator. This makes the QUIC connection fully bidirectional
    /// for sync — both sides can initiate streams.
    ///
    /// Use this for **both** incoming connections (from `SyncProtocolHandler`)
    /// and outgoing connections (from `connect_peer` or discovery).
    pub fn start_sync_connection(
        self: &Arc<Self>,
        connection: Connection,
        coordinator: Arc<AutomergeSyncCoordinator>,
    ) {
        let peer_id = connection.remote_id();
        self.insert_connection(peer_id, connection.clone());

        let transport = Arc::clone(self);

        // Log initial path info for the connection. iroh 1.0.0-rc.0
        // returns a `PathList<'_>` snapshot from `paths()`; `iter()` borrows
        // path entries scoped to the snapshot's lifetime.
        let initial_path_list = connection.paths();
        let path_summary: Vec<&str> = initial_path_list
            .iter()
            .map(|p| if p.is_relay() { "relay" } else { "direct" })
            .collect();
        info!(
            peer = %peer_id.fmt_short(),
            paths = ?path_summary,
            "Sync connection established with {} path(s)",
            initial_path_list.len()
        );
        // Drop the borrow before spawning the watcher (it would otherwise
        // hold the snapshot across an await).
        drop(initial_path_list);

        // Spawn path watcher to log multipath transitions. iroh 1.0.0-rc.0
        // replaced the `PathWatcher::updated().await` polling shape with
        // `Connection::paths_stream()` — a `Stream` that yields a fresh
        // `PathList` snapshot on every change. The stream ends when the
        // connection closes, which the `while let Some(...)` loop handles.
        let path_conn = connection.clone();
        let path_peer_id = peer_id;
        tokio::spawn(async move {
            // iroh 1.0.0-rc.0's `PathListStream` implements
            // `tokio_stream::Stream` (i.e. `futures_core::Stream`); pull the
            // `next()` adapter in from the `futures` umbrella crate
            // (already in peat-mesh's direct deps) so we don't enable
            // `tokio-stream` just for this one call.
            use futures::StreamExt;
            let mut stream = path_conn.paths_stream();
            while let Some(path_list) = stream.next().await {
                let count = path_list.len();
                // Bind both fields in a single match arm so the
                // `is_relay`/`rtt()` predicates only evaluate once per
                // stream tick. (iroh 1.0.0-rc.0's `PathListIter`
                // yields `Path<'_>` by value — Clone but not Copy —
                // so `find(...)` returns `Option<Path<'_>>` and a
                // naive `.map()` twice would move out.)
                // `Path::rtt()` returns `Duration` directly in iroh 1.0
                // (was `Option<Duration>` on the retired `PathInfo`).
                let (path_type, rtt) = match path_list.iter().find(|p| p.is_selected()) {
                    Some(p) => (
                        Some(if p.is_relay() { "relay" } else { "direct" }),
                        Some(p.rtt()),
                    ),
                    None => (None, None),
                };
                info!(
                    peer = %path_peer_id.fmt_short(),
                    active_path = ?path_type,
                    rtt = ?rtt,
                    total_paths = count,
                    "Connection paths changed"
                );
            }
        });

        tokio::spawn(async move {
            loop {
                match connection.accept_bi().await {
                    Ok((send, recv)) => {
                        debug!(
                            peer = %peer_id.fmt_short(),
                            "Accepted incoming sync stream"
                        );
                        let coord = coordinator.clone();
                        tokio::spawn(async move {
                            if let Err(e) =
                                coord.handle_incoming_sync_stream(peer_id, send, recv).await
                            {
                                warn!(
                                    peer = %peer_id.fmt_short(),
                                    error = %e,
                                    "Error handling incoming sync stream"
                                );
                            }
                        });
                    }
                    Err(e) => {
                        debug!(
                            peer = %peer_id.fmt_short(),
                            error = %e,
                            "Sync connection closed"
                        );
                        break;
                    }
                }
            }

            transport.remove_connection(&peer_id);
            coordinator.clear_peer_sync_state(peer_id);
            info!(
                peer = %peer_id.fmt_short(),
                "Cleaned up sync state for disconnected peer"
            );
        });
    }

    /// Get a reference to the underlying Iroh endpoint.
    pub fn endpoint(&self) -> &Endpoint {
        &self.endpoint
    }

    /// Get multipath info for a peer connection.
    ///
    /// Returns the current set of network paths (relay, direct IPv4/IPv6)
    /// as owned [`PathSnapshot`] entries — a small peat-mesh-defined struct
    /// that captures each path's identity bits (`is_selected`, `is_ip`,
    /// `is_relay`, `rtt`) plus the iroh-side [`PathStats`] at the moment
    /// of the call. Returns `None` if the peer is not connected.
    ///
    /// iroh 1.0.0-rc.0 retired the borrowing `PathInfo` type and split
    /// path state into a lifetime-bound `Path<'_>` (predicates + `rtt()`)
    /// plus an owned `PathStats` (counters). Returning a borrowed
    /// `Vec<Path<'_>>` would force callers to thread the snapshot's
    /// lifetime; `PathSnapshot` is the owned eqivalent of the pre-1.0
    /// `PathInfo` shape.
    pub fn peer_paths(&self, peer_id: &EndpointId) -> Option<Vec<PathSnapshot>> {
        let conns = self.connections.read().unwrap_or_else(|e| e.into_inner());
        let conn = conns.get(peer_id)?;
        Some(
            conn.paths()
                .iter()
                .map(|p| PathSnapshot {
                    id: p.id(),
                    is_selected: p.is_selected(),
                    is_ip: p.is_ip(),
                    is_relay: p.is_relay(),
                    rtt: p.rtt(),
                    stats: p.stats(),
                })
                .collect(),
        )
    }

    /// Get the current round-trip time estimate for a peer.
    ///
    /// Returns the RTT of the selected (active) path, or `None` if the
    /// peer is not connected or no path is selected.
    ///
    /// **Semantics change vs. peat-mesh < rc.20 (peat-mesh#160 / peat#923):**
    /// in iroh 1.0.0-rc.0, `Path::rtt()` returns `Duration` directly (no
    /// `Option`). Pre-rc.20, this method's `None` could also signal "RTT
    /// not yet measured on the selected path" — that meaning is gone. A
    /// peer with a selected path now always returns `Some(rtt)`, with a
    /// placeholder Duration (typically zero, or iroh's initial estimate)
    /// before the first round-trip measurement lands. Consumers that
    /// previously branched on `is_some()` to detect "do we have telemetry
    /// yet?" should switch to comparing the inner `Duration` against
    /// zero / a small threshold instead.
    pub fn peer_rtt(&self, peer_id: &EndpointId) -> Option<Duration> {
        let conns = self.connections.read().unwrap_or_else(|e| e.into_inner());
        let conn = conns.get(peer_id)?;
        // iroh 1.0.0-rc.0: `Path::rtt()` returns `Duration` directly (was
        // `Option<Duration>` on the retired `PathInfo`). Use `map` not
        // `and_then` so we wrap the bare Duration in the outer Option.
        conn.paths()
            .iter()
            .find(|p| p.is_selected())
            .map(|p| p.rtt())
    }

    /// Establish an outgoing QUIC connection to a peer with mandatory
    /// formation key authentication.
    ///
    /// Opens a `CAP_AUTOMERGE_ALPN` connection and runs the connector-side
    /// HMAC challenge-response handshake before storing the connection.
    ///
    /// This variant takes only an [`EndpointId`] and relies on iroh's
    /// `address_lookup` chain to resolve the peer's socket address. In
    /// airgapped or no-public-egress deployments where DNS resolution
    /// hangs to its 30s budget, prefer [`connect_and_authenticate_with_addr`]
    /// when the caller already has a concrete [`EndpointAddr`] in hand —
    /// it bypasses the lookup chain entirely.
    ///
    /// See peat-mesh#205 for the airgapped lookup-chain failure mode.
    ///
    /// [`connect_and_authenticate_with_addr`]: MeshSyncTransport::connect_and_authenticate_with_addr
    pub async fn connect_and_authenticate(
        &self,
        peer_id: EndpointId,
    ) -> anyhow::Result<Connection> {
        let conn = self
            .endpoint
            .connect(peer_id, CAP_AUTOMERGE_ALPN)
            .await
            .map_err(|e| anyhow::anyhow!("failed to connect to peer: {e}"))?;

        respond_to_formation_auth(&self.formation_key, &conn).await?;

        info!(
            peer = %peer_id.fmt_short(),
            "outgoing sync connection authenticated via formation key"
        );

        self.insert_connection(peer_id, conn.clone());
        Ok(conn)
    }

    /// Establish an outgoing QUIC connection using a fully-populated
    /// [`EndpointAddr`], bypassing iroh's `address_lookup` chain.
    ///
    /// This is the surgical-caller variant of [`connect_and_authenticate`].
    /// When the caller already knows the peer's direct socket address
    /// (e.g. from a `peat-cli` join handshake, a `peat-node` sidecar's
    /// `dial_and_attach`, or any other out-of-band address-exchange path),
    /// passing the `EndpointAddr` directly removes the dependency on the
    /// `address_lookup` chain that backs the `EndpointId`-only variant.
    ///
    /// # When to prefer this method
    ///
    /// - **Airgapped deployments.** iroh's default lookup chain wires
    ///   `DnsAddressLookup::n0_dns()` ahead of peat-mesh's `MemoryLookup`
    ///   (the chain is additive — `.address_lookup(...)` calls `push`,
    ///   not replace). Without internet egress, the DNS lookup hangs the
    ///   full 30 s budget before falling through to `MemoryLookup`, even
    ///   when the caller has already registered the peer's address via
    ///   `MemoryLookup::add_endpoint_info`. This variant skips the chain
    ///   entirely.
    /// - **Direct-address handoffs.** Callers that exchange `EndpointAddr`
    ///   through an out-of-band channel (formation join, sidecar pairing)
    ///   already have the authoritative socket; making iroh re-resolve via
    ///   the chain adds nothing.
    ///
    /// # When to use the [`EndpointId`]-only variant instead
    ///
    /// - The caller only has the `EndpointId` and is relying on a registered
    ///   `address_lookup` source (DNS, `MemoryLookup` registered by a
    ///   separate code path, future mDNS lookup, etc.) to resolve the
    ///   address.
    ///
    /// See peat-mesh#205 for the originating airgapped failure case.
    pub async fn connect_and_authenticate_with_addr(
        &self,
        peer_addr: EndpointAddr,
    ) -> anyhow::Result<Connection> {
        let peer_id = peer_addr.id;

        // Guard the entire purpose of this method: the caller is supposed
        // to already know the socket. An empty `addrs` would silently
        // fall back through iroh's `address_lookup` chain, re-introducing
        // exactly the peat-mesh#205 failure mode this method exists to
        // prevent. Convert that latent misuse into a loud, immediate
        // error at the call site.
        if peer_addr.addrs.is_empty() {
            anyhow::bail!(
                "connect_and_authenticate_with_addr called with empty addrs for peer {}; \
                 use connect_and_authenticate(EndpointId) if you need the lookup chain",
                peer_id.fmt_short()
            );
        }

        let conn = self
            .endpoint
            .connect(peer_addr, CAP_AUTOMERGE_ALPN)
            .await
            .map_err(|e| anyhow::anyhow!("failed to connect to peer: {e}"))?;

        respond_to_formation_auth(&self.formation_key, &conn).await?;

        info!(
            peer = %peer_id.fmt_short(),
            "outgoing sync connection authenticated via formation key (direct-addr)"
        );

        self.insert_connection(peer_id, conn.clone());
        Ok(conn)
    }
}

#[async_trait]
impl SyncTransport for MeshSyncTransport {
    fn get_connection(&self, peer_id: &EndpointId) -> Option<Connection> {
        let conns = self.connections.read().unwrap_or_else(|e| e.into_inner());
        conns.get(peer_id).cloned()
    }

    /// Get an existing connection or establish a new authenticated one.
    ///
    /// Checks the cache first, pruning stale (closed) connections.
    /// If no live connection exists, opens a new `CAP_AUTOMERGE_ALPN`
    /// connection and runs formation key authentication.
    async fn get_or_connect(&self, peer_id: &EndpointId) -> anyhow::Result<Connection> {
        // Check cache, pruning closed connections
        if let Some(conn) = self.get_connection(peer_id) {
            if conn.close_reason().is_none() {
                return Ok(conn);
            }
            self.remove_connection(peer_id);
        }
        self.connect_and_authenticate(*peer_id).await
    }

    fn connected_peers(&self) -> Vec<EndpointId> {
        let mut conns = self.connections.write().unwrap_or_else(|e| e.into_inner());
        // Prune closed connections while we enumerate.
        conns.retain(|_id, conn| conn.close_reason().is_none());
        conns.keys().copied().collect()
    }
}

// ────────────────────────────────────────────────────────────────────────────
// SyncProtocolHandler
// ────────────────────────────────────────────────────────────────────────────

/// Shared slot that lets the accept path register inbound-authenticated peers
/// into the blob store's `known_peers`.
///
/// The handler is constructed — and moved into the blob store's iroh Router —
/// *before* the blob store itself exists, so it can't simply hold an
/// `Arc<NetworkedIrohBlobStore>`. The builder hands it this `Arc<OnceLock<…>>`,
/// and the construction code fills the `OnceLock` with a `Weak` ref once the
/// store is built. `Weak` so the handler never keeps the store alive.
///
/// Why it matters: without inbound registration, `known_peers` is populated
/// only by *outbound* dials (discovery / `connect_peer`). A peer that dials
/// *us* is authenticated and synced but never recorded — so it's excluded from
/// distribution targeting (`resolve_targets`) and blob-provider lookup
/// (`fetch_blob`), which is exactly why an attachment can replicate as a CRDT
/// document yet never deliver as a file. See peat-node#170.
pub type InboundPeerRegistry = Arc<OnceLock<Weak<NetworkedIrohBlobStore>>>;

/// Iroh Router protocol handler for `CAP_AUTOMERGE_ALPN`.
///
/// When the Router receives a QUIC connection with our ALPN, it calls
/// [`accept`](iroh::protocol::ProtocolHandler::accept), which delegates to
/// [`MeshSyncTransport::start_sync_connection`] to set up full-duplex
/// sync on the connection.
///
/// ## Two-Phase Gating (Layer 2)
///
/// When a [`CertificateBundle`] is configured, incoming connections are
/// validated against it before sync begins. This is "Layer 2" of the
/// two-phase connection model:
///
/// - **Layer 0**: QUIC transport (formation_secret → EndpointId)
/// - **Layer 1**: `peat/enroll/1` ALPN (no cert required)
/// - **Layer 2**: `cap/automerge/1` ALPN — this handler (cert required when configured)
pub struct SyncProtocolHandler {
    transport: Arc<MeshSyncTransport>,
    coordinator: Arc<AutomergeSyncCoordinator>,
    /// Formation key for peer authentication (mandatory).
    /// All incoming connections must pass HMAC challenge-response
    /// before sync streams are accepted.
    formation_key: FormationKey,
    /// Optional certificate bundle for Layer 2 peer validation.
    /// When set, peers must have a valid, non-expired certificate
    /// or the connection is rejected.
    certificate_bundle: Option<Arc<RwLock<crate::security::certificate::CertificateBundle>>>,
    /// Optional sink for registering inbound-authenticated peers into the blob
    /// store's `known_peers`. See [`InboundPeerRegistry`]. `None` keeps the
    /// historical behavior (inbound peers not registered).
    peer_registry: Option<InboundPeerRegistry>,
}

impl fmt::Debug for SyncProtocolHandler {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SyncProtocolHandler")
            .field("has_certificate_bundle", &self.certificate_bundle.is_some())
            .finish()
    }
}

impl SyncProtocolHandler {
    /// Create a new protocol handler with mandatory formation key authentication.
    ///
    /// All incoming connections on `CAP_AUTOMERGE_ALPN` must pass an
    /// HMAC-SHA256 challenge-response handshake before sync begins.
    /// Connections that fail authentication are rejected.
    pub fn new(
        transport: Arc<MeshSyncTransport>,
        coordinator: Arc<AutomergeSyncCoordinator>,
        formation_key: FormationKey,
    ) -> Self {
        Self {
            transport,
            coordinator,
            formation_key,
            certificate_bundle: None,
            peer_registry: None,
        }
    }

    /// Enable Layer 2 certificate validation.
    ///
    /// When a certificate bundle is set, peers without a valid certificate
    /// are rejected. There is no warn-and-allow mode — certificates are
    /// either enforced or not configured.
    pub fn with_certificate_bundle(
        mut self,
        bundle: Arc<RwLock<crate::security::certificate::CertificateBundle>>,
    ) -> Self {
        self.certificate_bundle = Some(bundle);
        self
    }

    /// Register inbound-authenticated peers into the blob store's `known_peers`.
    ///
    /// Pass the same [`InboundPeerRegistry`] whose `OnceLock` the construction
    /// code fills with a `Weak<NetworkedIrohBlobStore>` once the store exists.
    /// After a peer passes formation auth on an accepted connection, the handler
    /// calls `add_peer`, making inbound peers eligible for distribution
    /// targeting and blob fetch — not just peers we dialed. See peat-node#170.
    pub fn with_inbound_peer_registry(mut self, registry: InboundPeerRegistry) -> Self {
        self.peer_registry = Some(registry);
        self
    }
}

impl iroh::protocol::ProtocolHandler for SyncProtocolHandler {
    /// Handle an incoming QUIC connection on the `CAP_AUTOMERGE_ALPN` ALPN.
    ///
    /// Authentication layers (in order):
    /// 1. **Certificate validation** (Layer 2): If a certificate bundle is
    ///    configured, the peer's EndpointId is checked against known certificates.
    ///    Peers without valid certificates are rejected.
    /// 2. **Formation key** (mandatory): Runs HMAC-SHA256 challenge-response
    ///    handshake. Connections that fail are rejected.
    async fn accept(&self, connection: Connection) -> Result<(), AcceptError> {
        let peer = connection.remote_id();

        // Layer 2: Certificate validation (hard-reject when configured)
        if let Some(ref bundle) = self.certificate_bundle {
            let bundle = bundle.read().unwrap_or_else(|e| e.into_inner());
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_millis() as u64)
                .unwrap_or(0);
            let peer_valid = bundle.validate_peer(peer.as_bytes(), now);

            if !peer_valid {
                warn!(
                    peer = %peer.fmt_short(),
                    "peer has no valid certificate, rejecting sync connection"
                );
                connection.close(2u32.into(), b"certificate required");
                return Ok(());
            }
            debug!(peer = %peer.fmt_short(), "peer certificate validated");
        }

        // Formation key authentication (mandatory)
        match Self::run_formation_auth(&self.formation_key, &connection).await {
            Ok(()) => {
                info!(peer = %peer.fmt_short(), "peer authenticated via formation key");
            }
            Err(e) => {
                warn!(
                    peer = %peer.fmt_short(),
                    error = %e,
                    "peer failed formation key authentication, rejecting"
                );
                connection.close(1u32.into(), b"formation auth failed");
                return Ok(());
            }
        }

        // Register this inbound, now-authenticated peer into the blob store's
        // `known_peers`. Without it, `known_peers` reflects only peers WE dialed
        // (discovery / connect_peer); a peer that dialed us would sync documents
        // but stay invisible to distribution targeting and blob fetch — the
        // "attachment doc syncs but the file never arrives" bug. `add_peer` is
        // idempotent, so a re-accept is harmless. peat-node#170.
        if let Some(registry) = &self.peer_registry {
            if let Some(store) = registry.get().and_then(Weak::upgrade) {
                store.add_peer(peer).await;
                debug!(
                    peer = %peer.fmt_short(),
                    "registered inbound-authenticated peer into known_peers"
                );
            }
        }

        self.transport
            .start_sync_connection(connection, self.coordinator.clone());

        // peat-mesh#205 follow-up: push our local docs to the newly-
        // connected peer.
        //
        // Pre-fix, only the DIAL side called `sync_all_documents_with_peer`
        // (via `peat-node::start_sync` and `peat-cli::join::connect_peer`
        // on the connector's existing-docs list). The ACCEPT side
        // (this handler) registered the connection and spawned the
        // sync-receiver task but never proactively pushed its own
        // docs. That worked fine for sidecar-to-sidecar bootstrap
        // because the dialer happened to also have docs to push, but
        // it broke the moment a fresh-empty-store peer (peat-cli, or
        // any newly-spawned ephemeral consumer) dialed an established
        // sidecar — the dialer's `sync_all_documents_with_peer` ran
        // against an empty store (`Syncing 0 existing documents…`),
        // the acceptor never reciprocated, and the CLI's local store
        // stayed empty even though the connection itself was up,
        // authenticated, and idle.
        //
        // Fix: spawn a one-shot push of our local docs to this peer
        // on a tokio task so the protocol-handler's `accept` future
        // returns promptly (per iroh's `ProtocolHandler` contract —
        // `accept` should hand off long-running work, not block the
        // router's accept loop). Errors are logged but don't fail
        // the accept — partial sync is still useful.
        let coord_for_initial_sync = self.coordinator.clone();
        let peer_for_initial_sync = peer;
        tokio::spawn(async move {
            if let Err(e) = coord_for_initial_sync
                .sync_all_documents_with_peer(peer_for_initial_sync)
                .await
            {
                warn!(
                    peer = %peer_for_initial_sync.fmt_short(),
                    error = %e,
                    "initial sync_all_documents_with_peer (accept side) failed"
                );
            }
        });

        Ok(())
    }
}

impl SyncProtocolHandler {
    /// Run formation key challenge-response on an accepted QUIC connection
    /// (acceptor/server side).
    ///
    /// Waits for the connecting peer to open a bidirectional auth stream,
    /// sends a challenge nonce, reads the HMAC response, and verifies it.
    /// All I/O operations are bounded by [`FORMATION_AUTH_TIMEOUT`].
    ///
    /// Returns `Ok(())` on successful authentication.
    async fn run_formation_auth(fk: &FormationKey, connection: &Connection) -> anyhow::Result<()> {
        // Acceptor waits for the connector to open the auth stream.
        // iroh 0.97: `accept_bi()` blocks until the opener writes at least
        // one byte, so the initiator sends a version byte first.
        let (mut send, mut recv) =
            tokio::time::timeout(FORMATION_AUTH_TIMEOUT, connection.accept_bi())
                .await
                .map_err(|_| {
                    anyhow::anyhow!("formation auth timed out waiting for auth stream")
                })??;

        // Read and validate the version byte sent by the initiator.
        let mut version = [0u8; 1];
        tokio::time::timeout(FORMATION_AUTH_TIMEOUT, recv.read_exact(&mut version))
            .await
            .map_err(|_| anyhow::anyhow!("formation auth timed out reading version"))??;
        if version[0] != FORMATION_AUTH_VERSION {
            anyhow::bail!("unsupported formation auth version: {}", version[0]);
        }

        // Send challenge
        let (nonce, _expected) = fk.create_challenge();
        let challenge = FormationChallenge {
            formation_id: fk.formation_id().to_string(),
            nonce,
        };
        let challenge_bytes = challenge.to_bytes();
        tokio::time::timeout(FORMATION_AUTH_TIMEOUT, async {
            send.write_all(&(challenge_bytes.len() as u32).to_le_bytes())
                .await?;
            send.write_all(&challenge_bytes).await?;
            Ok::<(), std::io::Error>(())
        })
        .await
        .map_err(|_| anyhow::anyhow!("formation auth timed out sending challenge"))??;

        // Read response
        let mut len_buf = [0u8; 4];
        tokio::time::timeout(FORMATION_AUTH_TIMEOUT, recv.read_exact(&mut len_buf))
            .await
            .map_err(|_| anyhow::anyhow!("formation auth timed out reading response length"))??;
        let resp_len = u32::from_le_bytes(len_buf) as usize;
        if resp_len > 256 {
            anyhow::bail!("response too large: {resp_len}");
        }
        let mut resp_buf = vec![0u8; resp_len];
        tokio::time::timeout(FORMATION_AUTH_TIMEOUT, recv.read_exact(&mut resp_buf))
            .await
            .map_err(|_| anyhow::anyhow!("formation auth timed out reading response body"))??;

        let resp = FormationChallengeResponse::from_bytes(&resp_buf)
            .map_err(|e| anyhow::anyhow!("invalid response: {e}"))?;

        // Verify and send result
        if fk.verify_response(&nonce, &resp.response) {
            tokio::time::timeout(
                FORMATION_AUTH_TIMEOUT,
                send.write_all(&[FormationAuthResult::Accepted.to_byte()]),
            )
            .await
            .map_err(|_| anyhow::anyhow!("formation auth timed out sending accept"))??;
            send.finish()?;
            Ok(())
        } else {
            tokio::time::timeout(
                FORMATION_AUTH_TIMEOUT,
                send.write_all(&[FormationAuthResult::Rejected.to_byte()]),
            )
            .await
            .map_err(|_| anyhow::anyhow!("formation auth timed out sending reject"))??;
            send.finish()?;
            anyhow::bail!("HMAC verification failed")
        }
    }
}

/// Connector-side formation key authentication.
///
/// The connecting peer opens a bidirectional auth stream, reads the challenge
/// from the acceptor, computes the HMAC response, and reads back the
/// accept/reject verdict. All I/O is bounded by [`FORMATION_AUTH_TIMEOUT`].
///
/// Call this immediately after establishing a QUIC connection to a peer that
/// requires formation key authentication (before opening sync streams).
///
/// Returns `Ok(())` if the acceptor verified our response successfully.
pub async fn respond_to_formation_auth(
    fk: &FormationKey,
    connection: &Connection,
) -> anyhow::Result<()> {
    // Connector opens the auth stream.
    let (mut send, mut recv) = tokio::time::timeout(FORMATION_AUTH_TIMEOUT, connection.open_bi())
        .await
        .map_err(|_| anyhow::anyhow!("formation auth timed out opening auth stream"))??;

    // Write version byte so the acceptor's `accept_bi()` can complete.
    // iroh 0.97 requires the opener to write before the peer can accept.
    tokio::time::timeout(
        FORMATION_AUTH_TIMEOUT,
        send.write_all(&[FORMATION_AUTH_VERSION]),
    )
    .await
    .map_err(|_| anyhow::anyhow!("formation auth timed out sending version"))??;

    // Read challenge from acceptor
    let mut len_buf = [0u8; 4];
    tokio::time::timeout(FORMATION_AUTH_TIMEOUT, recv.read_exact(&mut len_buf))
        .await
        .map_err(|_| anyhow::anyhow!("formation auth timed out reading challenge length"))??;
    let challenge_len = u32::from_le_bytes(len_buf) as usize;
    if challenge_len > 1024 {
        anyhow::bail!("challenge too large: {challenge_len}");
    }
    let mut challenge_buf = vec![0u8; challenge_len];
    tokio::time::timeout(FORMATION_AUTH_TIMEOUT, recv.read_exact(&mut challenge_buf))
        .await
        .map_err(|_| anyhow::anyhow!("formation auth timed out reading challenge body"))??;

    let challenge = FormationChallenge::from_bytes(&challenge_buf)
        .map_err(|e| anyhow::anyhow!("invalid challenge: {e}"))?;

    // Verify formation ID matches
    if challenge.formation_id != fk.formation_id() {
        anyhow::bail!(
            "formation ID mismatch: expected {}, got {}",
            fk.formation_id(),
            challenge.formation_id
        );
    }

    // Compute and send HMAC response
    let response = fk.respond_to_challenge(&challenge.nonce);
    let resp = FormationChallengeResponse { response };
    let resp_bytes = resp.to_bytes();
    tokio::time::timeout(FORMATION_AUTH_TIMEOUT, async {
        send.write_all(&(resp_bytes.len() as u32).to_le_bytes())
            .await?;
        send.write_all(&resp_bytes).await?;
        Ok::<(), std::io::Error>(())
    })
    .await
    .map_err(|_| anyhow::anyhow!("formation auth timed out sending response"))??;

    // Read accept/reject verdict
    let mut verdict = [0u8; 1];
    tokio::time::timeout(FORMATION_AUTH_TIMEOUT, recv.read_exact(&mut verdict))
        .await
        .map_err(|_| anyhow::anyhow!("formation auth timed out reading verdict"))??;

    match FormationAuthResult::from_byte(verdict[0]) {
        FormationAuthResult::Accepted => Ok(()),
        FormationAuthResult::Rejected => {
            anyhow::bail!("formation key authentication rejected by acceptor")
        }
    }
}

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

    /// Create a test `FormationKey` for use in unit tests.
    fn test_formation_key() -> FormationKey {
        FormationKey::new("test-formation", &[42u8; 32])
    }

    #[test]
    fn test_formation_auth_timeout_is_reasonable() {
        assert_eq!(FORMATION_AUTH_TIMEOUT, Duration::from_secs(30));
        assert!(
            FORMATION_AUTH_TIMEOUT >= Duration::from_secs(5),
            "Auth timeout should be at least 5s for slow networks"
        );
        assert!(
            FORMATION_AUTH_TIMEOUT <= Duration::from_secs(120),
            "Auth timeout should not exceed 2 minutes"
        );
    }

    /// `MeshSyncTransport` and `SyncProtocolHandler` require a running Iroh
    /// `Endpoint` (QUIC stack) to construct. Unit tests for the full protocol
    /// flow are covered in integration tests (tests/hierarchy_e2e.rs,
    /// tests/dual_active_transport_e2e.rs). Tests here verify configuration
    /// values and type-level properties.

    #[test]
    fn test_sync_protocol_handler_debug_impl() {
        // Verify the Debug impl exists and is callable (compile-time + runtime check).
        let _: fn(&SyncProtocolHandler, &mut std::fmt::Formatter) -> std::fmt::Result =
            <SyncProtocolHandler as std::fmt::Debug>::fmt;
    }

    #[test]
    fn test_mesh_sync_transport_debug_impl() {
        let _: fn(&MeshSyncTransport, &mut std::fmt::Formatter) -> std::fmt::Result =
            <MeshSyncTransport as std::fmt::Debug>::fmt;
    }

    #[test]
    fn test_formation_key_can_create_challenge_response() {
        // Test that the FormationKey type works for challenge-response without
        // needing a QUIC connection.
        let fk = FormationKey::new("test-formation", &[1u8; 32]);

        // Create challenge
        let (nonce, _expected_hmac) = fk.create_challenge();
        assert_eq!(nonce.len(), 32, "Challenge nonce should be 32 bytes");

        // Generate response
        let response = fk.respond_to_challenge(&nonce);
        assert_eq!(response.len(), 32, "HMAC response should be 32 bytes");

        // Verify round-trip
        assert!(
            fk.verify_response(&nonce, &response),
            "Response should verify against same formation key"
        );
    }

    #[test]
    fn test_formation_key_rejects_wrong_response() {
        let fk = FormationKey::new("test-formation", &[1u8; 32]);
        let (nonce, _) = fk.create_challenge();

        // Tampered response
        let mut bad_response = fk.respond_to_challenge(&nonce);
        bad_response[0] ^= 0xff;

        assert!(
            !fk.verify_response(&nonce, &bad_response),
            "Tampered response should fail verification"
        );
    }

    #[test]
    fn test_formation_key_different_formations_incompatible() {
        let fk_a = FormationKey::new("formation-alpha", &[2u8; 32]);
        let fk_b = FormationKey::new("formation-beta", &[3u8; 32]);

        let (nonce, _) = fk_a.create_challenge();
        let response_b = fk_b.respond_to_challenge(&nonce);

        assert!(
            !fk_a.verify_response(&nonce, &response_b),
            "Different formation keys should not verify each other"
        );
    }

    #[test]
    fn test_formation_challenge_serialization_roundtrip() {
        let fk = FormationKey::new("roundtrip-test", &[4u8; 32]);
        let (nonce, _) = fk.create_challenge();

        let challenge = FormationChallenge {
            formation_id: fk.formation_id().to_string(),
            nonce,
        };

        let bytes = challenge.to_bytes();
        assert!(!bytes.is_empty());

        let decoded = FormationChallenge::from_bytes(&bytes).unwrap();
        assert_eq!(decoded.formation_id, fk.formation_id());
        assert_eq!(decoded.nonce, nonce);
    }

    #[test]
    fn test_formation_challenge_response_serialization_roundtrip() {
        let fk = FormationKey::new("roundtrip-test", &[4u8; 32]);
        let (nonce, _) = fk.create_challenge();
        let response = fk.respond_to_challenge(&nonce);

        let resp = FormationChallengeResponse { response };
        let bytes = resp.to_bytes();
        assert!(!bytes.is_empty());

        let decoded = FormationChallengeResponse::from_bytes(&bytes).unwrap();
        assert_eq!(decoded.response, response);
    }

    #[test]
    fn test_formation_auth_result_byte_encoding() {
        assert_eq!(
            FormationAuthResult::from_byte(FormationAuthResult::Accepted.to_byte()),
            FormationAuthResult::Accepted
        );
        assert_eq!(
            FormationAuthResult::from_byte(FormationAuthResult::Rejected.to_byte()),
            FormationAuthResult::Rejected
        );
    }

    // ---- Multipath / noq feature tests ----
    //
    // QUIC endpoint tests must run sequentially — parallel bind() calls
    // compete for ports and cause intermittent timeouts under load.
    static QUIC_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    /// Helper: create a pair of connected iroh endpoints for testing.
    ///
    /// Returns (endpoint_a, endpoint_b, connection_from_b_to_a, router_handle).
    /// The router handle must be kept alive for the duration of the test.
    async fn create_connected_endpoints() -> (Endpoint, Endpoint, Connection, Router) {
        use iroh::address_lookup::memory::MemoryLookup;
        use iroh::protocol::Router;
        use std::sync::atomic::{AtomicBool, Ordering};

        let timeout = Duration::from_secs(10);
        let alpn: &[u8] = b"test/mesh/1";

        // Minimal protocol handler that just accepts
        #[derive(Debug)]
        struct AcceptAll(Arc<AtomicBool>);
        impl iroh::protocol::ProtocolHandler for AcceptAll {
            async fn accept(&self, _conn: Connection) -> Result<(), iroh::protocol::AcceptError> {
                self.0.store(true, Ordering::SeqCst);
                Ok(())
            }
        }

        let accepted = Arc::new(AtomicBool::new(false));

        // Build endpoints using empty_builder to avoid external DNS/relay
        // dependencies in tests.
        let lookup_a = MemoryLookup::new();
        let endpoint_a = tokio::time::timeout(
            timeout,
            Endpoint::builder(iroh::endpoint::presets::Empty)
                .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                .address_lookup(lookup_a.clone())
                .secret_key(iroh::SecretKey::from_bytes(&[10u8; 32]))
                .bind(),
        )
        .await
        .expect("endpoint_a bind timed out")
        .unwrap();

        // Router on A accepts incoming connections
        let router_a = Router::builder(endpoint_a.clone())
            .accept(alpn, AcceptAll(accepted.clone()))
            .spawn();

        let lookup_b = MemoryLookup::new();
        let endpoint_b = tokio::time::timeout(
            timeout,
            Endpoint::builder(iroh::endpoint::presets::Empty)
                .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                .address_lookup(lookup_b.clone())
                .secret_key(iroh::SecretKey::from_bytes(&[11u8; 32]))
                .bind(),
        )
        .await
        .expect("endpoint_b bind timed out")
        .unwrap();

        // Tell B about A's full endpoint address
        lookup_b.add_endpoint_info(endpoint_a.addr());

        // B connects to A
        let conn = tokio::time::timeout(timeout, endpoint_b.connect(endpoint_a.id(), alpn))
            .await
            .expect("connect timed out")
            .unwrap();

        (endpoint_a, endpoint_b, conn, router_a)
    }

    /// `peer_paths()` returns path info for a connected peer.
    #[tokio::test]
    async fn test_peer_paths_returns_paths_for_connected_peer() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let (endpoint_a, endpoint_b, conn, _router) = create_connected_endpoints().await;

        let transport = MeshSyncTransport::new(endpoint_b.clone(), test_formation_key());
        let peer_id = conn.remote_id();
        transport.insert_connection(peer_id, conn);

        let paths = transport.peer_paths(&peer_id);
        assert!(paths.is_some(), "Should return paths for connected peer");

        let paths = paths.unwrap();
        assert!(!paths.is_empty(), "Should have at least one path");

        // At least one path should be selected (PathSnapshot exposes
        // the predicates as public fields, not methods — see
        // crate::storage::mesh_sync_transport::PathSnapshot).
        let selected = paths.iter().filter(|p| p.is_selected).count();
        assert!(selected >= 1, "At least one path should be selected");

        // Local connections should be direct (IP), not relay.
        let has_ip = paths.iter().any(|p| p.is_ip);
        assert!(has_ip, "Local connection should have an IP path");

        endpoint_a.close().await;
        endpoint_b.close().await;
    }

    /// `peer_paths()` returns `None` for unknown peers.
    #[tokio::test]
    async fn test_peer_paths_returns_none_for_unknown_peer() {
        let endpoint = Endpoint::builder(iroh::endpoint::presets::Empty)
            .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
            .secret_key(iroh::SecretKey::from_bytes(&[20u8; 32]))
            .bind()
            .await
            .unwrap();

        let transport = MeshSyncTransport::new(endpoint.clone(), test_formation_key());
        let unknown_peer = iroh::SecretKey::from_bytes(&[99u8; 32]).public();

        assert!(transport.peer_paths(&unknown_peer).is_none());
        assert!(transport.peer_rtt(&unknown_peer).is_none());

        endpoint.close().await;
    }

    /// `peer_rtt()` returns a duration for a connected peer.
    #[tokio::test]
    async fn test_peer_rtt_returns_value_for_connected_peer() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let (endpoint_a, endpoint_b, conn, _router) = create_connected_endpoints().await;

        let transport = MeshSyncTransport::new(endpoint_b.clone(), test_formation_key());
        let peer_id = conn.remote_id();
        transport.insert_connection(peer_id, conn);

        // RTT may not be immediately available (needs a round-trip),
        // but peer_rtt should not panic and should return Some for
        // an established local connection.
        let rtt = transport.peer_rtt(&peer_id);
        // Local connections have near-zero RTT; it should be available
        // after the QUIC handshake.
        if let Some(rtt) = rtt {
            assert!(
                rtt < Duration::from_secs(5),
                "Local RTT should be well under 5s, got {:?}",
                rtt
            );
        }
        // Note: RTT may be None if no data has been sent yet on the
        // selected path — this is acceptable.

        endpoint_a.close().await;
        endpoint_b.close().await;
    }

    /// PathSnapshot exposes path type (relay vs direct) correctly.
    #[tokio::test]
    async fn test_path_info_type_detection() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let (endpoint_a, endpoint_b, conn, _router) = create_connected_endpoints().await;

        let transport = MeshSyncTransport::new(endpoint_b.clone(), test_formation_key());
        let peer_id = conn.remote_id();
        transport.insert_connection(peer_id, conn);

        let paths = transport.peer_paths(&peer_id).unwrap();
        for path in &paths {
            // Each path must be exactly one type — PathSnapshot's
            // predicates are public fields, not methods.
            assert!(
                path.is_ip || path.is_relay,
                "Path should be either IP or relay",
            );
        }

        endpoint_a.close().await;
        endpoint_b.close().await;
    }

    /// PathSnapshot's `id` and `stats` fields are populated from iroh's
    /// `Path::id()` / `Path::stats()` (peat-mesh#160 QA Suggestion). The
    /// predicate fields already have surface coverage above; this pin
    /// catches the case where iroh's `Path::id`/`Path::stats` signatures
    /// shift in a future bump (returning different inner types, taking
    /// arguments, etc.) — peat-mesh's re-export shape stays stable only
    /// if both keep returning shapes the producer copies as-is.
    #[tokio::test]
    async fn test_path_snapshot_id_and_stats_pin() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let (endpoint_a, endpoint_b, conn, _router) = create_connected_endpoints().await;

        let transport = MeshSyncTransport::new(endpoint_b.clone(), test_formation_key());
        let peer_id = conn.remote_id();
        transport.insert_connection(peer_id, conn);

        let paths = transport.peer_paths(&peer_id).unwrap();
        assert!(!paths.is_empty(), "must have at least one path");

        // `PathStats::udp_tx` / `udp_rx` are the UDP-counter sub-stats
        // iroh exposes on every path — exercising them pins the shape
        // of the re-exported `iroh::endpoint::PathStats` as the same
        // counter struct peat-mesh callers expect. (Pre-rc.20 there
        // was a `bytes_sent` field; iroh 1.0 reshaped to nested
        // `UdpStats` accessors.)
        let _ = &paths[0].stats.udp_tx;
        let _ = &paths[0].stats.udp_rx;

        // `PathId` is `Copy + Eq + Hash`. Pin equality + hashing so a
        // future iroh change that strips one of those derives fails
        // this test rather than breaking downstream consumers that
        // dedupe paths by id.
        let id_copy = paths[0].id;
        assert_eq!(paths[0].id, id_copy, "PathId must be Eq");
        let mut set = std::collections::HashSet::new();
        set.insert(id_copy);
        assert!(set.contains(&id_copy), "PathId must be Hash");

        endpoint_a.close().await;
        endpoint_b.close().await;
    }

    // ---- Formation auth handshake integration tests ----

    /// Helper: set up two iroh endpoints where endpoint_a runs a protocol
    /// handler that performs the acceptor-side formation auth, and return
    /// the connection from endpoint_b (the initiator).
    async fn create_formation_auth_endpoints(
        fk: FormationKey,
    ) -> (
        Endpoint,
        Endpoint,
        Connection,
        Arc<std::sync::atomic::AtomicBool>,
        Router,
    ) {
        use crate::storage::sync_transport::CAP_AUTOMERGE_ALPN;
        use iroh::address_lookup::memory::MemoryLookup;
        use iroh::protocol::Router;
        use std::sync::atomic::AtomicBool;

        let authenticated = Arc::new(AtomicBool::new(false));

        // Protocol handler that runs the acceptor-side formation auth.
        #[derive(Debug)]
        struct AuthAcceptor {
            fk: FormationKey,
            authenticated: Arc<AtomicBool>,
        }

        impl iroh::protocol::ProtocolHandler for AuthAcceptor {
            async fn accept(
                &self,
                connection: Connection,
            ) -> Result<(), iroh::protocol::AcceptError> {
                match SyncProtocolHandler::run_formation_auth(&self.fk, &connection).await {
                    Ok(()) => {
                        self.authenticated
                            .store(true, std::sync::atomic::Ordering::SeqCst);
                    }
                    Err(e) => {
                        connection.close(1u32.into(), b"formation auth failed");
                        return Err(iroh::protocol::AcceptError::from_err(std::io::Error::new(
                            std::io::ErrorKind::PermissionDenied,
                            e.to_string(),
                        )));
                    }
                }
                // Keep the connection alive so the verdict byte is delivered
                // before the Router drops the Connection.  In production,
                // SyncProtocolHandler::accept() hands off to
                // start_sync_connection() which holds the connection open.
                connection.closed().await;
                Ok(())
            }
        }

        let timeout = Duration::from_secs(10);

        let lookup_a = MemoryLookup::new();
        let endpoint_a = tokio::time::timeout(
            timeout,
            Endpoint::builder(iroh::endpoint::presets::Empty)
                .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                .address_lookup(lookup_a.clone())
                .secret_key(iroh::SecretKey::from_bytes(&[30u8; 32]))
                .bind(),
        )
        .await
        .expect("endpoint_a bind timed out")
        .unwrap();

        let router_a = Router::builder(endpoint_a.clone())
            .accept(
                CAP_AUTOMERGE_ALPN,
                AuthAcceptor {
                    fk: fk.clone(),
                    authenticated: authenticated.clone(),
                },
            )
            .spawn();

        let lookup_b = MemoryLookup::new();
        let endpoint_b = tokio::time::timeout(
            timeout,
            Endpoint::builder(iroh::endpoint::presets::Empty)
                .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                .address_lookup(lookup_b.clone())
                .secret_key(iroh::SecretKey::from_bytes(&[31u8; 32]))
                .bind(),
        )
        .await
        .expect("endpoint_b bind timed out")
        .unwrap();

        lookup_b.add_endpoint_info(endpoint_a.addr());

        let conn = tokio::time::timeout(
            timeout,
            endpoint_b.connect(endpoint_a.id(), CAP_AUTOMERGE_ALPN),
        )
        .await
        .expect("connect timed out")
        .unwrap();

        (endpoint_a, endpoint_b, conn, authenticated, router_a)
    }

    /// Full formation auth handshake completes over a live iroh connection.
    ///
    /// This is the key regression test for peat#759: prior to the fix,
    /// `accept_bi()` would deadlock because the initiator never wrote to
    /// the send stream before reading.
    #[tokio::test]
    async fn test_formation_auth_handshake_over_iroh_connection() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let fk = test_formation_key();
        let (endpoint_a, endpoint_b, conn, authenticated, _router) =
            create_formation_auth_endpoints(fk.clone()).await;

        // Initiator runs the connector-side handshake.
        let result = tokio::time::timeout(
            Duration::from_secs(10),
            respond_to_formation_auth(&fk, &conn),
        )
        .await;

        assert!(
            result.is_ok(),
            "Handshake should complete within 10s (not deadlock)"
        );
        let inner = result.unwrap();
        assert!(
            inner.is_ok(),
            "Handshake should succeed with matching keys: {:?}",
            inner.unwrap_err()
        );

        // Give the acceptor a moment to store the flag.
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(
            authenticated.load(std::sync::atomic::Ordering::SeqCst),
            "Acceptor should have authenticated the peer"
        );

        endpoint_a.close().await;
        endpoint_b.close().await;
    }

    /// Handshake fails when initiator uses a different formation key.
    #[tokio::test]
    async fn test_formation_auth_rejects_wrong_key() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let acceptor_fk = test_formation_key();
        let initiator_fk = FormationKey::new("wrong-formation", &[99u8; 32]);

        let (endpoint_a, endpoint_b, conn, authenticated, _router) =
            create_formation_auth_endpoints(acceptor_fk).await;

        let result = tokio::time::timeout(
            Duration::from_secs(10),
            respond_to_formation_auth(&initiator_fk, &conn),
        )
        .await;

        assert!(result.is_ok(), "Should complete without deadlock");
        // The initiator should get an error (either formation ID mismatch
        // or rejection verdict).
        assert!(
            result.unwrap().is_err(),
            "Handshake should fail with mismatched keys"
        );
        assert!(
            !authenticated.load(std::sync::atomic::Ordering::SeqCst),
            "Acceptor should NOT have authenticated the peer"
        );

        endpoint_a.close().await;
        endpoint_b.close().await;
    }
}
