Fleet terminal: WebRTC DataChannel direct path (low-latency) + WS fallback

Terminal keystrokes were ~400ms because every byte relayed browser→Cloudflare→
gw-04 (Europe)→tailscale→node, even when the node is on the user's own LAN. Add a
direct browser↔node WebRTC DataChannel so co-located terminals run at LAN speed;
the gateway is reduced to signaling; the WebSocket relay stays as the automatic
fallback (graceful degradation — never worse than before).

Daemon (clawmates-node v0.4.0, new src/rtc.rs):
- Add the `webrtc` crate (reuses the ring crypto provider we already install — no
  conflict). Browser is the offerer; we answer, trickle ICE back over the control
  channel, and on DataChannel open spawn a host PTY (tmux) bridged DIRECTLY to the
  channel. Refactor open_pty → spawn_terminal_pty shared by both transports.
  iceServers: STUN + auto host/tailnet candidates (direct, no relay, for LAN/tailnet).

Server (cm-api):
- NodeConn.signal_sinks; Uplink WebRtcAnswer/WebRtcIce/WebRtcFailed routed to the
  browser; NodeHub webrtc_offer/ice/close + open_session/open_pty (open_terminal
  split so the PTY opens only once the transport is chosen). bridge_terminal relays
  signaling over the existing ticket-authed WS and opens the relay PTY on
  {type:"fallback"}.

Browser (NodeTerminalApp):
- RTCPeerConnection + reliable/ordered DataChannel; offer/answer/ICE over the WS;
  2.5s race → use the DataChannel if it opens, else fall back to the WS relay.
  Reconnect wraps both. A direct⚡/relayed indicator shows the live transport.

Deployed; both nodes (morpheus, tank) updated to v0.4.0 and steady online. Direct-
path proof is a browser action (the ⚡ indicator + latency); confirmable from the
daemon's [rtc] logs.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

crates/bins/clawmates-node/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "clawmates-node"
-version = "0.3.0"
+version = "0.4.0"
 edition.workspace = true
 rust-version.workspace = true
 license.workspace = true
@@ -23,6 +23,8 @@ cm-sandbox = { path = "../../cm-sandbox" }
 # Linking cm-sandbox (bollard) brings a second rustls provider into the graph, so
 # rustls can't auto-pick one — we install `ring` explicitly at startup.
 rustls = { version = "0.23", default-features = false, features = ["ring"] }
+webrtc = "0.17.1"
+bytes = "1.12.0"
 
 [lints]
 workspace = true

crates/bins/clawmates-node/src/main.rs

@@ -19,6 +19,8 @@ use sysinfo::{Disks, System};
 use tokio::sync::{mpsc, Mutex};
 use tokio_tungstenite::tungstenite::Message;
 
+mod rtc;
+
 const B64: base64::engine::general_purpose::GeneralPurpose = base64::engine::general_purpose::STANDARD;
 
 /// A live host-shell PTY the gateway opened (keyed by session id).
@@ -104,6 +106,7 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
     // through one channel so PTY reader threads can push asynchronously.
     let (out_tx, mut out_rx) = mpsc::unbounded_channel::<String>();
     let ptys: Ptys = Arc::new(Mutex::new(HashMap::new()));
+    let peers: rtc::RtcPeers = Arc::new(Mutex::new(HashMap::new()));
     // Collect heartbeats on a dedicated thread: the metric helpers shell out to
     // docker/tailscale and stat disks (blocking), which must never stall the
     // async select loop (or heartbeats/pongs would starve during a slow op).
@@ -138,8 +141,9 @@ async fn run(ws_url: &str) -> Result<(), Box<dyn std::error::Error>> {
                         // blocks heartbeats, pongs, or other commands.
                         let out = out_tx.clone();
                         let ptys = ptys.clone();
+                        let peers = peers.clone();
                         let text = t.to_string();
-                        tokio::spawn(async move { handle_frame(&text, &out, &ptys).await; });
+                        tokio::spawn(async move { handle_frame(&text, &out, &ptys, &peers).await; });
                     }
                     Some(Ok(Message::Ping(p))) => write.send(Message::Pong(p)).await?,
                     Some(Ok(Message::Close(_))) | None => return Ok(()),
@@ -240,7 +244,12 @@ fn tailscale_ip() -> Option<String> {
 /// Handle a typed frame from the gateway. The gateway never sends arbitrary
 /// shell — only vetted ops (verify, and an interactive host terminal the user
 /// explicitly opened), so the host attack surface stays minimal.
-async fn handle_frame(text: &str, out: &mpsc::UnboundedSender<String>, ptys: &Ptys) {
+async fn handle_frame(
+    text: &str,
+    out: &mpsc::UnboundedSender<String>,
+    ptys: &Ptys,
+    peers: &rtc::RtcPeers,
+) {
     let Ok(v) = serde_json::from_str::<Value>(text) else {
         return;
     };
@@ -300,10 +309,50 @@ async fn handle_frame(text: &str, out: &mpsc::UnboundedSender<String>, ptys: &Pt
                 let _ = p.child.kill();
             }
         }
+        // WebRTC signaling (browser is offerer). The DataChannel, once open,
+        // carries terminal I/O directly peer-to-peer; resize/close still arrive
+        // as pty_resize/pty_close keyed by the same sid.
+        "webrtc_offer" => {
+            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
+            let sdp = v.get("sdp").and_then(Value::as_str).unwrap_or("").to_owned();
+            rtc::handle_offer(sid, sdp, out.clone(), ptys.clone(), peers.clone()).await;
+        }
+        "webrtc_ice" => {
+            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
+            rtc::handle_ice(sid, &v, peers).await;
+        }
+        "webrtc_close" => {
+            let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
+            rtc::handle_close(sid, peers, ptys).await;
+        }
         _ => {}
     }
 }
 
+/// (master for resize, cloned reader, writer, child) — one PTY spawn reused by
+/// both the WS-relay path and the WebRTC DataChannel path.
+type PtyParts = (
+    Box<dyn MasterPty + Send>,
+    Box<dyn Read + Send>,
+    Box<dyn Write + Send>,
+    Box<dyn portable_pty::Child + Send + Sync>,
+);
+
+/// Spawn a host login shell (tmux `-L clawmates`) in a fresh PTY.
+fn spawn_terminal_pty(cols: u16, rows: u16) -> Result<PtyParts, String> {
+    let pair = native_pty_system()
+        .openpty(PtySize { rows, cols, pixel_width: 0, pixel_height: 0 })
+        .map_err(|e| e.to_string())?;
+    let child = pair
+        .slave
+        .spawn_command(terminal_command())
+        .map_err(|e| e.to_string())?;
+    drop(pair.slave);
+    let reader = pair.master.try_clone_reader().map_err(|e| e.to_string())?;
+    let writer = pair.master.take_writer().map_err(|e| e.to_string())?;
+    Ok((pair.master, reader, writer, child))
+}
+
 /// Spawn a host login shell in a PTY; stream its output back as pty_out frames.
 async fn open_pty(
     sid: u64,
@@ -312,17 +361,8 @@ async fn open_pty(
     out: mpsc::UnboundedSender<String>,
     ptys: Ptys,
 ) -> Result<(), String> {
-    let pair = native_pty_system()
+    let (master, mut reader, writer, child) = spawn_terminal_pty(cols, rows)?;
-        .openpty(PtySize { rows, cols, pixel_width: 0, pixel_height: 0 })
+    ptys.lock().await.insert(sid, Pty { master, writer, child });
-        .map_err(|e| e.to_string())?;
-    let child = pair
-        .slave
-        .spawn_command(terminal_command())
-        .map_err(|e| e.to_string())?;
-    drop(pair.slave);
-    let mut reader = pair.master.try_clone_reader().map_err(|e| e.to_string())?;
-    let writer = pair.master.take_writer().map_err(|e| e.to_string())?;
-    ptys.lock().await.insert(sid, Pty { master: pair.master, writer, child });
     let tmux = has_tmux();
     eprintln!("[pty] open sid={sid} cols={cols} rows={rows} tmux={tmux}");
     // Immediate banner over the channel: if the browser shows this but no shell,

crates/bins/clawmates-node/src/rtc.rs

@@ -0,0 +1,230 @@
+//! WebRTC DataChannel terminal transport.
+//!
+//! The browser is the offerer; we answer and trickle ICE back over the gateway
+//! control channel. On DataChannel open we spawn a host PTY (tmux) bridged
+//! DIRECTLY to the channel — so terminal I/O flows browser↔node peer-to-peer
+//! (LAN speed via ICE host candidates) instead of relaying through the gateway.
+//! The WebSocket relay path stays as the automatic fallback (handled elsewhere).
+
+use std::collections::HashMap;
+use std::io::{Read, Write};
+use std::sync::Arc;
+
+use bytes::Bytes;
+use serde_json::{json, Value};
+use tokio::sync::{mpsc, Mutex};
+
+use webrtc::api::interceptor_registry::register_default_interceptors;
+use webrtc::api::media_engine::MediaEngine;
+use webrtc::api::APIBuilder;
+use webrtc::data_channel::data_channel_message::DataChannelMessage;
+use webrtc::data_channel::RTCDataChannel;
+use webrtc::ice_transport::ice_candidate::{RTCIceCandidate, RTCIceCandidateInit};
+use webrtc::ice_transport::ice_server::RTCIceServer;
+use webrtc::interceptor::registry::Registry;
+use webrtc::peer_connection::configuration::RTCConfiguration;
+use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
+use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
+use webrtc::peer_connection::RTCPeerConnection;
+
+use crate::{spawn_terminal_pty, Pty, Ptys};
+
+/// Live WebRTC peers keyed by terminal session id.
+pub type RtcPeers = Arc<Mutex<HashMap<u64, Arc<RTCPeerConnection>>>>;
+
+/// Handle a browser SDP offer: build a peer, answer, trickle ICE, and on
+/// DataChannel open bridge it to a host PTY. Errors are reported so the browser
+/// can fall back to the WebSocket relay.
+pub async fn handle_offer(
+    sid: u64,
+    sdp: String,
+    out: mpsc::UnboundedSender<String>,
+    ptys: Ptys,
+    peers: RtcPeers,
+) {
+    if let Err(e) = build_peer(sid, sdp, out.clone(), ptys, peers).await {
+        eprintln!("[rtc] sid={sid} offer failed: {e}");
+        let _ = out.send(json!({ "t": "webrtc_failed", "sid": sid, "error": e }).to_string());
+    }
+}
+
+async fn build_peer(
+    sid: u64,
+    sdp: String,
+    out: mpsc::UnboundedSender<String>,
+    ptys: Ptys,
+    peers: RtcPeers,
+) -> Result<(), String> {
+    let mut m = MediaEngine::default();
+    let registry = register_default_interceptors(Registry::new(), &mut m).map_err(|e| e.to_string())?;
+    let api = APIBuilder::new()
+        .with_media_engine(m)
+        .with_interceptor_registry(registry)
+        .build();
+
+    // STUN for server-reflexive discovery; ICE also gathers host candidates
+    // (LAN + tailscale 100.x) which connect co-located peers with no relay.
+    let config = RTCConfiguration {
+        ice_servers: vec![RTCIceServer {
+            urls: vec!["stun:stun.l.google.com:19302".to_owned()],
+            ..Default::default()
+        }],
+        ..Default::default()
+    };
+    let pc = Arc::new(api.new_peer_connection(config).await.map_err(|e| e.to_string())?);
+
+    // Trickle our local ICE candidates back to the browser over the control channel.
+    let out_ice = out.clone();
+    pc.on_ice_candidate(Box::new(move |cand: Option<RTCIceCandidate>| {
+        let out_ice = out_ice.clone();
+        Box::pin(async move {
+            if let Some(c) = cand {
+                if let Ok(init) = c.to_json() {
+                    let _ = out_ice.send(
+                        json!({
+                            "t": "webrtc_ice",
+                            "sid": sid,
+                            "candidate": init.candidate,
+                            "sdp_mid": init.sdp_mid,
+                            "sdp_mline_index": init.sdp_mline_index,
+                        })
+                        .to_string(),
+                    );
+                }
+            }
+        })
+    }));
+
+    // The browser creates the DataChannel; on open, bridge it to a host PTY.
+    let ptys_dc = ptys.clone();
+    pc.on_data_channel(Box::new(move |dc: Arc<RTCDataChannel>| {
+        let ptys_dc = ptys_dc.clone();
+        Box::pin(async move {
+            let dc_open = dc.clone();
+            let ptys_open = ptys_dc.clone();
+            dc.on_open(Box::new(move || {
+                Box::pin(async move {
+                    bridge_pty(sid, dc_open, ptys_open).await;
+                })
+            }));
+        })
+    }));
+
+    // Tear down the PTY when the peer drops.
+    let peers_st = peers.clone();
+    let ptys_st = ptys.clone();
+    pc.on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| {
+        let peers_st = peers_st.clone();
+        let ptys_st = ptys_st.clone();
+        Box::pin(async move {
+            if matches!(
+                s,
+                RTCPeerConnectionState::Failed
+                    | RTCPeerConnectionState::Closed
+                    | RTCPeerConnectionState::Disconnected
+            ) {
+                peers_st.lock().await.remove(&sid);
+                if let Some(mut p) = ptys_st.lock().await.remove(&sid) {
+                    let _ = p.child.kill();
+                }
+            }
+        })
+    }));
+
+    pc.set_remote_description(RTCSessionDescription::offer(sdp).map_err(|e| e.to_string())?)
+        .await
+        .map_err(|e| e.to_string())?;
+    let answer = pc.create_answer(None).await.map_err(|e| e.to_string())?;
+    pc.set_local_description(answer.clone()).await.map_err(|e| e.to_string())?;
+    let _ = out.send(json!({ "t": "webrtc_answer", "sid": sid, "sdp": answer.sdp }).to_string());
+
+    peers.lock().await.insert(sid, pc);
+    eprintln!("[rtc] sid={sid} answered, peer up");
+    Ok(())
+}
+
+/// Spawn the host PTY for this session and pump it both ways over the DataChannel.
+async fn bridge_pty(sid: u64, dc: Arc<RTCDataChannel>, ptys: Ptys) {
+    let (master, mut reader, writer, child) = match spawn_terminal_pty(120, 40) {
+        Ok(parts) => parts,
+        Err(e) => {
+            let _ = dc
+                .send_text(format!("\r\n[clawmates] could not start shell: {e}\r\n"))
+                .await;
+            return;
+        }
+    };
+    ptys.lock().await.insert(sid, Pty { master, writer, child });
+    eprintln!("[rtc] sid={sid} DataChannel open → PTY bridged (direct)");
+
+    // PTY output → DataChannel. Blocking reads on a thread feed an async sender.
+    let (tx, mut rx) = mpsc::unbounded_channel::<Vec<u8>>();
+    std::thread::spawn(move || {
+        let mut buf = [0u8; 8192];
+        loop {
+            match reader.read(&mut buf) {
+                Ok(0) | Err(_) => break,
+                Ok(n) => {
+                    if tx.send(buf[..n].to_vec()).is_err() {
+                        break;
+                    }
+                }
+            }
+        }
+    });
+    let dc_out = dc.clone();
+    tokio::spawn(async move {
+        while let Some(bytes) = rx.recv().await {
+            if dc_out.send(&Bytes::from(bytes)).await.is_err() {
+                break;
+            }
+        }
+    });
+
+    // DataChannel input (keystrokes) → PTY writer.
+    let ptys_in = ptys.clone();
+    dc.on_message(Box::new(move |msg: DataChannelMessage| {
+        let ptys_in = ptys_in.clone();
+        Box::pin(async move {
+            if let Some(p) = ptys_in.lock().await.get_mut(&sid) {
+                let _ = p.writer.write_all(&msg.data);
+                let _ = p.writer.flush();
+            }
+        })
+    }));
+}
+
+/// Add a trickled ICE candidate from the browser.
+pub async fn handle_ice(sid: u64, v: &Value, peers: &RtcPeers) {
+    let candidate = v
+        .get("candidate")
+        .and_then(Value::as_str)
+        .unwrap_or("")
+        .to_owned();
+    let sdp_mid = v.get("sdp_mid").and_then(Value::as_str).map(str::to_owned);
+    let sdp_mline_index = v
+        .get("sdp_mline_index")
+        .and_then(Value::as_u64)
+        .map(|n| n as u16);
+    let pc = peers.lock().await.get(&sid).cloned();
+    if let Some(pc) = pc {
+        let _ = pc
+            .add_ice_candidate(RTCIceCandidateInit {
+                candidate,
+                sdp_mid,
+                sdp_mline_index,
+                username_fragment: None,
+            })
+            .await;
+    }
+}
+
+/// Close a peer + its PTY (browser navigated away or fell back to the relay).
+pub async fn handle_close(sid: u64, peers: &RtcPeers, ptys: &Ptys) {
+    if let Some(pc) = peers.lock().await.remove(&sid) {
+        let _ = pc.close().await;
+    }
+    if let Some(mut p) = ptys.lock().await.remove(&sid) {
+        let _ = p.child.kill();
+    }
+}

crates/cm-api/src/fleet.rs

@@ -58,8 +58,12 @@ pub struct ExecOutput {
 struct NodeConn {
     tx: mpsc::UnboundedSender<String>,
     pending: Mutex<HashMap<u64, oneshot::Sender<ExecOutput>>>,
-    /// Live terminal sessions: sid → sink for the browser bridge.
+    /// Live terminal sessions: sid → byte sink for the browser bridge (WS-relay
+    /// PTY output).
     pty_sinks: Mutex<HashMap<u64, mpsc::UnboundedSender<Vec<u8>>>>,
+    /// WebRTC signaling: sid → text sink delivering the daemon's answer/ICE to
+    /// the browser bridge.
+    signal_sinks: Mutex<HashMap<u64, mpsc::UnboundedSender<String>>>,
     next_id: AtomicU64,
     /// Unique per physical connection — see `CONN_EPOCH`.
     epoch: u64,
@@ -143,24 +147,69 @@ impl NodeHub {
         }
     }
 
-    /// Open a host terminal on a node: allocate a session, ask the daemon to
+    /// Allocate a terminal session: a sid + a byte stream (WS-relay PTY output)
-    /// spawn a PTY, and return the sid + a stream of its output bytes.
+    /// + a text stream (WebRTC answer/ICE). The PTY is NOT opened yet — the
-    pub async fn open_terminal(
+    /// browser picks the transport (WebRTC direct, or `open_pty` fallback).
+    pub async fn open_session(
         &self,
         id: NodeId,
-        cols: u16,
+    ) -> Option<(
-        rows: u16,
+        u64,
-    ) -> Option<(u64, mpsc::UnboundedReceiver<Vec<u8>>)> {
+        mpsc::UnboundedReceiver<Vec<u8>>,
+        mpsc::UnboundedReceiver<String>,
+    )> {
         let conn = self.get(id).await?;
         let sid = conn.next_id.fetch_add(1, Ordering::Relaxed);
-        let (tx, rx) = mpsc::unbounded_channel();
+        let (ptx, prx) = mpsc::unbounded_channel();
-        conn.pty_sinks.lock().await.insert(sid, tx);
+        let (stx, srx) = mpsc::unbounded_channel();
-        let frame = json!({ "t": "pty_open", "sid": sid, "cols": cols, "rows": rows }).to_string();
+        conn.pty_sinks.lock().await.insert(sid, ptx);
-        if conn.tx.send(frame).is_err() {
+        conn.signal_sinks.lock().await.insert(sid, stx);
-            conn.pty_sinks.lock().await.remove(&sid);
+        Some((sid, prx, srx))
-            return None;
+    }
+
+    /// Open the WS-relay PTY for an allocated session (the fallback path).
+    pub async fn open_pty(&self, id: NodeId, sid: u64, cols: u16, rows: u16) {
+        if let Some(conn) = self.get(id).await {
+            let _ = conn.tx.send(
+                json!({ "t": "pty_open", "sid": sid, "cols": cols, "rows": rows }).to_string(),
+            );
+        }
+    }
+
+    /// Relay a browser SDP offer to the daemon (it answers + trickles ICE back).
+    pub async fn webrtc_offer(&self, id: NodeId, sid: u64, sdp: &str) {
+        if let Some(conn) = self.get(id).await {
+            let _ = conn
+                .tx
+                .send(json!({ "t": "webrtc_offer", "sid": sid, "sdp": sdp }).to_string());
+        }
+    }
+
+    /// Relay a browser ICE candidate to the daemon.
+    pub async fn webrtc_ice(
+        &self,
+        id: NodeId,
+        sid: u64,
+        candidate: &str,
+        sdp_mid: Option<&str>,
+        sdp_mline_index: Option<u16>,
+    ) {
+        if let Some(conn) = self.get(id).await {
+            let _ = conn.tx.send(
+                json!({ "t": "webrtc_ice", "sid": sid, "candidate": candidate,
+                        "sdp_mid": sdp_mid, "sdp_mline_index": sdp_mline_index })
+                .to_string(),
+            );
+        }
+    }
+
+    /// Tell the daemon to tear down a WebRTC peer for a session.
+    pub async fn webrtc_close(&self, id: NodeId, sid: u64) {
+        if let Some(conn) = self.get(id).await {
+            let _ = conn
+                .tx
+                .send(json!({ "t": "webrtc_close", "sid": sid }).to_string());
         }
-        Some((sid, rx))
     }
 
     pub async fn terminal_input(&self, id: NodeId, sid: u64, bytes: &[u8]) {
@@ -182,7 +231,9 @@ impl NodeHub {
     pub async fn terminal_close(&self, id: NodeId, sid: u64) {
         if let Some(conn) = self.get(id).await {
             conn.pty_sinks.lock().await.remove(&sid);
+            conn.signal_sinks.lock().await.remove(&sid);
             let _ = conn.tx.send(json!({ "t": "pty_close", "sid": sid }).to_string());
+            let _ = conn.tx.send(json!({ "t": "webrtc_close", "sid": sid }).to_string());
         }
     }
 
@@ -227,6 +278,17 @@ enum Uplink {
     PtyOut { sid: u64, data: String },
     #[serde(rename = "pty_exit")]
     PtyExit { sid: u64 },
+    #[serde(rename = "webrtc_answer")]
+    WebRtcAnswer { sid: u64, sdp: String },
+    #[serde(rename = "webrtc_ice")]
+    WebRtcIce {
+        sid: u64,
+        candidate: String,
+        sdp_mid: Option<String>,
+        sdp_mline_index: Option<u16>,
+    },
+    #[serde(rename = "webrtc_failed")]
+    WebRtcFailed { sid: u64 },
 }
 
 #[derive(Deserialize)]
@@ -254,6 +316,7 @@ pub async fn run_channel(pool: PgPool, hub: Arc<NodeHub>, node_id: NodeId, socke
         tx,
         pending: Mutex::new(HashMap::new()),
         pty_sinks: Mutex::new(HashMap::new()),
+        signal_sinks: Mutex::new(HashMap::new()),
         next_id: AtomicU64::new(0),
         epoch,
     });
@@ -338,6 +401,34 @@ pub async fn run_channel(pool: PgPool, hub: Arc<NodeHub>, node_id: NodeId, socke
                     Ok(Uplink::PtyExit { sid }) => {
                         conn.pty_sinks.lock().await.remove(&sid);
                     }
+                    // WebRTC signaling from the daemon → forward to the browser
+                    // bridge (re-tagged `type` for the browser) via signal_sinks.
+                    Ok(Uplink::WebRtcAnswer { sid, sdp }) => {
+                        if let Some(s) = conn.signal_sinks.lock().await.get(&sid) {
+                            let _ = s.send(
+                                json!({ "type": "webrtc_answer", "sdp": sdp }).to_string(),
+                            );
+                        }
+                    }
+                    Ok(Uplink::WebRtcIce {
+                        sid,
+                        candidate,
+                        sdp_mid,
+                        sdp_mline_index,
+                    }) => {
+                        if let Some(s) = conn.signal_sinks.lock().await.get(&sid) {
+                            let _ = s.send(
+                                json!({ "type": "webrtc_ice", "candidate": candidate,
+                                        "sdp_mid": sdp_mid, "sdp_mline_index": sdp_mline_index })
+                                .to_string(),
+                            );
+                        }
+                    }
+                    Ok(Uplink::WebRtcFailed { sid }) => {
+                        if let Some(s) = conn.signal_sinks.lock().await.get(&sid) {
+                            let _ = s.send(json!({ "type": "webrtc_failed" }).to_string());
+                        }
+                    }
                     Err(_) => {}
                 }
             },

crates/cm-api/src/routes/nodes.rs

@@ -202,35 +202,76 @@ pub async fn terminal_ws(
     upgrade.on_upgrade(move |socket| bridge_terminal(hub, node_id, socket))
 }
 
+/// Browser→server control frames over the terminal WS: a resize, the
+/// `fallback` request to open the WS-relay PTY, or WebRTC signaling (offer/ICE).
 #[derive(Deserialize)]
 struct TermCtrl {
     #[serde(rename = "type")]
     kind: String,
-    cols: u16,
+    cols: Option<u16>,
-    rows: u16,
+    rows: Option<u16>,
+    sdp: Option<String>,
+    candidate: Option<String>,
+    sdp_mid: Option<String>,
+    sdp_mline_index: Option<u16>,
 }
 
+/// The terminal WS is BOTH the WebRTC signaling channel and the fallback data
+/// path. The browser tries a direct DataChannel first (offer/ICE relayed here);
+/// if that fails it sends `{type:"fallback"}` and we open the WS-relay PTY.
 async fn bridge_terminal(hub: Arc<NodeHub>, node_id: NodeId, socket: WebSocket) {
-    let Some((sid, mut rx)) = hub.open_terminal(node_id, 80, 24).await else {
+    let Some((sid, mut pty_rx, mut sig_rx)) = hub.open_session(node_id).await else {
         return;
     };
     let (mut ws_tx, mut ws_rx) = socket.split();
     let to_browser = async {
-        while let Some(bytes) = rx.recv().await {
+        loop {
-            if ws_tx.send(Message::Binary(bytes.into())).await.is_err() {
+            tokio::select! {
-                break;
+                bytes = pty_rx.recv() => match bytes {
+                    Some(b) => {
+                        if ws_tx.send(Message::Binary(b.into())).await.is_err() { break; }
+                    }
+                    None => break,
+                },
+                text = sig_rx.recv() => match text {
+                    Some(t) => {
+                        if ws_tx.send(Message::Text(t.into())).await.is_err() { break; }
+                    }
+                    None => break,
+                },
             }
         }
     };
     let to_node = async {
         while let Some(Ok(msg)) = ws_rx.next().await {
             match msg {
+                // Binary = keystrokes over the WS fallback path only (the direct
+                // DataChannel carries its own input).
                 Message::Binary(b) => hub.terminal_input(node_id, sid, b.as_ref()).await,
                 Message::Text(t) => {
-                    if let Ok(c) = serde_json::from_str::<TermCtrl>(t.as_str()) {
+                    let Ok(c) = serde_json::from_str::<TermCtrl>(t.as_str()) else {
-                        if c.kind == "resize" {
+                        continue;
-                            hub.terminal_resize(node_id, sid, c.cols, c.rows).await;
+                    };
+                    let cols = c.cols.unwrap_or(80);
+                    let rows = c.rows.unwrap_or(24);
+                    match c.kind.as_str() {
+                        "resize" => hub.terminal_resize(node_id, sid, cols, rows).await,
+                        "fallback" => hub.open_pty(node_id, sid, cols, rows).await,
+                        "webrtc_offer" => {
+                            hub.webrtc_offer(node_id, sid, c.sdp.as_deref().unwrap_or("")).await
                         }
+                        "webrtc_ice" => {
+                            hub.webrtc_ice(
+                                node_id,
+                                sid,
+                                c.candidate.as_deref().unwrap_or(""),
+                                c.sdp_mid.as_deref(),
+                                c.sdp_mline_index,
+                            )
+                            .await
+                        }
+                        "webrtc_close" => hub.webrtc_close(node_id, sid).await,
+                        _ => {}
                     }
                 }
                 Message::Close(_) => break,

frontend/src/components/computer/apps/infra/NodeTerminalApp.tsx

@@ -1,12 +1,13 @@
 "use client";
 
 // The infra computer's Terminal app: a shell on a fleet node, rendered inside
-// the pull-out (not a separate window). Targets the node in the ?node= param;
+// the pull-out. Terminal I/O prefers a DIRECT WebRTC DataChannel (browser↔node
-// shows a picker when none is set. PTY is proxied over the node control channel.
+// peer-to-peer, LAN speed) and falls back to the gateway WebSocket relay if no
+// direct path forms. The same WS carries the WebRTC signaling and the fallback.
 
-import { Server, Terminal as TerminalIcon } from "lucide-react";
+import { Server, Terminal as TerminalIcon, Zap } from "lucide-react";
 import { useQueryStates } from "nuqs";
-import { useEffect, useRef } from "react";
+import { useEffect, useRef, useState } from "react";
 
 import { useFetchJson } from "@/lib/api/use-fetch";
 import { panelParsers } from "@/lib/url/panel-params";
@@ -14,18 +15,34 @@ import type { FleetNode } from "@/components/dashboard/fleet/FleetPanels";
 
 import "@xterm/xterm/css/xterm.css";
 
-/** xterm bridged to a node's host shell, filling the app window. */
+const ICE_SERVERS: RTCIceServer[] = [{ urls: ["stun:stun.l.google.com:19302"] }];
+type Transport = "connecting" | "direct" | "relayed";
+
+/** xterm bridged to a node's host shell, preferring a direct WebRTC DataChannel. */
 function NodeShell({ nodeId }: { nodeId: string }) {
   const hostRef = useRef<HTMLDivElement>(null);
+  const [transport, setTransport] = useState<Transport>("connecting");
 
   useEffect(() => {
     let disposed = false;
     let term: import("@xterm/xterm").Terminal | null = null;
     let fit: import("@xterm/addon-fit").FitAddon | null = null;
     let ws: WebSocket | null = null;
+    let pc: RTCPeerConnection | null = null;
+    let dc: RTCDataChannel | null = null;
     let ro: ResizeObserver | null = null;
     let resizeT: ReturnType<typeof setTimeout> | undefined;
     let reconnectT: ReturnType<typeof setTimeout> | undefined;
+    let raceT: ReturnType<typeof setTimeout> | undefined;
+    let mode: Transport = "connecting";
+    let remoteSet = false;
+    const pendingIce: RTCIceCandidateInit[] = [];
+    const inputQueue: string[] = [];
+
+    const setMode = (m: Transport) => {
+      mode = m;
+      if (!disposed) setTransport(m);
+    };
 
     const sendResize = () => {
       if (ws?.readyState === WebSocket.OPEN && term && fit && hostRef.current?.clientWidth) {
@@ -34,12 +51,113 @@ function NodeShell({ nodeId }: { nodeId: string }) {
       }
     };
 
-    // Reconnecting transport. The host session is a persistent `tmux` server, so
+    // Route keystrokes to whichever transport is live; buffer while connecting.
-    // each reconnect re-attaches and tmux redraws the live screen — mosh-style
+    const sendInput = (d: string) => {
-    // snap-to-state over our WS, no byte-backlog replay. The xterm instance and
+      const bytes = new TextEncoder().encode(d);
-    // its scrollback persist across drops (we never dispose it between tries).
+      if (mode === "direct" && dc?.readyState === "open") dc.send(bytes);
+      else if (mode === "relayed" && ws?.readyState === WebSocket.OPEN) ws.send(bytes);
+      else inputQueue.push(d);
+    };
+    const flushInput = () => {
+      const q = inputQueue.splice(0);
+      for (const d of q) sendInput(d);
+    };
+
+    // Give up on the direct path → use the WS relay (today's behavior).
+    const goRelayed = () => {
+      if (disposed || mode !== "connecting") return;
+      clearTimeout(raceT);
+      setMode("relayed");
+      try {
+        dc?.close();
+        pc?.close();
+      } catch {
+        /* ignore */
+      }
+      dc = null;
+      pc = null;
+      ws?.send(JSON.stringify({ type: "fallback", cols: term?.cols ?? 80, rows: term?.rows ?? 24 }));
+      flushInput();
+    };
+
+    const startWebrtc = () => {
+      if (typeof RTCPeerConnection === "undefined") {
+        goRelayed();
+        return;
+      }
+      try {
+        pc = new RTCPeerConnection({ iceServers: ICE_SERVERS });
+        dc = pc.createDataChannel("term", { ordered: true });
+        dc.binaryType = "arraybuffer";
+        dc.onopen = () => {
+          if (disposed || mode !== "connecting") return;
+          clearTimeout(raceT);
+          setMode("direct");
+          flushInput();
+          sendResize();
+        };
+        dc.onmessage = (e) => {
+          if (typeof e.data === "string") term?.write(e.data);
+          else term?.write(new Uint8Array(e.data as ArrayBuffer));
+        };
+        dc.onclose = () => {
+          if (mode === "direct" && !disposed) ws?.close(); // lost direct → reconnect
+        };
+        pc.onicecandidate = (e) => {
+          if (e.candidate && ws?.readyState === WebSocket.OPEN) {
+            ws.send(
+              JSON.stringify({
+                type: "webrtc_ice",
+                candidate: e.candidate.candidate,
+                sdp_mid: e.candidate.sdpMid,
+                sdp_mline_index: e.candidate.sdpMLineIndex,
+              }),
+            );
+          }
+        };
+        pc.onconnectionstatechange = () => {
+          if (pc?.connectionState === "failed") goRelayed();
+        };
+        pc.createOffer()
+          .then((offer) => pc!.setLocalDescription(offer).then(() => {
+            ws?.send(JSON.stringify({ type: "webrtc_offer", sdp: offer.sdp }));
+          }))
+          .catch(() => goRelayed());
+        // Race: if no direct DataChannel within 2.5s, fall back.
+        raceT = setTimeout(goRelayed, 2500);
+      } catch {
+        goRelayed();
+      }
+    };
+
+    const onSignal = (m: { type: string; sdp?: string; candidate?: string; sdp_mid?: string | null; sdp_mline_index?: number | null }) => {
+      if (!pc) return;
+      if (m.type === "webrtc_answer" && m.sdp) {
+        pc.setRemoteDescription({ type: "answer", sdp: m.sdp })
+          .then(() => {
+            remoteSet = true;
+            for (const c of pendingIce.splice(0)) pc?.addIceCandidate(c).catch(() => {});
+          })
+          .catch(() => goRelayed());
+      } else if (m.type === "webrtc_ice" && m.candidate != null) {
+        const cand: RTCIceCandidateInit = {
+          candidate: m.candidate,
+          sdpMid: m.sdp_mid ?? undefined,
+          sdpMLineIndex: m.sdp_mline_index ?? undefined,
+        };
+        if (remoteSet) pc.addIceCandidate(cand).catch(() => {});
+        else pendingIce.push(cand);
+      } else if (m.type === "webrtc_failed") {
+        goRelayed();
+      }
+    };
+
     const connect = async () => {
       if (disposed || !term) return;
+      setMode("connecting");
+      pc = null;
+      dc = null;
+      remoteSet = false;
       let ticket: string | undefined;
       try {
         const res = await fetch(`/api/nodes/${nodeId}/terminal/ticket`, { method: "POST" });
@@ -58,14 +176,28 @@ function NodeShell({ nodeId }: { nodeId: string }) {
       const proto = location.protocol === "https:" ? "wss:" : "ws:";
       ws = new WebSocket(`${proto}//${location.host}/api/nodes/${nodeId}/terminal/ws?token=${encodeURIComponent(ticket)}`);
       ws.binaryType = "arraybuffer";
-      ws.onopen = () => sendResize();
+      ws.onopen = () => startWebrtc();
       ws.onmessage = (e) => {
-        if (typeof e.data === "string") term?.write(e.data);
+        if (typeof e.data === "string") {
-        else term?.write(new Uint8Array(e.data as ArrayBuffer));
+          try {
+            onSignal(JSON.parse(e.data));
+          } catch {
+            /* ignore malformed signaling */
+          }
+        } else {
+          // Binary = PTY output over the WS relay (fallback path).
+          term?.write(new Uint8Array(e.data as ArrayBuffer));
+        }
       };
-      // onerror is always followed by onclose; reconnect from there only.
       ws.onclose = () => {
         if (disposed) return;
+        clearTimeout(raceT);
+        try {
+          dc?.close();
+          pc?.close();
+        } catch {
+          /* ignore */
+        }
         term?.writeln("\r\n\x1b[2m[reconnecting…]\x1b[0m");
         reconnectT = setTimeout(connect, 1500);
       };
@@ -88,10 +220,7 @@ function NodeShell({ nodeId }: { nodeId: string }) {
       term.open(hostRef.current);
       fit.fit();
       term.focus();
-      term.onData((d) => {
+      term.onData(sendInput);
-        if (ws?.readyState === WebSocket.OPEN) ws.send(new TextEncoder().encode(d));
-      });
-      // Debounce: the pull-out animates open, firing the observer on every pixel.
       ro = new ResizeObserver(() => {
         clearTimeout(resizeT);
         resizeT = setTimeout(sendResize, 150);
@@ -104,6 +233,13 @@ function NodeShell({ nodeId }: { nodeId: string }) {
       disposed = true;
       clearTimeout(resizeT);
       clearTimeout(reconnectT);
+      clearTimeout(raceT);
+      try {
+        dc?.close();
+        pc?.close();
+      } catch {
+        /* ignore */
+      }
       ro?.disconnect();
       ws?.close();
       term?.dispose();
@@ -111,7 +247,18 @@ function NodeShell({ nodeId }: { nodeId: string }) {
   }, [nodeId]);
 
   return (
-    <div className="flex h-full w-full flex-col bg-[#0a0a0c]">
+    <div className="relative flex h-full w-full flex-col bg-[#0a0a0c]">
+      <div
+        className="pointer-events-none absolute right-2 top-2 z-10 flex items-center gap-1 rounded-md px-1.5 py-0.5 font-mono text-[9px]"
+        style={{
+          background: transport === "direct" ? "rgba(95,208,138,.12)" : "rgba(255,255,255,.05)",
+          border: `1px solid ${transport === "direct" ? "rgba(95,208,138,.3)" : "rgba(255,255,255,.1)"}`,
+          color: transport === "direct" ? "#5fd08a" : transport === "relayed" ? "#8a8a92" : "#e8b465",
+        }}
+      >
+        {transport === "direct" ? <Zap size={9} /> : null}
+        {transport === "direct" ? "direct" : transport === "relayed" ? "relayed" : "connecting…"}
+      </div>
       <div className="relative min-h-0 flex-1">
         <div ref={hostRef} className="absolute inset-0 p-2" />
       </div>