rns_net/

rpc.rs

//! RPC server and client for cross-process daemon communication.
//!
//! Implements Python `multiprocessing.connection` wire protocol:
//! - 4-byte big-endian signed i32 length prefix + payload
//! - HMAC-SHA256 challenge-response authentication
//! - msgpack serialization for request/response dictionaries
//!
//! Server translates RPC dicts into [`QueryRequest`] events, sends
//! them through the driver event channel, and returns RPC responses.

use std::io::{self, Read, Write};
use std::net::{IpAddr, TcpListener, TcpStream};
#[cfg(target_os = "linux")]
use std::os::linux::net::SocketAddrExt;
#[cfg(target_os = "linux")]
use std::os::unix::net::{SocketAddr, UnixListener};
use std::sync::mpsc;
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};
use std::thread;
use std::time::Duration;

use rns_core::msgpack::{self, Value as MsgpackValue};
use rns_core::transport::types::InterfaceId;
use rns_crypto::hmac::hmac_sha256;
use rns_crypto::sha256::sha256;

#[cfg(test)]
use crate::event::TrafficDetail;
use crate::event::{
    BackboneInterfaceEntry, BackbonePeerStateEntry, BlackholeInfo, BurstClassStats, DrainStatus,
    DynamicBurstStats, Event, EventSender, HookInfo, InterfaceStatsResponse, KnownDestinationEntry,
    LifecycleState, PathTableEntry, ProviderBridgeStats, QueryRequest, QueryResponse,
    RateTableEntry, RuntimeConfigApplyMode, RuntimeConfigEntry, RuntimeConfigError,
    RuntimeConfigErrorCode, RuntimeConfigSource, RuntimeConfigValue, SingleInterfaceStat,
};
use crate::md5::hmac_md5;
use crate::pickle::{self, PickleValue};

const CHALLENGE_PREFIX: &[u8] = b"#CHALLENGE#";
const WELCOME: &[u8] = b"#WELCOME#";
const FAILURE: &[u8] = b"#FAILURE#";
const CHALLENGE_LEN: usize = 40;
const MAX_CONCURRENT_RPC_CONNECTIONS: usize = 32;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RpcCodec {
    Msgpack,
    Pickle,
}

/// RPC address types.
#[derive(Debug, Clone)]
pub enum RpcAddr {
    Tcp(String, u16),
}

/// RPC server that listens for incoming connections and handles queries.
pub struct RpcServer {
    shutdown: Arc<AtomicBool>,
    thread: Option<thread::JoinHandle<()>>,
}

impl RpcServer {
    /// Start the RPC server on the given address.
    pub fn start(addr: &RpcAddr, auth_key: [u8; 32], event_tx: EventSender) -> io::Result<Self> {
        let shutdown = Arc::new(AtomicBool::new(false));
        let shutdown2 = shutdown.clone();

        let listener = match addr {
            RpcAddr::Tcp(host, port) => {
                let l = TcpListener::bind((host.as_str(), *port))?;
                // Non-blocking so we can check shutdown flag
                l.set_nonblocking(true)?;
                l
            }
        };

        let thread = thread::Builder::new()
            .name("rpc-server".into())
            .spawn(move || {
                rpc_server_loop(listener, auth_key, event_tx, shutdown2);
            })
            .map_err(io::Error::other)?;

        Ok(RpcServer {
            shutdown,
            thread: Some(thread),
        })
    }

    /// Listen on the abstract socket used by Python Reticulum on Linux.
    #[cfg(target_os = "linux")]
    pub fn start_abstract_unix(
        instance_name: &str,
        auth_key: [u8; 32],
        event_tx: EventSender,
    ) -> io::Result<Self> {
        let addr = SocketAddr::from_abstract_name(format!("rns/{instance_name}/rpc"))?;
        let listener = UnixListener::bind_addr(&addr)?;
        listener.set_nonblocking(true)?;
        let shutdown = Arc::new(AtomicBool::new(false));
        let shutdown2 = shutdown.clone();
        let thread = thread::Builder::new()
            .name("rpc-unix-server".into())
            .spawn(move || rpc_server_loop_unix(listener, auth_key, event_tx, shutdown2))
            .map_err(io::Error::other)?;
        Ok(Self {
            shutdown,
            thread: Some(thread),
        })
    }

    /// Stop the RPC server.
    pub fn stop(&mut self) {
        self.shutdown.store(true, Ordering::Relaxed);
        if let Some(handle) = self.thread.take() {
            let _ = handle.join();
        }
    }
}

#[cfg(target_os = "linux")]
fn rpc_server_loop_unix(
    listener: UnixListener,
    auth_key: [u8; 32],
    event_tx: EventSender,
    shutdown: Arc<AtomicBool>,
) {
    let mut workers = Vec::new();
    loop {
        reap_finished_rpc_workers(&mut workers);
        if shutdown.load(Ordering::Relaxed) {
            break;
        }
        match listener.accept() {
            Ok((stream, _)) => {
                if workers.len() >= MAX_CONCURRENT_RPC_CONNECTIONS {
                    log::warn!("rejecting Unix RPC connection: {MAX_CONCURRENT_RPC_CONNECTIONS} handlers already active");
                    continue;
                }
                let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
                let _ = stream.set_write_timeout(Some(Duration::from_secs(10)));
                let worker_event_tx = event_tx.clone();
                match thread::Builder::new()
                    .name("rpc-unix-client".into())
                    .spawn(move || {
                        if let Err(e) = handle_connection(stream, &auth_key, &worker_event_tx) {
                            log::debug!("Unix RPC connection error: {e}");
                        }
                    }) {
                    Ok(worker) => workers.push(worker),
                    Err(e) => log::error!("failed to start Unix RPC connection handler: {e}"),
                }
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                thread::sleep(Duration::from_millis(10))
            }
            Err(e) => {
                log::error!("Unix RPC accept error: {e}");
                thread::sleep(Duration::from_millis(100));
            }
        }
    }
    for worker in workers {
        let _ = worker.join();
    }
}

impl Drop for RpcServer {
    fn drop(&mut self) {
        self.stop();
    }
}

fn rpc_server_loop(
    listener: TcpListener,
    auth_key: [u8; 32],
    event_tx: EventSender,
    shutdown: Arc<AtomicBool>,
) {
    let mut workers = Vec::new();

    loop {
        reap_finished_rpc_workers(&mut workers);

        if shutdown.load(Ordering::Relaxed) {
            break;
        }

        match listener.accept() {
            Ok((stream, addr)) => {
                if workers.len() >= MAX_CONCURRENT_RPC_CONNECTIONS {
                    log::warn!(
                        "rejecting RPC connection from {addr}: {MAX_CONCURRENT_RPC_CONNECTIONS} handlers already active"
                    );
                    continue;
                }

                // Set blocking for this connection
                let _ = stream.set_nonblocking(false);
                let _ = stream.set_read_timeout(Some(std::time::Duration::from_secs(10)));
                let _ = stream.set_write_timeout(Some(std::time::Duration::from_secs(10)));

                let worker_event_tx = event_tx.clone();
                match thread::Builder::new()
                    .name("rpc-client".into())
                    .spawn(move || {
                        if let Err(e) = handle_connection(stream, &auth_key, &worker_event_tx) {
                            log::debug!("RPC connection error: {e}");
                        }
                    }) {
                    Ok(worker) => workers.push(worker),
                    Err(e) => log::error!("failed to start RPC connection handler: {e}"),
                }
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                // No pending connection, sleep briefly and retry
                thread::sleep(std::time::Duration::from_millis(10));
            }
            Err(e) => {
                log::error!("RPC accept error: {}", e);
                thread::sleep(std::time::Duration::from_millis(100));
            }
        }
    }

    for worker in workers {
        let _ = worker.join();
    }
}

fn reap_finished_rpc_workers(workers: &mut Vec<thread::JoinHandle<()>>) {
    let mut index = 0;
    while index < workers.len() {
        if workers[index].is_finished() {
            let worker = workers.swap_remove(index);
            let _ = worker.join();
        } else {
            index += 1;
        }
    }
}

fn handle_connection<S: Read + Write>(
    mut stream: S,
    auth_key: &[u8; 32],
    event_tx: &EventSender,
) -> io::Result<()> {
    // Authentication: send challenge, verify response
    server_auth(&mut stream, auth_key)?;
    // multiprocessing.connection authenticates both peers before requests.
    answer_challenge(&mut stream, auth_key)?;

    // Read request.
    let request_bytes = recv_bytes(&mut stream)?;
    let (request, codec) = decode_rpc_payload(&request_bytes)?;

    // Translate request dict to query, send to driver, get response.
    let response = handle_rpc_request(&request, event_tx)?;

    // Encode response and send
    let response_bytes = encode_rpc_payload(&response, codec);
    send_bytes(&mut stream, &response_bytes)?;

    Ok(())
}

fn decode_rpc_payload(data: &[u8]) -> io::Result<(PickleValue, RpcCodec)> {
    if let Ok(value) = msgpack::unpack_exact(data) {
        return Ok((msgpack_to_pickle(&value), RpcCodec::Msgpack));
    }
    pickle::decode(data)
        .map(|value| (value, RpcCodec::Pickle))
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))
}

fn encode_rpc_payload(value: &PickleValue, codec: RpcCodec) -> Vec<u8> {
    match codec {
        RpcCodec::Msgpack => msgpack::pack(&pickle_to_msgpack(value)),
        RpcCodec::Pickle => pickle::encode(value),
    }
}

fn pickle_to_msgpack(value: &PickleValue) -> MsgpackValue {
    match value {
        PickleValue::None => MsgpackValue::Nil,
        PickleValue::Bool(v) => MsgpackValue::Bool(*v),
        PickleValue::Int(v) if *v >= 0 => MsgpackValue::UInt(*v as u64),
        PickleValue::Int(v) => MsgpackValue::Int(*v),
        PickleValue::Float(v) => MsgpackValue::Float(*v),
        PickleValue::String(v) => MsgpackValue::Str(v.clone()),
        PickleValue::Bytes(v) => MsgpackValue::Bin(v.clone()),
        PickleValue::List(values) => {
            MsgpackValue::Array(values.iter().map(pickle_to_msgpack).collect())
        }
        PickleValue::Dict(values) => MsgpackValue::Map(
            values
                .iter()
                .map(|(key, value)| (pickle_to_msgpack(key), pickle_to_msgpack(value)))
                .collect(),
        ),
    }
}

fn msgpack_to_pickle(value: &MsgpackValue) -> PickleValue {
    match value {
        MsgpackValue::Nil => PickleValue::None,
        MsgpackValue::Bool(v) => PickleValue::Bool(*v),
        MsgpackValue::UInt(v) if *v <= i64::MAX as u64 => PickleValue::Int(*v as i64),
        MsgpackValue::UInt(v) => PickleValue::Float(*v as f64),
        MsgpackValue::Int(v) => PickleValue::Int(*v),
        MsgpackValue::Float(v) => PickleValue::Float(*v),
        MsgpackValue::Bin(v) => PickleValue::Bytes(v.clone()),
        MsgpackValue::Str(v) => PickleValue::String(v.clone()),
        MsgpackValue::Array(values) => {
            PickleValue::List(values.iter().map(msgpack_to_pickle).collect())
        }
        MsgpackValue::Map(values) => PickleValue::Dict(
            values
                .iter()
                .map(|(key, value)| (msgpack_to_pickle(key), msgpack_to_pickle(value)))
                .collect(),
        ),
    }
}

/// Server-side authentication: challenge-response.
fn server_auth<S: Read + Write>(stream: &mut S, auth_key: &[u8; 32]) -> io::Result<()> {
    // Generate challenge: #CHALLENGE#{sha256}<40 random bytes>
    let mut random_bytes = [0u8; CHALLENGE_LEN];
    // Use /dev/urandom for randomness
    {
        let mut f = std::fs::File::open("/dev/urandom")?;
        f.read_exact(&mut random_bytes)?;
    }

    let mut challenge_message = Vec::with_capacity(CHALLENGE_PREFIX.len() + 8 + CHALLENGE_LEN);
    challenge_message.extend_from_slice(CHALLENGE_PREFIX);
    challenge_message.extend_from_slice(b"{sha256}");
    challenge_message.extend_from_slice(&random_bytes);

    send_bytes(stream, &challenge_message)?;

    // Read response (max 256 bytes)
    let response = recv_bytes(stream)?;

    // Verify response
    // The message to HMAC is everything after #CHALLENGE# (i.e. {sha256}<random>)
    let message = &challenge_message[CHALLENGE_PREFIX.len()..];

    if verify_response(auth_key, message, &response) {
        send_bytes(stream, WELCOME)?;
        Ok(())
    } else {
        send_bytes(stream, FAILURE)?;
        Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "auth failed",
        ))
    }
}

/// Verify a client's HMAC response.
fn verify_response(auth_key: &[u8; 32], message: &[u8], response: &[u8]) -> bool {
    // Modern protocol: response = {sha256}<hmac-sha256 digest>
    if response.starts_with(b"{sha256}") {
        let digest = &response[8..];
        let expected = hmac_sha256(auth_key, message);
        constant_time_eq(digest, &expected)
    }
    // Legacy protocol: response = raw 16-byte HMAC-MD5 digest
    else if response.len() == 16 {
        let expected = hmac_md5(auth_key, message);
        constant_time_eq(response, &expected)
    }
    // Legacy with {md5} prefix
    else if response.starts_with(b"{md5}") {
        let digest = &response[5..];
        let expected = hmac_md5(auth_key, message);
        constant_time_eq(digest, &expected)
    } else {
        false
    }
}

/// Constant-time byte comparison.
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Send bytes with 4-byte big-endian length prefix.
fn send_bytes<S: Write>(stream: &mut S, data: &[u8]) -> io::Result<()> {
    let len = data.len() as i32;
    stream.write_all(&len.to_be_bytes())?;
    stream.write_all(data)?;
    stream.flush()
}

/// Receive bytes with 4-byte big-endian length prefix.
fn recv_bytes<S: Read>(stream: &mut S) -> io::Result<Vec<u8>> {
    let mut len_buf = [0u8; 4];
    stream.read_exact(&mut len_buf)?;
    let len = i32::from_be_bytes(len_buf);

    if len < 0 {
        // Extended format: 8-byte length
        let mut len8_buf = [0u8; 8];
        stream.read_exact(&mut len8_buf)?;
        let len = u64::from_be_bytes(len8_buf) as usize;
        if len > 64 * 1024 * 1024 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "message too large",
            ));
        }
        let mut buf = vec![0u8; len];
        stream.read_exact(&mut buf)?;
        Ok(buf)
    } else {
        let len = len as usize;
        if len > 64 * 1024 * 1024 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "message too large",
            ));
        }
        let mut buf = vec![0u8; len];
        stream.read_exact(&mut buf)?;
        Ok(buf)
    }
}

/// Default probe payload size when the request omits `size`.
const DEFAULT_PROBE_PAYLOAD_SIZE: usize = 16;

/// Encode a failed `send_probe` for RPC clients.
///
/// `UnknownIdentity` stays `None` for compatibility with older clients. Other
/// failures return `{"error": <code>, ...}` so clients can tell them apart.
fn send_probe_error_pickle(err: crate::event::SendProbeError) -> PickleValue {
    use crate::event::SendProbeError;
    let code = match err {
        SendProbeError::UnknownIdentity => return PickleValue::None,
        SendProbeError::PayloadTooLarge { .. } => "payload_too_large",
        SendProbeError::EncryptFailed => "encrypt_failed",
        SendProbeError::PackFailed => "pack_failed",
    };
    let mut entries = vec![(
        PickleValue::String("error".into()),
        PickleValue::String(code.into()),
    )];
    if let SendProbeError::PayloadTooLarge { max } = err {
        entries.push((
            PickleValue::String("max_size".into()),
            PickleValue::Int(max as i64),
        ));
    }
    PickleValue::Dict(entries)
}

/// Translate an RPC request dict to a query event and get response.
fn handle_rpc_request(request: &PickleValue, event_tx: &EventSender) -> io::Result<PickleValue> {
    if let Some(operation) = request.get("manage").and_then(PickleValue::as_str) {
        let operation = match operation {
            "attach_interface" => crate::event::InterfaceManagementOperation::Attach,
            "detach_interface" => crate::event::InterfaceManagementOperation::Detach,
            "reload_interface" => crate::event::InterfaceManagementOperation::Reload,
            _ => return Ok(PickleValue::None),
        };
        let name = request
            .get("name")
            .and_then(PickleValue::as_str)
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing interface name"))?;
        let result = send_query(
            event_tx,
            QueryRequest::ManageInterface {
                operation,
                name: name.to_string(),
            },
        )?;
        return Ok(match result {
            QueryResponse::InterfaceManagement(Some(value)) => PickleValue::Bool(value),
            _ => PickleValue::None,
        });
    }
    // Handle "get" requests
    if let Some(get_val) = request.get("get") {
        if let Some(path) = get_val.as_str() {
            return match path {
                "interface_stats" => {
                    let resp = send_query(event_tx, QueryRequest::InterfaceStats)?;
                    if let QueryResponse::InterfaceStats(stats) = resp {
                        let mut response = interface_stats_to_pickle(&stats);
                        add_inbound_runtime_stats(&mut response, &stats, event_tx);
                        Ok(response)
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "profiling_results" => {
                    Ok(crate::profiling::results_pickle().unwrap_or(PickleValue::None))
                }
                "path_table" => {
                    let max_hops = request
                        .get("max_hops")
                        .and_then(|v| v.as_int().map(|n| n as u8));
                    let resp = send_query(event_tx, QueryRequest::PathTable { max_hops })?;
                    if let QueryResponse::PathTable(entries) = resp {
                        Ok(path_table_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "rate_table" => {
                    let resp = send_query(event_tx, QueryRequest::RateTable)?;
                    if let QueryResponse::RateTable(entries) = resp {
                        Ok(rate_table_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "lowest_interface_bitrate" => {
                    let resp = send_query(event_tx, QueryRequest::LowestInterfaceBitrate)?;
                    if let QueryResponse::LowestInterfaceBitrate(Some(bitrate)) = resp {
                        Ok(PickleValue::Int(bitrate as i64))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "medium_path_timeout" => {
                    let resp = send_query(event_tx, QueryRequest::MediumPathTimeout)?;
                    if let QueryResponse::MediumPathTimeout(timeout) = resp {
                        Ok(PickleValue::Float(timeout))
                    } else {
                        Ok(PickleValue::Int(0))
                    }
                }
                "first_hop_timeout" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp =
                        send_query(event_tx, QueryRequest::FirstHopTimeout { dest_hash: hash })?;
                    if let QueryResponse::FirstHopTimeout(timeout) = resp {
                        Ok(PickleValue::Float(timeout))
                    } else {
                        Ok(PickleValue::Float(
                            rns_core::constants::LINK_ESTABLISHMENT_TIMEOUT_PER_HOP,
                        ))
                    }
                }
                "next_hop" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp = send_query(event_tx, QueryRequest::NextHop { dest_hash: hash })?;
                    if let QueryResponse::NextHop(Some(nh)) = resp {
                        Ok(PickleValue::Bytes(nh.next_hop.to_vec()))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "next_hop_if_name" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp =
                        send_query(event_tx, QueryRequest::NextHopIfName { dest_hash: hash })?;
                    if let QueryResponse::NextHopIfName(Some(name)) = resp {
                        Ok(PickleValue::String(name))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "link_count" => {
                    let resp = send_query(event_tx, QueryRequest::LinkCount)?;
                    if let QueryResponse::LinkCount(n) = resp {
                        Ok(PickleValue::Int(n as i64))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "active_link_count" => {
                    let resp = send_query(event_tx, QueryRequest::ActiveLinkCount)?;
                    if let QueryResponse::ActiveLinkCount(n) = resp {
                        Ok(PickleValue::Int(n as i64))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "transport_identity" => {
                    let resp = send_query(event_tx, QueryRequest::TransportIdentity)?;
                    if let QueryResponse::TransportIdentity(Some(hash)) = resp {
                        Ok(PickleValue::Bytes(hash.to_vec()))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "blackholed" => {
                    let resp = send_query(event_tx, QueryRequest::GetBlackholed)?;
                    if let QueryResponse::Blackholed(entries) = resp {
                        Ok(blackholed_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "discovered_interfaces" => {
                    let only_available = request
                        .get("only_available")
                        .and_then(|v| v.as_bool())
                        .unwrap_or(false);
                    let only_transport = request
                        .get("only_transport")
                        .and_then(|v| v.as_bool())
                        .unwrap_or(false);
                    let resp = send_query(
                        event_tx,
                        QueryRequest::DiscoveredInterfaces {
                            only_available,
                            only_transport,
                        },
                    )?;
                    if let QueryResponse::DiscoveredInterfaces(interfaces) = resp {
                        Ok(discovered_interfaces_to_pickle(&interfaces))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "hooks" => {
                    let (response_tx, response_rx) = mpsc::channel();
                    try_send_event(event_tx, Event::ListHooks { response_tx })?;
                    let hooks = response_rx
                        .recv_timeout(std::time::Duration::from_secs(5))
                        .map_err(|_| {
                            io::Error::new(io::ErrorKind::TimedOut, "list hooks timed out")
                        })?;
                    Ok(hooks_to_pickle(&hooks))
                }
                "runtime_config" => {
                    let resp = send_query(event_tx, QueryRequest::ListRuntimeConfig)?;
                    if let QueryResponse::RuntimeConfigList(entries) = resp {
                        Ok(runtime_config_list_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "known_destinations" => {
                    let resp = send_query(event_tx, QueryRequest::KnownDestinations)?;
                    if let QueryResponse::KnownDestinations(entries) = resp {
                        Ok(known_destinations_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "runtime_config_entry" => {
                    let key = request
                        .get("key")
                        .and_then(|v| v.as_str())
                        .unwrap_or_default()
                        .to_string();
                    let resp = send_query(event_tx, QueryRequest::GetRuntimeConfig { key })?;
                    if let QueryResponse::RuntimeConfigEntry(entry) = resp {
                        Ok(entry
                            .as_ref()
                            .map(runtime_config_entry_to_pickle)
                            .unwrap_or(PickleValue::None))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "backbone_peer_state" => {
                    let interface_name = request
                        .get("interface")
                        .and_then(|v| v.as_str())
                        .map(|s| s.to_string());
                    let resp =
                        send_query(event_tx, QueryRequest::BackbonePeerState { interface_name })?;
                    if let QueryResponse::BackbonePeerState(entries) = resp {
                        Ok(backbone_peer_state_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "backbone_interfaces" => {
                    let resp = send_query(event_tx, QueryRequest::BackboneInterfaces)?;
                    if let QueryResponse::BackboneInterfaces(entries) = resp {
                        Ok(backbone_interfaces_to_pickle(&entries))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "provider_bridge_stats" => {
                    let resp = send_query(event_tx, QueryRequest::ProviderBridgeStats)?;
                    if let QueryResponse::ProviderBridgeStats(stats) = resp {
                        Ok(stats
                            .as_ref()
                            .map(provider_bridge_stats_to_pickle)
                            .unwrap_or(PickleValue::None))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "drain_status" => {
                    let resp = send_query(event_tx, QueryRequest::DrainStatus)?;
                    if let QueryResponse::DrainStatus(status) = resp {
                        Ok(drain_status_to_pickle(&status))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                _ => Ok(PickleValue::None),
            };
        }
    }

    if let Some(begin_val) = request.get("begin_drain") {
        let timeout_secs = begin_val
            .as_float()
            .or_else(|| begin_val.as_int().map(|value| value as f64))
            .unwrap_or(0.0)
            .max(0.0);
        let timeout = Duration::from_secs_f64(timeout_secs);
        try_send_event(event_tx, Event::BeginDrain { timeout })?;
        return Ok(PickleValue::Bool(true));
    }

    if let Some(set_val) = request.get("set").and_then(|v| v.as_str()) {
        if set_val == "known_destination_retained" {
            let dest_hash = extract_dest_hash(request, "dest_hash")?;
            let resp = send_query(event_tx, QueryRequest::RetainKnownDestination { dest_hash })?;
            return if let QueryResponse::RetainKnownDestination(ok) = resp {
                Ok(PickleValue::Bool(ok))
            } else {
                Ok(PickleValue::None)
            };
        }
        if set_val == "identity_retained" {
            let identity_hash = extract_dest_hash(request, "identity_hash")?;
            let resp = send_query(event_tx, QueryRequest::RetainIdentity { identity_hash })?;
            return if let QueryResponse::RetainIdentity(ok) = resp {
                Ok(PickleValue::Bool(ok))
            } else {
                Ok(PickleValue::None)
            };
        }
        if set_val == "known_destination_used" {
            let dest_hash = extract_dest_hash(request, "dest_hash")?;
            let resp = send_query(
                event_tx,
                QueryRequest::MarkKnownDestinationUsed { dest_hash },
            )?;
            return if let QueryResponse::MarkKnownDestinationUsed(ok) = resp {
                Ok(PickleValue::Bool(ok))
            } else {
                Ok(PickleValue::None)
            };
        }
        if set_val == "runtime_config" {
            let key = request
                .get("key")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string();
            let Some(value) = request
                .get("value")
                .and_then(runtime_config_value_from_pickle)
            else {
                return Ok(runtime_config_error_to_pickle(&RuntimeConfigError {
                    code: RuntimeConfigErrorCode::InvalidType,
                    message: "runtime-config set requires a scalar value".into(),
                }));
            };
            let resp = send_query(event_tx, QueryRequest::SetRuntimeConfig { key, value })?;
            return if let QueryResponse::RuntimeConfigSet(result) = resp {
                Ok(runtime_config_result_to_pickle(result))
            } else {
                Ok(PickleValue::None)
            };
        }
    }

    if let Some(reset_val) = request.get("reset").and_then(|v| v.as_str()) {
        if reset_val == "runtime_config" {
            let key = request
                .get("key")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string();
            let resp = send_query(event_tx, QueryRequest::ResetRuntimeConfig { key })?;
            return if let QueryResponse::RuntimeConfigReset(result) = resp {
                Ok(runtime_config_result_to_pickle(result))
            } else {
                Ok(PickleValue::None)
            };
        }
    }

    if let Some(clear_val) = request.get("clear").and_then(|v| v.as_str()) {
        if clear_val == "known_destination_retained" {
            let dest_hash = extract_dest_hash(request, "dest_hash")?;
            let resp = send_query(
                event_tx,
                QueryRequest::UnretainKnownDestination { dest_hash },
            )?;
            return if let QueryResponse::UnretainKnownDestination(ok) = resp {
                Ok(PickleValue::Bool(ok))
            } else {
                Ok(PickleValue::None)
            };
        }
        if clear_val == "backbone_peer_state" {
            let interface_name = required_string(request, "interface")?;
            let peer_ip = required_string(request, "ip")?;
            let peer_ip = peer_ip
                .parse()
                .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid peer IP"))?;
            let resp = send_query(
                event_tx,
                QueryRequest::ClearBackbonePeerState {
                    interface_name,
                    peer_ip,
                },
            )?;
            return if let QueryResponse::ClearBackbonePeerState(ok) = resp {
                Ok(PickleValue::Bool(ok))
            } else {
                Ok(PickleValue::None)
            };
        }
    }

    if let Some(set_val) = request.get("set").and_then(|v| v.as_str()) {
        if set_val == "backbone_peer_blacklist" {
            let interface_name = required_string(request, "interface")?;
            let peer_ip = required_string(request, "ip")?;
            let peer_ip: IpAddr = peer_ip
                .parse()
                .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "invalid peer IP"))?;
            let duration_secs = request
                .get("duration_secs")
                .and_then(|v| v.as_int())
                .ok_or_else(|| {
                    io::Error::new(io::ErrorKind::InvalidInput, "missing duration_secs")
                })?;
            let reason = request
                .get("reason")
                .and_then(|v| v.as_str())
                .unwrap_or("sentinel blacklist")
                .to_string();
            let penalty_level = request
                .get("penalty_level")
                .and_then(|v| v.as_int())
                .unwrap_or(0)
                .clamp(0, u8::MAX as i64) as u8;
            let resp = send_query(
                event_tx,
                QueryRequest::BlacklistBackbonePeer {
                    interface_name,
                    peer_ip,
                    duration: Duration::from_secs(duration_secs as u64),
                    reason,
                    penalty_level,
                },
            )?;
            return if let QueryResponse::BlacklistBackbonePeer(ok) = resp {
                Ok(PickleValue::Bool(ok))
            } else {
                Ok(PickleValue::None)
            };
        }
    }

    // Handle "request_path" -- trigger a path request to the network
    if let Some(hash_val) = request.get("request_path") {
        if let Some(hash_bytes) = hash_val.as_bytes() {
            if hash_bytes.len() >= 16 {
                let mut dest_hash = [0u8; 16];
                dest_hash.copy_from_slice(&hash_bytes[..16]);
                try_send_event(event_tx, crate::event::Event::RequestPath { dest_hash })?;
                return Ok(PickleValue::Bool(true));
            }
        }
    }

    // Handle "send_probe" requests
    if let Some(hash_val) = request.get("send_probe") {
        if let Some(hash_bytes) = hash_val.as_bytes() {
            if hash_bytes.len() >= 16 {
                let mut dest_hash = [0u8; 16];
                dest_hash.copy_from_slice(&hash_bytes[..16]);
                // Missing or non-positive sizes fall back to the default;
                // oversize requests are rejected rather than silently
                // replaced, so the caller learns the real limit.
                let payload_size = match request.get("size").and_then(|v| v.as_int()) {
                    Some(n) if n > 0 => usize::try_from(n).unwrap_or(usize::MAX),
                    _ => DEFAULT_PROBE_PAYLOAD_SIZE,
                };
                if payload_size > crate::event::MAX_PROBE_PAYLOAD_SIZE {
                    return Ok(send_probe_error_pickle(
                        crate::event::SendProbeError::PayloadTooLarge {
                            max: crate::event::MAX_PROBE_PAYLOAD_SIZE,
                        },
                    ));
                }
                let resp = send_query(
                    event_tx,
                    QueryRequest::SendProbe {
                        dest_hash,
                        payload_size,
                    },
                )?;
                return Ok(match resp {
                    QueryResponse::SendProbe(Ok((packet_hash, hops))) => PickleValue::Dict(vec![
                        (
                            PickleValue::String("packet_hash".into()),
                            PickleValue::Bytes(packet_hash.to_vec()),
                        ),
                        (
                            PickleValue::String("hops".into()),
                            PickleValue::Int(hops as i64),
                        ),
                    ]),
                    QueryResponse::SendProbe(Err(err)) => send_probe_error_pickle(err),
                    _ => PickleValue::None,
                });
            }
        }
    }

    // Handle "check_proof" requests
    if let Some(hash_val) = request.get("check_proof") {
        if let Some(hash_bytes) = hash_val.as_bytes() {
            if hash_bytes.len() >= 32 {
                let mut packet_hash = [0u8; 32];
                packet_hash.copy_from_slice(&hash_bytes[..32]);
                let resp = send_query(event_tx, QueryRequest::CheckProof { packet_hash })?;
                if let QueryResponse::CheckProof(Some(rtt)) = resp {
                    return Ok(PickleValue::Float(rtt));
                } else {
                    return Ok(PickleValue::None);
                }
            }
        }
    }

    // Handle "blackhole" requests
    if let Some(hash_val) = request.get("blackhole") {
        if let Some(hash_bytes) = hash_val.as_bytes() {
            if hash_bytes.len() >= 16 {
                let mut identity_hash = [0u8; 16];
                identity_hash.copy_from_slice(&hash_bytes[..16]);
                let duration_hours = request.get("duration").and_then(|v| v.as_float());
                let reason = request
                    .get("reason")
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string());
                let resp = send_query(
                    event_tx,
                    QueryRequest::BlackholeIdentity {
                        identity_hash,
                        duration_hours,
                        reason,
                    },
                )?;
                return Ok(PickleValue::Bool(matches!(
                    resp,
                    QueryResponse::BlackholeResult(true)
                )));
            }
        }
    }

    // Handle "unblackhole" requests
    if let Some(hash_val) = request.get("unblackhole") {
        if let Some(hash_bytes) = hash_val.as_bytes() {
            if hash_bytes.len() >= 16 {
                let mut identity_hash = [0u8; 16];
                identity_hash.copy_from_slice(&hash_bytes[..16]);
                let resp = send_query(
                    event_tx,
                    QueryRequest::UnblackholeIdentity { identity_hash },
                )?;
                return Ok(PickleValue::Bool(matches!(
                    resp,
                    QueryResponse::UnblackholeResult(true)
                )));
            }
        }
    }

    // Handle "drop" requests
    if let Some(drop_val) = request.get("drop") {
        if let Some(path) = drop_val.as_str() {
            return match path {
                "path" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp = send_query(event_tx, QueryRequest::DropPath { dest_hash: hash })?;
                    if let QueryResponse::DropPath(ok) = resp {
                        Ok(PickleValue::Bool(ok))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "all_via" => {
                    let hash = extract_dest_hash(request, "destination_hash")?;
                    let resp = send_query(
                        event_tx,
                        QueryRequest::DropAllVia {
                            transport_hash: hash,
                        },
                    )?;
                    if let QueryResponse::DropAllVia(n) = resp {
                        Ok(PickleValue::Int(n as i64))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                "announce_queues" => {
                    let resp = send_query(event_tx, QueryRequest::DropAnnounceQueues)?;
                    if let QueryResponse::DropAnnounceQueues = resp {
                        Ok(PickleValue::Bool(true))
                    } else {
                        Ok(PickleValue::None)
                    }
                }
                _ => Ok(PickleValue::None),
            };
        }
    }

    if let Some(hook_val) = request.get("hook").and_then(|v| v.as_str()) {
        return handle_hook_rpc_request(hook_val, request, event_tx);
    }

    Ok(PickleValue::None)
}

fn handle_hook_rpc_request(
    op: &str,
    request: &PickleValue,
    event_tx: &EventSender,
) -> io::Result<PickleValue> {
    match op {
        "load" => {
            let name = required_string(request, "name")?;
            let attach_point = required_string(request, "attach_point")?;
            let priority = request
                .get("priority")
                .and_then(|v| v.as_int())
                .unwrap_or(0) as i32;
            let wasm = request
                .get("wasm")
                .and_then(|v| v.as_bytes())
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing wasm"))?
                .to_vec();
            let (response_tx, response_rx) = mpsc::channel();
            try_send_event(
                event_tx,
                Event::LoadHook {
                    name,
                    wasm_bytes: wasm,
                    attach_point,
                    priority,
                    response_tx,
                },
            )?;
            let response = response_rx
                .recv_timeout(std::time::Duration::from_secs(5))
                .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "hook load timed out"))?;
            Ok(hook_result_to_pickle(response))
        }
        "load_builtin" => {
            let name = required_string(request, "name")?;
            let attach_point = required_string(request, "attach_point")?;
            let builtin_id = required_string(request, "builtin_id")?;
            let priority = request
                .get("priority")
                .and_then(|v| v.as_int())
                .unwrap_or(0) as i32;
            let (response_tx, response_rx) = mpsc::channel();
            try_send_event(
                event_tx,
                Event::LoadBuiltinHook {
                    name,
                    builtin_id,
                    attach_point,
                    priority,
                    response_tx,
                },
            )?;
            let response = response_rx
                .recv_timeout(std::time::Duration::from_secs(5))
                .map_err(|_| {
                    io::Error::new(io::ErrorKind::TimedOut, "built-in hook load timed out")
                })?;
            Ok(hook_result_to_pickle(response))
        }
        "unload" => {
            let name = required_string(request, "name")?;
            let attach_point = required_string(request, "attach_point")?;
            let (response_tx, response_rx) = mpsc::channel();
            try_send_event(
                event_tx,
                Event::UnloadHook {
                    name,
                    attach_point,
                    response_tx,
                },
            )?;
            let response = response_rx
                .recv_timeout(std::time::Duration::from_secs(5))
                .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "hook unload timed out"))?;
            Ok(hook_result_to_pickle(response))
        }
        "enable" | "disable" => {
            let name = required_string(request, "name")?;
            let attach_point = required_string(request, "attach_point")?;
            let enabled = op == "enable";
            let (response_tx, response_rx) = mpsc::channel();
            try_send_event(
                event_tx,
                Event::SetHookEnabled {
                    name,
                    attach_point,
                    enabled,
                    response_tx,
                },
            )?;
            let response = response_rx
                .recv_timeout(std::time::Duration::from_secs(5))
                .map_err(|_| {
                    io::Error::new(io::ErrorKind::TimedOut, "hook enable/disable timed out")
                })?;
            Ok(hook_result_to_pickle(response))
        }
        "set_priority" => {
            let name = required_string(request, "name")?;
            let attach_point = required_string(request, "attach_point")?;
            let priority = request
                .get("priority")
                .and_then(|v| v.as_int())
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing priority"))?
                as i32;
            let (response_tx, response_rx) = mpsc::channel();
            try_send_event(
                event_tx,
                Event::SetHookPriority {
                    name,
                    attach_point,
                    priority,
                    response_tx,
                },
            )?;
            let response = response_rx
                .recv_timeout(std::time::Duration::from_secs(5))
                .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "hook priority timed out"))?;
            Ok(hook_result_to_pickle(response))
        }
        _ => Ok(PickleValue::None),
    }
}

/// Send a query to the driver and wait for the response.
fn send_query(event_tx: &EventSender, request: QueryRequest) -> io::Result<QueryResponse> {
    let (resp_tx, resp_rx) = mpsc::channel();
    try_send_event(event_tx, Event::query(request, resp_tx))?;
    resp_rx
        .recv_timeout(std::time::Duration::from_secs(5))
        .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "query timed out"))
}

fn try_send_event(event_tx: &EventSender, event: Event) -> io::Result<()> {
    event_tx.try_send(event).map_err(|error| match error {
        mpsc::TrySendError::Full(_) => {
            io::Error::new(io::ErrorKind::WouldBlock, "driver event queue is full")
        }
        mpsc::TrySendError::Disconnected(_) => {
            io::Error::new(io::ErrorKind::BrokenPipe, "driver shut down")
        }
    })
}

/// Extract a 16-byte destination hash from an RPC dict field.
fn extract_dest_hash(request: &PickleValue, key: &str) -> io::Result<[u8; 16]> {
    let bytes = request
        .get(key)
        .and_then(|v| v.as_bytes())
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing destination_hash"))?;
    if bytes.len() < 16 {
        return Err(io::Error::new(io::ErrorKind::InvalidData, "hash too short"));
    }
    let mut hash = [0u8; 16];
    hash.copy_from_slice(&bytes[..16]);
    Ok(hash)
}

fn required_string(request: &PickleValue, key: &str) -> io::Result<String> {
    request
        .get(key)
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("missing {}", key)))
}

fn hook_result_to_pickle(result: Result<(), String>) -> PickleValue {
    match result {
        Ok(()) => PickleValue::Dict(vec![(
            PickleValue::String("ok".into()),
            PickleValue::Bool(true),
        )]),
        Err(error) => PickleValue::Dict(vec![
            (PickleValue::String("ok".into()), PickleValue::Bool(false)),
            (
                PickleValue::String("error".into()),
                PickleValue::String(error),
            ),
        ]),
    }
}

// --- Pickle response builders ---

fn interface_stats_to_pickle(stats: &InterfaceStatsResponse) -> PickleValue {
    let mut ifaces = Vec::new();
    for iface in &stats.interfaces {
        ifaces.push(single_iface_to_pickle(iface));
    }

    let mut dict = vec![
        (
            PickleValue::String("interfaces".into()),
            PickleValue::List(ifaces),
        ),
        (
            PickleValue::String("transport_enabled".into()),
            PickleValue::Bool(stats.transport_enabled),
        ),
        (
            PickleValue::String("transport_uptime".into()),
            PickleValue::Float(stats.transport_uptime),
        ),
        (
            PickleValue::String("rxb".into()),
            PickleValue::Int(stats.total_rxb as i64),
        ),
        (
            PickleValue::String("txb".into()),
            PickleValue::Int(stats.total_txb as i64),
        ),
        (
            PickleValue::String("rxpps".into()),
            PickleValue::Float(stats.rx_pps),
        ),
        (
            PickleValue::String("txpps".into()),
            PickleValue::Float(stats.tx_pps),
        ),
        (
            PickleValue::String("rxs".into()),
            PickleValue::Float(stats.traffic.rxs),
        ),
        (
            PickleValue::String("txs".into()),
            PickleValue::Float(stats.traffic.txs),
        ),
        (
            PickleValue::String("arxb".into()),
            PickleValue::Int(stats.traffic.arxb as i64),
        ),
        (
            PickleValue::String("atxb".into()),
            PickleValue::Int(stats.traffic.atxb as i64),
        ),
        (
            PickleValue::String("arxc".into()),
            PickleValue::Int(stats.traffic.arxc as i64),
        ),
        (
            PickleValue::String("atxc".into()),
            PickleValue::Int(stats.traffic.atxc as i64),
        ),
        (
            PickleValue::String("arxs".into()),
            PickleValue::Float(stats.traffic.arxs),
        ),
        (
            PickleValue::String("atxs".into()),
            PickleValue::Float(stats.traffic.atxs),
        ),
        (
            PickleValue::String("prxb".into()),
            PickleValue::Int(stats.traffic.prxb as i64),
        ),
        (
            PickleValue::String("ptxb".into()),
            PickleValue::Int(stats.traffic.ptxb as i64),
        ),
        (
            PickleValue::String("prxc".into()),
            PickleValue::Int(stats.traffic.prxc as i64),
        ),
        (
            PickleValue::String("ptxc".into()),
            PickleValue::Int(stats.traffic.ptxc as i64),
        ),
        (
            PickleValue::String("prxs".into()),
            PickleValue::Float(stats.traffic.prxs),
        ),
        (
            PickleValue::String("ptxs".into()),
            PickleValue::Float(stats.traffic.ptxs),
        ),
        (
            PickleValue::String("arxf".into()),
            PickleValue::Float(stats.traffic.arxf),
        ),
        (
            PickleValue::String("atxf".into()),
            PickleValue::Float(stats.traffic.atxf),
        ),
        (
            PickleValue::String("prxf".into()),
            PickleValue::Float(stats.traffic.prxf),
        ),
        (
            PickleValue::String("ptxf".into()),
            PickleValue::Float(stats.traffic.ptxf),
        ),
    ];

    if let Some(tid) = stats.transport_id {
        dict.push((
            PickleValue::String("transport_id".into()),
            PickleValue::Bytes(tid.to_vec()),
        ));
    } else {
        dict.push((
            PickleValue::String("transport_id".into()),
            PickleValue::None,
        ));
    }

    if let Some(pr) = stats.probe_responder {
        dict.push((
            PickleValue::String("probe_responder".into()),
            PickleValue::Bytes(pr.to_vec()),
        ));
    } else {
        dict.push((
            PickleValue::String("probe_responder".into()),
            PickleValue::None,
        ));
    }

    if let Some(pool) = &stats.backbone_peer_pool {
        let members = pool
            .members
            .iter()
            .map(|member| {
                let mut member_dict = vec![
                    (
                        PickleValue::String("name".into()),
                        PickleValue::String(member.name.clone()),
                    ),
                    (
                        PickleValue::String("remote".into()),
                        PickleValue::String(member.remote.clone()),
                    ),
                    (
                        PickleValue::String("source".into()),
                        PickleValue::String(member.source.clone()),
                    ),
                    (
                        PickleValue::String("priority".into()),
                        PickleValue::Int(member.priority as i64),
                    ),
                    (
                        PickleValue::String("state".into()),
                        PickleValue::String(member.state.clone()),
                    ),
                    (
                        PickleValue::String("failure_count".into()),
                        PickleValue::Int(member.failure_count as i64),
                    ),
                ];
                member_dict.push((
                    PickleValue::String("interface_id".into()),
                    member
                        .interface_id
                        .map(|id| PickleValue::Int(id as i64))
                        .unwrap_or(PickleValue::None),
                ));
                member_dict.push((
                    PickleValue::String("last_error".into()),
                    member
                        .last_error
                        .as_ref()
                        .map(|err| PickleValue::String(err.clone()))
                        .unwrap_or(PickleValue::None),
                ));
                member_dict.push((
                    PickleValue::String("cooldown_remaining_seconds".into()),
                    member
                        .cooldown_remaining_seconds
                        .map(PickleValue::Float)
                        .unwrap_or(PickleValue::None),
                ));
                PickleValue::Dict(member_dict)
            })
            .collect();
        dict.push((
            PickleValue::String("backbone_peer_pool".into()),
            PickleValue::Dict(vec![
                (
                    PickleValue::String("max_connected".into()),
                    PickleValue::Int(pool.max_connected as i64),
                ),
                (
                    PickleValue::String("active_count".into()),
                    PickleValue::Int(pool.active_count as i64),
                ),
                (
                    PickleValue::String("standby_count".into()),
                    PickleValue::Int(pool.standby_count as i64),
                ),
                (
                    PickleValue::String("cooldown_count".into()),
                    PickleValue::Int(pool.cooldown_count as i64),
                ),
                (
                    PickleValue::String("members".into()),
                    PickleValue::List(members),
                ),
            ]),
        ));
    } else {
        dict.push((
            PickleValue::String("backbone_peer_pool".into()),
            PickleValue::None,
        ));
    }

    PickleValue::Dict(dict)
}

fn add_inbound_runtime_stats(
    response: &mut PickleValue,
    stats: &InterfaceStatsResponse,
    event_tx: &EventSender,
) {
    let PickleValue::Dict(dict) = response else {
        return;
    };
    if let Some((_, PickleValue::List(ifaces))) = dict
        .iter_mut()
        .find(|(key, _)| key.as_str() == Some("interfaces"))
    {
        for (iface, stat) in ifaces.iter_mut().zip(&stats.interfaces) {
            let PickleValue::Dict(iface_dict) = iface else {
                continue;
            };
            let burst_stats = if stat.interface_type == "BackboneInterface" {
                event_tx
                    .dynamic_burst_stats(InterfaceId(stat.id))
                    .or(Some(DynamicBurstStats {
                        announce: BurstClassStats {
                            count: 0,
                            activated: None,
                        },
                        path_request: BurstClassStats {
                            count: 0,
                            activated: None,
                        },
                    }))
            } else {
                None
            };
            iface_dict.push((
                PickleValue::String("burst_count".into()),
                burst_stats
                    .map(|stats| PickleValue::Int(stats.announce.count as i64))
                    .unwrap_or(PickleValue::None),
            ));
            iface_dict.push((
                PickleValue::String("pr_burst_count".into()),
                burst_stats
                    .map(|stats| PickleValue::Int(stats.path_request.count as i64))
                    .unwrap_or(PickleValue::None),
            ));
            if let Some(stats) = burst_stats {
                set_pickle_dict_entry(
                    iface_dict,
                    "burst_active",
                    PickleValue::Bool(stats.announce.count > 0),
                );
                set_pickle_dict_entry(
                    iface_dict,
                    "burst_activated",
                    stats
                        .announce
                        .activated
                        .map(PickleValue::Float)
                        .unwrap_or(PickleValue::None),
                );
                set_pickle_dict_entry(
                    iface_dict,
                    "pr_burst_active",
                    PickleValue::Bool(stats.path_request.count > 0),
                );
                set_pickle_dict_entry(
                    iface_dict,
                    "pr_burst_activated",
                    stats
                        .path_request
                        .activated
                        .map(PickleValue::Float)
                        .unwrap_or(PickleValue::None),
                );
            }
        }
    }

    let snapshot = event_tx.inbound_queue_snapshot();
    let total_height = snapshot.total_height();
    let total_capacity = snapshot.total_capacity();
    let total_dropped = snapshot.total_dropped();
    let pressure = |height: usize, capacity: usize| {
        if height == 0 {
            0.0
        } else {
            height as f64 / capacity as f64
        }
    };
    dict.extend([
        (
            PickleValue::String("rxqt".into()),
            PickleValue::Int(total_height as i64),
        ),
        (
            PickleValue::String("rxqd".into()),
            PickleValue::Int(snapshot.heights[0] as i64),
        ),
        (
            PickleValue::String("rxqa".into()),
            PickleValue::Int(snapshot.heights[1] as i64),
        ),
        (
            PickleValue::String("rxqp".into()),
            PickleValue::Int(snapshot.heights[2] as i64),
        ),
        (
            PickleValue::String("rxqil".into()),
            PickleValue::Int(snapshot.heights[3] as i64),
        ),
        (
            PickleValue::String("rxqtd".into()),
            PickleValue::Int(total_dropped as i64),
        ),
        (
            PickleValue::String("rxqdd".into()),
            PickleValue::Int(snapshot.dropped[0] as i64),
        ),
        (
            PickleValue::String("rxqad".into()),
            PickleValue::Int(snapshot.dropped[1] as i64),
        ),
        (
            PickleValue::String("rxqpd".into()),
            PickleValue::Int(snapshot.dropped[2] as i64),
        ),
        (
            PickleValue::String("rxqild".into()),
            PickleValue::Int(snapshot.dropped[3] as i64),
        ),
        (
            PickleValue::String("tqpressure".into()),
            PickleValue::Float(pressure(total_height, total_capacity)),
        ),
        (
            PickleValue::String("dqpressure".into()),
            PickleValue::Float(pressure(snapshot.heights[0], snapshot.capacities[0])),
        ),
        (
            PickleValue::String("aqpressure".into()),
            PickleValue::Float(pressure(snapshot.heights[1], snapshot.capacities[1])),
        ),
        (
            PickleValue::String("pqpressure".into()),
            PickleValue::Float(pressure(snapshot.heights[2], snapshot.capacities[2])),
        ),
        (
            PickleValue::String("ilqpressure".into()),
            PickleValue::Float(pressure(snapshot.heights[3], snapshot.capacities[3])),
        ),
        (PickleValue::String("txq".into()), PickleValue::None),
    ]);
}

fn set_pickle_dict_entry(
    dict: &mut Vec<(PickleValue, PickleValue)>,
    key: &str,
    value: PickleValue,
) {
    if let Some((_, existing)) = dict
        .iter_mut()
        .find(|(candidate, _)| candidate.as_str() == Some(key))
    {
        *existing = value;
    } else {
        dict.push((PickleValue::String(key.into()), value));
    }
}

fn single_iface_to_pickle(s: &SingleInterfaceStat) -> PickleValue {
    let mut dict = vec![
        (
            PickleValue::String("id".into()),
            PickleValue::Int(s.id as i64),
        ),
        (
            PickleValue::String("name".into()),
            PickleValue::String(s.name.clone()),
        ),
        (
            PickleValue::String("status".into()),
            PickleValue::Bool(s.status),
        ),
        (
            PickleValue::String("mode".into()),
            PickleValue::Int(s.mode as i64),
        ),
        (
            PickleValue::String("gravity".into()),
            PickleValue::Int(s.gravity),
        ),
        (
            PickleValue::String("announces_to_internal".into()),
            s.announces_to_internal
                .map(PickleValue::Bool)
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("rxb".into()),
            PickleValue::Int(s.rxb as i64),
        ),
        (
            PickleValue::String("txb".into()),
            PickleValue::Int(s.txb as i64),
        ),
        (
            PickleValue::String("txdrp".into()),
            PickleValue::Int(s.tx_drops as i64),
        ),
        (
            PickleValue::String("txdrb".into()),
            PickleValue::Int(s.tx_dropped_bytes as i64),
        ),
        (
            PickleValue::String("txstalled".into()),
            PickleValue::Bool(s.tx_stalled),
        ),
        (
            PickleValue::String("txbuffered".into()),
            PickleValue::Int(s.tx_buffered as i64),
        ),
        (
            PickleValue::String("rxs".into()),
            PickleValue::Float(s.traffic.rxs),
        ),
        (
            PickleValue::String("txs".into()),
            PickleValue::Float(s.traffic.txs),
        ),
        (
            PickleValue::String("arxb".into()),
            PickleValue::Int(s.traffic.arxb as i64),
        ),
        (
            PickleValue::String("atxb".into()),
            PickleValue::Int(s.traffic.atxb as i64),
        ),
        (
            PickleValue::String("arxc".into()),
            PickleValue::Int(s.traffic.arxc as i64),
        ),
        (
            PickleValue::String("atxc".into()),
            PickleValue::Int(s.traffic.atxc as i64),
        ),
        (
            PickleValue::String("arxs".into()),
            PickleValue::Float(s.traffic.arxs),
        ),
        (
            PickleValue::String("atxs".into()),
            PickleValue::Float(s.traffic.atxs),
        ),
        (
            PickleValue::String("prxb".into()),
            PickleValue::Int(s.traffic.prxb as i64),
        ),
        (
            PickleValue::String("ptxb".into()),
            PickleValue::Int(s.traffic.ptxb as i64),
        ),
        (
            PickleValue::String("prxc".into()),
            PickleValue::Int(s.traffic.prxc as i64),
        ),
        (
            PickleValue::String("ptxc".into()),
            PickleValue::Int(s.traffic.ptxc as i64),
        ),
        (
            PickleValue::String("prxs".into()),
            PickleValue::Float(s.traffic.prxs),
        ),
        (
            PickleValue::String("ptxs".into()),
            PickleValue::Float(s.traffic.ptxs),
        ),
        (
            PickleValue::String("protocol_violations".into()),
            PickleValue::Int(s.protocol_violations as i64),
        ),
        (
            PickleValue::String("ifac_violations".into()),
            PickleValue::Int(s.ifac_violations as i64),
        ),
        (
            PickleValue::String("packet_filter_hits".into()),
            PickleValue::Int(s.packet_filter_hits as i64),
        ),
        (
            PickleValue::String("rx_packets".into()),
            PickleValue::Int(s.rx_packets as i64),
        ),
        (
            PickleValue::String("tx_packets".into()),
            PickleValue::Int(s.tx_packets as i64),
        ),
        (
            PickleValue::String("cpu_load".into()),
            s.cpu_load
                .map(PickleValue::Float)
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("mem_load".into()),
            s.mem_load
                .map(PickleValue::Float)
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("switch_id".into()),
            s.switch_id
                .map(|v| PickleValue::Bytes(v.to_vec()))
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("endpoint_id".into()),
            s.endpoint_id
                .map(|v| PickleValue::Bytes(v.to_vec()))
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("via_switch_id".into()),
            s.via_switch_id
                .map(|v| PickleValue::Bytes(v.to_vec()))
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("peers".into()),
            s.peers
                .map(|v| PickleValue::Int(v as i64))
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("started".into()),
            PickleValue::Float(s.started),
        ),
        (
            PickleValue::String("ia_freq".into()),
            PickleValue::Float(s.ia_freq),
        ),
        (
            PickleValue::String("oa_freq".into()),
            PickleValue::Float(s.oa_freq),
        ),
        (
            PickleValue::String("ip_freq".into()),
            PickleValue::Float(s.ip_freq),
        ),
        (
            PickleValue::String("op_freq".into()),
            PickleValue::Float(s.op_freq),
        ),
        (
            PickleValue::String("burst_active".into()),
            PickleValue::Bool(s.burst_active),
        ),
        (
            PickleValue::String("burst_activated".into()),
            PickleValue::Float(s.burst_activated),
        ),
        (
            PickleValue::String("pr_burst_active".into()),
            PickleValue::Bool(s.pr_burst_active),
        ),
        (
            PickleValue::String("pr_burst_activated".into()),
            PickleValue::Float(s.pr_burst_activated),
        ),
        (
            PickleValue::String("clients".into()),
            s.clients
                .map(|clients| PickleValue::Int(clients as i64))
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("blocked_ips".into()),
            s.blocked_ips
                .map(|blocked| PickleValue::Int(blocked as i64))
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("blocked_ip_list".into()),
            s.blocked_ip_list
                .as_ref()
                .map(|blocked| {
                    PickleValue::List(blocked.iter().cloned().map(PickleValue::String).collect())
                })
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("announce_rate_grace".into()),
            PickleValue::Int(s.announce_rate_grace as i64),
        ),
        (
            PickleValue::String("announce_rate_penalty".into()),
            PickleValue::Float(s.announce_rate_penalty),
        ),
    ];

    match s.announce_rate_target {
        Some(target) => dict.push((
            PickleValue::String("announce_rate_target".into()),
            PickleValue::Float(target),
        )),
        None => dict.push((
            PickleValue::String("announce_rate_target".into()),
            PickleValue::None,
        )),
    }

    match s.bitrate {
        Some(br) => dict.push((
            PickleValue::String("bitrate".into()),
            PickleValue::Int(br as i64),
        )),
        None => dict.push((PickleValue::String("bitrate".into()), PickleValue::None)),
    }

    match s.mtu {
        Some(mtu) => dict.push((
            PickleValue::String("mtu".into()),
            PickleValue::Int(mtu as i64),
        )),
        None => dict.push((PickleValue::String("mtu".into()), PickleValue::None)),
    }

    match s.ifac_size {
        Some(sz) => dict.push((
            PickleValue::String("ifac_size".into()),
            PickleValue::Int(sz as i64),
        )),
        None => dict.push((PickleValue::String("ifac_size".into()), PickleValue::None)),
    }

    PickleValue::Dict(dict)
}

fn path_table_to_pickle(entries: &[PathTableEntry]) -> PickleValue {
    let list: Vec<PickleValue> = entries
        .iter()
        .map(|e| {
            PickleValue::Dict(vec![
                (
                    PickleValue::String("hash".into()),
                    PickleValue::Bytes(e.hash.to_vec()),
                ),
                (
                    PickleValue::String("timestamp".into()),
                    PickleValue::Float(e.timestamp),
                ),
                (
                    PickleValue::String("via".into()),
                    PickleValue::Bytes(e.via.to_vec()),
                ),
                (
                    PickleValue::String("hops".into()),
                    PickleValue::Int(e.hops as i64),
                ),
                (
                    PickleValue::String("expires".into()),
                    PickleValue::Float(e.expires),
                ),
                (
                    PickleValue::String("interface".into()),
                    PickleValue::String(e.interface_name.clone()),
                ),
            ])
        })
        .collect();
    PickleValue::List(list)
}

fn rate_table_to_pickle(entries: &[RateTableEntry]) -> PickleValue {
    let list: Vec<PickleValue> = entries
        .iter()
        .map(|e| {
            PickleValue::Dict(vec![
                (
                    PickleValue::String("hash".into()),
                    PickleValue::Bytes(e.hash.to_vec()),
                ),
                (
                    PickleValue::String("last".into()),
                    PickleValue::Float(e.last),
                ),
                (
                    PickleValue::String("rate_violations".into()),
                    PickleValue::Int(e.rate_violations as i64),
                ),
                (
                    PickleValue::String("blocked_until".into()),
                    PickleValue::Float(e.blocked_until),
                ),
                (
                    PickleValue::String("timestamps".into()),
                    PickleValue::List(
                        e.timestamps
                            .iter()
                            .map(|&t| PickleValue::Float(t))
                            .collect(),
                    ),
                ),
            ])
        })
        .collect();
    PickleValue::List(list)
}

fn blackholed_to_pickle(entries: &[BlackholeInfo]) -> PickleValue {
    let list: Vec<PickleValue> = entries
        .iter()
        .map(|e| {
            let mut dict = vec![
                (
                    PickleValue::String("identity_hash".into()),
                    PickleValue::Bytes(e.identity_hash.to_vec()),
                ),
                (
                    PickleValue::String("created".into()),
                    PickleValue::Float(e.created),
                ),
                (
                    PickleValue::String("expires".into()),
                    PickleValue::Float(e.expires),
                ),
            ];
            if let Some(ref reason) = e.reason {
                dict.push((
                    PickleValue::String("reason".into()),
                    PickleValue::String(reason.clone()),
                ));
            } else {
                dict.push((PickleValue::String("reason".into()), PickleValue::None));
            }
            PickleValue::Dict(dict)
        })
        .collect();
    PickleValue::List(list)
}

fn discovered_interfaces_to_pickle(
    interfaces: &[crate::discovery::DiscoveredInterface],
) -> PickleValue {
    let list: Vec<PickleValue> = interfaces
        .iter()
        .map(|iface| {
            let mut dict = vec![
                (
                    PickleValue::String("type".into()),
                    PickleValue::String(iface.interface_type.clone()),
                ),
                (
                    PickleValue::String("impl_name".into()),
                    iface
                        .impl_name
                        .clone()
                        .map(PickleValue::String)
                        .unwrap_or(PickleValue::None),
                ),
                (
                    PickleValue::String("version".into()),
                    iface
                        .impl_version
                        .clone()
                        .map(PickleValue::String)
                        .unwrap_or(PickleValue::None),
                ),
                (
                    PickleValue::String("transport".into()),
                    PickleValue::Bool(iface.transport),
                ),
                (
                    PickleValue::String("name".into()),
                    PickleValue::String(iface.name.clone()),
                ),
                (
                    PickleValue::String("discovered".into()),
                    PickleValue::Float(iface.discovered),
                ),
                (
                    PickleValue::String("last_heard".into()),
                    PickleValue::Float(iface.last_heard),
                ),
                (
                    PickleValue::String("heard_count".into()),
                    PickleValue::Int(iface.heard_count as i64),
                ),
                (
                    PickleValue::String("status".into()),
                    PickleValue::String(iface.status.as_str().into()),
                ),
                (
                    PickleValue::String("stamp".into()),
                    PickleValue::Bytes(iface.stamp.clone()),
                ),
                (
                    PickleValue::String("value".into()),
                    PickleValue::Int(iface.stamp_value as i64),
                ),
                (
                    PickleValue::String("transport_id".into()),
                    PickleValue::Bytes(iface.transport_id.to_vec()),
                ),
                (
                    PickleValue::String("network_id".into()),
                    PickleValue::Bytes(iface.network_id.to_vec()),
                ),
                (
                    PickleValue::String("hops".into()),
                    PickleValue::Int(iface.hops as i64),
                ),
            ];

            // Optional location fields
            if let Some(v) = iface.latitude {
                dict.push((
                    PickleValue::String("latitude".into()),
                    PickleValue::Float(v),
                ));
            } else {
                dict.push((PickleValue::String("latitude".into()), PickleValue::None));
            }
            if let Some(v) = iface.longitude {
                dict.push((
                    PickleValue::String("longitude".into()),
                    PickleValue::Float(v),
                ));
            } else {
                dict.push((PickleValue::String("longitude".into()), PickleValue::None));
            }
            if let Some(v) = iface.height {
                dict.push((PickleValue::String("height".into()), PickleValue::Float(v)));
            } else {
                dict.push((PickleValue::String("height".into()), PickleValue::None));
            }
            if let Some(v) = iface.operator_lxmf_address {
                dict.push((
                    PickleValue::String("operator_lxmf_address".into()),
                    PickleValue::String(crate::discovery::hex_encode(&v)),
                ));
            }

            // Connection info
            if let Some(ref v) = iface.reachable_on {
                dict.push((
                    PickleValue::String("reachable_on".into()),
                    PickleValue::String(v.clone()),
                ));
            } else {
                dict.push((
                    PickleValue::String("reachable_on".into()),
                    PickleValue::None,
                ));
            }
            if let Some(v) = iface.port {
                dict.push((
                    PickleValue::String("port".into()),
                    PickleValue::Int(v as i64),
                ));
            } else {
                dict.push((PickleValue::String("port".into()), PickleValue::None));
            }

            // RNode/RF specific
            if let Some(v) = iface.frequency {
                dict.push((
                    PickleValue::String("frequency".into()),
                    PickleValue::Int(v as i64),
                ));
            } else {
                dict.push((PickleValue::String("frequency".into()), PickleValue::None));
            }
            if let Some(v) = iface.bandwidth {
                dict.push((
                    PickleValue::String("bandwidth".into()),
                    PickleValue::Int(v as i64),
                ));
            } else {
                dict.push((PickleValue::String("bandwidth".into()), PickleValue::None));
            }
            if let Some(v) = iface.spreading_factor {
                dict.push((PickleValue::String("sf".into()), PickleValue::Int(v as i64)));
            } else {
                dict.push((PickleValue::String("sf".into()), PickleValue::None));
            }
            if let Some(v) = iface.coding_rate {
                dict.push((PickleValue::String("cr".into()), PickleValue::Int(v as i64)));
            } else {
                dict.push((PickleValue::String("cr".into()), PickleValue::None));
            }
            if let Some(ref v) = iface.modulation {
                dict.push((
                    PickleValue::String("modulation".into()),
                    PickleValue::String(v.clone()),
                ));
            } else {
                dict.push((PickleValue::String("modulation".into()), PickleValue::None));
            }
            if let Some(v) = iface.channel {
                dict.push((
                    PickleValue::String("channel".into()),
                    PickleValue::Int(v as i64),
                ));
            } else {
                dict.push((PickleValue::String("channel".into()), PickleValue::None));
            }

            // IFAC info
            if let Some(ref v) = iface.ifac_netname {
                dict.push((
                    PickleValue::String("ifac_netname".into()),
                    PickleValue::String(v.clone()),
                ));
            } else {
                dict.push((
                    PickleValue::String("ifac_netname".into()),
                    PickleValue::None,
                ));
            }
            if let Some(ref v) = iface.ifac_netkey {
                dict.push((
                    PickleValue::String("ifac_netkey".into()),
                    PickleValue::String(v.clone()),
                ));
            } else {
                dict.push((PickleValue::String("ifac_netkey".into()), PickleValue::None));
            }

            // Config entry
            if let Some(ref v) = iface.config_entry {
                dict.push((
                    PickleValue::String("config_entry".into()),
                    PickleValue::String(v.clone()),
                ));
            } else {
                dict.push((
                    PickleValue::String("config_entry".into()),
                    PickleValue::None,
                ));
            }

            dict.push((
                PickleValue::String("discovery_hash".into()),
                PickleValue::Bytes(iface.discovery_hash.to_vec()),
            ));

            PickleValue::Dict(dict)
        })
        .collect();
    PickleValue::List(list)
}

fn hooks_to_pickle(hooks: &[HookInfo]) -> PickleValue {
    PickleValue::List(
        hooks
            .iter()
            .map(|hook| {
                PickleValue::Dict(vec![
                    (
                        PickleValue::String("name".into()),
                        PickleValue::String(hook.name.clone()),
                    ),
                    (
                        PickleValue::String("type".into()),
                        PickleValue::String(hook.hook_type.clone()),
                    ),
                    (
                        PickleValue::String("attach_point".into()),
                        PickleValue::String(hook.attach_point.clone()),
                    ),
                    (
                        PickleValue::String("priority".into()),
                        PickleValue::Int(hook.priority as i64),
                    ),
                    (
                        PickleValue::String("enabled".into()),
                        PickleValue::Bool(hook.enabled),
                    ),
                    (
                        PickleValue::String("consecutive_traps".into()),
                        PickleValue::Int(hook.consecutive_traps as i64),
                    ),
                ])
            })
            .collect(),
    )
}

fn backbone_peer_state_to_pickle(entries: &[BackbonePeerStateEntry]) -> PickleValue {
    PickleValue::List(
        entries
            .iter()
            .map(|entry| {
                PickleValue::Dict(vec![
                    (
                        PickleValue::String("interface".into()),
                        PickleValue::String(entry.interface_name.clone()),
                    ),
                    (
                        PickleValue::String("ip".into()),
                        PickleValue::String(entry.peer_ip.to_string()),
                    ),
                    (
                        PickleValue::String("connected_count".into()),
                        PickleValue::Int(entry.connected_count as i64),
                    ),
                    (
                        PickleValue::String("blacklisted_remaining_secs".into()),
                        entry
                            .blacklisted_remaining_secs
                            .map(PickleValue::Float)
                            .unwrap_or(PickleValue::None),
                    ),
                    (
                        PickleValue::String("blacklist_reason".into()),
                        entry
                            .blacklist_reason
                            .as_ref()
                            .map(|v: &String| PickleValue::String(v.clone()))
                            .unwrap_or(PickleValue::None),
                    ),
                    (
                        PickleValue::String("reject_count".into()),
                        PickleValue::Int(entry.reject_count as i64),
                    ),
                ])
            })
            .collect(),
    )
}

fn backbone_interfaces_to_pickle(entries: &[BackboneInterfaceEntry]) -> PickleValue {
    PickleValue::List(
        entries
            .iter()
            .map(|entry| {
                PickleValue::Dict(vec![
                    (
                        PickleValue::String("id".into()),
                        PickleValue::Int(entry.interface_id.0 as i64),
                    ),
                    (
                        PickleValue::String("name".into()),
                        PickleValue::String(entry.interface_name.clone()),
                    ),
                ])
            })
            .collect(),
    )
}

fn provider_bridge_stats_to_pickle(stats: &ProviderBridgeStats) -> PickleValue {
    PickleValue::Dict(vec![
        (
            PickleValue::String("connected".into()),
            PickleValue::Bool(stats.connected),
        ),
        (
            PickleValue::String("consumer_count".into()),
            PickleValue::Int(stats.consumer_count as i64),
        ),
        (
            PickleValue::String("queue_max_events".into()),
            PickleValue::Int(stats.queue_max_events as i64),
        ),
        (
            PickleValue::String("queue_max_bytes".into()),
            PickleValue::Int(stats.queue_max_bytes as i64),
        ),
        (
            PickleValue::String("backlog_len".into()),
            PickleValue::Int(stats.backlog_len as i64),
        ),
        (
            PickleValue::String("backlog_bytes".into()),
            PickleValue::Int(stats.backlog_bytes as i64),
        ),
        (
            PickleValue::String("backlog_dropped_pending".into()),
            PickleValue::Int(stats.backlog_dropped_pending as i64),
        ),
        (
            PickleValue::String("backlog_dropped_total".into()),
            PickleValue::Int(stats.backlog_dropped_total as i64),
        ),
        (
            PickleValue::String("total_disconnect_count".into()),
            PickleValue::Int(stats.total_disconnect_count as i64),
        ),
        (
            PickleValue::String("consumers".into()),
            PickleValue::List(
                stats
                    .consumers
                    .iter()
                    .map(|consumer| {
                        PickleValue::Dict(vec![
                            (
                                PickleValue::String("id".into()),
                                PickleValue::Int(consumer.id as i64),
                            ),
                            (
                                PickleValue::String("connected".into()),
                                PickleValue::Bool(consumer.connected),
                            ),
                            (
                                PickleValue::String("queue_len".into()),
                                PickleValue::Int(consumer.queue_len as i64),
                            ),
                            (
                                PickleValue::String("queued_bytes".into()),
                                PickleValue::Int(consumer.queued_bytes as i64),
                            ),
                            (
                                PickleValue::String("dropped_pending".into()),
                                PickleValue::Int(consumer.dropped_pending as i64),
                            ),
                            (
                                PickleValue::String("dropped_total".into()),
                                PickleValue::Int(consumer.dropped_total as i64),
                            ),
                            (
                                PickleValue::String("queue_max_events".into()),
                                PickleValue::Int(consumer.queue_max_events as i64),
                            ),
                            (
                                PickleValue::String("queue_max_bytes".into()),
                                PickleValue::Int(consumer.queue_max_bytes as i64),
                            ),
                        ])
                    })
                    .collect(),
            ),
        ),
    ])
}

fn lifecycle_state_name(state: LifecycleState) -> &'static str {
    match state {
        LifecycleState::Active => "active",
        LifecycleState::Draining => "draining",
        LifecycleState::Stopping => "stopping",
        LifecycleState::Stopped => "stopped",
    }
}

fn drain_status_to_pickle(status: &DrainStatus) -> PickleValue {
    PickleValue::Dict(vec![
        (
            PickleValue::String("state".into()),
            PickleValue::String(lifecycle_state_name(status.state).into()),
        ),
        (
            PickleValue::String("drain_age_seconds".into()),
            status
                .drain_age_seconds
                .map(PickleValue::Float)
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("deadline_remaining_seconds".into()),
            status
                .deadline_remaining_seconds
                .map(PickleValue::Float)
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("drain_complete".into()),
            PickleValue::Bool(status.drain_complete),
        ),
        (
            PickleValue::String("interface_writer_queued_frames".into()),
            PickleValue::Int(status.interface_writer_queued_frames as i64),
        ),
        (
            PickleValue::String("provider_backlog_events".into()),
            PickleValue::Int(status.provider_backlog_events as i64),
        ),
        (
            PickleValue::String("provider_consumer_queued_events".into()),
            PickleValue::Int(status.provider_consumer_queued_events as i64),
        ),
        (
            PickleValue::String("detail".into()),
            status
                .detail
                .as_ref()
                .map(|detail| PickleValue::String(detail.clone()))
                .unwrap_or(PickleValue::None),
        ),
    ])
}

fn runtime_config_value_to_pickle(value: &RuntimeConfigValue) -> PickleValue {
    match value {
        RuntimeConfigValue::Int(v) => PickleValue::Int(*v),
        RuntimeConfigValue::Float(v) => PickleValue::Float(*v),
        RuntimeConfigValue::Bool(v) => PickleValue::Bool(*v),
        RuntimeConfigValue::String(v) => PickleValue::String(v.clone()),
        RuntimeConfigValue::Null => PickleValue::None,
    }
}

fn runtime_config_value_from_pickle(value: &PickleValue) -> Option<RuntimeConfigValue> {
    match value {
        PickleValue::Int(v) => Some(RuntimeConfigValue::Int(*v)),
        PickleValue::Float(v) => Some(RuntimeConfigValue::Float(*v)),
        PickleValue::Bool(v) => Some(RuntimeConfigValue::Bool(*v)),
        PickleValue::String(v) => Some(RuntimeConfigValue::String(v.clone())),
        PickleValue::None => Some(RuntimeConfigValue::Null),
        _ => None,
    }
}

fn runtime_config_entry_to_pickle(entry: &RuntimeConfigEntry) -> PickleValue {
    PickleValue::Dict(vec![
        (
            PickleValue::String("key".into()),
            PickleValue::String(entry.key.clone()),
        ),
        (
            PickleValue::String("value".into()),
            runtime_config_value_to_pickle(&entry.value),
        ),
        (
            PickleValue::String("default".into()),
            runtime_config_value_to_pickle(&entry.default),
        ),
        (
            PickleValue::String("source".into()),
            PickleValue::String(match entry.source {
                RuntimeConfigSource::Startup => "startup".into(),
                RuntimeConfigSource::RuntimeOverride => "runtime_override".into(),
            }),
        ),
        (
            PickleValue::String("apply_mode".into()),
            PickleValue::String(match entry.apply_mode {
                RuntimeConfigApplyMode::Immediate => "immediate".into(),
                RuntimeConfigApplyMode::NewConnectionsOnly => "new_connections_only".into(),
                RuntimeConfigApplyMode::NextReconnect => "next_reconnect".into(),
                RuntimeConfigApplyMode::RestartRequired => "restart_required".into(),
            }),
        ),
        (
            PickleValue::String("description".into()),
            entry
                .description
                .as_ref()
                .map(|v| PickleValue::String(v.clone()))
                .unwrap_or(PickleValue::None),
        ),
    ])
}

fn runtime_config_list_to_pickle(entries: &[RuntimeConfigEntry]) -> PickleValue {
    PickleValue::List(entries.iter().map(runtime_config_entry_to_pickle).collect())
}

fn runtime_config_error_to_pickle(error: &RuntimeConfigError) -> PickleValue {
    PickleValue::Dict(vec![
        (
            PickleValue::String("error".into()),
            PickleValue::String(match error.code {
                RuntimeConfigErrorCode::UnknownKey => "unknown_key".into(),
                RuntimeConfigErrorCode::InvalidType => "invalid_type".into(),
                RuntimeConfigErrorCode::InvalidValue => "invalid_value".into(),
                RuntimeConfigErrorCode::Unsupported => "unsupported".into(),
                RuntimeConfigErrorCode::NotFound => "not_found".into(),
                RuntimeConfigErrorCode::ApplyFailed => "apply_failed".into(),
            }),
        ),
        (
            PickleValue::String("message".into()),
            PickleValue::String(error.message.clone()),
        ),
    ])
}

fn runtime_config_result_to_pickle(
    result: Result<RuntimeConfigEntry, RuntimeConfigError>,
) -> PickleValue {
    match result {
        Ok(entry) => runtime_config_entry_to_pickle(&entry),
        Err(error) => runtime_config_error_to_pickle(&error),
    }
}

fn known_destination_entry_to_pickle(entry: &KnownDestinationEntry) -> PickleValue {
    PickleValue::Dict(vec![
        (
            PickleValue::String("dest_hash".into()),
            PickleValue::Bytes(entry.dest_hash.to_vec()),
        ),
        (
            PickleValue::String("identity_hash".into()),
            PickleValue::Bytes(entry.identity_hash.to_vec()),
        ),
        (
            PickleValue::String("public_key".into()),
            PickleValue::Bytes(entry.public_key.to_vec()),
        ),
        (
            PickleValue::String("app_data".into()),
            entry
                .app_data
                .as_ref()
                .map(|data: &Vec<u8>| PickleValue::Bytes(data.clone()))
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("hops".into()),
            PickleValue::Int(entry.hops as i64),
        ),
        (
            PickleValue::String("received_at".into()),
            PickleValue::Float(entry.received_at),
        ),
        (
            PickleValue::String("receiving_interface".into()),
            PickleValue::Int(entry.receiving_interface.0 as i64),
        ),
        (
            PickleValue::String("was_used".into()),
            PickleValue::Bool(entry.was_used),
        ),
        (
            PickleValue::String("last_used_at".into()),
            entry
                .last_used_at
                .map(PickleValue::Float)
                .unwrap_or(PickleValue::None),
        ),
        (
            PickleValue::String("retained".into()),
            PickleValue::Bool(entry.retained),
        ),
    ])
}

fn known_destinations_to_pickle(entries: &[KnownDestinationEntry]) -> PickleValue {
    PickleValue::List(
        entries
            .iter()
            .map(known_destination_entry_to_pickle)
            .collect(),
    )
}

fn parse_known_destination_entry(value: &PickleValue) -> io::Result<KnownDestinationEntry> {
    let get_bytes = |key: &str, len: usize| -> io::Result<Vec<u8>> {
        let value = value.get(key).ok_or_else(|| {
            io::Error::new(io::ErrorKind::InvalidData, format!("missing {}", key))
        })?;
        let bytes = value.as_bytes().ok_or_else(|| {
            io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key))
        })?;
        if bytes.len() != len {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                format!("invalid {} length", key),
            ));
        }
        Ok(bytes.to_vec())
    };
    let get_int = |key: &str| -> io::Result<i64> {
        value
            .get(key)
            .and_then(|v| v.as_int())
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key)))
    };
    let get_float = |key: &str| -> io::Result<f64> {
        value
            .get(key)
            .and_then(|v| v.as_float())
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key)))
    };
    let get_bool = |key: &str| -> io::Result<bool> {
        value
            .get(key)
            .and_then(|v| v.as_bool())
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, format!("invalid {}", key)))
    };

    let mut dest_hash = [0u8; 16];
    dest_hash.copy_from_slice(&get_bytes("dest_hash", 16)?);
    let mut identity_hash = [0u8; 16];
    identity_hash.copy_from_slice(&get_bytes("identity_hash", 16)?);
    let mut public_key = [0u8; 64];
    public_key.copy_from_slice(&get_bytes("public_key", 64)?);
    let app_data = value
        .get("app_data")
        .and_then(|v| v.as_bytes())
        .map(|bytes| bytes.to_vec());
    let last_used_at = value.get("last_used_at").and_then(|v| v.as_float());

    Ok(KnownDestinationEntry {
        dest_hash,
        identity_hash,
        public_key,
        app_data,
        hops: get_int("hops")? as u8,
        received_at: get_float("received_at")?,
        receiving_interface: rns_core::transport::types::InterfaceId(
            get_int("receiving_interface")? as u64,
        ),
        was_used: get_bool("was_used")?,
        last_used_at,
        retained: get_bool("retained")?,
    })
}

fn parse_known_destination_list(value: &PickleValue) -> io::Result<Vec<KnownDestinationEntry>> {
    let list = value
        .as_list()
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "expected list"))?;
    list.iter().map(parse_known_destination_entry).collect()
}

// --- RPC Client ---

/// RPC client for connecting to a running daemon.
pub struct RpcClient {
    stream: TcpStream,
}

impl RpcClient {
    /// Connect to an RPC server and authenticate.
    pub fn connect(addr: &RpcAddr, auth_key: &[u8; 32]) -> io::Result<Self> {
        let mut stream = match addr {
            RpcAddr::Tcp(host, port) => TcpStream::connect((host.as_str(), *port))?,
        };

        stream.set_read_timeout(Some(std::time::Duration::from_secs(10)))?;
        stream.set_write_timeout(Some(std::time::Duration::from_secs(10)))?;

        // Client-side authentication
        client_auth(&mut stream, auth_key)?;

        Ok(RpcClient { stream })
    }

    /// Send an RPC request and receive a response.
    pub fn call(&mut self, request: &PickleValue) -> io::Result<PickleValue> {
        let request_bytes = encode_rpc_payload(request, RpcCodec::Msgpack);
        send_bytes(&mut self.stream, &request_bytes)?;

        let response_bytes = recv_bytes(&mut self.stream)?;
        decode_rpc_payload(&response_bytes).map(|(value, _)| value)
    }

    pub fn list_hooks(&mut self) -> io::Result<Vec<HookInfo>> {
        let response = self.call(&PickleValue::Dict(vec![(
            PickleValue::String("get".into()),
            PickleValue::String("hooks".into()),
        )]))?;
        parse_hook_list(&response)
    }

    pub fn begin_drain(&mut self, timeout: Duration) -> io::Result<bool> {
        let response = self.call(&PickleValue::Dict(vec![(
            PickleValue::String("begin_drain".into()),
            PickleValue::Float(timeout.as_secs_f64()),
        )]))?;
        Ok(response.as_bool().unwrap_or(false))
    }

    pub fn drain_status(&mut self) -> io::Result<Option<DrainStatus>> {
        let response = self.call(&PickleValue::Dict(vec![(
            PickleValue::String("get".into()),
            PickleValue::String("drain_status".into()),
        )]))?;
        parse_drain_status(&response)
    }

    pub fn provider_bridge_stats(&mut self) -> io::Result<PickleValue> {
        self.call(&PickleValue::Dict(vec![(
            PickleValue::String("get".into()),
            PickleValue::String("provider_bridge_stats".into()),
        )]))
    }

    pub fn load_hook(
        &mut self,
        name: &str,
        attach_point: &str,
        priority: i32,
        wasm: &[u8],
    ) -> io::Result<Result<(), String>> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("hook".into()),
                PickleValue::String("load".into()),
            ),
            (
                PickleValue::String("name".into()),
                PickleValue::String(name.to_string()),
            ),
            (
                PickleValue::String("attach_point".into()),
                PickleValue::String(attach_point.to_string()),
            ),
            (
                PickleValue::String("priority".into()),
                PickleValue::Int(priority as i64),
            ),
            (
                PickleValue::String("wasm".into()),
                PickleValue::Bytes(wasm.to_vec()),
            ),
        ]))?;
        parse_hook_result(&response)
    }

    pub fn load_builtin_hook(
        &mut self,
        name: &str,
        attach_point: &str,
        priority: i32,
        builtin_id: &str,
    ) -> io::Result<Result<(), String>> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("hook".into()),
                PickleValue::String("load_builtin".into()),
            ),
            (
                PickleValue::String("name".into()),
                PickleValue::String(name.to_string()),
            ),
            (
                PickleValue::String("attach_point".into()),
                PickleValue::String(attach_point.to_string()),
            ),
            (
                PickleValue::String("priority".into()),
                PickleValue::Int(priority as i64),
            ),
            (
                PickleValue::String("builtin_id".into()),
                PickleValue::String(builtin_id.to_string()),
            ),
        ]))?;
        parse_hook_result(&response)
    }

    pub fn unload_hook(
        &mut self,
        name: &str,
        attach_point: &str,
    ) -> io::Result<Result<(), String>> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("hook".into()),
                PickleValue::String("unload".into()),
            ),
            (
                PickleValue::String("name".into()),
                PickleValue::String(name.to_string()),
            ),
            (
                PickleValue::String("attach_point".into()),
                PickleValue::String(attach_point.to_string()),
            ),
        ]))?;
        parse_hook_result(&response)
    }

    pub fn set_hook_enabled(
        &mut self,
        name: &str,
        attach_point: &str,
        enabled: bool,
    ) -> io::Result<Result<(), String>> {
        let op = if enabled { "enable" } else { "disable" };
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("hook".into()),
                PickleValue::String(op.into()),
            ),
            (
                PickleValue::String("name".into()),
                PickleValue::String(name.to_string()),
            ),
            (
                PickleValue::String("attach_point".into()),
                PickleValue::String(attach_point.to_string()),
            ),
        ]))?;
        parse_hook_result(&response)
    }

    pub fn set_hook_priority(
        &mut self,
        name: &str,
        attach_point: &str,
        priority: i32,
    ) -> io::Result<Result<(), String>> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("hook".into()),
                PickleValue::String("set_priority".into()),
            ),
            (
                PickleValue::String("name".into()),
                PickleValue::String(name.to_string()),
            ),
            (
                PickleValue::String("attach_point".into()),
                PickleValue::String(attach_point.to_string()),
            ),
            (
                PickleValue::String("priority".into()),
                PickleValue::Int(priority as i64),
            ),
        ]))?;
        parse_hook_result(&response)
    }

    pub fn blacklist_backbone_peer(
        &mut self,
        interface: &str,
        ip: &str,
        duration_secs: u64,
        reason: Option<&str>,
        penalty_level: Option<u8>,
    ) -> io::Result<bool> {
        let mut request = vec![
            (
                PickleValue::String("set".into()),
                PickleValue::String("backbone_peer_blacklist".into()),
            ),
            (
                PickleValue::String("interface".into()),
                PickleValue::String(interface.to_string()),
            ),
            (
                PickleValue::String("ip".into()),
                PickleValue::String(ip.to_string()),
            ),
            (
                PickleValue::String("duration_secs".into()),
                PickleValue::Int(duration_secs as i64),
            ),
        ];
        if let Some(reason) = reason {
            request.push((
                PickleValue::String("reason".into()),
                PickleValue::String(reason.to_string()),
            ));
        }
        if let Some(level) = penalty_level {
            request.push((
                PickleValue::String("penalty_level".into()),
                PickleValue::Int(level as i64),
            ));
        }
        let response = self.call(&PickleValue::Dict(request))?;
        Ok(response.as_bool().unwrap_or(false))
    }

    pub fn known_destinations(&mut self) -> io::Result<Vec<KnownDestinationEntry>> {
        let response = self.call(&PickleValue::Dict(vec![(
            PickleValue::String("get".into()),
            PickleValue::String("known_destinations".into()),
        )]))?;
        parse_known_destination_list(&response)
    }

    pub fn retain_known_destination(&mut self, dest_hash: [u8; 16]) -> io::Result<bool> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("set".into()),
                PickleValue::String("known_destination_retained".into()),
            ),
            (
                PickleValue::String("dest_hash".into()),
                PickleValue::Bytes(dest_hash.to_vec()),
            ),
        ]))?;
        Ok(response.as_bool().unwrap_or(false))
    }

    pub fn retain_identity(&mut self, identity_hash: [u8; 16]) -> io::Result<bool> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("set".into()),
                PickleValue::String("identity_retained".into()),
            ),
            (
                PickleValue::String("identity_hash".into()),
                PickleValue::Bytes(identity_hash.to_vec()),
            ),
        ]))?;
        Ok(response.as_bool().unwrap_or(false))
    }

    pub fn unretain_known_destination(&mut self, dest_hash: [u8; 16]) -> io::Result<bool> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("clear".into()),
                PickleValue::String("known_destination_retained".into()),
            ),
            (
                PickleValue::String("dest_hash".into()),
                PickleValue::Bytes(dest_hash.to_vec()),
            ),
        ]))?;
        Ok(response.as_bool().unwrap_or(false))
    }

    pub fn mark_known_destination_used(&mut self, dest_hash: [u8; 16]) -> io::Result<bool> {
        let response = self.call(&PickleValue::Dict(vec![
            (
                PickleValue::String("set".into()),
                PickleValue::String("known_destination_used".into()),
            ),
            (
                PickleValue::String("dest_hash".into()),
                PickleValue::Bytes(dest_hash.to_vec()),
            ),
        ]))?;
        Ok(response.as_bool().unwrap_or(false))
    }
}

fn parse_lifecycle_state(value: &str) -> Option<LifecycleState> {
    match value {
        "active" => Some(LifecycleState::Active),
        "draining" => Some(LifecycleState::Draining),
        "stopping" => Some(LifecycleState::Stopping),
        "stopped" => Some(LifecycleState::Stopped),
        _ => None,
    }
}

fn parse_drain_status(value: &PickleValue) -> io::Result<Option<DrainStatus>> {
    if !matches!(value, PickleValue::Dict(_)) {
        return Ok(None);
    }
    let state = value
        .get("state")
        .and_then(|entry| entry.as_str())
        .and_then(parse_lifecycle_state)
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing drain state"))?;
    let drain_age_seconds = value.get("drain_age_seconds").and_then(|entry| {
        entry
            .as_float()
            .or_else(|| entry.as_int().map(|v| v as f64))
    });
    let deadline_remaining_seconds = value.get("deadline_remaining_seconds").and_then(|entry| {
        entry
            .as_float()
            .or_else(|| entry.as_int().map(|v| v as f64))
    });
    let drain_complete = value
        .get("drain_complete")
        .and_then(|entry| entry.as_bool())
        .unwrap_or(false);
    let interface_writer_queued_frames = value
        .get("interface_writer_queued_frames")
        .and_then(|entry| entry.as_int())
        .unwrap_or(0)
        .max(0) as usize;
    let provider_backlog_events = value
        .get("provider_backlog_events")
        .and_then(|entry| entry.as_int())
        .unwrap_or(0)
        .max(0) as usize;
    let provider_consumer_queued_events = value
        .get("provider_consumer_queued_events")
        .and_then(|entry| entry.as_int())
        .unwrap_or(0)
        .max(0) as usize;
    let detail = value
        .get("detail")
        .and_then(|entry| entry.as_str().map(|v| v.to_string()));
    Ok(Some(DrainStatus {
        state,
        drain_age_seconds,
        deadline_remaining_seconds,
        drain_complete,
        interface_writer_queued_frames,
        provider_backlog_events,
        provider_consumer_queued_events,
        detail,
    }))
}

fn parse_hook_result(response: &PickleValue) -> io::Result<Result<(), String>> {
    let ok = response
        .get("ok")
        .and_then(|v| v.as_bool())
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "invalid hook response"))?;
    if ok {
        Ok(Ok(()))
    } else {
        Ok(Err(response
            .get("error")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown hook error")
            .to_string()))
    }
}

fn parse_hook_list(response: &PickleValue) -> io::Result<Vec<HookInfo>> {
    let list = response
        .as_list()
        .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "invalid hooks response"))?;
    let mut hooks = Vec::with_capacity(list.len());
    for item in list {
        hooks.push(HookInfo {
            name: item
                .get("name")
                .and_then(|v| v.as_str())
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing hook name"))?
                .to_string(),
            hook_type: item
                .get("type")
                .and_then(|v| v.as_str())
                .unwrap_or(default_hook_type())
                .to_string(),
            attach_point: item
                .get("attach_point")
                .and_then(|v| v.as_str())
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing attach_point"))?
                .to_string(),
            priority: item
                .get("priority")
                .and_then(|v| v.as_int())
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing priority"))?
                as i32,
            enabled: item
                .get("enabled")
                .and_then(|v| v.as_bool())
                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "missing enabled"))?,
            consecutive_traps: item
                .get("consecutive_traps")
                .and_then(|v| v.as_int())
                .ok_or_else(|| {
                    io::Error::new(io::ErrorKind::InvalidData, "missing consecutive_traps")
                })? as u32,
        });
    }
    Ok(hooks)
}

fn default_hook_type() -> &'static str {
    #[cfg(feature = "rns-hooks-native")]
    {
        "native"
    }
    #[cfg(all(not(feature = "rns-hooks-native"), feature = "rns-hooks-wasm"))]
    {
        return "wasm";
    }
    #[cfg(all(not(feature = "rns-hooks-native"), not(feature = "rns-hooks-wasm")))]
    {
        "wasm"
    }
}

/// Client-side authentication: answer the server's challenge.
fn client_auth(stream: &mut TcpStream, auth_key: &[u8; 32]) -> io::Result<()> {
    answer_challenge(stream, auth_key)?;
    server_auth(stream, auth_key)
}

fn answer_challenge<S: Read + Write>(stream: &mut S, auth_key: &[u8; 32]) -> io::Result<()> {
    // Read challenge
    let challenge = recv_bytes(stream)?;

    if !challenge.starts_with(CHALLENGE_PREFIX) {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            "expected challenge",
        ));
    }

    let message = &challenge[CHALLENGE_PREFIX.len()..];

    // Create HMAC response
    let response = create_response(auth_key, message);
    send_bytes(stream, &response)?;

    // Read welcome/failure
    let result = recv_bytes(stream)?;
    if result == WELCOME {
        Ok(())
    } else {
        Err(io::Error::new(
            io::ErrorKind::PermissionDenied,
            "authentication failed",
        ))
    }
}

/// Create an HMAC response to a challenge message.
fn create_response(auth_key: &[u8; 32], message: &[u8]) -> Vec<u8> {
    // Check if message has {sha256} prefix (modern protocol)
    if message.starts_with(b"{sha256}") || message.len() > 20 {
        // Modern protocol: use HMAC-SHA256 with {sha256} prefix
        let digest = hmac_sha256(auth_key, message);
        let mut response = Vec::with_capacity(8 + 32);
        response.extend_from_slice(b"{sha256}");
        response.extend_from_slice(&digest);
        response
    } else {
        // Legacy protocol: raw HMAC-MD5
        let digest = hmac_md5(auth_key, message);
        digest.to_vec()
    }
}

/// Derive the RPC auth key from transport identity private key.
pub fn derive_auth_key(private_key: &[u8]) -> [u8; 32] {
    sha256(private_key)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn discovered_interface_rpc_includes_operator_lxmf_address() {
        let interface = crate::discovery::DiscoveredInterface {
            interface_type: "BackboneInterface".into(),
            impl_name: Some("Reticulum".into()),
            impl_version: Some("1.5.5".into()),
            transport: true,
            name: "Operator Test".into(),
            discovered: 1.0,
            last_heard: 2.0,
            heard_count: 1,
            status: crate::discovery::DiscoveredStatus::Available,
            stamp: vec![0; crate::discovery::STAMP_SIZE],
            stamp_value: 16,
            transport_id: [0x11; 16],
            network_id: [0x22; 16],
            hops: 1,
            latitude: None,
            longitude: None,
            height: None,
            operator_lxmf_address: Some([0xa5; 16]),
            reachable_on: Some("example.com".into()),
            port: Some(4242),
            frequency: None,
            bandwidth: None,
            spreading_factor: None,
            coding_rate: None,
            modulation: None,
            channel: None,
            ifac_netname: None,
            ifac_netkey: None,
            config_entry: None,
            discovery_hash: [0x33; 32],
        };

        let encoded = discovered_interfaces_to_pickle(&[interface]);
        let first = &encoded.as_list().unwrap()[0];

        assert_eq!(
            first.get("impl_name").and_then(PickleValue::as_str),
            Some("Reticulum")
        );
        assert_eq!(
            first.get("version").and_then(PickleValue::as_str),
            Some("1.5.5")
        );

        assert_eq!(
            first
                .get("operator_lxmf_address")
                .and_then(PickleValue::as_str),
            Some("a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5")
        );
    }

    #[test]
    fn send_recv_bytes_roundtrip() {
        let (mut c1, mut c2) = tcp_pair();
        let data = b"hello world";
        send_bytes(&mut c1, data).unwrap();
        let received = recv_bytes(&mut c2).unwrap();
        assert_eq!(&received, data);
    }

    #[test]
    fn send_recv_empty() {
        let (mut c1, mut c2) = tcp_pair();
        send_bytes(&mut c1, b"").unwrap();
        let received = recv_bytes(&mut c2).unwrap();
        assert!(received.is_empty());
    }

    #[test]
    fn auth_success() {
        let key = derive_auth_key(b"test-private-key");
        let (mut server, mut client) = tcp_pair();

        let key2 = key;
        let t = thread::spawn(move || {
            client_auth(&mut client, &key2).unwrap();
        });

        server_auth(&mut server, &key).unwrap();
        answer_challenge(&mut server, &key).unwrap();
        t.join().unwrap();
    }

    #[test]
    fn auth_failure_wrong_key() {
        let server_key = derive_auth_key(b"server-key");
        let client_key = derive_auth_key(b"wrong-key");
        let (mut server, mut client) = tcp_pair();

        let t = thread::spawn(move || {
            let result = client_auth(&mut client, &client_key);
            assert!(result.is_err());
        });

        let result = server_auth(&mut server, &server_key);
        assert!(result.is_err());
        t.join().unwrap();
    }

    #[test]
    fn verify_sha256_response() {
        let key = derive_auth_key(b"mykey");
        let message = b"{sha256}abcdefghijklmnopqrstuvwxyz0123456789ABCD";
        let response = create_response(&key, message);
        assert!(response.starts_with(b"{sha256}"));
        assert!(verify_response(&key, message, &response));
    }

    #[test]
    fn verify_legacy_md5_response() {
        let key = derive_auth_key(b"mykey");
        // Legacy message: 20 bytes, no prefix
        let message = b"01234567890123456789";
        // Create legacy response (raw HMAC-MD5)
        let digest = hmac_md5(&key, message);
        assert!(verify_response(&key, message, &digest));
    }

    #[test]
    fn constant_time_eq_works() {
        assert!(constant_time_eq(b"hello", b"hello"));
        assert!(!constant_time_eq(b"hello", b"world"));
        assert!(!constant_time_eq(b"hello", b"hell"));
    }

    #[test]
    fn rpc_payload_accepts_msgpack_and_legacy_pickle() {
        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("get".into()),
                PickleValue::String("link_count".into()),
            ),
            (PickleValue::String("max_hops".into()), PickleValue::Int(3)),
        ]);

        let msgpack_bytes = encode_rpc_payload(&request, RpcCodec::Msgpack);
        let (decoded_msgpack, msgpack_codec) = decode_rpc_payload(&msgpack_bytes).unwrap();
        assert_eq!(decoded_msgpack, request);
        assert_eq!(msgpack_codec, RpcCodec::Msgpack);

        let pickle_bytes = encode_rpc_payload(&request, RpcCodec::Pickle);
        let (decoded_pickle, pickle_codec) = decode_rpc_payload(&pickle_bytes).unwrap();
        assert_eq!(decoded_pickle, request);
        assert_eq!(pickle_codec, RpcCodec::Pickle);
    }

    #[test]
    fn rpc_roundtrip() {
        let key = derive_auth_key(b"test-key");
        let (event_tx, event_rx) = crate::event::channel();

        // Start server
        // Bind manually to get the actual port
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        listener.set_nonblocking(true).unwrap();

        let shutdown = Arc::new(AtomicBool::new(false));
        let shutdown2 = shutdown.clone();

        // Driver thread that handles queries
        let driver_thread = thread::spawn(move || loop {
            match event_rx.recv_timeout(std::time::Duration::from_secs(5)) {
                Ok(Event::Query(request, resp_tx))
                    if matches!(*request, QueryRequest::LinkCount) =>
                {
                    let _ = resp_tx.send(QueryResponse::LinkCount(42));
                }
                Ok(Event::Query(request, resp_tx))
                    if matches!(*request, QueryRequest::ActiveLinkCount) =>
                {
                    let _ = resp_tx.send(QueryResponse::ActiveLinkCount(7));
                }
                Ok(Event::Query(request, resp_tx))
                    if matches!(*request, QueryRequest::LowestInterfaceBitrate) =>
                {
                    let _ = resp_tx.send(QueryResponse::LowestInterfaceBitrate(Some(400)));
                }
                Ok(Event::Query(request, resp_tx))
                    if matches!(*request, QueryRequest::MediumPathTimeout) =>
                {
                    let _ = resp_tx.send(QueryResponse::MediumPathTimeout(26.0));
                }
                Ok(Event::Query(request, resp_tx))
                    if matches!(*request, QueryRequest::FirstHopTimeout { .. }) =>
                {
                    let QueryRequest::FirstHopTimeout { dest_hash } = *request else {
                        unreachable!()
                    };
                    assert_eq!(dest_hash, [0x77; 16]);
                    let _ = resp_tx.send(QueryResponse::FirstHopTimeout(6.4));
                }
                Ok(Event::Query(request, resp_tx))
                    if matches!(*request, QueryRequest::InterfaceStats) =>
                {
                    let _ = resp_tx.send(QueryResponse::InterfaceStats(InterfaceStatsResponse {
                        interfaces: vec![SingleInterfaceStat {
                            id: 7,
                            name: "TestInterface".into(),
                            status: true,
                            mode: 1,
                            gravity: -2,
                            announces_to_internal: Some(true),
                            rxb: 1000,
                            txb: 2000,
                            tx_drops: 0,
                            tx_dropped_bytes: 0,
                            tx_stalled: false,
                            tx_buffered: 0,
                            traffic: TrafficDetail::default(),
                            protocol_violations: 0,
                            ifac_violations: 0,
                            packet_filter_hits: 0,
                            rx_packets: 10,
                            tx_packets: 20,
                            cpu_load: None,
                            mem_load: None,
                            switch_id: None,
                            endpoint_id: None,
                            via_switch_id: None,
                            peers: None,
                            bitrate: Some(10_000_000),
                            mtu: Some(262_144),
                            ifac_size: None,
                            started: 1000.0,
                            ia_freq: 0.0,
                            ip_freq: 0.0,
                            op_freq: 0.0,
                            op_samples: 0,
                            burst_active: false,
                            burst_activated: 0.0,
                            pr_burst_active: false,
                            pr_burst_activated: 0.0,
                            oa_freq: 0.0,
                            clients: Some(2),
                            blocked_ips: None,
                            blocked_ip_list: None,
                            announce_rate_target: Some(3600.0),
                            announce_rate_grace: 5,
                            announce_rate_penalty: 0.0,
                            interface_type: "TestInterface".into(),
                        }],
                        transport_id: None,
                        transport_enabled: true,
                        transport_uptime: 3600.0,
                        total_rxb: 1000,
                        total_txb: 2000,
                        rx_pps: 3.0,
                        tx_pps: 4.0,
                        traffic: TrafficDetail::default(),
                        probe_responder: None,
                        backbone_peer_pool: None,
                    }));
                }
                _ => break,
            }
        });

        let key2 = key;
        let shutdown3 = shutdown2.clone();
        let server_thread = thread::spawn(move || {
            rpc_server_loop(listener, key2, event_tx, shutdown3);
        });

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

        // Client: connect and query link count
        let server_addr = RpcAddr::Tcp("127.0.0.1".into(), port);
        let mut client = RpcClient::connect(&server_addr, &key).unwrap();
        let response = client
            .call(&PickleValue::Dict(vec![(
                PickleValue::String("get".into()),
                PickleValue::String("link_count".into()),
            )]))
            .unwrap();
        assert_eq!(response.as_int().unwrap(), 42);
        drop(client);

        let mut active_client = RpcClient::connect(&server_addr, &key).unwrap();
        let active_response = active_client
            .call(&PickleValue::Dict(vec![(
                PickleValue::String("get".into()),
                PickleValue::String("active_link_count".into()),
            )]))
            .unwrap();
        assert_eq!(active_response.as_int().unwrap(), 7);
        drop(active_client);

        let mut timeout_client = RpcClient::connect(&server_addr, &key).unwrap();
        let bitrate = timeout_client
            .call(&PickleValue::Dict(vec![(
                PickleValue::String("get".into()),
                PickleValue::String("lowest_interface_bitrate".into()),
            )]))
            .unwrap();
        assert_eq!(bitrate.as_int(), Some(400));
        drop(timeout_client);
        let mut timeout_client = RpcClient::connect(&server_addr, &key).unwrap();
        let timeout = timeout_client
            .call(&PickleValue::Dict(vec![(
                PickleValue::String("get".into()),
                PickleValue::String("medium_path_timeout".into()),
            )]))
            .unwrap();
        assert_eq!(timeout.as_float(), Some(26.0));
        drop(timeout_client);
        let mut timeout_client = RpcClient::connect(&server_addr, &key).unwrap();
        let first_hop = timeout_client
            .call(&PickleValue::Dict(vec![
                (
                    PickleValue::String("get".into()),
                    PickleValue::String("first_hop_timeout".into()),
                ),
                (
                    PickleValue::String("destination_hash".into()),
                    PickleValue::Bytes(vec![0x77; 16]),
                ),
            ]))
            .unwrap();
        assert_eq!(first_hop.as_float(), Some(6.4));
        drop(timeout_client);

        // Client: query interface stats
        let mut client2 = RpcClient::connect(&server_addr, &key).unwrap();
        let response2 = client2
            .call(&PickleValue::Dict(vec![(
                PickleValue::String("get".into()),
                PickleValue::String("interface_stats".into()),
            )]))
            .unwrap();
        let ifaces = response2.get("interfaces").unwrap().as_list().unwrap();
        assert_eq!(ifaces.len(), 1);
        let iface = &ifaces[0];
        assert_eq!(
            iface.get("name").unwrap().as_str().unwrap(),
            "TestInterface"
        );
        assert_eq!(iface.get("rxb").unwrap().as_int().unwrap(), 1000);
        assert_eq!(iface.get("mtu").unwrap().as_int(), Some(262_144));
        drop(client2);

        // Shutdown
        shutdown2.store(true, Ordering::Relaxed);
        server_thread.join().unwrap();
        driver_thread.join().unwrap();
    }

    #[test]
    fn full_event_queue_fails_rpc_query_without_blocking() {
        let (event_tx, event_rx) = crate::event::channel_with_capacity(1);
        event_tx.send(Event::Tick).unwrap();

        let (result_tx, result_rx) = mpsc::channel();
        let query_thread = thread::spawn(move || {
            let _ = result_tx.send(send_query(&event_tx, QueryRequest::LinkCount));
        });

        let completed_promptly = match result_rx.recv_timeout(Duration::from_millis(250)) {
            Ok(result) => {
                assert_eq!(result.unwrap_err().kind(), io::ErrorKind::WouldBlock);
                true
            }
            Err(mpsc::RecvTimeoutError::Timeout) => false,
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                panic!("RPC query thread disconnected without a result")
            }
        };

        // Let the old blocking implementation finish so a regression failure does
        // not leak a thread into the rest of the test process.
        assert!(matches!(event_rx.recv().unwrap(), Event::Tick));
        if !completed_promptly {
            match event_rx.recv_timeout(Duration::from_secs(1)).unwrap() {
                Event::Query(request, response_tx)
                    if matches!(*request, QueryRequest::LinkCount) =>
                {
                    response_tx.send(QueryResponse::LinkCount(1)).unwrap();
                }
                other => panic!("unexpected event after releasing queue: {other:?}"),
            }
            let _ = result_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        }
        query_thread.join().unwrap();

        assert!(
            completed_promptly,
            "RPC query blocked while the driver event queue was full"
        );
    }

    #[test]
    fn stalled_rpc_request_does_not_block_later_clients() {
        let key = derive_auth_key(b"concurrent-rpc-test");
        let (event_tx, event_rx) = crate::event::channel();

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        listener.set_nonblocking(true).unwrap();

        let shutdown = Arc::new(AtomicBool::new(false));
        let server_shutdown = shutdown.clone();
        let server_thread = thread::spawn(move || {
            rpc_server_loop(listener, key, event_tx, server_shutdown);
        });

        let (first_seen_tx, first_seen_rx) = mpsc::channel();
        let (concurrent_tx, concurrent_rx) = mpsc::channel();
        let driver_thread = thread::spawn(move || {
            let first_response = match event_rx.recv_timeout(Duration::from_secs(1)).unwrap() {
                Event::Query(request, response_tx)
                    if matches!(*request, QueryRequest::LinkCount) =>
                {
                    response_tx
                }
                other => panic!("unexpected first RPC event: {other:?}"),
            };
            first_seen_tx.send(()).unwrap();

            let second = event_rx.recv_timeout(Duration::from_millis(500));
            let handled_concurrently = match second {
                Ok(Event::Query(request, response_tx))
                    if matches!(*request, QueryRequest::TransportIdentity) =>
                {
                    response_tx
                        .send(QueryResponse::TransportIdentity(Some([7; 16])))
                        .unwrap();
                    true
                }
                Ok(other) => panic!("unexpected second RPC event: {other:?}"),
                Err(mpsc::RecvTimeoutError::Timeout) => false,
                Err(mpsc::RecvTimeoutError::Disconnected) => {
                    panic!("RPC event channel disconnected")
                }
            };

            first_response.send(QueryResponse::LinkCount(1)).unwrap();

            if !handled_concurrently {
                match event_rx.recv_timeout(Duration::from_secs(1)).unwrap() {
                    Event::Query(request, response_tx)
                        if matches!(*request, QueryRequest::TransportIdentity) =>
                    {
                        response_tx
                            .send(QueryResponse::TransportIdentity(Some([7; 16])))
                            .unwrap();
                    }
                    other => panic!("unexpected delayed RPC event: {other:?}"),
                }
            }
            concurrent_tx.send(handled_concurrently).unwrap();
        });

        let first_addr = RpcAddr::Tcp("127.0.0.1".into(), port);
        let first_client = thread::spawn(move || {
            let mut client = RpcClient::connect(&first_addr, &key).unwrap();
            client
                .call(&PickleValue::Dict(vec![(
                    PickleValue::String("get".into()),
                    PickleValue::String("link_count".into()),
                )]))
                .unwrap()
        });

        first_seen_rx.recv_timeout(Duration::from_secs(1)).unwrap();

        let second_addr = RpcAddr::Tcp("127.0.0.1".into(), port);
        let second_client = thread::spawn(move || {
            let mut client = RpcClient::connect(&second_addr, &key).unwrap();
            client
                .call(&PickleValue::Dict(vec![(
                    PickleValue::String("get".into()),
                    PickleValue::String("transport_identity".into()),
                )]))
                .unwrap()
        });

        let handled_concurrently = concurrent_rx.recv_timeout(Duration::from_secs(2)).unwrap();
        assert_eq!(first_client.join().unwrap().as_int(), Some(1));
        assert_eq!(
            second_client.join().unwrap().as_bytes(),
            Some([7; 16].as_slice())
        );

        shutdown.store(true, Ordering::Relaxed);
        server_thread.join().unwrap();
        driver_thread.join().unwrap();

        assert!(
            handled_concurrently,
            "a stalled RPC request prevented the server from handling a later client"
        );
    }

    #[test]
    fn derive_auth_key_deterministic() {
        let key1 = derive_auth_key(b"test");
        let key2 = derive_auth_key(b"test");
        assert_eq!(key1, key2);
        // Different input → different key
        let key3 = derive_auth_key(b"other");
        assert_ne!(key1, key3);
    }

    #[test]
    fn pickle_request_handling() {
        // Test the request → query translation without networking
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((QueryRequest::DropPath { dest_hash }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(dest_hash, [1u8; 16]);
                let _ = resp_tx.send(QueryResponse::DropPath(true));
            }
        });

        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("drop".into()),
                PickleValue::String("path".into()),
            ),
            (
                PickleValue::String("destination_hash".into()),
                PickleValue::Bytes(vec![1u8; 16]),
            ),
        ]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::Bool(true));
        driver.join().unwrap();
    }

    #[test]
    fn hook_list_request_handling() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Ok(Event::ListHooks { response_tx }) = event_rx.recv() {
                let _ = response_tx.send(vec![HookInfo {
                    name: "stats".into(),
                    hook_type: "wasm".into(),
                    attach_point: "PreIngress".into(),
                    priority: 7,
                    enabled: true,
                    consecutive_traps: 0,
                }]);
            }
        });

        let request = PickleValue::Dict(vec![(
            PickleValue::String("get".into()),
            PickleValue::String("hooks".into()),
        )]);
        let response = handle_rpc_request(&request, &event_tx).unwrap();
        let hooks = parse_hook_list(&response).unwrap();
        assert_eq!(hooks.len(), 1);
        assert_eq!(hooks[0].name, "stats");
        driver.join().unwrap();
    }

    #[test]
    fn hook_load_request_handling() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Ok(Event::LoadHook {
                name,
                wasm_bytes,
                attach_point,
                priority,
                response_tx,
            }) = event_rx.recv()
            {
                assert_eq!(name, "stats");
                assert_eq!(attach_point, "PreIngress");
                assert_eq!(priority, 11);
                assert_eq!(wasm_bytes, vec![1, 2, 3]);
                let _ = response_tx.send(Ok(()));
            }
        });

        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("hook".into()),
                PickleValue::String("load".into()),
            ),
            (
                PickleValue::String("name".into()),
                PickleValue::String("stats".into()),
            ),
            (
                PickleValue::String("attach_point".into()),
                PickleValue::String("PreIngress".into()),
            ),
            (PickleValue::String("priority".into()), PickleValue::Int(11)),
            (
                PickleValue::String("wasm".into()),
                PickleValue::Bytes(vec![1, 2, 3]),
            ),
        ]);
        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(parse_hook_result(&response).unwrap(), Ok(()));
        driver.join().unwrap();
    }

    #[test]
    fn interface_stats_pickle_format() {
        let traffic = TrafficDetail {
            rxs: 800.0,
            txs: 1600.0,
            arxb: 11,
            atxb: 22,
            arxc: 1,
            atxc: 2,
            arxs: 88.0,
            atxs: 176.0,
            prxb: 33,
            ptxb: 44,
            prxc: 3,
            ptxc: 4,
            prxs: 264.0,
            ptxs: 352.0,
            arxf: 1.0,
            atxf: 2.0,
            prxf: 3.0,
            ptxf: 4.0,
        };
        let mut stats = InterfaceStatsResponse {
            interfaces: vec![SingleInterfaceStat {
                id: 1,
                name: "TCP".into(),
                status: true,
                mode: 1,
                gravity: -2,
                announces_to_internal: Some(true),
                rxb: 100,
                txb: 200,
                tx_drops: 3,
                tx_dropped_bytes: 4096,
                tx_stalled: true,
                tx_buffered: 8192,
                traffic,
                protocol_violations: 11,
                ifac_violations: 12,
                packet_filter_hits: 13,
                rx_packets: 5,
                tx_packets: 10,
                cpu_load: Some(12.0),
                mem_load: Some(34.5),
                switch_id: Some([1, 2, 3, 4]),
                endpoint_id: Some([5; 8]),
                via_switch_id: Some([6; 4]),
                peers: Some(2),
                bitrate: Some(1000000),
                mtu: Some(131_072),
                ifac_size: Some(16),
                started: 1000.0,
                ia_freq: 1.0,
                ip_freq: 2.0,
                op_freq: 3.0,
                op_samples: 0,
                burst_active: true,
                burst_activated: 1200.0,
                pr_burst_active: true,
                pr_burst_activated: 1300.0,
                oa_freq: 4.0,
                clients: Some(3),
                blocked_ips: Some(2),
                blocked_ip_list: Some(vec!["192.0.2.1".into(), "203.0.113.2".into()]),
                announce_rate_target: Some(3600.0),
                announce_rate_grace: 5,
                announce_rate_penalty: 0.0,
                interface_type: "TCPClientInterface".into(),
            }],
            transport_id: Some([0xAB; 16]),
            transport_enabled: true,
            transport_uptime: 3600.0,
            total_rxb: 100,
            total_txb: 200,
            rx_pps: 3.0,
            tx_pps: 4.0,
            traffic,
            probe_responder: None,
            backbone_peer_pool: None,
        };

        let pickle = interface_stats_to_pickle(&stats);

        // Verify it round-trips through encode/decode
        let encoded = pickle::encode(&pickle);
        let decoded = pickle::decode(&encoded).unwrap();
        assert!(decoded.get("transport_enabled").unwrap().as_bool().unwrap());
        let ifaces = decoded.get("interfaces").unwrap().as_list().unwrap();
        assert_eq!(decoded.get("arxf").unwrap().as_float(), Some(1.0));
        assert_eq!(decoded.get("atxf").unwrap().as_float(), Some(2.0));
        assert_eq!(decoded.get("prxf").unwrap().as_float(), Some(3.0));
        assert_eq!(decoded.get("ptxf").unwrap().as_float(), Some(4.0));
        assert_eq!(decoded.get("rxpps").unwrap().as_float(), Some(3.0));
        assert_eq!(decoded.get("txpps").unwrap().as_float(), Some(4.0));
        assert_eq!(ifaces[0].get("id").unwrap().as_int().unwrap(), 1);
        assert_eq!(ifaces[0].get("name").unwrap().as_str().unwrap(), "TCP");
        assert_eq!(ifaces[0].get("mtu").unwrap().as_int(), Some(131_072));
        assert_eq!(ifaces[0].get("gravity").unwrap().as_int().unwrap(), -2);
        assert_eq!(ifaces[0].get("txdrp").unwrap().as_int(), Some(3));
        assert_eq!(ifaces[0].get("txdrb").unwrap().as_int(), Some(4096));
        assert_eq!(ifaces[0].get("txstalled").unwrap().as_bool(), Some(true));
        assert_eq!(ifaces[0].get("txbuffered").unwrap().as_int(), Some(8192));
        assert!(ifaces[0]
            .get("announces_to_internal")
            .unwrap()
            .as_bool()
            .unwrap());
        assert_eq!(ifaces[0].get("ia_freq").unwrap().as_float().unwrap(), 1.0);
        assert_eq!(ifaces[0].get("ip_freq").unwrap().as_float().unwrap(), 2.0);
        assert_eq!(ifaces[0].get("op_freq").unwrap().as_float().unwrap(), 3.0);
        assert_eq!(ifaces[0].get("oa_freq").unwrap().as_float().unwrap(), 4.0);
        assert_eq!(ifaces[0].get("arxb").unwrap().as_int().unwrap(), 11);
        assert_eq!(ifaces[0].get("atxb").unwrap().as_int().unwrap(), 22);
        assert_eq!(ifaces[0].get("prxb").unwrap().as_int().unwrap(), 33);
        assert_eq!(ifaces[0].get("ptxb").unwrap().as_int().unwrap(), 44);
        assert_eq!(ifaces[0].get("arxs").unwrap().as_float().unwrap(), 88.0);
        assert_eq!(ifaces[0].get("ptxs").unwrap().as_float().unwrap(), 352.0);
        assert_eq!(
            ifaces[0]
                .get("protocol_violations")
                .unwrap()
                .as_int()
                .unwrap(),
            11
        );
        assert_eq!(ifaces[0].get("ifac_violations").unwrap().as_int(), Some(12));
        assert_eq!(
            ifaces[0].get("packet_filter_hits").unwrap().as_int(),
            Some(13)
        );
        assert_eq!(decoded.get("arxb").unwrap().as_int().unwrap(), 11);
        assert_eq!(decoded.get("prxb").unwrap().as_int().unwrap(), 33);
        assert!(ifaces[0].get("burst_active").unwrap().as_bool().unwrap());
        assert_eq!(
            ifaces[0]
                .get("burst_activated")
                .unwrap()
                .as_float()
                .unwrap(),
            1200.0
        );
        assert!(ifaces[0].get("pr_burst_active").unwrap().as_bool().unwrap());
        assert_eq!(
            ifaces[0]
                .get("pr_burst_activated")
                .unwrap()
                .as_float()
                .unwrap(),
            1300.0
        );
        assert_eq!(
            ifaces[0]
                .get("announce_rate_target")
                .unwrap()
                .as_float()
                .unwrap(),
            3600.0
        );
        assert_eq!(
            ifaces[0]
                .get("announce_rate_grace")
                .unwrap()
                .as_int()
                .unwrap(),
            5
        );
        assert_eq!(
            ifaces[0]
                .get("announce_rate_penalty")
                .unwrap()
                .as_float()
                .unwrap(),
            0.0
        );
        assert_eq!(ifaces[0].get("clients").unwrap().as_int().unwrap(), 3);
        assert_eq!(ifaces[0].get("blocked_ips").unwrap().as_int().unwrap(), 2);
        assert_eq!(
            ifaces[0]
                .get("blocked_ip_list")
                .unwrap()
                .as_list()
                .unwrap()
                .iter()
                .map(|value| value.as_str().unwrap())
                .collect::<Vec<_>>(),
            vec!["192.0.2.1", "203.0.113.2"]
        );

        stats.interfaces[0].interface_type = "BackboneInterface".into();
        let (event_tx, _event_rx) = crate::event::channel_with_capacity(1);
        event_tx.register_dynamic_parent(InterfaceId(2), InterfaceId(1));
        event_tx.set_ingress_bursts(InterfaceId(2), Some(10.0), Some(20.0));
        event_tx
            .try_send(Event::Frame {
                interface_id: InterfaceId(2),
                data: vec![0],
                rssi: None,
                snr: None,
            })
            .unwrap();
        assert!(event_tx
            .try_send(Event::Frame {
                interface_id: InterfaceId(2),
                data: vec![0],
                rssi: None,
                snr: None,
            })
            .is_err());
        // Reader `send` now waits for space in the data class instead of
        // dropping, so a second refused `try_send` provides the second drop.
        assert!(event_tx
            .try_send(Event::Frame {
                interface_id: InterfaceId(2),
                data: vec![0],
                rssi: None,
                snr: None,
            })
            .is_err());
        let mut runtime_pickle = interface_stats_to_pickle(&stats);
        add_inbound_runtime_stats(&mut runtime_pickle, &stats, &event_tx);
        let runtime_ifaces = runtime_pickle.get("interfaces").unwrap().as_list().unwrap();
        assert_eq!(
            runtime_ifaces[0]
                .get("burst_count")
                .unwrap()
                .as_int()
                .unwrap(),
            1
        );
        assert_eq!(
            runtime_ifaces[0]
                .get("pr_burst_count")
                .unwrap()
                .as_int()
                .unwrap(),
            1
        );
        assert!(runtime_ifaces[0]
            .get("burst_active")
            .unwrap()
            .as_bool()
            .unwrap());
        assert_eq!(
            runtime_ifaces[0]
                .get("burst_activated")
                .unwrap()
                .as_float()
                .unwrap(),
            10.0
        );
        assert!(runtime_ifaces[0]
            .get("pr_burst_active")
            .unwrap()
            .as_bool()
            .unwrap());
        assert_eq!(
            runtime_ifaces[0]
                .get("pr_burst_activated")
                .unwrap()
                .as_float()
                .unwrap(),
            20.0
        );
        assert_eq!(runtime_pickle.get("rxqt").unwrap().as_int().unwrap(), 1);
        assert_eq!(runtime_pickle.get("rxqd").unwrap().as_int().unwrap(), 1);
        assert_eq!(runtime_pickle.get("rxqtd").unwrap().as_int().unwrap(), 2);
        assert_eq!(runtime_pickle.get("rxqdd").unwrap().as_int().unwrap(), 2);
        assert_eq!(runtime_pickle.get("rxqad").unwrap().as_int().unwrap(), 0);
        assert_eq!(runtime_pickle.get("rxqpd").unwrap().as_int().unwrap(), 0);
        assert_eq!(runtime_pickle.get("rxqild").unwrap().as_int().unwrap(), 0);
        assert_eq!(
            runtime_pickle
                .get("dqpressure")
                .unwrap()
                .as_float()
                .unwrap(),
            1.0
        );
        assert_eq!(
            runtime_pickle
                .get("tqpressure")
                .unwrap()
                .as_float()
                .unwrap(),
            0.25
        );
    }

    #[test]
    fn profiling_results_rpc_exposes_completed_samples() {
        {
            let _sample = crate::profiling::profile("entry110.rpc");
        }
        let (event_tx, _event_rx) = crate::event::channel();
        let request = PickleValue::Dict(vec![(
            PickleValue::String("get".into()),
            PickleValue::String("profiling_results".into()),
        )]);
        let response = handle_rpc_request(&request, &event_tx).unwrap();
        let sample = response.get("entry110.rpc").unwrap();
        assert_eq!(
            sample.get("name").and_then(PickleValue::as_str),
            Some("entry110.rpc")
        );
        assert_eq!(
            sample
                .get("stats_all")
                .and_then(|stats| stats.get("count"))
                .and_then(PickleValue::as_int),
            Some(1)
        );
    }

    #[test]
    fn interface_stats_pickle_includes_backbone_peer_pool_priority() {
        let stats = InterfaceStatsResponse {
            interfaces: vec![],
            transport_id: None,
            transport_enabled: true,
            transport_uptime: 1.0,
            total_rxb: 0,
            total_txb: 0,
            rx_pps: 0.0,
            tx_pps: 0.0,
            traffic: TrafficDetail::default(),
            probe_responder: None,
            backbone_peer_pool: Some(crate::event::BackbonePeerPoolStatus {
                max_connected: 1,
                active_count: 1,
                standby_count: 0,
                cooldown_count: 0,
                members: vec![crate::event::BackbonePeerPoolMemberStatus {
                    name: "peer".into(),
                    remote: "127.0.0.1:4242".into(),
                    source: "configured".into(),
                    priority: 77,
                    state: "active".into(),
                    interface_id: Some(42),
                    failure_count: 0,
                    last_error: None,
                    cooldown_remaining_seconds: None,
                }],
            }),
        };

        let pickle = interface_stats_to_pickle(&stats);
        let encoded = pickle::encode(&pickle);
        let decoded = pickle::decode(&encoded).unwrap();
        let pool = decoded.get("backbone_peer_pool").unwrap();
        let members = pool.get("members").unwrap().as_list().unwrap();
        assert_eq!(members[0].get("priority").unwrap().as_int().unwrap(), 77);
    }

    #[test]
    fn send_probe_rpc_unknown_dest() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((
                QueryRequest::SendProbe {
                    dest_hash,
                    payload_size,
                },
                resp_tx,
            )) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(dest_hash, [0xAA; 16]);
                assert_eq!(payload_size, 16); // default
                let _ = resp_tx.send(QueryResponse::SendProbe(Err(
                    crate::event::SendProbeError::UnknownIdentity,
                )));
            }
        });

        let request = PickleValue::Dict(vec![(
            PickleValue::String("send_probe".into()),
            PickleValue::Bytes(vec![0xAA; 16]),
        )]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::None);
        driver.join().unwrap();
    }

    #[test]
    fn send_probe_rpc_with_result() {
        let (event_tx, event_rx) = crate::event::channel();

        let packet_hash = [0xBB; 32];
        let driver = thread::spawn(move || {
            if let Some((
                QueryRequest::SendProbe {
                    dest_hash,
                    payload_size,
                },
                resp_tx,
            )) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(dest_hash, [0xCC; 16]);
                assert_eq!(payload_size, 32);
                let _ = resp_tx.send(QueryResponse::SendProbe(Ok((packet_hash, 3))));
            }
        });

        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("send_probe".into()),
                PickleValue::Bytes(vec![0xCC; 16]),
            ),
            (PickleValue::String("size".into()), PickleValue::Int(32)),
        ]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        let ph = response.get("packet_hash").unwrap().as_bytes().unwrap();
        assert_eq!(ph, &[0xBB; 32]);
        assert_eq!(response.get("hops").unwrap().as_int().unwrap(), 3);
        driver.join().unwrap();
    }

    #[test]
    fn retain_identity_rpc_sends_identity_hash_query() {
        let (event_tx, event_rx) = crate::event::channel();
        let identity_hash = [0x44; 16];

        let driver = thread::spawn(move || {
            if let Some((
                QueryRequest::RetainIdentity {
                    identity_hash: hash,
                },
                resp_tx,
            )) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(hash, identity_hash);
                let _ = resp_tx.send(QueryResponse::RetainIdentity(true));
            }
        });

        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("set".into()),
                PickleValue::String("identity_retained".into()),
            ),
            (
                PickleValue::String("identity_hash".into()),
                PickleValue::Bytes(identity_hash.to_vec()),
            ),
        ]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::Bool(true));
        driver.join().unwrap();
    }

    #[test]
    fn send_probe_rpc_size_validation() {
        let (event_tx, event_rx) = crate::event::channel();

        // Negative size should be clamped to default (16)
        let driver = thread::spawn(move || {
            if let Some((QueryRequest::SendProbe { payload_size, .. }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(payload_size, 16); // default, not negative
                let _ = resp_tx.send(QueryResponse::SendProbe(Err(
                    crate::event::SendProbeError::UnknownIdentity,
                )));
            }
        });

        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("send_probe".into()),
                PickleValue::Bytes(vec![0xDD; 16]),
            ),
            (PickleValue::String("size".into()), PickleValue::Int(-1)),
        ]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::None);
        driver.join().unwrap();
    }

    #[test]
    fn send_probe_rpc_size_too_large() {
        let (event_tx, _event_rx) = crate::event::channel();

        // Oversize requests are rejected before reaching the driver instead
        // of being silently replaced by the default size.
        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("send_probe".into()),
                PickleValue::Bytes(vec![0xDD; 16]),
            ),
            (
                PickleValue::String("size".into()),
                PickleValue::Int(crate::event::MAX_PROBE_PAYLOAD_SIZE as i64 + 1),
            ),
        ]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(
            response.get("error").and_then(|v| v.as_str()),
            Some("payload_too_large")
        );
        assert_eq!(
            response.get("max_size").and_then(|v| v.as_int()),
            Some(crate::event::MAX_PROBE_PAYLOAD_SIZE as i64)
        );
    }

    #[test]
    fn send_probe_rpc_max_size_reaches_driver() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((QueryRequest::SendProbe { payload_size, .. }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(payload_size, crate::event::MAX_PROBE_PAYLOAD_SIZE);
                let _ = resp_tx.send(QueryResponse::SendProbe(Ok(([0xEE; 32], 2))));
            }
        });

        let request = PickleValue::Dict(vec![
            (
                PickleValue::String("send_probe".into()),
                PickleValue::Bytes(vec![0xDD; 16]),
            ),
            (
                PickleValue::String("size".into()),
                PickleValue::Int(crate::event::MAX_PROBE_PAYLOAD_SIZE as i64),
            ),
        ]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response.get("hops").unwrap().as_int().unwrap(), 2);
        driver.join().unwrap();
    }

    #[test]
    fn send_probe_rpc_driver_payload_too_large_is_reported() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((QueryRequest::SendProbe { .. }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                let _ = resp_tx.send(QueryResponse::SendProbe(Err(
                    crate::event::SendProbeError::PayloadTooLarge { max: 123 },
                )));
            }
        });

        let request = PickleValue::Dict(vec![(
            PickleValue::String("send_probe".into()),
            PickleValue::Bytes(vec![0xDD; 16]),
        )]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(
            response.get("error").and_then(|v| v.as_str()),
            Some("payload_too_large")
        );
        assert_eq!(response.get("max_size").and_then(|v| v.as_int()), Some(123));
        driver.join().unwrap();
    }

    #[test]
    fn check_proof_rpc_not_found() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((QueryRequest::CheckProof { packet_hash }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(packet_hash, [0xEE; 32]);
                let _ = resp_tx.send(QueryResponse::CheckProof(None));
            }
        });

        let request = PickleValue::Dict(vec![(
            PickleValue::String("check_proof".into()),
            PickleValue::Bytes(vec![0xEE; 32]),
        )]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::None);
        driver.join().unwrap();
    }

    #[test]
    fn check_proof_rpc_found() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((QueryRequest::CheckProof { packet_hash }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(packet_hash, [0xFF; 32]);
                let _ = resp_tx.send(QueryResponse::CheckProof(Some(0.352)));
            }
        });

        let request = PickleValue::Dict(vec![(
            PickleValue::String("check_proof".into()),
            PickleValue::Bytes(vec![0xFF; 32]),
        )]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        if let PickleValue::Float(rtt) = response {
            assert!((rtt - 0.352).abs() < 0.001);
        } else {
            panic!("Expected Float, got {:?}", response);
        }
        driver.join().unwrap();
    }

    #[test]
    fn request_path_rpc() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver =
            thread::spawn(
                move || match event_rx.recv_timeout(std::time::Duration::from_secs(5)) {
                    Ok(Event::RequestPath { dest_hash }) => {
                        assert_eq!(dest_hash, [0x11; 16]);
                    }
                    other => panic!("Expected RequestPath event, got {:?}", other),
                },
            );

        let request = PickleValue::Dict(vec![(
            PickleValue::String("request_path".into()),
            PickleValue::Bytes(vec![0x11; 16]),
        )]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::Bool(true));
        driver.join().unwrap();
    }

    #[test]
    fn begin_drain_rpc_emits_event() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(
            move || match event_rx.recv_timeout(Duration::from_secs(5)) {
                Ok(Event::BeginDrain { timeout }) => {
                    assert!((timeout.as_secs_f64() - 1.5).abs() < 0.001);
                }
                other => panic!("Expected BeginDrain event, got {:?}", other),
            },
        );

        let request = PickleValue::Dict(vec![(
            PickleValue::String("begin_drain".into()),
            PickleValue::Float(1.5),
        )]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response, PickleValue::Bool(true));
        driver.join().unwrap();
    }

    #[test]
    fn drain_status_rpc_roundtrips_fields() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((QueryRequest::DrainStatus, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                let _ = resp_tx.send(QueryResponse::DrainStatus(DrainStatus {
                    state: LifecycleState::Draining,
                    drain_age_seconds: Some(0.75),
                    deadline_remaining_seconds: Some(2.25),
                    drain_complete: false,
                    interface_writer_queued_frames: 3,
                    provider_backlog_events: 4,
                    provider_consumer_queued_events: 5,
                    detail: Some("node is draining existing work".into()),
                }));
            }
        });

        let request = PickleValue::Dict(vec![(
            PickleValue::String("get".into()),
            PickleValue::String("drain_status".into()),
        )]);

        let response = handle_rpc_request(&request, &event_tx).unwrap();
        assert_eq!(response.get("state").unwrap().as_str(), Some("draining"));
        assert_eq!(
            response.get("drain_complete").unwrap().as_bool(),
            Some(false)
        );
        assert_eq!(
            response
                .get("deadline_remaining_seconds")
                .unwrap()
                .as_float(),
            Some(2.25)
        );
        assert_eq!(
            response
                .get("interface_writer_queued_frames")
                .unwrap()
                .as_int(),
            Some(3)
        );
        assert_eq!(
            response.get("provider_backlog_events").unwrap().as_int(),
            Some(4)
        );
        assert_eq!(
            response
                .get("provider_consumer_queued_events")
                .unwrap()
                .as_int(),
            Some(5)
        );
        assert_eq!(
            response.get("detail").unwrap().as_str(),
            Some("node is draining existing work")
        );
        driver.join().unwrap();
    }

    #[test]
    fn interface_stats_with_probe_responder() {
        let probe_hash = [0x42; 16];
        let stats = InterfaceStatsResponse {
            interfaces: vec![],
            transport_id: None,
            transport_enabled: true,
            transport_uptime: 100.0,
            total_rxb: 0,
            total_txb: 0,
            rx_pps: 0.0,
            tx_pps: 0.0,
            traffic: TrafficDetail::default(),
            probe_responder: Some(probe_hash),
            backbone_peer_pool: None,
        };

        let pickle = interface_stats_to_pickle(&stats);
        let encoded = pickle::encode(&pickle);
        let decoded = pickle::decode(&encoded).unwrap();

        let pr = decoded.get("probe_responder").unwrap().as_bytes().unwrap();
        assert_eq!(pr, &probe_hash);
    }

    #[test]
    fn runtime_config_get_and_set_rpc() {
        let (event_tx, event_rx) = crate::event::channel();

        let driver = thread::spawn(move || {
            if let Some((QueryRequest::GetRuntimeConfig { key }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(key, "global.tick_interval_ms");
                let _ = resp_tx.send(QueryResponse::RuntimeConfigEntry(Some(
                    RuntimeConfigEntry {
                        key,
                        value: RuntimeConfigValue::Int(1000),
                        default: RuntimeConfigValue::Int(1000),
                        source: RuntimeConfigSource::Startup,
                        apply_mode: RuntimeConfigApplyMode::Immediate,
                        description: Some("tick".into()),
                    },
                )));
            } else {
                panic!("expected GetRuntimeConfig query");
            }

            if let Some((QueryRequest::SetRuntimeConfig { key, value }, resp_tx)) = event_rx
                .recv()
                .ok()
                .and_then(|event| event.into_query().ok())
            {
                assert_eq!(key, "global.tick_interval_ms");
                assert_eq!(value, RuntimeConfigValue::Int(250));
                let _ = resp_tx.send(QueryResponse::RuntimeConfigSet(Ok(RuntimeConfigEntry {
                    key,
                    value: RuntimeConfigValue::Int(250),
                    default: RuntimeConfigValue::Int(1000),
                    source: RuntimeConfigSource::RuntimeOverride,
                    apply_mode: RuntimeConfigApplyMode::Immediate,
                    description: Some("tick".into()),
                })));
            } else {
                panic!("expected SetRuntimeConfig query");
            }
        });

        let get_request = PickleValue::Dict(vec![
            (
                PickleValue::String("get".into()),
                PickleValue::String("runtime_config_entry".into()),
            ),
            (
                PickleValue::String("key".into()),
                PickleValue::String("global.tick_interval_ms".into()),
            ),
        ]);
        let get_response = handle_rpc_request(&get_request, &event_tx).unwrap();
        assert_eq!(
            get_response.get("key").and_then(|v| v.as_str()),
            Some("global.tick_interval_ms")
        );

        let set_request = PickleValue::Dict(vec![
            (
                PickleValue::String("set".into()),
                PickleValue::String("runtime_config".into()),
            ),
            (
                PickleValue::String("key".into()),
                PickleValue::String("global.tick_interval_ms".into()),
            ),
            (PickleValue::String("value".into()), PickleValue::Int(250)),
        ]);
        let set_response = handle_rpc_request(&set_request, &event_tx).unwrap();
        assert_eq!(
            set_response.get("value").and_then(|v| v.as_int()),
            Some(250)
        );

        driver.join().unwrap();
    }

    // Helper: create a connected TCP pair
    fn tcp_pair() -> (TcpStream, TcpStream) {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let client = TcpStream::connect(("127.0.0.1", port)).unwrap();
        let (server, _) = listener.accept().unwrap();
        client
            .set_read_timeout(Some(std::time::Duration::from_secs(5)))
            .unwrap();
        server
            .set_read_timeout(Some(std::time::Duration::from_secs(5)))
            .unwrap();
        (server, client)
    }
}