    #[test]
    fn session_runtime_receive_owns_fsp_open_bookkeeping_and_dispatch_metadata() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let (local_session, mut peer_sender) = make_xk_session_pair(&local, &peer);
        let mut entry = SessionEntry::new(
            *peer.node_addr(),
            peer.pubkey_full(),
            EndToEndState::Established(local_session),
            1_000,
            true,
        );
        entry.mark_established(1_000);
        entry.touch_inbound_frame(1_000);
        entry.record_decrypt_failure();

        let endpoint_payload = b"endpoint runtime receive".to_vec();
        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::EndpointData.to_byte(),
            0,
            &endpoint_payload,
        );
        let counter = peer_sender.current_send_counter();
        let header = build_fsp_header(counter, 0, plaintext.len() as u16);
        let ciphertext = peer_sender
            .encrypt_with_aad(&plaintext, &header)
            .expect("test frame should encrypt");
        let mut wire = header.to_vec();
        wire.extend_from_slice(&ciphertext);
        let parsed = FspEncryptedHeader::parse(&wire).expect("test frame should parse");

        let outcome = SessionRuntimeReceive::new(
            &mut entry,
            &parsed,
            &wire[FSP_HEADER_SIZE..],
            1_280,
            true,
            2_000,
        )
        .open_established();

        match outcome {
            FspFrameOutcome::Authentic(message) => {
                assert_eq!(message.source_peer().node_addr(), peer.node_addr());
                assert_eq!(message.plaintext(), plaintext);
                assert_eq!(
                    message.msg_type(),
                    SessionMessageType::EndpointData.to_byte()
                );
                assert_eq!(message.inner_flags_byte(), 0);
                assert_eq!(message.timestamp(), 0x0102_0304);
                assert_eq!(message.body(), endpoint_payload);
                assert!(message.is_application_data());
            }
            other => panic!("expected authentic FSP frame, got {other:?}"),
        }
        assert_eq!(entry.consecutive_decrypt_failures(), 0);
        assert_eq!(entry.last_inbound_frame_ms(), 2_000);
    }

    #[test]
    fn fsp_receive_sync_requests_worker_refresh_only_on_epoch_promotion() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let mut entry = established_entry(&local, &peer);
        entry.mark_established(1_000);
        let initial_k_bit = entry.current_k_bit();
        entry.set_pending_session(make_xk_session(&local, &peer));

        let promoted = entry.apply_fsp_receive_sync_result(
            crate::node::session::FspReceiveSync {
                counter: 7,
                slot: EpochSlot::Pending,
                received_k_bit: !initial_k_bit,
                timestamp: 0x0102_0304,
                plaintext_len: FSP_INNER_HEADER_SIZE + 16,
                ce_flag: false,
                path_mtu: 1_280,
                spin_bit: false,
            },
            2_000,
            Instant::now(),
        );

        assert!(promoted.is_applied());
        assert!(
            promoted.refresh_worker_session(),
            "pending promotion changes FSP epoch topology and must refresh worker snapshots"
        );
        assert_eq!(entry.current_k_bit(), !initial_k_bit);
        assert!(entry.pending_new_session().is_none());
        assert!(entry.previous_highest_counter().is_some());

        let current = entry.apply_fsp_receive_sync_result(
            crate::node::session::FspReceiveSync {
                counter: 8,
                slot: EpochSlot::Current,
                received_k_bit: entry.current_k_bit(),
                timestamp: 0x0102_0305,
                plaintext_len: FSP_INNER_HEADER_SIZE + 16,
                ce_flag: false,
                path_mtu: 1_280,
                spin_bit: false,
            },
            2_100,
            Instant::now(),
        );

        assert!(current.is_applied());
        assert!(
            !current.refresh_worker_session(),
            "ordinary current-epoch replay mirroring must not re-register workers"
        );
    }

    #[test]
    fn established_fsp_wire_owns_ciphertext_offset_and_coord_warmup() {
        use crate::tree::TreeCoordinate;

        let source_addr = node_addr(0x01);
        let local_addr = node_addr(0x02);
        let root_addr = node_addr(0xf0);
        let source_coords = TreeCoordinate::from_addrs(vec![source_addr, root_addr]).unwrap();
        let local_coords = TreeCoordinate::from_addrs(vec![local_addr, root_addr]).unwrap();

        let header = build_fsp_header(42, FSP_FLAG_CP | FSP_FLAG_K, 20);
        let mut wire_bytes = header.to_vec();
        encode_coords(&source_coords, &mut wire_bytes);
        encode_coords(&local_coords, &mut wire_bytes);
        let ciphertext_offset = wire_bytes.len();
        wire_bytes.extend_from_slice(&[0xcc; 36]);

        let mut wire = EstablishedFspWire::parse(&wire_bytes, source_addr, local_addr)
            .expect("wire should parse with coord warmup");
        assert_eq!(wire.header.counter, 42);
        assert_eq!(wire.header.flags, FSP_FLAG_CP | FSP_FLAG_K);
        assert_eq!(wire.ciphertext, &wire_bytes[ciphertext_offset..]);
        assert!(wire.has_coord_warmup());

        let receive = wire.receive(1_280, true, 2_000);
        assert_eq!(receive.header.counter, 42);
        assert_eq!(receive.ciphertext, &wire_bytes[ciphertext_offset..]);
        assert_eq!(receive.path_mtu, 1_280);
        assert!(receive.ce_flag);
        assert_eq!(receive.now_ms, 2_000);

        let mut coord_cache = crate::cache::CoordCache::new(16, 1_000);
        wire.apply_coord_warmup(&mut coord_cache, 1_500);
        assert!(!wire.has_coord_warmup());
        assert_eq!(
            coord_cache
                .get(&source_addr, 1_500)
                .expect("source coords should be cached")
                .root_id(),
            &root_addr
        );
        assert_eq!(
            coord_cache
                .get(&local_addr, 1_500)
                .expect("local coords should be cached")
                .root_id(),
            &root_addr
        );

        assert!(matches!(
            EstablishedFspWire::parse(&wire_bytes[..FSP_HEADER_SIZE - 1], source_addr, local_addr),
            Err(EstablishedFspWireError::BadHeader)
        ));
        let mut truncated_coords = build_fsp_header(43, FSP_FLAG_CP, 20).to_vec();
        truncated_coords.extend_from_slice(&2u16.to_le_bytes());
        truncated_coords.extend_from_slice(&[0u8; 14]);
        assert!(matches!(
            EstablishedFspWire::parse(&truncated_coords, source_addr, local_addr),
            Err(EstablishedFspWireError::BadCoords(_))
        ));
    }

    #[test]
    fn session_registry_owns_early_encrypted_handshake_resend_budget() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let handshake = HandshakeState::new_xk_initiator(local.keypair(), peer.pubkey_full());
        let mut entry = SessionEntry::new(
            *peer.node_addr(),
            peer.pubkey_full(),
            EndToEndState::Initiating(handshake),
            1_000,
            true,
        );
        let payload = vec![0x10, 0x20, 0x30];
        entry.set_handshake_payload(payload.clone(), 1_500);

        let mut sessions = crate::node::SessionRegistry::default();
        sessions.insert(*peer.node_addr(), entry);

        match sessions.prepare_handshake_resend_after_early_encrypted_data(peer.node_addr(), 2) {
            EarlyEncryptedHandshakeResend::Resend { payload: resend } => {
                assert_eq!(resend, payload);
            }
            other => panic!("expected resend decision, got {other:?}"),
        }
        assert!(sessions.record_handshake_resend(peer.node_addr(), 2_000));
        let entry = sessions
            .get(peer.node_addr())
            .expect("session should remain");
        assert_eq!(entry.resend_count(), 1);
        assert_eq!(entry.next_resend_at_ms(), 2_000);
        assert_eq!(entry.handshake_payload(), Some(payload.as_slice()));

        assert!(matches!(
            sessions.prepare_handshake_resend_after_early_encrypted_data(peer.node_addr(), 1),
            EarlyEncryptedHandshakeResend::BudgetExhausted
        ));
        let entry = sessions
            .get(peer.node_addr())
            .expect("session should remain");
        assert!(entry.handshake_payload().is_none());
        assert_eq!(entry.next_resend_at_ms(), 0);
        assert_eq!(
            entry.resend_count(),
            1,
            "clearing an exhausted payload must not rewrite resend history"
        );

        let missing = node_addr(0x77);
        assert!(matches!(
            sessions.prepare_handshake_resend_after_early_encrypted_data(&missing, 2),
            EarlyEncryptedHandshakeResend::NoPayload
        ));
        assert!(!sessions.record_handshake_resend(&missing, 3_000));
    }

    #[test]
    fn session_registry_owns_established_fsp_open_lookup_and_bookkeeping() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let (local_session, mut peer_sender) = make_xk_session_pair(&local, &peer);
        let mut entry = SessionEntry::new(
            *peer.node_addr(),
            peer.pubkey_full(),
            EndToEndState::Established(local_session),
            1_000,
            true,
        );
        entry.mark_established(1_000);
        entry.record_decrypt_failure();

        let endpoint_payload = b"registry runtime receive".to_vec();
        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::EndpointData.to_byte(),
            0,
            &endpoint_payload,
        );
        let counter = peer_sender.current_send_counter();
        let header = build_fsp_header(counter, 0, plaintext.len() as u16);
        let ciphertext = peer_sender
            .encrypt_with_aad(&plaintext, &header)
            .expect("test frame should encrypt");
        let mut wire = header.to_vec();
        wire.extend_from_slice(&ciphertext);
        let parsed = FspEncryptedHeader::parse(&wire).expect("test frame should parse");

        let mut sessions = crate::node::SessionRegistry::default();
        sessions.insert(*peer.node_addr(), entry);

        let outcome = sessions.open_established_fsp_frame(
            peer.node_addr(),
            EstablishedFspReceive::new(&parsed, &wire[FSP_HEADER_SIZE..], 1_280, true, 2_000),
        );

        match outcome {
            FspFrameOutcome::Authentic(message) => {
                assert_eq!(message.source_peer().node_addr(), peer.node_addr());
                assert_eq!(
                    message.msg_type(),
                    SessionMessageType::EndpointData.to_byte()
                );
                assert_eq!(message.body(), endpoint_payload);
            }
            other => panic!("expected authentic FSP frame, got {other:?}"),
        }
        let entry = sessions
            .get(peer.node_addr())
            .expect("session should remain");
        assert_eq!(entry.consecutive_decrypt_failures(), 0);
        assert_eq!(entry.last_inbound_frame_ms(), 2_000);

        let missing = node_addr(0x77);
        let outcome = sessions.open_established_fsp_frame(
            &missing,
            EstablishedFspReceive::new(&parsed, &wire[FSP_HEADER_SIZE..], 1_280, false, 2_001),
        );
        assert!(matches!(outcome, FspFrameOutcome::UnknownSession));
    }

    #[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,
            plaintext,
            SessionMessageType::EndpointData.to_byte(),
            0,
            0x0102_0304,
        );

        assert_eq!(message.body(), endpoint_payload);
        let delivery = message.into_endpoint_data_delivery();
        assert_eq!(delivery.source_peer, source_peer);
        assert_eq!(delivery.payload, endpoint_payload);
    }

    #[test]
    fn authenticated_session_message_can_own_plaintext_inside_wire_buffer() {
        let peer = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let endpoint_payload = b"buffer endpoint delivery".to_vec();
        let plaintext = fsp_prepend_inner_header(
            0x0102_0304,
            SessionMessageType::EndpointData.to_byte(),
            0,
            &endpoint_payload,
        );
        let mut buffer = b"outer-fmp-prefix".to_vec();
        let plaintext_offset = buffer.len();
        buffer.extend_from_slice(&plaintext);
        buffer.extend_from_slice(b"outer-fmp-trailer");

        let message = AuthenticatedSessionMessage::from_buffer(
            source_peer,
            buffer,
            plaintext_offset,
            plaintext.len(),
            SessionMessageType::EndpointData.to_byte(),
            0,
            0x0102_0304,
        );

        assert_eq!(message.plaintext(), plaintext);
        assert_eq!(message.body(), endpoint_payload);
        let delivery = message.into_endpoint_data_delivery();
        assert_eq!(delivery.source_peer, source_peer);
        assert_eq!(delivery.payload, endpoint_payload);
    }

    #[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,
                plaintext,
                SessionMessageType::EndpointData.to_byte(),
                0,
                0x0102_0304,
            ),
        );

        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,
                body_len: endpoint_payload.len()
            })
        );
        let commit = dispatch.commit();
        assert_eq!(commit.source_addr(), &source_addr);
        assert_eq!(
            commit.receive_completion(),
            Some(SessionReceiveCompletion {
                source_addr,
                body_len: endpoint_payload.len()
            })
        );
        let local = Identity::generate();
        let mut sessions = crate::node::SessionRegistry::default();
        sessions.insert(source_addr, established_entry(&local, &peer));
        assert!(commit.record_receive(&mut sessions, 0x0bad_cafe));
        let entry = sessions.get(&source_addr).expect("session should remain");
        assert_eq!(
            entry.traffic_counters(),
            (0, 1, 0, endpoint_payload.len() as u64)
        );
        assert_eq!(entry.last_activity(), 0x0bad_cafe);

        let delivery = dispatch.into_endpoint_data_delivery();
        assert_eq!(delivery.source_peer, source_peer);
        assert_eq!(delivery.payload, endpoint_payload);

        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,
                report_plaintext,
                SessionMessageType::SenderReport.to_byte(),
                0,
                0x0102_0304,
            ),
        );
        assert_eq!(
            report_dispatch.receive_completion(),
            None,
            "MMP reports must not reset session idle/traffic counters"
        );
        let report_commit = report_dispatch.commit();
        assert_eq!(report_commit.source_addr(), &source_addr);
        assert_eq!(
            report_commit.receive_completion(),
            None,
            "MMP reports still flush pending packets without recording receive progress"
        );
        assert!(!report_commit.record_receive(&mut sessions, 0x0bad_f00d));
        let entry = sessions.get(&source_addr).expect("session should remain");
        assert_eq!(
            entry.traffic_counters(),
            (0, 1, 0, endpoint_payload.len() as u64)
        );
        assert_eq!(entry.last_activity(), 0x0bad_cafe);
    }

    #[test]
    fn endpoint_data_fast_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,
                plaintext,
                SessionMessageType::EndpointData.to_byte(),
                0,
                0x0102_0304,
            ),
        );

        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 finish = dispatch.dispatch_endpoint_data_fast(&mut node);
        assert_eq!(finish.pending_flush_dest(), None);
        match endpoint_io.event_rx.try_recv().expect("endpoint event") {
            crate::node::NodeEndpointEvent::Data {
                source_peer: delivered_source,
                payload,
                ..
            } => {
                assert_eq!(delivered_source, source_peer);
                assert_eq!(payload, endpoint_payload);
            }
            event => panic!("expected single endpoint data event, got {event:?}"),
        }
        let entry = node
            .sessions
            .get(&source_addr)
            .expect("session should remain");
        assert_eq!(
            entry.traffic_counters(),
            (0, 1, 0, endpoint_payload.len() as u64)
        );
        assert!(
            !node.pending_session_traffic.has_traffic_for(&source_addr),
            "empty pending guard should keep the fast path synchronous"
        );
    }

    #[test]
    fn endpoint_data_fast_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,
                plaintext,
                SessionMessageType::EndpointData.to_byte(),
                0,
                0x0102_0304,
            ),
        );

        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(source_addr, vec![0xaa], 8, 8)
                .destination_dropped()
        );

        let finish = dispatch.dispatch_endpoint_data_fast(&mut node);

        assert_eq!(finish.pending_flush_dest(), Some(source_addr));
        assert!(
            node.pending_session_traffic.has_traffic_for(&source_addr),
            "fast dispatch should report, not synchronously drain, pending traffic"
        );
    }

    #[tokio::test]
    async fn worker_direct_session_data_commits_before_endpoint_delivery() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let previous_hop = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let previous_hop_peer = PeerIdentity::from_pubkey_full(previous_hop.pubkey_full());
        let source_addr = *peer.node_addr();
        let endpoint_payload = b"worker decoded endpoint".to_vec();
        let plaintext_len = FSP_INNER_HEADER_SIZE + endpoint_payload.len();

        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 direct = DecryptDirectSessionData::for_test(
            crate::node::decrypt_worker::DecryptFmpBookkeeping {
                source_peer: previous_hop_peer,
                transport_id: crate::transport::TransportId::new(1),
                remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
                packet_timestamp_ms: 2_000,
                packet_len: 256,
                fmp_counter: 11,
                inner_timestamp_ms: 22,
                fmp_flags: 0,
            },
            source_addr,
            previous_hop_peer,
            false,
            crate::node::session::FspReceiveSync {
                counter: 7,
                slot: EpochSlot::Current,
                received_k_bit: false,
                timestamp: 0x0102_0304,
                plaintext_len,
                ce_flag: false,
                path_mtu: 1_280,
                spin_bit: false,
            },
            endpoint_payload.len(),
            DecryptDirectSessionDelivery::EndpointData(EndpointDataDelivery::new(
                source_peer,
                endpoint_payload.clone(),
            )),
        );

        node.process_direct_session_data_from_worker(direct).await;

        match endpoint_io.event_rx.try_recv().expect("endpoint event") {
            crate::node::NodeEndpointEvent::Data {
                source_peer: delivered_source,
                payload,
                ..
            } => {
                assert_eq!(delivered_source, source_peer);
                assert_eq!(payload, endpoint_payload);
            }
            event => panic!("expected worker-decoded endpoint data event, got {event:?}"),
        }
        let entry = node
            .sessions
            .get(&source_addr)
            .expect("session should remain");
        assert_eq!(
            entry.traffic_counters(),
            (0, 1, 0, endpoint_payload.len() as u64)
        );
        assert_eq!(entry.current_highest_counter(), Some(7));
    }

    #[tokio::test]
    async fn worker_direct_session_commit_updates_metadata_without_payload_bounce() {
        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 body_len = b"already delivered".len();
        let plaintext_len = FSP_INNER_HEADER_SIZE + body_len;

        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 commit = DecryptDirectSessionCommit::for_test(
            crate::node::decrypt_worker::DecryptFmpBookkeeping {
                source_peer,
                transport_id: crate::transport::TransportId::new(1),
                remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
                packet_timestamp_ms: 2_000,
                packet_len: 256,
                fmp_counter: 11,
                inner_timestamp_ms: 22,
                fmp_flags: 0,
            },
            source_addr,
            source_peer,
            false,
            crate::node::session::FspReceiveSync {
                counter: 7,
                slot: EpochSlot::Current,
                received_k_bit: false,
                timestamp: 0x0102_0304,
                plaintext_len,
                ce_flag: false,
                path_mtu: 1_280,
                spin_bit: false,
            },
            body_len,
            false,
        );

        node.process_direct_session_commit_from_worker(commit).await;

        assert!(
            endpoint_io.event_rx.try_recv().is_err(),
            "compact direct commit must not bounce payload bytes through rx_loop"
        );
        let entry = node
            .sessions
            .get(&source_addr)
            .expect("session should remain");
        assert_eq!(entry.traffic_counters(), (0, 1, 0, body_len as u64));
        assert_eq!(entry.current_highest_counter(), Some(7));
    }

    #[tokio::test]
    async fn worker_direct_session_commit_batch_flushes_pending_after_ordered_commits() {
        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 body_len = b"already delivered".len();
        let plaintext_len = FSP_INNER_HEADER_SIZE + body_len;

        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));
        assert!(
            !node
                .pending_session_traffic
                .push_endpoint_data(source_addr, vec![0xaa], 8, 8)
                .destination_dropped()
        );

        let commit = |counter| {
            DecryptDirectSessionCommit::for_test(
                crate::node::decrypt_worker::DecryptFmpBookkeeping {
                    source_peer,
                    transport_id: crate::transport::TransportId::new(1),
                    remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
                    packet_timestamp_ms: 2_000,
                    packet_len: 256,
                    fmp_counter: counter,
                    inner_timestamp_ms: 22,
                    fmp_flags: 0,
                },
                source_addr,
                source_peer,
                false,
                crate::node::session::FspReceiveSync {
                    counter,
                    slot: EpochSlot::Current,
                    received_k_bit: false,
                    timestamp: 0x0102_0304,
                    plaintext_len,
                    ce_flag: false,
                    path_mtu: 1_280,
                    spin_bit: false,
                },
                body_len,
                false,
            )
        };

        node.process_direct_session_commit_batch_from_worker(vec![commit(7), commit(8)])
            .await;

        assert!(
            endpoint_io.event_rx.try_recv().is_err(),
            "compact direct commit batch must not bounce payload bytes through rx_loop"
        );
        assert!(
            !node.pending_session_traffic.has_traffic_for(&source_addr),
            "batch boundary should flush pending session traffic once receive commits are applied"
        );
        let entry = node
            .sessions
            .get(&source_addr)
            .expect("session should remain");
        assert_eq!(entry.current_highest_counter(), Some(8));
    }

    #[tokio::test]
    async fn worker_direct_session_data_batch_delivers_endpoint_data_then_flushes_pending() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let previous_hop = Identity::generate();
        let source_peer = PeerIdentity::from_pubkey_full(peer.pubkey_full());
        let previous_hop_peer = PeerIdentity::from_pubkey_full(previous_hop.pubkey_full());
        let source_addr = *peer.node_addr();

        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));
        assert!(
            !node
                .pending_session_traffic
                .push_endpoint_data(source_addr, vec![0xbb], 8, 8)
                .destination_dropped()
        );

        let direct = |counter, payload: &[u8]| {
            DecryptDirectSessionData::for_test(
                crate::node::decrypt_worker::DecryptFmpBookkeeping {
                    source_peer: previous_hop_peer,
                    transport_id: crate::transport::TransportId::new(1),
                    remote_addr: crate::transport::TransportAddr::from_string("127.0.0.1:1234"),
                    packet_timestamp_ms: 2_000,
                    packet_len: 256,
                    fmp_counter: counter,
                    inner_timestamp_ms: 22,
                    fmp_flags: 0,
                },
                source_addr,
                previous_hop_peer,
                false,
                crate::node::session::FspReceiveSync {
                    counter,
                    slot: EpochSlot::Current,
                    received_k_bit: false,
                    timestamp: 0x0102_0304,
                    plaintext_len: FSP_INNER_HEADER_SIZE + payload.len(),
                    ce_flag: false,
                    path_mtu: 1_280,
                    spin_bit: false,
                },
                payload.len(),
                DecryptDirectSessionDelivery::EndpointData(EndpointDataDelivery::new(
                    source_peer,
                    payload.to_vec(),
                )),
            )
        };

        node.process_direct_session_data_batch_from_worker(vec![
            direct(7, b"batch one"),
            direct(8, b"batch two"),
        ])
        .await;

        for expected in [b"batch one".to_vec(), b"batch two".to_vec()] {
            match endpoint_io.event_rx.try_recv().expect("endpoint event") {
                crate::node::NodeEndpointEvent::Data {
                    source_peer: delivered_source,
                    payload,
                    ..
                } => {
                    assert_eq!(delivered_source, source_peer);
                    assert_eq!(payload, expected);
                }
                event => panic!("expected worker-decoded endpoint data event, got {event:?}"),
            }
        }
        assert!(
            !node.pending_session_traffic.has_traffic_for(&source_addr),
            "batch boundary should flush pending session traffic after direct deliveries"
        );
        let entry = node
            .sessions
            .get(&source_addr)
            .expect("session should remain");
        assert_eq!(entry.current_highest_counter(), Some(8));
    }

    #[tokio::test]
    async fn reserved_link_type_0x03_does_not_deliver_endpoint_data() {
        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 payload = b"reserved endpoint-shaped payload".to_vec();
        let mut plaintext = 0x0102_0304u32.to_le_bytes().to_vec();
        plaintext.push(0x03);
        plaintext.extend_from_slice(&payload);
        let remote_addr = crate::transport::TransportAddr::from_string("127.0.0.1:1234");

        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));
        node.process_authentic_fmp_plaintext(crate::node::AuthenticatedFmpPlaintext::new(
            source_peer,
            crate::transport::TransportId::new(1),
            &remote_addr,
            2_100,
            plaintext.len() + crate::node::wire::ESTABLISHED_HEADER_SIZE + crate::noise::TAG_SIZE,
            12,
            0,
            &plaintext,
        ))
        .await;

        assert!(
            endpoint_io.event_rx.try_recv().is_err(),
            "reserved 0x03 link payload must not bypass FSP session data"
        );
        let entry = node
            .sessions
            .get(&source_addr)
            .expect("session should remain");
        assert_eq!(entry.traffic_counters(), (0, 0, 0, 0));
    }

    #[test]
    fn session_runtime_receive_owns_decrypt_failure_recovery_gate() {
        let local = Identity::generate();
        let peer = Identity::generate();
        let mut entry = established_entry(&local, &peer);
        entry.mark_established(1_000);
        let plaintext_len = FSP_INNER_HEADER_SIZE + 32;
        let forged_ciphertext = vec![0u8; plaintext_len + crate::noise::TAG_SIZE];

        for attempt in 1..=DECRYPT_FAILURE_RECOVERY_THRESHOLD {
            let header = build_fsp_header(attempt as u64, 0, plaintext_len as u16);
            let mut wire = header.to_vec();
            wire.extend_from_slice(&forged_ciphertext);
            let parsed = FspEncryptedHeader::parse(&wire).expect("forged frame should parse");
            let outcome = SessionRuntimeReceive::new(
                &mut entry,
                &parsed,
                &wire[FSP_HEADER_SIZE..],
                1_280,
                false,
                1_000 + DECRYPT_FAILURE_RECOVERY_QUIET_MS + attempt as u64,
            )
            .open_established();

            match outcome {
                FspFrameOutcome::DecryptFailed {
                    counter,
                    consecutive,
                    recover_session,
                    ..
                } => {
                    assert_eq!(counter, attempt as u64);
                    assert_eq!(consecutive, attempt);
                    assert_eq!(
                        recover_session,
                        attempt == DECRYPT_FAILURE_RECOVERY_THRESHOLD
                    );
                }
                other => panic!("expected decrypt failure, got {other:?}"),
            }
        }
    }
