net-mesh 0.24.0

//! Three-node integration tests for Net mesh protocol.
//!
//! These tests verify properties that only emerge with 3+ nodes:
//! forwarding, rerouting, multi-path communication, and bidirectional
//! simultaneous data flow. Each "node" is represented by two
//! `NetAdapter` instances (one per peer connection), since the adapter
//! is point-to-point.
//!
//! See `docs/THREE_NODE_TEST_PLAN.md` for the full plan.
//!
//! Run:
//!   cargo test --features net --test three_node_integration

#![cfg(feature = "net")]
#![allow(
    clippy::disallowed_methods,
    reason = "test code legitimately uses std::sync::Mutex for SUT setup; no real poison concern"
)]

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use net::adapter::net::{NetAdapterConfig, StaticKeypair};
use net::adapter::Adapter;
use net::event::{batch_process_nonce, Batch, InternalEvent};
use tokio::net::UdpSocket;
use tokio::sync::Barrier;

/// Buffer size for tests
const TEST_BUFFER_SIZE: usize = 256 * 1024;

/// Keypair and identity for a test node.
struct NodeIdentity {
    keypair: StaticKeypair,
    #[allow(dead_code)]
    port: u16,
    addr: SocketAddr,
}

impl NodeIdentity {
    fn new(port: u16) -> Self {
        let keypair = StaticKeypair::generate();
        let addr: SocketAddr = format!("127.0.0.1:{}", port).parse().unwrap();
        Self {
            keypair,
            port,
            addr,
        }
    }
}

/// A connected adapter pair (one link in the triangle).
struct Link {
    initiator: net::adapter::net::NetAdapter,
    responder: net::adapter::net::NetAdapter,
}

/// Find N available UDP ports by binding to `:0` and reading the
/// assigned port.
///
/// # Residual race
///
/// This test exercises the `NetAdapter` layer directly, whose
/// config API takes both `bind_addr` and `peer_addr` up front —
/// peers need each other's concrete port before anyone binds. So
/// unlike the `MeshNode`-based suites (which pass `:0` all the
/// way through and read `local_addr()` post-bind), this harness
/// has to pre-reserve ports, drop the sockets, then let
/// `NetAdapter::new` re-bind. That leaves a microsecond-wide
/// TOCTOU window on loopback where a parallel process could grab
/// a port between drop and re-bind.
///
/// Earlier revisions slept for 10 ms between drop and return,
/// which *widened* the window without buying anything — UDP
/// sockets have no TIME_WAIT on unix. The sleep is gone;
/// `local_addr()` is read immediately before the socket drops
/// and the caller binds the adapter on the very next scheduler
/// tick. Collapsing the race to zero requires handing a
/// pre-bound socket into `NetAdapter`, which is a bigger API
/// change than this test file is worth.
async fn find_ports(n: usize) -> Vec<u16> {
    let mut ports = Vec::with_capacity(n);
    let mut sockets = Vec::with_capacity(n);
    for _ in 0..n {
        let sock = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        ports.push(sock.local_addr().unwrap().port());
        sockets.push(sock);
    }
    drop(sockets);
    ports
}

/// Create an initiator/responder adapter config pair for two nodes.
fn create_link_configs(
    initiator: &NodeIdentity,
    responder: &NodeIdentity,
    psk: &[u8; 32],
) -> (NetAdapterConfig, NetAdapterConfig) {
    let init_cfg = NetAdapterConfig::initiator(
        initiator.addr,
        responder.addr,
        *psk,
        responder.keypair.public,
    )
    .with_handshake(3, Duration::from_secs(3))
    .with_heartbeat_interval(Duration::from_millis(200))
    .with_session_timeout(Duration::from_secs(5))
    .with_socket_buffers(TEST_BUFFER_SIZE, TEST_BUFFER_SIZE);

    let resp_cfg = NetAdapterConfig::responder(
        responder.addr,
        initiator.addr,
        *psk,
        responder.keypair.clone(),
    )
    .with_handshake(3, Duration::from_secs(3))
    .with_heartbeat_interval(Duration::from_millis(200))
    .with_session_timeout(Duration::from_secs(5))
    .with_socket_buffers(TEST_BUFFER_SIZE, TEST_BUFFER_SIZE);

    (init_cfg, resp_cfg)
}

/// Perform a handshake between two adapters concurrently. Returns the connected pair.
async fn connect_link(init_cfg: NetAdapterConfig, resp_cfg: NetAdapterConfig) -> Link {
    let barrier = Arc::new(Barrier::new(2));

    let rb = barrier.clone();
    let resp_handle = tokio::spawn(async move {
        let mut adapter = net::adapter::net::NetAdapter::new(resp_cfg).unwrap();
        rb.wait().await;
        adapter.init().await.expect("responder init failed");
        adapter
    });

    let ib = barrier.clone();
    let init_handle = tokio::spawn(async move {
        let mut adapter = net::adapter::net::NetAdapter::new(init_cfg).unwrap();
        ib.wait().await;
        adapter.init().await.expect("initiator init failed");
        adapter
    });

    let timeout = Duration::from_secs(10);
    let (resp, init) =
        tokio::time::timeout(timeout, futures::future::join(resp_handle, init_handle))
            .await
            .expect("handshake timed out");

    Link {
        initiator: init.expect("initiator task panicked"),
        responder: resp.expect("responder task panicked"),
    }
}

/// Three-node triangle: each node has links to the other two.
///
/// ```text
///          Node A
///         /      \
///    link_ab    link_ac
///       /          \
///   Node B ——————— Node C
///         link_bc
/// ```
///
/// Each link is an independent encrypted session. Each adapter in a link
/// binds to a unique port, so 6 adapters total (2 per link, 3 links).
struct Triangle {
    /// A↔B link (A=initiator, B=responder)
    link_ab: Link,
    /// A↔C link (A=initiator, C=responder)
    link_ac: Link,
    /// B↔C link (B=initiator, C=responder)
    link_bc: Link,
}

impl Triangle {
    /// Set up a fully connected three-node triangle.
    async fn setup() -> Self {
        // Each link needs its own pair of ports (6 total).
        // Links are independent encrypted sessions.
        let ports = find_ports(6).await;
        let psk = [0x42u8; 32];

        // Node identities (each node appears in 2 links with different ports)
        let a_for_ab = NodeIdentity::new(ports[0]);
        let b_for_ab = NodeIdentity::new(ports[1]);
        let a_for_ac = NodeIdentity::new(ports[2]);
        let c_for_ac = NodeIdentity::new(ports[3]);
        let b_for_bc = NodeIdentity::new(ports[4]);
        let c_for_bc = NodeIdentity::new(ports[5]);

        let (ab_init, ab_resp) = create_link_configs(&a_for_ab, &b_for_ab, &psk);
        let (ac_init, ac_resp) = create_link_configs(&a_for_ac, &c_for_ac, &psk);
        let (bc_init, bc_resp) = create_link_configs(&b_for_bc, &c_for_bc, &psk);

        // Connect all three links concurrently
        let (link_ab, link_ac, link_bc) = tokio::join!(
            connect_link(ab_init, ab_resp),
            connect_link(ac_init, ac_resp),
            connect_link(bc_init, bc_resp),
        );

        Triangle {
            link_ab,
            link_ac,
            link_bc,
        }
    }

    /// Shut down all 6 adapters.
    async fn shutdown(self) {
        let futs = vec![
            self.link_ab.initiator.shutdown(),
            self.link_ab.responder.shutdown(),
            self.link_ac.initiator.shutdown(),
            self.link_ac.responder.shutdown(),
            self.link_bc.initiator.shutdown(),
            self.link_bc.responder.shutdown(),
        ];
        for fut in futs {
            let _ = fut.await;
        }
    }
}

/// Create a batch of test events for a shard.
fn make_batch(shard_id: u16, count: usize, tag: &str) -> Batch {
    let events: Vec<InternalEvent> = (0..count)
        .map(|i| {
            InternalEvent::from_value(
                serde_json::json!({"tag": tag, "index": i}),
                i as u64,
                shard_id,
            )
        })
        .collect();

    Batch {
        shard_id,
        events,
        sequence_start: 0,
        process_nonce: batch_process_nonce(),
    }
}

// ============================================================================
// Phase 1: Mesh Formation
// ============================================================================

/// 1.1 — All 3 pairwise Noise handshakes complete within timeout.
///
/// Verifies key exchange when each node must manage 2 concurrent sessions.
/// This is the foundation — if this fails, nothing else works.
#[tokio::test]
async fn test_three_node_handshake() {
    let triangle = Triangle::setup().await;

    // If we got here, all 3 handshakes completed successfully.
    // Verify all adapters report healthy.
    assert!(
        triangle.link_ab.initiator.is_healthy().await,
        "A→B initiator unhealthy"
    );
    assert!(
        triangle.link_ab.responder.is_healthy().await,
        "A→B responder unhealthy"
    );
    assert!(
        triangle.link_ac.initiator.is_healthy().await,
        "A→C initiator unhealthy"
    );
    assert!(
        triangle.link_ac.responder.is_healthy().await,
        "A→C responder unhealthy"
    );
    assert!(
        triangle.link_bc.initiator.is_healthy().await,
        "B→C initiator unhealthy"
    );
    assert!(
        triangle.link_bc.responder.is_healthy().await,
        "B→C responder unhealthy"
    );

    triangle.shutdown().await;
}

/// 1.2 — After shutting down one node, the other two remain healthy.
///
/// Simulates node C dying. The A↔B link should remain healthy and
/// functional. This proves sessions are independent — a dead peer on
/// one link doesn't poison another.
#[tokio::test]
async fn test_three_node_health_after_one_shutdown() {
    let triangle = Triangle::setup().await;

    // Shut down "node C" (both of C's adapters)
    let _ = triangle.link_ac.responder.shutdown().await;
    let _ = triangle.link_bc.responder.shutdown().await;

    // Give heartbeats time to detect the loss
    tokio::time::sleep(Duration::from_millis(300)).await;

    // A↔B link should still be healthy
    assert!(
        triangle.link_ab.initiator.is_healthy().await,
        "A→B should remain healthy after C dies"
    );
    assert!(
        triangle.link_ab.responder.is_healthy().await,
        "B→A should remain healthy after C dies"
    );

    // Clean up remaining adapters
    let _ = triangle.link_ab.initiator.shutdown().await;
    let _ = triangle.link_ab.responder.shutdown().await;
    let _ = triangle.link_ac.initiator.shutdown().await;
    let _ = triangle.link_bc.initiator.shutdown().await;
}

// ============================================================================
// Phase 1: Data Flow
// ============================================================================

/// 2.1 — A sends events to B over the A↔B link; B receives them.
///
/// Basic point-to-point data flow in a three-node topology. Proves
/// that sending over one link doesn't interfere with the other links.
#[tokio::test]
async fn test_data_flow_a_to_b() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    // A sends to B via link_ab
    let batch = make_batch(0, 10, "a_to_b");
    triangle
        .link_ab
        .initiator
        .on_batch(batch.into())
        .await
        .expect("A→B send failed");

    // Wait for delivery
    tokio::time::sleep(Duration::from_millis(500)).await;

    // B receives on link_ab responder
    let result = triangle
        .link_ab
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("B poll failed");

    assert!(
        !result.events.is_empty(),
        "B should receive events from A, got {}",
        result.events.len()
    );

    triangle.shutdown().await;
}

/// 2.2 — A sends to B AND A sends to C simultaneously; both receive.
///
/// Proves concurrent sends over different links from the same logical
/// node don't interfere.
#[tokio::test]
async fn test_data_flow_a_to_b_and_c() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    // A sends to B and C concurrently
    let batch_ab = make_batch(0, 10, "to_b");
    let batch_ac = make_batch(0, 10, "to_c");

    let (send_ab, send_ac) = tokio::join!(
        triangle.link_ab.initiator.on_batch(batch_ab.into()),
        triangle.link_ac.initiator.on_batch(batch_ac.into()),
    );
    send_ab.expect("A→B send failed");
    send_ac.expect("A→C send failed");

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

    // B receives from A
    let b_result = triangle
        .link_ab
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("B poll failed");

    // C receives from A
    let c_result = triangle
        .link_ac
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("C poll failed");

    assert!(
        !b_result.events.is_empty(),
        "B should receive events from A"
    );
    assert!(
        !c_result.events.is_empty(),
        "C should receive events from A"
    );

    triangle.shutdown().await;
}

/// 4.1 — A sends to B while B sends to A simultaneously.
///
/// Full-duplex test. Verifies TX/RX key derivation is correct in
/// both directions within the same Noise session.
#[tokio::test]
async fn test_bidirectional_simultaneous() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    // A→B (shard 0) and B→A (shard 0) — both on shard 0 to match
    // the proven pattern from the 2-node integration tests.
    // Events are distinguished by tag, not shard.
    let batch_a_to_b = make_batch(0, 10, "a_sends");
    let batch_b_to_a = make_batch(0, 10, "b_sends");

    // Send sequentially: A first, then B. Concurrent sends on the same
    // session can race on the TX counter if both sides try simultaneously.
    triangle
        .link_ab
        .initiator
        .on_batch(batch_a_to_b.into())
        .await
        .expect("A→B send failed");
    tokio::time::sleep(Duration::from_millis(100)).await;
    triangle
        .link_ab
        .responder
        .on_batch(batch_b_to_a.into())
        .await
        .expect("B→A send failed");

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

    // B receives A's events on shard 0
    let b_received = triangle
        .link_ab
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("B poll shard 0 failed");

    // A receives B's events on shard 0
    let a_received = triangle
        .link_ab
        .initiator
        .poll_shard(0, None, 100)
        .await
        .expect("A poll shard 0 failed");

    assert!(
        !b_received.events.is_empty(),
        "B should receive A's events, got {}",
        b_received.events.len()
    );
    assert!(
        !a_received.events.is_empty(),
        "A should receive B's events, got {}",
        a_received.events.len()
    );

    triangle.shutdown().await;
}

/// 4.3 — A sends stream S1 to B and stream S2 to C; streams don't leak.
///
/// Events sent on the A↔B link should not appear on the A↔C link.
/// Verifies stream isolation between independent encrypted sessions.
#[tokio::test]
async fn test_independent_streams_no_interference() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    // S1: A→B on shard 0
    let batch_s1 = make_batch(0, 10, "stream_s1_for_b");
    triangle
        .link_ab
        .initiator
        .on_batch(batch_s1.into())
        .await
        .expect("S1 send failed");

    // S2: A→C on shard 0
    let batch_s2 = make_batch(0, 10, "stream_s2_for_c");
    triangle
        .link_ac
        .initiator
        .on_batch(batch_s2.into())
        .await
        .expect("S2 send failed");

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

    // B should only see S1 events
    let b_events = triangle
        .link_ab
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("B poll failed");

    // C should only see S2 events
    let c_events = triangle
        .link_ac
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("C poll failed");

    // Verify B got S1 events (tag: stream_s1_for_b)
    for event in &b_events.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(
            json["tag"], "stream_s1_for_b",
            "B received event from wrong stream: {:?}",
            json
        );
    }

    // Verify C got S2 events (tag: stream_s2_for_c)
    for event in &c_events.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(
            json["tag"], "stream_s2_for_c",
            "C received event from wrong stream: {:?}",
            json
        );
    }

    assert!(!b_events.events.is_empty(), "B should receive S1 events");
    assert!(!c_events.events.is_empty(), "C should receive S2 events");

    triangle.shutdown().await;
}

/// All three links carry data simultaneously. A→B, B→C, C→A.
///
/// Full ring: every node sends and receives on different links.
/// Proves the mesh sustains concurrent traffic on all edges without
/// interference or deadlock.
#[tokio::test]
async fn test_full_ring_traffic() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    // A→B, B→C, C→A — all on shard 0, distinguished by tags.
    // Send sequentially to avoid TX counter races on sessions where
    // both sides send concurrently.
    let batch_ab = make_batch(0, 10, "a_to_b");
    let batch_bc = make_batch(0, 10, "b_to_c");
    let batch_ca = make_batch(0, 10, "c_to_a");

    triangle
        .link_ab
        .initiator
        .on_batch(batch_ab.into())
        .await
        .expect("A→B failed");
    triangle
        .link_bc
        .initiator
        .on_batch(batch_bc.into())
        .await
        .expect("B→C failed");
    // C→A: C is responder on link_ac
    triangle
        .link_ac
        .responder
        .on_batch(batch_ca.into())
        .await
        .expect("C→A failed");

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

    // B receives from A (link_ab responder)
    let b_got = triangle
        .link_ab
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("B poll failed");

    // C receives from B (link_bc responder)
    let c_got = triangle
        .link_bc
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("C poll failed");

    // A receives from C (link_ac initiator)
    let a_got = triangle
        .link_ac
        .initiator
        .poll_shard(0, None, 100)
        .await
        .expect("A poll failed");

    assert!(
        !b_got.events.is_empty(),
        "B should receive from A, got {}",
        b_got.events.len()
    );
    assert!(
        !c_got.events.is_empty(),
        "C should receive from B, got {}",
        c_got.events.len()
    );
    assert!(
        !a_got.events.is_empty(),
        "A should receive from C, got {}",
        a_got.events.len()
    );

    triangle.shutdown().await;
}

/// Large batch across all links simultaneously.
///
/// Each link sends 100 events. Verifies the mesh handles sustained
/// throughput without packet loss or corruption under concurrent load.
#[tokio::test]
async fn test_sustained_throughput_all_links() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    let batch_ab = make_batch(0, 100, "ab");
    let batch_ac = make_batch(0, 100, "ac");
    let batch_bc = make_batch(0, 100, "bc");

    let (r1, r2, r3) = tokio::join!(
        triangle.link_ab.initiator.on_batch(batch_ab.into()),
        triangle.link_ac.initiator.on_batch(batch_ac.into()),
        triangle.link_bc.initiator.on_batch(batch_bc.into()),
    );
    r1.expect("A→B failed");
    r2.expect("A→C failed");
    r3.expect("B→C failed");

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

    let b_count = triangle
        .link_ab
        .responder
        .poll_shard(0, None, 1000)
        .await
        .expect("B poll failed")
        .events
        .len();

    let c_from_a = triangle
        .link_ac
        .responder
        .poll_shard(0, None, 1000)
        .await
        .expect("C poll A failed")
        .events
        .len();

    let c_from_b = triangle
        .link_bc
        .responder
        .poll_shard(0, None, 1000)
        .await
        .expect("C poll B failed")
        .events
        .len();

    // With fire-and-forget UDP on localhost, we should get the vast majority
    assert!(
        b_count >= 50,
        "B should receive most of A's 100 events, got {}",
        b_count
    );
    assert!(
        c_from_a >= 50,
        "C should receive most of A's 100 events, got {}",
        c_from_a
    );
    assert!(
        c_from_b >= 50,
        "C should receive most of B's 100 events, got {}",
        c_from_b
    );

    triangle.shutdown().await;
}

/// Failure detection: shut down one node and verify the peer detects it.
///
/// The most important test in the rerouting category. On localhost,
/// kernel scheduling adds microseconds of jitter — we use generous
/// bounds (2x configured timeout) to avoid CI flakiness. The value
/// is proving the detection mechanism works over real sockets.
#[tokio::test]
async fn test_failure_detection_on_node_shutdown() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Confirm healthy before
    assert!(triangle.link_ab.initiator.is_healthy().await);
    assert!(triangle.link_ab.responder.is_healthy().await);

    // Record time and shut down B's side of the A↔B link
    let shutdown_start = std::time::Instant::now();
    let _ = triangle.link_ab.responder.shutdown().await;

    // Wait for A to detect B's absence via heartbeat timeout.
    // Configured heartbeat interval is 200ms, session timeout is 5s.
    // We wait up to 2x session timeout to be safe on CI.
    let max_wait = Duration::from_secs(10);
    let poll_interval = Duration::from_millis(100);
    let mut detected = false;

    loop {
        if shutdown_start.elapsed() > max_wait {
            break;
        }
        if !triangle.link_ab.initiator.is_healthy().await {
            detected = true;
            break;
        }
        tokio::time::sleep(poll_interval).await;
    }

    let detection_time = shutdown_start.elapsed();
    assert!(
        detected,
        "A should detect B's failure within {:?}, gave up after {:?}",
        max_wait, detection_time
    );

    // Detection should happen well within the session timeout (5s)
    // plus generous CI headroom. Just verify it wasn't instant (0ms)
    // which would indicate a bug rather than real detection.
    assert!(
        detection_time > Duration::from_millis(50),
        "Detection too fast ({:?}) — likely a bug, not real heartbeat detection",
        detection_time
    );

    // Clean up
    let _ = triangle.link_ab.initiator.shutdown().await;
    let _ = triangle.link_ac.initiator.shutdown().await;
    let _ = triangle.link_ac.responder.shutdown().await;
    let _ = triangle.link_bc.initiator.shutdown().await;
    let _ = triangle.link_bc.responder.shutdown().await;
}

/// Data flow continues on surviving links after one node dies.
///
/// B dies, but A and C can still communicate over the A↔C link.
/// Events sent after B's death are received correctly.
#[tokio::test]
async fn test_data_flow_survives_node_death() {
    let triangle = Triangle::setup().await;
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Kill node B (both adapters)
    let _ = triangle.link_ab.responder.shutdown().await;
    let _ = triangle.link_bc.initiator.shutdown().await;

    // Wait for heartbeat detection
    tokio::time::sleep(Duration::from_millis(500)).await;

    // A↔C should still work
    let batch = make_batch(0, 20, "after_b_death");
    triangle
        .link_ac
        .initiator
        .on_batch(batch.into())
        .await
        .expect("A→C send should work after B dies");

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

    let c_got = triangle
        .link_ac
        .responder
        .poll_shard(0, None, 100)
        .await
        .expect("C poll failed");

    assert!(
        !c_got.events.is_empty(),
        "C should receive events from A after B dies, got {}",
        c_got.events.len()
    );

    // Clean up
    let _ = triangle.link_ab.initiator.shutdown().await;
    let _ = triangle.link_ac.initiator.shutdown().await;
    let _ = triangle.link_ac.responder.shutdown().await;
    let _ = triangle.link_bc.responder.shutdown().await;
}

// ============================================================================
// Phase 2: Router-Based Forwarding
// ============================================================================

use bytes::{BufMut, Bytes, BytesMut};
use net::adapter::net::{
    NetRouter, RouteAction, RouterConfig, RouterError, RoutingHeader, ROUTING_HEADER_SIZE,
};

/// Build a routed packet: routing header + opaque payload.
fn build_routed_packet(dest_id: u64, src_id: u32, ttl: u8, payload: &[u8]) -> Bytes {
    let header = RoutingHeader::new(dest_id, src_id, ttl);
    let mut buf = BytesMut::with_capacity(ROUTING_HEADER_SIZE + payload.len());
    header.write_to(&mut buf);
    buf.put_slice(payload);
    buf.freeze()
}

/// 2.1 — Router forwards a packet from A to C through B.
///
/// A sends a packet destined for C's node ID. B's router has a route
/// to C and forwards it. This tests the routing decision (not the
/// encrypted payload — the router only reads the routing header).
#[tokio::test]
async fn test_router_forwarding_through_middle_node() {
    let ports = find_ports(3).await;

    let node_a: u64 = 0x1111;
    let node_b: u64 = 0x2222;
    let node_c: u64 = 0x3333;

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    // B runs a router
    let router_b = NetRouter::new(RouterConfig::new(node_b, addr_b))
        .await
        .expect("router B failed to bind");

    // B has a route to C
    router_b.add_route(node_c, addr_c);

    // Start B's send loop
    let send_handle = router_b.start();

    // A sends a packet destined for C through B
    let payload = b"hello from A to C";
    let packet = build_routed_packet(node_c, node_a as u32, 4, payload);

    let sock_a = UdpSocket::bind(addr_a).await.unwrap();
    sock_a.send_to(&packet, addr_b).await.unwrap();

    // B receives and routes the packet
    let mut recv_buf = vec![0u8; 8192];
    let (n, from) = tokio::time::timeout(Duration::from_secs(2), router_b.recv_from(&mut recv_buf))
        .await
        .expect("recv timed out")
        .expect("recv failed");

    let data = Bytes::copy_from_slice(&recv_buf[..n]);
    let action = router_b.route_packet(data, from).expect("route failed");

    match action {
        RouteAction::Forwarded(dest) => {
            assert_eq!(dest, addr_c, "should forward to C's address");
        }
        RouteAction::Local(_) => panic!("should not be local delivery"),
    }

    // Verify stats
    let stats = router_b.stats();
    assert_eq!(stats.packets_received, 1);
    assert_eq!(stats.packets_forwarded, 1);
    assert_eq!(stats.packets_local, 0);

    router_b.stop();
    if let Some(h) = send_handle {
        let _ = h.await;
    }
}

/// 2.3 — Router delivers locally when dest_id matches local node.
///
/// A sends a packet addressed to B. B's router recognizes it as local
/// and returns `RouteAction::Local` with the payload (routing header
/// stripped).
#[tokio::test]
async fn test_router_local_delivery() {
    let ports = find_ports(2).await;

    let node_a: u64 = 0x1111;
    let node_b: u64 = 0x2222;

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let router_b = NetRouter::new(RouterConfig::new(node_b, addr_b))
        .await
        .expect("router B failed to bind");

    // A sends to B (dest_id = node_b)
    let payload = b"for B directly";
    let packet = build_routed_packet(node_b, node_a as u32, 4, payload);

    let sock_a = UdpSocket::bind(addr_a).await.unwrap();
    sock_a.send_to(&packet, addr_b).await.unwrap();

    let mut recv_buf = vec![0u8; 8192];
    let (n, from) = tokio::time::timeout(Duration::from_secs(2), router_b.recv_from(&mut recv_buf))
        .await
        .expect("recv timed out")
        .expect("recv failed");

    let data = Bytes::copy_from_slice(&recv_buf[..n]);
    let action = router_b.route_packet(data, from).expect("route failed");

    match action {
        RouteAction::Local(local_data) => {
            // Routing header is stripped; payload is the original bytes
            assert_eq!(&local_data[..], payload);
        }
        RouteAction::Forwarded(_) => panic!("should be local delivery, not forwarded"),
    }

    let stats = router_b.stats();
    assert_eq!(stats.packets_local, 1);
    assert_eq!(stats.packets_forwarded, 0);
}

/// 2.4 — TTL expiry: router drops packets with TTL=0.
///
/// A sends a packet with TTL=1. B decrements to 0 and rejects it
/// with `TtlExpired`. The packet is never forwarded.
#[tokio::test]
async fn test_router_ttl_expiry() {
    let ports = find_ports(3).await;

    let node_a: u64 = 0x1111;
    let node_b: u64 = 0x2222;
    let node_c: u64 = 0x3333;

    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let router_b = NetRouter::new(RouterConfig::new(node_b, addr_b))
        .await
        .expect("router B failed to bind");
    router_b.add_route(node_c, addr_c);

    // TTL=0: already expired, B should drop immediately
    let packet = build_routed_packet(node_c, node_a as u32, 0, b"should expire");

    // Simulate B receiving this packet
    let result = router_b.route_packet(packet, "127.0.0.1:0".parse().unwrap());

    assert!(
        matches!(result, Err(RouterError::TtlExpired)),
        "expected TtlExpired, got {:?}",
        result
    );

    let stats = router_b.stats();
    assert_eq!(
        stats.packets_dropped, 1,
        "TTL-expired packet should be counted as dropped"
    );
    assert_eq!(
        stats.packets_forwarded, 0,
        "TTL-expired packet should not be forwarded"
    );
}

/// 2.5 — Hop count is incremented on forwarding.
///
/// A sends a packet with hop_count=0 to C via B. After B forwards it,
/// the hop_count in the forwarded packet should be 1.
#[tokio::test]
async fn test_router_hop_count_incremented() {
    let ports = find_ports(3).await;

    let node_a: u64 = 0x1111;
    let node_b: u64 = 0x2222;
    let node_c: u64 = 0x3333;

    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let router_b = NetRouter::new(RouterConfig::new(node_b, addr_b))
        .await
        .expect("router B failed to bind");
    router_b.add_route(node_c, addr_c);

    // Start router to process the forwarded packet through the scheduler
    let send_handle = router_b.start();

    // A sends packet destined for C
    let packet = build_routed_packet(node_c, node_a as u32, 4, b"hop test");

    // Route it through B
    let action = router_b
        .route_packet(packet, "127.0.0.1:0".parse().unwrap())
        .unwrap();
    assert!(matches!(action, RouteAction::Forwarded(_)));

    // C receives the forwarded packet and checks hop_count
    let sock_c = UdpSocket::bind(addr_c).await.unwrap();
    let mut recv_buf = vec![0u8; 8192];
    let (n, _) = tokio::time::timeout(Duration::from_secs(2), sock_c.recv_from(&mut recv_buf))
        .await
        .expect("C recv timed out")
        .expect("C recv failed");

    // Parse routing header from the forwarded packet
    let fwd_header = RoutingHeader::from_bytes(&recv_buf[..n])
        .expect("invalid routing header in forwarded packet");

    assert_eq!(
        fwd_header.hop_count, 1,
        "hop_count should be 1 after one forward"
    );
    assert_eq!(fwd_header.ttl, 3, "TTL should be decremented from 4 to 3");
    assert_eq!(fwd_header.dest_id, node_c, "dest_id should be unchanged");

    router_b.stop();
    if let Some(h) = send_handle {
        let _ = h.await;
    }
}

/// 2.6 — Router rejects packet with no route to destination.
#[tokio::test]
async fn test_router_no_route() {
    let ports = find_ports(1).await;
    let node_b: u64 = 0x2222;
    let unknown_dest: u64 = 0x9999;

    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let router_b = NetRouter::new(RouterConfig::new(node_b, addr_b))
        .await
        .expect("router B failed to bind");

    // No route to 0x9999
    let packet = build_routed_packet(unknown_dest, 0x1111, 4, b"no route");
    let result = router_b.route_packet(packet, "127.0.0.1:0".parse().unwrap());

    assert!(
        matches!(result, Err(RouterError::NoRoute)),
        "expected NoRoute, got {:?}",
        result
    );
}

/// Full-stack: EventBus backed by Net adapter, ingest on one side,
/// poll on the other.
///
/// This is the highest-value test — it proves the entire pipeline works
/// end-to-end: EventBus → sharded ring buffers → drain workers →
/// batch workers → NetAdapter → encrypted UDP → NetAdapter → poll.
#[tokio::test]
async fn test_eventbus_over_net_full_stack() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];

    let responder_keypair = StaticKeypair::generate();

    let sender_addr: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let receiver_addr: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    // Sender EventBus config with Net adapter (initiator)
    let sender_net =
        NetAdapterConfig::initiator(sender_addr, receiver_addr, psk, responder_keypair.public)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
            .with_socket_buffers(TEST_BUFFER_SIZE, TEST_BUFFER_SIZE);

    let sender_config = net::config::EventBusConfig::builder()
        .num_shards(2)
        .ring_buffer_capacity(1024)
        .adapter(net::config::AdapterConfig::Net(Box::new(sender_net)))
        .without_scaling()
        .build()
        .unwrap();

    // Receiver: just a NetAdapter (not an EventBus — we poll the adapter directly)
    let receiver_net =
        NetAdapterConfig::responder(receiver_addr, sender_addr, psk, responder_keypair)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
            .with_socket_buffers(TEST_BUFFER_SIZE, TEST_BUFFER_SIZE);

    let barrier = Arc::new(Barrier::new(2));

    // Spawn receiver adapter
    let rb = barrier.clone();
    let receiver_handle = tokio::spawn(async move {
        let mut adapter = net::adapter::net::NetAdapter::new(receiver_net).unwrap();
        rb.wait().await;
        adapter.init().await.expect("receiver init failed");

        // Wait for events to arrive
        tokio::time::sleep(Duration::from_millis(2000)).await;

        // Poll for events across both shards
        let shard0 = adapter
            .poll_shard(0, None, 1000)
            .await
            .expect("poll 0 failed");
        let shard1 = adapter
            .poll_shard(1, None, 1000)
            .await
            .expect("poll 1 failed");

        adapter.shutdown().await.expect("receiver shutdown failed");
        shard0.events.len() + shard1.events.len()
    });

    // Spawn sender EventBus
    let sb = barrier.clone();
    let sender_handle = tokio::spawn(async move {
        sb.wait().await;
        let bus = net::EventBus::new(sender_config)
            .await
            .expect("sender EventBus failed");

        // Give handshake time to complete
        tokio::time::sleep(Duration::from_millis(500)).await;

        // Ingest events through the full EventBus pipeline
        for i in 0..50 {
            let event =
                net::event::Event::new(serde_json::json!({"index": i, "source": "eventbus"}));
            bus.ingest(event).unwrap();
        }

        // Flush to ensure events are dispatched to adapter
        bus.flush().await.expect("flush failed");

        // Wait for delivery
        tokio::time::sleep(Duration::from_millis(1000)).await;

        bus.shutdown().await.expect("sender shutdown failed");
    });

    let timeout = Duration::from_secs(15);
    let (recv_result, send_result) = tokio::time::timeout(
        timeout,
        futures::future::join(receiver_handle, sender_handle),
    )
    .await
    .expect("full-stack test timed out");

    send_result.expect("sender panicked");
    let received = recv_result.expect("receiver panicked");

    assert!(
        received >= 25,
        "receiver should get most of 50 events through the full stack, got {}",
        received
    );
}

// ============================================================================
// Phase 2 continued: Backpressure and full-stack stress
// ============================================================================

/// 10.1 — Backpressure: flood a node's ring buffer, verify it doesn't crash.
///
/// A sends events faster than the adapter can drain. The ring buffer
/// fills and events are dropped per the backpressure policy. After the
/// flood stops, the bus recovers and can still ingest new events.
/// Verifies "ring buffer is a speed buffer, not a waiting room."
#[tokio::test]
async fn test_backpressure_ring_buffer_survives_flood() {
    // Small ring buffer to trigger backpressure quickly
    let config = net::config::EventBusConfig::builder()
        .num_shards(2)
        .ring_buffer_capacity(1024) // minimum allowed — fills fast under flood
        .without_scaling()
        .build()
        .unwrap();

    let bus = net::EventBus::new(config).await.unwrap();

    // Flood: ingest far more events than the ring buffer can hold
    let mut ingested = 0u64;
    let mut _dropped = 0u64;
    for i in 0..10_000 {
        let event = net::event::Event::new(serde_json::json!({"flood": i}));
        match bus.ingest(event) {
            Ok(_) => ingested += 1,
            Err(_) => _dropped += 1,
        }
    }

    // Some events should have been dropped (backpressure)
    // With a 64-slot ring buffer and 2 shards, capacity is 128 total.
    // 10k events means most are dropped or evicted.
    assert!(ingested > 0, "at least some events should be ingested");

    // Bus should not be in a broken state — new events still work
    tokio::time::sleep(Duration::from_millis(100)).await;
    let post_flood = net::event::Event::new(serde_json::json!({"after": "flood"}));
    let result = bus.ingest(post_flood);
    assert!(
        result.is_ok(),
        "bus should accept events after flood subsides"
    );

    let stats = bus.stats();
    let total_ingested = stats
        .events_ingested
        .load(std::sync::atomic::Ordering::Relaxed);
    assert!(total_ingested > 0, "stats should reflect ingested events");

    bus.shutdown().await.unwrap();
}

/// Full-stack bidirectional: two EventBus instances connected via Net,
/// both ingest and poll.
///
/// A ingests events → Net → B polls them. B ingests events → Net → A
/// polls them. Proves the full EventBus pipeline works in both
/// directions over encrypted UDP.
#[tokio::test]
async fn test_eventbus_bidirectional_over_net() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let keypair_b = StaticKeypair::generate();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    // A: initiator EventBus
    let net_a = NetAdapterConfig::initiator(addr_a, addr_b, psk, keypair_b.public)
        .with_handshake(3, Duration::from_secs(3))
        .with_heartbeat_interval(Duration::from_millis(500))
        .with_session_timeout(Duration::from_secs(10))
        .with_socket_buffers(TEST_BUFFER_SIZE, TEST_BUFFER_SIZE);

    // B: responder — raw adapter (not EventBus) since each adapter has one peer
    let net_b = NetAdapterConfig::responder(addr_b, addr_a, psk, keypair_b)
        .with_handshake(3, Duration::from_secs(3))
        .with_heartbeat_interval(Duration::from_millis(500))
        .with_session_timeout(Duration::from_secs(10))
        .with_socket_buffers(TEST_BUFFER_SIZE, TEST_BUFFER_SIZE);

    let config_a = net::config::EventBusConfig::builder()
        .num_shards(2)
        .ring_buffer_capacity(1024)
        .adapter(net::config::AdapterConfig::Net(Box::new(net_a)))
        .without_scaling()
        .build()
        .unwrap();

    let barrier = Arc::new(Barrier::new(2));

    // B: adapter
    let bb = barrier.clone();
    let b_handle = tokio::spawn(async move {
        let mut adapter = net::adapter::net::NetAdapter::new(net_b).unwrap();
        bb.wait().await;
        adapter.init().await.expect("B init failed");

        // B sends events to A
        tokio::time::sleep(Duration::from_millis(500)).await;
        let events: Vec<InternalEvent> = (0..20)
            .map(|i| InternalEvent::from_value(serde_json::json!({"from": "B", "i": i}), i, 0))
            .collect();
        adapter
            .on_batch(
                Batch {
                    shard_id: 0,
                    events,
                    sequence_start: 0,
                    process_nonce: batch_process_nonce(),
                }
                .into(),
            )
            .await
            .expect("B send failed");

        // Wait for A's events
        tokio::time::sleep(Duration::from_millis(1500)).await;
        let received = adapter
            .poll_shard(0, None, 1000)
            .await
            .expect("B poll failed");
        adapter.shutdown().await.expect("B shutdown failed");
        received.events.len()
    });

    // A: EventBus
    let ab = barrier.clone();
    let a_handle = tokio::spawn(async move {
        ab.wait().await;
        let bus = net::EventBus::new(config_a)
            .await
            .expect("A EventBus failed");
        tokio::time::sleep(Duration::from_millis(500)).await;

        // A ingests events (→ Net → B)
        for i in 0..20 {
            bus.ingest(net::event::Event::new(
                serde_json::json!({"from": "A", "i": i}),
            ))
            .unwrap();
        }
        bus.flush().await.expect("A flush failed");

        // Wait for B's events, then poll
        tokio::time::sleep(Duration::from_millis(1500)).await;
        let response = bus
            .poll(net::ConsumeRequest::new(1000))
            .await
            .expect("A poll failed");

        bus.shutdown().await.expect("A shutdown failed");
        response.events.len()
    });

    let timeout = Duration::from_secs(15);
    let (b_result, a_result) =
        tokio::time::timeout(timeout, futures::future::join(b_handle, a_handle))
            .await
            .expect("bidirectional test timed out");

    let b_received = b_result.expect("B panicked");
    let _a_received = a_result.expect("A panicked");

    assert!(
        b_received > 0,
        "B should receive A's events through EventBus pipeline, got {}",
        b_received
    );
    // A polls from its own EventBus, which uses the Net adapter for storage.
    // Events from B arrive in the adapter's inbound queue, but the EventBus
    // polls from its own adapter — so A may or may not see B's events depending
    // on whether the adapter's inbound feeds into the EventBus poll path.
    // The key assertion is that B receives A's events (full stack works).
}

/// Router forwarding: A → B → C over actual UDP sockets.
///
/// Unlike the unit-level router test, this sends the forwarded packet
/// over real UDP and verifies C actually receives it.
#[tokio::test]
async fn test_router_end_to_end_forwarding() {
    let ports = find_ports(3).await;

    let node_a: u64 = 0x1111;
    let node_b: u64 = 0x2222;
    let node_c: u64 = 0x3333;

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    // B is the router
    let router_b = NetRouter::new(RouterConfig::new(node_b, addr_b))
        .await
        .expect("router B bind failed");
    router_b.add_route(node_c, addr_c);
    let send_handle = router_b.start();

    // C listens
    let sock_c = UdpSocket::bind(addr_c).await.unwrap();

    // A sends 10 packets to C via B
    let sock_a = UdpSocket::bind(addr_a).await.unwrap();
    for i in 0..10u8 {
        let payload = format!("packet-{}", i);
        let packet = build_routed_packet(node_c, node_a as u32, 4, payload.as_bytes());
        sock_a.send_to(&packet, addr_b).await.unwrap();
    }

    // B receives and routes each packet
    let mut recv_buf = vec![0u8; 8192];
    for _ in 0..10 {
        let (n, from) =
            tokio::time::timeout(Duration::from_secs(2), router_b.recv_from(&mut recv_buf))
                .await
                .expect("B recv timed out")
                .expect("B recv failed");

        let data = Bytes::copy_from_slice(&recv_buf[..n]);
        let _ = router_b.route_packet(data, from);
    }

    // Give the send loop time to flush
    tokio::time::sleep(Duration::from_millis(200)).await;

    // C receives the forwarded packets
    let mut received = 0;
    while let Ok(Ok((n, _))) =
        tokio::time::timeout(Duration::from_millis(500), sock_c.recv_from(&mut recv_buf)).await
    {
        // Verify it has a valid routing header
        let hdr = RoutingHeader::from_bytes(&recv_buf[..n]);
        assert!(
            hdr.is_some(),
            "forwarded packet should have valid routing header"
        );
        let hdr = hdr.unwrap();
        assert_eq!(hdr.dest_id, node_c);
        assert_eq!(hdr.hop_count, 1, "should have 1 hop from B");
        received += 1;
    }

    assert!(
        received >= 5,
        "C should receive most of 10 forwarded packets, got {}",
        received
    );

    router_b.stop();
    if let Some(h) = send_handle {
        let _ = h.await;
    }
}

/// Multi-hop forwarding: A → B → C with two routers.
///
/// Both B and C run routers. A sends to a destination known only to C
/// (node D = 0x4444). B forwards to C, C delivers locally.
/// Proves the hop_count increments correctly across two hops.
#[tokio::test]
async fn test_router_multi_hop_two_routers() {
    let ports = find_ports(3).await;

    let node_a: u64 = 0x1111;
    let node_b: u64 = 0x2222;
    let node_c: u64 = 0x3333;

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    // B routes to C
    let router_b = NetRouter::new(RouterConfig::new(node_b, addr_b))
        .await
        .expect("router B bind failed");
    router_b.add_route(node_c, addr_c);
    let send_b = router_b.start();

    // C is the destination (local delivery)
    let router_c = NetRouter::new(RouterConfig::new(node_c, addr_c))
        .await
        .expect("router C bind failed");

    // A sends packet destined for C, TTL=4
    let sock_a = UdpSocket::bind(addr_a).await.unwrap();
    let packet = build_routed_packet(node_c, node_a as u32, 4, b"multi-hop-test");
    sock_a.send_to(&packet, addr_b).await.unwrap();

    // B receives and forwards
    let mut buf = vec![0u8; 8192];
    let (n, from) = tokio::time::timeout(Duration::from_secs(2), router_b.recv_from(&mut buf))
        .await
        .expect("B recv timed out")
        .expect("B recv failed");

    let data = Bytes::copy_from_slice(&buf[..n]);
    let action = router_b.route_packet(data, from).expect("B route failed");
    assert!(matches!(action, RouteAction::Forwarded(_)));

    // Give B's send loop time to transmit
    tokio::time::sleep(Duration::from_millis(200)).await;

    // C receives the forwarded packet
    let (n, from) = tokio::time::timeout(Duration::from_secs(2), router_c.recv_from(&mut buf))
        .await
        .expect("C recv timed out")
        .expect("C recv failed");

    let data = Bytes::copy_from_slice(&buf[..n]);
    let action = router_c.route_packet(data, from).expect("C route failed");

    match action {
        RouteAction::Local(payload) => {
            assert_eq!(
                &payload[..],
                b"multi-hop-test",
                "payload should survive 2 hops"
            );
        }
        RouteAction::Forwarded(_) => panic!("C should deliver locally, not forward"),
    }

    let stats_b = router_b.stats();
    assert_eq!(stats_b.packets_forwarded, 1);

    let stats_c = router_c.stats();
    assert_eq!(stats_c.packets_local, 1);

    router_b.stop();
    if let Some(h) = send_b {
        let _ = h.await;
    }
}

// ============================================================================
// Phase 4: Subnet Gateway Enforcement
// ============================================================================

use net::adapter::net::{
    ChannelConfig, ChannelConfigRegistry, ChannelId, DropReason, ForwardDecision, SubnetGateway,
    SubnetId, Visibility,
};

/// Helper: register a channel with the given visibility and return its
/// wire `u16` hash (matches `SubnetGateway::should_forward`'s
/// `channel_hash` argument, which is the per-packet wire field).
fn register_channel(registry: &ChannelConfigRegistry, name: &str, vis: Visibility) -> u16 {
    let id = ChannelId::parse(name).expect("invalid channel name");
    let wire = id.wire_hash();
    let config = ChannelConfig::new(id).with_visibility(vis);
    registry.insert(config);
    wire
}

/// 9.1 — Gateway blocks SubnetLocal traffic at boundary.
#[tokio::test]
async fn test_subnet_gateway_blocks_local_traffic() {
    let subnet_ab = SubnetId::new(&[3, 1]);
    let subnet_c = SubnetId::new(&[3, 2]);

    let registry = std::sync::Arc::new(ChannelConfigRegistry::new());
    let local_hash = register_channel(&registry, "internal/metrics", Visibility::SubnetLocal);

    let gateway = SubnetGateway::new(subnet_ab, registry);

    let decision = gateway.should_forward(subnet_ab, subnet_c, local_hash, 8, 0);
    assert_eq!(
        decision,
        ForwardDecision::Drop(DropReason::SubnetLocal),
        "SubnetLocal traffic must not cross subnet boundary"
    );
}

/// 9.2 — Gateway forwards Global traffic across subnets.
#[tokio::test]
async fn test_subnet_gateway_forwards_global_traffic() {
    let subnet_ab = SubnetId::new(&[3, 1]);
    let subnet_c = SubnetId::new(&[3, 2]);

    let registry = std::sync::Arc::new(ChannelConfigRegistry::new());
    let global_hash = register_channel(&registry, "events/global", Visibility::Global);

    let gateway = SubnetGateway::new(subnet_ab, registry);

    let decision = gateway.should_forward(subnet_ab, subnet_c, global_hash, 8, 0);
    assert_eq!(
        decision,
        ForwardDecision::Forward,
        "Global traffic must cross subnet boundary"
    );
}

/// 9.3 — Gateway forwards Exported traffic only to listed subnets.
#[tokio::test]
async fn test_subnet_gateway_exported_selective() {
    let subnet_ab = SubnetId::new(&[3, 1]);
    let subnet_c = SubnetId::new(&[3, 2]);
    let subnet_d = SubnetId::new(&[3, 3]);

    let registry = std::sync::Arc::new(ChannelConfigRegistry::new());
    let export_hash = register_channel(&registry, "data/shared", Visibility::Exported);

    let gateway = SubnetGateway::new(subnet_ab, registry);
    gateway.add_peer(subnet_c);
    gateway.add_peer(subnet_d);
    gateway.export_channel(export_hash, vec![subnet_c]);

    let to_c = gateway.should_forward(subnet_ab, subnet_c, export_hash, 8, 0);
    assert_eq!(
        to_c,
        ForwardDecision::Forward,
        "exported to C should forward"
    );

    let to_d = gateway.should_forward(subnet_ab, subnet_d, export_hash, 8, 0);
    assert_eq!(
        to_d,
        ForwardDecision::Drop(DropReason::NotExported),
        "not exported to D should drop"
    );
}

/// 9.4 — ParentVisible traffic forwards to ancestor subnets only.
#[tokio::test]
async fn test_subnet_gateway_parent_visible() {
    let child = SubnetId::new(&[3, 1, 2]);
    let parent = SubnetId::new(&[3, 1]);
    let sibling = SubnetId::new(&[3, 2]);

    let registry = std::sync::Arc::new(ChannelConfigRegistry::new());
    let hash = register_channel(&registry, "status/reports", Visibility::ParentVisible);

    let gateway = SubnetGateway::new(child, registry);

    let to_parent = gateway.should_forward(child, parent, hash, 8, 0);
    assert_eq!(
        to_parent,
        ForwardDecision::Forward,
        "child to parent should forward"
    );

    let to_sibling = gateway.should_forward(child, sibling, hash, 8, 0);
    assert_eq!(
        to_sibling,
        ForwardDecision::Drop(DropReason::NotAncestor),
        "child to sibling should drop"
    );
}

/// 9.5 — Gateway stats track forwarded and dropped accurately.
#[tokio::test]
async fn test_subnet_gateway_stats() {
    let subnet_a = SubnetId::new(&[1]);
    let subnet_b = SubnetId::new(&[2]);

    let registry = std::sync::Arc::new(ChannelConfigRegistry::new());
    let local_hash = register_channel(&registry, "chan/local", Visibility::SubnetLocal);
    let global_hash = register_channel(&registry, "chan/global", Visibility::Global);

    let gateway = SubnetGateway::new(subnet_a, registry);

    for _ in 0..5 {
        gateway.should_forward(subnet_a, subnet_b, local_hash, 8, 0);
    }
    for _ in 0..3 {
        gateway.should_forward(subnet_a, subnet_b, global_hash, 8, 0);
    }

    assert_eq!(gateway.forwarded_count(), 3, "3 global packets forwarded");
    assert_eq!(gateway.dropped_count(), 5, "5 local packets dropped");
}

// ============================================================================
// Phase 4: Correlated Failure Detection
// ============================================================================

use net::adapter::net::{CorrelatedFailureConfig, CorrelatedFailureDetector, CorrelationVerdict};

/// 8.1 — Independent failures classified correctly.
///
/// A few nodes fail — below the mass failure threshold.
/// Should return Independent verdict.
#[tokio::test]
async fn test_correlated_failure_independent() {
    let config = CorrelatedFailureConfig {
        correlation_window: Duration::from_secs(2),
        mass_failure_threshold: 0.3,
        subnet_correlation_threshold: 0.8,
        max_concurrent_migrations: 3,
    };
    let mut detector = CorrelatedFailureDetector::new(config);

    // Register 10 nodes across subnets
    for i in 0..10u64 {
        detector.register_node(i, SubnetId::new(&[(i as u8) % 4]));
    }

    // 2 of 10 fail — 20%, below 30% threshold
    let verdict = detector.record_failures(&[0, 1], 10);
    assert!(
        matches!(verdict, CorrelationVerdict::Independent { .. }),
        "2/10 = 20% < 30% threshold = Independent, got {:?}",
        verdict
    );

    // Independent → unlimited recovery budget
    let budget = detector.recovery_budget();
    assert_eq!(
        budget,
        usize::MAX,
        "independent failures get unlimited budget"
    );
}

/// 8.2 — Mass failure classified when threshold exceeded.
///
/// 4 of 10 nodes fail (40%) — above the 30% threshold.
/// Should return MassFailure verdict with throttled recovery.
#[tokio::test]
async fn test_correlated_failure_mass() {
    let config = CorrelatedFailureConfig {
        correlation_window: Duration::from_secs(2),
        mass_failure_threshold: 0.3,
        subnet_correlation_threshold: 0.8,
        max_concurrent_migrations: 3,
    };
    let mut detector = CorrelatedFailureDetector::new(config);

    for i in 0..10u64 {
        detector.register_node(i, SubnetId::new(&[1]));
    }

    // 4 of 10 = 40% > 30%
    let verdict = detector.record_failures(&[0, 1, 2, 3], 10);
    assert!(
        matches!(verdict, CorrelationVerdict::MassFailure { .. }),
        "4/10 = 40% > 30% threshold = MassFailure, got {:?}",
        verdict
    );

    // Mass failure → throttled recovery
    let budget = detector.recovery_budget();
    assert_eq!(
        budget, 3,
        "mass failure throttles to max_concurrent_migrations"
    );
    assert!(detector.in_mass_failure());
}

/// 8.3 — Recovery budget resets after window clears.
#[tokio::test]
async fn test_correlated_failure_recovery_resets() {
    let config = CorrelatedFailureConfig {
        correlation_window: Duration::from_secs(2),
        mass_failure_threshold: 0.3,
        subnet_correlation_threshold: 0.8,
        max_concurrent_migrations: 2,
    };
    let mut detector = CorrelatedFailureDetector::new(config);

    for i in 0..10u64 {
        detector.register_node(i, SubnetId::new(&[1]));
    }

    // Trigger mass failure
    detector.record_failures(&[0, 1, 2, 3, 4], 10);
    assert!(detector.in_mass_failure());
    assert_eq!(detector.recovery_budget(), 2);

    // Clear the window (simulating time passing / conditions normalizing)
    detector.clear_window();

    // New failures below threshold → back to independent
    let verdict = detector.record_failures(&[8], 10);
    assert!(matches!(verdict, CorrelationVerdict::Independent { .. }));
    assert!(!detector.in_mass_failure());
    assert_eq!(detector.recovery_budget(), usize::MAX);
}

// ============================================================================
// Phase 2 continued: Failure Detector lifecycle
// ============================================================================

use net::adapter::net::{
    FailureDetector, FailureDetectorConfig, MeshNode, MeshNodeConfig, NodeStatus,
};

/// 3.3 — Failure detector: heartbeat → suspect → fail → recover.
///
/// Three nodes send heartbeats. One stops. The detector transitions
/// through Healthy → Suspected → Failed. When heartbeats resume,
/// the node recovers to Healthy.
#[tokio::test]
async fn test_failure_detector_lifecycle() {
    let config = FailureDetectorConfig {
        timeout: Duration::from_millis(100),
        miss_threshold: 3,
        suspicion_threshold: 1,
        cleanup_interval: Duration::from_secs(60),
    };

    let failed_nodes = Arc::new(std::sync::Mutex::new(Vec::<u64>::new()));
    let recovered_nodes = Arc::new(std::sync::Mutex::new(Vec::<u64>::new()));

    let failed_cb = failed_nodes.clone();
    let recovered_cb = recovered_nodes.clone();

    let detector = FailureDetector::with_config(config)
        .on_failure(move |id| failed_cb.lock().unwrap().push(id))
        .on_recovery(move |id| recovered_cb.lock().unwrap().push(id));

    let node_a: u64 = 0x1111;
    let node_b: u64 = 0x2222;
    let node_c: u64 = 0x3333;
    let addr: SocketAddr = "127.0.0.1:1234".parse().unwrap();

    // All three heartbeat — all healthy
    detector.heartbeat(node_a, addr);
    detector.heartbeat(node_b, addr);
    detector.heartbeat(node_c, addr);

    assert_eq!(detector.status(node_a), NodeStatus::Healthy);
    assert_eq!(detector.status(node_b), NodeStatus::Healthy);
    assert_eq!(detector.status(node_c), NodeStatus::Healthy);

    // B stops heartbeating. Wait past the timeout.
    tokio::time::sleep(Duration::from_millis(150)).await;

    // A and C heartbeat, B does not
    detector.heartbeat(node_a, addr);
    detector.heartbeat(node_c, addr);

    // Check — B should be suspected or failed
    let newly_failed = detector.check_all();
    let b_status = detector.status(node_b);
    assert!(
        b_status == NodeStatus::Suspected || b_status == NodeStatus::Failed,
        "B should be suspected or failed after timeout, got {:?}",
        b_status
    );

    // Wait longer and check again — B should be fully failed
    tokio::time::sleep(Duration::from_millis(300)).await;
    detector.heartbeat(node_a, addr);
    detector.heartbeat(node_c, addr);
    let _ = detector.check_all();

    assert_eq!(
        detector.status(node_b),
        NodeStatus::Failed,
        "B should be failed after sustained silence"
    );
    assert!(
        failed_nodes.lock().unwrap().contains(&node_b),
        "failure callback should have been called for B"
    );

    // B recovers — sends heartbeat
    detector.heartbeat(node_b, addr);
    assert_eq!(
        detector.status(node_b),
        NodeStatus::Healthy,
        "B should recover after heartbeat"
    );
    assert!(
        recovered_nodes.lock().unwrap().contains(&node_b),
        "recovery callback should have been called for B"
    );

    // Other nodes unaffected throughout
    assert_eq!(detector.status(node_a), NodeStatus::Healthy);
    assert_eq!(detector.status(node_c), NodeStatus::Healthy);
    assert!(newly_failed.is_empty() || newly_failed == vec![node_b]);
}

// ============================================================================
// Phase 2 continued: Swarm / Pingwave
// ============================================================================

use net::adapter::net::{Pingwave, PINGWAVE_SIZE};

/// Pingwave serialization roundtrip and forwarding mechanics.
///
/// A creates a pingwave with TTL=3. B receives it, forwards (TTL→2,
/// hop_count→1). C receives it, forwards (TTL→1, hop_count→2).
/// A third forward would set TTL→0 and expire it.
#[tokio::test]
async fn test_pingwave_forwarding_chain() {
    let node_a: u64 = 0x1111;

    // A creates a pingwave
    let pw = Pingwave::new(node_a, 1, 3);
    assert_eq!(pw.origin_id, node_a);
    assert_eq!(pw.ttl, 3);
    assert_eq!(pw.hop_count, 0);
    assert!(!pw.is_expired());

    // Serialize and deserialize (simulates wire transfer to B)
    let bytes = pw.to_bytes();
    assert_eq!(bytes.len(), PINGWAVE_SIZE);
    let mut pw_at_b = Pingwave::from_bytes(&bytes).expect("deserialization failed");
    assert_eq!(pw_at_b.origin_id, node_a);

    // B forwards: TTL 3→2, hop_count 0→1
    assert!(pw_at_b.forward());
    assert_eq!(pw_at_b.ttl, 2);
    assert_eq!(pw_at_b.hop_count, 1);

    // C receives and forwards: TTL 2→1, hop_count 1→2
    let mut pw_at_c = pw_at_b;
    assert!(pw_at_c.forward());
    assert_eq!(pw_at_c.ttl, 1);
    assert_eq!(pw_at_c.hop_count, 2);

    // D receives and forwards: TTL 1→0, hop_count 2→3
    let mut pw_at_d = pw_at_c;
    assert!(pw_at_d.forward());
    assert_eq!(pw_at_d.ttl, 0);
    assert_eq!(pw_at_d.hop_count, 3);

    // Next forward should fail — TTL is 0
    assert!(!pw_at_d.forward(), "TTL=0 should refuse to forward");
    assert!(pw_at_d.is_expired());
}

/// Pingwave over real UDP: A broadcasts, B and C receive.
#[tokio::test]
async fn test_pingwave_over_udp() {
    let ports = find_ports(3).await;

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let sock_a = UdpSocket::bind(addr_a).await.unwrap();
    let sock_b = UdpSocket::bind(addr_b).await.unwrap();
    let sock_c = UdpSocket::bind(addr_c).await.unwrap();

    // A broadcasts a pingwave to B and C
    let pw = Pingwave::new(0x1111, 42, 3);
    let bytes = pw.to_bytes();
    sock_a.send_to(&bytes, addr_b).await.unwrap();
    sock_a.send_to(&bytes, addr_c).await.unwrap();

    // B receives
    let mut buf = vec![0u8; 64];
    let (n, from) = tokio::time::timeout(Duration::from_secs(2), sock_b.recv_from(&mut buf))
        .await
        .expect("B recv timed out")
        .expect("B recv failed");

    let pw_b = Pingwave::from_bytes(&buf[..n]).expect("B: invalid pingwave");
    assert_eq!(pw_b.origin_id, 0x1111);
    assert_eq!(pw_b.seq, 42);
    assert_eq!(from, addr_a);

    // C receives
    let (n, from) = tokio::time::timeout(Duration::from_secs(2), sock_c.recv_from(&mut buf))
        .await
        .expect("C recv timed out")
        .expect("C recv failed");

    let pw_c = Pingwave::from_bytes(&buf[..n]).expect("C: invalid pingwave");
    assert_eq!(pw_c.origin_id, 0x1111);
    assert_eq!(from, addr_a);
}

// ============================================================================
// MeshNode tests
// ============================================================================

use net::adapter::net::identity::EntityKeypair;

/// MeshNode: two nodes connect, exchange data through the mesh runtime.
///
/// This is the first test of the composed protocol stack: single socket,
/// multi-session, encrypted data flow.
#[tokio::test]
async fn test_mesh_node_two_node_data_exchange() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];

    let identity_a = EntityKeypair::generate();
    let identity_b = EntityKeypair::generate();
    let node_id_a = identity_a.node_id();
    let node_id_b = identity_b.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let config_a = MeshNodeConfig::new(addr_a, psk)
        .with_num_shards(2)
        .with_handshake(3, Duration::from_secs(3))
        .with_heartbeat_interval(Duration::from_millis(500))
        .with_session_timeout(Duration::from_secs(10));

    let config_b = MeshNodeConfig::new(addr_b, psk)
        .with_num_shards(2)
        .with_handshake(3, Duration::from_secs(3))
        .with_heartbeat_interval(Duration::from_millis(500))
        .with_session_timeout(Duration::from_secs(10));

    let node_a = MeshNode::new(identity_a, config_a).await.unwrap();
    let node_b = MeshNode::new(identity_b, config_b).await.unwrap();

    // Get B's Noise public key (Curve25519, not ed25519)
    let pubkey_b = *node_b.public_key();

    // Connect: A initiates, B accepts (concurrently via join)
    let (accept_result, connect_result) = tokio::join!(node_b.accept(node_id_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pubkey_b, node_id_b).await
    },);
    accept_result.expect("B accept failed");
    connect_result.expect("A connect failed");

    // Start receive loops
    node_a.start();
    node_b.start();

    assert_eq!(node_a.peer_count(), 1);
    assert_eq!(node_b.peer_count(), 1);

    // A sends events to B
    tokio::time::sleep(Duration::from_millis(100)).await;
    let batch = make_batch(0, 20, "mesh_test");
    node_a.send_to_peer(addr_b, &batch).await.unwrap();

    // Wait for delivery
    tokio::time::sleep(Duration::from_millis(1000)).await;

    // B polls for events
    let result = node_b.poll_shard(0, None, 100).await.unwrap();

    assert!(
        !result.events.is_empty(),
        "B should receive events from A via MeshNode, got {}",
        result.events.len()
    );

    // Verify event content
    for event in &result.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(json["tag"], "mesh_test");
    }

    // Shutdown
    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
}

/// MeshNode: three nodes form a triangle, all pairs exchange data.
#[tokio::test]
async fn test_mesh_node_triangle() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();

    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk_config = |addr: SocketAddr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    let node_a = MeshNode::new(id_a, mk_config(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk_config(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk_config(addr_c)).await.unwrap();

    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    // Connect A→B, then A→C (sequential handshake pairs via join)
    let (accept_result, connect_result) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    },);
    accept_result.expect("B accept A failed");
    connect_result.expect("A connect B failed");

    // A→C: C accepts, A connects
    let (accept_result, connect_result) = tokio::join!(node_c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_c, &pub_c, nid_c).await
    },);
    accept_result.expect("C accept A failed");
    connect_result.expect("A connect C failed");

    // Start all nodes
    node_a.start();
    node_b.start();
    node_c.start();

    assert_eq!(node_a.peer_count(), 2, "A should have 2 peers");
    assert_eq!(node_b.peer_count(), 1, "B should have 1 peer (A)");
    assert_eq!(node_c.peer_count(), 1, "C should have 1 peer (A)");

    // A sends to B and C
    tokio::time::sleep(Duration::from_millis(100)).await;
    node_a
        .send_to_peer(addr_b, &make_batch(0, 10, "to_b"))
        .await
        .unwrap();
    node_a
        .send_to_peer(addr_c, &make_batch(0, 10, "to_c"))
        .await
        .unwrap();

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

    let b_events = node_b.poll_shard(0, None, 100).await.unwrap();
    let c_events = node_c.poll_shard(0, None, 100).await.unwrap();

    assert!(
        !b_events.events.is_empty(),
        "B should receive from A, got {}",
        b_events.events.len()
    );
    assert!(
        !c_events.events.is_empty(),
        "C should receive from A, got {}",
        c_events.events.len()
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

/// MeshNode relay: A sends to C through B.
///
/// A and C don't have a direct UDP path — all traffic goes through B.
/// A has sessions with both B and C. A encrypts for C, prepends a routing
/// header, and sends to B. B forwards without decrypting. C receives and
/// decrypts.
///
/// This is the core "untrusted relay" test — B never sees the plaintext.
#[tokio::test]
async fn test_mesh_node_relay_through_middle() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();

    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk_config = |addr: SocketAddr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    let node_a = MeshNode::new(id_a, mk_config(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk_config(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk_config(addr_c)).await.unwrap();

    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    // A↔B: establish session
    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    },);
    r1.expect("B accept A failed");
    r2.expect("A connect B failed");

    // A↔C: establish session (A connects directly to C for key exchange,
    // but in production this could be relayed — we need the session keys)
    let (r1, r2) = tokio::join!(node_c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_c, &pub_c, nid_c).await
    },);
    r1.expect("C accept A failed");
    r2.expect("A connect C failed");

    // B↔C: establish session (so B can forward to C)
    let (r1, r2) = tokio::join!(node_c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_b.connect(addr_c, &pub_c, nid_c).await
    },);
    r1.expect("C accept B failed");
    r2.expect("B connect C failed");

    // Set up routing: A's route to C goes through B
    node_a.router().add_route(nid_c, addr_b);
    // B's route to C is direct
    node_b.router().add_route(nid_c, addr_c);

    // Start all nodes
    node_a.start();
    node_b.start();
    node_c.start();

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

    // A sends to C via routing (encrypted for C, routed through B)
    let batch = make_batch(0, 10, "relayed_from_a");
    node_a.send_routed(nid_c, &batch).await.unwrap();

    // Wait for relay
    tokio::time::sleep(Duration::from_millis(1500)).await;

    // C should receive the events
    let c_result = node_c.poll_shard(0, None, 100).await.unwrap();
    assert!(
        !c_result.events.is_empty(),
        "C should receive relayed events from A through B, got {}",
        c_result.events.len()
    );

    // Verify content
    for event in &c_result.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(json["tag"], "relayed_from_a");
    }

    // B should not have the events (it only forwarded, never decrypted)
    let b_result = node_b.poll_shard(0, None, 100).await.unwrap();
    assert_eq!(
        b_result.events.len(),
        0,
        "B should NOT have decrypted events — it's only a relay"
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

/// 2.2 — Relay preserves payload integrity over 100 events.
#[tokio::test]
async fn test_mesh_relay_preserves_payload_integrity() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();

    // A↔B, A↔C, B↔C
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    a.router().add_route(nid_c, addr_b);
    b.router().add_route(nid_c, addr_c);
    a.start();
    b.start();
    c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    let batch = make_batch(0, 100, "integrity_check");
    a.send_routed(nid_c, &batch).await.unwrap();
    tokio::time::sleep(Duration::from_millis(2000)).await;

    let result = c.poll_shard(0, None, 1000).await.unwrap();
    assert!(
        result.events.len() >= 50,
        "C should receive most of 100 relayed events, got {}",
        result.events.len()
    );
    for event in &result.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(
            json["tag"], "integrity_check",
            "payload corrupted during relay"
        );
    }

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

/// 2.3 — Relay tamper detection: malicious relay flips a byte, AEAD rejects.
#[tokio::test]
async fn test_mesh_relay_tamper_detected() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_c = *c.public_key();

    // A↔C session (for encryption keys)
    let (r1, r2) = tokio::join!(c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    a.router().add_route(nid_c, addr_b);
    a.start();
    c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // B is a malicious relay — raw UDP socket
    let evil_b = UdpSocket::bind(addr_b).await.unwrap();

    // A sends a routed packet
    let batch = make_batch(0, 5, "tamper_test");
    a.send_routed(nid_c, &batch).await.unwrap();

    // B receives and tampers
    let mut buf = vec![0u8; 8192];
    let (n, _) = tokio::time::timeout(Duration::from_secs(2), evil_b.recv_from(&mut buf))
        .await
        .expect("B recv timed out")
        .expect("B recv failed");

    use net::adapter::net::HEADER_SIZE;
    let tamper_offset = ROUTING_HEADER_SIZE + HEADER_SIZE + 10;
    if n > tamper_offset {
        buf[tamper_offset] ^= 0xFF;
    }

    evil_b.send_to(&buf[..n], addr_c).await.unwrap();

    tokio::time::sleep(Duration::from_millis(1000)).await;
    let result = c.poll_shard(0, None, 100).await.unwrap();
    assert_eq!(
        result.events.len(),
        0,
        "C should reject tampered packets — AEAD tag mismatch. Got {} events",
        result.events.len()
    );

    a.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

/// 3.1 — MeshNode failure detection over real encrypted sessions.
#[tokio::test]
async fn test_mesh_node_failure_detection() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_millis(500))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *b.public_key();

    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    a.start();
    b.start();

    tokio::time::sleep(Duration::from_millis(300)).await;
    assert_eq!(a.failure_detector().status(nid_b), NodeStatus::Healthy);

    // Kill B
    b.shutdown().await.unwrap();

    // Wait for detection
    tokio::time::sleep(Duration::from_millis(1500)).await;
    a.failure_detector().check_all();

    let status = a.failure_detector().status(nid_b);
    assert!(
        status == NodeStatus::Failed || status == NodeStatus::Suspected,
        "A should detect B's failure, got {:?}",
        status
    );

    a.shutdown().await.unwrap();
}

/// 3.1 — Reroute: A sends to B, B dies, A reroutes to C.
///
/// A has sessions with B and C. A routes to a destination D (logical)
/// through B. B dies. A updates the routing table to route D through C.
/// Subsequent events reach C. Proves the routing table update + send_routed
/// path supports rerouting without rebuilding sessions.
#[tokio::test]
async fn test_mesh_node_reroute_on_failure() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_secs(5))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();

    // A↔B
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();

    // A↔C
    let (r1, r2) = tokio::join!(c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    // B↔C (so B could forward — but B will die)
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    // Route to C goes through B initially
    a.router().add_route(nid_c, addr_b);
    b.router().add_route(nid_c, addr_c);

    a.start();
    b.start();
    c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Phase 1: A sends to C via B — works
    let batch1 = make_batch(0, 5, "before_failure");
    a.send_routed(nid_c, &batch1).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    let c_before = c.poll_shard(0, None, 100).await.unwrap();
    assert!(
        !c_before.events.is_empty(),
        "C should receive events via B before failure, got {}",
        c_before.events.len()
    );

    // Phase 2: B dies
    b.shutdown().await.unwrap();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Phase 3: A reroutes — update routing table to send directly to C
    a.router().remove_route(nid_c);
    a.router().add_route(nid_c, addr_c);

    // A sends again — should reach C directly now
    let batch2 = make_batch(0, 5, "after_reroute");
    a.send_routed(nid_c, &batch2).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    let c_after = c.poll_shard(0, None, 100).await.unwrap();
    assert!(
        !c_after.events.is_empty(),
        "C should receive events after reroute, got {}",
        c_after.events.len()
    );

    // Verify the rerouted events have the correct tag
    let has_rerouted = c_after.events.iter().any(|e| {
        e.parse()
            .map(|v: serde_json::Value| v["tag"] == "after_reroute")
            .unwrap_or(false)
    });
    assert!(has_rerouted, "C should have events tagged 'after_reroute'");

    a.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

/// Data survives reroute: events before and after B's death both reach C.
///
/// This tests the full sequence: send via relay, relay dies, reroute,
/// send directly. No events lost (verified by collecting all tags).
#[tokio::test]
async fn test_mesh_node_reroute_no_data_loss() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();

    // Full triangle
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    a.router().add_route(nid_c, addr_b);
    b.router().add_route(nid_c, addr_c);
    a.start();
    b.start();
    c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Send 10 events via relay
    let batch1 = make_batch(0, 10, "phase1_via_relay");
    a.send_routed(nid_c, &batch1).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    // Kill B, reroute direct
    b.shutdown().await.unwrap();
    a.router().remove_route(nid_c);
    a.router().add_route(nid_c, addr_c);

    // Send 10 more events directly
    let batch2 = make_batch(0, 10, "phase2_direct");
    a.send_routed(nid_c, &batch2).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    // Collect all events C received
    let result = c.poll_shard(0, None, 1000).await.unwrap();

    let phase1_count = result
        .events
        .iter()
        .filter(|e| {
            e.parse()
                .map(|v: serde_json::Value| v["tag"] == "phase1_via_relay")
                .unwrap_or(false)
        })
        .count();

    let phase2_count = result
        .events
        .iter()
        .filter(|e| {
            e.parse()
                .map(|v: serde_json::Value| v["tag"] == "phase2_direct")
                .unwrap_or(false)
        })
        .count();

    assert!(
        phase1_count > 0,
        "C should have received phase 1 (relayed) events, got {}",
        phase1_count
    );
    assert!(
        phase2_count > 0,
        "C should have received phase 2 (direct) events, got {}",
        phase2_count
    );

    a.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

// ============================================================================
// Phase 3: Migration over wire
// ============================================================================

use net::adapter::net::behavior::capability::CapabilityFilter;
use net::adapter::net::compute::orchestrator::wire as migration_wire;
use net::adapter::net::state::causal::CausalEvent;
use net::adapter::net::{
    DaemonError, DaemonFactoryRegistry, DaemonHost, DaemonHostConfig, DaemonRegistry, MeshDaemon,
    MigrationMessage, MigrationOrchestrator, MigrationPhase, MigrationSourceHandler,
    MigrationSubprotocolHandler, MigrationTargetHandler, SUBPROTOCOL_MIGRATION,
};

/// Simple stateful daemon for migration testing.
struct CounterDaemon {
    count: u64,
}

impl CounterDaemon {
    fn with_count(count: u64) -> Self {
        Self { count }
    }
}

impl MeshDaemon for CounterDaemon {
    fn name(&self) -> &str {
        "counter"
    }
    fn requirements(&self) -> CapabilityFilter {
        CapabilityFilter::default()
    }
    fn process(&mut self, _event: &CausalEvent) -> Result<Vec<Bytes>, DaemonError> {
        self.count += 1;
        Ok(vec![Bytes::from(self.count.to_le_bytes().to_vec())])
    }
    fn snapshot(&self) -> Option<Bytes> {
        Some(Bytes::from(self.count.to_le_bytes().to_vec()))
    }
    fn restore(&mut self, state: Bytes) -> Result<(), DaemonError> {
        if state.len() != 8 {
            return Err(DaemonError::RestoreFailed("bad state size".into()));
        }
        self.count = u64::from_le_bytes(state[..8].try_into().unwrap());
        Ok(())
    }
}

/// 7.1 — Migration snapshot request/response over encrypted UDP.
///
/// A sends TakeSnapshot to B over encrypted UDP. B's handler processes
/// it and takes the snapshot. Verified by checking B's source handler
/// state changed — proving the message survived wire encoding, encryption,
/// UDP, decryption, and subprotocol dispatch.
#[tokio::test]
async fn test_migration_snapshot_over_wire() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    // B: source node with a daemon
    let registry_b = Arc::new(DaemonRegistry::new());
    let daemon_kp = EntityKeypair::generate();
    let daemon_origin = daemon_kp.origin_hash();
    let host = DaemonHost::new(
        Box::new(CounterDaemon::with_count(42)),
        daemon_kp.clone(),
        DaemonHostConfig::default(),
    );
    registry_b.register(host).unwrap();

    let orchestrator_b = Arc::new(MigrationOrchestrator::new(registry_b.clone(), nid_b));
    let source_b = Arc::new(MigrationSourceHandler::new(registry_b.clone()));
    let target_b = Arc::new(MigrationTargetHandler::new(registry_b.clone()));
    let handler_b = Arc::new(MigrationSubprotocolHandler::new(
        orchestrator_b,
        source_b.clone(),
        target_b,
        nid_b,
    ));

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *node_b.public_key();

    node_b.set_migration_handler(handler_b);

    // A also needs a handler installed — and a matching factory
    // for `daemon_origin` — so the incoming `SnapshotReady` lands
    // successfully rather than getting bounced back with a
    // MigrationFailed reply that would clear B's migration record
    // before the test gets a chance to observe it. Post the
    // runtime-readiness "default handler" work, a bare `MeshNode`
    // with no migration handler synthesizes `ComputeNotSupported`;
    // a handler with no factory-for-origin emits `FactoryNotFound`;
    // either outcome aborts B's source_handler mid-flow. Wiring a
    // full A-side handler that can actually restore keeps the
    // original assertion observable.
    let registry_a = Arc::new(DaemonRegistry::new());
    let factories_a = Arc::new(DaemonFactoryRegistry::new());
    factories_a
        .register(daemon_kp.clone(), DaemonHostConfig::default(), || {
            Box::new(CounterDaemon::with_count(0))
        })
        .unwrap();
    let orch_a = Arc::new(MigrationOrchestrator::new(registry_a.clone(), nid_a));
    let source_a = Arc::new(MigrationSourceHandler::new(registry_a.clone()));
    let target_a = Arc::new(MigrationTargetHandler::new_with_factories(
        registry_a.clone(),
        factories_a,
    ));
    let handler_a = Arc::new(MigrationSubprotocolHandler::new(
        orch_a, source_a, target_a, nid_a,
    ));
    node_a.set_migration_handler(handler_a);

    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();

    node_a.start();
    node_b.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // A sends TakeSnapshot to B
    let msg = MigrationMessage::TakeSnapshot {
        daemon_origin,
        target_node: nid_a,
    };
    let encoded = migration_wire::encode(&msg).unwrap();
    node_a
        .send_subprotocol(addr_b, SUBPROTOCOL_MIGRATION, &encoded)
        .await
        .unwrap();

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

    // Verify B processed TakeSnapshot: source handler is in snapshot state
    let second_try = source_b.start_snapshot(daemon_origin, 0x9999, 0x1111);
    assert!(
        second_try.is_err(),
        "B should have processed TakeSnapshot over wire — source handler in snapshot state"
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
}

/// 7.2 — Full migration lifecycle: orchestrator starts migration, snapshot
/// flows from source to orchestrator over encrypted UDP.
///
/// A=orchestrator, B=source. A starts migration via the orchestrator API,
/// sends TakeSnapshot to B. B processes it and sends SnapshotReady back.
/// A's handler receives it and advances the orchestrator past Snapshot.
///
/// This proves the full round-trip: orchestrator → wire → source handler →
/// snapshot → wire → orchestrator state machine.
#[tokio::test]
async fn test_migration_full_lifecycle_over_wire() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    // B: source node with a daemon
    let registry_b = Arc::new(DaemonRegistry::new());
    let daemon_kp = EntityKeypair::generate();
    let daemon_origin = daemon_kp.origin_hash();
    let host = DaemonHost::new(
        Box::new(CounterDaemon::with_count(42)),
        daemon_kp.clone(),
        DaemonHostConfig::default(),
    );
    registry_b.register(host).unwrap();

    // A: orchestrator
    let registry_a = Arc::new(DaemonRegistry::new());

    // Create handlers
    let orch_a = Arc::new(MigrationOrchestrator::new(registry_a.clone(), nid_a));
    let handler_a = Arc::new(MigrationSubprotocolHandler::new(
        orch_a.clone(),
        Arc::new(MigrationSourceHandler::new(registry_a.clone())),
        Arc::new(MigrationTargetHandler::new(registry_a.clone())),
        nid_a,
    ));

    let orch_b = Arc::new(MigrationOrchestrator::new(registry_b.clone(), nid_b));
    let source_b = Arc::new(MigrationSourceHandler::new(registry_b.clone()));
    let handler_b = Arc::new(MigrationSubprotocolHandler::new(
        orch_b,
        source_b.clone(),
        Arc::new(MigrationTargetHandler::new(registry_b.clone())),
        nid_b,
    ));

    // C: target node — register a factory so the subprotocol handler can
    // auto-restore the daemon when SnapshotReady arrives over the wire.
    let registry_c = Arc::new(DaemonRegistry::new());
    let factories_c = Arc::new(DaemonFactoryRegistry::new());
    factories_c
        .register(daemon_kp.clone(), DaemonHostConfig::default(), || {
            Box::new(CounterDaemon::with_count(0))
        })
        .unwrap();
    let handler_c = Arc::new(MigrationSubprotocolHandler::new(
        Arc::new(MigrationOrchestrator::new(registry_c.clone(), nid_c)),
        Arc::new(MigrationSourceHandler::new(registry_c.clone())),
        Arc::new(MigrationTargetHandler::new_with_factories(
            registry_c.clone(),
            factories_c.clone(),
        )),
        nid_c,
    ));

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    node_a.set_migration_handler(handler_a);
    node_b.set_migration_handler(handler_b);
    node_c.set_migration_handler(handler_c);

    // Connect A↔B
    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();

    // Connect A↔C (needed for orchestrator to route SnapshotReady to target)
    let (r1, r2) = tokio::join!(node_c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    node_a.start();
    node_b.start();
    node_c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Start migration on orchestrator (remote source at B, target at C)
    orch_a.start_migration(daemon_origin, nid_b, nid_c).unwrap();
    assert_eq!(orch_a.status(daemon_origin), Some(MigrationPhase::Snapshot));

    // Send TakeSnapshot to B
    let msg = MigrationMessage::TakeSnapshot {
        daemon_origin,
        target_node: nid_c,
    };
    let encoded = migration_wire::encode(&msg).unwrap();
    node_a
        .send_subprotocol(addr_b, SUBPROTOCOL_MIGRATION, &encoded)
        .await
        .unwrap();

    // Wait for the full round trip: B takes snapshot, ships to A, A forwards
    // to C, C restores, Restore/Replay/Cutover/Cleanup/Activate chain back.
    tokio::time::sleep(Duration::from_millis(3000)).await;

    // Verify B processed TakeSnapshot (source_handler has an in-flight record).
    let source_check = source_b.start_snapshot(daemon_origin, 0x9999, 0x1111);
    assert!(
        source_check.is_err(),
        "B should have processed TakeSnapshot"
    );

    // After the full lifecycle chains end-to-end, A's orchestrator record
    // should have been torn down at ActivateAck. The daemon lives on C and
    // has been unregistered from B by source_handler.cleanup.
    assert!(
        !orch_a.is_migrating(daemon_origin),
        "orchestrator record should be removed after ActivateAck"
    );
    assert!(
        registry_c.contains(daemon_origin),
        "daemon should be registered on target C after full lifecycle"
    );
    assert!(
        !registry_b.contains(daemon_origin),
        "daemon should be unregistered from source B after cleanup"
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

// ============================================================================
// Phase 4: Partition simulation
// ============================================================================

/// 8.1 — Partition: block traffic between A and B, verify detection.
///
/// A and B are connected. A blocks B's address via PartitionFilter.
/// Heartbeats stop. A's failure detector marks B as failed.
/// Proves partition simulation works through the MeshNode runtime.
#[tokio::test]
async fn test_partition_detection_via_filter() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_millis(500))
    };

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *node_b.public_key();

    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    node_a.start();
    node_b.start();

    // Verify healthy
    tokio::time::sleep(Duration::from_millis(300)).await;
    assert_eq!(node_a.failure_detector().status(nid_b), NodeStatus::Healthy);

    // Partition: A blocks B
    node_a.block_peer(addr_b);
    assert!(node_a.is_blocked(&addr_b));

    // Wait for heartbeats to time out
    tokio::time::sleep(Duration::from_millis(1500)).await;
    node_a.failure_detector().check_all();

    let status = node_a.failure_detector().status(nid_b);
    assert!(
        status == NodeStatus::Failed || status == NodeStatus::Suspected,
        "A should detect B as failed during partition, got {:?}",
        status
    );

    // Data should be silently dropped during partition
    let batch = make_batch(0, 5, "during_partition");
    node_a.send_to_peer(addr_b, &batch).await.unwrap(); // succeeds silently
    tokio::time::sleep(Duration::from_millis(500)).await;
    let b_events = node_b.poll_shard(0, None, 100).await.unwrap();
    // B might have events from before the partition but none tagged "during_partition"
    let partition_events: Vec<_> = b_events
        .events
        .iter()
        .filter(|e| {
            e.parse()
                .map(|v: serde_json::Value| v["tag"] == "during_partition")
                .unwrap_or(false)
        })
        .collect();
    assert_eq!(
        partition_events.len(),
        0,
        "B should not receive events sent during partition"
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
}

/// 8.2 — Partition healing: unblock traffic, verify recovery.
///
/// A blocks B, then unblocks. After unblocking, heartbeats resume
/// and A can send events to B again.
#[tokio::test]
async fn test_partition_healing() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_millis(500))
    };

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *node_b.public_key();

    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    node_a.start();
    node_b.start();
    tokio::time::sleep(Duration::from_millis(300)).await;

    // Partition
    node_a.block_peer(addr_b);
    tokio::time::sleep(Duration::from_millis(1000)).await;

    // Heal
    node_a.unblock_peer(&addr_b);
    assert!(!node_a.is_blocked(&addr_b));

    // Wait for heartbeats to resume (B is still sending heartbeats;
    // A was blocking them. Once unblocked, A will receive B's next heartbeat
    // and the failure detector will recover B to Healthy.)
    tokio::time::sleep(Duration::from_millis(1000)).await;

    // B sends heartbeats from its loop → A receives → failure_detector.heartbeat()
    let status = node_a.failure_detector().status(nid_b);
    assert_eq!(
        status,
        NodeStatus::Healthy,
        "B should recover to Healthy after partition heals, got {:?}",
        status
    );

    // Data should flow again
    let batch = make_batch(0, 10, "after_healing");
    node_a.send_to_peer(addr_b, &batch).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    let b_events = node_b.poll_shard(0, None, 100).await.unwrap();
    let healed_events: Vec<_> = b_events
        .events
        .iter()
        .filter(|e| {
            e.parse()
                .map(|v: serde_json::Value| v["tag"] == "after_healing")
                .unwrap_or(false)
        })
        .collect();
    assert!(
        !healed_events.is_empty(),
        "B should receive events after partition heals"
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
}

/// 8.3 — Three-node partition: A isolated from B and C.
///
/// A blocks both B and C. B and C can still communicate with each other.
/// Proves asymmetric partition — one node isolated, the other two unaffected.
#[tokio::test]
async fn test_partition_asymmetric_three_node() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_secs(5))
    };

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    // A↔B, A↔C, B↔C
    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(node_c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(node_c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    node_a.start();
    node_b.start();
    node_c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Partition: A is isolated (blocks both B and C)
    node_a.block_peer(addr_b);
    node_a.block_peer(addr_c);
    // Also block A on B and C side (partition is bidirectional)
    node_b.block_peer(addr_a);
    node_c.block_peer(addr_a);

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

    // B→C should still work (they're on the same side of the partition)
    let batch = make_batch(0, 10, "b_to_c_during_partition");
    node_b.send_to_peer(addr_c, &batch).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    let c_events = node_c.poll_shard(0, None, 100).await.unwrap();
    let bc_events: Vec<_> = c_events
        .events
        .iter()
        .filter(|e| {
            e.parse()
                .map(|v: serde_json::Value| v["tag"] == "b_to_c_during_partition")
                .unwrap_or(false)
        })
        .collect();
    assert!(
        !bc_events.is_empty(),
        "B→C should work during A's partition, got {} events",
        bc_events.len()
    );

    // A→B should be blocked
    let batch = make_batch(0, 5, "a_to_b_blocked");
    node_a.send_to_peer(addr_b, &batch).await.unwrap();
    tokio::time::sleep(Duration::from_millis(500)).await;
    let b_events = node_b.poll_shard(0, None, 100).await.unwrap();
    let blocked_events: Vec<_> = b_events
        .events
        .iter()
        .filter(|e| {
            e.parse()
                .map(|v: serde_json::Value| v["tag"] == "a_to_b_blocked")
                .unwrap_or(false)
        })
        .collect();
    assert_eq!(
        blocked_events.len(),
        0,
        "A→B should be blocked during partition"
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

// ============================================================================
// Automatic rerouting
// ============================================================================

/// Auto-reroute: A sends to C via B. B dies. The reroute policy
/// automatically updates the route to C directly. No manual
/// add_route/remove_route calls — the failure detector triggers it.
#[tokio::test]
async fn test_mesh_node_auto_reroute() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_millis(600))
    };

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    // Full triangle
    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(node_c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(node_c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    // Route to C goes through B
    node_a.router().add_route(nid_c, addr_b);
    node_b.router().add_route(nid_c, addr_c);

    node_a.start();
    node_b.start();
    node_c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Verify reroute policy has no active reroutes
    assert_eq!(node_a.reroute_policy().active_reroutes(), 0);

    // Phase 1: send via B — works
    let batch1 = make_batch(0, 5, "before_auto_reroute");
    node_a.send_routed(nid_c, &batch1).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    let c_before = node_c.poll_shard(0, None, 100).await.unwrap();
    assert!(!c_before.events.is_empty(), "C should receive via B");

    // Phase 2: kill B
    node_b.shutdown().await.unwrap();

    // Wait for failure detection + automatic reroute
    // Heartbeat interval=200ms, timeout=600ms, miss_threshold=3
    // Detection should happen within ~2s
    tokio::time::sleep(Duration::from_millis(2500)).await;
    node_a.failure_detector().check_all();

    // The reroute policy should have triggered automatically
    assert!(
        node_a.reroute_policy().active_reroutes() > 0,
        "reroute policy should have rerouted after B's failure"
    );

    // Phase 3: send again — should reach C via auto-rerouted path (direct)
    // NO manual route update — the reroute policy did it automatically
    let batch2 = make_batch(0, 5, "after_auto_reroute");
    node_a.send_routed(nid_c, &batch2).await.unwrap();
    tokio::time::sleep(Duration::from_millis(1000)).await;

    let c_after = node_c.poll_shard(0, None, 100).await.unwrap();
    let auto_events: Vec<_> = c_after
        .events
        .iter()
        .filter(|e| {
            e.parse()
                .map(|v: serde_json::Value| v["tag"] == "after_auto_reroute")
                .unwrap_or(false)
        })
        .collect();
    assert!(
        !auto_events.is_empty(),
        "C should receive events after automatic reroute (no manual route update)"
    );

    node_a.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

/// Auto-reroute recovery: B dies → auto-reroute → B comes back →
/// original route restored automatically.
#[tokio::test]
async fn test_mesh_node_auto_reroute_recovery() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_millis(600))
    };

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    // Full triangle
    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(node_c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    // Route to C goes through B
    node_a.router().add_route(nid_c, addr_b);

    node_a.start();
    node_b.start();
    node_c.start();
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Simulate B failure via partition (not shutdown — B stays alive for recovery)
    node_a.block_peer(addr_b);
    node_b.block_peer(addr_a);

    // Wait for detection + auto-reroute
    tokio::time::sleep(Duration::from_millis(2500)).await;
    node_a.failure_detector().check_all();
    assert!(
        node_a.reroute_policy().active_reroutes() > 0,
        "should auto-reroute"
    );

    // Heal partition — B sends heartbeats again → failure detector recovers B
    node_a.unblock_peer(&addr_b);
    node_b.unblock_peer(&addr_a);

    // Wait for recovery heartbeats
    tokio::time::sleep(Duration::from_millis(1000)).await;

    // Recovery should restore the original route
    assert_eq!(
        node_a.reroute_policy().active_reroutes(),
        0,
        "original route should be restored after B recovers"
    );

    // Verify the route points back to B
    let next_hop = node_a.router().routing_table().lookup(nid_c);
    assert_eq!(
        next_hop,
        Some(addr_b),
        "route to C should be restored through B"
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

/// Proximity graph: A↔B↔C (no direct A↔C link). A discovers C through
/// B's relayed pingwaves.
#[tokio::test]
async fn test_proximity_graph_pingwave_discovery() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_secs(10))
    };

    let node_a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    // A↔B only
    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();

    // B↔C only (A has NO direct connection to C)
    let (r1, r2) = tokio::join!(node_c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    node_a.start();
    node_b.start();
    node_c.start();

    // A initially knows 1 peer (B)
    assert_eq!(node_a.proximity_graph().node_count(), 1);

    // Wait for pingwave propagation (C→B→A)
    tokio::time::sleep(Duration::from_millis(1500)).await;

    // A should now know about C via B's relayed pingwave
    let a_nodes = node_a.proximity_graph().node_count();
    assert!(
        a_nodes >= 2,
        "A should discover C via B's pingwave relay, knows {} nodes",
        a_nodes
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

// ============================================================================
// Handshake relay tests
// ============================================================================

/// A establishes a session with C via relay B without any direct A↔C
/// UDP path. After the relayed handshake, A can send data to C via
/// `send_routed`, and the payload arrives intact.
#[tokio::test]
async fn test_mesh_handshake_via_relay() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();

    // A↔B direct session.
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.expect("B accept A failed");
    r2.expect("A connect B failed");

    // B↔C direct session (B is the relay — must have a path to C).
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.expect("C accept B failed");
    r2.expect("B connect C failed");

    // Start all nodes — receive loops must be running before connect_via.
    a.start();
    b.start();
    c.start();

    // A has no direct UDP path to C. Establish a session via B.
    let result = a.connect_via(addr_b, &pub_c, nid_c).await;
    result.expect("A connect_via B to C failed");

    // Route for forwarded data: on B, A→C already exists from b.connect(C).
    // On A, connect_via already inserted a route for C via B.
    // On B, add a route for A→B so data C→A gets forwarded correctly.
    b.router().add_route(nid_a, addr_a);

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

    // A sends a batch to C over the newly established A↔C session, routed
    // through B.
    let batch = make_batch(0, 10, "via_relay_handshake");
    a.send_routed(nid_c, &batch)
        .await
        .expect("A send_routed to C failed");

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

    let c_result = c.poll_shard(0, None, 100).await.unwrap();
    assert!(
        !c_result.events.is_empty(),
        "C should receive events from A via relayed-handshake session, got {}",
        c_result.events.len()
    );
    for event in &c_result.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(json["tag"], "via_relay_handshake");
    }

    let b_result = b.poll_shard(0, None, 100).await.unwrap();
    assert_eq!(
        b_result.events.len(),
        0,
        "B should not decrypt A→C data — it only relays"
    );

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

/// After a handshake established via relay B, data flows in both directions
/// between A and C through B.
#[tokio::test]
async fn test_mesh_handshake_relay_bidirectional() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();

    // A↔B and B↔C direct sessions.
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    a.start();
    b.start();
    c.start();

    a.connect_via(addr_b, &pub_c, nid_c)
        .await
        .expect("connect_via failed");

    // Routing: B already has A (from b.accept/add_route) and C (from
    // b.connect/add_route). On C, the responder side added a route for A
    // via B's addr during the handshake. On A, connect_via added the route
    // for C via B. All four directions are covered.
    b.router().add_route(nid_a, addr_a);

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

    // A → C
    a.send_routed(nid_c, &make_batch(0, 5, "a_to_c"))
        .await
        .unwrap();
    // C → A
    c.send_routed(nid_a, &make_batch(0, 5, "c_to_a"))
        .await
        .unwrap();

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

    let c_events = c.poll_shard(0, None, 100).await.unwrap();
    let a_events = a.poll_shard(0, None, 100).await.unwrap();

    assert!(
        !c_events.events.is_empty(),
        "C should receive events from A via B"
    );
    assert!(
        !a_events.events.is_empty(),
        "A should receive events from C via B"
    );
    for event in &c_events.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(json["tag"], "a_to_c");
    }
    for event in &a_events.events {
        let json: serde_json::Value = event.parse().unwrap();
        assert_eq!(json["tag"], "c_to_a");
    }

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

// ============================================================================
// Stream multiplexing
// ============================================================================

/// Two streams between the same pair of peers, distinct stream IDs,
/// distinct stream stats, and data flows on both. Verifies that:
/// 1. `open_stream` returns a handle backed by per-stream state.
/// 2. `send_on_stream` ships events that land in the right shard inbound.
/// 3. `all_stream_stats` reports both streams' tx_seq independently.
#[tokio::test]
async fn test_stream_multiplex_two_streams_same_peer() {
    use net::adapter::net::{Reliability, StreamConfig};

    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(4)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *b.public_key();

    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();

    a.start();
    b.start();

    // Open two streams with distinct reliability modes.
    let s_fire = a
        .open_stream(
            nid_b,
            111,
            StreamConfig::new().with_reliability(Reliability::FireAndForget),
        )
        .unwrap();
    let s_rel = a
        .open_stream(
            nid_b,
            222,
            StreamConfig::new().with_reliability(Reliability::Reliable),
        )
        .unwrap();

    let events_fire: Vec<Bytes> = (0..3)
        .map(|i| Bytes::from(format!(r#"{{"stream":"fire","i":{}}}"#, i)))
        .collect();
    let events_rel: Vec<Bytes> = (0..5)
        .map(|i| Bytes::from(format!(r#"{{"stream":"rel","i":{}}}"#, i)))
        .collect();

    a.send_on_stream(&s_fire, &events_fire).await.unwrap();
    a.send_on_stream(&s_rel, &events_rel).await.unwrap();

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

    // Both streams recorded their sends independently.
    let all = a.all_stream_stats(nid_b);
    let fire_stats = all
        .iter()
        .find(|(sid, _)| *sid == 111)
        .map(|(_, s)| *s)
        .expect("stream 111 stats");
    let rel_stats = all
        .iter()
        .find(|(sid, _)| *sid == 222)
        .map(|(_, s)| *s)
        .expect("stream 222 stats");
    assert_eq!(
        fire_stats.tx_seq, 1,
        "fire stream sent 1 packet (3 events fit in one)"
    );
    assert_eq!(
        rel_stats.tx_seq, 1,
        "rel stream sent 1 packet (5 events fit in one)"
    );
    assert!(fire_stats.active);
    assert!(rel_stats.active);

    // Single-stream accessor matches.
    let fire_solo = a.stream_stats(nid_b, 111).unwrap();
    assert_eq!(fire_solo.tx_seq, 1);

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
}

/// `open_stream` is idempotent for a given `(peer, stream_id)`: re-opens
/// return handles backed by the same underlying state. Closing and then
/// re-opening creates fresh state.
#[tokio::test]
async fn test_stream_open_close_idempotency() {
    use net::adapter::net::{Reliability, StreamConfig};

    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *b.public_key();

    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();

    // First open creates state.
    let first = a
        .open_stream(
            nid_b,
            77,
            StreamConfig::new().with_reliability(Reliability::Reliable),
        )
        .unwrap();
    assert_eq!(first.stream_id(), 77);
    a.send_on_stream(&first, &[Bytes::from_static(b"{}")])
        .await
        .unwrap();
    let stats1 = a.stream_stats(nid_b, 77).unwrap();
    assert_eq!(stats1.tx_seq, 1);

    // Second open: same tx_seq state — open is idempotent.
    let second = a
        .open_stream(
            nid_b,
            77,
            StreamConfig::new().with_reliability(Reliability::Reliable),
        )
        .unwrap();
    a.send_on_stream(&second, &[Bytes::from_static(b"{}")])
        .await
        .unwrap();
    let stats2 = a.stream_stats(nid_b, 77).unwrap();
    assert_eq!(
        stats2.tx_seq, 2,
        "second send on re-opened stream continues tx_seq"
    );

    // Close + re-open creates fresh state.
    a.close_stream(nid_b, 77);
    assert!(a.stream_stats(nid_b, 77).is_none());
    let third = a.open_stream(nid_b, 77, StreamConfig::new()).unwrap();
    a.send_on_stream(&third, &[Bytes::from_static(b"{}")])
        .await
        .unwrap();
    let stats3 = a.stream_stats(nid_b, 77).unwrap();
    assert_eq!(stats3.tx_seq, 1, "after close+reopen tx_seq resets");

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
}

/// Regression: `send_on_stream` used to call `session.get_or_create_stream`,
/// which would silently revive a closed stream with *default* config
/// (losing the caller's original `Reliability` / `tx_window` /
/// `fairness_weight`) on the next send. A stale `Stream` handle
/// reaching back into a closed session thus produced a stream the
/// caller never explicitly opened.
///
/// Fix: `send_on_stream` now checks `session.try_stream(stream_id)`
/// and returns `StreamError::NotConnected` if the stream isn't
/// currently open. Callers that want auto-create behavior use
/// `send_to_peer` / `send_routed` instead of the typed handle API.
#[tokio::test]
async fn test_regression_send_on_stream_rejects_closed_stream() {
    use net::adapter::net::{Reliability, StreamConfig, StreamError};

    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
    };
    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *b.public_key();

    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();

    // Open a reliable stream, send once to prime state.
    let stream = a
        .open_stream(
            nid_b,
            123,
            StreamConfig::new().with_reliability(Reliability::Reliable),
        )
        .unwrap();
    a.send_on_stream(&stream, &[Bytes::from_static(b"{}")])
        .await
        .unwrap();

    // Close the stream. The `stream` handle is now stale — the
    // session no longer tracks stream_id 123.
    a.close_stream(nid_b, 123);
    assert!(a.stream_stats(nid_b, 123).is_none());

    // Send on the stale handle. Must return NotConnected — NOT
    // silently recreate with default (FireAndForget, tx_window=0)
    // config and succeed.
    let result = a
        .send_on_stream(&stream, &[Bytes::from_static(b"{}")])
        .await;
    assert!(
        matches!(result, Err(StreamError::NotConnected)),
        "expected NotConnected for send on closed stream; got {:?}",
        result
    );
    // And the session still has no entry for 123 — no silent
    // recreation happened.
    assert!(
        a.stream_stats(nid_b, 123).is_none(),
        "send on closed stream must NOT revive the stream"
    );

    // Now the trickier case: close a stream, reopen it with fresh
    // config, and verify the ORIGINAL stale handle still refuses to
    // operate on the new stream. The new `Stream` handle works
    // normally; only the stale one is inert.
    let fresh = a
        .open_stream(
            nid_b,
            123,
            StreamConfig::new().with_reliability(Reliability::FireAndForget),
        )
        .unwrap();
    let stale_result = a
        .send_on_stream(&stream, &[Bytes::from_static(b"{}")])
        .await;
    assert!(
        matches!(stale_result, Err(StreamError::NotConnected)),
        "stale handle from pre-reopen lifetime must still refuse; got {:?}",
        stale_result
    );
    a.send_on_stream(&fresh, &[Bytes::from_static(b"{}")])
        .await
        .expect("fresh handle after reopen must work");

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
}

// ============================================================================
// Multi-hop routing discovery
// ============================================================================

/// Pingwave-driven route install: a node learns routes to non-direct
/// peers from pingwaves that arrive with `hop_count > 0`. On a 4-node
/// chain A↔B↔C↔D where only adjacent pairs have direct sessions, A
/// should eventually learn a next-hop toward D purely from pingwave
/// propagation — no manual `add_route` needed.
///
/// The installer places `origin → from` (the immediate forwarder is the
/// next hop) with metric `hop_count + 2`. Direct routes at metric 1
/// remain authoritative and are not downgraded.
#[tokio::test]
async fn test_multi_hop_routing_pingwave_installs_indirect_route() {
    let ports = find_ports(4).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let id_d = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();
    let nid_d = id_d.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();
    let addr_d: SocketAddr = format!("127.0.0.1:{}", ports[3]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let d = MeshNode::new(id_d, mk(addr_d)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();
    let pub_d = *d.public_key();

    // Build the chain: A↔B, B↔C, C↔D. No other direct sessions.
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(d.accept(nid_c), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        c.connect(addr_d, &pub_d, nid_d).await
    });
    r1.unwrap();
    r2.unwrap();

    // Start all nodes so pingwaves start flowing.
    a.start();
    b.start();
    c.start();
    d.start();

    // Wait for enough heartbeat intervals that pingwaves have reached
    // A from D (3 hops × 200ms plus some slack). The heartbeat loop
    // fires on its interval; let a handful of them propagate.
    tokio::time::sleep(Duration::from_millis(1500)).await;

    // A's routing table must have a route for D pointing at B (the
    // only peer A has that can get a pingwave from D to us first).
    let lookup = a.router().routing_table().lookup(nid_d);
    assert_eq!(
        lookup,
        Some(addr_b),
        "A should have learned D via B from pingwave propagation; got {:?}",
        lookup
    );

    // Direct routes must not have been disturbed. A's route to B is
    // metric 1 (direct); the pingwave-carried route to B arriving via
    // B itself would have metric 2 and must NOT win.
    let b_lookup = a.router().routing_table().lookup(nid_b);
    assert_eq!(b_lookup, Some(addr_b), "direct B route preserved");
    let _ = nid_a;
    let _ = nid_c;

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
    d.shutdown().await.unwrap();
}

/// DV plan: after pingwave propagation, `ProximityGraph::path_to` must
/// return a real multi-hop path — not `None`. Previously `edges` was
/// never populated, so `path_to` always returned `None` and
/// `ReroutePolicy` fell back to "any direct peer". This is the primary
/// before/after for the DV plan: `path_to` is now usable.
#[tokio::test]
async fn test_regression_dv_path_to_returns_multi_hop() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(200))
            .with_session_timeout(Duration::from_secs(30))
    };
    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();

    // Chain: A↔B, B↔C. A has no direct session with C.
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    a.start();
    b.start();
    c.start();

    // Wait for pingwaves from C to reach A via B.
    tokio::time::sleep(Duration::from_millis(1500)).await;

    // A's proximity graph should now have a multi-hop path to C.
    // Graph-id encoding: low 8 bytes = u64 node_id LE, high 24 bytes
    // zero (matches the `node_id_to_graph_id` convention used
    // everywhere node_ids enter the proximity graph).
    let mut c_graph_id = [0u8; 32];
    c_graph_id[0..8].copy_from_slice(&nid_c.to_le_bytes());
    let path = a.proximity_graph().path_to(&c_graph_id);
    assert!(
        path.is_some(),
        "path_to(C) must be Some now that edges populate on pingwave receipt"
    );
    let path = path.unwrap();
    assert!(
        path.len() >= 2,
        "path_to(C) should have at least 2 nodes (self + next hop); got {:?}",
        path.len()
    );

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

/// Regression: the pingwave dispatch path used to accept pingwaves from
/// any UDP source — including addresses that had never completed a
/// handshake — and install both a `RoutingTable` entry and a
/// `ProximityGraph` node for the forged origin. That let anyone on the
/// wire poison the victim's routing/topology state by fabricating a
/// pingwave claiming to be a next-hop for an arbitrary origin.
///
/// Fix: the dispatch now looks up `source` in `addr_to_node` and drops
/// the pingwave if the source isn't a registered direct peer.
///
/// Before the fix, this test would have seen the forged origin appear
/// in both the routing table and the proximity graph. After the fix,
/// neither surface changes — the pingwave is silently dropped at the
/// dispatch boundary.
#[tokio::test]
async fn test_regression_pingwave_from_unregistered_source_is_dropped() {
    use net::adapter::net::behavior::EnhancedPingwave;
    use tokio::net::UdpSocket;

    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let attacker_addr: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let a = MeshNode::new(
        id_a,
        MeshNodeConfig::new(addr_a, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30)),
    )
    .await
    .unwrap();
    a.start();

    // Bind a raw UDP socket that has NOT completed any handshake with A.
    let attacker = UdpSocket::bind(attacker_addr).await.unwrap();

    // Forge a pingwave claiming origin 0xDEADBEEF — a node the attacker
    // wants A to install a route for. `hop_count = 1` is a lie; an
    // honest intermediate would only be 1 hop away, but A has no way
    // to verify the claim without authenticating the source.
    let mut forged_origin_graph_id = [0u8; 32];
    let forged_origin_nid: u64 = 0xDEAD_BEEF_CAFE_F00D;
    forged_origin_graph_id[0..8].copy_from_slice(&forged_origin_nid.to_le_bytes());
    let mut pw = EnhancedPingwave::new(forged_origin_graph_id, 1, 3);
    pw.hop_count = 1;
    let pw_bytes = pw.to_bytes();

    // Send from the unregistered socket directly to A. A will dispatch
    // it through the pingwave path.
    attacker.send_to(&pw_bytes, addr_a).await.unwrap();

    // Give A's receive loop a generous moment to process. The dispatch
    // is synchronous after parse; 200 ms is plenty.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Assertion 1: no route for the forged origin was installed.
    let lookup = a.router().routing_table().lookup(forged_origin_nid);
    assert!(
        lookup.is_none(),
        "routing table must NOT install a route for a forged origin \
         advertised from an unregistered source; got next_hop={:?}",
        lookup
    );

    // Assertion 2: no proximity graph node materialized for the forged
    // origin. Before the fix, `on_pingwave_from` would have added it.
    assert!(
        a.proximity_graph()
            .get_node(&forged_origin_graph_id)
            .is_none(),
        "proximity graph must NOT contain a node for a forged origin \
         advertised from an unregistered source"
    );

    // Assertion 3: no graph edge materialized. Even though the
    // attacker's addr could resolve to some synthetic node, we expect
    // the graph to stay empty of edges rooted at the attacker.
    // (Checked indirectly: `path_to(forged_origin)` must be None.)
    assert!(
        a.proximity_graph()
            .path_to(&forged_origin_graph_id)
            .is_none(),
        "path_to(forged_origin) must be None — no edge should have \
         been installed by the dropped pingwave"
    );

    drop(attacker);
    a.shutdown().await.unwrap();
}

// ============================================================================
// Regressions
// ============================================================================

/// Regression: `MeshNode` used to seed its local proximity-graph identity as
/// `entity_id().as_bytes()` while peers were seeded via the zero-padded
/// `node_id` encoding (`node_id_to_graph_id`). That mismatch meant the
/// graph stored the same logical node under two different keys — any
/// `path_to(...)` or `all_nodes()` query that joined self with peers would
/// see inconsistent identities.
///
/// The fix is to use `node_id_to_graph_id(node_id)` in both places. This
/// test exercises the fix via the observable `create_pingwave()` output:
/// the pingwave's `origin_id` is the local graph identity, and must equal
/// `node_id.to_le_bytes()` zero-padded to 32 bytes.
#[tokio::test]
async fn test_regression_proximity_graph_local_id_matches_peer_encoding() {
    use net::adapter::net::behavior::loadbalance::HealthStatus;

    let ports = find_ports(1).await;
    let psk = [0x42u8; 32];
    let id = EntityKeypair::generate();
    let expected_nid = id.node_id();
    let addr: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();

    let node = MeshNode::new(id, MeshNodeConfig::new(addr, psk))
        .await
        .unwrap();

    let pw = node
        .proximity_graph()
        .create_pingwave(HealthStatus::Healthy);

    let mut expected = [0u8; 32];
    expected[0..8].copy_from_slice(&expected_nid.to_le_bytes());
    assert_eq!(
        pw.origin_id, expected,
        "proximity graph's local id must use the zero-padded node_id \
         encoding so it matches what peers see when they seed this node \
         into their own graph"
    );

    node.shutdown().await.unwrap();
}

/// Regression: `HandshakeAction::Forward` used to require a direct peer
/// session for the `to_node` — if the next hop toward the destination
/// was a different intermediate node, the packet was silently dropped.
/// That meant `connect_via` could only traverse one relay, contradicting
/// the documented "chain of already-connected peers" design.
///
/// The fix looks up `to_node` in the routing table first; only if that
/// misses does it fall back to the direct-peer entry. This test sets up
/// a four-node chain A↔B↔C↔D, gives each relay the route to the far
/// destination, and confirms that `a.connect_via(relay=B, D)` completes
/// end-to-end with A gaining a peer entry for D.
#[tokio::test]
async fn test_regression_handshake_relay_multi_hop_via_routing_table() {
    let ports = find_ports(4).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let id_d = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();
    let nid_d = id_d.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();
    let addr_d: SocketAddr = format!("127.0.0.1:{}", ports[3]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let d = MeshNode::new(id_d, mk(addr_d)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();
    let pub_d = *d.public_key();

    // Build the chain: A↔B, B↔C, C↔D. No other direct sessions.
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(d.accept(nid_c), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        c.connect(addr_d, &pub_d, nid_d).await
    });
    r1.unwrap();
    r2.unwrap();

    // Routing table entries for the relays. The `connect`/`accept`
    // calls above auto-install routes for direct peers only; we have to
    // seed the cross-relay routes manually.
    //
    // msg1 path A→B→C→D:  B needs route D via C (C has D direct).
    // msg2 path D→C→B→A:  C needs route A via B (B has A direct).
    b.router().add_route(nid_d, addr_c);
    c.router().add_route(nid_a, addr_b);

    a.start();
    b.start();
    c.start();
    d.start();

    // Register a handshake handler factory on D? No — handshake relay
    // doesn't use DaemonFactoryRegistry. D just needs to be running.
    //
    // Register D as a reachable peer on A's side (no direct session
    // yet). After connect_via, A should have one.
    let _before = a.peer_count();

    // Drive the relayed handshake across two intermediate hops.
    a.connect_via(addr_b, &pub_d, nid_d)
        .await
        .expect("A connect_via D (through B→C) failed");

    // Give the spawned msg2-send tasks on D and C time to resolve.
    tokio::time::sleep(Duration::from_millis(500)).await;

    // A now has sessions for B and for D (but not C — C was purely a
    // transit hop as far as A is concerned).
    assert!(
        a.peer_count() >= 2,
        "A must have at least B + D after multi-hop connect_via; got {}",
        a.peer_count()
    );
    // D now has sessions for C and for A.
    assert!(
        d.peer_count() >= 2,
        "D must have at least C + A after the relayed handshake completes; got {}",
        d.peer_count()
    );
    let _ = nid_a;

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
    d.shutdown().await.unwrap();
}

/// Regression: `execute_handshake_action::RegisterResponderPeer` used to
/// insert the peer + session + route *before* the msg2 send was spawned,
/// and the send result was ignored. If the spawned send failed (partition
/// filter set, socket error, etc.), the responder would keep a half-open
/// peer record: the initiator never sees msg2 and the responder has a
/// session it will never be able to use.
///
/// The fix is to register the peer only after the msg2 send succeeds,
/// inside the spawned task. This test asserts the happy path still
/// registers — the converse (failure-path silence) is structurally
/// guaranteed by the spawned-task ordering in
/// `execute_handshake_action` and covered in a code review.
#[tokio::test]
async fn test_regression_handshake_relay_registers_peer_after_msg2_sent() {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = MeshNode::new(id_a, mk(addr_a)).await.unwrap();
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let c = MeshNode::new(id_c, mk(addr_c)).await.unwrap();
    let pub_b = *b.public_key();
    let pub_c = *c.public_key();

    // A↔B and B↔C direct sessions.
    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    let (r1, r2) = tokio::join!(c.accept(nid_b), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        b.connect(addr_c, &pub_c, nid_c).await
    });
    r1.unwrap();
    r2.unwrap();

    // After direct connects, both sides should see exactly one peer.
    assert_eq!(a.peer_count(), 1);
    assert_eq!(c.peer_count(), 1);

    a.start();
    b.start();
    c.start();

    a.connect_via(addr_b, &pub_c, nid_c).await.unwrap();

    // Wait for the spawned msg2-send on C to resolve before we inspect.
    // Poll rather than sleep — scheduler jitter can easily exceed a
    // 300 ms fixed delay under load.
    tokio::time::timeout(Duration::from_secs(2), async {
        while c.peer_count() < 2 {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("timed out waiting for C to register A after connect_via");

    // Happy path: A now has C registered, C now has A registered. Both
    // registrations ride on msg2 actually having been sent; without the
    // fix, C's A-peer insertion races the send and can happen even if
    // the send silently fails.
    assert_eq!(a.peer_count(), 2, "A must have B and C");
    assert_eq!(
        c.peer_count(),
        2,
        "C must have B and A — the A entry is only inserted after the \
         spawned msg2-send succeeds"
    );
    let _ = nid_a;

    a.shutdown().await.unwrap();
    b.shutdown().await.unwrap();
    c.shutdown().await.unwrap();
}

/// Regression: the receive loop used to dispatch any 72-byte UDP packet as
/// a pingwave based on length alone. A packet whose leading bytes were the
/// Net-header magic (`0x4E45`) would still be mis-handled as a pingwave,
/// bypassing normal decryption and session-id validation.
///
/// The fix is a structural check: only dispatch as a pingwave when the
/// leading two bytes are NOT the Net magic. A length-72 blob starting with
/// the magic must NOT increment `pingwaves_received`. (This test does not
/// address the broader issue of pingwave authentication, which is a
/// separate protocol concern; it verifies that the length-only heuristic
/// no longer swallows legitimate Net-header-shaped traffic.)
#[tokio::test]
async fn test_regression_pingwave_not_dispatched_on_net_magic_packet() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    // Handshaked peer B so we have a registered source for the
    // "legitimate pingwave" sanity check below. Pingwaves from
    // unregistered sources are dropped at the dispatch boundary
    // (see `test_regression_pingwave_from_unregistered_source_is_dropped`),
    // so the sanity leg needs a real peer.
    let mk_config = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };
    let node_a = MeshNode::new(id_a, mk_config(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk_config(addr_b)).await.unwrap();
    let pub_b = *node_b.public_key();

    let (r1, r2) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    node_a.start();
    node_b.start();

    let before = node_a.proximity_graph().stats().pingwaves_received;

    // Leg 1: a 72-byte packet whose first two bytes are the Net magic
    // (0x4E45 little-endian = [0x45, 0x4E] = "EN"), sent from an
    // UNHANDSHAKED socket. Post-fix this must be rejected at the
    // pingwave parse guard and NOT reach the proximity graph. The
    // unregistered-source guard would also drop it, but the parse
    // guard fires first and is what this test is about.
    let attacker = UdpSocket::bind("127.0.0.1:0").await.unwrap();
    let mut pkt = [0u8; 72];
    pkt[0] = 0x45;
    pkt[1] = 0x4E;
    attacker.send_to(&pkt, addr_a).await.unwrap();

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

    let after = node_a.proximity_graph().stats().pingwaves_received;
    assert_eq!(
        before, after,
        "a 72-byte packet starting with the Net magic must not be \
         dispatched as a pingwave"
    );

    // Leg 2: legitimate pingwaves from B (a handshaked peer) must
    // still be accepted. B's heartbeat loop emits a pingwave every
    // `heartbeat_interval_ms` (500 ms by config); wait long enough
    // for at least one to cross the wire. A's dispatch must parse it
    // as a pingwave (the guard lets non-magic packets through) AND
    // accept it (B is a registered peer). If this counter doesn't
    // move, either the parse guard is too aggressive or the
    // unregistered-source guard is rejecting traffic from a real
    // peer.
    tokio::time::sleep(Duration::from_millis(1200)).await;

    let after_valid = node_a.proximity_graph().stats().pingwaves_received;
    assert!(
        after_valid > after,
        "legitimate pingwaves from a handshaked peer must be accepted; \
         before_valid_leg={}, after_valid_leg={}",
        after,
        after_valid
    );

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
}

// ============================================================================
// Channel fan-out (ChannelPublisher) — three-node tests
// ============================================================================

use net::adapter::net::{ChannelName, OnFailure, PublishConfig, Reliability};

/// Helper: build three nodes connected in a star (A as publisher, B and C
/// as subscribers). Returns the three nodes plus their node_ids.
async fn setup_publisher_with_two_subscribers() -> (MeshNode, MeshNode, MeshNode, u64, u64, u64) {
    let ports = find_ports(3).await;
    let psk = [0x42u8; 32];

    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let id_c = EntityKeypair::generate();

    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let nid_c = id_c.node_id();

    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();
    let addr_c: SocketAddr = format!("127.0.0.1:{}", ports[2]).parse().unwrap();

    let mk_config = |addr: SocketAddr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(2)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(10))
    };

    let node_a = MeshNode::new(id_a, mk_config(addr_a)).await.unwrap();
    let node_b = MeshNode::new(id_b, mk_config(addr_b)).await.unwrap();
    let node_c = MeshNode::new(id_c, mk_config(addr_c)).await.unwrap();

    let pub_b = *node_b.public_key();
    let pub_c = *node_c.public_key();

    // B subscribes to A → B accepts A, A connects B
    let (accept_result, connect_result) = tokio::join!(node_b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_b, &pub_b, nid_b).await
    });
    accept_result.expect("B accept A failed");
    connect_result.expect("A connect B failed");

    // Same for C
    let (accept_result, connect_result) = tokio::join!(node_c.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        node_a.connect(addr_c, &pub_c, nid_c).await
    });
    accept_result.expect("C accept A failed");
    connect_result.expect("A connect C failed");

    node_a.start();
    node_b.start();
    node_c.start();
    tokio::time::sleep(Duration::from_millis(100)).await;

    (node_a, node_b, node_c, nid_a, nid_b, nid_c)
}

/// `ChannelPublisher::publish` reaches every subscriber; the roster
/// evicts peers that go `Failed` so the next publish skips them.
#[tokio::test]
async fn test_channel_publisher_fanout_reaches_all_subscribers() {
    let (node_a, node_b, node_c, nid_a, _nid_b, _nid_c) =
        setup_publisher_with_two_subscribers().await;

    // B and C both subscribe to sensors/lidar on publisher A.
    let channel = ChannelName::new("sensors/lidar").unwrap();
    node_b
        .subscribe_channel(nid_a, channel.clone())
        .await
        .expect("B subscribe failed");
    node_c
        .subscribe_channel(nid_a, channel.clone())
        .await
        .expect("C subscribe failed");

    // Roster now has 2 subscribers.
    let ch_id = net::adapter::net::ChannelId::new(channel.clone());
    let members = node_a.roster().members(&ch_id);
    assert_eq!(members.len(), 2, "A's roster should have 2 subscribers");

    // Publish and verify the report.
    let publisher = node_a.channel_publisher(
        channel.clone(),
        PublishConfig::new()
            .with_reliability(Reliability::FireAndForget)
            .with_on_failure(OnFailure::Collect),
    );
    let payload = bytes::Bytes::from_static(b"lidar-scan-0");
    let report = node_a.publish(&publisher, payload).await.unwrap();

    assert_eq!(
        report.attempted, 2,
        "should have attempted both subscribers"
    );
    assert_eq!(report.delivered, 2, "both per-peer sends should succeed");
    assert!(report.errors.is_empty(), "no errors expected");

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

/// `Unsubscribe` removes the peer from the roster; the next publish
/// does not target it.
#[tokio::test]
async fn test_channel_publisher_unsubscribe_evicts_from_roster() {
    let (node_a, node_b, node_c, nid_a, nid_b, nid_c) =
        setup_publisher_with_two_subscribers().await;

    let channel = ChannelName::new("alerts").unwrap();
    node_b
        .subscribe_channel(nid_a, channel.clone())
        .await
        .unwrap();
    node_c
        .subscribe_channel(nid_a, channel.clone())
        .await
        .unwrap();

    let ch_id = net::adapter::net::ChannelId::new(channel.clone());
    assert_eq!(node_a.roster().members(&ch_id).len(), 2);

    // B unsubscribes; A's roster should drop B.
    node_b
        .unsubscribe_channel(nid_a, channel.clone())
        .await
        .unwrap();

    // Ack is processed asynchronously on A; poll briefly.
    let mut members = Vec::new();
    for _ in 0..20 {
        members = node_a.roster().members(&ch_id);
        if members.len() == 1 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert_eq!(
        members.len(),
        1,
        "after B unsubscribes, only C should remain"
    );
    assert_eq!(members[0], nid_c, "surviving subscriber should be C");

    // Publish goes only to C.
    let publisher = node_a.channel_publisher(
        channel.clone(),
        PublishConfig::new().with_on_failure(OnFailure::Collect),
    );
    let report = node_a
        .publish(&publisher, bytes::Bytes::from_static(b"only-c"))
        .await
        .unwrap();
    assert_eq!(report.attempted, 1);
    assert_eq!(report.delivered, 1);

    let _ = nid_a;
    let _ = nid_b;
    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

/// Empty roster → Ok with `attempted == 0`; not an error.
#[tokio::test]
async fn test_channel_publisher_empty_roster_is_ok() {
    let (node_a, node_b, node_c, _nid_a, _nid_b, _nid_c) =
        setup_publisher_with_two_subscribers().await;

    let channel = ChannelName::new("nobody/listens").unwrap();
    let publisher = node_a.channel_publisher(
        channel,
        PublishConfig::new().with_on_failure(OnFailure::Collect),
    );
    let report = node_a
        .publish(&publisher, bytes::Bytes::from_static(b"x"))
        .await
        .unwrap();
    assert_eq!(report.attempted, 0);
    assert_eq!(report.delivered, 0);
    assert!(report.errors.is_empty());
    assert!(report.is_empty());

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

/// `MembershipAck` flows: an Unsubscribe for an unregistered subscriber
/// is still accepted (idempotent), so the call returns Ok.
#[tokio::test]
async fn test_channel_publisher_unsubscribe_idempotent() {
    let (node_a, node_b, node_c, nid_a, _nid_b, _nid_c) =
        setup_publisher_with_two_subscribers().await;

    let channel = ChannelName::new("ghosts").unwrap();
    // B never subscribed; unsubscribe should still succeed.
    node_b
        .unsubscribe_channel(nid_a, channel.clone())
        .await
        .expect("unsubscribe must be idempotent");

    node_a.shutdown().await.unwrap();
    node_b.shutdown().await.unwrap();
    node_c.shutdown().await.unwrap();
}

// ============================================================================
// Stream backpressure (v1 local in-flight window)
// ============================================================================

use net::adapter::net::{StreamConfig, StreamError};

/// Concurrent callers on the same stream with a window-1 cap: at least
/// one caller per concurrent burst hits `StreamError::Backpressure` and
/// `backpressure_events` increments. Multi-threaded runtime is required
/// to force real concurrency between the acquire-send-release sequence
/// across the window boundary (localhost UDP sends are too fast on a
/// single-threaded runtime for the race to materialize).
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_send_on_stream_backpressure_when_concurrent() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(4)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = Arc::new(MeshNode::new(id_a, mk(addr_a)).await.unwrap());
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *b.public_key();

    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    a.start();
    b.start();

    // 128-byte window (v2 wire-bytes semantics). Each packet on the
    // wire = 64 B Net header + 16 B AEAD tag + payload = 80 B of
    // fixed overhead + 14 B payload (4 B frame length + 10 B event)
    // = 94 B per packet. One packet fits (94 of 128); the second
    // races for only 34 B of credit and sees Backpressure before
    // the receiver's grant arrives.
    let stream = a
        .open_stream(nid_b, 7777, StreamConfig::new().with_window_bytes(128))
        .unwrap();

    // Spin up a batch of concurrent tasks sending on the same stream.
    // Each event is a small payload.
    let n_tasks: usize = 16;
    let event = Bytes::from_static(b"{\"t\":\"bp\"}");
    let mut handles = Vec::new();
    for _ in 0..n_tasks {
        let mesh = Arc::clone(&a);
        let stream = stream.clone();
        let payload = event.clone();
        handles.push(tokio::spawn(async move {
            mesh.send_on_stream(&stream, &[payload]).await
        }));
    }

    let mut ok = 0usize;
    let mut backpressure = 0usize;
    let mut transport = 0usize;
    for h in handles {
        match h.await.unwrap() {
            Ok(()) => ok += 1,
            Err(StreamError::Backpressure) => backpressure += 1,
            Err(StreamError::Transport(_)) => transport += 1,
            Err(StreamError::NotConnected) => panic!("unexpected NotConnected"),
        }
    }
    // At least one caller must have hit the cap. We don't assert an
    // exact count — the scheduler's interleaving depends on the
    // runtime — but with a window of 1 and 16 concurrent senders,
    // backpressure is virtually guaranteed.
    assert!(
        backpressure > 0,
        "expected at least one Backpressure in {} concurrent sends; got ok={}, bp={}, transport={}",
        n_tasks,
        ok,
        backpressure,
        transport
    );
    assert!(ok > 0, "expected at least one successful send");

    // Stats mirror the backpressure counter we just observed.
    let stats = a.stream_stats(nid_b, 7777).expect("stream stats");
    assert!(
        stats.backpressure_events >= backpressure as u64,
        "stats.backpressure_events ({}) must be >= observed ({})",
        stats.backpressure_events,
        backpressure
    );
    assert_eq!(stats.tx_window, 128);

    Arc::try_unwrap(a).ok().unwrap().shutdown().await.unwrap();
    b.shutdown().await.unwrap();
}

/// `send_with_retry` backs off on Backpressure and eventually succeeds
/// once the window clears. Transport errors are returned immediately,
/// not retried.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_send_with_retry_eventually_succeeds_through_backpressure() {
    let ports = find_ports(2).await;
    let psk = [0x42u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(4)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = Arc::new(MeshNode::new(id_a, mk(addr_a)).await.unwrap());
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *b.public_key();

    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    a.start();
    b.start();

    // v2 wire-bytes window: each packet is 80 B overhead + small
    // payload ≈ 95 B. A 512-byte window fits ~5 packets in flight,
    // small enough that retries hit Backpressure, large enough that
    // receiver-side grants flowing back replenish credit for
    // subsequent attempts.
    let stream = a
        .open_stream(nid_b, 8888, StreamConfig::new().with_window_bytes(512))
        .unwrap();

    // Run 32 concurrent retry-sends. All must eventually succeed;
    // none should surface as Backpressure at the caller because
    // send_with_retry absorbs the transient pressure.
    let n: usize = 32;
    let mut handles = Vec::new();
    for i in 0..n {
        let mesh = Arc::clone(&a);
        let stream = stream.clone();
        let payload = Bytes::from(format!(r#"{{"i":{}}}"#, i));
        handles.push(tokio::spawn(async move {
            mesh.send_with_retry(&stream, &[payload], 64).await
        }));
    }
    for h in handles {
        h.await
            .unwrap()
            .expect("send_with_retry must eventually succeed");
    }

    // After the storm, credit must still be available — no leaks.
    let stats = a.stream_stats(nid_b, 8888).expect("stream stats");
    assert!(stats.tx_credit_remaining > 0);

    Arc::try_unwrap(a).ok().unwrap().shutdown().await.unwrap();
    b.shutdown().await.unwrap();
}

/// v2 flagship regression: a **single serial sender** with a small
/// window outruns the receiver's grant cadence and surfaces
/// `StreamError::Backpressure` — the exact case v1 (local in-flight
/// counter) could not catch.
///
/// In v1 this test would have surfaced as `Transport(io::Error)` once
/// the kernel send buffer saturated; v2's byte-credit window exhausts
/// deterministically and returns the clean Backpressure variant. The
/// `credit_grants_received` counter confirms the loop is active (if
/// it stayed at zero the sender would just be stalled, not
/// backpressure-aware).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_v2_serial_sender_sees_backpressure_on_slow_receiver() {
    let ports = find_ports(2).await;
    let psk = [0x13u8; 32];
    let id_a = EntityKeypair::generate();
    let id_b = EntityKeypair::generate();
    let nid_a = id_a.node_id();
    let nid_b = id_b.node_id();
    let addr_a: SocketAddr = format!("127.0.0.1:{}", ports[0]).parse().unwrap();
    let addr_b: SocketAddr = format!("127.0.0.1:{}", ports[1]).parse().unwrap();

    let mk = |addr| {
        MeshNodeConfig::new(addr, psk)
            .with_num_shards(4)
            .with_handshake(3, Duration::from_secs(3))
            .with_heartbeat_interval(Duration::from_millis(500))
            .with_session_timeout(Duration::from_secs(30))
    };

    let a = Arc::new(MeshNode::new(id_a, mk(addr_a)).await.unwrap());
    let b = MeshNode::new(id_b, mk(addr_b)).await.unwrap();
    let pub_b = *b.public_key();

    let (r1, r2) = tokio::join!(b.accept(nid_a), async {
        tokio::time::sleep(Duration::from_millis(50)).await;
        a.connect(addr_b, &pub_b, nid_b).await
    });
    r1.unwrap();
    r2.unwrap();
    a.start();
    b.start();

    // 256-byte wire-bytes window. Each packet on the wire is 80 B
    // overhead + 14 B payload = 94 B, so two packets drain to 68 B
    // remaining — the third serial send needs 94 B more credit than
    // it has and has to wait for a grant from B. Without a grant it
    // Backpressures; the critical thing v2 does that v1 couldn't.
    let stream = a
        .open_stream(nid_b, 9999, StreamConfig::new().with_window_bytes(256))
        .unwrap();

    let event = Bytes::from_static(b"{\"k\":\"v\"}"); // 10 bytes, frame = 14
    let mut ok = 0usize;
    let mut backpressure = 0usize;

    // Rip through 64 serial sends with NO intervening await beyond
    // what send_on_stream itself does. The first 2 succeed, the rest
    // race the grant RTT and trip Backpressure. v1 would eventually
    // surface Transport(io::Error) when the kernel buffer filled; v2
    // delivers clean Backpressure on the in-protocol signal.
    for _ in 0..64 {
        match a
            .send_on_stream(&stream, std::slice::from_ref(&event))
            .await
        {
            Ok(()) => ok += 1,
            Err(StreamError::Backpressure) => backpressure += 1,
            Err(StreamError::Transport(_)) => panic!(
                "v2 must not surface kernel-buffer-full as Transport; \
                 credit exhaustion should always present as Backpressure"
            ),
            Err(StreamError::NotConnected) => panic!("unexpected NotConnected"),
        }
    }

    assert!(
        backpressure > 0,
        "serial sender must hit Backpressure once credit drains faster \
         than grants arrive; got ok={}, bp={}",
        ok,
        backpressure,
    );
    assert!(ok > 0, "at least the initial handful must succeed");

    // Let any in-flight grants settle so the stats snapshot reflects
    // both sides of the loop.
    tokio::time::sleep(Duration::from_millis(200)).await;

    let stats = a.stream_stats(nid_b, 9999).expect("stats");
    assert_eq!(stats.tx_window, 256);
    assert!(
        stats.backpressure_events >= backpressure as u64,
        "stats.backpressure_events ({}) should be >= observed ({})",
        stats.backpressure_events,
        backpressure,
    );
    assert!(
        stats.credit_grants_received > 0,
        "at least one StreamWindow grant must have flowed back from B — \
         otherwise the v2 loop is not actually active"
    );

    Arc::try_unwrap(a).ok().unwrap().shutdown().await.unwrap();
    b.shutdown().await.unwrap();
}