rns_net/

node.rs

//! RnsNode: high-level lifecycle management.
//!
//! Wires together the driver, interfaces, and timer thread.

use std::collections::{HashMap, HashSet};
use std::io::{self, Read};
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::thread::{self, JoinHandle};
use std::time::Duration;

use rns_core::transport::announce_verify_queue::OverflowPolicy as AnnounceQueueOverflowPolicy;
use rns_core::transport::types::TransportConfig;
use rns_crypto::identity::Identity;
use rns_crypto::{OsRng, Rng};

use crate::config;
use crate::driver::{AnnounceRateDefaults, Callbacks, Driver};
#[cfg(feature = "iface-backbone")]
use crate::driver::{BackbonePeerPoolCandidateConfig, BackbonePeerPoolSettings};
use crate::event::{self, Event, EventSender};
use crate::ifac;
#[cfg(feature = "iface-auto")]
use crate::interface::auto::{auto_runtime_handle_from_config, AutoConfig};
#[cfg(feature = "iface-backbone")]
use crate::interface::backbone::{
    client_config_from_mode, client_priority_from_mode, client_runtime_handle_from_mode,
    peer_state_handle_from_mode, runtime_handle_from_mode, BackboneMode,
};
#[cfg(feature = "iface-i2p")]
use crate::interface::i2p::{i2p_runtime_handle_from_config, I2pConfig};
#[cfg(feature = "iface-local")]
use crate::interface::local::LocalServerConfig;
#[cfg(feature = "iface-pipe")]
use crate::interface::pipe::{pipe_runtime_handle_from_config, PipeConfig};
#[cfg(feature = "iface-rnode")]
use crate::interface::rnode::{rnode_runtime_handle_from_config, RNodeConfig};
#[cfg(feature = "iface-tcp")]
use crate::interface::tcp::{tcp_client_runtime_handle_from_config, TcpClientConfig};
#[cfg(feature = "iface-tcp")]
use crate::interface::tcp_server::{
    runtime_handle_from_config as tcp_runtime_handle_from_config, TcpServerConfig,
};
#[cfg(feature = "iface-udp")]
use crate::interface::udp::{udp_runtime_handle_from_config, UdpConfig};
use crate::interface::{InterfaceEntry, InterfaceStats};
use crate::storage;
use crate::time;

static NEXT_RESOURCE_TRANSFER_ID: AtomicU64 = AtomicU64::new(1);

#[cfg(test)]
const DEFAULT_KNOWN_DESTINATIONS_TTL: Duration = Duration::from_secs(48 * 60 * 60);
const DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES: usize = 8192;

/// Parse an interface mode string to the corresponding constant.
/// Matches Python's `_synthesize_interface()` in `RNS/Reticulum.py`.
fn parse_interface_mode(mode: &str) -> u8 {
    match mode.to_lowercase().as_str() {
        "full" => rns_core::constants::MODE_FULL,
        "access_point" | "accesspoint" | "ap" => rns_core::constants::MODE_ACCESS_POINT,
        "pointtopoint" | "ptp" => rns_core::constants::MODE_POINT_TO_POINT,
        "roaming" => rns_core::constants::MODE_ROAMING,
        "boundary" => rns_core::constants::MODE_BOUNDARY,
        "gateway" | "gw" => rns_core::constants::MODE_GATEWAY,
        "internal" => rns_core::constants::MODE_INTERNAL,
        _ => rns_core::constants::MODE_FULL,
    }
}

fn parse_optional_interface_mode(mode: &str) -> Option<u8> {
    match mode.to_lowercase().as_str() {
        "full" => Some(rns_core::constants::MODE_FULL),
        "access_point" | "accesspoint" | "ap" => Some(rns_core::constants::MODE_ACCESS_POINT),
        "pointtopoint" | "ptp" => Some(rns_core::constants::MODE_POINT_TO_POINT),
        "roaming" => Some(rns_core::constants::MODE_ROAMING),
        "boundary" => Some(rns_core::constants::MODE_BOUNDARY),
        "gateway" | "gw" => Some(rns_core::constants::MODE_GATEWAY),
        "internal" => Some(rns_core::constants::MODE_INTERNAL),
        _ => None,
    }
}

fn parse_interface_gravity(value: Option<&str>, default_gravity: i64) -> i64 {
    value
        .and_then(|value| value.parse().ok())
        .unwrap_or(default_gravity)
}

/// Apply Reticulum's mode normalization for discoverable interfaces.
fn normalize_discovery_mode(
    interface_type: &str,
    configured_mode: u8,
    discoverable: bool,
    ignore_config_warnings: bool,
) -> u8 {
    if !discoverable
        || ignore_config_warnings
        || matches!(
            configured_mode,
            rns_core::constants::MODE_ACCESS_POINT
                | rns_core::constants::MODE_GATEWAY
                | rns_core::constants::MODE_INTERNAL
        )
    {
        return configured_mode;
    }

    if matches!(interface_type, "RNodeInterface" | "RNodeMultiInterface") {
        rns_core::constants::MODE_ACCESS_POINT
    } else {
        rns_core::constants::MODE_GATEWAY
    }
}

fn default_ingress_control_for_type(
    iface_type: &str,
) -> rns_core::transport::types::IngressControlConfig {
    match iface_type {
        "AutoInterface" | "BackboneInterface" | "TCPClientInterface" | "TCPServerInterface"
        | "UDPInterface" | "I2PInterface" => {
            rns_core::transport::types::IngressControlConfig::enabled()
        }
        _ => rns_core::transport::types::IngressControlConfig::disabled(),
    }
}

fn parse_ingress_control_config(
    iface_type: &str,
    params: &std::collections::HashMap<String, String>,
) -> Result<rns_core::transport::types::IngressControlConfig, String> {
    let mut config = default_ingress_control_for_type(iface_type);

    if let Some(v) = params.get("ingress_control") {
        config.enabled = config::parse_bool_pub(v)
            .ok_or_else(|| format!("ingress_control must be a boolean, got '{}'", v))?;
    }
    if let Some(v) = params.get("ic_max_held_announces") {
        config.max_held_announces = v
            .parse::<usize>()
            .map_err(|_| format!("ic_max_held_announces must be an integer, got '{}'", v))?;
    }
    if let Some(v) = params.get("ic_burst_hold") {
        config.burst_hold = parse_nonnegative_f64("ic_burst_hold", v)?;
    }
    if let Some(v) = params.get("ic_burst_freq_new") {
        config.burst_freq_new = parse_nonnegative_f64("ic_burst_freq_new", v)?;
    }
    if let Some(v) = params.get("ic_burst_freq") {
        config.burst_freq = parse_nonnegative_f64("ic_burst_freq", v)?;
    }
    if let Some(v) = params.get("ic_pr_burst_freq_new") {
        config.pr_burst_freq_new = parse_nonnegative_f64("ic_pr_burst_freq_new", v)?;
    }
    if let Some(v) = params.get("ic_pr_burst_freq") {
        config.pr_burst_freq = parse_nonnegative_f64("ic_pr_burst_freq", v)?;
    }
    if let Some(v) = params.get("egress_control") {
        config.egress_enabled = config::parse_bool_pub(v)
            .ok_or_else(|| format!("egress_control must be a boolean, got '{}'", v))?;
    }
    if let Some(v) = params.get("ec_pr_freq") {
        config.egress_pr_freq = parse_nonnegative_f64("ec_pr_freq", v)?;
    }
    if let Some(v) = params.get("ic_new_time") {
        config.new_time = parse_nonnegative_f64("ic_new_time", v)?;
    }
    if let Some(v) = params.get("ic_burst_penalty") {
        config.burst_penalty = parse_nonnegative_f64("ic_burst_penalty", v)?;
    }
    if let Some(v) = params.get("ic_held_release_interval") {
        config.held_release_interval = parse_nonnegative_f64("ic_held_release_interval", v)?;
    }

    Ok(config)
}

fn parse_nonnegative_f64(key: &str, value: &str) -> Result<f64, String> {
    let parsed = value
        .parse::<f64>()
        .map_err(|_| format!("{} must be numeric, got '{}'", key, value))?;
    if parsed < 0.0 {
        return Err(format!("{} must be >= 0, got '{}'", key, value));
    }
    Ok(parsed)
}

/// Reject empty and the nonsensical literal `"None"` IFAC config values.
///
/// Upstream treats `"None"` as an unset value and logs a warning, because a
/// quoted TOML/JSON `None` is commonly mistaken for a real network name or
/// passphrase. Empty strings are also ignored.
fn sanitize_ifac_config_string(kind: &str, value: &str) -> Option<String> {
    if value == "None" {
        log::warn!(
            "Ambiguous IFAC {} \"None\", this value is ignored and an IFAC {} has NOT been set",
            kind,
            kind
        );
        return None;
    }
    if value.is_empty() {
        return None;
    }
    Some(value.to_string())
}

/// Extract IFAC configuration from interface params, if present.
/// Returns None if neither networkname/network_name nor passphrase/pass_phrase is set.
fn extract_ifac_config(
    params: &std::collections::HashMap<String, String>,
    default_size: usize,
) -> Option<IfacConfig> {
    let netname = params
        .get("networkname")
        .or_else(|| params.get("network_name"))
        .cloned()
        .and_then(|value| sanitize_ifac_config_string("network name", &value));
    let netkey = params
        .get("passphrase")
        .or_else(|| params.get("pass_phrase"))
        .cloned()
        .and_then(|value| sanitize_ifac_config_string("passphrase", &value));

    if netname.is_none() && netkey.is_none() {
        return None;
    }

    // ifac_size is specified in bits in config, divide by 8 for bytes
    let size = params
        .get("ifac_size")
        .and_then(|v| v.parse::<usize>().ok())
        .map(|bits| (bits / 8).max(1))
        .unwrap_or(default_size);

    Some(IfacConfig {
        netname,
        netkey,
        size,
    })
}

/// Upstream enables the static transport identity when a discoverable interface
/// is configured on a non-transport instance, so the announced transport
/// identity is stable across restarts instead of ephemeral.
fn effective_static_transport_identity(
    configured: bool,
    transport_enabled: bool,
    has_discoverable_interface: bool,
) -> bool {
    configured || (!transport_enabled && has_discoverable_interface)
}

/// Extract discovery configuration from interface params, if `discoverable` is set.
fn extract_discovery_config(
    iface_name: &str,
    iface_type: &str,
    params: &std::collections::HashMap<String, String>,
) -> Option<crate::discovery::DiscoveryConfig> {
    let discoverable = params
        .get("discoverable")
        .and_then(|v| config::parse_bool_pub(v))
        .unwrap_or(false);
    if !discoverable {
        return None;
    }

    if iface_type == "TCPClientInterface" {
        log::error!(
            "Invalid interface discovery configuration for {}, aborting discovery announce",
            iface_name
        );
        return None;
    }

    let discovery_name = params
        .get("discovery_name")
        .cloned()
        .unwrap_or_else(|| iface_name.to_string());

    // Config value is in seconds. Min 300s (5min), default 21600s (6h).
    let announce_interval = params
        .get("announce_interval")
        .and_then(|v| v.parse::<u64>().ok())
        .map(|secs| secs.max(300))
        .unwrap_or(21600);

    // `as_float` since upstream 796ef2d4.
    let stamp_value = params
        .get("discovery_stamp_value")
        .and_then(|v| config::parse_config_float(v))
        .map(|v| rns_core::stamp::clamp_cost(v, rns_core::constants::STAMP_COST_CAP))
        .unwrap_or(crate::discovery::DEFAULT_STAMP_VALUE);

    let reachable_on = params.get("reachable_on").cloned();

    let listen_port = params
        .get("listen_port")
        .or_else(|| params.get("port"))
        .and_then(|v| v.parse().ok());

    let location_cmd = params.get("location_cmd").cloned();

    let latitude = params
        .get("latitude")
        .or_else(|| params.get("lat"))
        .and_then(|v| v.parse().ok());
    let longitude = params
        .get("longitude")
        .or_else(|| params.get("lon"))
        .and_then(|v| v.parse().ok());
    let height = params.get("height").and_then(|v| v.parse().ok());
    let operator_lxmf_address = params.get("discovery_lxmf_address").and_then(|value| {
        let parsed = parse_discovery_lxmf_address(value);
        if parsed.is_none() {
            log::error!(
                "Invalid interface discovery LXMF address for '{}': {}",
                iface_name,
                value
            );
        }
        parsed
    });

    Some(crate::discovery::DiscoveryConfig {
        discovery_name,
        announce_interval,
        stamp_value,
        reachable_on,
        interface_type: iface_type.to_string(),
        listen_port,
        location_cmd,
        latitude,
        longitude,
        height,
        operator_lxmf_address,
        // Invalid values were already rejected with the interface itself.
        stamp_costs: parse_interface_stamp_costs(params).unwrap_or_default(),
    })
}

fn default_discovery_runtime_config(
    interface_name: &str,
    interface_type: &str,
    listen_port: Option<u16>,
) -> crate::discovery::DiscoveryConfig {
    crate::discovery::DiscoveryConfig {
        stamp_costs: Default::default(),
        discovery_name: interface_name.to_string(),
        announce_interval: 21600,
        stamp_value: crate::discovery::DEFAULT_STAMP_VALUE,
        reachable_on: None,
        interface_type: interface_type.to_string(),
        listen_port,
        location_cmd: None,
        latitude: None,
        longitude: None,
        height: None,
        operator_lxmf_address: None,
    }
}

fn parse_discovery_lxmf_address(value: &str) -> Option<[u8; 16]> {
    if value.len() != 32 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        return None;
    }
    let mut address = [0u8; 16];
    for (index, byte) in address.iter_mut().enumerate() {
        *byte = u8::from_str_radix(&value[index * 2..index * 2 + 2], 16).ok()?;
    }
    Some(address)
}

fn discovery_runtime_ifac_fields(ifac: Option<&IfacConfig>) -> (Option<String>, Option<String>) {
    (
        ifac.and_then(|cfg| cfg.netname.clone()),
        ifac.and_then(|cfg| cfg.netkey.clone()),
    )
}

#[cfg(feature = "iface-backbone")]
pub(crate) fn backbone_discovery_runtime_from_interface(
    interface_name: &str,
    mode: &BackboneMode,
    discovery: Option<&crate::discovery::DiscoveryConfig>,
    transport_enabled: bool,
    ifac: Option<&IfacConfig>,
) -> Option<crate::driver::BackboneDiscoveryRuntimeHandle> {
    let config = match mode {
        BackboneMode::Server(config) => config,
        BackboneMode::Client(_, _) => return None,
    };

    let startup_config = discovery.cloned().unwrap_or_else(|| {
        default_discovery_runtime_config(
            interface_name,
            "BackboneInterface",
            Some(config.listen_port),
        )
    });
    let (ifac_netname, ifac_netkey) = discovery_runtime_ifac_fields(ifac);

    Some(crate::driver::BackboneDiscoveryRuntimeHandle::from_parts(
        config.name.clone(),
        startup_config,
        transport_enabled,
        ifac_netname,
        ifac_netkey,
        discovery.is_some(),
    ))
}

#[cfg(feature = "iface-tcp")]
pub(crate) fn tcp_server_discovery_runtime_from_interface(
    interface_name: &str,
    config: &crate::interface::tcp_server::TcpServerConfig,
    discovery: Option<&crate::discovery::DiscoveryConfig>,
    transport_enabled: bool,
    ifac: Option<&IfacConfig>,
) -> crate::driver::TcpServerDiscoveryRuntimeHandle {
    let startup_config = discovery.cloned().unwrap_or_else(|| {
        default_discovery_runtime_config(
            interface_name,
            "TCPServerInterface",
            Some(config.listen_port),
        )
    });
    let (ifac_netname, ifac_netkey) = discovery_runtime_ifac_fields(ifac);

    crate::driver::TcpServerDiscoveryRuntimeHandle::from_parts(
        config.name.clone(),
        startup_config,
        transport_enabled,
        ifac_netname,
        ifac_netkey,
        discovery.is_some(),
    )
}

pub(crate) fn ifac_runtime_from_config(
    ifac: Option<&IfacConfig>,
    default_size: usize,
) -> crate::driver::IfacRuntimeConfig {
    crate::driver::IfacRuntimeConfig::from_parts(
        ifac.and_then(|cfg| cfg.netname.clone()),
        ifac.and_then(|cfg| cfg.netkey.clone()),
        ifac.map(|cfg| cfg.size).unwrap_or(default_size),
    )
}

pub(crate) fn discoverable_interface_from_config(
    interface_name: &str,
    discovery: &crate::discovery::DiscoveryConfig,
    transport_enabled: bool,
    ifac: Option<&IfacConfig>,
) -> crate::discovery::DiscoverableInterface {
    crate::discovery::DiscoverableInterface {
        interface_name: interface_name.to_string(),
        config: discovery.clone(),
        transport_enabled,
        ifac_netname: ifac.and_then(|cfg| cfg.netname.clone()),
        ifac_netkey: ifac.and_then(|cfg| cfg.netkey.clone()),
    }
}

pub(crate) fn derive_ifac_state(
    ifac: Option<&IfacConfig>,
    interface_name: &str,
) -> io::Result<Option<crate::ifac::IfacState>> {
    let Some(ifac) = ifac else {
        return Ok(None);
    };
    if ifac.netname.is_none() && ifac.netkey.is_none() {
        return Ok(None);
    }

    ifac::derive_ifac(ifac.netname.as_deref(), ifac.netkey.as_deref(), ifac.size)
        .map(Some)
        .map_err(|err| {
            io::Error::other(format!(
                "failed to derive IFAC for {}: {}",
                interface_name, err
            ))
        })
}

pub(crate) struct StartedInterface<'a> {
    pub(crate) driver: &'a mut Driver,
    pub(crate) tx: &'a EventSender,
    pub(crate) queue_capacity: usize,
    pub(crate) id: rns_core::transport::types::InterfaceId,
    pub(crate) info: rns_core::transport::types::InterfaceInfo,
    pub(crate) writer: Box<dyn crate::interface::Writer>,
    pub(crate) interface_type_name: String,
    pub(crate) ifac_state: Option<crate::ifac::IfacState>,
    pub(crate) ifac_runtime: &'a crate::driver::IfacRuntimeConfig,
    pub(crate) stamp_costs: rns_core::transport::types::InterfaceStampCosts,
}

pub(crate) fn register_started_interface(params: StartedInterface<'_>) {
    let StartedInterface {
        driver,
        tx,
        queue_capacity,
        id,
        mut info,
        writer,
        interface_type_name,
        ifac_state,
        ifac_runtime,
        stamp_costs,
    } = params;
    info.stamp_costs = stamp_costs;
    driver.apply_announce_rate_defaults(&mut info);
    driver.apply_ingress_control_defaults(&mut info);
    let (writer, async_writer_metrics) =
        crate::interface::wrap_async_writer(writer, id, &info.name, tx.clone(), queue_capacity);
    driver.register_interface_runtime_defaults(&info);
    driver.register_interface_ifac_runtime(&info.name, ifac_runtime.clone());
    driver.engine.register_interface(info.clone());
    driver.interfaces.insert(
        id,
        InterfaceEntry {
            id,
            info,
            writer,
            async_writer_metrics: Some(async_writer_metrics),
            enabled: true,
            online: false,
            dynamic: false,
            ifac: ifac_state,
            stats: InterfaceStats {
                started: time::now(),
                ..Default::default()
            },
            interface_type: interface_type_name,
            send_retry_at: None,
            send_retry_backoff: Duration::ZERO,
        },
    );
}

/// Sources and startup defaults needed to reread a named interface from disk.
pub(crate) struct InterfaceManagementConfig {
    pub(crate) config_file: std::path::PathBuf,
    pub(crate) storage_dir: std::path::PathBuf,
    pub(crate) initial_ids: HashMap<String, rns_core::transport::types::InterfaceId>,
    pub(crate) queue_capacity: usize,
    pub(crate) transport_enabled: bool,
    pub(crate) underlay_mark: Option<u32>,
}

/// Top-level node configuration.
pub struct NodeConfig {
    pub transport_enabled: bool,
    pub static_transport_identity: bool,
    pub local_hops_delta: bool,
    pub identity: Option<Identity>,
    /// Interface configurations (parsed via registry factories).
    pub interfaces: Vec<InterfaceConfig>,
    /// Enable shared instance server for local clients (rns-ctl, etc.)
    pub share_instance: bool,
    /// Instance name for Unix socket namespace (default: "default").
    pub instance_name: String,
    /// Shared instance port for local client connections (default 37428).
    pub shared_instance_port: u16,
    /// RPC control port (default 37429). Only used when share_instance is true.
    pub rpc_port: u16,
    /// Cache directory for announce cache. If None, announce caching is disabled.
    pub cache_dir: Option<std::path::PathBuf>,
    /// Store for received ratchets. If None, received ratchets are not persisted or used.
    pub ratchet_store: Option<Arc<dyn storage::RatchetStore>>,
    /// TTL for received ratchets.
    pub ratchet_expiry: Duration,
    /// Remote management configuration.
    pub management: crate::management::ManagementConfig,
    /// Port to run the STUN probe server on (for facilitator nodes).
    pub probe_port: Option<u16>,
    /// Addresses of STUN/RNSP probe servers (tried sequentially with failover).
    pub probe_addrs: Vec<std::net::SocketAddr>,
    /// Protocol for endpoint discovery: "rnsp" (default) or "stun".
    pub probe_protocol: rns_core::holepunch::ProbeProtocol,
    /// Policy for incoming direct-connect proposals.
    pub direct_connect_policy: crate::event::HolePunchPolicy,
    /// Accept direct-connect endpoints in private, shared, link-local,
    /// and unique-local ranges (LAN-only deployments).
    pub direct_connect_allow_private_endpoints: bool,
    /// Network interface to bind outbound sockets to (e.g. "usb0").
    pub device: Option<String>,
    /// Linux `SO_MARK` value applied to Reticulum IP underlay sockets.
    pub underlay_mark: Option<u32>,
    /// Hook configurations loaded from the config file.
    pub hooks: Vec<config::ParsedHook>,
    /// Enable interface discovery.
    pub discover_interfaces: bool,
    /// Mode assigned to auto-connected discovered interfaces, when configured.
    pub autoconnect_interface_mode: Option<u8>,
    /// Gravity assigned to auto-connected discovered interfaces.
    pub autoconnect_interface_gravity: i64,
    /// Allow auto-connected interfaces to propagate announces to internal interfaces.
    pub autoconnect_announces_to_internal: bool,
    /// Allow auto-connecting discovered interfaces from unverified implementations.
    pub autoconnect_unverified_implementations: bool,
    /// Minimum stamp value for accepting discovered interfaces (default: 16).
    pub discovery_required_value: Option<rns_core::stamp::StampCost>,
    /// Respond to probe packets with automatic proof (like Python's respond_to_probes).
    pub respond_to_probes: bool,
    /// Accept an announce with strictly fewer hops even when the random_blob
    /// is a duplicate of the existing path entry.  Default `false` preserves
    /// Python-compatible anti-replay behaviour.
    pub prefer_shorter_path: bool,
    /// Maximum number of alternative paths stored per destination.
    /// Default 1 (single path, backward-compatible).
    pub max_paths_per_destination: usize,
    /// Maximum number of packet hashes retained for duplicate suppression.
    pub packet_hashlist_max_entries: usize,
    /// Startup allocation policy for the packet hashlist payload pages.
    pub packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation,
    /// Maximum number of discovery path-request tags remembered.
    pub max_discovery_pr_tags: usize,
    /// Maximum number of destination hashes retained in the live path table.
    pub max_path_destinations: usize,
    /// Maximum number of retained tunnel-known destinations.
    pub max_tunnel_destinations_total: usize,
    /// TTL for recalled known destinations without an active path.
    pub known_destinations_ttl: Duration,
    /// Maximum number of recalled known destinations retained.
    pub known_destinations_max_entries: usize,
    /// TTL for announce retransmission state.
    pub announce_table_ttl: Duration,
    /// Maximum retained bytes for announce retransmission state.
    pub announce_table_max_bytes: usize,
    /// Maximum queued events awaiting driver processing.
    pub driver_event_queue_capacity: usize,
    /// Maximum queued outbound frames per interface writer worker.
    pub interface_writer_queue_capacity: usize,
    /// Default announce-rate controls applied to interfaces when transport is enabled.
    pub announce_rate_defaults: AnnounceRateDefaults,
    /// Default ingress/egress controls applied to interfaces when transport is enabled.
    pub ingress_control_defaults: rns_core::transport::types::IngressControlConfig,
    /// Outbound Backbone peer-pool settings. Disabled when `None`.
    #[cfg(feature = "iface-backbone")]
    pub backbone_peer_pool: Option<BackbonePeerPoolSettings>,
    /// Whether the announce signature verification cache is enabled.
    pub announce_sig_cache_enabled: bool,
    /// Maximum entries in the announce signature verification cache.
    pub announce_sig_cache_max_entries: usize,
    /// TTL for announce signature cache entries.
    pub announce_sig_cache_ttl: Duration,
    /// Custom interface registry. If `None`, uses `InterfaceRegistry::with_builtins()`.
    pub registry: Option<crate::interface::registry::InterfaceRegistry>,
    /// If true, a single interface failing to start will abort the entire node.
    /// If false (default), the error is logged and remaining interfaces continue.
    pub panic_on_interface_error: bool,
    /// Threads that verify and sign link packet proofs off the driver thread
    /// (`Auto`: half the CPUs the process may use, from one to four).
    pub link_proof_verify_threads: crate::ProofVerifyThreads,
    /// Stamp generation, enforcement, costs and timeouts: the `[stamps]`
    /// section, which [`RnsNode::from_config`] reads from the config file.
    /// Upstream's `Reticulum(enable_stamping=True)` corresponds to setting
    /// `stamp_destinations`, `stamp_prs` and `stamp_lrs`, and
    /// `enforce_stamps=True` to `stamps_enforced` and `lr_stamps_enforced`.
    pub stamps: config::StampsSection,
    /// External provider bridge for hook-emitted events.
    #[cfg(feature = "hooks")]
    pub provider_bridge: Option<crate::provider_bridge::ProviderBridgeConfig>,
}

struct NodeQueueConfig {
    rpc_auth_key: Option<[u8; 32]>,
    announce_max_entries: usize,
    announce_max_interfaces: usize,
    announce_max_bytes: usize,
    announce_ttl_secs: f64,
    announce_overflow_policy: AnnounceQueueOverflowPolicy,
    inbound_capacities: crate::event::InboundQueueCapacities,
    link_mtu_discovery: bool,
    /// The `[filter]` config section.
    filter: config::FilterSection,
    /// `[reticulum] hop_limiter`.
    hop_limiter: u8,
}

impl Default for NodeQueueConfig {
    fn default() -> Self {
        Self {
            rpc_auth_key: None,
            announce_max_entries: 256,
            announce_max_interfaces: 1024,
            announce_max_bytes: 256 * 1024,
            announce_ttl_secs: 30.0,
            announce_overflow_policy: AnnounceQueueOverflowPolicy::DropWorst,
            inbound_capacities: crate::event::InboundQueueCapacities::default(),
            link_mtu_discovery: true,
            filter: config::FilterSection::default(),
            hop_limiter: rns_core::constants::HOP_LIMITER,
        }
    }
}

impl Default for NodeConfig {
    fn default() -> Self {
        Self {
            transport_enabled: false,
            static_transport_identity: false,
            local_hops_delta: false,
            identity: None,
            interfaces: Vec::new(),
            share_instance: false,
            instance_name: "default".into(),
            shared_instance_port: 37428,
            rpc_port: 0,
            cache_dir: None,
            ratchet_store: None,
            ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
            management: Default::default(),
            probe_port: None,
            probe_addrs: vec![],
            probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
            direct_connect_policy: crate::event::HolePunchPolicy::IdentifiedOnly,
            direct_connect_allow_private_endpoints: false,
            device: None,
            underlay_mark: None,
            hooks: Vec::new(),
            discover_interfaces: false,
            autoconnect_interface_mode: None,
            autoconnect_interface_gravity: 0,
            autoconnect_announces_to_internal: false,
            autoconnect_unverified_implementations: false,
            discovery_required_value: None,
            respond_to_probes: false,
            prefer_shorter_path: false,
            max_paths_per_destination: 1,
            packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
            packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation::Lazy,
            max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
            max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
            max_tunnel_destinations_total: usize::MAX,
            known_destinations_ttl: Duration::from_secs(48 * 60 * 60),
            known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
            announce_table_ttl: Duration::from_secs(rns_core::constants::ANNOUNCE_TABLE_TTL as u64),
            announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
            driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
            interface_writer_queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
            announce_rate_defaults: AnnounceRateDefaults::default(),
            ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(),
            #[cfg(feature = "iface-backbone")]
            backbone_peer_pool: None,
            announce_sig_cache_enabled: true,
            announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
            announce_sig_cache_ttl: Duration::from_secs(
                rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
            ),
            registry: None,
            panic_on_interface_error: false,
            link_proof_verify_threads: Default::default(),
            stamps: config::StampsSection::default(),
            #[cfg(feature = "hooks")]
            provider_bridge: None,
        }
    }
}

/// IFAC configuration for an interface.
pub struct IfacConfig {
    pub netname: Option<String>,
    pub netkey: Option<String>,
    pub size: usize,
}

/// Interface configuration, parsed via an [`InterfaceFactory`] from the registry.
pub struct InterfaceConfig {
    pub name: String,
    pub type_name: String,
    pub config_data: Box<dyn crate::interface::InterfaceConfigData>,
    pub mode: u8,
    pub gravity: i64,
    pub recursive_prs: bool,
    pub announces_from_internal: bool,
    pub announces_to_internal: Option<bool>,
    pub ingress_control: rns_core::transport::types::IngressControlConfig,
    /// Per-interface `destination_cost` and `pr_cost` (upstream 9b464382).
    pub stamp_costs: rns_core::transport::types::InterfaceStampCosts,
    pub ifac: Option<IfacConfig>,
    pub discovery: Option<crate::discovery::DiscoveryConfig>,
}

/// Parse the per-interface `destination_cost` and `pr_cost` options
/// (upstream 9b464382): `as_float`, clamped to `0..=STAMP_COST_CAP`.
pub(crate) fn parse_interface_stamp_costs(
    params: &HashMap<String, String>,
) -> Result<rns_core::transport::types::InterfaceStampCosts, String> {
    let cost = |key: &str| -> Result<Option<rns_core::stamp::StampCost>, String> {
        let Some(value) = params.get(key) else {
            return Ok(None);
        };
        let value = config::parse_config_float(value)
            .ok_or_else(|| format!("{key} must be a number, got '{value}'"))?;
        Ok(Some(rns_core::stamp::clamp_cost(
            value,
            rns_core::constants::STAMP_COST_CAP,
        )))
    };
    Ok(rns_core::transport::types::InterfaceStampCosts {
        destination_cost: cost("destination_cost")?,
        pr_cost: cost("pr_cost")?,
    })
}

pub(crate) fn parse_managed_interface(
    iface: &config::ParsedInterface,
    parsed: &config::RnsConfig,
    storage_dir: &Path,
    id: rns_core::transport::types::InterfaceId,
    registry: &crate::interface::registry::InterfaceRegistry,
) -> io::Result<InterfaceConfig> {
    let factory = registry.get(&iface.interface_type).ok_or_else(|| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            format!("unsupported interface type '{}'", iface.interface_type),
        )
    })?;
    let discoverable = iface
        .params
        .get("discoverable")
        .and_then(|value| config::parse_bool_pub(value))
        .unwrap_or(false);
    let ignore_warnings = iface
        .params
        .get("ignore_config_warnings")
        .and_then(|value| config::parse_bool_pub(value))
        .unwrap_or(false);
    let mode = normalize_discovery_mode(
        &iface.interface_type,
        parse_interface_mode(&iface.mode),
        discoverable,
        ignore_warnings,
    );
    let ifac = extract_ifac_config(&iface.params, factory.default_ifac_size());
    let discovery = extract_discovery_config(&iface.name, &iface.interface_type, &iface.params);
    let ingress_control = parse_ingress_control_config(&iface.interface_type, &iface.params)
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
    let stamp_costs = parse_interface_stamp_costs(&iface.params)
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
    let recursive_prs = iface
        .params
        .get("recursive_prs")
        .and_then(|value| config::parse_bool_pub(value))
        .unwrap_or(false);
    let announces_from_internal = iface
        .params
        .get("announces_from_internal")
        .and_then(|value| config::parse_bool_pub(value))
        .unwrap_or(true);
    let announces_to_internal = iface
        .params
        .get("announces_to_internal")
        .and_then(|value| config::parse_bool_pub(value));
    let gravity = parse_interface_gravity(
        iface.params.get("gravity").map(String::as_str),
        parsed.reticulum.default_gravity,
    );
    let mut params = iface.params.clone();
    params
        .entry("storage_dir".to_string())
        .or_insert_with(|| storage_dir.to_string_lossy().to_string());
    if let Some(device) = &parsed.reticulum.device {
        params
            .entry("device".to_string())
            .or_insert_with(|| device.clone());
    }
    let config_data = factory
        .parse_config_section(
            &iface.name,
            id,
            crate::interface::ConfigSection {
                params: &params,
                children: &iface.subinterfaces,
            },
        )
        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
    Ok(InterfaceConfig {
        name: iface.name.clone(),
        type_name: iface.interface_type.clone(),
        config_data,
        mode,
        gravity,
        recursive_prs,
        announces_from_internal,
        announces_to_internal,
        ingress_control,
        stamp_costs,
        ifac,
        discovery,
    })
}

use crate::event::{QueryRequest, QueryResponse};

/// Error returned when the driver thread has shut down.
#[derive(Debug)]
pub struct SendError;

impl std::fmt::Display for SendError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "driver shut down")
    }
}

impl std::error::Error for SendError {}

/// A channel send can be temporarily blocked without the link having failed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ChannelSendError {
    NotReady,
    Rejected(String),
    Unavailable,
}

impl From<String> for ChannelSendError {
    fn from(error: String) -> Self {
        if error == rns_core::channel::ChannelError::NotReady.to_string() {
            Self::NotReady
        } else {
            Self::Rejected(error)
        }
    }
}

impl std::fmt::Display for ChannelSendError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotReady => write!(f, "channel is not ready to send"),
            Self::Rejected(error) => write!(f, "{error}"),
            Self::Unavailable => write!(f, "driver is stopped or draining"),
        }
    }
}

impl std::error::Error for ChannelSendError {}

/// A running RNS node.
pub struct RnsNode {
    tx: EventSender,
    driver_handle: Option<JoinHandle<()>>,
    verify_handle: Option<JoinHandle<()>>,
    verify_shutdown: Arc<crate::driver::WorkSignal>,
    persist_handle: Option<JoinHandle<()>>,
    persist_stop: Option<std::sync::mpsc::Sender<()>>,
    rpc_server: Option<crate::rpc::RpcServer>,
    #[cfg(target_os = "linux")]
    rpc_unix_server: Option<crate::rpc::RpcServer>,
    tick_interval_ms: Arc<AtomicU64>,
    /// The next [`SendTicket`](crate::SendTicket) number.
    next_send_ticket: AtomicU64,
    /// Whether deferred sends are encrypted on the caller's thread: when
    /// this process may run on one CPU only, another thread would share it.
    encrypt_deferred_inline: bool,
    /// Cleared by the driver when it stops accepting new work.
    accepting_work: Arc<AtomicBool>,
    #[allow(dead_code)]
    probe_server: Option<crate::holepunch::probe::ProbeServerHandle>,
    known_destinations_path: Option<std::path::PathBuf>,
    known_destinations_store: Option<Arc<crate::storage_sqlite::KnownDestinationsStore>>,
    stamp_store: Option<crate::stamps::DestinationStampStore>,
    /// `Transport.STAMP_DESTINATIONS`: whether announces from this node carry
    /// a destination stamp. Defaults to the compile-time constant; shared
    /// with the driver, which applies the `global.stamp_destinations`
    /// runtime config key.
    stamp_destinations: Arc<AtomicBool>,
    stamp_targets: crate::stamper::StampTargets,
    /// Shared with the driver; grinds stamps off the calling thread.
    stamper: Option<Arc<crate::stamper::Stamper>>,
    transport_state_dir: Option<std::path::PathBuf>,
    ratchet_store: Option<Arc<dyn storage::RatchetStore>>,
    ratchet_expiry: Duration,
    /// Destination types this application needs (upstream 825c2e19).
    #[cfg(feature = "iface-local")]
    pub(crate) destination_types: crate::shared_client::DestinationTypeRequester,
}

impl RnsNode {
    /// Start the node from a config file path.
    /// If `config_path` is None, uses `~/.reticulum/`.
    pub fn from_config(
        config_path: Option<&Path>,
        callbacks: Box<dyn Callbacks>,
    ) -> io::Result<Self> {
        let config_dir = storage::resolve_config_dir(config_path);
        Self::from_config_dirs(&config_dir, &config_dir, None, false, callbacks)
    }

    /// Start an application-private node from a dedicated configuration and
    /// state root.
    ///
    /// Private nodes never expose the shared-instance or management servers,
    /// never load hooks, fail on unsupported/malformed enabled interfaces, and
    /// force runtime interface failures to stop startup. `underlay_mark`
    /// overrides the value in the Reticulum configuration.
    pub fn from_private_config(
        config_dir: &Path,
        state_dir: &Path,
        underlay_mark: Option<u32>,
        callbacks: Box<dyn Callbacks>,
    ) -> io::Result<Self> {
        Self::from_config_dirs(config_dir, state_dir, underlay_mark, true, callbacks)
    }

    fn from_config_dirs(
        config_dir: &Path,
        state_dir: &Path,
        underlay_mark: Option<u32>,
        private: bool,
        callbacks: Box<dyn Callbacks>,
    ) -> io::Result<Self> {
        let paths = storage::ensure_storage_dirs(state_dir)?;
        let known_destinations_path = paths.storage.join("known_destinations");

        // Parse config file
        let config_file = config_dir.join("config");
        let config_content = if config_file.exists() {
            std::fs::read_to_string(&config_file)?
        } else {
            String::new()
        };
        let rns_config = config::parse(&config_content)
            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?;
        let link_mtu_discovery = config::parse_link_mtu_discovery(&config_content)
            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?;

        // Load or create identity
        let identity = if let Some(ref id_path_str) = rns_config.reticulum.network_identity {
            let id_path = std::path::PathBuf::from(id_path_str);
            if id_path.exists() {
                storage::load_identity(&id_path)?
            } else {
                let id = Identity::new(&mut OsRng);
                storage::save_identity(&id, &id_path)?;
                id
            }
        } else {
            storage::load_or_create_identity(&paths.identities)?
        };
        // Python clients derive their RPC key from this persisted identity, even
        // when their non-transport instance uses an ephemeral network identity.
        let rpc_auth_key = if !private && rns_config.reticulum.share_instance {
            let transport_identity_path = storage::transport_identity_path(&paths);
            let rpc_identity = if transport_identity_path.exists() {
                storage::load_identity(&transport_identity_path)?
            } else {
                storage::save_identity(&identity, &transport_identity_path)?;
                Identity::from_private_key(
                    &identity
                        .get_private_key()
                        .expect("node identity has private key"),
                )
            };
            Some(crate::rpc::derive_auth_key(
                &rpc_identity
                    .get_private_key()
                    .expect("RPC identity has private key"),
            ))
        } else {
            None
        };

        // Build interface configs from parsed config using registry
        let registry = crate::interface::registry::InterfaceRegistry::with_builtins();
        let mut interface_configs = Vec::new();
        let mut initial_ids = HashMap::new();
        let mut next_id_val = 1u64;

        for iface in &rns_config.interfaces {
            if !iface.enabled {
                continue;
            }

            let iface_id = rns_core::transport::types::InterfaceId(next_id_val);
            next_id_val += 1;

            let factory = match registry.get(&iface.interface_type) {
                Some(f) => f,
                None => {
                    if private {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidData,
                            format!(
                                "unsupported enabled interface type '{}' for '{}'",
                                iface.interface_type, iface.name
                            ),
                        ));
                    }
                    log::warn!(
                        "Unsupported interface type '{}' for '{}'",
                        iface.interface_type,
                        iface.name,
                    );
                    continue;
                }
            };

            let mut iface_mode = parse_interface_mode(&iface.mode);

            let discoverable = iface
                .params
                .get("discoverable")
                .and_then(|value| config::parse_bool_pub(value))
                .unwrap_or(false);
            let ignore_config_warnings = iface
                .params
                .get("ignore_config_warnings")
                .and_then(|value| config::parse_bool_pub(value))
                .unwrap_or(false);
            let normalized_mode = normalize_discovery_mode(
                &iface.interface_type,
                iface_mode,
                discoverable,
                ignore_config_warnings,
            );
            if normalized_mode != iface_mode {
                log::info!(
                    "Interface '{}' has discovery enabled, auto-configuring mode to {}",
                    iface.name,
                    if normalized_mode == rns_core::constants::MODE_ACCESS_POINT {
                        "ACCESS_POINT"
                    } else {
                        "GATEWAY"
                    }
                );
                iface_mode = normalized_mode;
            }

            let default_ifac_size = factory.default_ifac_size();
            let ifac_config = extract_ifac_config(&iface.params, default_ifac_size);
            let discovery_config =
                extract_discovery_config(&iface.name, &iface.interface_type, &iface.params);
            let ingress_control =
                match parse_ingress_control_config(&iface.interface_type, &iface.params) {
                    Ok(config) => config,
                    Err(e) => {
                        if private {
                            return Err(io::Error::new(
                                io::ErrorKind::InvalidData,
                                format!("invalid ingress control for '{}': {e}", iface.name),
                            ));
                        }
                        log::warn!(
                            "Failed to parse ingress control config for '{}': {}",
                            iface.name,
                            e
                        );
                        continue;
                    }
                };
            let stamp_costs = match parse_interface_stamp_costs(&iface.params) {
                Ok(costs) => costs,
                Err(e) => {
                    if private {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidData,
                            format!("invalid stamp cost for '{}': {e}", iface.name),
                        ));
                    }
                    log::warn!("Failed to parse stamp costs for '{}': {}", iface.name, e);
                    continue;
                }
            };
            let recursive_prs = iface
                .params
                .get("recursive_prs")
                .and_then(|v| config::parse_bool_pub(v))
                .unwrap_or(false);
            let announces_from_internal = iface
                .params
                .get("announces_from_internal")
                .and_then(|v| config::parse_bool_pub(v))
                .unwrap_or(true);
            let gravity = parse_interface_gravity(
                iface.params.get("gravity").map(String::as_str),
                rns_config.reticulum.default_gravity,
            );
            let announces_to_internal = iface
                .params
                .get("announces_to_internal")
                .and_then(|v| config::parse_bool_pub(v));

            // Inject storage_dir for I2P (and any future factories that need it)
            let mut params = iface.params.clone();
            if !params.contains_key("storage_dir") {
                params.insert(
                    "storage_dir".to_string(),
                    paths.storage.to_string_lossy().to_string(),
                );
            }
            // Inject device for TCP client
            if let Some(ref device) = rns_config.reticulum.device {
                if !params.contains_key("device") {
                    params.insert("device".to_string(), device.clone());
                }
            }

            let config_data = match factory.parse_config_section(
                &iface.name,
                iface_id,
                crate::interface::ConfigSection {
                    params: &params,
                    children: &iface.subinterfaces,
                },
            ) {
                Ok(data) => data,
                Err(e) => {
                    if private {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidData,
                            format!("invalid interface '{}': {e}", iface.name),
                        ));
                    }
                    log::warn!("Failed to parse config for '{}': {}", iface.name, e);
                    continue;
                }
            };

            interface_configs.push(InterfaceConfig {
                name: iface.name.clone(),
                type_name: iface.interface_type.clone(),
                config_data,
                mode: iface_mode,
                gravity,
                recursive_prs,
                announces_from_internal,
                announces_to_internal,
                ingress_control,
                stamp_costs,
                ifac: ifac_config,
                discovery: discovery_config,
            });
            initial_ids.insert(iface.name.clone(), iface_id);
        }

        // Parse management config
        let mut mgmt_allowed = Vec::new();
        for hex_hash in &rns_config.reticulum.remote_management_allowed {
            if hex_hash.len() == 32 {
                if let Ok(bytes) = (0..hex_hash.len())
                    .step_by(2)
                    .map(|i| u8::from_str_radix(&hex_hash[i..i + 2], 16))
                    .collect::<Result<Vec<u8>, _>>()
                {
                    if bytes.len() == 16 {
                        let mut h = [0u8; 16];
                        h.copy_from_slice(&bytes);
                        mgmt_allowed.push(h);
                    }
                } else {
                    log::warn!("Invalid hex in remote_management_allowed: {}", hex_hash);
                }
            } else {
                log::warn!(
                    "Invalid entry in remote_management_allowed (expected 32 hex chars, got {}): {}",
                    hex_hash.len(), hex_hash,
                );
            }
        }

        // Parse probe_addr (comma-separated list of SocketAddr)
        let probe_addrs: Vec<std::net::SocketAddr> = rns_config
            .reticulum
            .probe_addr
            .as_ref()
            .map(|s| {
                s.split(',')
                    .filter_map(|entry| {
                        let trimmed = entry.trim();
                        if trimmed.is_empty() {
                            return None;
                        }
                        trimmed
                            .parse::<std::net::SocketAddr>()
                            .map_err(|e| {
                                log::warn!("Invalid probe_addr entry '{}': {}", trimmed, e);
                                e
                            })
                            .ok()
                    })
                    .collect()
            })
            .unwrap_or_default();

        // Parse probe_protocol (default: rnsp)
        let probe_protocol = match rns_config
            .reticulum
            .probe_protocol
            .as_deref()
            .map(|s| s.to_lowercase())
        {
            Some(ref s) if s == "stun" => rns_core::holepunch::ProbeProtocol::Stun,
            _ => rns_core::holepunch::ProbeProtocol::Rnsp,
        };

        let known_destinations_db = crate::storage_sqlite::default_path(&paths.storage);
        let mut known_destinations_store: Option<
            Arc<crate::storage_sqlite::KnownDestinationsStore>,
        > = None;
        let (known_destinations, ratchet_store): (
            std::collections::HashMap<[u8; 16], storage::KnownDestination>,
            Arc<dyn storage::RatchetStore>,
        ) = match crate::storage_sqlite::KnownDestinationsStore::open(&known_destinations_db, 16) {
            Ok(store) => {
                let store = Arc::new(store);
                let mut destinations: std::collections::HashMap<
                    [u8; 16],
                    storage::KnownDestination,
                > = store
                    .load_all()
                    .map(|entries| entries.into_iter().collect())
                    .unwrap_or_else(|err| {
                        log::warn!("failed to load known destinations from database: {}", err);
                        Default::default()
                    });
                if destinations.is_empty() && known_destinations_path.is_file() {
                    match store.import_legacy(&known_destinations_path) {
                        Ok(imported) if imported > 0 => {
                            log::info!(
                                "imported {} known destinations from legacy storage",
                                imported
                            );
                            let _ = std::fs::rename(
                                &known_destinations_path,
                                known_destinations_path.with_extension("imported"),
                            );
                            destinations = store
                                .load_all()
                                .map(|entries| entries.into_iter().collect())
                                .unwrap_or_default();
                        }
                        Ok(_) => {}
                        Err(err) => {
                            log::warn!("could not import legacy known destinations: {}", err)
                        }
                    }
                }
                let known_destination_hashes: HashSet<[u8; 16]> =
                    destinations.keys().copied().collect();
                if let Err(err) = store.import_legacy_ratchets(
                    &paths.ratchets,
                    time::now() - rns_config.reticulum.ratchet_expiry as f64,
                    &known_destination_hashes,
                ) {
                    log::warn!("could not import legacy ratchets: {}", err);
                }
                known_destinations_store = Some(Arc::clone(&store));
                (destinations, store)
            }
            Err(err) => {
                log::warn!(
                        "could not initialise known destinations database, falling back to legacy storage: {}",
                        err
                    );
                let destinations = match storage::load_known_destinations(&known_destinations_path)
                {
                    Ok(destinations) => destinations,
                    Err(err) if err.kind() == io::ErrorKind::NotFound => Default::default(),
                    Err(err) => {
                        log::warn!("failed to load known destinations: {}", err);
                        Default::default()
                    }
                };
                (
                    destinations,
                    Arc::new(storage::FsRatchetStore::new(paths.ratchets.clone())),
                )
            }
        };
        let known_destination_hashes: HashSet<[u8; 16]> =
            known_destinations.keys().copied().collect();
        match ratchet_store.cleanup(
            &known_destination_hashes,
            time::now(),
            rns_config.reticulum.ratchet_expiry as f64,
        ) {
            Ok(stats) if stats.processed > 0 => log::debug!(
                "Processed {} ratchets, not in use {}, removed {}",
                stats.processed,
                stats.not_known,
                stats.removed
            ),
            Ok(_) => {}
            Err(err) => log::warn!("failed to clean ratchets: {}", err),
        }

        let interface_management = if private || !rns_config.reticulum.enable_interface_management {
            None
        } else {
            Some(InterfaceManagementConfig {
                config_file: config_file.clone(),
                storage_dir: paths.storage.clone(),
                initial_ids,
                queue_capacity: rns_config.reticulum.interface_writer_queue_capacity,
                transport_enabled: rns_config.reticulum.enable_transport,
                underlay_mark: underlay_mark.or(rns_config.reticulum.underlay_mark),
            })
        };
        let node_config = NodeConfig {
            transport_enabled: rns_config.reticulum.enable_transport,
            static_transport_identity: rns_config.reticulum.static_transport_identity,
            local_hops_delta: rns_config.reticulum.local_hops_delta,
            identity: Some(identity),
            share_instance: !private && rns_config.reticulum.share_instance,
            instance_name: rns_config.reticulum.instance_name.clone(),
            shared_instance_port: rns_config.reticulum.shared_instance_port,
            rpc_port: rns_config.reticulum.instance_control_port,
            cache_dir: Some(paths.cache),
            ratchet_store: Some(Arc::clone(&ratchet_store)),
            ratchet_expiry: Duration::from_secs(rns_config.reticulum.ratchet_expiry),
            management: if private {
                crate::management::ManagementConfig::default()
            } else {
                crate::management::ManagementConfig {
                    enable_remote_management: rns_config.reticulum.enable_remote_management,
                    remote_management_allowed: mgmt_allowed,
                    publish_blackhole: rns_config.reticulum.publish_blackhole,
                }
            },
            probe_port: rns_config.reticulum.probe_port,
            probe_addrs,
            probe_protocol,
            direct_connect_policy: rns_config.reticulum.direct_connect_policy,
            direct_connect_allow_private_endpoints: rns_config
                .reticulum
                .direct_connect_allow_private_endpoints,
            device: rns_config.reticulum.device.clone(),
            underlay_mark: underlay_mark.or(rns_config.reticulum.underlay_mark),
            hooks: if private {
                Vec::new()
            } else {
                rns_config.hooks.clone()
            },
            discover_interfaces: rns_config.reticulum.discover_interfaces,
            autoconnect_interface_mode: rns_config
                .reticulum
                .autoconnect_interface_mode
                .as_deref()
                .and_then(parse_optional_interface_mode),
            autoconnect_interface_gravity: rns_config.reticulum.autoconnect_interface_gravity,
            autoconnect_announces_to_internal: rns_config
                .reticulum
                .autoconnect_announces_to_internal,
            autoconnect_unverified_implementations: rns_config
                .reticulum
                .autoconnect_unverified_implementations,
            discovery_required_value: rns_config.reticulum.required_discovery_value,
            respond_to_probes: rns_config.reticulum.respond_to_probes,
            prefer_shorter_path: rns_config.reticulum.prefer_shorter_path,
            max_paths_per_destination: rns_config.reticulum.max_paths_per_destination,
            packet_hashlist_max_entries: rns_config.reticulum.packet_hashlist_max_entries,
            packet_hashlist_allocation: rns_config.reticulum.packet_hashlist_allocation,
            max_discovery_pr_tags: rns_config.reticulum.max_discovery_pr_tags,
            max_path_destinations: rns_config.reticulum.max_path_destinations,
            max_tunnel_destinations_total: rns_config.reticulum.max_tunnel_destinations_total,
            known_destinations_ttl: Duration::from_secs(
                rns_config.reticulum.known_destinations_ttl,
            ),
            known_destinations_max_entries: rns_config.reticulum.known_destinations_max_entries,
            announce_table_ttl: Duration::from_secs(rns_config.reticulum.announce_table_ttl),
            announce_table_max_bytes: rns_config.reticulum.announce_table_max_bytes,
            driver_event_queue_capacity: rns_config.reticulum.driver_event_queue_capacity,
            interface_writer_queue_capacity: rns_config.reticulum.interface_writer_queue_capacity,
            announce_rate_defaults: AnnounceRateDefaults {
                target: rns_config.reticulum.default_ar_target,
                penalty: rns_config.reticulum.default_ar_penalty,
                grace: rns_config.reticulum.default_ar_grace,
            },
            ingress_control_defaults: rns_core::transport::types::IngressControlConfig {
                enabled: true,
                egress_enabled: rns_config.reticulum.default_egress_control,
                max_held_announces: rns_config.reticulum.default_ic_max_held_announces,
                burst_freq_new: rns_config.reticulum.default_ic_burst_freq_new,
                burst_freq: rns_config.reticulum.default_ic_burst_freq,
                pr_burst_freq_new: rns_config.reticulum.default_ic_pr_burst_freq_new,
                pr_burst_freq: rns_config.reticulum.default_ic_pr_burst_freq,
                egress_pr_freq: rns_config.reticulum.default_ec_pr_freq,
                new_time: rns_config.reticulum.default_ic_new_time,
                burst_hold: rns_config.reticulum.default_ic_burst_hold,
                burst_penalty: rns_config.reticulum.default_ic_burst_penalty,
                held_release_interval: rns_config.reticulum.default_ic_held_release_interval,
            },
            #[cfg(feature = "iface-backbone")]
            backbone_peer_pool: if rns_config.reticulum.backbone_peer_pool_max_connected > 0 {
                Some(BackbonePeerPoolSettings {
                    max_connected: rns_config.reticulum.backbone_peer_pool_max_connected,
                    failure_threshold: rns_config.reticulum.backbone_peer_pool_failure_threshold,
                    failure_window: Duration::from_secs(
                        rns_config.reticulum.backbone_peer_pool_failure_window,
                    ),
                    cooldown: Duration::from_secs(rns_config.reticulum.backbone_peer_pool_cooldown),
                })
            } else {
                None
            },
            announce_sig_cache_enabled: rns_config.reticulum.announce_sig_cache_enabled,
            announce_sig_cache_max_entries: rns_config.reticulum.announce_sig_cache_max_entries,
            announce_sig_cache_ttl: Duration::from_secs(
                rns_config.reticulum.announce_sig_cache_ttl,
            ),
            interfaces: interface_configs,
            registry: None,
            panic_on_interface_error: private || rns_config.reticulum.panic_on_interface_error,
            link_proof_verify_threads: Default::default(),
            stamps: rns_config.stamps.clone(),
            #[cfg(feature = "hooks")]
            provider_bridge: if !private && rns_config.reticulum.provider_bridge {
                Some(crate::provider_bridge::ProviderBridgeConfig {
                    enabled: true,
                    socket_path: rns_config
                        .reticulum
                        .provider_socket_path
                        .as_ref()
                        .map(std::path::PathBuf::from)
                        .unwrap_or_else(|| config_dir.join("provider.sock")),
                    queue_max_events: rns_config.reticulum.provider_queue_max_events,
                    queue_max_bytes: rns_config.reticulum.provider_queue_max_bytes,
                    overflow_policy: match rns_config.reticulum.provider_overflow_policy.as_str() {
                        "drop_oldest" => crate::provider_bridge::OverflowPolicy::DropOldest,
                        _ => crate::provider_bridge::OverflowPolicy::DropNewest,
                    },
                    node_instance: rns_config.reticulum.instance_name.clone(),
                })
            } else {
                None
            },
        };

        let mut node = Self::start_with_queue_config(
            node_config,
            callbacks,
            NodeQueueConfig {
                rpc_auth_key,
                announce_max_entries: rns_config.reticulum.announce_queue_max_entries,
                announce_max_interfaces: rns_config.reticulum.announce_queue_max_interfaces,
                announce_max_bytes: rns_config.reticulum.announce_queue_max_bytes,
                announce_ttl_secs: rns_config.reticulum.announce_queue_ttl as f64,
                announce_overflow_policy: match rns_config
                    .reticulum
                    .announce_queue_overflow_policy
                    .as_str()
                {
                    "drop_newest" => AnnounceQueueOverflowPolicy::DropNewest,
                    "drop_oldest" => AnnounceQueueOverflowPolicy::DropOldest,
                    _ => AnnounceQueueOverflowPolicy::DropWorst,
                },
                inbound_capacities: rns_config.reticulum.inbound_queue_capacities,
                link_mtu_discovery,
                filter: rns_config.filter.clone(),
                hop_limiter: rns_config.reticulum.hop_limiter,
            },
            interface_management,
        )?;

        node.known_destinations_path = Some(known_destinations_path.clone());
        node.known_destinations_store = known_destinations_store;
        node.set_stamp_store(crate::stamps::DestinationStampStore::new(&paths.storage));
        for (dest_hash, known) in known_destinations {
            let _ = node.query(QueryRequest::RestoreKnownDestination(
                crate::event::KnownDestinationEntry {
                    dest_hash,
                    identity_hash: known.identity_hash,
                    public_key: known.public_key,
                    app_data: known.app_data,
                    hops: known.hops,
                    received_at: known.received_at,
                    receiving_interface: rns_core::transport::types::InterfaceId(
                        known.receiving_interface,
                    ),
                    was_used: known.was_used,
                    last_used_at: known.last_used_at,
                    retained: known.retained,
                    stamp: known
                        .stamp
                        .as_deref()
                        .and_then(|stamp| stamp.try_into().ok()),
                    stamp_value: known.stamp_value,
                    lr_cost: known.lr_cost,
                },
            ));
        }

        Ok(node)
    }

    /// Start the node. Connects all interfaces, starts driver and timer threads.
    pub fn start(config: NodeConfig, callbacks: Box<dyn Callbacks>) -> io::Result<Self> {
        crate::interface::validate_underlay_mark(config.underlay_mark)?;
        let inbound_queue_capacities = crate::event::InboundQueueCapacities::from_shared_capacity(
            config.driver_event_queue_capacity,
        );
        Self::start_with_queue_config(
            config,
            callbacks,
            NodeQueueConfig {
                inbound_capacities: inbound_queue_capacities,
                ..NodeQueueConfig::default()
            },
            None,
        )
    }

    fn start_with_queue_config(
        config: NodeConfig,
        callbacks: Box<dyn Callbacks>,
        queue_config: NodeQueueConfig,
        interface_management: Option<InterfaceManagementConfig>,
    ) -> io::Result<Self> {
        let storage_dir = config.cache_dir.as_ref().map(|cache_dir| {
            cache_dir
                .parent()
                .unwrap_or_else(|| std::path::Path::new("."))
                .join("storage")
        });
        let transport_state_dir = if config.transport_enabled {
            storage_dir.clone()
        } else {
            None
        };
        let identity = config.identity.unwrap_or_else(|| Identity::new(&mut OsRng));
        let rpc_auth_key = queue_config.rpc_auth_key.unwrap_or_else(|| {
            crate::rpc::derive_auth_key(
                &identity
                    .get_private_key()
                    .expect("node identity has private key"),
            )
        });
        let has_discoverable_interface = config
            .interfaces
            .iter()
            .any(|iface| iface.discovery.is_some());
        let static_transport_identity = effective_static_transport_identity(
            config.static_transport_identity,
            config.transport_enabled,
            has_discoverable_interface,
        );
        if static_transport_identity && !config.static_transport_identity {
            log::warn!(
                "Discoverable interface was configured, enabling static transport identity on non-transport instance"
            );
        }
        let transport_identity = if config.transport_enabled || static_transport_identity {
            Identity::from_private_key(
                &identity
                    .get_private_key()
                    .expect("node identity must expose a private key"),
            )
        } else {
            Identity::new(&mut OsRng)
        };

        let local_hops_delta = if config.local_hops_delta {
            let mut byte = [0u8; 1];
            OsRng.fill_bytes(&mut byte);
            byte[0] % 6 + 2
        } else {
            0
        };

        let transport_config = TransportConfig {
            transport_enabled: config.transport_enabled,
            identity_hash: Some(*transport_identity.hash()),
            local_hops_delta,
            prefer_shorter_path: config.prefer_shorter_path,
            max_paths_per_destination: config.max_paths_per_destination,
            packet_hashlist_max_entries: config.packet_hashlist_max_entries,
            packet_hashlist_allocation: config.packet_hashlist_allocation,
            max_discovery_pr_tags: config.max_discovery_pr_tags,
            max_path_destinations: config.max_path_destinations,
            max_tunnel_destinations_total: config.max_tunnel_destinations_total,
            destination_timeout_secs: config.known_destinations_ttl.as_secs_f64(),
            announce_table_ttl_secs: config.announce_table_ttl.as_secs_f64(),
            announce_table_max_bytes: config.announce_table_max_bytes,
            announce_sig_cache_enabled: config.announce_sig_cache_enabled,
            announce_sig_cache_max_entries: config.announce_sig_cache_max_entries,
            announce_sig_cache_ttl_secs: config.announce_sig_cache_ttl.as_secs_f64(),
            announce_queue_max_entries: queue_config.announce_max_entries,
            announce_queue_max_interfaces: queue_config.announce_max_interfaces,
        };

        let (tx, rx) = event::channel_with_queue_capacities(
            config.driver_event_queue_capacity,
            queue_config.inbound_capacities,
        );
        let tick_interval_ms = Arc::new(AtomicU64::new(1000));
        let mut driver = Driver::new(transport_config, rx, tx.clone(), callbacks);
        let stamper = Arc::new(crate::stamper::Stamper::new(tx.clone(), None));
        driver.set_stamper(Arc::clone(&stamper));
        let stamp_destinations = Arc::new(AtomicBool::new(rns_core::constants::STAMP_DESTINATIONS));
        driver.set_stamp_destinations_flag(Arc::clone(&stamp_destinations));
        driver.apply_stamp_config(&config.stamps, queue_config.hop_limiter);
        *driver.engine.destination_filter_mut() = queue_config.filter.destination_filter();
        let stamp_targets = crate::stamper::StampTargets::from(&config.stamps);
        driver.underlay_mark = config.underlay_mark;
        driver
            .link_manager
            .set_link_mtu_discovery(queue_config.link_mtu_discovery);
        driver
            .link_manager
            .set_proof_verify_threads(config.link_proof_verify_threads);
        driver.set_announce_verify_queue_config(
            queue_config.announce_max_entries,
            queue_config.announce_max_bytes,
            queue_config.announce_ttl_secs,
            queue_config.announce_overflow_policy,
        );
        driver.async_announce_verification = true;
        driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
        let accepting_work = driver.accepting_work_handle();
        driver.known_destinations_ttl = config.known_destinations_ttl.as_secs_f64();
        driver.known_destinations_max_entries = config.known_destinations_max_entries;
        driver.ratchet_store = config.ratchet_store.clone();
        driver.ratchet_expiry_secs = config.ratchet_expiry.as_secs_f64();
        driver.interface_writer_queue_capacity = config.interface_writer_queue_capacity;
        driver.set_announce_rate_defaults(config.announce_rate_defaults);
        driver.set_ingress_control_defaults(config.ingress_control_defaults);
        driver.runtime_config_defaults.known_destinations_ttl =
            config.known_destinations_ttl.as_secs_f64();
        driver
            .holepunch_manager
            .set_policy(config.direct_connect_policy);
        driver
            .holepunch_manager
            .set_allow_private_endpoints(config.direct_connect_allow_private_endpoints);
        driver.runtime_config_defaults.direct_connect_policy = config.direct_connect_policy;
        #[cfg(feature = "hooks")]
        if let Some(provider_config) = config.provider_bridge.clone() {
            driver.runtime_config_defaults.provider_queue_max_events =
                provider_config.queue_max_events;
            driver.runtime_config_defaults.provider_queue_max_bytes =
                provider_config.queue_max_bytes;
            if provider_config.enabled {
                let started = if crate::engine::FOLD_SERVICES {
                    crate::provider_bridge::ProviderBridge::start_folded(
                        provider_config,
                        Arc::clone(tx.poller()),
                    )
                } else {
                    crate::provider_bridge::ProviderBridge::start(provider_config)
                };
                match started {
                    Ok(bridge) => driver.provider_bridge = Some(bridge),
                    Err(err) => log::warn!("failed to start provider bridge: {}", err),
                }
            }
        }

        // Set up announce cache if cache directory is configured
        if let Some(ref cache_dir) = config.cache_dir {
            let announces_dir = cache_dir.join("announces");
            let _ = std::fs::create_dir_all(&announces_dir);
            driver.announce_cache = Some(crate::announce_cache::AnnounceCache::new(announces_dir));
        }

        // Configure probe addresses and device for hole punching
        if !config.probe_addrs.is_empty()
            || config.device.is_some()
            || config.underlay_mark.is_some()
        {
            driver.set_probe_config(
                config.probe_addrs.clone(),
                config.probe_protocol,
                config.device.clone(),
            );
        }

        // Start probe server if configured
        let probe_server = if let Some(port) = config.probe_port {
            let listen_addr: std::net::SocketAddr = ([0, 0, 0, 0], port).into();
            if crate::engine::FOLD_SERVICES {
                match crate::holepunch::probe::FoldedProbeServer::bind(
                    listen_addr,
                    config.underlay_mark,
                    Arc::clone(tx.poller()),
                ) {
                    Ok(server) => {
                        log::info!("Probe server started on 0.0.0.0:{}", port);
                        driver.probe_server = Some(server);
                    }
                    Err(e) => log::error!("Failed to start probe server on port {}: {}", port, e),
                }
                None
            } else {
                match crate::holepunch::probe::start_probe_server_with_mark(
                    listen_addr,
                    config.underlay_mark,
                ) {
                    Ok(handle) => {
                        log::info!("Probe server started on 0.0.0.0:{}", port);
                        Some(handle)
                    }
                    Err(e) => {
                        log::error!("Failed to start probe server on port {}: {}", port, e);
                        None
                    }
                }
            }
        } else {
            None
        };

        // Store management config on driver for ACL enforcement
        driver.management_config = config.management.clone();

        // Store transport identity for tunnel synthesis and management announces.
        driver.transport_identity = Some(transport_identity);

        // Load hooks from config
        #[cfg(feature = "hooks")]
        {
            for hook_cfg in &config.hooks {
                if !hook_cfg.enabled {
                    continue;
                }
                let point_idx = match config::parse_hook_point(&hook_cfg.attach_point) {
                    Some(idx) => idx,
                    None => {
                        log::warn!(
                            "Unknown hook point '{}' for hook '{}'",
                            hook_cfg.attach_point,
                            hook_cfg.name,
                        );
                        continue;
                    }
                };
                let mgr = match driver.hook_manager.as_ref() {
                    Some(m) => m,
                    None => {
                        log::warn!(
                            "Hook manager not available, skipping hook '{}'",
                            hook_cfg.name
                        );
                        continue;
                    }
                };
                let hook_backend = match config::parse_hook_backend(&hook_cfg.hook_type) {
                    Ok(backend) => backend,
                    Err(e) => {
                        log::warn!(
                            "Invalid hook type '{}' for hook '{}': {}",
                            hook_cfg.hook_type,
                            hook_cfg.name,
                            e,
                        );
                        continue;
                    }
                };
                let load_result = if hook_backend == rns_hooks::HookBackend::Builtin {
                    let builtin_id = hook_cfg
                        .builtin_id
                        .as_deref()
                        .filter(|id| !id.is_empty())
                        .or_else(|| (!hook_cfg.path.is_empty()).then_some(hook_cfg.path.as_str()));
                    match builtin_id {
                        Some(id) => mgr.load_builtin(hook_cfg.name.clone(), id, hook_cfg.priority),
                        None => Err(rns_hooks::HookError::CompileError(
                            "built-in hook requires builtin/id or path".to_string(),
                        )),
                    }
                } else {
                    mgr.load_file_backend(
                        hook_cfg.name.clone(),
                        std::path::Path::new(&hook_cfg.path),
                        hook_cfg.priority,
                        hook_backend,
                    )
                };
                match load_result {
                    Ok(program) => {
                        driver.hook_slots[point_idx].attach(program);
                        log::info!(
                            "Loaded hook '{}' at point {} (priority {})",
                            hook_cfg.name,
                            hook_cfg.attach_point,
                            hook_cfg.priority,
                        );
                    }
                    Err(e) => {
                        log::error!(
                            "Failed to load hook '{}' from '{}': {}",
                            hook_cfg.name,
                            hook_cfg.path,
                            e,
                        );
                    }
                }
            }
        }

        // Configure discovery
        driver.discover_interfaces = config.discover_interfaces;
        driver.autoconnect_interface_mode = config.autoconnect_interface_mode;
        driver.autoconnect_interface_gravity = config.autoconnect_interface_gravity;
        driver.autoconnect_announces_to_internal =
            config.autoconnect_announces_to_internal.then_some(true);
        driver.autoconnect_unverified_implementations =
            config.autoconnect_unverified_implementations;
        if let Some(val) = config.discovery_required_value {
            driver.discovery_required_value = val;
        }

        // Shared counter for dynamic interface IDs
        let next_dynamic_id = Arc::new(AtomicU64::new(10000));
        driver.next_dynamic_interface_id = Arc::clone(&next_dynamic_id);

        // Collect discoverable interface configs for the announcer
        let mut discoverable_interfaces = Vec::new();
        #[cfg(feature = "iface-backbone")]
        let mut backbone_peer_pool_candidates = Vec::new();

        let initial_managed_ids = interface_management
            .as_ref()
            .map(|management| management.initial_ids.clone())
            .unwrap_or_default();
        driver.interface_management = interface_management;

        // --- Registry-based startup for interfaces ---
        let registry = config
            .registry
            .unwrap_or_else(crate::interface::registry::InterfaceRegistry::with_builtins);
        for iface_config in config.interfaces {
            let managed_id = initial_managed_ids.get(&iface_config.name).copied();
            #[cfg(feature = "iface-backbone")]
            if iface_config.type_name == "BackboneInterface" {
                if let Some(mode) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<BackboneMode>()
                {
                    if let Some(handle) = runtime_handle_from_mode(mode) {
                        driver.register_backbone_runtime(handle);
                    }
                    if let Some(handle) = peer_state_handle_from_mode(mode) {
                        driver.register_backbone_peer_state(handle);
                    }
                    if let Some(handle) = client_runtime_handle_from_mode(mode) {
                        driver.register_backbone_client_runtime(handle);
                    }
                    if let Some(handle) = backbone_discovery_runtime_from_interface(
                        &iface_config.name,
                        mode,
                        iface_config.discovery.as_ref(),
                        config.transport_enabled,
                        iface_config.ifac.as_ref(),
                    ) {
                        driver.register_backbone_discovery_runtime(handle);
                    }
                }
            }
            #[cfg(feature = "iface-tcp")]
            if iface_config.type_name == "TCPClientInterface" {
                if let Some(tcp_config) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<TcpClientConfig>()
                {
                    driver.register_tcp_client_runtime(tcp_client_runtime_handle_from_config(
                        tcp_config,
                    ));
                }
            }
            #[cfg(feature = "iface-tcp")]
            if iface_config.type_name == "TCPServerInterface" {
                if let Some(tcp_config) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<TcpServerConfig>()
                {
                    driver.register_tcp_server_runtime(tcp_runtime_handle_from_config(tcp_config));
                    driver.register_tcp_server_discovery_runtime(
                        tcp_server_discovery_runtime_from_interface(
                            &iface_config.name,
                            tcp_config,
                            iface_config.discovery.as_ref(),
                            config.transport_enabled,
                            iface_config.ifac.as_ref(),
                        ),
                    );
                }
            }
            #[cfg(feature = "iface-udp")]
            if iface_config.type_name == "UDPInterface" {
                if let Some(udp_config) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<UdpConfig>()
                {
                    driver.register_udp_runtime(udp_runtime_handle_from_config(udp_config));
                }
            }
            #[cfg(feature = "iface-auto")]
            if iface_config.type_name == "AutoInterface" {
                if let Some(auto_config) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<AutoConfig>()
                {
                    driver.register_auto_runtime(auto_runtime_handle_from_config(auto_config));
                }
            }
            #[cfg(feature = "iface-i2p")]
            if iface_config.type_name == "I2PInterface" {
                if let Some(i2p_config) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<I2pConfig>()
                {
                    driver.register_i2p_runtime(i2p_runtime_handle_from_config(i2p_config));
                }
            }
            #[cfg(feature = "iface-pipe")]
            if iface_config.type_name == "PipeInterface" {
                if let Some(pipe_config) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<PipeConfig>()
                {
                    driver.register_pipe_runtime(pipe_runtime_handle_from_config(pipe_config));
                }
            }
            #[cfg(feature = "iface-rnode")]
            if iface_config.type_name == "RNodeInterface" {
                if let Some(rnode_config) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<RNodeConfig>()
                {
                    driver.register_rnode_runtime(rnode_runtime_handle_from_config(rnode_config));
                }
            }

            let factory = match registry.get(&iface_config.type_name) {
                Some(f) => f,
                None => {
                    log::warn!(
                        "No factory registered for interface type '{}'",
                        iface_config.type_name
                    );
                    continue;
                }
            };

            let mut ifac_state = derive_ifac_state(iface_config.ifac.as_ref(), &iface_config.name)?;
            let ifac_runtime =
                ifac_runtime_from_config(iface_config.ifac.as_ref(), factory.default_ifac_size());

            #[cfg(feature = "iface-backbone")]
            if config.backbone_peer_pool.is_some() && iface_config.type_name == "BackboneInterface"
            {
                if let Some(mode) = iface_config
                    .config_data
                    .as_any()
                    .downcast_ref::<BackboneMode>()
                {
                    if let Some(client) = client_config_from_mode(mode) {
                        let priority = client_priority_from_mode(mode).unwrap_or(
                            crate::driver::BACKBONE_PEER_POOL_CONFIGURED_DEFAULT_PRIORITY,
                        );
                        backbone_peer_pool_candidates.push(BackbonePeerPoolCandidateConfig {
                            client,
                            mode: iface_config.mode,
                            gravity: iface_config.gravity,
                            recursive_prs: iface_config.recursive_prs,
                            announces_from_internal: iface_config.announces_from_internal,
                            announces_to_internal: iface_config.announces_to_internal,
                            ingress_control: iface_config.ingress_control,
                            stamp_costs: iface_config.stamp_costs,
                            ifac_runtime: ifac_runtime.clone(),
                            ifac_enabled: ifac_state.is_some(),
                            interface_type_name: iface_config.type_name.clone(),
                            source: crate::driver::BackbonePeerPoolCandidateSource::Configured,
                            priority,
                            discovery: None,
                        });
                        if let Some(ref disc) = iface_config.discovery {
                            discoverable_interfaces.push(discoverable_interface_from_config(
                                &iface_config.name,
                                disc,
                                config.transport_enabled,
                                iface_config.ifac.as_ref(),
                            ));
                        }
                        continue;
                    }
                }
            }

            let ctx = crate::interface::StartContext {
                tx: tx.clone(),
                next_dynamic_id: next_dynamic_id.clone(),
                mode: iface_config.mode,
                gravity: iface_config.gravity,
                recursive_prs: iface_config.recursive_prs,
                announces_from_internal: iface_config.announces_from_internal,
                announces_to_internal: iface_config.announces_to_internal,
                ingress_control: iface_config.ingress_control,
                stamp_costs: iface_config.stamp_costs,
                ifac: ifac_state.clone(),
                underlay_mark: config.underlay_mark,
            };

            let result = match factory.start(iface_config.config_data, ctx) {
                Ok(r) => r,
                Err(e) => {
                    if config.panic_on_interface_error {
                        return Err(e);
                    }
                    log::error!(
                        "Interface '{}' ({}) failed to start: {}",
                        iface_config.name,
                        iface_config.type_name,
                        e
                    );
                    continue;
                }
            };

            if let Some(ref disc) = iface_config.discovery {
                discoverable_interfaces.push(discoverable_interface_from_config(
                    &iface_config.name,
                    disc,
                    config.transport_enabled,
                    iface_config.ifac.as_ref(),
                ));
            }

            match result {
                crate::interface::StartResult::Simple {
                    id,
                    info,
                    writer,
                    interface_type_name,
                    control,
                } => {
                    let managed_control = control.clone();
                    if let Some(control) = control {
                        driver.register_listener_control(control);
                    }
                    if let Some(parent_id) = managed_id {
                        driver.managed_interfaces.insert(
                            iface_config.name.clone(),
                            crate::driver::ManagedInterface {
                                parent_id,
                                type_name: iface_config.type_name.clone(),
                                control: managed_control,
                                static_ids: vec![id],
                            },
                        );
                    }
                    register_started_interface(StartedInterface {
                        driver: &mut driver,
                        tx: &tx,
                        queue_capacity: config.interface_writer_queue_capacity,
                        id,
                        info,
                        writer,
                        interface_type_name,
                        ifac_state,
                        ifac_runtime: &ifac_runtime,
                        stamp_costs: iface_config.stamp_costs,
                    });
                }
                crate::interface::StartResult::Listener { control } => {
                    // Listener-type interface (TcpServer, Auto, I2P, etc.)
                    // registers dynamic interfaces via InterfaceUp events.
                    let managed_control = control.clone();
                    if let Some(control) = control {
                        driver.register_listener_control(control);
                    }
                    if let Some(parent_id) = managed_id {
                        driver.managed_interfaces.insert(
                            iface_config.name.clone(),
                            crate::driver::ManagedInterface {
                                parent_id,
                                type_name: iface_config.type_name.clone(),
                                control: managed_control,
                                static_ids: Vec::new(),
                            },
                        );
                    }
                }
                crate::interface::StartResult::Multi {
                    subinterfaces: subs,
                    control,
                } => {
                    let managed_control = control.clone();
                    if let Some(control) = control {
                        driver.register_listener_control(control);
                    }
                    let static_ids = subs.iter().map(|sub| sub.id).collect();
                    if let Some(parent_id) = managed_id {
                        driver.managed_interfaces.insert(
                            iface_config.name.clone(),
                            crate::driver::ManagedInterface {
                                parent_id,
                                type_name: iface_config.type_name.clone(),
                                control: managed_control,
                                static_ids,
                            },
                        );
                    }
                    let ifac_cfg = &iface_config.ifac;
                    let mut first = true;
                    for sub in subs {
                        let sub_ifac = if first {
                            first = false;
                            ifac_state.take()
                        } else {
                            derive_ifac_state(ifac_cfg.as_ref(), &sub.info.name)?
                        };
                        register_started_interface(StartedInterface {
                            driver: &mut driver,
                            tx: &tx,
                            queue_capacity: config.interface_writer_queue_capacity,
                            id: sub.id,
                            info: sub.info,
                            writer: sub.writer,
                            interface_type_name: sub.interface_type_name,
                            ifac_state: sub_ifac,
                            ifac_runtime: &ifac_runtime,
                            stamp_costs: iface_config.stamp_costs,
                        });
                    }
                }
            }
        }

        // Set up interface announcer if we have discoverable interfaces
        if !discoverable_interfaces.is_empty() {
            let transport_id = *identity.hash();
            let announcer =
                crate::discovery::InterfaceAnnouncer::new(transport_id, discoverable_interfaces);
            log::info!("Interface discovery announcer initialized");
            driver.interface_announcer = Some(announcer);
        }

        // Set up discovered interfaces storage path
        if let Some(ref cache_dir) = config.cache_dir {
            let disc_path = std::path::PathBuf::from(cache_dir)
                .parent()
                .unwrap_or(std::path::Path::new("."))
                .join("storage")
                .join("discovery")
                .join("interfaces");
            let _ = std::fs::create_dir_all(&disc_path);
            driver.discovered_interfaces =
                crate::discovery::DiscoveredInterfaceStorage::new(disc_path);
        }

        // Restore after static interfaces and the announce cache are available,
        // but before the driver can process inbound traffic.
        if let Some(ref storage_dir) = transport_state_dir {
            match storage::load_transport_state(storage_dir) {
                Ok(snapshot) => {
                    let announce_cache = driver.announce_cache.as_ref();
                    let stats = driver.engine.restore_persistence_snapshot(
                        snapshot,
                        time::now(),
                        |packet_hash| {
                            announce_cache
                                .and_then(|cache| cache.get(packet_hash).ok().flatten())
                                .map(|(raw, _)| raw)
                        },
                    );
                    log::debug!(
                        "restored transport state: {} packet hashes, {} paths, {} tunnels ({} paths skipped)",
                        stats.packet_hashes,
                        stats.paths,
                        stats.tunnels,
                        stats.skipped_paths,
                    );
                }
                Err(err) => log::warn!("failed to restore transport state: {}", err),
            }
        }

        #[cfg(feature = "iface-backbone")]
        if let Some(settings) = config.backbone_peer_pool.clone() {
            driver.configure_backbone_peer_pool(settings, backbone_peer_pool_candidates);
            driver.seed_backbone_peer_pool_from_discovery_cache();
        }

        // Set up management destinations if enabled
        if config.management.enable_remote_management {
            if let Some(prv_key) = identity.get_private_key() {
                let identity_hash = *identity.hash();
                let mgmt_dest = crate::management::management_dest_hash(&identity_hash);

                // Extract Ed25519 signing keys from the identity
                let sig_prv = rns_crypto::ed25519::Ed25519PrivateKey::from_bytes(
                    &prv_key[32..64].try_into().unwrap(),
                );
                let sig_pub_bytes: [u8; 32] = identity.get_public_key().unwrap()[32..64]
                    .try_into()
                    .unwrap();

                // Register as SINGLE destination in transport engine
                driver
                    .engine
                    .register_destination(mgmt_dest, rns_core::constants::DESTINATION_SINGLE);
                driver
                    .local_destinations
                    .insert(mgmt_dest, rns_core::constants::DESTINATION_SINGLE);

                // Register as link destination in link manager
                driver.link_manager.register_link_destination(
                    mgmt_dest,
                    sig_prv,
                    sig_pub_bytes,
                    crate::link_manager::ResourceStrategy::AcceptNone,
                );

                // Register management path hashes
                driver
                    .link_manager
                    .register_management_path(crate::management::status_path_hash());
                driver
                    .link_manager
                    .register_management_path(crate::management::path_path_hash());

                log::info!("Remote management enabled on {:02x?}", &mgmt_dest[..4],);

                // Set up allowed list
                if !config.management.remote_management_allowed.is_empty() {
                    log::info!(
                        "Remote management allowed for {} identities",
                        config.management.remote_management_allowed.len(),
                    );
                }
            }
        }

        if config.management.publish_blackhole {
            if let Some(prv_key) = identity.get_private_key() {
                let identity_hash = *identity.hash();
                let bh_dest = crate::management::blackhole_dest_hash(&identity_hash);

                let sig_prv = rns_crypto::ed25519::Ed25519PrivateKey::from_bytes(
                    &prv_key[32..64].try_into().unwrap(),
                );
                let sig_pub_bytes: [u8; 32] = identity.get_public_key().unwrap()[32..64]
                    .try_into()
                    .unwrap();

                driver
                    .engine
                    .register_destination(bh_dest, rns_core::constants::DESTINATION_SINGLE);
                driver.link_manager.register_link_destination(
                    bh_dest,
                    sig_prv,
                    sig_pub_bytes,
                    crate::link_manager::ResourceStrategy::AcceptNone,
                );
                driver
                    .link_manager
                    .register_management_path(crate::management::list_path_hash());
                // Upstream 426673e9: requesting the list costs a 5.25 stamp,
                // below the `LR_COST` default.
                driver
                    .link_manager
                    .link_request_costs()
                    .set(bh_dest, rns_core::constants::BLACKHOLE_LINK_COST.as_f64());

                log::info!(
                    "Blackhole list publishing enabled on {:02x?}",
                    &bh_dest[..4],
                );
            }
        }

        // Set up probe responder if enabled
        if config.respond_to_probes && config.transport_enabled {
            let identity_hash = *identity.hash();
            let probe_dest = crate::management::probe_dest_hash(&identity_hash);

            // Register as SINGLE destination in transport engine
            driver
                .engine
                .register_destination(probe_dest, rns_core::constants::DESTINATION_SINGLE);
            driver
                .local_destinations
                .insert(probe_dest, rns_core::constants::DESTINATION_SINGLE);

            // Register PROVE_ALL proof strategy with transport identity
            let probe_identity = rns_crypto::identity::Identity::from_private_key(
                &identity.get_private_key().unwrap(),
            );
            driver.proof_strategies.insert(
                probe_dest,
                (
                    rns_core::types::ProofStrategy::ProveAll,
                    Some(probe_identity),
                ),
            );

            driver.probe_responder_hash = Some(probe_dest);

            log::info!("Probe responder enabled on {:02x?}", &probe_dest[..4],);
        }

        // Spawn timer thread with configurable tick interval, unless the
        // driver keeps the tick itself (compact engine).
        if !crate::engine::FOLD_TIMERS {
            let timer_tx = tx.clone();
            let timer_interval = Arc::clone(&tick_interval_ms);
            thread::Builder::new()
                .name("rns-timer".into())
                .spawn(move || loop {
                    let ms = timer_interval.load(Ordering::Relaxed);
                    thread::sleep(Duration::from_millis(ms));
                    if timer_tx.send(Event::Tick).is_err() {
                        break; // receiver dropped
                    }
                })?;
        }

        // Start LocalServer for shared instance clients if share_instance is enabled
        #[cfg(feature = "iface-local")]
        if config.share_instance {
            let local_server_config = LocalServerConfig {
                instance_name: config.instance_name.clone(),
                port: config.shared_instance_port,
                // Keys the listener on the driver loop when interface I/O is
                // folded; clients get ids of their own.
                interface_id: rns_core::transport::types::InterfaceId(
                    next_dynamic_id.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
                ),
            };
            match crate::interface::local::start_server(
                local_server_config,
                tx.clone(),
                next_dynamic_id.clone(),
            ) {
                Ok(control) => {
                    driver.register_listener_control(control);
                    log::info!(
                        "Local shared instance server started (instance={}, port={})",
                        config.instance_name,
                        config.shared_instance_port
                    );
                }
                Err(e) => {
                    log::error!("Failed to start local shared instance server: {}", e);
                }
            }
        }

        // Start the RPC servers if share_instance is enabled: on threads, or
        // on the driver loop when the compact engine folds services.
        let serve_rpc_threads = config.share_instance && !crate::engine::FOLD_SERVICES;
        if config.share_instance && crate::engine::FOLD_SERVICES {
            let mut service = crate::driver::RpcService::new(rpc_auth_key, Arc::clone(tx.poller()));
            match std::net::TcpListener::bind(("127.0.0.1", config.rpc_port)).and_then(|listener| {
                service.add_listener(crate::driver::RpcListener::Tcp(listener))
            }) {
                Ok(()) => log::info!("RPC server started on 127.0.0.1:{}", config.rpc_port),
                Err(e) => log::error!("Failed to start RPC server: {}", e),
            }
            #[cfg(target_os = "linux")]
            match crate::rpc::bind_abstract_unix(&config.instance_name).and_then(|listener| {
                service.add_listener(crate::driver::RpcListener::Unix(listener))
            }) {
                Ok(()) => log::info!(
                    "RPC server started on abstract Unix socket for {}",
                    config.instance_name
                ),
                Err(e) => log::error!("Failed to start abstract Unix RPC server: {e}"),
            }
            driver.rpc_service = Some(service);
        }
        let rpc_server = if serve_rpc_threads {
            let rpc_addr = crate::rpc::RpcAddr::Tcp("127.0.0.1".into(), config.rpc_port);
            match crate::rpc::RpcServer::start(&rpc_addr, rpc_auth_key, tx.clone()) {
                Ok(server) => {
                    log::info!("RPC server started on 127.0.0.1:{}", config.rpc_port);
                    Some(server)
                }
                Err(e) => {
                    log::error!("Failed to start RPC server: {}", e);
                    None
                }
            }
        } else {
            None
        };
        #[cfg(target_os = "linux")]
        let rpc_unix_server = if serve_rpc_threads {
            match crate::rpc::RpcServer::start_abstract_unix(
                &config.instance_name,
                rpc_auth_key,
                tx.clone(),
            ) {
                Ok(server) => {
                    log::info!(
                        "RPC server started on abstract Unix socket for {}",
                        config.instance_name
                    );
                    Some(server)
                }
                Err(e) => {
                    log::error!("Failed to start abstract Unix RPC server: {e}");
                    None
                }
            }
        } else {
            None
        };

        let announce_verify_queue = Arc::clone(&driver.announce_verify_queue);
        let verify_shutdown = Arc::clone(&driver.announce_verify_signal);
        let verify_shutdown_thread = Arc::clone(&verify_shutdown);
        let verify_tx = tx.clone();
        // The verification thread, unless the compact engine verifies
        // announces on its driver loop.
        let verify_handle = if crate::engine::FOLD_WORK {
            None
        } else {
            Some(
                thread::Builder::new()
                    .name("rns-verify".into())
                    .spawn(move || {
                        #[cfg(target_family = "unix")]
                        {
                            unsafe {
                                libc::nice(5);
                            }
                        }

                        while !verify_shutdown_thread.stopped() {
                            let batch = {
                                let mut queue = announce_verify_queue
                                    .lock()
                                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                                queue.take_pending(time::now())
                            };

                            if batch.is_empty() {
                                // Until the driver queues an announce.
                                verify_shutdown_thread.wait();
                                continue;
                            }

                            for (key, pending) in batch {
                                if verify_shutdown_thread.stopped() {
                                    break;
                                }
                                let event = match crate::driver::check_pending_announce(&pending) {
                                    crate::driver::AnnounceCheck::Valid {
                                        validated,
                                        sig_cache_key,
                                    } => Event::AnnounceVerified {
                                        key,
                                        validated,
                                        sig_cache_key,
                                    },
                                    crate::driver::AnnounceCheck::Invalid { sig_cache_key } => {
                                        Event::AnnounceVerifyFailed { key, sig_cache_key }
                                    }
                                };
                                if verify_tx.send(event).is_err() {
                                    return;
                                }
                            }
                        }
                    })?,
            )
        };

        // Spawn the driver after startup has registered local destinations and static interfaces.
        // Interface readers can enqueue frames before this point; the bounded event queue preserves
        // ordering and backpressures instead of dropping startup traffic.
        log::info!("rns-net {} engine", crate::engine::ENGINE);
        // The compact engine persists transport state on the driver loop.
        if crate::engine::FOLD_TIMERS {
            driver.folded.persist_dir = transport_state_dir.clone();
        }
        let driver_handle = thread::Builder::new()
            .name("rns-driver".into())
            .spawn(move || {
                driver.run();
            })?;

        // Match upstream's 12-hour background persistence cadence. The worker
        // performs disk serialization at low priority and never touches engine
        // state directly; it obtains a consistent snapshot through the driver.
        let (persist_handle, persist_stop) = if let Some(storage_dir) = transport_state_dir
            .clone()
            .filter(|_| !crate::engine::FOLD_TIMERS)
        {
            let persist_tx = tx.clone();
            let (stop_tx, stop_rx) = std::sync::mpsc::channel();
            let handle = thread::Builder::new()
                .name("rns-persist".into())
                .spawn(move || {
                    #[cfg(target_family = "unix")]
                    unsafe {
                        libc::nice(5);
                    }

                    loop {
                        match stop_rx.recv_timeout(Duration::from_secs(12 * 60 * 60)) {
                            Ok(()) | Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
                            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {}
                        }
                        let (response_tx, response_rx) = std::sync::mpsc::channel();
                        if persist_tx
                            .send(Event::query(
                                QueryRequest::TransportStateSnapshot,
                                response_tx,
                            ))
                            .is_err()
                        {
                            break;
                        }
                        let Ok(QueryResponse::TransportStateSnapshot(snapshot)) =
                            response_rx.recv()
                        else {
                            continue;
                        };
                        if let Err(err) =
                            storage::save_transport_state(&snapshot, &storage_dir, true)
                        {
                            log::warn!("failed to persist transport state: {}", err);
                        }
                    }
                })?;
            (Some(handle), Some(stop_tx))
        } else {
            (None, None)
        };

        let mut node = RnsNode {
            tx,
            driver_handle: Some(driver_handle),
            verify_handle,
            verify_shutdown,
            persist_handle,
            persist_stop,
            rpc_server,
            #[cfg(target_os = "linux")]
            rpc_unix_server,
            tick_interval_ms,
            next_send_ticket: AtomicU64::new(0),
            encrypt_deferred_inline: std::thread::available_parallelism()
                .map_or(true, |cpus| cpus.get() == 1),
            accepting_work,
            probe_server,
            known_destinations_path: None,
            known_destinations_store: None,
            stamp_store: None,
            stamp_destinations,
            stamp_targets,
            stamper: Some(stamper),
            transport_state_dir,
            ratchet_store: config.ratchet_store,
            ratchet_expiry: config.ratchet_expiry,
            #[cfg(feature = "iface-local")]
            destination_types: Default::default(),
        };
        // Destination stamps live under `storage/stamps` next to the cache
        // directory, the layout `from_config_dirs` uses; a node without a
        // cache directory cannot stamp its announces.
        if let Some(storage_dir) = storage_dir {
            node.set_stamp_store(crate::stamps::DestinationStampStore::new(&storage_dir));
        }
        Ok(node)
    }

    /// Query the driver for state information.
    pub fn query(&self, request: QueryRequest) -> Result<QueryResponse, SendError> {
        let (resp_tx, resp_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::query(request, resp_tx))
            .map_err(|_| SendError)?;
        resp_rx.recv().map_err(|_| SendError)
    }

    /// Attach a named interface from the current on-disk configuration.
    pub fn attach_interface(&self, name: &str) -> Result<Option<bool>, SendError> {
        self.manage_interface(crate::event::InterfaceManagementOperation::Attach, name)
    }

    /// Detach a named interface and its spawned children.
    pub fn detach_interface(&self, name: &str) -> Result<Option<bool>, SendError> {
        self.manage_interface(crate::event::InterfaceManagementOperation::Detach, name)
    }

    /// Detach and attach a named interface using the current on-disk configuration.
    pub fn reload_interface(&self, name: &str) -> Result<Option<bool>, SendError> {
        self.manage_interface(crate::event::InterfaceManagementOperation::Reload, name)
    }

    fn manage_interface(
        &self,
        operation: crate::event::InterfaceManagementOperation,
        name: &str,
    ) -> Result<Option<bool>, SendError> {
        match self.query(QueryRequest::ManageInterface {
            operation,
            name: name.to_string(),
        })? {
            QueryResponse::InterfaceManagement(result) => Ok(result),
            _ => Err(SendError),
        }
    }

    /// Snapshot active Link metadata, including the negotiated direct-data MDU.
    pub fn links(&self) -> Result<Vec<crate::event::LinkInfoEntry>, SendError> {
        match self.query(QueryRequest::Links)? {
            QueryResponse::Links(links) => Ok(links),
            _ => Err(SendError),
        }
    }

    /// Snapshot one Link, including local send backlog, or None if unknown.
    pub fn query_link(
        &self,
        link_id: [u8; 16],
    ) -> Result<Option<crate::event::LinkInfoEntry>, SendError> {
        Ok(self
            .links()?
            .into_iter()
            .find(|link| link.link_id == link_id))
    }

    /// The link and interface statistics the driver last published, read
    /// without a driver round trip.
    ///
    /// Unlike [`query`](Self::query), [`links`](Self::links) and
    /// [`query_link`](Self::query_link), which queue behind every earlier
    /// control event and answer exactly at that point, this returns at once
    /// even while the driver works through a long backlog or is blocked, for
    /// example in a callback. The price is freshness and ordering: the driver
    /// republishes on every tick (`global.tick_interval_ms`, 1 s by default)
    /// and as soon as a link or interface changes state, so the snapshot may
    /// lag by up to one tick, plus however long the driver has been blocked,
    /// and it is not a barrier: events sent before the call, sends included,
    /// may not be reflected yet. Before the first publish the snapshot is
    /// empty with `sequence == 0`.
    pub fn state_snapshot(&self) -> Arc<crate::state_snapshot::StateSnapshot> {
        self.tx.state_snapshot().load()
    }

    /// Links as of the last published [`state_snapshot`](Self::state_snapshot).
    ///
    /// May lag by up to one tick and is not ordered after earlier events; use
    /// [`links`](Self::links) for an exact, ordered answer.
    pub fn snapshot_links(&self) -> Vec<crate::event::LinkInfoEntry> {
        self.state_snapshot().links.clone()
    }

    /// One link as of the last published
    /// [`state_snapshot`](Self::state_snapshot), or None if it was not known
    /// then.
    ///
    /// May lag by up to one tick and is not ordered after earlier events; use
    /// [`query_link`](Self::query_link) for an exact, ordered answer.
    pub fn snapshot_link(&self, link_id: [u8; 16]) -> Option<crate::event::LinkInfoEntry> {
        self.state_snapshot().link(&link_id).cloned()
    }

    /// Interface statistics as of the last published
    /// [`state_snapshot`](Self::state_snapshot).
    ///
    /// May lag by up to one tick and is not ordered after earlier events; use
    /// `query(QueryRequest::InterfaceStats)` for an exact, ordered answer.
    pub fn snapshot_interfaces(&self) -> crate::event::InterfaceStatsResponse {
        self.state_snapshot().interfaces.clone()
    }

    /// Return the slowest positive bitrate among currently registered interfaces.
    ///
    /// Returns `Ok(None)` when no registered interface advertises a usable
    /// bitrate.
    pub fn lowest_interface_bitrate(&self) -> Result<Option<u64>, SendError> {
        match self.query(QueryRequest::LowestInterfaceBitrate)? {
            QueryResponse::LowestInterfaceBitrate(bitrate) => Ok(bitrate),
            _ => Err(SendError),
        }
    }

    /// Return the adaptive medium path timeout advertised by the transport engine.
    ///
    /// The estimate covers a full MTU round trip on the slowest registered
    /// positive-bitrate interface plus the per-hop allowance, or is zero when
    /// no usable bitrate is known.
    pub fn medium_path_timeout(&self) -> Result<f64, SendError> {
        match self.query(QueryRequest::MediumPathTimeout)? {
            QueryResponse::MediumPathTimeout(timeout) => Ok(timeout),
            _ => Err(SendError),
        }
    }

    /// Enter drain mode and stop admitting new work.
    pub fn begin_drain(&self, timeout: Duration) -> Result<(), SendError> {
        // Draining is one-way. Refuse new work from this call on, before the
        // driver has processed the event.
        self.accepting_work.store(false, Ordering::Release);
        self.tx
            .send(Event::BeginDrain { timeout })
            .map_err(|_| SendError)
    }

    /// Query current drain/lifecycle status.
    pub fn drain_status(&self) -> Result<crate::event::DrainStatus, SendError> {
        match self.query(QueryRequest::DrainStatus)? {
            QueryResponse::DrainStatus(status) => Ok(status),
            _ => Err(SendError),
        }
    }

    fn reject_new_work_if_draining(&self) -> Result<(), SendError> {
        // A flag the driver clears when draining starts, not a driver query:
        // this runs on every send.
        if self.accepting_work.load(Ordering::Acquire) {
            Ok(())
        } else {
            Err(SendError)
        }
    }

    /// Send a raw outbound packet.
    pub fn send_raw(
        &self,
        raw: Vec<u8>,
        dest_type: u8,
        attached_interface: Option<rns_core::transport::types::InterfaceId>,
    ) -> Result<(), SendError> {
        self.tx
            .send(Event::SendOutbound {
                raw,
                dest_type,
                attached_interface,
            })
            .map_err(|_| SendError)
    }

    /// Register a local destination with the transport engine.
    pub fn register_destination(
        &self,
        dest_hash: [u8; 16],
        dest_type: u8,
    ) -> Result<(), SendError> {
        self.tx
            .send(Event::RegisterDestination {
                dest_hash,
                dest_type,
            })
            .map_err(|_| SendError)
    }

    /// Deregister a local destination.
    pub fn deregister_destination(&self, dest_hash: [u8; 16]) -> Result<(), SendError> {
        self.tx
            .send(Event::DeregisterDestination { dest_hash })
            .map_err(|_| SendError)
    }

    /// Deregister a link destination (stop accepting incoming links).
    pub fn deregister_link_destination(&self, dest_hash: [u8; 16]) -> Result<(), SendError> {
        self.tx
            .send(Event::DeregisterLinkDestination { dest_hash })
            .map_err(|_| SendError)
    }

    /// Register a link destination that can accept incoming links.
    ///
    /// `dest_hash`: the destination hash
    /// `sig_prv_bytes`: Ed25519 private signing key (32 bytes)
    /// `sig_pub_bytes`: Ed25519 public signing key (32 bytes)
    pub fn register_link_destination(
        &self,
        dest_hash: [u8; 16],
        sig_prv_bytes: [u8; 32],
        sig_pub_bytes: [u8; 32],
        resource_strategy: u8,
    ) -> Result<(), SendError> {
        self.register_link_destination_with_max_request_size(
            dest_hash,
            sig_prv_bytes,
            sig_pub_bytes,
            resource_strategy,
            None,
        )
    }

    /// Register a link destination with an optional maximum accepted request size.
    pub fn register_link_destination_with_max_request_size(
        &self,
        dest_hash: [u8; 16],
        sig_prv_bytes: [u8; 32],
        sig_pub_bytes: [u8; 32],
        resource_strategy: u8,
        max_request_size: Option<usize>,
    ) -> Result<(), SendError> {
        self.tx
            .send(Event::RegisterLinkDestination {
                dest_hash,
                sig_prv_bytes,
                sig_pub_bytes,
                resource_strategy,
                max_request_size,
            })
            .map_err(|_| SendError)
    }

    /// Set the required link establishment stamp cost for a destination.
    ///
    /// Mirrors `Destination.set_link_cost`: the cost is clamped to
    /// `0..=LINK_COST_CAP` and the clamped value is returned. Set it to 0 to
    /// disable. The cost is carried in the destination's announce extension
    /// (emitted when destination stamping is on) so initiators grind their
    /// link request stamps to it, and, when link request stamps are enforced,
    /// incoming link requests whose stamp value is below it are dropped.
    pub fn set_link_cost(&self, dest_hash: [u8; 16], cost: f64) -> f64 {
        self.tx.link_request_costs().set(dest_hash, cost).as_f64()
    }

    /// The link request stamp cost announced and required for a destination.
    ///
    /// Mirrors `Destination.get_link_cost`: the configured cost, or the
    /// default `LINK_REQUEST_STAMP_COST` (`Transport.LR_COST`) when none was
    /// set. Costs are fractional since upstream 796ef2d4 and held on the
    /// 1/256 grid they are announced and checked on, so 14.2 reads back as
    /// 14.19921875.
    pub fn link_cost(&self, dest_hash: &[u8; 16]) -> f64 {
        self.tx.link_request_costs().get(dest_hash).as_f64()
    }

    /// Override `Transport.LR_STAMPS_ENFORCED` for this node.
    ///
    /// When enabled, incoming link requests whose stamp value is below the
    /// destination's [`link_cost`](Self::link_cost) are dropped. Defaults to
    /// `rns_core::constants::LR_STAMPS_ENFORCED` (off, as upstream).
    pub fn set_link_request_stamps_enforced(&self, enforced: bool) {
        self.tx.link_request_costs().set_enforced(enforced);
    }

    /// Whether incoming link request stamps are enforced.
    pub fn link_request_stamps_enforced(&self) -> bool {
        self.tx.link_request_costs().enforced()
    }

    /// Register a request handler for a given path on established links.
    pub fn register_request_handler<F>(
        &self,
        path: &str,
        allowed_list: Option<Vec<[u8; 16]>>,
        handler: F,
    ) -> Result<(), SendError>
    where
        F: Fn([u8; 16], &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<Vec<u8>>
            + Send
            + 'static,
    {
        self.tx
            .send(Event::RegisterRequestHandler {
                path: path.to_string(),
                allowed_list,
                handler: Box::new(handler),
            })
            .map_err(|_| SendError)
    }

    /// Register a request handler that can return resource responses with metadata.
    pub fn register_request_handler_response<F>(
        &self,
        path: &str,
        allowed_list: Option<Vec<[u8; 16]>>,
        handler: F,
    ) -> Result<(), SendError>
    where
        F: Fn(
                [u8; 16],
                &str,
                &[u8],
                Option<&([u8; 16], [u8; 64])>,
            ) -> Option<crate::link_manager::RequestResponse>
            + Send
            + 'static,
    {
        self.tx
            .send(Event::RegisterRequestHandlerResponse {
                path: path.to_string(),
                allowed_list,
                handler: Box::new(handler),
            })
            .map_err(|_| SendError)
    }

    /// Register a request handler whose response is completed asynchronously
    /// with [`RnsNode::send_deferred_response`].
    pub fn register_deferred_request_handler<F>(
        &self,
        path: &str,
        allowed_list: Option<Vec<[u8; 16]>>,
        handler: F,
    ) -> Result<(), SendError>
    where
        F: Fn([u8; 16], &str, [u8; 16], &[u8], Option<&([u8; 16], [u8; 64])>) + Send + 'static,
    {
        self.tx
            .send(Event::RegisterDeferredRequestHandler {
                path: path.to_string(),
                allowed_list,
                handler: Box::new(handler),
            })
            .map_err(|_| SendError)
    }

    /// Complete a request accepted by a deferred request handler.
    pub fn send_deferred_response(
        &self,
        link_id: [u8; 16],
        request_id: [u8; 16],
        data: Vec<u8>,
    ) -> Result<(), SendError> {
        self.tx
            .send(Event::SendDeferredResponse {
                link_id,
                request_id,
                data,
            })
            .map_err(|_| SendError)
    }

    /// Create an outbound link to a destination.
    ///
    /// Returns the link_id on success.
    pub fn create_link(
        &self,
        dest_hash: [u8; 16],
        dest_sig_pub_bytes: [u8; 32],
    ) -> Result<[u8; 16], SendError> {
        self.reject_new_work_if_draining()?;
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::CreateLink {
                dest_hash,
                dest_sig_pub_bytes,
                response_tx,
            })
            .map_err(|_| SendError)?;
        let link_id = response_rx.recv().map_err(|_| SendError)?;
        if link_id == [0u8; 16] {
            Err(SendError)
        } else {
            Ok(link_id)
        }
    }

    /// Send a request on an established link.
    pub fn send_request(
        &self,
        link_id: [u8; 16],
        path: &str,
        data: &[u8],
    ) -> Result<(), SendError> {
        self.send_request_with_max_response_size(link_id, path, data, None)
    }

    /// Send a request with an optional maximum accepted response size.
    pub fn send_request_with_max_response_size(
        &self,
        link_id: [u8; 16],
        path: &str,
        data: &[u8],
        max_response_size: Option<usize>,
    ) -> Result<(), SendError> {
        self.reject_new_work_if_draining()?;
        self.tx
            .send(Event::SendRequest {
                link_id,
                path: path.to_string(),
                data: data.to_vec(),
                max_response_size,
            })
            .map_err(|_| SendError)
    }

    /// Identify on a link (reveal identity to remote peer).
    pub fn identify_on_link(
        &self,
        link_id: [u8; 16],
        identity_prv_key: [u8; 64],
    ) -> Result<(), SendError> {
        self.reject_new_work_if_draining()?;
        self.tx
            .send(Event::IdentifyOnLink {
                link_id,
                identity_prv_key,
            })
            .map_err(|_| SendError)
    }

    /// Tear down a link.
    pub fn teardown_link(&self, link_id: [u8; 16]) -> Result<(), SendError> {
        self.tx
            .send(Event::TeardownLink { link_id })
            .map_err(|_| SendError)
    }

    /// Send a resource on an established link.
    pub fn send_resource(
        &self,
        link_id: [u8; 16],
        data: Vec<u8>,
        metadata: Option<Vec<u8>>,
    ) -> Result<(), SendError> {
        self.send_resource_with_auto_compress(link_id, data, metadata, true)
    }

    /// Send a resource on an established link, controlling automatic compression.
    ///
    /// Set `auto_compress` to `false` when the application knows its payload is
    /// already compressed or encrypted. This skips the compression attempt;
    /// link encryption and Resource integrity checks still apply. With `true`,
    /// eligible payloads are fully compressed and the result is used only when
    /// smaller. Even a rejected attempt costs CPU, while disabling compression
    /// for compressible content can substantially increase transmitted bytes.
    ///
    /// Buffered sends of at least 16 KiB that fit one Resource segment can
    /// prepare compression on a node-owned worker. Its admission budget is eight
    /// jobs and 8 MiB of owned input capacity plus reserved compressed output.
    /// Codec workspace and the running job's temporary input are additional.
    /// Saturation, small/split Resources and worker startup failure use synchronous
    /// preparation, preserving earlier buffered sends on the same link. This
    /// improves responsiveness for eligible sends; it does not provide a latency
    /// bound under overload. Reader transfers and request responses keep their
    /// existing synchronous preparation paths.
    pub fn send_resource_with_auto_compress(
        &self,
        link_id: [u8; 16],
        data: Vec<u8>,
        metadata: Option<Vec<u8>>,
        auto_compress: bool,
    ) -> Result<(), SendError> {
        self.send_resource_with_options(
            link_id,
            data,
            metadata,
            crate::resource::ResourceSendOptions {
                auto_compress,
                ..Default::default()
            },
        )
    }

    /// Send a resource on an established link with explicit send options,
    /// including the application flag the receiver sees with the data.
    pub fn send_resource_with_options(
        &self,
        link_id: [u8; 16],
        data: Vec<u8>,
        metadata: Option<Vec<u8>>,
        options: crate::resource::ResourceSendOptions,
    ) -> Result<(), SendError> {
        self.reject_new_work_if_draining()?;
        self.tx
            .send(Event::SendResource {
                link_id,
                data,
                metadata,
                options,
            })
            .map_err(|_| SendError)
    }

    /// Send a Resource from a sequential reader without retaining the full
    /// logical payload in memory.
    pub fn send_resource_reader<R: Read + Send + 'static>(
        &self,
        link_id: [u8; 16],
        reader: R,
        declared_length: u64,
        metadata: Option<Vec<u8>>,
        auto_compress: bool,
    ) -> Result<crate::resource::ResourceTransferId, SendError> {
        self.send_resource_reader_with_options(
            link_id,
            reader,
            declared_length,
            metadata,
            crate::resource::ResourceSendOptions {
                auto_compress,
                ..Default::default()
            },
        )
    }

    /// Send a Resource from a sequential reader with explicit send options.
    pub fn send_resource_reader_with_options<R: Read + Send + 'static>(
        &self,
        link_id: [u8; 16],
        reader: R,
        declared_length: u64,
        metadata: Option<Vec<u8>>,
        options: crate::resource::ResourceSendOptions,
    ) -> Result<crate::resource::ResourceTransferId, SendError> {
        self.reject_new_work_if_draining()?;
        let transfer_id = crate::resource::ResourceTransferId(
            NEXT_RESOURCE_TRANSFER_ID.fetch_add(1, Ordering::Relaxed),
        );
        self.tx
            .send(Event::SendResourceStream {
                link_id,
                transfer_id,
                reader: Box::new(reader),
                declared_length,
                metadata,
                options,
            })
            .map_err(|_| SendError)?;
        Ok(transfer_id)
    }

    /// Send a file as a bounded-memory streaming Resource.
    pub fn send_resource_file(
        &self,
        link_id: [u8; 16],
        path: impl AsRef<Path>,
        metadata: Option<Vec<u8>>,
        auto_compress: bool,
    ) -> io::Result<crate::resource::ResourceTransferId> {
        let file = std::fs::File::open(path.as_ref())?;
        let length = file.metadata()?.len();
        self.send_resource_reader(link_id, file, length, metadata, auto_compress)
            .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "RNS driver shut down"))
    }

    /// Select delivery policy for independent incoming Resources on a link.
    pub fn set_resource_receive_mode(
        &self,
        link_id: [u8; 16],
        mode: crate::resource::ResourceReceiveMode,
    ) -> Result<(), SendError> {
        self.tx
            .send(Event::SetResourceReceiveMode { link_id, mode })
            .map_err(|_| SendError)
    }

    /// Set the resource acceptance strategy for a link.
    ///
    /// 0 = AcceptNone, 1 = AcceptAll, 2 = AcceptApp
    pub fn set_resource_strategy(&self, link_id: [u8; 16], strategy: u8) -> Result<(), SendError> {
        self.tx
            .send(Event::SetResourceStrategy { link_id, strategy })
            .map_err(|_| SendError)
    }

    /// Accept or reject a pending resource (for AcceptApp strategy).
    pub fn accept_resource(
        &self,
        link_id: [u8; 16],
        resource_hash: Vec<u8>,
        accept: bool,
    ) -> Result<(), SendError> {
        if accept {
            self.reject_new_work_if_draining()?;
        }
        self.tx
            .send(Event::AcceptResource {
                link_id,
                resource_hash,
                accept,
            })
            .map_err(|_| SendError)
    }

    /// Send a channel message on a link (compatibility API).
    pub fn send_channel_message(
        &self,
        link_id: [u8; 16],
        msgtype: u16,
        payload: Vec<u8>,
    ) -> Result<(), SendError> {
        self.try_send_channel_message(link_id, msgtype, payload)
            .map_err(|_| SendError)
    }

    /// Attempt to enqueue a channel message, preserving temporary backpressure.
    /// A NotReady result means no sequence was reserved; callers can retry the
    /// same payload when capacity is available. Other failures are terminal.
    pub fn try_send_channel_message(
        &self,
        link_id: [u8; 16],
        msgtype: u16,
        payload: Vec<u8>,
    ) -> Result<(), ChannelSendError> {
        self.reject_new_work_if_draining()
            .map_err(|_| ChannelSendError::Unavailable)?;
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::SendChannelMessage {
                link_id,
                msgtype,
                payload,
                response_tx,
            })
            .map_err(|_| ChannelSendError::Unavailable)?;
        response_rx
            .recv()
            .map_err(|_| ChannelSendError::Unavailable)?
            .map_err(ChannelSendError::from)
    }

    /// Propose a direct P2P connection to a peer via NAT hole punching.
    ///
    /// The link must be active and connected through a backbone node.
    /// If successful, a direct UDP connection will be established, bypassing the backbone.
    pub fn propose_direct_connect(&self, link_id: [u8; 16]) -> Result<(), SendError> {
        self.reject_new_work_if_draining()?;
        self.tx
            .send(Event::ProposeDirectConnect { link_id })
            .map_err(|_| SendError)
    }

    /// Set the policy for handling incoming direct-connect proposals.
    pub fn set_direct_connect_policy(
        &self,
        policy: crate::holepunch::orchestrator::HolePunchPolicy,
    ) -> Result<(), SendError> {
        self.tx
            .send(Event::SetDirectConnectPolicy { policy })
            .map_err(|_| SendError)
    }

    /// Wait for capacity and transmit a Link packet through its interface writer.
    ///
    /// Success means the underlying transport accepted the complete packet,
    /// not that the peer received or acknowledged it. Cancellation before
    /// admission sends nothing; after admission, dropping this future does not
    /// cancel transmission. This future is independent of any async runtime.
    /// The driver's configured event capacity also bounds the total number of
    /// admitted confirmed sends (Links, packets and announcements), including
    /// frames waiting for an interface or write.
    /// A transmission error can follow a partial write; only QueueFull from the
    /// try API guarantees that retrying cannot duplicate an admitted packet.
    pub async fn send_on_link(
        &self,
        link_id: [u8; 16],
        data: Vec<u8>,
        context: u8,
    ) -> Result<(), crate::link_send::LinkSendError> {
        use crate::link_send::Completion;
        let tracking = self.tx.link_send_pool().track(link_id, true);
        let permit = self.tx.link_send_pool().acquire().await?;
        let (completion, receipt) = Completion::new_tracked(permit, tracking);
        self.tx
            .send_async(Event::SendLinkTracked {
                link_id,
                data,
                context,
                completion,
            })
            .await?;
        receipt.await
    }

    /// Try to admit a Link send without waiting for capacity.
    ///
    /// A successful return is a receipt, not a transmission result. Await the
    /// receipt for exactly the same completion contract as `send_on_link`.
    /// QueueFull means nothing was admitted and the caller may retry.
    pub fn try_send_on_link(
        &self,
        link_id: [u8; 16],
        data: Vec<u8>,
        context: u8,
    ) -> Result<crate::link_send::LinkSendReceipt, crate::link_send::LinkSendError> {
        use crate::link_send::{Completion, LinkSendError};
        let permit = self.tx.link_send_pool().try_acquire()?;
        let tracking = self.tx.link_send_pool().track(link_id, false);
        let (completion, receipt) = Completion::new_tracked(permit, tracking);
        self.tx
            .try_send(Event::SendLinkTracked {
                link_id,
                data,
                context,
                completion,
            })
            .map_err(|error| match error {
                std::sync::mpsc::TrySendError::Full(_) => LinkSendError::QueueFull,
                std::sync::mpsc::TrySendError::Disconnected(_) => LinkSendError::DriverStopped,
            })?;
        Ok(receipt)
    }

    /// Admit a best-effort direct Link datagram without blocking on queue space.
    ///
    /// Success means the driver accepted and encoded the packet. It does not
    /// imply remote delivery, acknowledgement, or interface transmission.
    pub fn try_send_link_datagram(
        &self,
        link_id: [u8; 16],
        data: Vec<u8>,
        context: u8,
        timeout: Duration,
    ) -> Result<(), crate::event::LinkDatagramError> {
        use crate::event::LinkDatagramError;
        if self.reject_new_work_if_draining().is_err() {
            return Err(LinkDatagramError::DriverStopped);
        }
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .try_send(Event::SendOnLink {
                link_id,
                data,
                context,
                response_tx: Some(response_tx),
            })
            .map_err(|error| match error {
                std::sync::mpsc::TrySendError::Full(_) => LinkDatagramError::DriverQueueFull,
                std::sync::mpsc::TrySendError::Disconnected(_) => LinkDatagramError::DriverStopped,
            })?;
        response_rx
            .recv_timeout(timeout)
            .map_err(|error| match error {
                std::sync::mpsc::RecvTimeoutError::Timeout => LinkDatagramError::AdmissionTimedOut,
                std::sync::mpsc::RecvTimeoutError::Disconnected => LinkDatagramError::DriverStopped,
            })?
    }

    /// Attach the same private ratchet owner used by the application's receiver.
    /// Reattaching that Arc is idempotent; conflicting owners are rejected. The
    /// owner remains attached across destination deregistration/re-registration.
    pub fn attach_local_ratchets(&self, owner: Arc<crate::LocalRatchets>) -> io::Result<()> {
        let mut owners = self.tx.local_ratchets().lock().unwrap();
        if let Some(existing) = owners.get(&owner.destination_hash()) {
            if !Arc::ptr_eq(existing, &owner) {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    "destination already has a ratchet owner",
                ));
            }
        }
        owners.insert(owner.destination_hash(), owner);
        Ok(())
    }

    /// Wait for capacity, then transmit an announcement on all selected local interfaces.
    ///
    /// Success means all selected writers accepted the complete packet, not
    /// remote delivery or subsequent relay. An error can follow partial delivery
    /// on one or more interfaces. No eligible interface returns `NoRoute`.
    /// Cancelling before admission sends nothing; after admission, transmission
    /// continues. Shared-client replay metadata travels in the same admitted
    /// event as the packet, never in a separate event from a rejected send.
    pub async fn announce(
        &self,
        dest: &crate::destination::Destination,
        identity: &Identity,
        app_data: Option<&[u8]>,
    ) -> Result<(), crate::TransmissionError> {
        let permit = self.tx.link_send_pool().acquire().await?;
        if self.announce_needs_stamp(dest, identity) {
            // Destination.announce with STAMP_DESTINATIONS and no stamp yet:
            // the stamper grinds it and submits the announce; the receipt
            // resolves when that announce has been written.
            let (completion, receipt) = crate::link_send::Completion::new(permit);
            self.enqueue_stamped_announce(dest, identity, app_data, Some(completion))
                .map_err(|_| crate::TransmissionError::QueueFull)?;
            return receipt.await;
        }
        let (packet, replay) = self
            .build_announce(dest, identity, app_data)
            .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
        let (completion, receipt) = crate::link_send::Completion::new(permit);
        self.tx
            .send_async(Event::SendOutboundTracked {
                raw: packet.raw,
                dest_type: dest.dest_type.to_wire_constant(),
                replay,
                completion,
            })
            .await?;
        receipt.await
    }

    /// Try admission immediately; await the receipt for the exact same result
    /// as `announce`. `QueueFull` guarantees nothing was admitted.
    pub fn try_announce(
        &self,
        dest: &crate::destination::Destination,
        identity: &Identity,
        app_data: Option<&[u8]>,
    ) -> Result<crate::TransmissionReceipt, crate::TransmissionError> {
        let permit = self.tx.link_send_pool().try_acquire()?;
        if self.announce_needs_stamp(dest, identity) {
            // Deferred to the stamper; the receipt resolves once the stamped
            // announce has been written, or with `Interrupted` if a newer
            // announce for the same destination replaces this one in the
            // queue.
            let (completion, receipt) = crate::link_send::Completion::new(permit);
            self.enqueue_stamped_announce(dest, identity, app_data, Some(completion))
                .map_err(|_| crate::TransmissionError::QueueFull)?;
            return Ok(receipt);
        }
        let (packet, replay) = self
            .build_announce(dest, identity, app_data)
            .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
        let (completion, receipt) = crate::link_send::Completion::new(permit);
        self.try_confirmed_event(Event::SendOutboundTracked {
            raw: packet.raw,
            dest_type: dest.dest_type.to_wire_constant(),
            replay,
            completion,
        })?;
        Ok(receipt)
    }

    /// Wait for capacity and local transmission, returning the hash for proof tracking.
    ///
    /// Uses the same all-selected-writers contract as `announce` and the
    /// same admission/completion capacity as `send_on_link`. No remote delivery
    /// is promised; errors may follow partial transmission. Dropping an admitted
    /// future does not cancel its send. No async runtime is required.
    pub async fn send_packet(
        &self,
        dest: &crate::destination::Destination,
        data: &[u8],
    ) -> Result<rns_core::types::PacketHash, crate::TransmissionError> {
        let permit = self.tx.link_send_pool().acquire().await?;
        let packet = self
            .build_packet(dest, data)
            .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
        let hash = rns_core::types::PacketHash(packet.packet_hash);
        let (completion, receipt) = crate::link_send::Completion::new(permit);
        self.tx
            .send_async(Event::SendOutboundTracked {
                raw: packet.raw,
                dest_type: dest.dest_type.to_wire_constant(),
                replay: None,
                completion,
            })
            .await?;
        receipt.await?;
        Ok(hash)
    }

    /// Try admission immediately. The receipt exposes the packet hash and
    /// resolves to the same result as `send_packet`, after local writes.
    /// `QueueFull` guarantees nothing was admitted and is safe to retry.
    pub fn try_send_packet(
        &self,
        dest: &crate::destination::Destination,
        data: &[u8],
    ) -> Result<crate::PacketSendReceipt, crate::TransmissionError> {
        let permit = self.tx.link_send_pool().try_acquire()?;
        let packet = self
            .build_packet(dest, data)
            .map_err(|_| crate::TransmissionError::PacketBuildFailed)?;
        let hash = rns_core::types::PacketHash(packet.packet_hash);
        let (completion, receipt) = crate::link_send::Completion::new(permit);
        self.try_confirmed_event(Event::SendOutboundTracked {
            raw: packet.raw,
            dest_type: dest.dest_type.to_wire_constant(),
            replay: None,
            completion,
        })?;
        Ok(crate::PacketSendReceipt {
            completion: receipt,
            hash,
        })
    }

    fn try_confirmed_event(&self, event: Event) -> Result<(), crate::TransmissionError> {
        self.tx.try_send(event).map_err(|error| match error {
            std::sync::mpsc::TrySendError::Full(_) => crate::TransmissionError::QueueFull,
            std::sync::mpsc::TrySendError::Disconnected(_) => {
                crate::TransmissionError::DriverStopped
            }
        })
    }

    /// Build and broadcast an announce for a destination (queue-only compatibility API).
    ///
    /// Success means driver submission, not transmission. Use `announce`
    /// or `try_announce` for confirmed local transmission.
    ///
    /// The identity is used to sign the announce. Must be the identity that
    /// owns the destination (i.e. `identity.hash()` matches `dest.identity_hash`).
    pub fn announce_queued(
        &self,
        dest: &crate::destination::Destination,
        identity: &Identity,
        app_data: Option<&[u8]>,
    ) -> Result<(), SendError> {
        self.reject_new_work_if_draining()?;
        if self.announce_needs_stamp(dest, identity) {
            // Destination.announce with STAMP_DESTINATIONS and no stamp yet
            // queues the announce for background stamping and emission.
            return self
                .enqueue_stamped_announce(dest, identity, app_data, None)
                .map_err(|_| SendError);
        }
        let (packet, replay) = self.build_announce(dest, identity, app_data)?;
        if let Some(replay) = replay {
            self.tx
                .send(Event::StoreSharedAnnounce {
                    dest_hash: replay.dest_hash,
                    name_hash: replay.name_hash,
                    identity_prv_key: replay.identity_prv_key,
                    app_data: replay.app_data,
                })
                .map_err(|_| SendError)?;
        }
        self.send_raw(packet.raw, dest.dest_type.to_wire_constant(), None)
    }

    fn build_announce(
        &self,
        dest: &crate::destination::Destination,
        identity: &Identity,
        app_data: Option<&[u8]>,
    ) -> Result<
        (
            rns_core::packet::RawPacket,
            Option<Box<crate::common::event::AnnounceReplay>>,
        ),
        SendError,
    > {
        build_announce_packet(
            &self.tx,
            self.stamp_store.as_ref(),
            self.stamp_destinations.load(Ordering::Relaxed),
            self.stamp_targets,
            dest,
            identity,
            app_data,
        )
    }

    /// Whether `Destination.announce` would hand this announce to the
    /// stamper: destination stamping is on and no persisted stamp reaches
    /// the destination's target value (Destination.py `_announce`).
    fn announce_needs_stamp(
        &self,
        dest: &crate::destination::Destination,
        identity: &Identity,
    ) -> bool {
        if !self.stamp_destinations.load(Ordering::Relaxed) || self.stamper.is_none() {
            return false;
        }
        let (Some(store), Some(public_key)) =
            (self.stamp_store.as_ref(), identity.get_public_key())
        else {
            return false;
        };
        let name_hash = rns_core::destination::name_hash(
            &dest.app_name,
            &dest.aspects.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
        );
        let material =
            crate::stamps::destination_stamp_material(&dest.hash.0, &public_key, &name_hash);
        store
            .load(
                &dest.hash.0,
                &material,
                self.stamp_targets.destination_target(dest),
            )
            .is_none()
    }

    /// `Transport.enqueue_stamped_announce`: the worker grinds the stamp and
    /// emits the announce; a full queue drops it.
    fn enqueue_stamped_announce(
        &self,
        dest: &crate::destination::Destination,
        identity: &Identity,
        app_data: Option<&[u8]>,
        completion: Option<crate::link_send::Completion>,
    ) -> Result<(), crate::stamper::QueueFull> {
        let stamper = self.stamper.as_ref().ok_or(crate::stamper::QueueFull)?;
        let identity_prv_key = identity
            .get_private_key()
            .ok_or(crate::stamper::QueueFull)?;
        let result = stamper.enqueue_announce(crate::stamper::AnnounceStampJob {
            dest: dest.clone(),
            identity_prv_key,
            app_data: app_data.map(|data| data.to_vec()),
            completion,
        });
        if result.is_err() {
            log::debug!(
                "Stamp queue full, dropping announce for {:02x?}",
                &dest.hash.0[..4]
            );
        }
        result
    }

    /// Use `store` for destination stamps, here and in the stamper.
    pub(crate) fn set_stamp_store(&mut self, store: crate::stamps::DestinationStampStore) {
        if let Some(stamper) = self.stamper.as_ref() {
            stamper.set_stamp_store(Some(store.clone()));
        }
        self.stamp_store = Some(store);
    }

    /// Override `STAMP_DESTINATIONS` and the stamp targets for this node and
    /// its stamper.
    #[doc(hidden)]
    pub fn set_destination_stamping(
        &mut self,
        enabled: bool,
        targets: crate::stamper::StampTargets,
    ) {
        self.stamp_destinations.store(enabled, Ordering::Relaxed);
        self.stamp_targets = targets;
        if let Some(stamper) = self.stamper.as_ref() {
            stamper.set_targets(targets);
        }
    }

    /// The stamper shared with the driver.
    #[doc(hidden)]
    pub fn stamper(&self) -> Option<&Arc<crate::stamper::Stamper>> {
        self.stamper.as_ref()
    }

    /// Queue an encrypted (SINGLE) or plaintext (PLAIN) packet to a destination.
    ///
    /// Queue-only compatibility API: use `send_packet` or `try_send_packet`
    /// for confirmed local transmission. The returned hash does not mean sent.
    ///
    /// For SINGLE destinations, `dest.public_key` must be set (OUT direction).
    /// Returns the packet hash for proof tracking.
    pub fn send_packet_queued(
        &self,
        dest: &crate::destination::Destination,
        data: &[u8],
    ) -> Result<rns_core::types::PacketHash, SendError> {
        self.reject_new_work_if_draining()?;
        let packet = self.build_packet(dest, data)?;
        let packet_hash = rns_core::types::PacketHash(packet.packet_hash);
        self.tx
            .send(Event::SendOutbound {
                raw: packet.raw,
                dest_type: dest.dest_type.to_wire_constant(),
                attached_interface: None,
            })
            .map_err(|_| SendError)?;
        Ok(packet_hash)
    }

    /// Deliver single packets addressed to `dest`, a local destination,
    /// decrypted: through [`Callbacks::on_packet_received`](crate::Callbacks::on_packet_received)
    /// in place of `on_local_delivery`, decrypted on the node's proof pool
    /// when it has one, so a receiver is not bound by decrypting on its own
    /// thread. They are decrypted with the destination's attached ratchets
    /// ([`attach_local_ratchets`](Self::attach_local_ratchets)) when it has
    /// them, else with `identity`, which must hold its private keys. Packets
    /// that do not decrypt are dropped and not proved. Deregistering the
    /// destination ends it.
    pub fn deliver_decrypted(
        &self,
        dest: &crate::destination::Destination,
        identity: &rns_crypto::identity::Identity,
    ) -> Result<(), SendError> {
        let private = identity.get_private_key().ok_or(SendError)?;
        self.tx
            .send(Event::DeliverDecrypted {
                dest_hash: dest.hash.0,
                identity: Box::new(rns_crypto::identity::Identity::from_private_key(&private)),
            })
            .map_err(|_| SendError)
    }

    /// Queue a packet for a SINGLE destination without encrypting it on this
    /// thread, unlike [`send_packet_queued`](Self::send_packet_queued): the
    /// node encrypts it (on its proof pool when it has one; on this thread
    /// when the process may use one CPU only) and reports the
    /// packet hash through [`Callbacks::on_packet_sent`](crate::Callbacks::on_packet_sent)
    /// with the returned ticket, before sending it. A sender is then not
    /// bound by its own thread's encryption, one X25519 key generation and
    /// exchange per packet.
    ///
    /// `dest.public_key` must be set (OUT direction). Like
    /// `send_packet_queued`, queueing does not mean sent.
    pub fn send_packet_deferred(
        &self,
        dest: &crate::destination::Destination,
        data: &[u8],
    ) -> Result<crate::SendTicket, SendError> {
        self.reject_new_work_if_draining()?;
        if dest.dest_type != rns_core::types::DestinationType::Single {
            return Err(SendError);
        }
        let public_key = dest.public_key.ok_or(SendError)?;
        let ticket = self.next_send_ticket.fetch_add(1, Ordering::Relaxed);
        let ratchet = self.single_ratchet(dest);
        let (data, encrypted) = if self.encrypt_deferred_inline {
            let payload = rns_crypto::identity::Identity::encrypt_for_public_key(
                &public_key,
                data,
                ratchet.as_ref(),
                &mut OsRng,
            )
            .map_err(|_| SendError)?;
            (payload, true)
        } else {
            (data.to_vec(), false)
        };
        self.tx
            .send(Event::SendDeferred(Box::new(
                crate::common::event::DeferredSend {
                    ticket,
                    destination: dest.hash.0,
                    public_key,
                    ratchet,
                    data,
                    encrypted,
                },
            )))
            .map_err(|_| SendError)?;
        Ok(crate::SendTicket(ticket))
    }

    fn build_packet(
        &self,
        dest: &crate::destination::Destination,
        data: &[u8],
    ) -> Result<rns_core::packet::RawPacket, SendError> {
        use rns_core::types::DestinationType;

        let payload = match dest.dest_type {
            DestinationType::Single => self.encrypt_single_payload(dest, data)?,
            DestinationType::Plain => data.to_vec(),
            DestinationType::Group => dest.encrypt(data).map_err(|_| SendError)?,
        };

        let flags = rns_core::packet::PacketFlags {
            header_type: rns_core::constants::HEADER_1,
            context_flag: rns_core::constants::FLAG_UNSET,
            transport_type: rns_core::constants::TRANSPORT_BROADCAST,
            destination_type: dest.dest_type.to_wire_constant(),
            packet_type: rns_core::constants::PACKET_TYPE_DATA,
        };

        let packet = rns_core::packet::RawPacket::pack(
            flags,
            0,
            &dest.hash.0,
            None,
            rns_core::constants::CONTEXT_NONE,
            &payload,
        )
        .map_err(|_| SendError)?;

        Ok(packet)
    }

    fn encrypt_single_payload(
        &self,
        dest: &crate::destination::Destination,
        data: &[u8],
    ) -> Result<Vec<u8>, SendError> {
        let pub_key = dest.public_key.ok_or(SendError)?;
        let ratchet = self.single_ratchet(dest);
        // Encrypted straight from the key bytes: building the identity would
        // decompress its signing key on every send, and encryption does not
        // use it.
        rns_crypto::identity::Identity::encrypt_for_public_key(
            &pub_key,
            data,
            ratchet.as_ref(),
            &mut OsRng,
        )
        .map_err(|_| SendError)
    }

    /// The ratchet to encrypt a packet for `dest` with: this node's own for
    /// its destinations, else the latest one `dest` announced.
    fn single_ratchet(&self, dest: &crate::destination::Destination) -> Option<[u8; 32]> {
        let local_ratchet = self
            .tx
            .local_ratchets()
            .lock()
            .unwrap()
            .get(&dest.hash.0)
            .and_then(|owner| owner.current_public());
        local_ratchet.or_else(|| {
            self.ratchet_store.as_ref().and_then(|store| {
                match store.current(&dest.hash.0, time::now(), self.ratchet_expiry.as_secs_f64()) {
                    Ok(entry) => entry.map(|entry| entry.ratchet),
                    Err(err) => {
                        log::warn!(
                            "failed to load ratchet for {:02x}{:02x}{:02x}{:02x}..: {}",
                            dest.hash.0[0],
                            dest.hash.0[1],
                            dest.hash.0[2],
                            dest.hash.0[3],
                            err
                        );
                        None
                    }
                }
            })
        })
    }

    /// Register a destination with the transport engine and set its proof strategy.
    ///
    /// `signing_key` is the full 64-byte identity private key (X25519 32 bytes +
    /// Ed25519 32 bytes), needed for ProveAll/ProveApp to sign proof packets.
    pub fn register_destination_with_proof(
        &self,
        dest: &crate::destination::Destination,
        signing_key: Option<[u8; 64]>,
    ) -> Result<(), SendError> {
        // Register with transport engine
        self.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())?;

        // Register proof strategy if not ProveNone
        if dest.proof_strategy != rns_core::types::ProofStrategy::ProveNone {
            self.tx
                .send(Event::RegisterProofStrategy {
                    dest_hash: dest.hash.0,
                    strategy: dest.proof_strategy,
                    signing_key,
                })
                .map_err(|_| SendError)?;
        }

        Ok(())
    }

    /// Request a path to a destination from the network.
    pub fn request_path(&self, dest_hash: &rns_core::types::DestHash) -> Result<(), SendError> {
        self.reject_new_work_if_draining()?;
        self.tx
            .send(Event::RequestPath {
                dest_hash: dest_hash.0,
            })
            .map_err(|_| SendError)
    }

    /// Check if a path exists to a destination (synchronous query).
    pub fn has_path(&self, dest_hash: &rns_core::types::DestHash) -> Result<bool, SendError> {
        match self.query(QueryRequest::HasPath {
            dest_hash: dest_hash.0,
        })? {
            QueryResponse::HasPath(v) => Ok(v),
            _ => Ok(false),
        }
    }

    /// Get hop count to a destination (synchronous query).
    pub fn hops_to(&self, dest_hash: &rns_core::types::DestHash) -> Result<Option<u8>, SendError> {
        match self.query(QueryRequest::HopsTo {
            dest_hash: dest_hash.0,
        })? {
            QueryResponse::HopsTo(v) => Ok(v),
            _ => Ok(None),
        }
    }

    /// Recall the identity information for a previously announced destination.
    pub fn recall_identity(
        &self,
        dest_hash: &rns_core::types::DestHash,
    ) -> Result<Option<crate::destination::AnnouncedIdentity>, SendError> {
        match self.query(QueryRequest::RecallIdentity {
            dest_hash: dest_hash.0,
        })? {
            QueryResponse::RecallIdentity(v) => Ok(v),
            _ => Ok(None),
        }
    }

    /// Stamp value and announced link request cost recorded from a known
    /// destination's last announce.
    pub fn recall_destination_stamp(
        &self,
        dest_hash: &rns_core::types::DestHash,
    ) -> Result<Option<crate::event::DestinationStampInfo>, SendError> {
        match self.query(QueryRequest::RecallDestinationStamp {
            dest_hash: dest_hash.0,
        })? {
            QueryResponse::RecallDestinationStamp(v) => Ok(v),
            _ => Ok(None),
        }
    }

    /// List known destinations and their lifecycle state.
    pub fn known_destinations(
        &self,
    ) -> Result<Vec<crate::event::KnownDestinationEntry>, SendError> {
        match self.query(QueryRequest::KnownDestinations)? {
            QueryResponse::KnownDestinations(entries) => Ok(entries),
            _ => Ok(Vec::new()),
        }
    }

    /// Mark a known destination as retained.
    pub fn retain_known_destination(
        &self,
        dest_hash: &rns_core::types::DestHash,
    ) -> Result<bool, SendError> {
        match self.query(QueryRequest::RetainKnownDestination {
            dest_hash: dest_hash.0,
        })? {
            QueryResponse::RetainKnownDestination(ok) => Ok(ok),
            _ => Ok(false),
        }
    }

    /// Mark all known destinations announced by an identity as retained.
    pub fn retain_identity(
        &self,
        identity_hash: &rns_core::types::IdentityHash,
    ) -> Result<bool, SendError> {
        match self.query(QueryRequest::RetainIdentity {
            identity_hash: identity_hash.0,
        })? {
            QueryResponse::RetainIdentity(ok) => Ok(ok),
            _ => Ok(false),
        }
    }

    /// Clear the retained flag on a known destination.
    pub fn unretain_known_destination(
        &self,
        dest_hash: &rns_core::types::DestHash,
    ) -> Result<bool, SendError> {
        match self.query(QueryRequest::UnretainKnownDestination {
            dest_hash: dest_hash.0,
        })? {
            QueryResponse::UnretainKnownDestination(ok) => Ok(ok),
            _ => Ok(false),
        }
    }

    /// Mark a known destination as used.
    pub fn mark_known_destination_used(
        &self,
        dest_hash: &rns_core::types::DestHash,
    ) -> Result<bool, SendError> {
        match self.query(QueryRequest::MarkKnownDestinationUsed {
            dest_hash: dest_hash.0,
        })? {
            QueryResponse::MarkKnownDestinationUsed(ok) => Ok(ok),
            _ => Ok(false),
        }
    }

    fn persist_known_destinations(&self) {
        if self.known_destinations_store.is_none() && self.known_destinations_path.is_none() {
            return;
        }

        let Ok(entries) = self.known_destinations() else {
            return;
        };

        let destinations: std::collections::HashMap<[u8; 16], storage::KnownDestination> = entries
            .into_iter()
            .map(|entry| {
                (
                    entry.dest_hash,
                    storage::KnownDestination {
                        identity_hash: entry.identity_hash,
                        public_key: entry.public_key,
                        app_data: entry.app_data,
                        hops: entry.hops,
                        received_at: entry.received_at,
                        receiving_interface: entry.receiving_interface.0,
                        was_used: entry.was_used,
                        last_used_at: entry.last_used_at,
                        retained: entry.retained,
                        stamp: entry.stamp.map(|stamp| stamp.to_vec()),
                        stamp_value: entry.stamp_value,
                        lr_cost: entry.lr_cost,
                    },
                )
            })
            .collect();

        if let Some(store) = self.known_destinations_store.as_ref() {
            if let Err(err) = store.save_all(&destinations) {
                log::warn!("failed to persist known destinations: {}", err);
            }
        } else if let Some(path) = self.known_destinations_path.as_ref() {
            if let Err(err) = storage::save_known_destinations(&destinations, path) {
                log::warn!("failed to persist known destinations: {}", err);
            }
        }
    }

    fn persist_transport_state(&self) {
        let Some(storage_dir) = self.transport_state_dir.as_ref() else {
            return;
        };
        let Ok(QueryResponse::TransportStateSnapshot(snapshot)) =
            self.query(QueryRequest::TransportStateSnapshot)
        else {
            return;
        };
        if let Err(err) = storage::save_transport_state(&snapshot, storage_dir, true) {
            log::warn!("failed to persist transport state: {}", err);
        }
    }

    /// Load an in-memory WASM hook at runtime.
    pub fn load_hook(
        &self,
        name: String,
        wasm_bytes: Vec<u8>,
        attach_point: String,
        priority: i32,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::LoadHook {
                name,
                wasm_bytes,
                attach_point,
                priority,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Load a hook from a server-local filesystem path at runtime.
    pub fn load_hook_file(
        &self,
        name: String,
        path: String,
        hook_type: String,
        attach_point: String,
        priority: i32,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::LoadHookFile {
                name,
                path,
                hook_type,
                attach_point,
                priority,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Load a registered built-in hook at runtime.
    pub fn load_builtin_hook(
        &self,
        name: String,
        builtin_id: String,
        attach_point: String,
        priority: i32,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::LoadBuiltinHook {
                name,
                builtin_id,
                attach_point,
                priority,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Unload a hook at runtime.
    pub fn unload_hook(
        &self,
        name: String,
        attach_point: String,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::UnloadHook {
                name,
                attach_point,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Reload an in-memory WASM hook at runtime (detach + recompile + reattach with same priority).
    pub fn reload_hook(
        &self,
        name: String,
        attach_point: String,
        wasm_bytes: Vec<u8>,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::ReloadHook {
                name,
                attach_point,
                wasm_bytes,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Reload a hook from a server-local filesystem path at runtime.
    pub fn reload_hook_file(
        &self,
        name: String,
        attach_point: String,
        path: String,
        hook_type: String,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::ReloadHookFile {
                name,
                attach_point,
                path,
                hook_type,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Reload a registered built-in hook at runtime.
    pub fn reload_builtin_hook(
        &self,
        name: String,
        attach_point: String,
        builtin_id: String,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::ReloadBuiltinHook {
                name,
                attach_point,
                builtin_id,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Enable or disable a loaded hook at runtime.
    pub fn set_hook_enabled(
        &self,
        name: String,
        attach_point: String,
        enabled: bool,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::SetHookEnabled {
                name,
                attach_point,
                enabled,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Update the priority of a loaded hook at runtime.
    pub fn set_hook_priority(
        &self,
        name: String,
        attach_point: String,
        priority: i32,
    ) -> Result<Result<(), String>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::SetHookPriority {
                name,
                attach_point,
                priority,
                response_tx,
            })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// List all loaded hooks.
    pub fn list_hooks(&self) -> Result<Vec<crate::event::HookInfo>, SendError> {
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        self.tx
            .send(Event::ListHooks { response_tx })
            .map_err(|_| SendError)?;
        response_rx.recv().map_err(|_| SendError)
    }

    /// Construct an RnsNode from its constituent parts.
    /// Used by `shared_client` to build a client-mode node.
    #[cfg(any(feature = "iface-local", test))]
    pub(crate) fn from_parts(
        tx: EventSender,
        driver_handle: thread::JoinHandle<()>,
        rpc_server: Option<crate::rpc::RpcServer>,
        tick_interval_ms: Arc<AtomicU64>,
        accepting_work: Arc<AtomicBool>,
    ) -> Self {
        RnsNode {
            tx,
            driver_handle: Some(driver_handle),
            verify_handle: None,
            verify_shutdown: Arc::default(),
            persist_handle: None,
            persist_stop: None,
            rpc_server,
            #[cfg(target_os = "linux")]
            rpc_unix_server: None,
            tick_interval_ms,
            next_send_ticket: AtomicU64::new(0),
            encrypt_deferred_inline: std::thread::available_parallelism()
                .map_or(true, |cpus| cpus.get() == 1),
            accepting_work,
            probe_server: None,
            known_destinations_path: None,
            known_destinations_store: None,
            stamp_store: None,
            stamp_destinations: Arc::new(AtomicBool::new(rns_core::constants::STAMP_DESTINATIONS)),
            stamp_targets: crate::stamper::StampTargets::default(),
            stamper: None,
            transport_state_dir: None,
            ratchet_store: None,
            ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
            #[cfg(feature = "iface-local")]
            destination_types: Default::default(),
        }
    }

    /// Get the event sender for direct event injection.
    pub fn event_sender(&self) -> &EventSender {
        &self.tx
    }

    /// Set the tick interval in milliseconds.
    /// Default is 1000 (1 second). Changes take effect on the next tick cycle.
    /// Values are clamped to the range 100..=10000.
    /// Returns the actual stored value (which may differ from `ms` if clamped).
    pub fn set_tick_interval(&self, ms: u64) -> u64 {
        let clamped = ms.clamp(100, 10_000);
        if clamped != ms {
            log::warn!(
                "tick interval {}ms out of range, clamped to {}ms",
                ms,
                clamped
            );
        }
        self.tick_interval_ms.store(clamped, Ordering::Relaxed);
        clamped
    }

    /// Get the current tick interval in milliseconds.
    pub fn tick_interval(&self) -> u64 {
        self.tick_interval_ms.load(Ordering::Relaxed)
    }

    /// Shut down the node. Blocks until the driver thread exits.
    pub fn shutdown(mut self) {
        self.shutdown_inner();
    }

    fn shutdown_inner(&mut self) {
        #[cfg(feature = "iface-local")]
        self.destination_types.withdraw_from_shared_instance();
        if self.driver_handle.is_none()
            && self.verify_handle.is_none()
            && self.persist_handle.is_none()
            && self.rpc_server.is_none()
            && {
                #[cfg(target_os = "linux")]
                {
                    self.rpc_unix_server.is_none()
                }
                #[cfg(not(target_os = "linux"))]
                {
                    true
                }
            }
        {
            return;
        }
        // Stop RPC server first
        if let Some(mut rpc) = self.rpc_server.take() {
            rpc.stop();
        }
        #[cfg(target_os = "linux")]
        if let Some(mut rpc) = self.rpc_unix_server.take() {
            rpc.stop();
        }
        self.persist_known_destinations();
        self.persist_transport_state();
        if let Some(stop) = self.persist_stop.take() {
            let _ = stop.send(());
        }
        if let Some(handle) = self.persist_handle.take() {
            let _ = handle.join();
        }
        self.verify_shutdown.stop();
        let _ = self.tx.send(Event::Shutdown);
        if let Some(handle) = self.driver_handle.take() {
            let _ = handle.join();
        }
        // Reader/ticker EventSender clones may outlive the driver. They must not
        // retain private owners (and their file locks) after node shutdown.
        self.tx.local_ratchets().lock().unwrap().clear();
        if let Some(handle) = self.verify_handle.take() {
            let _ = handle.join();
        }
    }
}

/// Make sure a destination stamp reaching the destination's target value is
/// persisted for `dest`, grinding one at that target when none is. This is
/// the stamp half of `Destination._announce`, run by the stamper off the
/// calling thread.
///
/// Since d7749b0b upstream reloads or regenerates the held stamp whenever
/// its value is below `target_value`, and both the store lookup and the
/// grind use `target_value` rather than `DESTINATION_COST` /
/// `DESTINATION_VALUE` (Destination.py:319-323). A persisted stamp that
/// still meets the announce cost but not the target is therefore replaced.
pub(crate) fn ensure_destination_stamp(
    store: &crate::stamps::DestinationStampStore,
    targets: crate::stamper::StampTargets,
    dest: &crate::destination::Destination,
    identity: &Identity,
) -> Result<(), String> {
    let public_key = identity
        .get_public_key()
        .ok_or_else(|| "identity has no public key".to_string())?;
    let name_hash = rns_core::destination::name_hash(
        &dest.app_name,
        &dest.aspects.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
    );
    let material = crate::stamps::destination_stamp_material(&dest.hash.0, &public_key, &name_hash);
    let target = targets.destination_target(dest);
    if store.load(&dest.hash.0, &material, target).is_some() {
        return Ok(());
    }
    let started = std::time::Instant::now();
    let (stamp, value) =
        crate::stamps::generate_stamp(&material, target, rns_core::constants::STAMP_LENGTH, 0)
            .ok_or_else(|| "stamp cost is above the protocol cap".to_string())?;
    if let Err(error) = store.store(&dest.hash.0, &stamp, value) {
        log::warn!("failed to persist destination stamp: {error}");
    }
    log::debug!(
        "Generated and persisted stamp with value {value} for {:02x?} in {:?}",
        &dest.hash.0[..4],
        started.elapsed()
    );
    Ok(())
}

/// Build an announce packet for `dest`, signed by `identity`.
///
/// With `stamp_destinations` the app data is extended with the destination
/// stamp from `stamp_store`, grinding one inline when none is persisted, as
/// `Destination._announce` does. The node API checks for a missing stamp
/// first and defers to the stamper, so the inline grind only runs when a
/// stamp disappeared between that check and this call.
pub(crate) fn build_announce_packet(
    tx: &EventSender,
    stamp_store: Option<&crate::stamps::DestinationStampStore>,
    stamp_destinations: bool,
    targets: crate::stamper::StampTargets,
    dest: &crate::destination::Destination,
    identity: &Identity,
    app_data: Option<&[u8]>,
) -> Result<
    (
        rns_core::packet::RawPacket,
        Option<Box<crate::common::event::AnnounceReplay>>,
    ),
    SendError,
> {
    let name_hash = rns_core::destination::name_hash(
        &dest.app_name,
        &dest.aspects.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
    );

    let mut random_hash = [0u8; 10];
    OsRng.fill_bytes(&mut random_hash[..5]);
    // Bytes [5:10] must be the emission timestamp (seconds since epoch,
    // big-endian, truncated to 5 bytes) so that path table dedup can
    // compare announce freshness.  Matches Python: int(time.time()).to_bytes(5, "big")
    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();
    random_hash[5..10].copy_from_slice(&now_secs.to_be_bytes()[3..8]);

    let owner = tx
        .local_ratchets()
        .lock()
        .unwrap()
        .get(&dest.hash.0)
        .cloned();
    let ratchet = if let Some(owner) = owner {
        if dest.direction != rns_core::types::Direction::In
            || owner.identity_hash() != *identity.hash()
        {
            return Err(SendError);
        }
        Some(owner.public_for_announce(now_secs).map_err(|error| {
            log::error!("local ratchet announcement failed: {error}");
            SendError
        })?)
    } else {
        None
    };
    let wire_app_data: Option<Vec<u8>> = if stamp_destinations {
        let public_key = identity.get_public_key().ok_or(SendError)?;
        let material =
            crate::stamps::destination_stamp_material(&dest.hash.0, &public_key, &name_hash);
        let store = stamp_store.ok_or(SendError)?;
        let target = targets.destination_target(dest);
        let (stamp, _value) = match store.load(&dest.hash.0, &material, target) {
            Some(entry) => entry,
            None => {
                let (stamp, value) = crate::stamps::generate_stamp(
                    &material,
                    target,
                    rns_core::constants::STAMP_LENGTH,
                    0,
                )
                .ok_or(SendError)?;
                if let Err(error) = store.store(&dest.hash.0, &stamp, value) {
                    log::warn!("failed to persist destination stamp: {error}");
                }
                (stamp, value)
            }
        };
        let mut wire = app_data.map(|data| data.to_vec()).unwrap_or_default();
        // `cost_to_bytes(self.get_link_cost())`: the destination's own cost,
        // or the default `LR_COST` when none was set, fraction included.
        wire.extend_from_slice(&crate::stamps::announce_ext_payload(
            &stamp,
            0,
            tx.link_request_costs().get(&dest.hash.0),
        ));
        Some(wire)
    } else {
        app_data.map(|data| data.to_vec())
    };

    let (announce_data, has_ratchet) = rns_core::announce::AnnounceData::pack(
        identity,
        &dest.hash.0,
        &name_hash,
        &random_hash,
        ratchet.as_ref(),
        wire_app_data.as_deref(),
    )
    .map_err(|_| SendError)?;

    let context_flag = if has_ratchet {
        rns_core::constants::FLAG_SET
    } else {
        rns_core::constants::FLAG_UNSET
    };

    let flags = rns_core::packet::PacketFlags {
        header_type: rns_core::constants::HEADER_1,
        context_flag,
        transport_type: rns_core::constants::TRANSPORT_BROADCAST,
        destination_type: rns_core::constants::DESTINATION_SINGLE,
        packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
    };

    let packet = rns_core::packet::RawPacket::pack(
        flags,
        0,
        &dest.hash.0,
        None,
        rns_core::constants::CONTEXT_NONE,
        &announce_data,
    )
    .map_err(|_| SendError)?;

    let replay = if dest.dest_type == rns_core::types::DestinationType::Single {
        identity.get_private_key().map(|identity_prv_key| {
            Box::new(crate::common::event::AnnounceReplay {
                dest_hash: dest.hash.0,
                name_hash,
                identity_prv_key,
                app_data: wire_app_data,
            })
        })
    } else {
        None
    };
    Ok((packet, replay))
}

impl Drop for RnsNode {
    fn drop(&mut self) {
        self.shutdown_inner();
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::driver::IfacRuntimeConfig;
    use crate::storage::RatchetStore;
    use rns_core::stamp::StampCost;
    use std::fs;
    use tempfile::tempdir;

    #[test]
    fn local_ratchet_announcements_and_self_send() {
        let node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();
        let identity = Identity::from_private_key(&[42; 64]);
        let dest = crate::Destination::single_in(
            "test",
            &["ratchet"],
            rns_core::types::IdentityHash(*identity.hash()),
        );
        let owner = crate::LocalRatchets::with_store(
            &dest,
            Identity::from_private_key(&[42; 64]),
            Arc::new(crate::MemoryRatchetStore::default()),
        )
        .unwrap();
        owner.enforce();
        node.attach_local_ratchets(owner.clone()).unwrap();
        node.attach_local_ratchets(owner.clone()).unwrap();
        let conflicting = crate::LocalRatchets::with_store(
            &dest,
            Identity::from_private_key(&[42; 64]),
            Arc::new(crate::MemoryRatchetStore::default()),
        )
        .unwrap();
        assert!(node.attach_local_ratchets(conflicting).is_err());
        let (packet, _) = node.build_announce(&dest, &identity, Some(b"app")).unwrap();
        assert_eq!(packet.flags.context_flag, rns_core::constants::FLAG_SET);
        let announce = rns_core::announce::AnnounceData::unpack(packet.data(), true).unwrap();
        announce.validate(&dest.hash.0).unwrap();
        assert_eq!(announce.ratchet, owner.current_public());
        assert!(node
            .build_announce(&dest, &Identity::from_private_key(&[43; 64]), None)
            .is_err());
        assert!(node
            .build_announce(&dest, &identity, Some(&vec![0; 500]))
            .is_err());
        let mut outgoing = dest.clone();
        outgoing.direction = rns_core::types::Direction::Out;
        outgoing.public_key = identity.get_public_key();
        let encrypted = node.encrypt_single_payload(&outgoing, b"self").unwrap();
        assert_eq!(owner.decrypt(&encrypted).unwrap().plaintext, b"self");
        assert!(identity.decrypt(&encrypted).is_err());
    }

    struct NoopCallbacks;

    impl Callbacks for NoopCallbacks {
        fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
        fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
        fn on_local_delivery(
            &mut self,
            _: rns_core::types::DestHash,
            _: Vec<u8>,
            _: rns_core::types::PacketHash,
        ) {
        }
    }

    #[test]
    fn managed_interface_attaches_disabled_section_and_reloads_from_disk() {
        use crate::event::InterfaceManagementOperation as Operation;

        let dir = tempdir().unwrap();
        let config_file = dir.path().join("config");
        let write_config = |port| {
            fs::write(&config_file, format!("[reticulum]\nshare_instance = no\n\n[interfaces]\n  [[Managed UDP]]\n    type = UDPInterface\n    enabled = no\n    forward_ip = 127.0.0.1\n    forward_port = {port}\n"))
            .unwrap();
        };
        write_config(9);
        let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
        let manage = |operation| {
            node.query(QueryRequest::ManageInterface {
                operation,
                name: "Managed UDP".into(),
            })
            .unwrap()
        };
        assert!(matches!(
            manage(Operation::Detach),
            QueryResponse::InterfaceManagement(None)
        ));
        assert!(matches!(
            manage(Operation::Attach),
            QueryResponse::InterfaceManagement(Some(true))
        ));
        assert!(matches!(
            manage(Operation::Attach),
            QueryResponse::InterfaceManagement(Some(false))
        ));
        write_config(10);
        assert!(matches!(
            manage(Operation::Reload),
            QueryResponse::InterfaceManagement(Some(true))
        ));
        assert!(matches!(
            manage(Operation::Detach),
            QueryResponse::InterfaceManagement(Some(true))
        ));
        assert!(matches!(
            manage(Operation::Detach),
            QueryResponse::InterfaceManagement(None)
        ));
        node.shutdown();
    }

    #[test]
    fn interface_management_can_be_disabled() {
        use crate::event::InterfaceManagementOperation as Operation;

        let dir = tempdir().unwrap();
        fs::write(
            dir.path().join("config"),
            "[reticulum]\nshare_instance = no\nenable_interface_management = no\n",
        )
        .unwrap();
        let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
        assert!(matches!(
            node.query(QueryRequest::ManageInterface {
                operation: Operation::Attach,
                name: "missing".into(),
            })
            .unwrap(),
            QueryResponse::InterfaceManagement(Some(false))
        ));
        node.shutdown();
    }

    #[cfg(feature = "iface-tcp")]
    #[test]
    fn managed_tcp_listener_releases_port_on_detach() {
        use crate::event::InterfaceManagementOperation as Operation;
        use std::io::Read;

        let dir = tempdir().unwrap();
        let port = crate::test_support::port();
        fs::write(dir.path().join("config"), format!(
            "[reticulum]\nshare_instance = no\n\n[interfaces]\n  [[Managed TCP]]\n    type = TCPServerInterface\n    enabled = no\n    listen_ip = 127.0.0.1\n    listen_port = {port}\n"
        )).unwrap();
        let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
        let manage = |operation| {
            node.query(QueryRequest::ManageInterface {
                operation,
                name: "Managed TCP".into(),
            })
            .unwrap()
        };
        assert!(matches!(
            manage(Operation::Attach),
            QueryResponse::InterfaceManagement(Some(true))
        ));
        let mut client = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
        client
            .set_read_timeout(Some(Duration::from_secs(2)))
            .unwrap();
        thread::sleep(Duration::from_millis(100));
        assert!(matches!(
            manage(Operation::Detach),
            QueryResponse::InterfaceManagement(Some(true))
        ));
        let deadline = std::time::Instant::now() + Duration::from_secs(2);
        while std::net::TcpStream::connect(("127.0.0.1", port)).is_ok()
            && std::time::Instant::now() < deadline
        {
            thread::sleep(Duration::from_millis(25));
        }
        assert!(std::net::TcpStream::connect(("127.0.0.1", port)).is_err());
        assert_eq!(client.read(&mut [0u8; 1]).unwrap(), 0);
        assert!(matches!(
            manage(Operation::Attach),
            QueryResponse::InterfaceManagement(Some(true))
        ));
        assert!(std::net::TcpStream::connect(("127.0.0.1", port)).is_ok());
        node.shutdown();
    }

    #[test]
    fn shared_instance_rpc_manages_interface_by_name() {
        use crate::pickle::PickleValue;
        use crate::rpc::{derive_auth_key, RpcAddr, RpcClient};

        let dir = tempdir().unwrap();
        let rpc_port = crate::test_support::port();
        let shared_port = crate::test_support::port();
        fs::write(dir.path().join("config"), format!(
            "[reticulum]\nshare_instance = yes\ninstance_control_port = {rpc_port}\nshared_instance_port = {shared_port}\n\n[interfaces]\n  [[Managed UDP]]\n    type = UDPInterface\n    enabled = no\n    forward_ip = 127.0.0.1\n    forward_port = 9\n"
        )).unwrap();
        let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
        let paths = storage::ensure_storage_dirs(dir.path()).unwrap();
        let identity = storage::load_or_create_identity(&paths.identities).unwrap();
        let key = derive_auth_key(&identity.get_private_key().unwrap());
        let call = |action: &str| {
            let mut client =
                RpcClient::connect(&RpcAddr::Tcp("127.0.0.1".into(), rpc_port), &key).unwrap();
            client
                .call(&PickleValue::Dict(vec![
                    (
                        PickleValue::String("manage".into()),
                        PickleValue::String(action.into()),
                    ),
                    (
                        PickleValue::String("name".into()),
                        PickleValue::String("Managed UDP".into()),
                    ),
                ]))
                .unwrap()
        };
        assert_eq!(call("attach_interface"), PickleValue::Bool(true));
        assert_eq!(call("reload_interface"), PickleValue::Bool(true));
        assert_eq!(call("detach_interface"), PickleValue::Bool(true));
        assert_eq!(call("detach_interface"), PickleValue::None);
        node.shutdown();
    }

    struct TestWriter;

    impl crate::interface::Writer for TestWriter {
        fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
            Ok(())
        }
    }

    fn test_transport_config(transport_enabled: bool) -> TransportConfig {
        TransportConfig {
            transport_enabled,
            identity_hash: None,
            local_hops_delta: 0,
            prefer_shorter_path: false,
            max_paths_per_destination: 1,
            packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
            packet_hashlist_allocation: rns_core::transport::types::PacketHashlistAllocation::Eager,
            max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
            max_path_destinations: usize::MAX,
            max_tunnel_destinations_total: usize::MAX,
            destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
            announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
            announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
            announce_sig_cache_enabled: true,
            announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
            announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
            announce_queue_max_entries: 256,
            announce_queue_max_interfaces: 1024,
        }
    }

    fn test_interface_info(id: u64) -> rns_core::transport::types::InterfaceInfo {
        rns_core::transport::types::InterfaceInfo {
            stamp_costs: Default::default(),
            id: rns_core::transport::types::InterfaceId(id),
            name: format!("test-{id}"),
            mode: rns_core::constants::MODE_FULL,
            gravity: 0,
            recursive_prs: false,
            announces_from_internal: true,
            announces_to_internal: None,
            out_capable: true,
            in_capable: true,
            bitrate: None,
            airtime_profile: None,
            announce_rate_target: None,
            announce_rate_grace: 0,
            announce_rate_penalty: 0.0,
            announce_cap: rns_core::constants::ANNOUNCE_CAP,
            is_local_client: false,
            wants_tunnel: false,
            tunnel_id: None,
            mtu: rns_core::constants::MTU as u32,
            ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
            ia_freq: 0.0,
            ip_freq: 0.0,
            op_freq: 0.0,
            op_samples: 0,
            started: 0.0,
        }
    }

    #[test]
    fn static_interface_registration_applies_transport_announce_rate_defaults() {
        let (tx, rx) = crate::event::channel();
        let mut driver = Driver::new(
            test_transport_config(true),
            rx,
            tx.clone(),
            Box::new(NoopCallbacks),
        );
        driver.set_announce_rate_defaults(AnnounceRateDefaults {
            target: Some(7200.0),
            penalty: 15.0,
            grace: 7,
        });

        let id = rns_core::transport::types::InterfaceId(7);
        register_started_interface(StartedInterface {
            driver: &mut driver,
            tx: &tx,
            queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
            id,
            info: test_interface_info(id.0),
            writer: Box::new(TestWriter),
            interface_type_name: "TestInterface".to_string(),
            ifac_state: None,
            ifac_runtime: &IfacRuntimeConfig {
                netname: None,
                netkey: None,
                size: 16,
            },
            stamp_costs: Default::default(),
        });

        let info = &driver.interfaces[&id].info;
        assert_eq!(info.announce_rate_target, Some(7200.0));
        assert_eq!(info.announce_rate_penalty, 15.0);
        assert_eq!(info.announce_rate_grace, 7);
    }

    #[test]
    fn static_interface_registration_applies_transport_ingress_control_defaults() {
        let (tx, rx) = crate::event::channel();
        let mut driver = Driver::new(
            test_transport_config(true),
            rx,
            tx.clone(),
            Box::new(NoopCallbacks),
        );
        let mut defaults = rns_core::transport::types::IngressControlConfig::enabled();
        defaults.burst_hold = 1.5;
        defaults.pr_burst_freq = 5.5;
        defaults.egress_enabled = true;
        defaults.egress_pr_freq = 9.5;
        driver.set_ingress_control_defaults(defaults);

        let id = rns_core::transport::types::InterfaceId(8);
        register_started_interface(StartedInterface {
            driver: &mut driver,
            tx: &tx,
            queue_capacity: crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
            id,
            info: test_interface_info(id.0),
            writer: Box::new(TestWriter),
            interface_type_name: "TestInterface".to_string(),
            ifac_state: None,
            ifac_runtime: &IfacRuntimeConfig {
                netname: None,
                netkey: None,
                size: 16,
            },
            stamp_costs: rns_core::transport::types::InterfaceStampCosts {
                destination_cost: None,
                pr_cost: Some(rns_core::stamp::StampCost::from_f64(12.25)),
            },
        });

        let ic = driver.interfaces[&id].info.ingress_control;
        assert_eq!(ic.burst_hold, 1.5);
        assert_eq!(ic.pr_burst_freq, 5.5);
        assert!(ic.egress_enabled);
        assert_eq!(ic.egress_pr_freq, 9.5);
        // The configured per-interface costs reach the engine (9b464382).
        assert_eq!(
            driver
                .engine
                .interface_info(&id)
                .unwrap()
                .stamp_costs
                .pr_cost,
            Some(rns_core::stamp::StampCost::from_f64(12.25))
        );
    }

    struct TestNodeRatchetStore {
        entry: storage::RatchetEntry,
        current_calls: std::sync::Mutex<Vec<[u8; 16]>>,
    }

    impl storage::RatchetStore for TestNodeRatchetStore {
        fn remember(&self, _dest_hash: [u8; 16], _entry: storage::RatchetEntry) -> io::Result<()> {
            Ok(())
        }

        fn current(
            &self,
            dest_hash: &[u8; 16],
            _now: f64,
            _expiry_secs: f64,
        ) -> io::Result<Option<storage::RatchetEntry>> {
            self.current_calls.lock().unwrap().push(*dest_hash);
            Ok(Some(self.entry))
        }

        fn cleanup(
            &self,
            _known_destinations: &HashSet<[u8; 16]>,
            _now: f64,
            _expiry_secs: f64,
        ) -> io::Result<storage::RatchetCleanupStats> {
            Ok(Default::default())
        }
    }

    #[test]
    fn send_packet_checks_ratchet_store_for_single_destinations() {
        let store = Arc::new(TestNodeRatchetStore {
            entry: storage::RatchetEntry {
                ratchet: [0x55; 32],
                received_at: time::now(),
            },
            current_calls: std::sync::Mutex::new(Vec::new()),
        });
        let ratchet_store: Arc<dyn storage::RatchetStore> = store.clone();
        let node = RnsNode::start(
            NodeConfig {
                ratchet_store: Some(ratchet_store),
                ..Default::default()
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let mut rng = OsRng;
        let remote_identity = Identity::new(&mut rng);
        let public_key = remote_identity.get_public_key().unwrap();
        let announced = crate::destination::AnnouncedIdentity {
            dest_hash: rns_core::types::DestHash([0x22; 16]),
            identity_hash: rns_core::types::IdentityHash(*remote_identity.hash()),
            public_key,
            app_data: None,
            hops: 1,
            received_at: time::now(),
            receiving_interface: rns_core::transport::types::InterfaceId(0),
            rssi: Some(-100),
            snr: Some(10.5),
        };
        let dest = crate::destination::Destination::single_out("test", &["ratchet"], &announced);

        node.send_packet_queued(&dest, b"hello").unwrap();
        assert_eq!(
            store.current_calls.lock().unwrap().as_slice(),
            &[dest.hash.0]
        );

        node.shutdown();
    }

    #[test]
    fn single_payload_uses_stored_ratchet_key_material() {
        let ratchet_prv = rns_crypto::x25519::X25519PrivateKey::from_bytes(&[0x42; 32]);
        let ratchet_pub = ratchet_prv.public_key().public_bytes();
        let store = Arc::new(TestNodeRatchetStore {
            entry: storage::RatchetEntry {
                ratchet: ratchet_pub,
                received_at: time::now(),
            },
            current_calls: std::sync::Mutex::new(Vec::new()),
        });
        let ratchet_store: Arc<dyn storage::RatchetStore> = store.clone();
        let (tx, _rx) = crate::event::channel();
        let node = RnsNode {
            tx,
            driver_handle: None,
            verify_handle: None,
            verify_shutdown: Arc::default(),
            persist_handle: None,
            persist_stop: None,
            rpc_server: None,
            #[cfg(target_os = "linux")]
            rpc_unix_server: None,
            tick_interval_ms: Arc::new(AtomicU64::new(1000)),
            next_send_ticket: AtomicU64::new(0),
            encrypt_deferred_inline: std::thread::available_parallelism()
                .map_or(true, |cpus| cpus.get() == 1),
            accepting_work: Arc::new(AtomicBool::new(true)),
            probe_server: None,
            known_destinations_path: None,
            known_destinations_store: None,
            stamp_store: None,
            stamp_destinations: Arc::new(AtomicBool::new(rns_core::constants::STAMP_DESTINATIONS)),
            stamp_targets: crate::stamper::StampTargets::default(),
            stamper: None,
            transport_state_dir: None,
            ratchet_store: Some(ratchet_store),
            ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
            #[cfg(feature = "iface-local")]
            destination_types: Default::default(),
        };

        let mut rng = OsRng;
        let remote_identity = Identity::new(&mut rng);
        let public_key = remote_identity.get_public_key().unwrap();
        let announced = crate::destination::AnnouncedIdentity {
            dest_hash: rns_core::types::DestHash([0x33; 16]),
            identity_hash: rns_core::types::IdentityHash(*remote_identity.hash()),
            public_key,
            app_data: None,
            hops: 1,
            received_at: time::now(),
            receiving_interface: rns_core::transport::types::InterfaceId(0),
            rssi: Some(-100),
            snr: Some(10.5),
        };
        let dest = crate::destination::Destination::single_out("test", &["ratchet"], &announced);
        let payload = node
            .encrypt_single_payload(&dest, b"ratchet message")
            .unwrap();

        assert_eq!(
            store.current_calls.lock().unwrap().as_slice(),
            &[dest.hash.0]
        );
        assert!(remote_identity.decrypt(&payload).is_err());

        let peer_pub_bytes: [u8; 32] = payload[..32].try_into().unwrap();
        let peer_pub = rns_crypto::x25519::X25519PublicKey::from_bytes(&peer_pub_bytes);
        let shared_key = ratchet_prv.exchange(&peer_pub);
        let derived_key = rns_crypto::hkdf::hkdf(
            rns_crypto::identity::DERIVED_KEY_LENGTH,
            &shared_key,
            Some(remote_identity.hash()),
            None,
        )
        .unwrap();
        let token = rns_crypto::token::Token::new(&derived_key).unwrap();
        let plaintext = token.decrypt(&payload[32..]).unwrap();
        assert_eq!(plaintext, b"ratchet message");
    }

    #[test]
    fn tcp_client_interface_is_not_discoverable_without_kiss_framing() {
        let mut params = std::collections::HashMap::new();
        params.insert("discoverable".to_string(), "yes".to_string());
        params.insert(
            "discovery_name".to_string(),
            "invalid-tcp-client".to_string(),
        );
        params.insert("reachable_on".to_string(), "example.com".to_string());
        params.insert("target_port".to_string(), "4242".to_string());

        let discovery =
            super::extract_discovery_config("tcp-client", "TCPClientInterface", &params);

        assert!(
            discovery.is_none(),
            "TCPClientInterface discovery must be rejected unless KISS framing is supported"
        );
    }

    #[test]
    fn discovery_config_parses_location_command() {
        let mut params = std::collections::HashMap::new();
        params.insert("discoverable".into(), "yes".into());
        params.insert("location_cmd".into(), "~/bin/reticulum-location".into());

        let discovery =
            super::extract_discovery_config("public", "BackboneInterface", &params).unwrap();
        assert_eq!(
            discovery.location_cmd.as_deref(),
            Some("~/bin/reticulum-location")
        );
    }

    #[test]
    fn discovery_config_parses_valid_operator_lxmf_address() {
        let mut params = std::collections::HashMap::new();
        params.insert("discoverable".into(), "yes".into());
        params.insert("discovery_lxmf_address".into(), "ab".repeat(16));

        let discovery =
            super::extract_discovery_config("public", "BackboneInterface", &params).unwrap();

        assert_eq!(discovery.operator_lxmf_address, Some([0xab; 16]));
    }

    #[test]
    fn discovery_config_ignores_invalid_operator_lxmf_addresses() {
        for address in ["ab".repeat(15), "zz".repeat(16)] {
            let mut params = std::collections::HashMap::new();
            params.insert("discoverable".into(), "yes".into());
            params.insert("discovery_lxmf_address".into(), address);

            let discovery =
                super::extract_discovery_config("public", "BackboneInterface", &params).unwrap();

            assert_eq!(discovery.operator_lxmf_address, None);
        }
    }

    #[test]
    fn ingress_control_config_defaults_by_interface_type() {
        let params = std::collections::HashMap::new();

        let tcp = super::parse_ingress_control_config("TCPServerInterface", &params).unwrap();
        assert!(tcp.enabled);
        assert_eq!(
            tcp.max_held_announces,
            rns_core::constants::IC_MAX_HELD_ANNOUNCES
        );
        assert_eq!(tcp.burst_hold, rns_core::constants::IC_BURST_HOLD);

        let pipe = super::parse_ingress_control_config("PipeInterface", &params).unwrap();
        assert!(!pipe.enabled);
        assert_eq!(
            pipe.held_release_interval,
            rns_core::constants::IC_HELD_RELEASE_INTERVAL
        );
    }

    #[test]
    fn ingress_control_config_parses_python_ic_keys() {
        let mut params = std::collections::HashMap::new();
        params.insert("ingress_control".to_string(), "No".to_string());
        params.insert("ic_max_held_announces".to_string(), "17".to_string());
        params.insert("ic_burst_hold".to_string(), "1.5".to_string());
        params.insert("ic_burst_freq_new".to_string(), "2.5".to_string());
        params.insert("ic_burst_freq".to_string(), "3.5".to_string());
        params.insert("ic_pr_burst_freq_new".to_string(), "3.25".to_string());
        params.insert("ic_pr_burst_freq".to_string(), "8.25".to_string());
        params.insert("egress_control".to_string(), "Yes".to_string());
        params.insert("ec_pr_freq".to_string(), "5.25".to_string());
        params.insert("ic_new_time".to_string(), "4.5".to_string());
        params.insert("ic_burst_penalty".to_string(), "5.5".to_string());
        params.insert("ic_held_release_interval".to_string(), "6.5".to_string());

        let config = super::parse_ingress_control_config("TCPServerInterface", &params).unwrap();

        assert!(!config.enabled);
        assert_eq!(config.max_held_announces, 17);
        assert_eq!(config.burst_hold, 1.5);
        assert_eq!(config.burst_freq_new, 2.5);
        assert_eq!(config.burst_freq, 3.5);
        assert_eq!(config.pr_burst_freq_new, 3.25);
        assert_eq!(config.pr_burst_freq, 8.25);
        assert!(config.egress_enabled);
        assert_eq!(config.egress_pr_freq, 5.25);
        assert_eq!(config.new_time, 4.5);
        assert_eq!(config.burst_penalty, 5.5);
        assert_eq!(config.held_release_interval, 6.5);
    }

    #[test]
    fn ingress_control_config_rejects_invalid_values() {
        let mut params = std::collections::HashMap::new();
        params.insert("ic_burst_hold".to_string(), "-1".to_string());

        let err = super::parse_ingress_control_config("TCPServerInterface", &params).unwrap_err();

        assert!(err.contains("ic_burst_hold"));
    }

    #[test]
    fn interface_stamp_costs_are_fractional_and_clamped() {
        // Reticulum.py since 9b464382: `as_float`, `max(0, min(v, cap))`.
        use rns_core::stamp::StampCost;
        let mut params = std::collections::HashMap::new();
        let none = super::parse_interface_stamp_costs(&params).unwrap();
        assert_eq!(none, Default::default());

        params.insert("destination_cost".to_string(), "17.5".to_string());
        params.insert("pr_cost".to_string(), "250".to_string());
        let costs = super::parse_interface_stamp_costs(&params).unwrap();
        assert_eq!(costs.destination_cost, Some(StampCost::from_f64(17.5)));
        assert_eq!(costs.pr_cost, Some(StampCost::from_whole(100)));

        params.insert("pr_cost".to_string(), "-3".to_string());
        let costs = super::parse_interface_stamp_costs(&params).unwrap();
        assert_eq!(costs.pr_cost, Some(StampCost::ZERO));

        params.insert("destination_cost".to_string(), "steep".to_string());
        let err = super::parse_interface_stamp_costs(&params).unwrap_err();
        assert!(err.contains("destination_cost"));
    }

    #[test]
    fn discovery_stamp_default_is_16_but_explicit_override_is_preserved() {
        let mut params = std::collections::HashMap::new();
        params.insert("discoverable".to_string(), "yes".to_string());
        let defaults =
            super::extract_discovery_config("Discovery Test", "BackboneInterface", &params)
                .unwrap();
        assert_eq!(
            defaults.stamp_value,
            rns_core::stamp::StampCost::from_whole(16)
        );

        params.insert("discovery_stamp_value".to_string(), "14".to_string());
        let overridden =
            super::extract_discovery_config("Discovery Test", "BackboneInterface", &params)
                .unwrap();
        assert_eq!(
            overridden.stamp_value,
            rns_core::stamp::StampCost::from_whole(14)
        );
    }

    #[test]
    fn start_and_shutdown() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();
        node.shutdown();
    }

    #[test]
    fn known_destinations_persist_across_restart() {
        let dir = tempdir().unwrap();
        let dest_hash = [0x91; 16];
        let identity = Identity::new(&mut OsRng);
        let last_used_at = 77.0;
        let receiving_interface = rns_core::transport::types::InterfaceId(42);

        let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
        let (response_tx, response_rx) = std::sync::mpsc::channel();
        node.event_sender()
            .send(crate::event::Event::query(
                QueryRequest::RestoreKnownDestination(crate::event::KnownDestinationEntry {
                    dest_hash,
                    identity_hash: *identity.hash(),
                    public_key: identity.get_public_key().unwrap(),
                    app_data: Some(b"persisted".to_vec()),
                    hops: 2,
                    received_at: 55.0,
                    receiving_interface,
                    was_used: true,
                    last_used_at: Some(last_used_at),
                    retained: true,
                    stamp: None,
                    stamp_value: rns_core::stamp::StampValue::ZERO,
                    lr_cost: None,
                }),
                response_tx,
            ))
            .unwrap();
        assert!(matches!(
            response_rx.recv().unwrap(),
            QueryResponse::RestoreKnownDestination(true)
        ));
        node.shutdown();

        let restarted = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
        let entries = restarted.known_destinations().unwrap();
        let entry = entries
            .iter()
            .find(|entry| entry.dest_hash == dest_hash)
            .expect("reloaded destination should appear in lifecycle listing");
        assert!(entry.retained);
        assert!(entry.was_used);
        assert_eq!(entry.hops, 2);
        assert_eq!(entry.receiving_interface, receiving_interface);
        assert_eq!(entry.last_used_at, Some(last_used_at));

        let recalled = restarted
            .recall_identity(&rns_core::types::DestHash(dest_hash))
            .unwrap()
            .expect("known destination should reload from storage");
        assert_eq!(recalled.identity_hash.0, *identity.hash());
        assert_eq!(recalled.app_data, Some(b"persisted".to_vec()));
        restarted.shutdown();
    }

    #[test]
    fn transport_paths_hashes_and_detached_tunnels_restore_from_upstream_files() {
        use rns_core::packet::{PacketFlags, RawPacket};
        use rns_core::transport::persistence::{
            PersistedPath, PersistedTunnel, TransportStateSnapshot,
        };

        let dir = tempdir().unwrap();
        fs::write(
            dir.path().join("config"),
            r#"[reticulum]
enable_transport = True
share_instance = False

[interfaces]
  [[Persistence Test]]
    type = UDPInterface
    forward_ip = 127.0.0.1
    forward_port = 9
    interface_mode = full
"#,
        )
        .unwrap();
        let paths = storage::ensure_storage_dirs(dir.path()).unwrap();
        let destination_hash = [0x31; 16];
        let packet = RawPacket::pack(
            PacketFlags {
                header_type: rns_core::constants::HEADER_1,
                context_flag: rns_core::constants::FLAG_UNSET,
                transport_type: rns_core::constants::TRANSPORT_BROADCAST,
                destination_type: rns_core::constants::DESTINATION_SINGLE,
                packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
            },
            0,
            &destination_hash,
            None,
            rns_core::constants::CONTEXT_NONE,
            &[0xA5; 32],
        )
        .unwrap();
        crate::announce_cache::AnnounceCache::new(paths.cache.join("announces"))
            .store(&packet.packet_hash, &packet.raw, None)
            .unwrap();
        let now = time::now();
        let route = PersistedPath {
            destination_hash,
            timestamp: now,
            received_from: [0x41; 16],
            hops: 2,
            expires: now + 600.0,
            random_blobs: vec![[0x51; 10]],
            interface_hash: Some(rns_core::hash::full_hash(b"Persistence Test")),
            packet_hash: packet.packet_hash,
            used: 0.0,
        };
        storage::save_transport_state(
            &TransportStateSnapshot {
                packet_hashes: vec![[0x61; 32], [0x62; 32]],
                paths: vec![route.clone()],
                tunnels: vec![PersistedTunnel {
                    tunnel_id: [0x71; 32],
                    interface_hash: None,
                    paths: vec![PersistedPath {
                        interface_hash: None,
                        ..route
                    }],
                    expires: now + 1_200.0,
                }],
            },
            &paths.storage,
            false,
        )
        .unwrap();

        let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();
        let QueryResponse::TransportStateSnapshot(restored) =
            node.query(QueryRequest::TransportStateSnapshot).unwrap()
        else {
            panic!("unexpected transport snapshot response");
        };
        // Upstream's full hashes load; the table keeps their first 16 bytes.
        let stored = |byte: u8| {
            let mut hash = [0u8; 32];
            hash[..16].fill(byte);
            hash
        };
        assert_eq!(restored.packet_hashes, vec![stored(0x61), stored(0x62)]);
        assert_eq!(restored.paths.len(), 1);
        assert_eq!(restored.paths[0].destination_hash, destination_hash);
        assert_eq!(restored.paths[0].received_from, [0x41; 16]);
        assert_eq!(restored.tunnels.len(), 1);
        assert_eq!(restored.tunnels[0].tunnel_id, [0x71; 32]);
        assert_eq!(restored.tunnels[0].interface_hash, None);
        node.shutdown();
    }

    #[test]
    fn from_config_cleans_persistent_ratchets_after_loading_known_destinations() {
        fn hex(bytes: &[u8]) -> String {
            bytes.iter().map(|byte| format!("{:02x}", byte)).collect()
        }

        let dir = tempdir().unwrap();
        fs::write(
            dir.path().join("config"),
            "[reticulum]\nenable_transport = False\nshare_instance = False\nratchet_expiry = 300\n",
        )
        .unwrap();
        let paths = storage::ensure_storage_dirs(dir.path()).unwrap();
        let known_live = [0x01; 16];
        let known_expired = [0x02; 16];
        let unknown = [0x03; 16];
        let known_corrupt = [0x04; 16];
        let now = time::now();

        let identity = Identity::new(&mut OsRng);
        let known = std::collections::HashMap::from([
            (
                known_live,
                storage::KnownDestination {
                    identity_hash: *identity.hash(),
                    public_key: identity.get_public_key().unwrap(),
                    app_data: None,
                    hops: 1,
                    received_at: now,
                    receiving_interface: 0,
                    was_used: false,
                    last_used_at: None,
                    retained: false,
                    stamp: None,
                    stamp_value: rns_core::stamp::StampValue::ZERO,
                    lr_cost: None,
                },
            ),
            (
                known_expired,
                storage::KnownDestination {
                    identity_hash: *identity.hash(),
                    public_key: identity.get_public_key().unwrap(),
                    app_data: None,
                    hops: 1,
                    received_at: now,
                    receiving_interface: 0,
                    was_used: false,
                    last_used_at: None,
                    retained: false,
                    stamp: None,
                    stamp_value: rns_core::stamp::StampValue::ZERO,
                    lr_cost: None,
                },
            ),
            (
                known_corrupt,
                storage::KnownDestination {
                    identity_hash: *identity.hash(),
                    public_key: identity.get_public_key().unwrap(),
                    app_data: None,
                    hops: 1,
                    received_at: now,
                    receiving_interface: 0,
                    was_used: false,
                    last_used_at: None,
                    retained: false,
                    stamp: None,
                    stamp_value: rns_core::stamp::StampValue::ZERO,
                    lr_cost: None,
                },
            ),
        ]);
        storage::save_known_destinations(&known, &paths.storage.join("known_destinations"))
            .unwrap();

        let store = storage::FsRatchetStore::new(paths.ratchets.clone());
        store
            .remember(
                known_live,
                storage::RatchetEntry {
                    ratchet: [0x11; 32],
                    received_at: now,
                },
            )
            .unwrap();
        store
            .remember(
                known_expired,
                storage::RatchetEntry {
                    ratchet: [0x22; 32],
                    received_at: now - 1000.0,
                },
            )
            .unwrap();
        store
            .remember(
                unknown,
                storage::RatchetEntry {
                    ratchet: [0x33; 32],
                    received_at: now,
                },
            )
            .unwrap();
        fs::write(paths.ratchets.join(hex(&known_corrupt)), b"not msgpack").unwrap();
        fs::write(paths.ratchets.join("0102.out"), b"temp").unwrap();

        let node = RnsNode::from_config(Some(dir.path()), Box::new(NoopCallbacks)).unwrap();

        assert!(!paths.ratchets.join(hex(&known_live)).exists());
        assert!(!paths.ratchets.join(hex(&known_expired)).exists());
        assert!(!paths.ratchets.join(hex(&unknown)).exists());
        assert!(!paths.ratchets.join(hex(&known_corrupt)).exists());
        assert!(!paths.ratchets.join("0102.out").exists());
        assert!(paths.storage.join("known_destinations.imported").exists());

        let store = crate::storage_sqlite::KnownDestinationsStore::open_read_only(
            &crate::storage_sqlite::default_path(&paths.storage),
            16,
        )
        .unwrap();
        assert_eq!(store.load_all().unwrap().len(), 3);
        assert!(store.current(&known_live, now, 300.0).unwrap().is_some());
        assert!(store.current(&known_expired, now, 300.0).unwrap().is_none());
        assert!(store.current(&unknown, now, 300.0).unwrap().is_none());
        assert!(store.current(&known_corrupt, now, 300.0).unwrap().is_none());

        node.shutdown();
    }

    #[test]
    fn start_with_identity() {
        let identity = Identity::new(&mut OsRng);
        let hash = *identity.hash();
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: true,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: Some(identity),
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();
        // The identity hash should have been used
        let _ = hash;
        node.shutdown();
    }

    #[test]
    fn start_generates_identity() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();
        // Should not panic - identity was auto-generated
        node.shutdown();
    }

    #[test]
    fn from_config_creates_identity() {
        let dir = std::env::temp_dir().join(format!("rns-test-fc-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        // Write a minimal config file
        fs::write(
            dir.join("config"),
            "[reticulum]\nenable_transport = False\n",
        )
        .unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();

        // Identity file should have been created
        assert!(dir.join("storage/identities/identity").exists());

        node.shutdown();
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn private_config_keeps_state_out_of_config_directory() {
        let root = tempfile::tempdir().unwrap();
        let config_dir = root.path().join("config");
        let state_dir = root.path().join("state");
        fs::create_dir_all(&config_dir).unwrap();
        fs::write(
            config_dir.join("config"),
            "[reticulum]\nenable_transport = False\nshare_instance = Yes\n",
        )
        .unwrap();
        let node =
            RnsNode::from_private_config(&config_dir, &state_dir, None, Box::new(NoopCallbacks))
                .unwrap();
        assert!(state_dir.join("storage/identities/identity").exists());
        assert!(!config_dir.join("storage").exists());
        node.shutdown();
    }

    #[test]
    fn from_config_loads_identity() {
        let dir = std::env::temp_dir().join(format!("rns-test-fl-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(dir.join("storage/identities")).unwrap();

        // Pre-create an identity
        let identity = Identity::new(&mut OsRng);
        let hash = *identity.hash();
        storage::save_identity(&identity, &dir.join("storage/identities/identity")).unwrap();

        fs::write(
            dir.join("config"),
            "[reticulum]\nenable_transport = False\n",
        )
        .unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();

        // Verify the same identity was loaded (hash matches)
        let loaded = storage::load_identity(&dir.join("storage/identities/identity")).unwrap();
        assert_eq!(*loaded.hash(), hash);

        node.shutdown();
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn from_config_tcp_server() {
        let dir = std::env::temp_dir().join(format!("rns-test-fts-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let port = crate::test_support::port();

        let config = format!(
            r#"
[reticulum]
enable_transport = False

[interfaces]
  [[Test TCP Server]]
    type = TCPServerInterface
    listen_ip = 127.0.0.1
    listen_port = {}
"#,
            port
        );

        fs::write(dir.join("config"), config).unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();

        // Give server time to start
        thread::sleep(Duration::from_millis(100));

        // Should be able to connect
        let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();

        node.shutdown();
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn from_config_starts_rpc_when_share_instance_enabled() {
        let dir = std::env::temp_dir().join(format!("rns-test-rpc-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let rpc_port = crate::test_support::port();

        let config = format!(
            r#"
[reticulum]
enable_transport = False
share_instance = Yes
instance_control_port = {}

[interfaces]
"#,
            rpc_port
        );

        fs::write(dir.join("config"), config).unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();

        thread::sleep(Duration::from_millis(100));

        let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", rpc_port)).unwrap();

        node.shutdown();
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn from_config_starts_rpc_when_transport_enabled() {
        let dir =
            std::env::temp_dir().join(format!("rns-test-rpc-transport-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let rpc_port = crate::test_support::port();

        let config = format!(
            r#"
[reticulum]
enable_transport = True
share_instance = Yes
instance_control_port = {}

[interfaces]
"#,
            rpc_port
        );

        fs::write(dir.join("config"), config).unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();

        thread::sleep(Duration::from_millis(100));

        let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", rpc_port)).unwrap();

        node.shutdown();
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn from_config_starts_rpc_when_tcp_client_is_unreachable() {
        let dir =
            std::env::temp_dir().join(format!("rns-test-rpc-unreachable-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let rpc_port = crate::test_support::port();
        let unreachable_port = crate::test_support::port();

        let config = format!(
            r#"
[reticulum]
enable_transport = True
share_instance = Yes
instance_control_port = {}

[interfaces]
  [[Unreachable Upstream]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = {}
"#,
            rpc_port, unreachable_port
        );

        fs::write(dir.join("config"), config).unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();

        thread::sleep(Duration::from_millis(100));

        let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", rpc_port)).unwrap();

        node.shutdown();
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn test_parse_interface_mode() {
        use rns_core::constants::*;

        assert_eq!(parse_interface_mode("full"), MODE_FULL);
        assert_eq!(parse_interface_mode("Full"), MODE_FULL);
        assert_eq!(parse_interface_mode("access_point"), MODE_ACCESS_POINT);
        assert_eq!(parse_interface_mode("accesspoint"), MODE_ACCESS_POINT);
        assert_eq!(parse_interface_mode("ap"), MODE_ACCESS_POINT);
        assert_eq!(parse_interface_mode("AP"), MODE_ACCESS_POINT);
        assert_eq!(parse_interface_mode("pointtopoint"), MODE_POINT_TO_POINT);
        assert_eq!(parse_interface_mode("ptp"), MODE_POINT_TO_POINT);
        assert_eq!(parse_interface_mode("roaming"), MODE_ROAMING);
        assert_eq!(parse_interface_mode("boundary"), MODE_BOUNDARY);
        assert_eq!(parse_interface_mode("gateway"), MODE_GATEWAY);
        assert_eq!(parse_interface_mode("gw"), MODE_GATEWAY);
        assert_eq!(parse_interface_mode("internal"), MODE_INTERNAL);
        assert_eq!(parse_interface_mode("Internal"), MODE_INTERNAL);
        // Unknown defaults to FULL
        assert_eq!(parse_interface_mode("invalid"), MODE_FULL);
    }

    #[test]
    fn discovery_mode_normalization_matrix() {
        use rns_core::constants::*;

        let interface_types = [
            "RNodeInterface",
            "RNodeMultiInterface",
            "BackboneInterface",
            "TCPServerInterface",
            "AutoInterface",
        ];
        let modes = [
            MODE_FULL,
            MODE_ACCESS_POINT,
            MODE_POINT_TO_POINT,
            MODE_ROAMING,
            MODE_BOUNDARY,
            MODE_GATEWAY,
            MODE_INTERNAL,
        ];

        for interface_type in interface_types {
            for configured_mode in modes {
                for discoverable in [false, true] {
                    for ignore_warnings in [false, true] {
                        let expected = if !discoverable
                            || ignore_warnings
                            || matches!(
                                configured_mode,
                                MODE_ACCESS_POINT | MODE_GATEWAY | MODE_INTERNAL
                            ) {
                            configured_mode
                        } else if matches!(interface_type, "RNodeInterface" | "RNodeMultiInterface")
                        {
                            MODE_ACCESS_POINT
                        } else {
                            MODE_GATEWAY
                        };
                        assert_eq!(
                            normalize_discovery_mode(
                                interface_type,
                                configured_mode,
                                discoverable,
                                ignore_warnings,
                            ),
                            expected,
                            "type={interface_type}, mode={configured_mode}, discoverable={discoverable}, ignore={ignore_warnings}",
                        );
                    }
                }
            }
        }
    }

    #[test]
    fn to_node_config_serial() {
        // Verify from_config parses SerialInterface correctly.
        // The serial port won't exist, so start() will fail, but the config
        // parsing path is exercised. We verify via the error (not a config error).
        let dir = std::env::temp_dir().join(format!("rns-test-serial-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let config = r#"
[reticulum]
enable_transport = False

[interfaces]
  [[Test Serial Port]]
    type = SerialInterface
    port = /dev/nonexistent_rns_test_serial
    speed = 115200
    databits = 8
    parity = E
    stopbits = 1
    interface_mode = ptp
    networkname = testnet
"#;
        fs::write(dir.join("config"), config).unwrap();

        // Interface error is non-fatal: the node starts but logs the error.
        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks))
            .expect("Config should parse; interface failure is non-fatal");
        node.shutdown();

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn to_node_config_kiss() {
        // Verify from_config parses KISSInterface correctly.
        let dir = std::env::temp_dir().join(format!("rns-test-kiss-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let config = r#"
[reticulum]
enable_transport = False

[interfaces]
  [[Test KISS TNC]]
    type = KISSInterface
    port = /dev/nonexistent_rns_test_kiss
    speed = 9600
    preamble = 500
    txtail = 30
    persistence = 128
    slottime = 40
    flow_control = True
    id_interval = 600
    id_callsign = TEST0
    interface_mode = full
    passphrase = secretkey
"#;
        fs::write(dir.join("config"), config).unwrap();

        // Interface error is non-fatal: the node starts but logs the error.
        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks))
            .expect("Config should parse; interface failure is non-fatal");
        node.shutdown();

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn config_parser_accepts_recursive_prs_interface_option() {
        let config = r#"
[reticulum]
enable_transport = True

[interfaces]
  [[Recursive Client]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = 4242
    interface_mode = full
    recursive_prs = yes

  [[Default Client]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = 4243
    interface_mode = gateway
"#;

        let parsed = config::parse(config).unwrap();

        assert_eq!(parsed.interfaces.len(), 2);
        assert_eq!(parsed.interfaces[0].name, "Recursive Client");
        assert_eq!(parsed.interfaces[0].mode, "full");
        assert_eq!(
            parsed.interfaces[0]
                .params
                .get("recursive_prs")
                .map(String::as_str),
            Some("yes")
        );
        assert_eq!(parsed.interfaces[1].name, "Default Client");
        assert_eq!(parsed.interfaces[1].mode, "gateway");
        assert!(!parsed.interfaces[1].params.contains_key("recursive_prs"));
    }

    #[test]
    fn config_parser_accepts_announces_from_internal_interface_option() {
        let config = r#"
[reticulum]
enable_transport = True

[interfaces]
  [[Filtered Client]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = 4242
    interface_mode = gateway
    announces_from_internal = no

  [[Default Client]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = 4243
    interface_mode = full
"#;

        let parsed = config::parse(config).unwrap();

        assert_eq!(parsed.interfaces.len(), 2);
        assert_eq!(parsed.interfaces[0].name, "Filtered Client");
        assert_eq!(parsed.interfaces[0].mode, "gateway");
        assert_eq!(
            parsed.interfaces[0]
                .params
                .get("announces_from_internal")
                .map(String::as_str),
            Some("no")
        );
        assert_eq!(parsed.interfaces[1].name, "Default Client");
        assert_eq!(parsed.interfaces[1].mode, "full");
        assert!(!parsed.interfaces[1]
            .params
            .contains_key("announces_from_internal"));
    }

    #[test]
    fn config_parser_preserves_announces_to_internal_tristate() {
        let config = r#"
[interfaces]
  [[Allowed Boundary]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = 4242
    announces_to_internal = yes

  [[Blocked Boundary]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = 4243
    announces_to_internal = no

  [[Default Boundary]]
    type = TCPClientInterface
    target_host = 127.0.0.1
    target_port = 4244
"#;

        let parsed = config::parse(config).unwrap();
        let parsed_value = |index: usize| {
            parsed.interfaces[index]
                .params
                .get("announces_to_internal")
                .and_then(|value| config::parse_bool_pub(value))
        };

        assert_eq!(parsed_value(0), Some(true));
        assert_eq!(parsed_value(1), Some(false));
        assert_eq!(parsed_value(2), None);
    }

    #[test]
    fn interface_gravity_parser_accepts_signed_values_and_uses_global_default() {
        assert_eq!(parse_interface_gravity(Some("7"), -2), 7);
        assert_eq!(parse_interface_gravity(Some("-4"), 2), -4);
        assert_eq!(parse_interface_gravity(None, -6), -6);
    }

    #[test]
    fn config_parser_accepts_local_hops_delta_option() {
        let cfg = r#"
[reticulum]
local_hops_delta = yes
"#;
        let parsed = config::parse(cfg).unwrap();
        assert!(parsed.reticulum.local_hops_delta);
    }

    #[test]
    fn config_parser_accepts_static_transport_identity_option() {
        let config = r#"
[reticulum]
enable_transport = false
static_transport_identity = yes
"#;

        let parsed = config::parse(config).unwrap();

        assert!(!parsed.reticulum.enable_transport);
        assert!(parsed.reticulum.static_transport_identity);
    }

    #[test]
    fn non_transport_node_uses_ephemeral_transport_identity_by_default() {
        let identity = Identity::new(&mut rns_crypto::FixedRng::new(&[0x41; 32]));
        let stored_hash = *identity.hash();
        let node = RnsNode::start(
            NodeConfig {
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: Some(identity),
                share_instance: false,
                ..Default::default()
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let response = node.query(QueryRequest::TransportIdentity).unwrap();
        let QueryResponse::TransportIdentity(Some(hash)) = response else {
            panic!("transport identity query returned {response:?}");
        };
        assert_ne!(hash, stored_hash);

        node.shutdown();
    }

    #[test]
    fn non_transport_node_can_use_static_transport_identity() {
        let identity = Identity::new(&mut rns_crypto::FixedRng::new(&[0x42; 32]));
        let stored_hash = *identity.hash();
        let node = RnsNode::start(
            NodeConfig {
                transport_enabled: false,
                static_transport_identity: true,
                local_hops_delta: false,
                identity: Some(identity),
                share_instance: false,
                ..Default::default()
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let response = node.query(QueryRequest::TransportIdentity).unwrap();
        let QueryResponse::TransportIdentity(Some(hash)) = response else {
            panic!("transport identity query returned {response:?}");
        };
        assert_eq!(hash, stored_hash);

        node.shutdown();
    }

    #[test]
    fn discoverable_interface_forces_static_transport_identity_on_non_transport() {
        assert!(!effective_static_transport_identity(false, false, false));
        assert!(effective_static_transport_identity(true, false, false));
        assert!(!effective_static_transport_identity(false, true, false));
        assert!(effective_static_transport_identity(false, false, true));
        assert!(effective_static_transport_identity(true, false, true));
    }

    #[test]
    fn test_extract_ifac_config() {
        use std::collections::HashMap;

        // No IFAC params → None
        let params: HashMap<String, String> = HashMap::new();
        assert!(extract_ifac_config(&params, 16).is_none());

        // networkname only
        let mut params = HashMap::new();
        params.insert("networkname".into(), "testnet".into());
        let ifac = extract_ifac_config(&params, 16).unwrap();
        assert_eq!(ifac.netname.as_deref(), Some("testnet"));
        assert!(ifac.netkey.is_none());
        assert_eq!(ifac.size, 16);

        // passphrase only with custom size (in bits)
        let mut params = HashMap::new();
        params.insert("passphrase".into(), "secret".into());
        params.insert("ifac_size".into(), "64".into()); // 64 bits = 8 bytes
        let ifac = extract_ifac_config(&params, 16).unwrap();
        assert!(ifac.netname.is_none());
        assert_eq!(ifac.netkey.as_deref(), Some("secret"));
        assert_eq!(ifac.size, 8);

        // Both with alternate key names
        let mut params = HashMap::new();
        params.insert("network_name".into(), "mynet".into());
        params.insert("pass_phrase".into(), "mykey".into());
        let ifac = extract_ifac_config(&params, 8).unwrap();
        assert_eq!(ifac.netname.as_deref(), Some("mynet"));
        assert_eq!(ifac.netkey.as_deref(), Some("mykey"));
        assert_eq!(ifac.size, 8);

        // Nonsensical "None" network name and passphrase are ignored.
        let mut params = HashMap::new();
        params.insert("networkname".into(), "None".into());
        params.insert("passphrase".into(), "None".into());
        assert!(extract_ifac_config(&params, 16).is_none());

        // Empty values are ignored, but a valid sibling value still applies.
        let mut params = HashMap::new();
        params.insert("networkname".into(), String::new());
        params.insert("passphrase".into(), "secret".into());
        let ifac = extract_ifac_config(&params, 16).unwrap();
        assert!(ifac.netname.is_none());
        assert_eq!(ifac.netkey.as_deref(), Some("secret"));
    }

    #[test]
    fn to_node_config_rnode() {
        // Verify from_config parses RNodeInterface correctly.
        // The serial port won't exist, so start() will fail at open time.
        let dir = std::env::temp_dir().join(format!("rns-test-rnode-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let config = r#"
[reticulum]
enable_transport = False

[interfaces]
  [[Test RNode]]
    type = RNodeInterface
    port = /dev/nonexistent_rns_test_rnode
    frequency = 867200000
    bandwidth = 125000
    txpower = 7
    spreadingfactor = 8
    codingrate = 5
    flow_control = True
    st_alock = 5.0
    lt_alock = 2.5
    interface_mode = full
    networkname = testnet
"#;
        fs::write(dir.join("config"), config).unwrap();

        // Interface error is non-fatal: the node starts but logs the error.
        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks))
            .expect("Config should parse; interface failure is non-fatal");
        node.shutdown();

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn to_node_config_pipe() {
        // Verify from_config parses PipeInterface correctly.
        // Use `cat` as a real command so it actually starts.
        let dir = std::env::temp_dir().join(format!("rns-test-pipe-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let config = r#"
[reticulum]
enable_transport = False

[interfaces]
  [[Test Pipe]]
    type = PipeInterface
    command = cat
    respawn_delay = 5000
    interface_mode = full
"#;
        fs::write(dir.join("config"), config).unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();
        // If we got here, config parsing and start() succeeded
        node.shutdown();

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn to_node_config_backbone() {
        // Verify from_config parses BackboneInterface correctly.
        let dir = std::env::temp_dir().join(format!("rns-test-backbone-{}", std::process::id()));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();

        let port = crate::test_support::port();

        let config = format!(
            r#"
[reticulum]
enable_transport = False

[interfaces]
  [[Test Backbone]]
    type = BackboneInterface
    listen_ip = 127.0.0.1
    listen_port = {}
    interface_mode = full
"#,
            port
        );

        fs::write(dir.join("config"), config).unwrap();

        let node = RnsNode::from_config(Some(&dir), Box::new(NoopCallbacks)).unwrap();

        // Give server time to start
        thread::sleep(Duration::from_millis(100));

        // Should be able to connect
        {
            let _client = std::net::TcpStream::connect(format!("127.0.0.1:{}", port)).unwrap();
            // client drops here, closing the connection cleanly
        }

        // Small delay to let epoll process the disconnect
        thread::sleep(Duration::from_millis(50));

        node.shutdown();
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn rnode_config_defaults() {
        use crate::interface::rnode::{RNodeConfig, RNodeSubConfig};

        let config = RNodeConfig::default();
        assert_eq!(config.speed, 115200);
        assert!(config.subinterfaces.is_empty());
        assert!(config.id_interval.is_none());
        assert!(config.id_callsign.is_none());

        let sub = RNodeSubConfig {
            name: "test".into(),
            vport: 0,
            outgoing: true,
            frequency: 868_000_000,
            bandwidth: 125_000,
            txpower: 7,
            spreading_factor: 8,
            coding_rate: 5,
            flow_control: false,
            st_alock: None,
            lt_alock: None,
        };
        assert_eq!(sub.frequency, 868_000_000);
        assert_eq!(sub.bandwidth, 125_000);
        assert!(!sub.flow_control);
    }

    // =========================================================================
    // Phase 9c: Announce + Discovery node-level tests
    // =========================================================================

    #[test]
    fn announce_without_persisted_stamp_is_deferred_to_the_stamper() {
        let dir = tempdir().unwrap();
        let store = crate::stamps::DestinationStampStore::new(dir.path());
        let targets = crate::stamper::StampTargets {
            destination_value: StampCost::from_whole(6),
            path_request_value: StampCost::from_whole(6),
        };
        let mut node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();
        node.set_stamp_store(store.clone());
        node.set_destination_stamping(true, targets);
        let stamper = Arc::clone(node.stamper().expect("started nodes share a stamper"));
        assert_eq!(
            stamper.targets(),
            targets,
            "the stamper grinds at the node's targets"
        );

        let identity = Identity::from_private_key(&[0x51; 64]);
        let dest = crate::Destination::single_in(
            "stamper",
            &["node"],
            rns_core::types::IdentityHash(*identity.hash()),
        );
        node.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())
            .unwrap();
        assert!(node.announce_needs_stamp(&dest, &identity));

        // Destination.announce with STAMP_DESTINATIONS and no stamp: the call
        // returns at once and the worker grinds, persists and emits.
        let started = std::time::Instant::now();
        node.announce_queued(&dest, &identity, Some(b"first"))
            .unwrap();
        assert!(started.elapsed() < Duration::from_secs(1));

        let name_hash = rns_core::destination::name_hash("stamper", &["node"]);
        let material = crate::stamps::destination_stamp_material(
            &dest.hash.0,
            &identity.get_public_key().unwrap(),
            &name_hash,
        );
        let deadline = std::time::Instant::now() + Duration::from_secs(30);
        while store
            .load(&dest.hash.0, &material, targets.destination_value)
            .is_none()
        {
            assert!(
                std::time::Instant::now() < deadline,
                "stamp was never persisted"
            );
            thread::sleep(Duration::from_millis(20));
        }

        // From now on announces are synchronous: the persisted stamp is reused.
        assert!(!node.announce_needs_stamp(&dest, &identity));
        node.announce_queued(&dest, &identity, Some(b"second"))
            .unwrap();
        assert_eq!(stamper.queued(), 0);
        let receipt = node
            .try_announce(&dest, &identity, None)
            .expect("synchronous stamped announce is admitted");
        // No interfaces: the receipt settles either way, it must not hang.
        let _ = futures::executor::block_on(receipt);

        // A confirmed-transmission announce for a destination still without a
        // stamp also defers, and its receipt settles once the worker emitted.
        let other_identity = Identity::from_private_key(&[0x52; 64]);
        let other = crate::Destination::single_in(
            "stamper",
            &["node", "other"],
            rns_core::types::IdentityHash(*other_identity.hash()),
        );
        node.register_destination(other.hash.0, other.dest_type.to_wire_constant())
            .unwrap();
        let started = std::time::Instant::now();
        let receipt = node
            .try_announce(&other, &other_identity, Some(b"deferred"))
            .expect("deferred announce is admitted");
        assert!(started.elapsed() < Duration::from_secs(1));
        let _ = futures::executor::block_on(receipt);
        assert!(!node.announce_needs_stamp(&other, &other_identity));

        node.shutdown();
    }

    #[test]
    fn announce_builds_valid_packet() {
        let identity = Identity::new(&mut OsRng);
        let identity_hash = rns_core::types::IdentityHash(*identity.hash());

        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let dest = crate::destination::Destination::single_in("test", &["echo"], identity_hash);

        // Register destination first
        node.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())
            .unwrap();

        // Announce should succeed (though no interfaces to send on)
        let result = node.announce_queued(&dest, &identity, Some(b"hello"));
        assert!(result.is_ok());

        node.shutdown();
    }

    #[test]
    fn has_path_and_hops_to() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let dh = rns_core::types::DestHash([0xAA; 16]);

        // No path should exist
        assert!(!node.has_path(&dh).unwrap());
        assert_eq!(node.hops_to(&dh).unwrap(), None);

        node.shutdown();
    }

    #[test]
    fn recall_identity_none_when_unknown() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let dh = rns_core::types::DestHash([0xBB; 16]);
        assert!(node.recall_identity(&dh).unwrap().is_none());

        node.shutdown();
    }

    #[test]
    fn request_path_does_not_crash() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let dh = rns_core::types::DestHash([0xCC; 16]);
        assert!(node.request_path(&dh).is_ok());

        // Small wait for the event to be processed
        thread::sleep(Duration::from_millis(50));

        node.shutdown();
    }

    #[test]
    fn create_link_returns_error_while_draining() {
        let node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();

        node.begin_drain(Duration::from_secs(1)).unwrap();
        assert!(node.create_link([0xAB; 16], [0xCD; 32]).is_err());

        node.shutdown();
    }

    #[test]
    fn request_path_returns_error_while_draining() {
        let node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();

        node.begin_drain(Duration::from_secs(1)).unwrap();
        assert!(node
            .request_path(&rns_core::types::DestHash([0xAB; 16]))
            .is_err());

        node.shutdown();
    }

    // =========================================================================
    // Phase 9d: send_packet + register_destination_with_proof tests
    // =========================================================================

    #[test]
    fn send_packet_returns_error_while_draining() {
        let node = RnsNode::start(NodeConfig::default(), Box::new(NoopCallbacks)).unwrap();
        let dest = crate::destination::Destination::plain("drain-test", &["send"]);

        node.begin_drain(Duration::from_secs(1)).unwrap();
        assert!(node.send_packet_queued(&dest, b"hello").is_err());

        node.shutdown();
    }

    #[test]
    fn send_packet_plain() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let dest = crate::destination::Destination::plain("test", &["echo"]);
        let result = node.send_packet_queued(&dest, b"hello world");
        assert!(result.is_ok());

        let packet_hash = result.unwrap();
        // Packet hash should be non-zero
        assert_ne!(packet_hash.0, [0u8; 32]);

        // Small wait for the event to be processed
        thread::sleep(Duration::from_millis(50));

        node.shutdown();
    }

    #[test]
    fn send_packet_single_requires_public_key() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        // single_in has no public_key — sending should fail
        let dest = crate::destination::Destination::single_in(
            "test",
            &["echo"],
            rns_core::types::IdentityHash([0x42; 16]),
        );
        let result = node.send_packet_queued(&dest, b"hello");
        assert!(result.is_err(), "single_in has no public_key, should fail");

        node.shutdown();
    }

    #[test]
    fn send_packet_single_encrypts() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        // Create a proper OUT SINGLE destination with a real identity's public key
        let remote_identity = Identity::new(&mut OsRng);
        let recalled = crate::destination::AnnouncedIdentity {
            dest_hash: rns_core::types::DestHash([0xAA; 16]),
            identity_hash: rns_core::types::IdentityHash(*remote_identity.hash()),
            public_key: remote_identity.get_public_key().unwrap(),
            app_data: None,
            hops: 1,
            received_at: 0.0,
            receiving_interface: rns_core::transport::types::InterfaceId(0),
            rssi: Some(-100),
            snr: Some(10.5),
        };
        let dest = crate::destination::Destination::single_out("test", &["echo"], &recalled);

        let result = node.send_packet_queued(&dest, b"secret message");
        assert!(result.is_ok());

        let packet_hash = result.unwrap();
        assert_ne!(packet_hash.0, [0u8; 32]);

        thread::sleep(Duration::from_millis(50));
        node.shutdown();
    }

    #[test]
    fn register_destination_with_proof_prove_all() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        let identity = Identity::new(&mut OsRng);
        let ih = rns_core::types::IdentityHash(*identity.hash());
        let dest = crate::destination::Destination::single_in("echo", &["request"], ih)
            .set_proof_strategy(rns_core::types::ProofStrategy::ProveAll);
        let prv_key = identity.get_private_key().unwrap();

        let result = node.register_destination_with_proof(&dest, Some(prv_key));
        assert!(result.is_ok());

        // Small wait for the events to be processed
        thread::sleep(Duration::from_millis(50));

        node.shutdown();
    }

    #[test]
    fn register_destination_with_proof_prove_none() {
        let node = RnsNode::start(
            NodeConfig {
                panic_on_interface_error: false,
                link_proof_verify_threads: Default::default(),
                stamps: Default::default(),
                transport_enabled: false,
                static_transport_identity: false,
                local_hops_delta: false,
                identity: None,
                interfaces: vec![],
                share_instance: false,
                instance_name: "default".into(),
                shared_instance_port: 37428,
                rpc_port: 0,
                cache_dir: None,
                ratchet_store: None,
                ratchet_expiry: Duration::from_secs(rns_core::constants::RATCHET_EXPIRY),
                management: Default::default(),
                probe_port: None,
                probe_addrs: vec![],
                probe_protocol: rns_core::holepunch::ProbeProtocol::Rnsp,
                direct_connect_policy: Default::default(),
                direct_connect_allow_private_endpoints: false,
                device: None,
                underlay_mark: None,
                hooks: Vec::new(),
                discover_interfaces: false,
                autoconnect_interface_mode: None,
                autoconnect_interface_gravity: 0,
                autoconnect_announces_to_internal: false,
                autoconnect_unverified_implementations: false,
                discovery_required_value: None,
                respond_to_probes: false,
                prefer_shorter_path: false,
                max_paths_per_destination: 1,
                packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
                packet_hashlist_allocation:
                    rns_core::transport::types::PacketHashlistAllocation::Eager,
                max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
                max_path_destinations: usize::MAX,
                max_tunnel_destinations_total: usize::MAX,
                known_destinations_ttl: DEFAULT_KNOWN_DESTINATIONS_TTL,
                known_destinations_max_entries: DEFAULT_KNOWN_DESTINATIONS_MAX_ENTRIES,
                announce_table_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_TABLE_TTL as u64,
                ),
                announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
                driver_event_queue_capacity: crate::event::DEFAULT_EVENT_QUEUE_CAPACITY,
                interface_writer_queue_capacity:
                    crate::interface::DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY,
                announce_rate_defaults: AnnounceRateDefaults::default(),
                ingress_control_defaults: rns_core::transport::types::IngressControlConfig::enabled(
                ),
                #[cfg(feature = "iface-backbone")]
                backbone_peer_pool: None,
                announce_sig_cache_enabled: true,
                announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
                announce_sig_cache_ttl: Duration::from_secs(
                    rns_core::constants::ANNOUNCE_SIG_CACHE_TTL as u64,
                ),
                registry: None,
                #[cfg(feature = "hooks")]
                provider_bridge: None,
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        // ProveNone should not send RegisterProofStrategy event
        let dest = crate::destination::Destination::plain("test", &["data"])
            .set_proof_strategy(rns_core::types::ProofStrategy::ProveNone);

        let result = node.register_destination_with_proof(&dest, None);
        assert!(result.is_ok());

        thread::sleep(Duration::from_millis(50));
        node.shutdown();
    }

    #[test]
    fn destination_stamp_below_target_value_is_regenerated() {
        // Destination._announce (d7749b0b): the held stamp is reloaded or
        // regenerated when its value is below `target_value`, and both the
        // lookup and the grind use the target (Destination.py:319-323).
        let dir = tempdir().unwrap();
        let store = crate::stamps::DestinationStampStore::new(dir.path());
        let targets = crate::stamper::StampTargets {
            destination_value: StampCost::from_whole(4),
            path_request_value: StampCost::from_whole(4),
        };
        let identity = Identity::from_private_key(&[0x52; 64]);
        let mut dest = crate::Destination::single_in(
            "stamper",
            &["target"],
            rns_core::types::IdentityHash(*identity.hash()),
        );
        let name_hash = rns_core::destination::name_hash("stamper", &["target"]);
        let material = crate::stamps::destination_stamp_material(
            &dest.hash.0,
            &identity.get_public_key().unwrap(),
            &name_hash,
        );

        ensure_destination_stamp(&store, targets, &dest, &identity).unwrap();
        let (first, first_value) = store
            .load(&dest.hash.0, &material, StampCost::ZERO)
            .expect("stamp persisted at the default target");
        assert!(first_value.meets(StampCost::from_whole(4)));

        // A target the held stamp already meets reuses it.
        dest.set_value(f64::from(first_value.whole()));
        ensure_destination_stamp(&store, targets, &dest, &identity).unwrap();
        assert_eq!(
            store
                .load(&dest.hash.0, &material, StampCost::ZERO)
                .unwrap()
                .0,
            first
        );

        // A target above the held value regenerates and persists a stamp
        // that reaches it.
        let raised = first_value.whole() as u8 + 1;
        dest.set_value(f64::from(raised));
        ensure_destination_stamp(&store, targets, &dest, &identity).unwrap();
        let (second, second_value) = store
            .load(&dest.hash.0, &material, StampCost::ZERO)
            .unwrap();
        assert_ne!(second, first);
        assert!(second_value.meets(StampCost::from_whole(u32::from(raised))));

        // The announce builder reloads at the same target and carries the
        // regenerated stamp and the configured default link cost.
        let (tx, _rx) = crate::event::channel();
        tx.link_request_costs()
            .set_default_cost(StampCost::from_whole(7));
        let (packet, _) =
            build_announce_packet(&tx, Some(&store), true, targets, &dest, &identity, None)
                .unwrap();
        let announce = rns_core::announce::AnnounceData::unpack(packet.data(), false).unwrap();
        let app_data = announce
            .app_data
            .expect("stamped announce carries app data");
        let ext = &app_data[app_data.len() - rns_core::constants::EXT_PAYLOAD_LENGTH..];
        assert_eq!(ext[0], rns_core::constants::EXT_PAYLOAD_MARKER);
        assert_eq!(
            &ext[rns_core::constants::EXT_HEADER_LENGTH
                ..rns_core::constants::EXT_HEADER_LENGTH + rns_core::constants::STAMP_LENGTH],
            &second[..]
        );
        assert_eq!(ext[rns_core::constants::EXT_PAYLOAD_LENGTH - 1], 7);
    }

    #[test]
    fn from_config_applies_the_stamps_section_and_hop_limiter() {
        let dir = tempdir().unwrap();
        std::fs::write(
            dir.path().join("config"),
            "[reticulum]\n\
             hop_limiter = 12\n\
             [stamps]\n\
             enforce_stamps = yes\n\
             stamp_destinations = yes\n\
             stamp_prs = yes\n\
             always_stamp_lrs = yes\n\
             destination_cost = 18\n\
             destination_value = 27\n\
             pr_cost = 11\n\
             pr_value = 13\n\
             lr_cost = 9\n\
             pr_stamp_timeout = 3\n",
        )
        .unwrap();
        let node =
            RnsNode::from_private_config(dir.path(), dir.path(), None, Box::new(NoopCallbacks))
                .unwrap();

        assert!(node.stamp_destinations.load(Ordering::Relaxed));
        assert_eq!(
            node.stamp_targets,
            crate::stamper::StampTargets {
                destination_value: StampCost::from_whole(27),
                path_request_value: StampCost::from_whole(13),
            }
        );
        // `lr_cost` and `enforce_stamps` reach the shared link cost table.
        assert_eq!(node.link_cost(&[0u8; 16]), 9.0);
        assert!(node.link_request_stamps_enforced());
        let stamper = node.stamper().unwrap();
        assert_eq!(stamper.targets(), node.stamp_targets);
        assert_eq!(stamper.path_request_timeout(), Duration::from_secs(3));

        // The runtime keys report the configured values as their startup
        // defaults, so a reset returns to the config file and not to the
        // compile-time constants.
        for key in [
            "global.stamp_destinations",
            "global.stamp_path_requests",
            "global.stamps_enforced",
            "global.stamp_link_requests",
            "global.always_stamp_link_requests",
        ] {
            let QueryResponse::RuntimeConfigEntry(Some(entry)) = node
                .query(QueryRequest::GetRuntimeConfig { key: key.into() })
                .unwrap()
            else {
                panic!("expected runtime config entry for {key}");
            };
            assert_eq!(
                entry.value,
                crate::event::RuntimeConfigValue::Bool(true),
                "{key}"
            );
            assert_eq!(
                entry.default,
                crate::event::RuntimeConfigValue::Bool(true),
                "{key}"
            );
        }
        node.shutdown();
    }

    #[test]
    fn start_applies_the_node_config_stamps() {
        // Upstream eb826cde: `Reticulum(enable_stamping=True,
        // enforce_stamps=True)` for a node configured in code.
        let node = RnsNode::start(
            NodeConfig {
                stamps: config::StampsSection {
                    stamp_destinations: true,
                    stamp_prs: true,
                    stamps_enforced: true,
                    lr_stamps_enforced: true,
                    lr_cost: StampCost::from_whole(9),
                    ..Default::default()
                },
                ..NodeConfig::default()
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        assert!(node.stamp_destinations.load(Ordering::Relaxed));
        assert_eq!(node.link_cost(&[0u8; 16]), 9.0);
        assert!(node.link_request_stamps_enforced());
        for key in [
            "global.stamp_destinations",
            "global.stamp_path_requests",
            "global.stamps_enforced",
            "global.lr_stamps_enforced",
        ] {
            let QueryResponse::RuntimeConfigEntry(Some(entry)) = node
                .query(QueryRequest::GetRuntimeConfig { key: key.into() })
                .unwrap()
            else {
                panic!("expected runtime config entry for {key}");
            };
            assert_eq!(
                entry.value,
                crate::event::RuntimeConfigValue::Bool(true),
                "{key}"
            );
            assert_eq!(
                entry.default,
                crate::event::RuntimeConfigValue::Bool(true),
                "{key}"
            );
        }
        node.shutdown();
    }

    #[test]
    fn published_blackhole_destination_requires_its_link_cost() {
        let identity = Identity::new(&mut OsRng);
        let blackhole = crate::management::blackhole_dest_hash(identity.hash());
        let node = RnsNode::start(
            NodeConfig {
                identity: Some(Identity::from_private_key(
                    &identity.get_private_key().unwrap(),
                )),
                management: crate::ManagementConfig {
                    publish_blackhole: true,
                    ..Default::default()
                },
                ..NodeConfig::default()
            },
            Box::new(NoopCallbacks),
        )
        .unwrap();

        assert_eq!(node.link_cost(&blackhole), 5.25);
        // Other destinations keep the `LR_COST` default.
        assert_eq!(node.link_cost(&[0u8; 16]), 12.5);
        node.shutdown();
    }
}