    #[cfg(unix)]
    #[test]
    fn session_datagram_runtime_route_owns_next_hop_path_mtu_and_bookkeeping() {
        use crate::PeerIdentity;
        use crate::config::RoutingMode;
        use crate::peer::ActivePeer;
        use crate::transport::udp::UdpTransport;
        use crate::transport::{LinkId, TransportAddr, TransportHandle, TransportId};
        use crate::utils::index::SessionIndex;

        let local = Identity::generate();
        let dest = Identity::generate();
        let transit = Identity::generate();
        let transit_identity = PeerIdentity::from_pubkey_full(transit.pubkey_full());
        let dest_addr = *dest.node_addr();
        let transit_addr = *transit_identity.node_addr();
        let transport_id = TransportId::new(0x57);

        let mut config = crate::config::Config::new();
        config.node.routing.mode = RoutingMode::ReplyLearned;
        let mut node = Node::with_identity(local, config).expect("node");

        let mut session = established_entry(&node.identity, &dest);
        session.mark_established(0x1000);
        session.init_mmp(&node.config.node.session_mmp);
        assert_eq!(
            session.mmp().expect("session mmp").path_mtu.current_mtu(),
            u16::MAX
        );
        assert!(node.sessions.insert(dest_addr, session).is_none());

        let active_peer = ActivePeer::with_session(
            transit_identity,
            LinkId::new(9),
            1_000,
            make_xk_session(&node.identity, &transit),
            SessionIndex::new(0x1010),
            SessionIndex::new(0x2020),
            transport_id,
            TransportAddr::from_string("127.0.0.1:9"),
            crate::transport::LinkStats::new(),
            true,
            &node.config.node.mmp,
            Some([0x02; 8]),
        );
        node.peers
            .insert_with_current_session_index(transit_addr, active_peer);
        let (packet_tx, _packet_rx) = crate::transport::packet_channel(8);
        let udp = UdpTransport::new(
            transport_id,
            None,
            crate::config::UdpConfig {
                bind_addr: Some("127.0.0.1:0".to_string()),
                mtu: Some(1234),
                ..Default::default()
            },
            packet_tx,
        );
        assert!(
            node.transports
                .insert(transport_id, TransportHandle::Udp(udp))
                .is_none()
        );
        node.learn_reverse_route(dest_addr, transit_addr);

        let mut datagram = SessionDatagram::new(
            *node.node_addr(),
            dest_addr,
            vec![SessionMessageType::DataPacket.to_byte(), 0, 0, 0],
        )
        .with_ttl(9);
        let route = node
            .resolve_session_datagram_runtime_route(&mut datagram)
            .expect("learned transit route should resolve");

        assert_eq!(route.dest_addr(), dest_addr);
        assert_eq!(route.next_hop_addr(), transit_addr);
        assert_eq!(route.path_mtu(), 1234);
        assert!(
            route.source_mmp_seeded(),
            "route owner should seed the session source-side MMP path MTU"
        );
        assert_eq!(
            datagram.path_mtu, 1234,
            "route owner should min-fold the outgoing transport MTU into the datagram"
        );
        assert_eq!(
            node.sessions
                .get(&dest_addr)
                .and_then(|entry| entry.mmp())
                .expect("session mmp")
                .path_mtu
                .current_mtu(),
            1234
        );

        let originated_before = node.stats().forwarding.originated_packets;
        let originated_bytes_before = node.stats().forwarding.originated_bytes;
        let encoded_len = datagram.encode().len();
        route.record_success(&mut node, encoded_len);
        let session = node.sessions.get(&dest_addr).expect("session exists");
        assert_eq!(
            session.last_outbound_next_hop(),
            Some(transit_addr),
            "route owner should record the successful outbound next hop"
        );
        assert_eq!(
            node.stats().forwarding.originated_packets,
            originated_before + 1
        );
        assert_eq!(
            node.stats().forwarding.originated_bytes,
            originated_bytes_before + encoded_len as u64
        );

        let route = node
            .resolve_session_datagram_runtime_route(&mut datagram)
            .expect("learned transit route should still resolve");
        route.record_failure(&mut node);
        let snapshot = node.learned_route_table_snapshot(Node::now_ms());
        let learned = snapshot
            .destinations
            .iter()
            .find(|dest| dest.destination == dest_addr.to_string())
            .and_then(|dest| {
                dest.routes
                    .iter()
                    .find(|route| route.next_hop == transit_addr.to_string())
            })
            .expect("learned transit route should remain visible");
        assert_eq!(
            learned.failures, 1,
            "route owner should record send failure against the selected learned next hop"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn endpoint_bulk_send_lease_dispatches_feedback_before_worker_batch() {
        use crate::PeerIdentity;
        use crate::peer::ActivePeer;
        use crate::transport::udp::UdpTransport;
        use crate::transport::{LinkId, TransportAddr, TransportId, packet_channel};
        use crate::utils::index::SessionIndex;

        let local = Identity::generate();
        let dest = Identity::generate();
        let next_hop = Identity::generate();
        let dest_identity = PeerIdentity::from_pubkey_full(dest.pubkey_full());
        let next_hop_identity = PeerIdentity::from_pubkey_full(next_hop.pubkey_full());
        let dest_addr = *dest_identity.node_addr();
        let next_hop_addr = *next_hop_identity.node_addr();
        let transport_id = TransportId::new(0x5A);

        let mut node = Node::with_identity(local, crate::config::Config::new()).expect("node");
        let mut session = established_entry(&node.identity, &dest);
        session.mark_established(1_000);
        session.init_mmp(&node.config.node.session_mmp);
        assert!(node.sessions.insert(dest_addr, session).is_none());

        let active_peer = ActivePeer::with_session(
            next_hop_identity,
            LinkId::new(9),
            1_000,
            make_xk_session(&node.identity, &next_hop),
            SessionIndex::new(0x1010),
            SessionIndex::new(0x2020),
            transport_id,
            TransportAddr::from_string("127.0.0.1:9"),
            crate::transport::LinkStats::new(),
            true,
            &node.config.node.mmp,
            Some([0x02; 8]),
        );
        node.peers
            .insert_with_current_session_index(next_hop_addr, active_peer);

        let (packet_tx, _packet_rx) = packet_channel(8);
        let mut udp = UdpTransport::new(
            transport_id,
            None,
            crate::config::UdpConfig {
                bind_addr: Some("127.0.0.1:0".to_string()),
                mtu: Some(1234),
                ..Default::default()
            },
            packet_tx,
        );
        udp.start_async().await.expect("start UDP transport");
        let prepared = crate::node::FmpSendPreparation {
            their_index: SessionIndex::new(0x2020),
            transport_id,
            remote_addr: TransportAddr::from_string("127.0.0.1:9"),
            #[cfg(any(target_os = "linux", target_os = "macos"))]
            connected_socket: None,
            timestamp_ms: 0,
            flags: 0,
            payload_len: 0,
        };
        let send_target = PipelinedEndpointSendTarget::resolve(&udp, &prepared)
            .await
            .expect("started UDP transport resolves send target")
            .into_selected_send_target();
        let workers = crate::node::encrypt_worker::EncryptWorkerPool::spawn(1);
        let (runtime, mut feedback_rx) = crate::node::EndpointBulkSendRuntime::channel(8);
        let fsp = node
            .sessions
            .get(&dest_addr)
            .and_then(|entry| entry.endpoint_bulk_fsp_lease())
            .expect("established session should export FSP lease state");
        let fmp = node
            .peers
            .endpoint_bulk_fmp_lease(&next_hop_addr)
            .expect("active peer should export FMP lease state");
        runtime.publish(crate::node::EndpointBulkSendLease::new(
            *node.node_addr(),
            dest_addr,
            next_hop_addr,
            1234,
            9,
            3,
            false,
            fsp,
            fmp,
            send_target,
            workers,
            std::time::Duration::from_secs(1),
        ));

        let payload = EndpointDataPayload::new(vec![0xee; 96]);
        assert!(
            runtime.try_send_bulk_batch_to_peer(dest_identity, std::slice::from_ref(&payload)),
            "published lease should dispatch the bulk batch"
        );
        let feedback = feedback_rx
            .try_recv()
            .expect("endpoint mover must enqueue feedback before worker dispatch");
        assert_eq!(feedback.records.len(), 1);
        assert_eq!(feedback.records[0].dest_addr, dest_addr);
        assert_eq!(feedback.records[0].next_hop_addr, next_hop_addr);
        let crate::node::EndpointBulkSendSessionBookkeeping::Fsp { path_mtu, .. } =
            feedback.records[0].session_bookkeeping;
        {
            assert_eq!(path_mtu, 1234);
        }

        node.apply_endpoint_bulk_send_feedback(feedback);
        let session = node.sessions.get(&dest_addr).expect("session exists");
        let (packets_sent, _, bytes_sent, _) = session.traffic_counters();
        assert_eq!(packets_sent, 1);
        assert_eq!(bytes_sent, payload.len() as u64);
        assert_eq!(session.last_outbound_next_hop(), Some(next_hop_addr));
        assert_eq!(
            session
                .mmp()
                .expect("session MMP")
                .path_mtu
                .current_mtu(),
            1234
        );
        let peer = node.peers.get(&next_hop_addr).expect("peer exists");
        assert_eq!(peer.link_stats().packets_sent, 1);
        assert_eq!(node.stats().forwarding.originated_packets, 1);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn pipelined_endpoint_peer_runtime_send_owns_transport_path_mtu_route_plan_and_runtime_dispatch()
     {
        use crate::PeerIdentity;
        use crate::peer::ActivePeer;
        use crate::transport::udp::UdpTransport;
        use crate::transport::{LinkId, TransportAddr, TransportId, packet_channel};
        use crate::utils::index::SessionIndex;
        use std::collections::HashMap;

        let local = Identity::generate();
        let peer = Identity::generate();
        let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let dest_addr = *peer_identity.node_addr();
        let source_addr = node_addr(0x10);
        let transport_id = TransportId::new(0x55);
        let fallback_addr: std::net::SocketAddr = "127.0.0.1:9".parse().unwrap();

        let mut sessions = crate::node::SessionRegistry::default();
        assert!(
            sessions
                .insert(dest_addr, established_entry(&local, &peer))
                .is_none()
        );

        let mut peers = crate::node::PeerLifecycleRegistry::default();
        let active_peer = ActivePeer::with_session(
            peer_identity,
            LinkId::new(9),
            1_000,
            make_xk_session(&local, &peer),
            SessionIndex::new(0x1010),
            SessionIndex::new(0x2020),
            transport_id,
            TransportAddr::from_string(&fallback_addr.to_string()),
            crate::transport::LinkStats::new(),
            true,
            &crate::mmp::MmpConfig::default(),
            Some([0x02; 8]),
        );
        peers.insert_with_current_session_index(dest_addr, active_peer);

        let (packet_tx, _packet_rx) = packet_channel(8);
        let mut udp = UdpTransport::new(
            transport_id,
            None,
            crate::config::UdpConfig {
                bind_addr: Some("127.0.0.1:0".to_string()),
                mtu: Some(1234),
                ..Default::default()
            },
            packet_tx,
        );
        udp.start_async().await.expect("start UDP transport");

        let mut transports = HashMap::new();
        assert!(
            transports
                .insert(transport_id, crate::transport::TransportHandle::Udp(udp))
                .is_none()
        );

        let payload = EndpointDataPayload::new(vec![0xee; 64]);
        let inner_plaintext = vec![0xaa; 80];
        let send = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &payload,
            now_ms: 0x1122_3344,
            timestamp: 0x5566_7788,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&inner_plaintext),
            my_coords: None,
            dest_coords: None,
        };

        let route_snapshot = peers
            .prepare_peer_runtime_route_snapshot(&dest_addr)
            .expect("active peer should prepare route snapshot");
        let runtime_route =
            PipelinedEndpointPeerRuntimeRoute::new(source_addr, route_snapshot, 9, 7, false);

        let fsp_before = sessions
            .get(&dest_addr)
            .expect("session exists")
            .send_counter();
        let fmp_before = peers
            .get(&dest_addr)
            .and_then(|peer| peer.noise_session())
            .expect("active peer session exists")
            .current_send_counter();

        let dispatch = PipelinedEndpointPeerRuntimeSend::new(runtime_route, send)
            .resolve_dispatch(&transports, &mut sessions, &mut peers)
            .await
            .expect("peer runtime send owner should build runtime plan and dispatch")
            .expect("established peer runtime send should dispatch");

        assert_eq!(dispatch.dest_addr(), dest_addr);
        assert_eq!(dispatch.next_hop_addr(), dest_addr);
        assert_eq!(
            dispatch.fsp_reservation_input().path_mtu,
            1234,
            "peer runtime send owner should derive path MTU from the selected transport"
        );
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists")
                .send_counter(),
            fsp_before + 1,
            "peer runtime send owner should consume exactly one FSP counter"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists")
                .current_send_counter(),
            fmp_before + 1,
            "peer runtime send owner should consume exactly one FMP counter"
        );

        let prepared = dispatch.into_prepared_send(None);
        assert_eq!(prepared.dest_addr, dest_addr);
        assert_eq!(prepared.next_hop_addr, dest_addr);
        assert_eq!(prepared.fsp_bookkeeping.counter, fsp_before);
        assert_eq!(prepared.fmp_counter, fmp_before);

        let missing_transport_send = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &payload,
            now_ms: 0x1122_3344,
            timestamp: 0x5566_7788,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&inner_plaintext),
            my_coords: None,
            dest_coords: None,
        };
        let missing_transport_snapshot = crate::node::PeerRuntimeRouteSnapshot::new(
            dest_addr,
            SessionIndex::new(0x2020),
            TransportId::new(0x99),
            TransportAddr::from_string(&fallback_addr.to_string()),
            #[cfg(any(target_os = "linux", target_os = "macos"))]
            None,
            0x0102_0304,
            0,
            true,
        );
        let missing_transport_route = PipelinedEndpointPeerRuntimeRoute::new(
            source_addr,
            missing_transport_snapshot,
            9,
            7,
            false,
        );

        assert!(matches!(
            PipelinedEndpointPeerRuntimeSend::new(
                missing_transport_route,
                missing_transport_send,
            )
            .resolve_dispatch(&transports, &mut sessions, &mut peers)
            .await,
            Err(PipelinedEndpointPeerRuntimeSendError::RuntimeSend(
                PipelinedEndpointRuntimeSendError::TransportNotFound(id),
            )) if id == TransportId::new(0x99)
        ));
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists after missing transport")
                .send_counter(),
            fsp_before + 1,
            "missing transport must fail before consuming another FSP counter"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists after missing transport")
                .current_send_counter(),
            fmp_before + 1,
            "missing transport must fail before consuming another FMP counter"
        );

        let batched_route_snapshot = peers
            .prepare_peer_runtime_route_snapshot(&dest_addr)
            .expect("active peer should prepare route snapshot for batch target");
        let batched_route =
            PipelinedEndpointPeerRuntimeRoute::new(source_addr, batched_route_snapshot, 9, 7, false);
        let batch_target = batched_route
            .batch_target(&transports)
            .await
            .expect("batch target should resolve")
            .expect("UDP route should provide a reusable batch target");
        assert_eq!(
            batch_target.path_mtu, 1234,
            "batch target should cache the selected transport path MTU"
        );

        let batched_payload = EndpointDataPayload::new(vec![0xdd; 96]);
        let batched_inner_plaintext = vec![0xbb; 112];
        let batched_send = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &batched_payload,
            now_ms: 0x2233_4455,
            timestamp: 0x6677_8899,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&batched_inner_plaintext),
            my_coords: None,
            dest_coords: None,
        };
        let fsp_before_batched = sessions
            .get(&dest_addr)
            .expect("session exists before batched send")
            .send_counter();
        let fmp_before_batched = peers
            .get(&dest_addr)
            .and_then(|peer| peer.noise_session())
            .expect("active peer session exists before batched send")
            .current_send_counter();

        let batched_dispatch = PipelinedEndpointPeerRuntimeSend::resolve_dispatch_with_batch_target(
            &batched_route,
            batched_send,
            &batch_target,
            &mut sessions,
            &mut peers,
        )
        .expect("batch target dispatch should reserve")
        .expect("established batched peer runtime send should dispatch");

        assert_eq!(batched_dispatch.dest_addr(), dest_addr);
        assert_eq!(batched_dispatch.next_hop_addr(), dest_addr);
        assert_eq!(
            batched_dispatch.fsp_reservation_input().path_mtu,
            1234,
            "batched dispatch should reuse the cached transport path MTU"
        );
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists after batched send")
                .send_counter(),
            fsp_before_batched + 1,
            "batch target path should still consume exactly one FSP counter"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists after batched send")
                .current_send_counter(),
            fmp_before_batched + 1,
            "batch target path should still consume exactly one FMP counter"
        );

        let batched_prepared = batched_dispatch.into_prepared_send(None);
        assert_eq!(batched_prepared.dest_addr, dest_addr);
        assert_eq!(batched_prepared.next_hop_addr, dest_addr);
        assert_eq!(
            batched_prepared.fsp_bookkeeping.counter,
            fsp_before_batched
        );
        assert_eq!(batched_prepared.fmp_counter, fmp_before_batched);
        assert_eq!(batched_prepared.worker_job.counter, fmp_before_batched);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn pipelined_endpoint_runtime_send_owns_transport_target_and_reservation_handoff() {
        use crate::PeerIdentity;
        use crate::peer::ActivePeer;
        use crate::transport::udp::UdpTransport;
        use crate::transport::{LinkId, TransportAddr, TransportId, packet_channel};
        use crate::utils::index::SessionIndex;
        use std::collections::HashMap;

        let local = Identity::generate();
        let peer = Identity::generate();
        let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let dest_addr = *peer_identity.node_addr();
        let source_addr = node_addr(0x10);
        let transport_id = TransportId::new(0x55);
        let fallback_addr: std::net::SocketAddr = "127.0.0.1:9".parse().unwrap();

        let mut sessions = crate::node::SessionRegistry::default();
        assert!(
            sessions
                .insert(dest_addr, established_entry(&local, &peer))
                .is_none()
        );

        let mut peers = crate::node::PeerLifecycleRegistry::default();
        let active_peer = ActivePeer::with_session(
            peer_identity,
            LinkId::new(9),
            1_000,
            make_xk_session(&local, &peer),
            SessionIndex::new(0x1010),
            SessionIndex::new(0x2020),
            transport_id,
            TransportAddr::from_string(&fallback_addr.to_string()),
            crate::transport::LinkStats::new(),
            true,
            &crate::mmp::MmpConfig::default(),
            Some([0x02; 8]),
        );
        peers.insert_with_current_session_index(dest_addr, active_peer);

        let (packet_tx, _packet_rx) = packet_channel(8);
        let mut udp = UdpTransport::new(
            transport_id,
            None,
            crate::config::UdpConfig {
                bind_addr: Some("127.0.0.1:0".to_string()),
                ..Default::default()
            },
            packet_tx,
        );
        udp.start_async().await.expect("start UDP transport");

        let mut transports = HashMap::new();
        assert!(
            transports
                .insert(transport_id, crate::transport::TransportHandle::Udp(udp))
                .is_none()
        );

        let payload = EndpointDataPayload::new(vec![0xee; 64]);
        let inner_plaintext = vec![0xaa; 80];
        let send = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &payload,
            now_ms: 0x1122_3344,
            timestamp: 0x5566_7788,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&inner_plaintext),
            my_coords: None,
            dest_coords: None,
        };

        let route_snapshot = peers
            .prepare_peer_runtime_route_snapshot(&dest_addr)
            .expect("active peer should prepare route snapshot");
        let transport = transports
            .get(&transport_id)
            .expect("transport should exist for runtime plan");
        let runtime =
            PipelinedEndpointPeerRuntimeRoute::new(source_addr, route_snapshot, 9, 7, false)
                .into_runtime_send_plan(&send, transport)
                .expect("runtime route should build send plan");

        let fsp_before = sessions
            .get(&dest_addr)
            .expect("session exists")
            .send_counter();
        let fmp_before = peers
            .get(&dest_addr)
            .and_then(|peer| peer.noise_session())
            .expect("active peer session exists")
            .current_send_counter();

        let dispatch = PipelinedEndpointRuntimeSend::new(runtime)
            .resolve_dispatch(&transports, &mut sessions, &mut peers)
            .await
            .expect("runtime send owner should resolve transport and reserve")
            .expect("established runtime send should dispatch");

        assert_eq!(dispatch.dest_addr(), dest_addr);
        assert_eq!(dispatch.next_hop_addr(), dest_addr);
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists")
                .send_counter(),
            fsp_before + 1,
            "runtime send owner should consume exactly one FSP counter"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists")
                .current_send_counter(),
            fmp_before + 1,
            "runtime send owner should consume exactly one FMP counter"
        );

        let prepared = dispatch.into_prepared_send(None);
        assert_eq!(prepared.dest_addr, dest_addr);
        assert_eq!(prepared.next_hop_addr, dest_addr);
        assert_eq!(prepared.fsp_bookkeeping.counter, fsp_before);
        assert_eq!(prepared.fmp_counter, fmp_before);

        let missing_transport_snapshot = crate::node::PeerRuntimeRouteSnapshot::new(
            dest_addr,
            SessionIndex::new(0x2020),
            TransportId::new(0x99),
            TransportAddr::from_string(&fallback_addr.to_string()),
            #[cfg(any(target_os = "linux", target_os = "macos"))]
            None,
            0x0102_0304,
            0,
            true,
        );
        let missing_transport_route =
            PipelinedEndpointRoutePlan::new(source_addr, dest_addr, 1234, 9, 7, false);
        let missing_transport_plan = missing_transport_route
            .build_send_plan(&send)
            .expect("missing-transport send plan should build");
        let missing_transport_runtime = PipelinedEndpointRuntimeSendPlan::from_peer_route_snapshot(
            missing_transport_route,
            missing_transport_plan,
            missing_transport_snapshot,
        )
        .expect("missing-transport runtime should still build send plan");

        assert!(matches!(
            PipelinedEndpointRuntimeSend::new(missing_transport_runtime)
                .resolve_dispatch(&transports, &mut sessions, &mut peers)
                .await,
            Err(PipelinedEndpointRuntimeSendError::TransportNotFound(id))
                if id == TransportId::new(0x99)
        ));
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists after missing transport")
                .send_counter(),
            fsp_before + 1,
            "missing transport must fail before consuming another FSP counter"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists after missing transport")
                .current_send_counter(),
            fmp_before + 1,
            "missing transport must fail before consuming another FMP counter"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn pipelined_endpoint_runtime_send_attempt_owns_target_and_reservations() {
        use crate::PeerIdentity;
        use crate::peer::ActivePeer;
        use crate::transport::udp::UdpTransport;
        use crate::transport::{LinkId, TransportAddr, TransportId, packet_channel};
        use crate::utils::index::SessionIndex;

        let local = Identity::generate();
        let peer = Identity::generate();
        let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let dest_addr = *peer_identity.node_addr();
        let source_addr = node_addr(0x10);
        let transport_id = TransportId::new(0x55);
        let fallback_addr: std::net::SocketAddr = "127.0.0.1:9".parse().unwrap();

        let mut sessions = crate::node::SessionRegistry::default();
        assert!(
            sessions
                .insert(dest_addr, established_entry(&local, &peer))
                .is_none()
        );

        let mut peers = crate::node::PeerLifecycleRegistry::default();
        let active_peer = ActivePeer::with_session(
            peer_identity,
            LinkId::new(9),
            1_000,
            make_xk_session(&local, &peer),
            SessionIndex::new(0x1010),
            SessionIndex::new(0x2020),
            transport_id,
            TransportAddr::from_string(&fallback_addr.to_string()),
            crate::transport::LinkStats::new(),
            true,
            &crate::mmp::MmpConfig::default(),
            Some([0x02; 8]),
        );
        peers.insert_with_current_session_index(dest_addr, active_peer);

        let (packet_tx, _packet_rx) = packet_channel(8);
        let mut udp = UdpTransport::new(
            transport_id,
            None,
            crate::config::UdpConfig {
                bind_addr: Some("127.0.0.1:0".to_string()),
                ..Default::default()
            },
            packet_tx,
        );
        udp.start_async().await.expect("start UDP transport");
        let transport = TransportHandle::Udp(udp);
        let TransportHandle::Udp(udp) = &transport else {
            unreachable!("test transport is UDP");
        };

        let payload = EndpointDataPayload::new(vec![0xee; 64]);
        let inner_plaintext = vec![0xaa; 80];
        let send = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &payload,
            now_ms: 0x1122_3344,
            timestamp: 0x5566_7788,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&inner_plaintext),
            my_coords: None,
            dest_coords: None,
        };

        let route_snapshot = peers
            .prepare_peer_runtime_route_snapshot(&dest_addr)
            .expect("active peer should prepare route snapshot");
        let runtime =
            PipelinedEndpointPeerRuntimeRoute::new(source_addr, route_snapshot, 9, 7, false)
                .into_runtime_send_plan(&send, &transport)
                .expect("runtime route should build send plan");
        let send_target = runtime
            .resolve_send_target(udp)
            .await
            .expect("started UDP transport resolves send target");

        let fsp_before = sessions
            .get(&dest_addr)
            .expect("session exists")
            .send_counter();
        let fmp_before = peers
            .get(&dest_addr)
            .and_then(|peer| peer.noise_session())
            .expect("active peer session exists")
            .current_send_counter();

        let send_target_for_batch = send_target.clone();
        let dispatch = PipelinedEndpointRuntimeSendAttempt::new(runtime, send_target)
            .reserve(&mut sessions, &mut peers)
            .expect("runtime send attempt should reserve from both registries")
            .expect("established runtime send attempt should dispatch");

        assert_eq!(dispatch.dest_addr(), dest_addr);
        assert_eq!(dispatch.next_hop_addr(), dest_addr);
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists")
                .send_counter(),
            fsp_before + 1,
            "attempt should consume exactly one FSP counter"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists")
                .current_send_counter(),
            fmp_before + 1,
            "attempt should consume exactly one FMP counter"
        );

        let prepared = dispatch.into_prepared_send(None);
        assert_eq!(prepared.dest_addr, dest_addr);
        assert_eq!(prepared.next_hop_addr, dest_addr);
        assert_eq!(prepared.fsp_bookkeeping.counter, fsp_before);
        assert_eq!(prepared.fmp_counter, fmp_before);
        assert_eq!(prepared.worker_job.counter, fmp_before);

        let batch_payload_a = EndpointDataPayload::new(vec![0x11; 96]);
        let batch_payload_b = EndpointDataPayload::new(vec![0x22; 48]);
        let batch_inner_a = vec![0xa1; 112];
        let batch_inner_b = vec![0xb2; 64];
        let batch_send_a = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &batch_payload_a,
            now_ms: 0x1122_3345,
            timestamp: 0x5566_7789,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&batch_inner_a),
            my_coords: None,
            dest_coords: None,
        };
        let batch_send_b = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &batch_payload_b,
            now_ms: 0x1122_3346,
            timestamp: 0x5566_7790,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&batch_inner_b),
            my_coords: None,
            dest_coords: None,
        };
        let route_snapshot = peers
            .prepare_peer_runtime_route_snapshot(&dest_addr)
            .expect("active peer should prepare route snapshot for batch");
        let batch_target = PipelinedEndpointBatchTarget {
            send_target: send_target_for_batch,
            path_mtu: route_snapshot.path_mtu(&transport),
        };
        let batch_route =
            PipelinedEndpointPeerRuntimeRoute::new(source_addr, route_snapshot, 9, 7, false);
        let batch_fsp_before = sessions
            .get(&dest_addr)
            .expect("session exists before batch")
            .send_counter();
        let batch_fmp_before = peers
            .get(&dest_addr)
            .and_then(|peer| peer.noise_session())
            .expect("active peer session exists before batch")
            .current_send_counter();

        let prepared_batch =
            PipelinedEndpointPeerRuntimeBatchSend::resolve_prepared_sends_with_batch_target(
                &batch_route,
                [batch_send_a, batch_send_b],
                &batch_target,
                &mut sessions,
                &mut peers,
            )
            .expect("batch send should reserve from both registries")
            .expect("available worker batch should dispatch");
        assert_eq!(prepared_batch.len(), 2);
        assert_eq!(prepared_batch[0].fsp_bookkeeping.counter, batch_fsp_before);
        assert_eq!(
            prepared_batch[1].fsp_bookkeeping.counter,
            batch_fsp_before + 1
        );
        assert_eq!(prepared_batch[0].fmp_counter, batch_fmp_before);
        assert_eq!(prepared_batch[1].fmp_counter, batch_fmp_before + 1);
        assert_eq!(prepared_batch[0].worker_job.counter, batch_fmp_before);
        assert_eq!(prepared_batch[1].worker_job.counter, batch_fmp_before + 1);
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists after batch")
                .send_counter(),
            batch_fsp_before + 2,
            "batch should consume exactly one FSP counter per packet"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists after batch")
                .current_send_counter(),
            batch_fmp_before + 2,
            "batch should consume exactly one FMP counter per packet"
        );

        let blocked_snapshot = crate::node::PeerRuntimeRouteSnapshot::new(
            dest_addr,
            SessionIndex::new(0x2020),
            transport_id,
            TransportAddr::from_string(&fallback_addr.to_string()),
            #[cfg(any(target_os = "linux", target_os = "macos"))]
            None,
            0x0102_0304,
            0,
            false,
        );
        let blocked_runtime =
            PipelinedEndpointPeerRuntimeRoute::new(source_addr, blocked_snapshot, 9, 7, false)
                .into_runtime_send_plan(&send, &transport)
                .expect("blocked worker runtime should still build send plan");
        let blocked_target = blocked_runtime
            .resolve_send_target(udp)
            .await
            .expect("started UDP transport resolves blocked send target");
        let blocked_attempt_fsp_before = sessions
            .get(&dest_addr)
            .expect("session exists before blocked attempt")
            .send_counter();
        let blocked_attempt_fmp_before = peers
            .get(&dest_addr)
            .and_then(|peer| peer.noise_session())
            .expect("active peer session exists before blocked attempt")
            .current_send_counter();

        assert!(
            PipelinedEndpointRuntimeSendAttempt::new(blocked_runtime, blocked_target)
                .reserve(&mut sessions, &mut peers)
                .expect("unavailable worker is a recoverable no-dispatch result")
                .is_none()
        );
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists after blocked attempt")
                .send_counter(),
            blocked_attempt_fsp_before,
            "blocked attempt must not consume another FSP counter"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists after blocked attempt")
                .current_send_counter(),
            blocked_attempt_fmp_before,
            "blocked attempt must not consume another FMP counter"
        );

        let blocked_batch_fsp_before = sessions
            .get(&dest_addr)
            .expect("session exists before blocked batch")
            .send_counter();
        let blocked_batch_fmp_before = peers
            .get(&dest_addr)
            .and_then(|peer| peer.noise_session())
            .expect("active peer session exists before blocked batch")
            .current_send_counter();
        let blocked_snapshot = crate::node::PeerRuntimeRouteSnapshot::new(
            dest_addr,
            SessionIndex::new(0x2020),
            transport_id,
            TransportAddr::from_string(&fallback_addr.to_string()),
            #[cfg(any(target_os = "linux", target_os = "macos"))]
            None,
            0x0102_0304,
            0,
            false,
        );
        let blocked_batch_route =
            PipelinedEndpointPeerRuntimeRoute::new(source_addr, blocked_snapshot, 9, 7, false);
        assert!(
            PipelinedEndpointPeerRuntimeBatchSend::resolve_prepared_sends_with_batch_target(
                &blocked_batch_route,
                [batch_send_a],
                &batch_target,
                &mut sessions,
                &mut peers,
            )
            .expect("unavailable worker batch is a recoverable no-dispatch result")
            .is_none()
        );
        assert_eq!(
            sessions
                .get(&dest_addr)
                .expect("session still exists after blocked batch")
                .send_counter(),
            blocked_batch_fsp_before,
            "blocked batch must not consume FSP counters"
        );
        assert_eq!(
            peers
                .get(&dest_addr)
                .and_then(|peer| peer.noise_session())
                .expect("active peer session still exists after blocked batch")
                .current_send_counter(),
            blocked_batch_fmp_before,
            "blocked batch must not consume FMP counters"
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn pipelined_endpoint_runtime_dispatch_owns_target_reservations_and_prepared_send() {
        use crate::node::wire::{FLAG_SP, build_established_header};
        use crate::node::{PreparedFmpWorkerReservation, session::FspSendReservation};
        use crate::transport::udp::UdpTransport;
        use crate::transport::{TransportAddr, TransportId, packet_channel};
        use crate::utils::index::SessionIndex;
        use ring::aead::{LessSafeKey, UnboundKey};

        fn test_cipher(byte: u8) -> LessSafeKey {
            let unbound =
                UnboundKey::new(&ring::aead::CHACHA20_POLY1305, &[byte; 32]).expect("test key");
            LessSafeKey::new(unbound)
        }

        let source_addr = node_addr(0x10);
        let dest_addr = node_addr(0x20);
        let next_hop_addr = node_addr(0x30);
        let payload = EndpointDataPayload::new(vec![0xee; 64]);
        let inner_plaintext = vec![0xaa; 80];
        let send = PipelinedEndpointSend {
            dest_addr: &dest_addr,
            payload: &payload,
            now_ms: 0x1122_3344,
            timestamp: 0x5566_7788,
            fsp_flags: 0,
            body: PipelinedEndpointWireBody::InnerPlaintext(&inner_plaintext),
            my_coords: None,
            dest_coords: None,
        };
        let route = PipelinedEndpointRoutePlan::new(source_addr, next_hop_addr, 1234, 9, 7, false);
        let plan = route
            .build_send_plan(&send)
            .expect("route plan should build send plan");
        let expected_originated_bytes = plan.link_plaintext_len() + crate::noise::TAG_SIZE;
        let expected_fsp_reservation = plan.fsp_reservation_input();

        let transport_id = TransportId::new(0x55);
        let (packet_tx, _packet_rx) = packet_channel(8);
        let mut udp = UdpTransport::new(
            transport_id,
            None,
            crate::config::UdpConfig {
                bind_addr: Some("127.0.0.1:0".to_string()),
                ..Default::default()
            },
            packet_tx,
        );
        udp.start_async().await.expect("start UDP transport");
        let fallback_addr: std::net::SocketAddr = "127.0.0.1:9".parse().unwrap();
        let fmp_prepared = crate::node::FmpSendPreparation {
            their_index: SessionIndex::new(0xA0B0_C0D0),
            transport_id,
            remote_addr: TransportAddr::from_string(&fallback_addr.to_string()),
            #[cfg(any(target_os = "linux", target_os = "macos"))]
            connected_socket: None,
            timestamp_ms: 0x0102_0304,
            flags: FLAG_SP,
            payload_len: plan.fmp_payload_len(),
        };
        let snapshot = crate::node::PeerRuntimeSendSnapshot::new(next_hop_addr, fmp_prepared, true);
        let runtime = PipelinedEndpointRuntimeSendPlan::from_parts(route, plan, snapshot)
            .expect("matching route/send/FMP preparation should form runtime plan");

        let send_target = runtime
            .resolve_send_target(&udp)
            .await
            .expect("started UDP transport resolves send target");
        let fmp_counter = 0x1112_1314_1516_1718;
        let fsp_counter = 0x0102_0304_0506_0708;
        let fmp_header = build_established_header(
            runtime.fmp_prepared().their_index,
            fmp_counter,
            runtime.fmp_prepared().flags,
            runtime.fmp_payload_len(),
        );
        let fsp_header = build_fsp_header(
            fsp_counter,
            send.fsp_flags,
            expected_fsp_reservation.payload_len,
        );
        let fmp_reservation = PreparedFmpWorkerReservation {
            counter: fmp_counter,
            header: fmp_header,
            cipher: test_cipher(7),
            predicted_bytes: ESTABLISHED_HEADER_SIZE
                + runtime.fmp_payload_len() as usize
                + crate::noise::TAG_SIZE,
        };
        let fsp_reservation = FspSendReservation {
            counter: fsp_counter,
            header: fsp_header,
            cipher: test_cipher(8),
        };

        let dispatch = PipelinedEndpointRuntimeSendDispatch::new(
            runtime,
            send_target,
            fmp_reservation,
            fsp_reservation,
        );
        assert_eq!(dispatch.dest_addr(), dest_addr);
        assert_eq!(dispatch.next_hop_addr(), next_hop_addr);
        assert_eq!(dispatch.fsp_reservation_input(), expected_fsp_reservation);

        let prepared = dispatch.into_prepared_send(None);
        assert_eq!(prepared.dest_addr, dest_addr);
        assert_eq!(prepared.next_hop_addr, next_hop_addr);
        assert_eq!(prepared.fmp_counter, fmp_counter);
        assert_eq!(prepared.fmp_timestamp_ms, 0x0102_0304);
        assert_eq!(prepared.originated_bytes, expected_originated_bytes);
        assert_eq!(prepared.fsp_bookkeeping.counter, fsp_counter);
        assert_eq!(prepared.fsp_bookkeeping.next_hop, Some(next_hop_addr));
        assert_eq!(prepared.worker_job.counter, fmp_counter);
        assert!(prepared.worker_job.bulk_endpoint_data);
        assert!(!prepared.worker_job.drop_on_backpressure);
        assert_eq!(prepared.worker_job.scheduling_weight, 7);
        assert!(prepared.worker_job.queued_at.is_none());
        assert_eq!(
            &prepared.worker_job.wire_buf[..ESTABLISHED_HEADER_SIZE],
            &fmp_header
        );
    }
