umsh_node/

lib.rs

#![allow(async_fn_in_trait)]
#![cfg_attr(not(feature = "std"), no_std)]

//! Application-facing node layer built on top of [`umsh-mac`](umsh_mac).
//!
//! > Note: This reference implementation is a work in progress and was developed
//! > with the assistance of an LLM. It should be considered experimental.
//!
//! `umsh-node` sits between the radio-facing MAC coordinator in `umsh-mac` and the
//! application. Where `umsh-mac` thinks in raw frames, keys, replay windows, and transmit
//! queues, `umsh-node` provides composable abstractions for sending and receiving messages,
//! tracking in-flight sends, and managing channel membership.
//!
//! The receive boundary is intentionally low-level: raw subscriptions get a
//! [`ReceivedPacketRef`] that stays close to the accepted on-wire packet. Payload-specific
//! helpers such as those in the `umsh-text` crate live one layer up and are built on top of
//! those raw packet callbacks.
//!
//! This crate requires `alloc` (heap allocation for `String`, `Vec`, etc.). It is
//! otherwise `no_std` compatible.
//!
//! # Architecture overview
//!
//! ```text
//! ┌──────────────────────────────────────────────────────────────┐
//! │  Application                                                 │
//! │  Host · LocalNode · PeerConnection · BoundChannel            │
//! ┌──────────────────────┴───────────────────────────────────────┐
//! │  Host                                                        │
//! │    ├── drives the shared MAC/runtime event loop              │
//! │    └── owns multiple LocalNode handles                       │
//! └──────────────────────┬───────────────────────────────────────┘
//!                        │
//! ┌──────────────────────┴───────────────────────────────────────┐
//! │  LocalNode<M>                                                │
//! │    ├── sends down through MacBackend                         │
//! │    ├── owns per-identity PFS state                           │
//! │    └── dispatches node/peer callback subscriptions           │
//! └──────────────────────┬───────────────────────────────────────┘
//!                        │  MacBackend trait
//! ┌──────────────────────┴───────────────────────────────────────┐
//! │  MacHandle → Mac<P>  (no_std, heapless)                      │
//! └──────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Key types
//!
//! - [`Host`] — preferred multi-identity driver. Owns the shared MAC event loop and routes
//!   inbound traffic to the right [`LocalNode`].
//! - [`LocalNode`] — per-identity application handle. Implements [`Transport`] (unicast /
//!   broadcast), owns PFS state, and exposes raw packet plus control-side subscriptions.
//! - [`BoundChannel`] — a channel bound to a `LocalNode`. Implements [`Transport`]
//!   (blind unicast / multicast). Available with the `software-crypto` feature.
//! - [`PeerConnection`] — relationship with one remote peer, generic over transport context,
//!   with peer-scoped callback subscriptions.
//! - [`Transport`] — shared send interface (`send` / `send_all`).
//! - [`SendProgressTicket`] — lightweight polling handle for observing in-flight send
//!   progress (`was_transmitted`, `was_acked`, `is_finished`).
//! - [`Subscription`] — owned callback registration that auto-unsubscribes on drop.
//! - [`ReceivedPacketRef`] — borrowed receive view passed into low-level `on_receive(...)`
//!   handlers and wrappers, including local RX observations such as RSSI, SNR, LQI, and
//!   receive timestamp.
//! - [`MacBackend`] — pluggable MAC backend trait for testability.
//!
//! # Control payload types
//!
//! [`umsh_text::OwnedTextMessage`], [`NodeIdentityPayload`], and [`OwnedMacCommand`] are
//! optional heap-allocated conveniences for callers that need
//! to retain parsed payloads across task boundaries. Most receive-side code should prefer the
//! borrowed views from the payload crates and [`ReceivedPacketRef`].
//!
//! # MAC abstraction
//!
//! [`MacBackend`] exposes the public send/configure surface of the MAC coordinator.
//! Safe PFS session management is available with `software-crypto` and builds on
//! that public surface directly.
//!
//! [`MacHandle`](umsh_mac::MacHandle) implements `MacBackend`, and test code can provide
//! a fake implementation to drive the node layer deterministically.
//!
//! # Typical usage
//!
//! For most applications, register callbacks and then let [`Host::run`] own the shared
//! MAC event loop:
//!
//! ```rust,ignore
//! let mut host = Host::new(mac_handle);
//! let node = host.add_node(identity_id);
//! let peer = node.peer(peer_key)?;
//! let chat = umsh_text::UnicastTextChatWrapper::from_peer(&peer);
//!
//! let _messages = chat.on_text(|packet, text| {
//!     println!(
//!         "peer says: {} (hops={})",
//!         text.body,
//!         packet.flood_hops().map(|h| h.remaining()).unwrap_or(0),
//!     );
//! });
//!
//! let _ticket = chat.send_text("hello", &SendOptions::default()).await?;
//! host.run().await?;
//! ```
//!
//! If you need to multiplex UMSH progress with another async source such as user input, use
//! [`Host::pump_once`] as a single wake-driven step. It already waits on radio activity and
//! protocol deadlines; you should not add a manual poll/sleep loop around it.
//!
//! ```rust,ignore
//! loop {
//!     tokio::select! {
//!         line = stdin.next_line() => { /* handle input */ }
//!         result = host.pump_once() => result?,
//!     }
//! }
//! ```
//!
//! If you need protocol fidelity instead of a payload wrapper, subscribe directly on the node
//! or peer and inspect the raw packet view:
//!
//! ```rust,ignore
//! let _raw = peer.on_receive(|packet| {
//!     if packet.packet_family() == umsh::mac::PacketFamily::Unicast {
//!         println!(
//!             "from={:?} encrypted={} mic_len={}",
//!             packet.from_key(),
//!             packet.encrypted(),
//!             packet.mic_len(),
//!         );
//!     }
//!     false
//! });
//! ```

#[cfg(not(feature = "alloc"))]
compile_error!("umsh-node currently requires the alloc feature");

extern crate alloc;

mod app_error;
mod app_payload;
mod app_util;
#[cfg(feature = "software-crypto")]
mod channel;
mod dispatch;
mod host;
mod identity;
pub mod identity_responder;
pub mod location;
mod mac;
pub mod mac_command;
mod node;
mod peer;
#[cfg(feature = "software-crypto")]
mod pfs;
mod receive;
mod ticket;
mod transport;

pub use app_error::{AppEncodeError, AppParseError};
pub use app_payload::{
    expect_payload_type, parse_mac_command_payload, parse_node_identity_payload, split_payload_type,
};
#[cfg(feature = "software-crypto")]
pub use channel::Channel;
pub use host::{Host, HostError};
pub use identity::{NodeCapabilities, NodeIdentityPayload, NodeRole};
pub use identity_responder::{
    IdentityRequestContext, NodeIdentityProfile, RespondDecision, default_respond_policy,
    never_respond_policy,
};
pub use mac::{MacBackend, MacBackendError};
pub use mac_command::OwnedMacCommand;
pub use mac_command::{CommandId, MacCommand};
#[cfg(feature = "software-crypto")]
pub use node::BoundChannel;
#[cfg(feature = "software-crypto")]
pub use node::PfsStatus;
pub use node::{LocalNode, NodeError, PfsFailure, PongMetadata, Subscription};
pub use peer::{PING_MIC_SIZE, PeerConnection};
pub use receive::{ChannelInfoRef, PacketFamily, ReceivedPacketRef, RouteHops, RxMetadata, Snr};
pub use ticket::{SendProgressTicket, SendToken};
pub use transport::Transport;

#[cfg(test)]
mod tests {
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use std::{
        cell::RefCell,
        collections::VecDeque,
        future::Future,
        num::NonZeroU8,
        pin::pin,
        rc::Rc,
        task::{Context, Poll, RawWaker, RawWakerVTable, Waker},
    };
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_core::{NodeHint, PublicKey};
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_crypto::NodeIdentity;
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_crypto::software::SoftwareIdentity;
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_hal::Snr;
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_mac::MacEventRef;
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_mac::{CapacityError, LocalIdentityId, PeerId, SendError, SendOptions, SendReceipt};

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use crate::ReceivedPacketRef;
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use crate::{MacBackend, MacBackendError, OwnedMacCommand, SendToken};
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_core::ChannelId;
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    use umsh_text::OwnedTextMessage;
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn peer_receive_handlers_precede_node_receive_handlers() {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};

        let mac = FakeMac::new(Vec::new());
        let dispatcher = Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new()));
        let membership = Rc::new(RefCell::new(NodeMembership::new()));
        let state = Rc::new(RefCell::new(LocalNodeState::new()));
        let node = LocalNode::new(LocalIdentityId(1), mac, dispatcher, membership, state);
        let peer = PublicKey([0x41; 32]);
        let peer_connection = block_on_ready(node.peer(peer)).unwrap();

        let call_order = Rc::new(RefCell::new(Vec::new()));
        let peer_call_order = call_order.clone();
        let _peer_subscription = peer_connection.on_receive(move |_| {
            peer_call_order.borrow_mut().push("peer");
            true
        });
        let node_call_order = call_order.clone();
        let _node_subscription = node.on_receive(move |_| {
            node_call_order.borrow_mut().push("node");
            true
        });

        assert!(node.dispatch_received_packet(&test_unicast_packet(peer, &[0x01, 0x02])));
        assert_eq!(call_order.borrow().as_slice(), ["peer"]);
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    fn test_node(mac: FakeMac) -> crate::node::LocalNode<FakeMac> {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};

        LocalNode::new(
            LocalIdentityId(1),
            mac,
            Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new())),
            Rc::new(RefCell::new(NodeMembership::new())),
            Rc::new(RefCell::new(LocalNodeState::new())),
        )
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn leaving_a_channel_unregisters_its_key_from_the_mac() {
        let mac = FakeMac::new(Vec::new());
        let node = test_node(mac.clone());
        let channel = crate::Channel::private(umsh_core::ChannelKey([0x11; 32]), "trail");

        block_on_ready(node.join(&channel)).unwrap();
        assert!(mac.holds_channel(channel.key()));

        block_on_ready(node.leave(&channel)).unwrap();
        assert!(!mac.holds_channel(channel.key()));
        assert!(node.bound_channel(&channel).is_none());
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn rejoining_after_leave_re_registers_the_key_with_the_mac() {
        let mac = FakeMac::new(Vec::new());
        let node = test_node(mac.clone());
        let channel = crate::Channel::private(umsh_core::ChannelKey([0x22; 32]), "camp");

        let first = block_on_ready(node.join(&channel)).unwrap();
        block_on_ready(node.leave(&channel)).unwrap();
        let second = block_on_ready(node.join(&channel)).unwrap();

        // The key is back in the MAC, the fresh handle is live, and the stale
        // one from before the leave is not.
        assert!(mac.holds_channel(channel.key()));
        assert!(second.is_active());
        assert!(!first.is_active());
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn leaving_a_channel_that_was_never_joined_is_a_no_op() {
        let mac = FakeMac::new(Vec::new());
        let node = test_node(mac.clone());
        let joined = crate::Channel::private(umsh_core::ChannelKey([0x33; 32]), "joined");
        let stranger = crate::Channel::private(umsh_core::ChannelKey([0x44; 32]), "stranger");

        block_on_ready(node.join(&joined)).unwrap();
        block_on_ready(node.leave(&stranger)).unwrap();

        assert!(mac.holds_channel(joined.key()));
        assert!(node.bound_channel(&joined).is_some());
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn receive_callbacks_can_observe_rx_metadata() {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};

        let mac = FakeMac::new(Vec::new());
        let dispatcher = Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new()));
        let membership = Rc::new(RefCell::new(NodeMembership::new()));
        let state = Rc::new(RefCell::new(LocalNodeState::new()));
        let node = LocalNode::new(LocalIdentityId(1), mac, dispatcher, membership, state);
        let peer = PublicKey([0x44; 32]);

        let observed = Rc::new(RefCell::new(None));
        let observed_for_callback = observed.clone();
        let _subscription = node.on_receive(move |packet| {
            *observed_for_callback.borrow_mut() = Some((
                packet.rssi(),
                packet.snr(),
                packet.lqi(),
                packet.received_at_ms(),
            ));
            true
        });

        let payload = encode_text_payload("metadata");
        let packet = test_unicast_packet_with_rx(
            peer,
            &payload,
            umsh_mac::RxMetadata::new(
                Some(-73),
                Some(Snr::from_centibels(123)),
                NonZeroU8::new(200),
                Some(123_456),
            ),
        );

        assert!(node.dispatch_received_packet(&packet));
        assert_eq!(
            *observed.borrow(),
            Some((
                Some(-73),
                Some(Snr::from_centibels(123)),
                NonZeroU8::new(200),
                Some(123_456),
            ))
        );
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn subscription_guard_unregisters_on_drop() {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};

        let mac = FakeMac::new(Vec::new());
        let dispatcher = Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new()));
        let membership = Rc::new(RefCell::new(NodeMembership::new()));
        let state = Rc::new(RefCell::new(LocalNodeState::new()));
        let node = LocalNode::new(LocalIdentityId(1), mac, dispatcher, membership, state);
        let peer = PublicKey([0x33; 32]);

        let hits = Rc::new(RefCell::new(0u32));
        {
            let hits = hits.clone();
            let _subscription = node.on_receive(move |_| {
                *hits.borrow_mut() += 1;
                true
            });
            assert!(node.dispatch_received_packet(&test_unicast_packet(peer, &[0x01, 0x02])));
        }

        assert_eq!(*hits.borrow(), 1);
        assert!(!node.dispatch_received_packet(&test_unicast_packet(peer, &[0x01, 0x02])));
        assert_eq!(*hits.borrow(), 1);
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn callbacks_observe_control_side_events_and_peer_ack_state() {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};

        let mac = FakeMac::new(Vec::new());
        let dispatcher = Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new()));
        let membership = Rc::new(RefCell::new(NodeMembership::new()));
        let state = Rc::new(RefCell::new(LocalNodeState::new()));
        let node = LocalNode::new(LocalIdentityId(1), mac, dispatcher, membership, state);

        let peer = PublicKey([0x42; 32]);
        let peer_connection = block_on_ready(node.peer(peer)).unwrap();
        let node_discovery = Rc::new(RefCell::new(Vec::new()));
        let beacons = Rc::new(RefCell::new(Vec::new()));
        let commands = Rc::new(RefCell::new(Vec::new()));
        let peer_acks = Rc::new(RefCell::new(Vec::new()));
        let peer_timeouts = Rc::new(RefCell::new(Vec::new()));

        let discovery_log = node_discovery.clone();
        let _discovered_subscription = node.on_node_discovered(move |key, name| {
            discovery_log
                .borrow_mut()
                .push((key, name.map(str::to_string)));
        });
        let beacon_log = beacons.clone();
        let _beacon_subscription = node.on_beacon(move |from_hint, from_key| {
            beacon_log.borrow_mut().push((from_hint, from_key));
        });
        let command_log = commands.clone();
        let _command_subscription = node.on_mac_command(move |from, command| {
            command_log.borrow_mut().push((from, command.clone()));
        });
        let peer_ack_log = peer_acks.clone();
        let _ack_subscription = peer_connection.on_ack_received(move |token| {
            peer_ack_log.borrow_mut().push(token);
        });
        let peer_timeout_log = peer_timeouts.clone();
        let _timeout_subscription = peer_connection.on_ack_timeout(move |token| {
            peer_timeout_log.borrow_mut().push(token);
        });

        let token = SendToken::new(LocalIdentityId(1), SendReceipt(12));
        let timeout_token = SendToken::new(LocalIdentityId(1), SendReceipt(13));
        let hint = NodeHint([1, 2, 3]);
        let command = OwnedMacCommand::EchoRequest {
            data: vec![9, 8, 7],
        };

        node.dispatch_node_discovered(peer, Some("alice"));
        node.dispatch_beacon(hint, Some(peer));
        node.dispatch_mac_command(peer, &command);
        node.dispatch_ack_received(peer, token);
        node.dispatch_ack_timeout(peer, timeout_token);

        assert_eq!(
            node_discovery.borrow().as_slice(),
            &[(peer, Some(String::from("alice")))]
        );
        assert_eq!(beacons.borrow().as_slice(), &[(hint, Some(peer))]);
        assert_eq!(commands.borrow().as_slice(), &[(peer, command)]);
        assert_eq!(peer_acks.borrow().as_slice(), &[token]);
        assert_eq!(peer_timeouts.borrow().as_slice(), &[timeout_token]);
    }

    /// A ping must travel the way the traffic it is measuring would, so the
    /// caller's options carry through untouched. The one exception is the ack
    /// request: the echo response already acknowledges the ping, so asking
    /// for a MAC ack too would put a second frame on the air for nothing.
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn ping_honours_caller_options_but_never_requests_a_mac_ack() {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};
        use umsh_core::MicSize;

        let mac = FakeMac::new(vec![[7u8; 32], [9u8; 32]]);
        let node = LocalNode::new(
            LocalIdentityId(1),
            mac.clone(),
            Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new())),
            Rc::new(RefCell::new(NodeMembership::new())),
            Rc::new(RefCell::new(LocalNodeState::new())),
        );
        let peer_connection = block_on_ready(node.peer(PublicKey([0x55; 32]))).unwrap();

        let options = SendOptions::default()
            .with_mic_size(MicSize::Mic4)
            .with_ack_requested(true)
            .with_trace_route()
            .with_flood_hops(3)
            .with_region_code([0x78, 0x53]);
        block_on_ready(peer_connection.ping(6, &options, 1_000)).unwrap();

        let sent = mac.take_unicasts().pop().expect("ping send");
        assert_eq!(sent.options.mic_size, MicSize::Mic4);
        assert!(sent.options.trace_route);
        assert_eq!(sent.options.flood_hops, Some(3));
        assert_eq!(sent.options.region_code, Some([0x78, 0x53]));
        assert!(!sent.options.ack_requested, "the echo response is the ack");

        // `no_flood` is a distinct state from an unset budget and must also
        // survive, rather than collapsing back to the wide default.
        block_on_ready(peer_connection.ping(0, &SendOptions::default().no_flood(), 1_000)).unwrap();
        let sent = mac.take_unicasts().pop().expect("ping send");
        assert_eq!(sent.options.flood_hops, None);
    }

    /// The MIC size pings are normally sent with. A ping frame is otherwise
    /// nearly half authenticator.
    #[test]
    fn ping_mic_size_is_eight_bytes() {
        assert_eq!(crate::PING_MIC_SIZE, umsh_core::MicSize::Mic8);
        assert_eq!(crate::PING_MIC_SIZE.byte_len(), 8);
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn pfs_routed_send_tracks_ack_against_ephemeral_identity() {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};

        let mac = FakeMac::new(vec![[7u8; 32], [9u8; 32]]);
        let dispatcher = Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new()));
        let membership = Rc::new(RefCell::new(NodeMembership::new()));
        let state = Rc::new(RefCell::new(LocalNodeState::new()));
        let node = LocalNode::new(
            LocalIdentityId(1),
            mac.clone(),
            dispatcher.clone(),
            membership,
            state,
        );

        let peer = PublicKey([0x55; 32]);
        let peer_connection = block_on_ready(node.peer(peer)).unwrap();
        let options = SendOptions::default().with_ack_requested(true);

        block_on_ready(node.request_pfs(&peer, 60, &options)).unwrap();
        let request = mac.take_unicasts().pop().expect("request send");
        let request_command = parse_owned_mac_command(&request.payload);
        let request_ephemeral = match request_command {
            OwnedMacCommand::PfsSessionRequest { ephemeral_key, .. } => ephemeral_key,
            other => panic!("unexpected request payload: {other:?}"),
        };

        block_on_ready(node.handle_pfs_command(
            &peer,
            &OwnedMacCommand::PfsSessionResponse {
                ephemeral_key: PublicKey([0x44; 32]),
                duration_minutes: 60,
            },
            &options,
        ))
        .unwrap();

        let payload = encode_text_payload("hello over pfs");
        let ticket = block_on_ready(peer_connection.send(&payload, &options)).unwrap();
        let sent = mac.take_unicasts().pop().expect("pfs-routed send");
        assert_eq!(sent.from, LocalIdentityId(10));
        assert_eq!(sent.to, PublicKey([0x44; 32]));

        let pairwise_from_pfs = PublicKey([0x44; 32]);
        let _ = request_ephemeral; // Keeps the request path explicit in the test setup.
        dispatcher.borrow_mut().dispatch_ticket_state(
            sent.from,
            &MacEventRef::AckReceived {
                peer: pairwise_from_pfs,
                receipt: SendReceipt(42),
            },
        );
        assert!(ticket.was_acked());
        assert!(ticket.is_finished());
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn pfs_session_manager_request_and_teardown() {
        use crate::pfs::PfsSessionManager;

        let mac = FakeMac::new(vec![[3u8; 32]]);
        let peer_long_term = PublicKey([0x55; 32]);
        let options = SendOptions::default().with_ack_requested(true);

        let mut pfs = PfsSessionManager::new();
        block_on_ready(pfs.request_session(
            &mac,
            LocalIdentityId(1),
            &peer_long_term,
            60,
            &options,
        ))
        .unwrap();

        let sent = mac.take_unicasts();
        assert_eq!(sent.len(), 1);
        assert_eq!(sent[0].from, LocalIdentityId(1));
        assert_eq!(sent[0].to, peer_long_term);
        assert_eq!(
            parse_owned_mac_command(&sent[0].payload),
            OwnedMacCommand::PfsSessionRequest {
                ephemeral_key: *SoftwareIdentity::from_secret_bytes(&[3u8; 32]).public_key(),
                duration_minutes: 60,
            }
        );

        assert!(
            block_on_ready(pfs.end_session(
                &mac,
                LocalIdentityId(1),
                &peer_long_term,
                true,
                &options,
            ))
            .unwrap()
        );
        let sent = mac.take_unicasts();
        assert_eq!(sent.len(), 1);
        assert_eq!(
            parse_owned_mac_command(&sent[0].payload),
            OwnedMacCommand::EndPfsSession
        );
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn pfs_end_session_errors_when_missing() {
        use crate::pfs::PfsSessionManager;

        let mac = FakeMac::new(Vec::new());
        let options = SendOptions::default();
        let mut pfs = PfsSessionManager::new();

        let error = block_on_ready(pfs.end_session(
            &mac,
            LocalIdentityId(1),
            &PublicKey([0x77; 32]),
            true,
            &options,
        ))
        .unwrap_err();
        assert!(matches!(error, crate::NodeError::PfsSessionMissing));
    }

    #[cfg(feature = "unsafe-advanced")]
    fn encode_text_payload(text: &str) -> Vec<u8> {
        let message = OwnedTextMessage::basic(text);
        let mut body = [0u8; 512];
        let len = umsh_text::text_message::encode(&message.as_borrowed(), &mut body).unwrap();
        let mut payload = Vec::with_capacity(len + 1);
        payload.push(umsh_core::PayloadType::TextMessage as u8);
        payload.extend_from_slice(&body[..len]);
        payload
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    fn test_unicast_packet<'a>(from: PublicKey, payload: &'a [u8]) -> ReceivedPacketRef<'a> {
        test_unicast_packet_with_rx(from, payload, umsh_mac::RxMetadata::default())
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    fn test_unicast_packet_with_rx<'a>(
        from: PublicKey,
        payload: &'a [u8],
        rx: umsh_mac::RxMetadata,
    ) -> ReceivedPacketRef<'a> {
        let wire = Box::leak(payload.to_vec().into_boxed_slice());
        let header = umsh_core::PacketHeader {
            fcf: umsh_core::Fcf::new(umsh_core::PacketType::Unicast, false, false),
            options_range: 0..0,
            flood_hops: None,
            dst: None,
            channel: None,
            source: umsh_core::SourceAddrRef::Hint(from.hint()),
            sec_info: None,
            body_range: 0..wire.len(),
            mic_range: wire.len()..wire.len(),
            total_len: wire.len(),
        };
        ReceivedPacketRef::new(
            wire,
            wire,
            header,
            umsh_core::ParsedOptions::default(),
            Some(from),
            Some(from.hint()),
            true,
            None,
            rx,
        )
    }

    /// A received broadcast frame, as a solicitation would arrive: optionally
    /// carrying an FHOPS byte and a Route option. Only the ranges matter, so
    /// the wire is just the route bytes.
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    fn test_broadcast_packet(
        from: PublicKey,
        flood_hops: Option<u8>,
        route: Option<&'static [u8]>,
    ) -> ReceivedPacketRef<'static> {
        let route_len = route.map_or(0, <[u8]>::len);
        let wire: &'static [u8] = Box::leak(
            route
                .map_or_else(Vec::new, <[u8]>::to_vec)
                .into_boxed_slice(),
        );
        let mut options = umsh_core::ParsedOptions::default();
        if route.is_some() {
            options.source_route = Some(0..route_len);
        }
        let header = umsh_core::PacketHeader {
            fcf: umsh_core::Fcf::new(umsh_core::PacketType::Broadcast, false, false),
            options_range: 0..route_len,
            flood_hops: flood_hops.map(umsh_core::FloodHops),
            dst: None,
            channel: None,
            source: umsh_core::SourceAddrRef::Hint(from.hint()),
            sec_info: None,
            body_range: route_len..wire.len(),
            mic_range: wire.len()..wire.len(),
            total_len: wire.len(),
        };
        ReceivedPacketRef::new(
            wire,
            &wire[route_len..],
            header,
            options,
            Some(from),
            Some(from.hint()),
            false,
            None,
            umsh_mac::RxMetadata::default(),
        )
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[derive(Clone, Default)]
    struct FakeMac {
        state: Rc<RefCell<FakeMacState>>,
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[derive(Default)]
    struct FakeMacState {
        random_blocks: VecDeque<[u8; 32]>,
        next_peer_id: u8,
        next_ephemeral_id: u8,
        now_ms: u64,
        unicasts: Vec<SentUnicast>,
        removed_ephemerals: Vec<LocalIdentityId>,
        peers: Vec<(PublicKey, PeerId)>,
        channels: Vec<umsh_core::ChannelKey>,
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[derive(Clone, Debug, PartialEq, Eq)]
    struct SentUnicast {
        from: LocalIdentityId,
        to: PublicKey,
        payload: Vec<u8>,
        options: SendOptions,
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    impl FakeMac {
        fn new(random_blocks: Vec<[u8; 32]>) -> Self {
            Self {
                state: Rc::new(RefCell::new(FakeMacState {
                    random_blocks: random_blocks.into(),
                    next_peer_id: 0,
                    next_ephemeral_id: 10,
                    now_ms: 1_000,
                    ..FakeMacState::default()
                })),
            }
        }

        fn take_unicasts(&self) -> Vec<SentUnicast> {
            core::mem::take(&mut self.state.borrow_mut().unicasts)
        }

        fn holds_channel(&self, key: &umsh_core::ChannelKey) -> bool {
            self.state.borrow().channels.iter().any(|k| k.0 == key.0)
        }
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    impl MacBackend for FakeMac {
        type SendError = SendError;
        type CapacityError = CapacityError;
        type RunError = core::convert::Infallible;

        async fn next_event(
            &self,
            _on_event: impl FnMut(LocalIdentityId, umsh_mac::MacEventRef<'_>),
        ) -> Result<(), Self::RunError> {
            // FakeMac is used to drive LocalNode directly in tests, never as a
            // Host run loop, so this stub is never invoked.
            Ok(())
        }

        async fn add_peer(
            &self,
            key: PublicKey,
        ) -> Result<PeerId, MacBackendError<Self::SendError, Self::CapacityError>> {
            let mut state = self.state.borrow_mut();
            if let Some((_, existing)) = state
                .peers
                .iter()
                .find(|(existing_key, _)| *existing_key == key)
            {
                return Ok(*existing);
            }
            let peer_id = PeerId(state.next_peer_id);
            state.next_peer_id = state.next_peer_id.wrapping_add(1);
            state.peers.push((key, peer_id));
            Ok(peer_id)
        }

        async fn add_private_channel(
            &self,
            key: umsh_core::ChannelKey,
        ) -> Result<(), MacBackendError<Self::SendError, Self::CapacityError>> {
            let mut state = self.state.borrow_mut();
            if !state.channels.iter().any(|k| k.0 == key.0) {
                state.channels.push(key);
            }
            Ok(())
        }

        async fn add_named_channel(
            &self,
            _name: &str,
        ) -> Result<(), MacBackendError<Self::SendError, Self::CapacityError>> {
            Ok(())
        }

        async fn remove_channel(&self, key: &umsh_core::ChannelKey) -> bool {
            let mut state = self.state.borrow_mut();
            let before = state.channels.len();
            state.channels.retain(|k| k.0 != key.0);
            state.channels.len() != before
        }

        async fn send_broadcast(
            &self,
            _from: LocalIdentityId,
            _payload: &[u8],
            _options: &SendOptions,
        ) -> Result<SendReceipt, MacBackendError<Self::SendError, Self::CapacityError>> {
            Ok(SendReceipt(99))
        }

        async fn send_multicast(
            &self,
            _from: LocalIdentityId,
            _channel: &ChannelId,
            _payload: &[u8],
            _options: &SendOptions,
        ) -> Result<SendReceipt, MacBackendError<Self::SendError, Self::CapacityError>> {
            Ok(SendReceipt(99))
        }

        async fn send_unicast(
            &self,
            from: LocalIdentityId,
            dst: &PublicKey,
            payload: &[u8],
            options: &SendOptions,
        ) -> Result<Option<SendReceipt>, MacBackendError<Self::SendError, Self::CapacityError>>
        {
            self.state.borrow_mut().unicasts.push(SentUnicast {
                from,
                to: *dst,
                payload: payload.to_vec(),
                options: options.clone(),
            });
            Ok(Some(SendReceipt(42)))
        }

        async fn send_blind_unicast(
            &self,
            from: LocalIdentityId,
            dst: &PublicKey,
            _channel: &ChannelId,
            payload: &[u8],
            options: &SendOptions,
        ) -> Result<Option<SendReceipt>, MacBackendError<Self::SendError, Self::CapacityError>>
        {
            self.send_unicast(from, dst, payload, options).await
        }

        async fn fill_random(&self, dest: &mut [u8]) {
            let mut state = self.state.borrow_mut();
            let next = state.random_blocks.pop_front().expect("test rng exhausted");
            dest.copy_from_slice(&next[..dest.len()]);
        }

        async fn now_ms(&self) -> u64 {
            self.state.borrow().now_ms
        }

        async fn register_ephemeral(
            &self,
            _parent: LocalIdentityId,
            _identity: SoftwareIdentity,
        ) -> Result<LocalIdentityId, MacBackendError<Self::SendError, Self::CapacityError>>
        {
            let mut state = self.state.borrow_mut();
            let id = LocalIdentityId(state.next_ephemeral_id);
            state.next_ephemeral_id = state.next_ephemeral_id.wrapping_add(1);
            Ok(id)
        }

        async fn remove_ephemeral(&self, id: LocalIdentityId) -> bool {
            self.state.borrow_mut().removed_ephemerals.push(id);
            true
        }
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    fn parse_owned_mac_command(payload: &[u8]) -> OwnedMacCommand {
        OwnedMacCommand::from(
            crate::parse_mac_command_payload(umsh_core::PacketType::Unicast, payload).unwrap(),
        )
    }

    #[cfg(feature = "unsafe-advanced")]
    fn block_on_ready<F: Future>(future: F) -> F::Output {
        fn raw_waker() -> RawWaker {
            fn clone(_: *const ()) -> RawWaker {
                raw_waker()
            }
            fn wake(_: *const ()) {}
            fn wake_by_ref(_: *const ()) {}
            fn drop(_: *const ()) {}

            RawWaker::new(
                core::ptr::null(),
                &RawWakerVTable::new(clone, wake, wake_by_ref, drop),
            )
        }

        let waker = unsafe { Waker::from_raw(raw_waker()) };
        let mut context = Context::from_waker(&waker);
        let mut future = pin!(future);
        match future.as_mut().poll(&mut context) {
            Poll::Ready(value) => value,
            Poll::Pending => panic!("test future unexpectedly returned Poll::Pending"),
        }
    }

    // --- Identity Request responder ---

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    fn responder_node(mac: &FakeMac) -> crate::LocalNode<FakeMac> {
        use crate::node::{LocalNode, LocalNodeState, NodeMembership};
        let dispatcher = Rc::new(RefCell::new(crate::dispatch::EventDispatcher::new()));
        let membership = Rc::new(RefCell::new(NodeMembership::new()));
        let state = Rc::new(RefCell::new(LocalNodeState::new()));
        LocalNode::new(
            LocalIdentityId(1),
            mac.clone(),
            dispatcher,
            membership,
            state,
        )
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    fn test_profile(public_key: PublicKey) -> crate::NodeIdentityProfile {
        crate::NodeIdentityProfile::new(
            public_key,
            crate::NodeRole::Repeater,
            crate::NodeCapabilities::REPEATER | crate::NodeCapabilities::TEXT_MESSAGES,
        )
        .with_name("repeater-1")
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn responder_answers_selected_request_with_unicast_identity() {
        let mac = FakeMac::new(Vec::new());
        let node = responder_node(&mac);
        let our_key = PublicKey([0x11; 32]);
        let requester = PublicKey([0x41; 32]);
        node.enable_identity_responder_default(test_profile(our_key));

        // Broadcast-style request that filters on our hint and carries a nonce.
        let options = crate::mac_command::IdentityRequestBuilder::new()
            .nonce(0xCAFE_F00D)
            .unwrap()
            .filter_hint(&our_key.hint())
            .unwrap()
            .build();
        let packet = test_unicast_packet(requester, &[]);

        let plan = node
            .evaluate_identity_request(&packet, requester, &options)
            .expect("responder should produce a reply plan");
        block_on_ready(node.send_identity_response(plan));

        let unicasts = mac.take_unicasts();
        assert_eq!(unicasts.len(), 1, "exactly one unicast reply");
        let reply = &unicasts[0];
        assert_eq!(reply.to, requester, "reply is addressed to the requester");
        // test_unicast_packet is source_authenticated → requester already has
        // our key → hint source suffices.
        assert!(!reply.options.full_source);
        // A targeted request gets an immediate reply; only broadcast and
        // multicast solicitations are jittered.
        assert_eq!(reply.options.tx_delay_ms, None);

        assert_eq!(reply.payload[0], umsh_core::PayloadType::NodeIdentity as u8);
        let identity = crate::NodeIdentityPayload::from_bytes(&reply.payload[1..]).unwrap();
        assert_eq!(identity.role, crate::NodeRole::Repeater);
        assert_eq!(identity.name.as_deref(), Some("repeater-1"));
        assert_eq!(identity.nonce, Some(0xCAFE_F00D), "request nonce echoed");
        assert!(identity.signature.is_none(), "responses are unsigned");
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn responder_answers_broadcast_solicitation_with_delayed_full_source_reply() {
        let mac = FakeMac::new(vec![[0x12; 32]]);
        let node = responder_node(&mac);
        let our_key = PublicKey([0x11; 32]);
        let requester = PublicKey([0x41; 32]);
        node.enable_identity_responder_default(test_profile(our_key));

        let options = crate::mac_command::IdentityRequestBuilder::new()
            .nonce(0x0000_BEEF)
            .unwrap()
            .filter_role(crate::NodeRole::Repeater)
            .unwrap()
            .build();
        let packet = test_broadcast_packet(requester, None, None);

        let plan = node
            .evaluate_identity_request(&packet, requester, &options)
            .expect("selected broadcast solicitation produces a plan");
        block_on_ready(node.send_identity_response(plan));

        let unicasts = mac.take_unicasts();
        assert_eq!(unicasts.len(), 1);
        let reply = &unicasts[0];
        assert_eq!(reply.to, requester);
        // A broadcast is unauthenticated, so the requester may lack our key.
        assert!(reply.options.full_source);
        // The reply is held for a random slice of the 30-second window so the
        // selected nodes do not all answer the same frame at once.
        let delay = reply
            .options
            .tx_delay_ms
            .expect("broadcast replies are jittered");
        assert!((500..=30_000).contains(&delay));
        // No FILTER_NODE_HINT narrowed the solicitation, so the reply stays
        // inside the one hop the request was allowed: no FHOPS field.
        assert_eq!(reply.options.flood_hops, None);
    }

    /// A `FILTER_NODE_HINT` names one answering node, so the solicitation may
    /// be flood routed and the reply keeps its normal flood budget.
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn responder_answers_flood_routed_hint_filtered_solicitation() {
        let mac = FakeMac::new(vec![[0x12; 32]]);
        let node = responder_node(&mac);
        let our_key = PublicKey([0x11; 32]);
        let requester = PublicKey([0x41; 32]);
        node.enable_identity_responder_default(test_profile(our_key));

        let options = crate::mac_command::IdentityRequestBuilder::new()
            .filter_hint(&our_key.hint())
            .unwrap()
            .build();
        // FHOPS_REM=2, FHOPS_ACC=1: repeated once, two hops of budget left.
        let repeated = test_broadcast_packet(requester, Some(0x21), None);

        let plan = node
            .evaluate_identity_request(&repeated, requester, &options)
            .expect("a hint-filtered solicitation may be flood routed");
        block_on_ready(node.send_identity_response(plan));

        let unicasts = mac.take_unicasts();
        assert_eq!(unicasts.len(), 1);
        assert!(
            unicasts[0].options.flood_hops.is_some(),
            "the reply may be flooded back to a requester that named us"
        );
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn responder_drops_repeated_or_routed_broadcast_solicitations() {
        let mac = FakeMac::new(Vec::new());
        let node = responder_node(&mac);
        let our_key = PublicKey([0x11; 32]);
        let requester = PublicKey([0x41; 32]);
        node.enable_identity_responder_default(test_profile(our_key));
        let options = crate::mac_command::IdentityRequestBuilder::new()
            .filter_role(crate::NodeRole::Repeater)
            .unwrap()
            .build();

        // A nonzero FHOPS byte marks a request that was flood routed, and no
        // FILTER_NODE_HINT narrows this one to a single answering node.
        let repeated = test_broadcast_packet(requester, Some(0x21), None);
        assert!(
            node.evaluate_identity_request(&repeated, requester, &options)
                .is_none()
        );

        // A non-empty Route option is a steered request. That holds whatever
        // the filters say, so check it with a hint filter too.
        let routed = test_broadcast_packet(requester, None, Some(&[0xAB, 0xCD]));
        assert!(
            node.evaluate_identity_request(&routed, requester, &options)
                .is_none()
        );
        let hint_options = crate::mac_command::IdentityRequestBuilder::new()
            .filter_hint(&our_key.hint())
            .unwrap()
            .build();
        assert!(
            node.evaluate_identity_request(&routed, requester, &hint_options)
                .is_none()
        );

        // A zeroed FHOPS byte and a present-but-empty Route option are fine.
        let clean = test_broadcast_packet(requester, Some(0x00), Some(&[]));
        assert!(
            node.evaluate_identity_request(&clean, requester, &options)
                .is_some()
        );
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn responder_ignores_request_that_filters_exclude() {
        let mac = FakeMac::new(Vec::new());
        let node = responder_node(&mac);
        let our_key = PublicKey([0x11; 32]);
        node.enable_identity_responder_default(test_profile(our_key));

        // Filter targets a different hint → we are not selected.
        let other_hint = PublicKey([0x99; 32]).hint();
        let options = crate::mac_command::IdentityRequestBuilder::new()
            .filter_hint(&other_hint)
            .unwrap()
            .build();
        let packet = test_unicast_packet(PublicKey([0x41; 32]), &[]);

        assert!(
            node.evaluate_identity_request(&packet, PublicKey([0x41; 32]), &options)
                .is_none()
        );
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn responder_respects_ignore_policy() {
        let mac = FakeMac::new(Vec::new());
        let node = responder_node(&mac);
        let our_key = PublicKey([0x11; 32]);
        node.enable_identity_responder(test_profile(our_key), |_ctx| {
            crate::RespondDecision::Ignore
        });

        // No filters → selects everyone, so only the policy can decline.
        let options = crate::mac_command::IdentityRequestBuilder::new().build();
        let packet = test_unicast_packet(PublicKey([0x41; 32]), &[]);

        assert!(
            node.evaluate_identity_request(&packet, PublicKey([0x41; 32]), &options)
                .is_none()
        );
    }

    /// Advertisements are built from the installed profile, so a node that
    /// has opted out of being discovered must still be able to read it back.
    /// Silencing the responder with a policy rather than uninstalling it is
    /// what keeps the two independent.
    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn never_respond_policy_keeps_the_profile_readable() {
        let mac = FakeMac::new(Vec::new());
        let node = responder_node(&mac);
        let our_key = PublicKey([0x11; 32]);
        node.enable_identity_responder(test_profile(our_key), crate::never_respond_policy);

        let options = crate::mac_command::IdentityRequestBuilder::new().build();
        let packet = test_unicast_packet(PublicKey([0x41; 32]), &[]);
        assert!(
            node.evaluate_identity_request(&packet, PublicKey([0x41; 32]), &options)
                .is_none(),
            "a silenced responder answers nothing"
        );
        assert_eq!(
            node.with_identity_profile(|profile| profile.public_key),
            Some(our_key),
            "yet the profile an advertisement is built from is still there"
        );
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn responder_disabled_yields_no_plan() {
        let mac = FakeMac::new(Vec::new());
        let node = responder_node(&mac);
        let options = crate::mac_command::IdentityRequestBuilder::new().build();
        let packet = test_unicast_packet(PublicKey([0x41; 32]), &[]);
        assert!(
            node.evaluate_identity_request(&packet, PublicKey([0x41; 32]), &options)
                .is_none()
        );
    }

    #[cfg(all(feature = "software-crypto", feature = "unsafe-advanced"))]
    #[test]
    fn default_policy_full_source_tracks_authentication() {
        use crate::identity_responder::{IdentityRequestContext, default_respond_policy};
        use crate::mac_command::IdentityRequestFilters;

        let base = |source_authenticated: bool| IdentityRequestContext {
            from_key: PublicKey([0x41; 32]),
            from_hint: None,
            source_authenticated,
            has_full_source: false,
            channel: None,
            family: crate::PacketFamily::Unicast,
            filters: IdentityRequestFilters::new(&[]),
            rssi: None,
            snr: None,
        };

        // Authenticated request → sender already holds our key → hint reply.
        assert_eq!(
            default_respond_policy(&base(true)),
            crate::RespondDecision::Respond { full_source: false }
        );
        // Unauthenticated request → sender may lack our key → include it.
        assert_eq!(
            default_respond_policy(&base(false)),
            crate::RespondDecision::Respond { full_source: true }
        );
    }
}