Add Unix domain control socket for runtime observability

Add a Unix domain socket interface for querying node state at runtime.
A spawned tokio task accepts connections and communicates with the main
event loop via mpsc/oneshot channels, keeping all Node access
single-threaded.

Includes:
- src/control/ module with socket lifecycle, JSON protocol, and 11
  query handlers (status, peers, links, tree, sessions, bloom, mmp,
  cache, connections, transports, routing)
- Separate fipsctl binary for CLI queries (fipsctl show <command>)
- ControlConfig in node configuration (enabled, socket_path)
- Integration into the main select! event loop

Cargo.lock

@@ -543,6 +543,7 @@ dependencies = [
  "rtnetlink",
  "secp256k1",
  "serde",
+ "serde_json",
  "serde_yaml",
  "sha2",
  "simple-dns",

Cargo.toml

@@ -17,6 +17,7 @@ rand = "0.8"
 thiserror = "2.0"
 bech32 = "0.11"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
 serde_yaml = "0.9"
 dirs = "6.0"
 hex = "0.4"
@@ -35,6 +36,10 @@ socket2 = { version = "0.5", features = ["all"] }
 tempfile = "3.15"
 criterion = { version = "0.5", features = ["html_reports"] }
 
+[[bin]]
+name = "fipsctl"
+path = "src/bin/fipsctl.rs"
+
 [[bench]]
 name = "bloom"
 harness = false

src/bin/fipsctl.rs

@@ -0,0 +1,165 @@
+//! fipsctl — FIPS control client
+//!
+//! Connects to the FIPS daemon's Unix domain control socket, sends a
+//! query command, and pretty-prints the JSON response.
+
+use clap::{Parser, Subcommand};
+use std::io::{BufRead, BufReader, Write};
+use std::os::unix::net::UnixStream;
+use std::path::PathBuf;
+use std::time::Duration;
+
+/// FIPS control client
+#[derive(Parser, Debug)]
+#[command(name = "fipsctl", version, about = "Query a running FIPS daemon")]
+struct Cli {
+    /// Control socket path override
+    #[arg(short = 's', long)]
+    socket: Option<PathBuf>,
+
+    #[command(subcommand)]
+    command: Commands,
+}
+
+#[derive(Subcommand, Debug)]
+enum Commands {
+    /// Show node information
+    Show {
+        #[command(subcommand)]
+        what: ShowCommands,
+    },
+}
+
+#[derive(Subcommand, Debug)]
+enum ShowCommands {
+    /// Node status overview
+    Status,
+    /// Authenticated peers
+    Peers,
+    /// Active links
+    Links,
+    /// Spanning tree state
+    Tree,
+    /// End-to-end sessions
+    Sessions,
+    /// Bloom filter state
+    Bloom,
+    /// MMP metrics summary
+    Mmp,
+    /// Coordinate cache stats
+    Cache,
+    /// Pending handshake connections
+    Connections,
+    /// Transport instances
+    Transports,
+    /// Routing table summary
+    Routing,
+}
+
+impl ShowCommands {
+    fn command_name(&self) -> &'static str {
+        match self {
+            ShowCommands::Status => "show_status",
+            ShowCommands::Peers => "show_peers",
+            ShowCommands::Links => "show_links",
+            ShowCommands::Tree => "show_tree",
+            ShowCommands::Sessions => "show_sessions",
+            ShowCommands::Bloom => "show_bloom",
+            ShowCommands::Mmp => "show_mmp",
+            ShowCommands::Cache => "show_cache",
+            ShowCommands::Connections => "show_connections",
+            ShowCommands::Transports => "show_transports",
+            ShowCommands::Routing => "show_routing",
+        }
+    }
+}
+
+/// Determine the default socket path.
+fn default_socket_path() -> PathBuf {
+    if let Ok(runtime_dir) = std::env::var("XDG_RUNTIME_DIR") {
+        PathBuf::from(format!("{}/fips/control.sock", runtime_dir))
+    } else {
+        PathBuf::from("/tmp/fips-control.sock")
+    }
+}
+
+fn main() {
+    let cli = Cli::parse();
+
+    let socket_path = cli.socket.unwrap_or_else(default_socket_path);
+    let command_name = match &cli.command {
+        Commands::Show { what } => what.command_name(),
+    };
+
+    // Connect to the control socket
+    let mut stream = match UnixStream::connect(&socket_path) {
+        Ok(s) => s,
+        Err(e) => {
+            eprintln!(
+                "error: cannot connect to {}: {}",
+                socket_path.display(),
+                e
+            );
+            eprintln!("Is the FIPS daemon running?");
+            std::process::exit(1);
+        }
+    };
+
+    // Set timeouts
+    let timeout = Duration::from_secs(2);
+    let _ = stream.set_read_timeout(Some(timeout));
+    let _ = stream.set_write_timeout(Some(timeout));
+
+    // Send request
+    let request = format!("{{\"command\":\"{}\"}}\n", command_name);
+    if let Err(e) = stream.write_all(request.as_bytes()) {
+        eprintln!("error: failed to send request: {}", e);
+        std::process::exit(1);
+    }
+
+    // Shutdown write half to signal end of request
+    let _ = stream.shutdown(std::net::Shutdown::Write);
+
+    // Read response
+    let reader = BufReader::new(&stream);
+    let response_line = match reader.lines().next() {
+        Some(Ok(line)) => line,
+        Some(Err(e)) => {
+            eprintln!("error: failed to read response: {}", e);
+            std::process::exit(1);
+        }
+        None => {
+            eprintln!("error: no response from daemon");
+            std::process::exit(1);
+        }
+    };
+
+    // Parse and pretty-print
+    let Ok(value) = serde_json::from_str::<serde_json::Value>(&response_line) else {
+        // Not JSON, print raw
+        println!("{}", response_line);
+        return;
+    };
+
+    let status = value
+        .get("status")
+        .and_then(|v| v.as_str())
+        .unwrap_or("unknown");
+
+    if status == "error" {
+        let msg = value
+            .get("message")
+            .and_then(|v| v.as_str())
+            .unwrap_or("unknown error");
+        eprintln!("error: {}", msg);
+        std::process::exit(1);
+    }
+
+    // Pretty-print the data field (or whole response if no data field)
+    let output = if let Some(data) = value.get("data") {
+        serde_json::to_string_pretty(data).unwrap_or(response_line)
+    } else {
+        serde_json::to_string_pretty(&value).unwrap_or(response_line)
+    };
+    println!("{}", output);
+}

src/config/mod.rs

@@ -29,8 +29,8 @@ use std::path::{Path, PathBuf};
 use thiserror::Error;
 
 pub use node::{
-    BloomConfig, BuffersConfig, CacheConfig, DiscoveryConfig, LimitsConfig, NodeConfig,
-    RateLimitConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig,
+    BloomConfig, BuffersConfig, CacheConfig, ControlConfig, DiscoveryConfig, LimitsConfig,
+    NodeConfig, RateLimitConfig, RetryConfig, SessionConfig, SessionMmpConfig, TreeConfig,
 };
 pub use peer::{ConnectPolicy, PeerAddress, PeerConfig};
 pub use transport::{TransportInstances, TransportsConfig, UdpConfig};

src/config/node.rs

@@ -314,6 +314,38 @@ impl SessionMmpConfig {
     fn default_owd_window_size() -> usize { DEFAULT_OWD_WINDOW_SIZE }
 }
 
+/// Control socket configuration (`node.control.*`).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ControlConfig {
+    /// Enable the control socket (`node.control.enabled`).
+    #[serde(default = "ControlConfig::default_enabled")]
+    pub enabled: bool,
+    /// Unix socket path (`node.control.socket_path`).
+    #[serde(default = "ControlConfig::default_socket_path")]
+    pub socket_path: String,
+}
+
+impl Default for ControlConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            socket_path: Self::default_socket_path(),
+        }
+    }
+}
+
+impl ControlConfig {
+    fn default_enabled() -> bool { true }
+
+    fn default_socket_path() -> String {
+        if let Ok(runtime_dir) = std::env::var("XDG_RUNTIME_DIR") {
+            format!("{}/fips/control.sock", runtime_dir)
+        } else {
+            "/tmp/fips-control.sock".to_string()
+        }
+    }
+}
+
 /// Internal buffers (`node.buffers.*`).
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BuffersConfig {
@@ -412,6 +444,10 @@ pub struct NodeConfig {
     #[serde(default)]
     pub buffers: BuffersConfig,
 
+    /// Control socket (`node.control.*`).
+    #[serde(default)]
+    pub control: ControlConfig,
+
     /// Metrics Measurement Protocol — link layer (`node.mmp.*`).
     #[serde(default)]
     pub mmp: MmpConfig,
@@ -439,6 +475,7 @@ impl Default for NodeConfig {
             bloom: BloomConfig::default(),
             session: SessionConfig::default(),
             buffers: BuffersConfig::default(),
+            control: ControlConfig::default(),
             mmp: MmpConfig::default(),
             session_mmp: SessionMmpConfig::default(),
         }

src/control/mod.rs

@@ -0,0 +1,216 @@
+//! Control socket for runtime observability.
+//!
+//! Provides a Unix domain socket that accepts query commands and returns
+//! structured JSON data about the node's current state. Read-only queries
+//! only — no state mutation through this channel.
+
+pub mod protocol;
+pub mod queries;
+
+use crate::config::ControlConfig;
+use protocol::{Request, Response};
+use std::path::{Path, PathBuf};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::UnixListener;
+use tokio::sync::{mpsc, oneshot};
+use tracing::{debug, info, warn};
+
+/// Maximum request size in bytes (4 KB).
+const MAX_REQUEST_SIZE: usize = 4096;
+
+/// I/O timeout for client connections.
+const IO_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
+
+/// A message sent from the accept loop to the main event loop.
+pub type ControlMessage = (Request, oneshot::Sender<Response>);
+
+/// Control socket listener.
+///
+/// Manages the Unix domain socket lifecycle: bind, accept, cleanup.
+pub struct ControlSocket {
+    listener: UnixListener,
+    socket_path: PathBuf,
+}
+
+impl ControlSocket {
+    /// Bind a new control socket.
+    ///
+    /// Creates parent directories if needed, removes stale socket files,
+    /// and binds the Unix listener.
+    pub fn bind(config: &ControlConfig) -> Result<Self, std::io::Error> {
+        let socket_path = PathBuf::from(&config.socket_path);
+
+        // Create parent directory if it doesn't exist
+        if let Some(parent) = socket_path.parent()
+            && !parent.exists()
+        {
+            std::fs::create_dir_all(parent)?;
+            debug!(path = %parent.display(), "Created control socket directory");
+        }
+
+        // Remove stale socket if it exists
+        if socket_path.exists() {
+            Self::remove_stale_socket(&socket_path)?;
+        }
+
+        let listener = UnixListener::bind(&socket_path)?;
+        info!(path = %socket_path.display(), "Control socket listening");
+
+        Ok(Self {
+            listener,
+            socket_path,
+        })
+    }
+
+    /// Remove a stale socket file.
+    ///
+    /// If the file exists but no one is listening, remove it so we can
+    /// bind. This handles unclean daemon exits.
+    fn remove_stale_socket(path: &Path) -> Result<(), std::io::Error> {
+        // Try connecting to see if someone is listening
+        match std::os::unix::net::UnixStream::connect(path) {
+            Ok(_) => {
+                // Someone is listening — don't remove it
+                Err(std::io::Error::new(
+                    std::io::ErrorKind::AddrInUse,
+                    format!("control socket already in use: {}", path.display()),
+                ))
+            }
+            Err(_) => {
+                // No one listening — remove the stale socket
+                debug!(path = %path.display(), "Removing stale control socket");
+                std::fs::remove_file(path)?;
+                Ok(())
+            }
+        }
+    }
+
+    /// Run the accept loop, forwarding requests to the main event loop via mpsc.
+    ///
+    /// Each accepted connection is handled in a spawned task:
+    /// 1. Read one line of JSON (the request)
+    /// 2. Send (Request, oneshot::Sender) to the main loop
+    /// 3. Wait for the response via oneshot
+    /// 4. Write the response as one line of JSON
+    /// 5. Close the connection
+    pub async fn accept_loop(self, control_tx: mpsc::Sender<ControlMessage>) {
+        loop {
+            let (stream, _addr) = match self.listener.accept().await {
+                Ok(conn) => conn,
+                Err(e) => {
+                    warn!(error = %e, "Control socket accept failed");
+                    continue;
+                }
+            };
+
+            let tx = control_tx.clone();
+            tokio::spawn(async move {
+                if let Err(e) = Self::handle_connection(stream, tx).await {
+                    debug!(error = %e, "Control connection error");
+                }
+            });
+        }
+    }
+
+    /// Handle a single client connection.
+    async fn handle_connection(
+        stream: tokio::net::UnixStream,
+        control_tx: mpsc::Sender<ControlMessage>,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let (reader, mut writer) = stream.into_split();
+        let mut buf_reader = BufReader::new(reader);
+        let mut line = String::new();
+
+        // Read one line with timeout and size limit
+        let read_result = tokio::time::timeout(IO_TIMEOUT, async {
+            let mut total = 0usize;
+            loop {
+                let n = buf_reader.read_line(&mut line).await?;
+                if n == 0 {
+                    break; // EOF
+                }
+                total += n;
+                if total > MAX_REQUEST_SIZE {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::InvalidData,
+                        "request too large",
+                    ));
+                }
+                if line.ends_with('\n') {
+                    break;
+                }
+            }
+            Ok(())
+        })
+        .await;
+
+        let response = match read_result {
+            Ok(Ok(())) if line.is_empty() => {
+                Response::error("empty request")
+            }
+            Ok(Ok(())) => {
+                // Parse the request
+                match serde_json::from_str::<Request>(line.trim()) {
+                    Ok(request) => {
+                        // Send to main loop and wait for response
+                        let (resp_tx, resp_rx) = oneshot::channel();
+                        if control_tx.send((request, resp_tx)).await.is_err() {
+                            Response::error("node shutting down")
+                        } else {
+                            match tokio::time::timeout(IO_TIMEOUT, resp_rx).await {
+                                Ok(Ok(resp)) => resp,
+                                Ok(Err(_)) => Response::error("response channel closed"),
+                                Err(_) => Response::error("query timeout"),
+                            }
+                        }
+                    }
+                    Err(e) => Response::error(format!("invalid request: {}", e)),
+                }
+            }
+            Ok(Err(e)) => Response::error(format!("read error: {}", e)),
+            Err(_) => Response::error("read timeout"),
+        };
+
+        // Write response with timeout
+        let json = serde_json::to_string(&response)?;
+        let write_result = tokio::time::timeout(IO_TIMEOUT, async {
+            writer.write_all(json.as_bytes()).await?;
+            writer.write_all(b"\n").await?;
+            writer.shutdown().await?;
+            Ok::<_, std::io::Error>(())
+        })
+        .await;
+
+        if let Err(_) | Ok(Err(_)) = write_result {
+            debug!("Control socket write failed or timed out");
+        }
+
+        Ok(())
+    }
+
+    /// Get the socket path.
+    pub fn socket_path(&self) -> &Path {
+        &self.socket_path
+    }
+
+    /// Clean up the socket file.
+    fn cleanup(&self) {
+        if self.socket_path.exists() {
+            if let Err(e) = std::fs::remove_file(&self.socket_path) {
+                warn!(
+                    path = %self.socket_path.display(),
+                    error = %e,
+                    "Failed to remove control socket"
+                );
+            } else {
+                debug!(path = %self.socket_path.display(), "Control socket removed");
+            }
+        }
+    }
+}
+
+impl Drop for ControlSocket {
+    fn drop(&mut self) {
+        self.cleanup();
+    }
+}

src/control/protocol.rs

@@ -0,0 +1,90 @@
+//! Control socket protocol types.
+//!
+//! Line-delimited JSON protocol for the Unix domain socket.
+//! Each request is one JSON line, each response is one JSON line.
+
+use serde::{Deserialize, Serialize};
+
+/// A control request from a client.
+#[derive(Debug, Deserialize)]
+pub struct Request {
+    /// The command to execute (e.g., "show_status", "show_peers").
+    pub command: String,
+}
+
+/// A control response to a client.
+#[derive(Debug, Serialize)]
+pub struct Response {
+    /// "ok" or "error".
+    pub status: String,
+    /// Response data (present on success).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub data: Option<serde_json::Value>,
+    /// Error message (present on failure).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub message: Option<String>,
+}
+
+impl Response {
+    /// Create a success response with data.
+    pub fn ok(data: serde_json::Value) -> Self {
+        Self {
+            status: "ok".to_string(),
+            data: Some(data),
+            message: None,
+        }
+    }
+
+    /// Create an error response with a message.
+    pub fn error(message: impl Into<String>) -> Self {
+        Self {
+            status: "error".to_string(),
+            data: None,
+            message: Some(message.into()),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_deserialize_request() {
+        let json = r#"{"command": "show_status"}"#;
+        let req: Request = serde_json::from_str(json).unwrap();
+        assert_eq!(req.command, "show_status");
+    }
+
+    #[test]
+    fn test_serialize_ok_response() {
+        let resp = Response::ok(serde_json::json!({"state": "running"}));
+        let json = serde_json::to_string(&resp).unwrap();
+        assert!(json.contains("\"status\":\"ok\""));
+        assert!(json.contains("\"state\":\"running\""));
+        assert!(!json.contains("\"message\""));
+    }
+
+    #[test]
+    fn test_serialize_error_response() {
+        let resp = Response::error("unknown command: foo");
+        let json = serde_json::to_string(&resp).unwrap();
+        assert!(json.contains("\"status\":\"error\""));
+        assert!(json.contains("unknown command: foo"));
+        assert!(!json.contains("\"data\""));
+    }
+
+    #[test]
+    fn test_deserialize_unknown_fields_ignored() {
+        let json = r#"{"command": "show_peers", "extra": true}"#;
+        let req: Request = serde_json::from_str(json).unwrap();
+        assert_eq!(req.command, "show_peers");
+    }
+
+    #[test]
+    fn test_deserialize_malformed_request() {
+        let json = r#"{"not_command": "foo"}"#;
+        let result: Result<Request, _> = serde_json::from_str(json);
+        assert!(result.is_err());
+    }
+}

src/control/queries.rs

@@ -0,0 +1,414 @@
+//! Control query implementations.
+//!
+//! Each function takes `&Node` and returns a `serde_json::Value`.
+//! Query logic is kept separate from socket handling.
+
+use crate::node::Node;
+use serde_json::{json, Value};
+
+/// Helper: get current Unix time in milliseconds.
+fn now_ms() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+/// Classify a DualEwma trend as "rising", "falling", or "stable".
+fn trend_label(short: f64, long: f64) -> &'static str {
+    if !short.is_finite() || !long.is_finite() || long == 0.0 {
+        return "stable";
+    }
+    let ratio = short / long;
+    if ratio > 1.05 {
+        "rising"
+    } else if ratio < 0.95 {
+        "falling"
+    } else {
+        "stable"
+    }
+}
+
+/// `show_status` — Node overview.
+pub fn show_status(node: &Node) -> Value {
+    json!({
+        "npub": node.npub(),
+        "node_addr": hex::encode(node.node_addr().as_bytes()),
+        "fips_address": format!("{}", node.identity().address()),
+        "state": format!("{}", node.state()),
+        "is_leaf_only": node.is_leaf_only(),
+        "peer_count": node.peer_count(),
+        "session_count": node.session_count(),
+        "link_count": node.link_count(),
+        "transport_count": node.transport_count(),
+        "connection_count": node.connection_count(),
+        "tun_state": format!("{}", node.tun_state()),
+        "effective_ipv6_mtu": node.effective_ipv6_mtu(),
+    })
+}
+
+/// `show_peers` — Authenticated peers.
+pub fn show_peers(node: &Node) -> Value {
+    let peers: Vec<Value> = node.peers().map(|peer| {
+        let node_addr = *peer.node_addr();
+        let addr_hex = hex::encode(node_addr.as_bytes());
+
+        let mut peer_json = json!({
+            "node_addr": addr_hex,
+            "npub": peer.npub(),
+            "display_name": node.peer_display_name(&node_addr),
+            "fips_address": format!("{}", peer.address()),
+            "connectivity": format!("{}", peer.connectivity()),
+            "link_id": peer.link_id().as_u64(),
+            "authenticated_at_ms": peer.authenticated_at(),
+            "last_seen_ms": peer.last_seen(),
+            "has_tree_position": peer.has_tree_position(),
+            "has_bloom_filter": peer.filter_sequence() > 0,
+            "filter_sequence": peer.filter_sequence(),
+        });
+
+        // Add transport address if available
+        if let Some(addr) = peer.current_addr() {
+            peer_json["transport_addr"] = json!(format!("{}", addr));
+        }
+
+        // Add tree depth if available
+        if let Some(coords) = peer.coords() {
+            peer_json["tree_depth"] = json!(coords.depth());
+        }
+
+        // Add link stats
+        let stats = peer.link_stats();
+        peer_json["stats"] = json!({
+            "packets_sent": stats.packets_sent,
+            "packets_recv": stats.packets_recv,
+            "bytes_sent": stats.bytes_sent,
+            "bytes_recv": stats.bytes_recv,
+        });
+
+        // Add MMP metrics if available
+        if let Some(mmp) = peer.mmp() {
+            let mut mmp_json = json!({
+                "mode": format!("{}", mmp.mode()),
+            });
+            if let Some(srtt) = mmp.metrics.srtt_ms() {
+                mmp_json["srtt_ms"] = json!(srtt);
+            }
+            mmp_json["loss_rate"] = json!(mmp.metrics.loss_rate());
+            mmp_json["etx"] = json!(mmp.metrics.etx);
+            mmp_json["goodput_bps"] = json!(mmp.metrics.goodput_bps);
+            mmp_json["delivery_ratio_forward"] = json!(mmp.metrics.delivery_ratio_forward);
+            mmp_json["delivery_ratio_reverse"] = json!(mmp.metrics.delivery_ratio_reverse);
+            peer_json["mmp"] = mmp_json;
+        }
+
+        peer_json
+    }).collect();
+
+    json!({ "peers": peers })
+}
+
+/// `show_links` — Active links.
+pub fn show_links(node: &Node) -> Value {
+    let links: Vec<Value> = node.links().map(|link| {
+        let stats = link.stats();
+        json!({
+            "link_id": link.link_id().as_u64(),
+            "transport_id": link.transport_id().as_u32(),
+            "remote_addr": format!("{}", link.remote_addr()),
+            "direction": format!("{}", link.direction()),
+            "state": format!("{}", link.state()),
+            "created_at_ms": link.created_at(),
+            "stats": {
+                "packets_sent": stats.packets_sent,
+                "packets_recv": stats.packets_recv,
+                "bytes_sent": stats.bytes_sent,
+                "bytes_recv": stats.bytes_recv,
+                "last_recv_ms": stats.last_recv_ms,
+            },
+        })
+    }).collect();
+
+    json!({ "links": links })
+}
+
+/// `show_tree` — Spanning tree state.
+pub fn show_tree(node: &Node) -> Value {
+    let tree = node.tree_state();
+    let my_coords = tree.my_coords();
+    let decl = tree.my_declaration();
+
+    // Build coords array as hex strings
+    let coords: Vec<String> = my_coords.entries()
+        .iter()
+        .map(|e| hex::encode(e.node_addr.as_bytes()))
+        .collect();
+
+    // Build peer tree data
+    let peers: Vec<Value> = tree.peer_ids().map(|peer_id| {
+        let mut peer_json = json!({
+            "node_addr": hex::encode(peer_id.as_bytes()),
+            "display_name": node.peer_display_name(peer_id),
+        });
+        if let Some(coords) = tree.peer_coords(peer_id) {
+            peer_json["depth"] = json!(coords.depth());
+            peer_json["root"] = json!(hex::encode(coords.root_id().as_bytes()));
+            peer_json["distance_to_us"] = json!(my_coords.distance_to(coords));
+        }
+        peer_json
+    }).collect();
+
+    // Determine parent display name
+    let parent_addr = my_coords.parent_id();
+    let parent_hex = hex::encode(parent_addr.as_bytes());
+    let parent_display = node.peer_display_name(parent_addr);
+
+    json!({
+        "my_node_addr": hex::encode(tree.my_node_addr().as_bytes()),
+        "root": hex::encode(tree.root().as_bytes()),
+        "is_root": tree.is_root(),
+        "depth": my_coords.depth(),
+        "my_coords": coords,
+        "parent": parent_hex,
+        "parent_display_name": parent_display,
+        "declaration_sequence": decl.sequence(),
+        "declaration_signed": decl.is_signed(),
+        "peer_tree_count": tree.peer_count(),
+        "peers": peers,
+    })
+}
+
+/// `show_sessions` — End-to-end sessions.
+pub fn show_sessions(node: &Node) -> Value {
+    let sessions: Vec<Value> = node.session_entries().map(|(addr, entry)| {
+        let state_str = if entry.is_established() {
+            "established"
+        } else if entry.is_initiating() {
+            "initiating"
+        } else if entry.is_responding() {
+            "responding"
+        } else {
+            "unknown"
+        };
+
+        let mut session_json = json!({
+            "remote_addr": hex::encode(addr.as_bytes()),
+            "display_name": node.peer_display_name(addr),
+            "state": state_str,
+            "is_initiator": entry.is_initiator(),
+            "last_activity_ms": entry.last_activity(),
+        });
+
+        // Add session MMP if available
+        if let Some(mmp) = entry.mmp() {
+            let mut mmp_json = json!({
+                "mode": format!("{}", mmp.mode()),
+            });
+            if let Some(srtt) = mmp.metrics.srtt_ms() {
+                mmp_json["srtt_ms"] = json!(srtt);
+            }
+            mmp_json["path_mtu"] = json!(mmp.path_mtu.current_mtu());
+            session_json["mmp"] = mmp_json;
+        }
+
+        session_json
+    }).collect();
+
+    json!({ "sessions": sessions })
+}
+
+/// `show_bloom` — Bloom filter state.
+pub fn show_bloom(node: &Node) -> Value {
+    let bloom = node.bloom_state();
+
+    let leaf_deps: Vec<String> = bloom.leaf_dependents()
+        .iter()
+        .map(|addr| hex::encode(addr.as_bytes()))
+        .collect();
+
+    // Build per-peer filter info
+    let peer_filters: Vec<Value> = node.peers().map(|peer| {
+        let addr = *peer.node_addr();
+        json!({
+            "peer": hex::encode(addr.as_bytes()),
+            "display_name": node.peer_display_name(&addr),
+            "has_filter": peer.filter_sequence() > 0,
+            "filter_sequence": peer.filter_sequence(),
+        })
+    }).collect();
+
+    json!({
+        "own_node_addr": hex::encode(node.node_addr().as_bytes()),
+        "is_leaf_only": node.is_leaf_only(),
+        "sequence": bloom.sequence(),
+        "leaf_dependent_count": bloom.leaf_dependents().len(),
+        "leaf_dependents": leaf_deps,
+        "peer_filters": peer_filters,
+    })
+}
+
+/// `show_mmp` — MMP metrics summary.
+pub fn show_mmp(node: &Node) -> Value {
+    // Link-layer MMP per peer
+    let peers: Vec<Value> = node.peers().filter_map(|peer| {
+        let mmp = peer.mmp()?;
+        let addr = *peer.node_addr();
+        let metrics = &mmp.metrics;
+
+        let mut link_layer = json!({
+            "loss_rate": metrics.loss_rate(),
+            "etx": metrics.etx,
+            "goodput_bps": metrics.goodput_bps,
+            "spin_bit_role": if mmp.spin_bit.is_initiator() { "initiator" } else { "responder" },
+        });
+
+        if let Some(srtt) = metrics.srtt_ms() {
+            link_layer["srtt_ms"] = json!(srtt);
+        }
+
+        // Trend indicators
+        if metrics.rtt_trend.initialized() {
+            link_layer["rtt_trend"] = json!(trend_label(metrics.rtt_trend.short(), metrics.rtt_trend.long()));
+        }
+        if metrics.loss_trend.initialized() {
+            link_layer["loss_trend"] = json!(trend_label(metrics.loss_trend.short(), metrics.loss_trend.long()));
+        }
+        if metrics.goodput_trend.initialized() {
+            link_layer["goodput_trend"] = json!(trend_label(metrics.goodput_trend.short(), metrics.goodput_trend.long()));
+        }
+        if metrics.jitter_trend.initialized() {
+            link_layer["jitter_trend"] = json!(trend_label(metrics.jitter_trend.short(), metrics.jitter_trend.long()));
+        }
+
+        link_layer["delivery_ratio_forward"] = json!(metrics.delivery_ratio_forward);
+        link_layer["delivery_ratio_reverse"] = json!(metrics.delivery_ratio_reverse);
+
+        Some(json!({
+            "peer": hex::encode(addr.as_bytes()),
+            "display_name": node.peer_display_name(&addr),
+            "mode": format!("{}", mmp.mode()),
+            "link_layer": link_layer,
+        }))
+    }).collect();
+
+    // Session-layer MMP
+    let sessions: Vec<Value> = node.session_entries().filter_map(|(addr, entry)| {
+        let mmp = entry.mmp()?;
+        let metrics = &mmp.metrics;
+
+        let mut session_layer = json!({
+            "loss_rate": metrics.loss_rate(),
+            "etx": metrics.etx,
+            "path_mtu": mmp.path_mtu.current_mtu(),
+        });
+
+        if let Some(srtt) = metrics.srtt_ms() {
+            session_layer["srtt_ms"] = json!(srtt);
+        }
+
+        Some(json!({
+            "remote": hex::encode(addr.as_bytes()),
+            "display_name": node.peer_display_name(addr),
+            "mode": format!("{}", mmp.mode()),
+            "session_layer": session_layer,
+        }))
+    }).collect();
+
+    json!({
+        "peers": peers,
+        "sessions": sessions,
+    })
+}
+
+/// `show_cache` — Coordinate cache stats.
+pub fn show_cache(node: &Node) -> Value {
+    let cache = node.coord_cache();
+    let stats = cache.stats(now_ms());
+
+    json!({
+        "entries": stats.entries,
+        "max_entries": stats.max_entries,
+        "fill_ratio": stats.fill_ratio(),
+        "default_ttl_ms": cache.default_ttl_ms(),
+        "expired": stats.expired,
+        "avg_age_ms": stats.avg_age_ms,
+    })
+}
+
+/// `show_connections` — Pending handshakes.
+pub fn show_connections(node: &Node) -> Value {
+    let now = now_ms();
+    let connections: Vec<Value> = node.connections().map(|conn| {
+        let mut conn_json = json!({
+            "link_id": conn.link_id().as_u64(),
+            "direction": format!("{}", conn.direction()),
+            "handshake_state": format!("{}", conn.handshake_state()),
+            "started_at_ms": conn.started_at(),
+            "idle_ms": now.saturating_sub(conn.last_activity()),
+            "resend_count": conn.resend_count(),
+        });
+
+        if let Some(identity) = conn.expected_identity() {
+            conn_json["expected_peer"] = json!(identity.npub());
+        }
+
+        conn_json
+    }).collect();
+
+    json!({ "connections": connections })
+}
+
+/// `show_transports` — Transport instances.
+pub fn show_transports(node: &Node) -> Value {
+    let transports: Vec<Value> = node.transport_ids().map(|id| {
+        let handle = node.get_transport(id).unwrap();
+        let mut t_json = json!({
+            "transport_id": id.as_u32(),
+            "type": handle.transport_type().name,
+            "state": format!("{}", handle.state()),
+            "mtu": handle.mtu(),
+        });
+
+        if let Some(name) = handle.name() {
+            t_json["name"] = json!(name);
+        }
+        if let Some(addr) = handle.local_addr() {
+            t_json["local_addr"] = json!(format!("{}", addr));
+        }
+
+        t_json
+    }).collect();
+
+    json!({ "transports": transports })
+}
+
+/// `show_routing` — Routing table summary.
+pub fn show_routing(node: &Node) -> Value {
+    let cache = node.coord_cache();
+    let stats = cache.stats(now_ms());
+
+    json!({
+        "coord_cache_entries": stats.entries,
+        "identity_cache_entries": node.identity_cache_len(),
+        "pending_lookups": node.pending_lookup_count(),
+        "recent_requests": node.recent_request_count(),
+    })
+}
+
+/// Dispatch a command string to the appropriate query function.
+pub fn dispatch(node: &Node, command: &str) -> super::protocol::Response {
+    match command {
+        "show_status" => super::protocol::Response::ok(show_status(node)),
+        "show_peers" => super::protocol::Response::ok(show_peers(node)),
+        "show_links" => super::protocol::Response::ok(show_links(node)),
+        "show_tree" => super::protocol::Response::ok(show_tree(node)),
+        "show_sessions" => super::protocol::Response::ok(show_sessions(node)),
+        "show_bloom" => super::protocol::Response::ok(show_bloom(node)),
+        "show_mmp" => super::protocol::Response::ok(show_mmp(node)),
+        "show_cache" => super::protocol::Response::ok(show_cache(node)),
+        "show_connections" => super::protocol::Response::ok(show_connections(node)),
+        "show_transports" => super::protocol::Response::ok(show_transports(node)),
+        "show_routing" => super::protocol::Response::ok(show_routing(node)),
+        _ => super::protocol::Response::error(format!("unknown command: {}", command)),
+    }
+}

src/lib.rs

@@ -6,6 +6,7 @@
 pub mod bloom;
 pub mod cache;
 pub mod config;
+pub mod control;
 pub mod identity;
 pub mod mmp;
 pub mod noise;

src/mmp/algorithms.rs

@@ -299,6 +299,11 @@ impl SpinBitState {
         }
     }
 
+    /// Check if this is the spin bit initiator.
+    pub fn is_initiator(&self) -> bool {
+        self.is_initiator
+    }
+
     /// Get the spin bit value to set on an outgoing frame.
     pub fn tx_bit(&self) -> bool {
         self.current_value

src/node/handlers/rx_loop.rs

@@ -1,10 +1,12 @@
 //! RX event loop and packet dispatch.
 
+use crate::control::ControlSocket;
+use crate::control::queries;
 use crate::node::{Node, NodeError};
 use crate::transport::ReceivedPacket;
 use crate::node::wire::{CommonPrefix, PHASE_ESTABLISHED, PHASE_MSG1, PHASE_MSG2, FMP_VERSION, COMMON_PREFIX_SIZE};
 use std::time::Duration;
-use tracing::{debug, info};
+use tracing::{debug, info, warn};
 
 impl Node {
     /// Run the receive event loop.
@@ -53,6 +55,28 @@ impl Node {
 
         let mut tick = tokio::time::interval(Duration::from_secs(self.config.node.tick_interval_secs));
 
+        // Set up control socket channel
+        let (control_tx, mut control_rx) = tokio::sync::mpsc::channel::<
+            crate::control::ControlMessage,
+        >(32);
+
+        if self.config.node.control.enabled {
+            let config = self.config.node.control.clone();
+            let tx = control_tx.clone();
+            tokio::spawn(async move {
+                match ControlSocket::bind(&config) {
+                    Ok(socket) => {
+                        socket.accept_loop(tx).await;
+                    }
+                    Err(e) => {
+                        warn!(error = %e, "Failed to bind control socket");
+                    }
+                }
+            });
+        }
+        // Drop unused sender to avoid keeping channel open if control is disabled
+        drop(control_tx);
+
         info!("RX event loop started");
 
         loop {
@@ -73,6 +97,10 @@ impl Node {
                     );
                     self.register_identity(identity.node_addr, identity.pubkey);
                 }
+                Some((request, response_tx)) = control_rx.recv() => {
+                    let response = queries::dispatch(self, &request.command);
+                    let _ = response_tx.send(response);
+                }
                 _ = tick.tick() => {
                     self.check_timeouts();
                     let now_ms = std::time::SystemTime::now()

src/node/mod.rs

@@ -919,6 +919,11 @@ impl Node {
         self.sessions.len()
     }
 
+    /// Iterate over all session entries (for control queries).
+    pub(crate) fn session_entries(&self) -> impl Iterator<Item = (&NodeAddr, &SessionEntry)> {
+        self.sessions.iter()
+    }
+
     // === Identity Cache ===
 
     /// Register a node in the identity cache for FipsAddress → NodeAddr lookup.
@@ -952,6 +957,16 @@ impl Node {
         self.identity_cache.len()
     }
 
+    /// Number of pending discovery lookups.
+    pub fn pending_lookup_count(&self) -> usize {
+        self.pending_lookups.len()
+    }
+
+    /// Number of recent discovery requests tracked.
+    pub fn recent_request_count(&self) -> usize {
+        self.recent_requests.len()
+    }
+
     // === Routing ===
 
     /// Find next hop for a destination node address.