    use crate::node::session_wire::fsp_prepend_inner_header;

    #[test]
    fn application_receive_refreshes_previous_hop_peer_without_direct_source_trust() {
        use crate::PeerIdentity;
        use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
        use crate::node::retry::RetryState;
        use crate::peer::{ActivePeer, ActivePeerSession};
        use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
        use crate::utils::index::SessionIndex;

        let local = Identity::generate();
        let source = Identity::generate();
        let previous_hop = Identity::generate();
        let previous_hop_identity = PeerIdentity::from_pubkey_full(previous_hop.pubkey_full());
        let previous_hop_addr = *previous_hop_identity.node_addr();
        let source_addr = *source.node_addr();
        let previous_hop_config = PeerConfig {
            npub: previous_hop.npub(),
            alias: None,
            addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
            connect_policy: ConnectPolicy::AutoConnect,
            auto_reconnect: true,
            discovery_fallback_transit: true,
        };

        let mut config = crate::config::Config::new();
        config.peers.push(previous_hop_config.clone());
        let mut node = Node::with_identity(local, config).expect("node");
        node.config.node.heartbeat_interval_secs = 10;

        let stale_seen_ms = Node::now_ms().saturating_sub(11_000);
        let mut active_peer = ActivePeer::with_session(
            previous_hop_identity,
            LinkId::new(9),
            1_000,
            ActivePeerSession {
                session: make_xk_session(&node.identity, &previous_hop),
                our_index: SessionIndex::new(0x1010),
                their_index: SessionIndex::new(0x2020),
                transport_id: TransportId::new(0x55),
                current_addr: TransportAddr::from_string("203.0.113.9:2121"),
                link_stats: LinkStats::new(),
                is_initiator: true,
                remote_epoch: None,
            },
        );
        active_peer.touch(stale_seen_ms);
        node.peers
            .insert_with_current_session_index(previous_hop_addr, active_peer);

        let session = SessionEntry::new(
            source_addr,
            source.pubkey_full(),
            EndToEndState::Established(make_xk_session(&node.identity, &source)),
            1_000,
            true,
        );
        node.sessions.insert(source_addr, session);

        let mut retry = RetryState::new(previous_hop_config);
        retry.reconnect = true;
        node.retry_pending.insert(previous_hop_addr, retry);

        SessionDispatchCommit {
            source_addr,
            receive_completion: Some(SessionReceiveCompletion {
                source_addr,
                previous_hop_addr,
                direct_path: false,
            }),
        }
        .finish_receive(&mut node);

        let previous_hop_peer = node
            .peers
            .get(&previous_hop_addr)
            .expect("previous hop should remain active");
        assert!(
            previous_hop_peer.idle_time(Node::now_ms()) <= 1_000,
            "accepted application data should refresh the direct previous-hop link"
        );
        assert!(
            !node.retry_pending.contains_key(&previous_hop_addr),
            "fresh authenticated data from the direct previous hop should stop link refresh churn"
        );
        assert!(node.sessions.get(&source_addr).is_some());
    }

    #[test]
    fn authenticated_fmp_receive_clears_direct_probe_retry_on_direct_path() {
        use crate::PeerIdentity;
        use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
        use crate::node::retry::RetryState;
        use crate::peer::{ActivePeer, ActivePeerSession};
        use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
        use crate::utils::index::SessionIndex;

        let local = Identity::generate();
        let peer = Identity::generate();
        let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let peer_addr = *peer_identity.node_addr();
        let peer_config = PeerConfig {
            npub: peer.npub(),
            alias: None,
            addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
            connect_policy: ConnectPolicy::AutoConnect,
            auto_reconnect: true,
            discovery_fallback_transit: true,
        };
        let transport_id = TransportId::new(0x56);
        let transport_addr = TransportAddr::from_string("198.51.100.20:61062");

        let mut config = crate::config::Config::new();
        config.peers.push(peer_config.clone());
        let mut node = Node::with_identity(local, config).expect("node");
        node.config.node.heartbeat_interval_secs = 10;

        let mut active_peer = ActivePeer::with_session(
            peer_identity,
            LinkId::new(9),
            1_000,
            ActivePeerSession {
                session: make_xk_session(&node.identity, &peer),
                our_index: SessionIndex::new(0x1011),
                their_index: SessionIndex::new(0x2021),
                transport_id,
                current_addr: transport_addr.clone(),
                link_stats: LinkStats::new(),
                is_initiator: true,
                remote_epoch: None,
            },
        );
        active_peer.touch(Node::now_ms().saturating_sub(11_000));
        node.peers
            .insert_with_current_session_index(peer_addr, active_peer);

        let mut retry = RetryState::new(peer_config);
        retry.reconnect = true;
        node.retry_pending.insert(peer_addr, retry);

        node.record_authenticated_fmp_receive_facts(
            crate::node::AuthenticatedFmpReceiveFacts {
                source_peer: peer_identity,
                transport_id,
                remote_addr: &transport_addr,
                packet_timestamp_ms: Node::now_ms(),
                packet_len: 256,
                fmp_counter: 11,
                inner_timestamp_ms: 22,
                fmp_flags: 0,
            },
            Some(&peer_addr),
        );

        assert!(
            !node.retry_pending.contains_key(&peer_addr),
            "fresh authenticated FMP return on the direct peer path should stop direct-probe churn"
        );
    }

    #[test]
    fn authenticated_fmp_receive_keeps_direct_probe_retry_for_forwarded_path() {
        use crate::PeerIdentity;
        use crate::config::{ConnectPolicy, PeerAddress, PeerConfig};
        use crate::node::retry::RetryState;
        use crate::peer::{ActivePeer, ActivePeerSession};
        use crate::transport::{LinkId, LinkStats, TransportAddr, TransportId};
        use crate::utils::index::SessionIndex;

        let local = Identity::generate();
        let peer = Identity::generate();
        let relay = Identity::generate();
        let peer_identity = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let peer_addr = *peer_identity.node_addr();
        let relay_addr = *relay.node_addr();
        let peer_config = PeerConfig {
            npub: peer.npub(),
            alias: None,
            addresses: vec![PeerAddress::with_priority("udp", "203.0.113.9:2121", 1)],
            connect_policy: ConnectPolicy::AutoConnect,
            auto_reconnect: true,
            discovery_fallback_transit: true,
        };
        let transport_id = TransportId::new(0x57);
        let transport_addr = TransportAddr::from_string("198.51.100.20:61062");

        let mut config = crate::config::Config::new();
        config.peers.push(peer_config.clone());
        let mut node = Node::with_identity(local, config).expect("node");
        node.config.node.heartbeat_interval_secs = 10;

        let mut active_peer = ActivePeer::with_session(
            peer_identity,
            LinkId::new(9),
            1_000,
            ActivePeerSession {
                session: make_xk_session(&node.identity, &peer),
                our_index: SessionIndex::new(0x1012),
                their_index: SessionIndex::new(0x2022),
                transport_id,
                current_addr: transport_addr.clone(),
                link_stats: LinkStats::new(),
                is_initiator: true,
                remote_epoch: None,
            },
        );
        active_peer.touch(Node::now_ms().saturating_sub(11_000));
        node.peers
            .insert_with_current_session_index(peer_addr, active_peer);

        let mut retry = RetryState::new(peer_config);
        retry.reconnect = true;
        node.retry_pending.insert(peer_addr, retry);

        node.record_authenticated_fmp_receive_facts(
            crate::node::AuthenticatedFmpReceiveFacts {
                source_peer: peer_identity,
                transport_id,
                remote_addr: &transport_addr,
                packet_timestamp_ms: Node::now_ms(),
                packet_len: 256,
                fmp_counter: 11,
                inner_timestamp_ms: 22,
                fmp_flags: 0,
            },
            Some(&relay_addr),
        );

        assert!(
            node.retry_pending.contains_key(&peer_addr),
            "forwarded authenticated FMP traffic must not prove the direct peer path is healthy"
        );
    }

    #[test]
    fn authenticated_session_message_owns_endpoint_delivery_conversion() {
        let peer = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let endpoint_payload = b"endpoint delivery".to_vec();
        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::EndpointData.to_byte(),
            0,
            &endpoint_payload,
        );

        let message = AuthenticatedSessionMessage::new(
            source_peer,
            crate::transport::PacketBuffer::new(plaintext),
            SessionMessageType::EndpointData.to_byte(),
        );

        assert_eq!(message.body(), endpoint_payload);
        let deliveries = message.into_endpoint_data_deliveries();
        assert_eq!(deliveries.len(), 1);
        let delivery = &deliveries[0];
        assert_eq!(delivery.source_peer, source_peer);
        assert_eq!(delivery.payload.as_slice(), endpoint_payload.as_slice());
    }

    #[test]
    fn authenticated_session_dispatch_owns_route_ce_and_completion_facts() {
        let peer = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let source_addr = *peer.node_addr();
        let previous_hop_addr = node_addr(0x55);
        let endpoint_payload = b"endpoint completion".to_vec();
        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::EndpointData.to_byte(),
            0,
            &endpoint_payload,
        );
        let dispatch = AuthenticatedSessionDispatch::new(
            source_addr,
            previous_hop_addr,
            true,
            AuthenticatedSessionMessage::new(
                source_peer,
                crate::transport::PacketBuffer::new(plaintext),
                SessionMessageType::EndpointData.to_byte(),
            ),
        );

        assert_eq!(dispatch.source_addr(), &source_addr);
        assert_eq!(dispatch.previous_hop_addr(), &previous_hop_addr);
        assert!(dispatch.ce_flag());
        assert_eq!(
            dispatch.msg_type(),
            SessionMessageType::EndpointData.to_byte()
        );
        assert_eq!(dispatch.body(), endpoint_payload);
        assert_eq!(
            dispatch.receive_completion(),
            Some(SessionReceiveCompletion {
                source_addr,
                previous_hop_addr,
                direct_path: false,
            })
        );
        let deliveries = dispatch.into_endpoint_data_deliveries();
        assert_eq!(deliveries.len(), 1);
        let delivery = &deliveries[0];
        assert_eq!(delivery.source_peer, source_peer);
        assert_eq!(delivery.payload.as_slice(), endpoint_payload.as_slice());

        let report_plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::SenderReport.to_byte(),
            0,
            b"report",
        );
        let report_dispatch = AuthenticatedSessionDispatch::new(
            source_addr,
            previous_hop_addr,
            false,
            AuthenticatedSessionMessage::new(
                source_peer,
                crate::transport::PacketBuffer::new(report_plaintext),
                SessionMessageType::SenderReport.to_byte(),
            ),
        );
        assert_eq!(
            report_dispatch.receive_completion(),
            None,
            "MMP reports must not reset session idle"
        );
    }

    #[test]
    fn endpoint_data_batched_dispatch_finishes_receive_without_pending_flush() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let source_addr = *peer.node_addr();
        let previous_hop_addr = node_addr(0x55);
        let endpoint_payload = b"fast endpoint delivery".to_vec();
        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::EndpointData.to_byte(),
            0,
            &endpoint_payload,
        );
        let dispatch = AuthenticatedSessionDispatch::new(
            source_addr,
            previous_hop_addr,
            false,
            AuthenticatedSessionMessage::new(
                source_peer,
                crate::transport::PacketBuffer::new(plaintext),
                SessionMessageType::EndpointData.to_byte(),
            ),
        );

        let mut node = Node::new(crate::config::Config::new()).expect("node");
        let mut endpoint_io = node
            .attach_endpoint_data_io(8)
            .expect("endpoint I/O should attach");
        node.sessions
            .insert(source_addr, established_entry(&local, &peer));

        let mut commit = SessionReceiveBatchCommit::default();
        let deliveries = dispatch.dispatch_endpoint_data_batched(&mut node, &mut commit);
        let pending_flush = commit.finish(&mut node);
        node.deliver_endpoint_data_batch(deliveries);
        assert!(pending_flush.is_empty());
        let crate::node::NodeEndpointEvent { messages, .. } =
            endpoint_io.event_rx.try_recv().expect("endpoint event");
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].source_peer, source_peer);
        assert_eq!(messages[0].payload.as_slice(), endpoint_payload.as_slice());
        assert!(node.sessions.get(&source_addr).is_some());
        assert!(
            !node.pending_session_traffic.has_traffic_for(&source_addr),
            "empty pending guard should keep the fast path synchronous"
        );
    }

    #[test]
    fn endpoint_data_batched_dispatch_reports_pending_flush_owner() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let source_addr = *peer.node_addr();
        let previous_hop_addr = node_addr(0x55);
        let endpoint_payload = b"fast endpoint pending".to_vec();
        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::EndpointData.to_byte(),
            0,
            &endpoint_payload,
        );
        let dispatch = AuthenticatedSessionDispatch::new(
            source_addr,
            previous_hop_addr,
            false,
            AuthenticatedSessionMessage::new(
                source_peer,
                crate::transport::PacketBuffer::new(plaintext),
                SessionMessageType::EndpointData.to_byte(),
            ),
        );

        let mut node = Node::new(crate::config::Config::new()).expect("node");
        let _endpoint_io = node
            .attach_endpoint_data_io(8)
            .expect("endpoint I/O should attach");
        node.sessions
            .insert(source_addr, established_entry(&local, &peer));
        assert!(
            !node
                .pending_session_traffic
                .push_endpoint_data_batch_with_enqueued_at_ms(
                    source_addr,
                    vec![crate::node::EndpointDataPayload::from_packet_payload(vec![0xaa])
                        .expect("test endpoint payload")],
                    8,
                    8,
                    1_000,
                )
                .destination_dropped()
        );

        let mut commit = SessionReceiveBatchCommit::default();
        let _delivery = dispatch.dispatch_endpoint_data_batched(&mut node, &mut commit);
        let pending_flush = commit.finish(&mut node);

        assert_eq!(pending_flush, vec![source_addr]);
        assert!(
            node.pending_session_traffic.has_traffic_for(&source_addr),
            "batched dispatch should report, not synchronously drain, pending traffic"
        );
    }

    #[tokio::test]
    async fn ipv6_shim_batched_dispatch_queues_tun_packets_and_reports_pending_flush() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let source_addr = *peer.node_addr();
        let previous_hop_addr = node_addr(0x55);
        let mut node = Node::with_identity(local, crate::config::Config::new()).expect("node");
        let (tun_tx, tun_rx) = crate::upper::tun::write_channel();
        node.tun_tx = Some(tun_tx);

        assert!(
            !node
                .pending_session_traffic
                .push_tun_packet(source_addr, vec![0xaa], 8, 8)
                .destination_dropped()
        );

        let mut ipv6 = Vec::new();
        ipv6.extend_from_slice(&[0x60, 0, 0, 0]);
        ipv6.extend_from_slice(&4u16.to_be_bytes());
        ipv6.push(59);
        ipv6.push(64);
        ipv6.extend_from_slice(
            &crate::FipsAddress::from_node_addr(&source_addr)
                .to_ipv6()
                .octets(),
        );
        ipv6.extend_from_slice(
            &crate::FipsAddress::from_node_addr(node.node_addr())
                .to_ipv6()
                .octets(),
        );
        ipv6.extend_from_slice(&[1, 2, 3, 4]);
        let expected_ipv6 = ipv6.clone();
        assert!(crate::upper::ipv6_shim::compress_ipv6_with_port_header_in_place(
            &mut ipv6,
            crate::node::session_wire::FSP_PORT_IPV6_SHIM,
            crate::node::session_wire::FSP_PORT_IPV6_SHIM,
        ));

        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::DataPacket.to_byte(),
            0,
            &ipv6,
        );
        let dispatch = AuthenticatedSessionDispatch::new(
            source_addr,
            previous_hop_addr,
            false,
            AuthenticatedSessionMessage::new(
                source_peer,
                crate::transport::PacketBuffer::new(plaintext),
                SessionMessageType::DataPacket.to_byte(),
            ),
        );

        assert!(dispatch.is_ipv6_shim_data_packet());
        let mut packets = Vec::new();
        let mut commit = SessionReceiveBatchCommit::default();
        dispatch.dispatch_ipv6_shim_batched(&mut node, &mut packets, &mut commit);
        assert_eq!(packets.len(), 1);
        assert!(!commit.is_empty());

        node.flush_dataplane_tun_session_batch(&mut packets, &mut commit)
            .await;

        assert!(packets.is_empty());
        assert!(commit.is_empty());
        let packet = tun_rx
            .try_recv_packet()
            .expect("batched shim packet should be queued to TUN");
        assert_eq!(packet.as_slice(), expected_ipv6.as_slice());
        assert!(
            node.pending_session_traffic.has_traffic_for(&source_addr),
            "batched TUN dispatch should report pending flush; without a dataplane owner the pending packet remains queued"
        );
    }
