feat(daemon): accept runtime commands and promote exit nodes to relay

MCP/REST/REPL users can now dynamically connect, listen, and disconnect
without restarting the daemon. A NodeCommand channel from Handler to the
mode task feeds a tokio::select! loop in auto mode.

When both --connect and --listen are specified, the node starts as exit
and promotes to relay when the listener accepts a second peer — multi-hop
topologies form organically without pre-declaring relay mode.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: maxholman

crates/core/src/control/handler.rs

@@ -6,7 +6,7 @@
 use std::{net::SocketAddr, sync::Arc, time::Instant};
 
 use arc_swap::ArcSwap;
-use tokio::sync::watch;
+use tokio::sync::{mpsc, watch};
 use wallhack_wire::{
     control::{
         ControlRequest, ControlResponse, ErrorResponse, PeerInfo, PingResponse, RouteInfo,
@@ -18,7 +18,11 @@ use wallhack_wire::{
 use crate::NodeRole;
 
 use super::{
-    log_buffer::LogBuffer, metrics::SharedMetrics, peers::SharedRegistry, routes::SharedRouteTable,
+    log_buffer::LogBuffer,
+    metrics::SharedMetrics,
+    node_command::{NodeCommand, reply_channel},
+    peers::SharedRegistry,
+    routes::SharedRouteTable,
 };
 
 /// Mutable runtime state that can change after construction.
@@ -122,6 +126,14 @@ pub struct Handler {
     route_updates: tokio::sync::broadcast::Sender<super::routes::RouteUpdate>,
     state: SharedNodeState,
     start_time: Instant,
+    /// Sender used by `connect/listen/disconnect` to dispatch commands to the
+    /// active mode task. `None` if the channel has been dropped (mode does not
+    /// support dynamic operations).
+    cmd_tx: mpsc::Sender<NodeCommand>,
+    /// Receiver extracted once by the mode task before it starts running.
+    /// Wrapped in `Mutex<Option<…>>` so it can be moved out without requiring
+    /// `&mut self`.
+    cmd_rx: std::sync::Mutex<Option<mpsc::Receiver<NodeCommand>>>,
 }
 
 impl Handler {
@@ -141,6 +153,7 @@ impl Handler {
     ) -> Self {
         let state = SharedNodeState::new(config.node_role);
         let (hint_tx, _) = watch::channel(None);
+        let (cmd_tx, cmd_rx) = mpsc::channel(8);
         Self {
             config,
             hint_tx,
@@ -151,6 +164,8 @@ impl Handler {
             route_updates,
             state,
             start_time: Instant::now(),
+            cmd_tx,
+            cmd_rx: std::sync::Mutex::new(Some(cmd_rx)),
         }
     }
 
@@ -170,6 +185,19 @@ impl Handler {
         self.hint_tx.subscribe()
     }
 
+    /// Extracts the [`NodeCommand`] receiver so the mode task can poll it.
+    ///
+    /// May only be called once. Returns `None` if already taken.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the internal mutex is poisoned (indicates a prior panic in
+    /// another thread while holding the lock, which should never happen in
+    /// normal operation).
+    pub fn take_cmd_rx(&self) -> Option<mpsc::Receiver<NodeCommand>> {
+        self.cmd_rx.lock().expect("cmd_rx mutex poisoned").take()
+    }
+
     /// Handles a control request and returns a response.
     ///
     /// # Errors
@@ -419,25 +447,70 @@ impl crate::node_api::NodeApi for Handler {
         }
     }
 
-    fn connect(&self, _addr: &str) -> crate::node_api::Result<crate::node_api::ConnectInfo> {
-        Err(crate::node_api::NodeApiError::NotSupported(
-            "dynamic connect not yet implemented — specify --connect at startup".into(),
-        ))
+    fn connect(&self, addr: &str) -> crate::node_api::Result<crate::node_api::ConnectInfo> {
+        let (reply_tx, reply_rx) = reply_channel();
+        self.cmd_tx
+            .try_send(NodeCommand::Connect {
+                addr: addr.to_string(),
+                reply: reply_tx,
+            })
+            .map_err(|err| match err {
+                tokio::sync::mpsc::error::TrySendError::Closed(_) => {
+                    crate::node_api::NodeApiError::NotSupported(
+                        "dynamic connect not supported in this mode".into(),
+                    )
+                }
+                tokio::sync::mpsc::error::TrySendError::Full(_) => {
+                    crate::node_api::NodeApiError::Internal("command queue full, try again".into())
+                }
+            })?;
+        tokio::task::block_in_place(|| reply_rx.recv()).map_err(|_| {
+            crate::node_api::NodeApiError::Internal("mode task dropped reply".into())
+        })?
     }
 
     fn listen(
         &self,
-        _addr: std::net::SocketAddr,
+        addr: std::net::SocketAddr,
     ) -> crate::node_api::Result<crate::node_api::ListenInfo> {
-        Err(crate::node_api::NodeApiError::NotSupported(
-            "dynamic listen not yet implemented — specify --listen at startup".into(),
-        ))
+        let (reply_tx, reply_rx) = reply_channel();
+        self.cmd_tx
+            .try_send(NodeCommand::Listen {
+                addr,
+                reply: reply_tx,
+            })
+            .map_err(|err| match err {
+                tokio::sync::mpsc::error::TrySendError::Closed(_) => {
+                    crate::node_api::NodeApiError::NotSupported(
+                        "dynamic listen not supported in this mode".into(),
+                    )
+                }
+                tokio::sync::mpsc::error::TrySendError::Full(_) => {
+                    crate::node_api::NodeApiError::Internal("command queue full, try again".into())
+                }
+            })?;
+        tokio::task::block_in_place(|| reply_rx.recv()).map_err(|_| {
+            crate::node_api::NodeApiError::Internal("mode task dropped reply".into())
+        })?
     }
 
     fn disconnect(&self) -> crate::node_api::Result<()> {
-        Err(crate::node_api::NodeApiError::NotSupported(
-            "dynamic disconnect not yet implemented".into(),
-        ))
+        let (reply_tx, reply_rx) = reply_channel();
+        self.cmd_tx
+            .try_send(NodeCommand::Disconnect { reply: reply_tx })
+            .map_err(|err| match err {
+                tokio::sync::mpsc::error::TrySendError::Closed(_) => {
+                    crate::node_api::NodeApiError::NotSupported(
+                        "dynamic disconnect not supported in this mode".into(),
+                    )
+                }
+                tokio::sync::mpsc::error::TrySendError::Full(_) => {
+                    crate::node_api::NodeApiError::Internal("command queue full, try again".into())
+                }
+            })?;
+        tokio::task::block_in_place(|| reply_rx.recv()).map_err(|_| {
+            crate::node_api::NodeApiError::Internal("mode task dropped reply".into())
+        })?
     }
 
     fn add_route(

crates/core/src/control/mod.rs

@@ -3,6 +3,7 @@ pub mod client;
 pub mod handler;
 pub mod log_buffer;
 pub mod metrics;
+pub mod node_command;
 pub mod peers;
 pub mod routes;
 #[cfg(feature = "quic")]

crates/core/src/control/node_command.rs

@@ -0,0 +1,54 @@
+//! Commands sent from the control API layer to the running mode task.
+//!
+//! [`NodeCommand`] is the message type used by [`super::handler::Handler`] to
+//! request dynamic operations (connect, listen, disconnect) from the active
+//! mode task. Replies are delivered via a one-shot channel embedded in each
+//! variant.
+//!
+//! The mode task receives commands from an `mpsc::Receiver<NodeCommand>` and
+//! the handler sends them via the paired `mpsc::Sender<NodeCommand>`. For
+//! modes that do not support dynamic operations (entry, exit, relay), the
+//! receiver end is simply dropped, which causes the handler's send to fail and
+//! return a `NotSupported` error.
+
+use std::net::SocketAddr;
+
+use crate::node_api::{ConnectInfo, ListenInfo, NodeApiError};
+
+/// Reply channel for a single command.
+///
+/// Uses a standard-library sync channel so the handler side (which may be
+/// called from a synchronous context) can block waiting for the reply without
+/// needing an async runtime handle.
+pub type ReplySender<T> = std::sync::mpsc::SyncSender<Result<T, NodeApiError>>;
+pub type ReplyReceiver<T> = std::sync::mpsc::Receiver<Result<T, NodeApiError>>;
+
+/// Create a reply channel pair for a node command.
+#[must_use]
+pub fn reply_channel<T>() -> (ReplySender<T>, ReplyReceiver<T>) {
+    std::sync::mpsc::sync_channel(1)
+}
+
+/// A command sent from the handler/API layer to the mode task.
+#[derive(Debug)]
+pub enum NodeCommand {
+    /// Connect to a remote peer at the given address.
+    Connect {
+        /// Target address (host, host:port, etc.).
+        addr: String,
+        /// Channel for sending the result back to the caller.
+        reply: ReplySender<ConnectInfo>,
+    },
+    /// Start listening for incoming peer connections.
+    Listen {
+        /// Address to bind.
+        addr: SocketAddr,
+        /// Channel for sending the result back to the caller.
+        reply: ReplySender<ListenInfo>,
+    },
+    /// Disconnect from the currently connected peer.
+    Disconnect {
+        /// Channel for sending the result back to the caller.
+        reply: ReplySender<()>,
+    },
+}

crates/daemon/src/lib.rs

@@ -164,6 +164,7 @@ pub fn start_node(
         log_buffer,
     );
     let node_state = handler.node_state();
+    let cmd_rx = handler.take_cmd_rx();
     let node_api: Arc<dyn NodeApi> = Arc::new(handler);
 
     let (shutdown_tx, _shutdown_rx) = watch::channel(());
@@ -177,6 +178,7 @@ pub fn start_node(
         route_updates: route_update_rx,
         route_updates_tx: route_update_tx,
         node_state,
+        cmd_rx,
     };
     let task = tokio::spawn(async move { mode::run(&config, resources).await.map_err(Into::into) });