Fleet: robust real-time connectivity + mosh-inspired reconnecting terminal

Nodes flapped online/offline and the terminal died on the first blip. WebSockets
are the right transport (outbound, NAT-friendly); the fixes harden around it.

Server (cm-api):
- Anti-clobber connection epoch: a reconnecting daemon gets a fresh epoch; a stale
  run_channel's teardown only clears the hub + sets offline if it still owns the
  slot — so a lingering old channel can't flip a live reconnection offline (the
  main false-offline cause).
- WS keepalive: run_channel now pings every 15s and tears down if no inbound
  frame (incl. pong) for 35s — dead links detected in seconds, not minutes.
- Staleness sweeper backstop: spawn_node_sweeper (8s tick / 20s window) wired in
  clawmates-server, so a vanished node goes offline within ~28s even if its
  channel hangs (mark_stale_offline was defined but never called).

Daemon (clawmates-node v0.3.0):
- Heartbeats off the select thread (dedicated thread owns System + blocking
  docker/tailscale/disk CLIs) so a slow op never starves heartbeats/pongs.
- Each handle_frame runs on its own task; added a 40s inbound idle deadline so a
  half-open socket triggers a reconnect.

Frontend:
- useNodes streams /api/nodes/live (SSE push) instead of a 3s poll; isLive()
  derives online from lastSeen freshness (<15s) so a transient column flip never
  shows a healthy node down.
- Node terminal: clean auto-reconnect loop (re-mint ticket -> reconnect -> tmux
  re-attaches and redraws the live screen = mosh-style snap-to-state over TCP),
  replacing the [disconnected] dead-end.

Mosh evaluated: harvest principles (session/transport decoupling, snap-to-state,
already given by tmux), don't adopt — UDP is incompatible with our browser+CF+NAT
topology and it's GPLv3. Removed temporary terminal debug traces + /api/debug route.

Verified: node holds steadily online (heartbeat 1-3s, no flap) and goes cleanly
offline when the daemon stops.

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

Cargo.lock

@@ -737,7 +737,7 @@ dependencies = [
 
 [[package]]
 name = "clawmates-node"
-version = "0.2.2"
+version = "0.3.0"
 dependencies = [
  "base64",
  "cm-sandbox",

crates/bins/clawmates-node/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "clawmates-node"
-version = "0.2.2"
+version = "0.3.0"
 edition.workspace = true
 rust-version.workspace = true
 license.workspace = true

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

@@ -104,19 +104,49 @@ 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 mut sys = System::new_all();
+    // Collect heartbeats on a dedicated thread: the metric helpers shell out to
-    let mut ticker = tokio::time::interval(Duration::from_secs(5));
+    // docker/tailscale and stat disks (blocking), which must never stall the
+    // async select loop (or heartbeats/pongs would starve during a slow op).
+    let hb_tx = out_tx.clone();
+    std::thread::spawn(move || {
+        let mut sys = System::new_all();
+        loop {
+            if hb_tx.send(heartbeat(&mut sys)).is_err() {
+                break; // connection gone
+            }
+            std::thread::sleep(Duration::from_secs(5));
+        }
+    });
+
+    // Liveness: the server pings every 15s. If nothing inbound arrives for 40s
+    // the socket is dead — return so main() reconnects.
+    let mut idle_tick = tokio::time::interval(Duration::from_secs(5));
+    let mut last_rx = std::time::Instant::now();
     loop {
         tokio::select! {
-            _ = ticker.tick() => { let _ = out_tx.send(heartbeat(&mut sys)); }
             Some(frame) = out_rx.recv() => { write.send(Message::Text(frame.into())).await?; }
-            msg = read.next() => match msg {
+            _ = idle_tick.tick() => {
-                Some(Ok(Message::Text(t))) => handle_frame(t.as_str(), &out_tx, &ptys).await,
+                if last_rx.elapsed() > Duration::from_secs(40) {
-                Some(Ok(Message::Ping(p))) => write.send(Message::Pong(p)).await?,
+                    return Ok(());
-                Some(Ok(Message::Close(_))) | None => return Ok(()),
+                }
-                Some(Err(e)) => return Err(e.into()),
+            }
-                _ => {}
+            msg = read.next() => {
-            },
+                last_rx = std::time::Instant::now();
+                match msg {
+                    Some(Ok(Message::Text(t))) => {
+                        // Run each op concurrently so long docker/PTY work never
+                        // blocks heartbeats, pongs, or other commands.
+                        let out = out_tx.clone();
+                        let ptys = ptys.clone();
+                        let text = t.to_string();
+                        tokio::spawn(async move { handle_frame(&text, &out, &ptys).await; });
+                    }
+                    Some(Ok(Message::Ping(p))) => write.send(Message::Pong(p)).await?,
+                    Some(Ok(Message::Close(_))) | None => return Ok(()),
+                    Some(Err(e)) => return Err(e.into()),
+                    _ => {}
+                }
+            }
         }
     }
 }

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

@@ -266,6 +266,9 @@ async fn run() -> Result<(), String> {
     // Expiry/retention sweep: expires stale auth/oauth rows and prunes old
     // journal/audit rows hourly so unbounded tables don't accumulate.
     cm_api::cleanup_sweeper::spawn(pool.clone(), std::time::Duration::from_secs(3600));
+    // Fleet backstop: a node whose heartbeats stop (without a clean channel
+    // close) goes offline within ~28s even if its control channel hangs.
+    cm_api::fleet::spawn_node_sweeper(pool.clone(), std::time::Duration::from_secs(8), 20);
 
     // Hosted identity (Clerk / OIDC): pin the issuer and load its JWKS.
     let auth_verifier = match config.auth.mode {

crates/cm-api/src/fleet.rs

@@ -23,6 +23,31 @@ use tokio::sync::{mpsc, oneshot, Mutex};
 
 const B64: base64::engine::general_purpose::GeneralPurpose = base64::engine::general_purpose::STANDARD;
 
+/// Monotonic per-connection id. A reconnecting daemon gets a fresh epoch so a
+/// stale channel's teardown can't clobber the newer connection's online status.
+static CONN_EPOCH: AtomicU64 = AtomicU64::new(1);
+
+/// Server keepalive: ping the daemon this often; if no inbound frame (incl. the
+/// matching pong) arrives within the deadline, treat the socket as dead.
+const PING_EVERY: Duration = Duration::from_secs(15);
+const IDLE_DEADLINE: Duration = Duration::from_secs(35);
+
+/// Backstop sweeper: flip any node whose heartbeat stopped (without a clean
+/// channel close) to offline. Heartbeats refresh `last_seen`, so a live node
+/// (5s heartbeat) is never older than the window and is never swept; a vanished
+/// node goes offline within `stale_secs` + one tick even if its channel hangs.
+pub fn spawn_node_sweeper(pool: PgPool, interval: Duration, stale_secs: i64) {
+    tokio::spawn(async move {
+        let mut tick = tokio::time::interval(interval);
+        loop {
+            tick.tick().await;
+            if let Err(e) = nodes::mark_stale_offline(&pool, stale_secs).await {
+                eprintln!("node_sweeper: mark_stale_offline failed: {e}");
+            }
+        }
+    });
+}
+
 /// The result of running a command on a node.
 #[derive(Debug, Clone)]
 pub struct ExecOutput {
@@ -36,6 +61,8 @@ struct NodeConn {
     /// Live terminal sessions: sid → sink for the browser bridge.
     pty_sinks: Mutex<HashMap<u64, mpsc::UnboundedSender<Vec<u8>>>>,
     next_id: AtomicU64,
+    /// Unique per physical connection — see `CONN_EPOCH`.
+    epoch: u64,
 }
 
 /// Registry of live daemon channels, keyed by node id.
@@ -129,7 +156,6 @@ impl NodeHub {
         let (tx, rx) = mpsc::unbounded_channel();
         conn.pty_sinks.lock().await.insert(sid, tx);
         let frame = json!({ "t": "pty_open", "sid": sid, "cols": cols, "rows": rows }).to_string();
-        eprintln!("[fleet-term] open_terminal sid={sid} node={id} cols={cols} rows={rows}");
         if conn.tx.send(frame).is_err() {
             conn.pty_sinks.lock().await.remove(&sid);
             return None;
@@ -223,91 +249,109 @@ struct HealthMsg {
 pub async fn run_channel(pool: PgPool, hub: Arc<NodeHub>, node_id: NodeId, socket: WebSocket) {
     let (mut ws_tx, mut ws_rx) = socket.split();
     let (tx, mut rx) = mpsc::unbounded_channel::<String>();
+    let epoch = CONN_EPOCH.fetch_add(1, Ordering::Relaxed);
     let conn = Arc::new(NodeConn {
         tx,
         pending: Mutex::new(HashMap::new()),
         pty_sinks: Mutex::new(HashMap::new()),
         next_id: AtomicU64::new(0),
+        epoch,
     });
+    // Inserting overwrites any stale conn for this node — dropping the old conn's
+    // `tx`, so its writer's `rx.recv()` returns None and that channel tears down.
     hub.conns.lock().await.insert(node_id, conn.clone());
     hub.online.lock().unwrap().insert(node_id);
 
-    let writer = async {
+    // One select loop carries outbound frames, inbound frames, and a keepalive
-        while let Some(frame) = rx.recv().await {
+    // ping. `last_inbound` tracks liveness: any frame (incl. Pong) refreshes it;
-            if ws_tx.send(Message::Text(frame.into())).await.is_err() {
+    // if it goes stale past the deadline, the socket is dead and we tear down.
-                break;
+    let mut ping_tick = tokio::time::interval(PING_EVERY);
-            }
+    ping_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-        }
+    let mut last_inbound = Instant::now();
-    };
+    loop {
-    let reader = async {
+        tokio::select! {
-        while let Some(Ok(msg)) = ws_rx.next().await {
+            frame = rx.recv() => match frame {
-            let Message::Text(t) = msg else { continue };
+                Some(f) => {
-            match serde_json::from_str::<Uplink>(t.as_str()) {
+                    if ws_tx.send(Message::Text(f.into())).await.is_err() {
-                Ok(Uplink::Heartbeat {
+                        break;
-                    version,
-                    tailscale_ip,
-                    hostname,
-                    local_ip,
-                    health,
-                }) => {
-                    let h = NodeHealth {
-                        cpu_pct: health.cpu_pct,
-                        mem_total: health.mem_total,
-                        mem_used: health.mem_used,
-                        mem_pressure: health.mem_pressure,
-                        swap_used: health.swap_used,
-                        disk_total: health.disk_total,
-                        disk_free: health.disk_free,
-                        load1: health.load1,
-                        load5: health.load5,
-                        load15: health.load15,
-                        container_count: health.container_count,
-                    };
-                    let _ = nodes::heartbeat(
-                        &pool,
-                        node_id,
-                        version.as_deref(),
-                        tailscale_ip.as_deref(),
-                        hostname.as_deref(),
-                        local_ip.as_deref(),
-                        &h,
-                    )
-                    .await;
-                }
-                Ok(Uplink::Result { id, ok, output }) => {
-                    if let Some(s) = conn.pending.lock().await.remove(&id) {
-                        let _ = s.send(ExecOutput { ok, output });
                     }
                 }
-                Ok(Uplink::PtyOut { sid, data }) => {
+                None => break,
-                    if let Ok(bytes) = B64.decode(&data) {
+            },
-                        let sink = conn.pty_sinks.lock().await.get(&sid).cloned();
+            _ = ping_tick.tick() => {
-                        eprintln!(
+                if last_inbound.elapsed() > IDLE_DEADLINE {
-                            "[fleet-term] pty_out sid={sid} bytes={} sink={}",
+                    break;
-                            bytes.len(),
+                }
-                            sink.is_some()
+                if ws_tx.send(Message::Ping(Vec::new().into())).await.is_err() {
-                        );
+                    break;
-                        if let Some(s) = sink {
+                }
-                            let _ = s.send(bytes);
+            },
+            msg = ws_rx.next() => {
+                let Some(Ok(msg)) = msg else { break };
+                last_inbound = Instant::now();
+                let Message::Text(t) = msg else { continue };
+                match serde_json::from_str::<Uplink>(t.as_str()) {
+                    Ok(Uplink::Heartbeat {
+                        version,
+                        tailscale_ip,
+                        hostname,
+                        local_ip,
+                        health,
+                    }) => {
+                        let h = NodeHealth {
+                            cpu_pct: health.cpu_pct,
+                            mem_total: health.mem_total,
+                            mem_used: health.mem_used,
+                            mem_pressure: health.mem_pressure,
+                            swap_used: health.swap_used,
+                            disk_total: health.disk_total,
+                            disk_free: health.disk_free,
+                            load1: health.load1,
+                            load5: health.load5,
+                            load15: health.load15,
+                            container_count: health.container_count,
+                        };
+                        let _ = nodes::heartbeat(
+                            &pool,
+                            node_id,
+                            version.as_deref(),
+                            tailscale_ip.as_deref(),
+                            hostname.as_deref(),
+                            local_ip.as_deref(),
+                            &h,
+                        )
+                        .await;
+                    }
+                    Ok(Uplink::Result { id, ok, output }) => {
+                        if let Some(s) = conn.pending.lock().await.remove(&id) {
+                            let _ = s.send(ExecOutput { ok, output });
                         }
                     }
+                    Ok(Uplink::PtyOut { sid, data }) => {
+                        if let Ok(bytes) = B64.decode(&data) {
+                            let sink = conn.pty_sinks.lock().await.get(&sid).cloned();
+                            if let Some(s) = sink {
+                                let _ = s.send(bytes);
+                            }
+                        }
+                    }
+                    Ok(Uplink::PtyExit { sid }) => {
+                        conn.pty_sinks.lock().await.remove(&sid);
+                    }
+                    Err(_) => {}
                 }
-                Ok(Uplink::PtyExit { sid }) => {
+            },
-                    conn.pty_sinks.lock().await.remove(&sid);
-                }
-                Err(_) => {}
-            }
         }
-    };
-
-    tokio::select! {
-        _ = writer => {},
-        _ = reader => {},
     }
 
-    hub.conns.lock().await.remove(&node_id);
+    // Teardown — only clear if we're STILL the registered connection. A daemon
-    hub.online.lock().unwrap().remove(&node_id);
+    // that reconnected has a newer epoch; a stale channel must not flip it offline.
-    let _ = nodes::set_status(&pool, node_id, "offline").await;
+    let mut conns = hub.conns.lock().await;
+    if conns.get(&node_id).map(|c| c.epoch) == Some(conn.epoch) {
+        conns.remove(&node_id);
+        hub.online.lock().unwrap().remove(&node_id);
+        let _ = nodes::set_status(&pool, node_id, "offline").await;
+    }
 }
 
 // ── RemoteDriver: run agent sandboxes on a connected node over the channel ────

crates/cm-api/src/lib.rs

@@ -123,7 +123,6 @@ pub fn router(state: AppState) -> Router {
         .route("/api/nodes/{id}/sandbox-check", post(routes::nodes::sandbox_check))
         .route("/api/nodes/{id}/terminal/ticket", post(routes::nodes::terminal_ticket))
         .route("/api/nodes/{id}/terminal/ws", get(routes::nodes::terminal_ws))
-        .route("/api/debug/node-pty/{id}", get(routes::nodes::debug_node_pty))
         .route("/api/nodes/{id}", delete(routes::nodes::remove))
         .route(
             "/api/fleet/tailscale",

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

@@ -136,39 +136,6 @@ pub async fn sandbox_check(
     }
 }
 
-#[derive(Deserialize)]
-pub struct DbgQuery {
-    pub dbg: String,
-}
-
-/// TEMP debug: open a terminal on a node and read ~2s of its output, no browser
-/// or auth. Lets us test the daemon→server PTY relay in isolation.
-pub async fn debug_node_pty(
-    State(state): State<AppState>,
-    Path(id): Path<Uuid>,
-    Query(q): Query<DbgQuery>,
-) -> Json<Value> {
-    if q.dbg != "clawdbg1" {
-        return Json(json!({ "ok": false, "error": "denied" }));
-    }
-    let node_id = NodeId::from(id);
-    let Some((sid, mut rx)) = state.node_hub.open_terminal(node_id, 120, 40).await else {
-        return Json(json!({ "ok": false, "error": "open_terminal None (node not connected)" }));
-    };
-    let mut total = 0usize;
-    let mut preview: Vec<u8> = Vec::new();
-    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
-    while let Ok(Some(bytes)) = tokio::time::timeout_at(deadline, rx.recv()).await {
-        total += bytes.len();
-        if preview.len() < 300 {
-            preview.extend_from_slice(&bytes);
-            preview.truncate(300);
-        }
-    }
-    state.node_hub.terminal_close(node_id, sid).await;
-    Json(json!({ "ok": true, "sid": sid, "bytes": total, "preview": String::from_utf8_lossy(&preview) }))
-}
-
 /// `DELETE /api/nodes/{id}` — deregister a node (workspace-scoped).
 pub async fn remove(
     State(state): State<AppState>,
@@ -212,9 +179,7 @@ pub async fn terminal_ticket(
     nodes::get(&state.pool, node_id, user.workspace_id)
         .await?
         .ok_or(ApiError::NotFound)?;
-    let online = state.node_hub.is_online(node_id).await;
+    if !state.node_hub.is_online(node_id).await {
-    eprintln!("[fleet-term] ticket request node={node_id} online={online}");
-    if !online {
         return Ok(Json(json!({ "error": "node is offline" })));
     }
     Ok(Json(json!({ "ticket": state.node_hub.mint_ticket(node_id).await })))
@@ -229,9 +194,7 @@ pub async fn terminal_ws(
     upgrade: WebSocketUpgrade,
 ) -> Response {
     let node_id = NodeId::from(id);
-    let redeemed = state.node_hub.redeem_ticket(&q.token).await;
+    match state.node_hub.redeem_ticket(&q.token).await {
-    eprintln!("[fleet-term] ws upgrade node={node_id} ticket_ok={}", redeemed == Some(node_id));
-    match redeemed {
         Some(t) if t == node_id => {}
         _ => return ApiError::Unauthorized.into_response(),
     }
@@ -249,21 +212,15 @@ struct TermCtrl {
 
 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 {
-        eprintln!("[fleet-term] bridge: open_terminal None (node offline?) node={node_id}");
         return;
     };
-    eprintln!("[fleet-term] bridge start node={node_id} sid={sid}");
     let (mut ws_tx, mut ws_rx) = socket.split();
     let to_browser = async {
-        let mut sent = 0usize;
         while let Some(bytes) = rx.recv().await {
-            sent += bytes.len();
             if ws_tx.send(Message::Binary(bytes.into())).await.is_err() {
-                eprintln!("[fleet-term] bridge: browser send failed after {sent} bytes");
                 break;
             }
         }
-        eprintln!("[fleet-term] bridge to_browser ended, {sent} bytes total");
     };
     let to_node = async {
         while let Some(Ok(msg)) = ws_rx.next().await {

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

@@ -21,9 +21,55 @@ function NodeShell({ nodeId }: { nodeId: string }) {
   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 ro: ResizeObserver | null = null;
     let resizeT: ReturnType<typeof setTimeout> | undefined;
+    let reconnectT: ReturnType<typeof setTimeout> | undefined;
+
+    const sendResize = () => {
+      if (ws?.readyState === WebSocket.OPEN && term && fit && hostRef.current?.clientWidth) {
+        fit.fit();
+        ws.send(JSON.stringify({ type: "resize", cols: term.cols, rows: term.rows }));
+      }
+    };
+
+    // Reconnecting transport. The host session is a persistent `tmux` server, so
+    // each reconnect re-attaches and tmux redraws the live screen — mosh-style
+    // snap-to-state over our WS, no byte-backlog replay. The xterm instance and
+    // its scrollback persist across drops (we never dispose it between tries).
+    const connect = async () => {
+      if (disposed || !term) return;
+      let ticket: string | undefined;
+      try {
+        const res = await fetch(`/api/nodes/${nodeId}/terminal/ticket`, { method: "POST" });
+        const body = res.ok ? ((await res.json()) as { ticket?: string; error?: string }) : {};
+        ticket = body.ticket;
+        if (!ticket) {
+          term.writeln(`\r\n\x1b[2m[${body.error ?? "node offline"} — retrying…]\x1b[0m`);
+          reconnectT = setTimeout(connect, 2000);
+          return;
+        }
+      } catch {
+        reconnectT = setTimeout(connect, 2000);
+        return;
+      }
+      if (disposed) return;
+      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.onmessage = (e) => {
+        if (typeof e.data === "string") term?.write(e.data);
+        else term?.write(new Uint8Array(e.data as ArrayBuffer));
+      };
+      // onerror is always followed by onclose; reconnect from there only.
+      ws.onclose = () => {
+        if (disposed) return;
+        term?.writeln("\r\n\x1b[2m[reconnecting…]\x1b[0m");
+        reconnectT = setTimeout(connect, 1500);
+      };
+    };
 
     (async () => {
       const [{ Terminal }, { FitAddon }] = await Promise.all([
@@ -37,65 +83,27 @@ function NodeShell({ nodeId }: { nodeId: string }) {
         cursorBlink: true,
         theme: { background: "#0a0a0c", foreground: "#d4d4d8" },
       });
-      const fit = new FitAddon();
+      fit = new FitAddon();
       term.loadAddon(fit);
       term.open(hostRef.current);
       fit.fit();
       term.focus();
-      console.log("[node-term] xterm opened", { w: hostRef.current?.clientWidth, h: hostRef.current?.clientHeight, cols: term.cols, rows: term.rows, node: nodeId });
-
-      console.log("[node-term] requesting ticket…");
-      const res = await fetch(`/api/nodes/${nodeId}/terminal/ticket`, { method: "POST" });
-      const body = res.ok ? ((await res.json()) as { ticket?: string; error?: string }) : {};
-      console.log("[node-term] ticket response", res.status, body.ticket ? "(ticket ok)" : body);
-      if (!body.ticket) {
-        term.writeln(`\r\n\x1b[31m${body.error ?? `could not open terminal (${res.status})`}\x1b[0m`);
-        return;
-      }
-      const proto = location.protocol === "https:" ? "wss:" : "ws:";
-      const url = `${proto}//${location.host}/api/nodes/${nodeId}/terminal/ws?token=${encodeURIComponent(body.ticket)}`;
-      console.log("[node-term] opening WS", url);
-      ws = new WebSocket(url);
-      ws.binaryType = "arraybuffer";
-      let rx = 0;
-      const sendResize = () => {
-        if (ws?.readyState === WebSocket.OPEN && term && hostRef.current?.clientWidth) {
-          fit.fit();
-          console.log("[node-term] resize ->", term.cols, "x", term.rows);
-          ws.send(JSON.stringify({ type: "resize", cols: term.cols, rows: term.rows }));
-        }
-      };
-      ws.onopen = () => {
-        console.log("[node-term] WS OPEN");
-        sendResize();
-      };
-      ws.onmessage = (e) => {
-        const n = typeof e.data === "string" ? e.data.length : (e.data as ArrayBuffer).byteLength;
-        rx += n;
-        if (rx === n || rx % 1000 < n) console.log("[node-term] WS data", n, "bytes (total", rx + ")");
-        if (typeof e.data === "string") term?.write(e.data);
-        else term?.write(new Uint8Array(e.data as ArrayBuffer));
-      };
-      ws.onerror = (ev) => console.log("[node-term] WS ERROR", ev);
-      ws.onclose = (ev) => {
-        console.log("[node-term] WS CLOSE code", ev.code, "reason", ev.reason || "(none)", "rxBytes", rx);
-        term?.writeln("\r\n\x1b[33m[disconnected]\x1b[0m");
-      };
       term.onData((d) => {
         if (ws?.readyState === WebSocket.OPEN) ws.send(new TextEncoder().encode(d));
       });
       // Debounce: the pull-out animates open, firing the observer on every pixel.
-      // Without this we'd spam dozens of resize frames (and SIGWINCH the PTY).
       ro = new ResizeObserver(() => {
         clearTimeout(resizeT);
         resizeT = setTimeout(sendResize, 150);
       });
       ro.observe(hostRef.current);
+      connect();
     })();
 
     return () => {
       disposed = true;
       clearTimeout(resizeT);
+      clearTimeout(reconnectT);
       ro?.disconnect();
       ws?.close();
       term?.dispose();

frontend/src/components/dashboard/fleet/FleetConsole.tsx

@@ -13,7 +13,7 @@ import { useQueryStates } from "nuqs";
 import { useFetchJson } from "@/lib/api/use-fetch";
 import { panelParsers } from "@/lib/url/panel-params";
 
-import { TailscaleSection, useNodes, type FleetNode } from "./FleetPanels";
+import { TailscaleSection, useNodes, isLive, displayStatus, type FleetNode } from "./FleetPanels";
 
 const mono = "'JetBrains Mono', ui-monospace, monospace";
 
@@ -62,7 +62,7 @@ export function InfraNav({
   onConnectHost: () => void;
 }) {
   const { nodes } = useNodes();
-  const online = nodes.filter((n) => n.status === "online").length;
+  const online = nodes.filter(isLive).length;
   const { data: ts } = useFetchJson<{ devices: TsDevice[] }>("/api/fleet/tailscale/devices");
   const tsCount = ts?.devices?.length ?? 0;
   const isLocal = view === "local";
@@ -141,18 +141,19 @@ function HostCard({ node, onRemoved }: { node: FleetNode; onRemoved: () => void
     });
   }, [sshHost]);
 
-  const accentBorder = node.status === "online" ? "rgba(95,208,138,.18)" : node.status === "pending" ? "rgba(232,196,106,.2)" : "rgba(255,255,255,.07)";
+  const st = displayStatus(node);
+  const accentBorder = st === "online" ? "rgba(95,208,138,.18)" : st === "pending" ? "rgba(232,196,106,.2)" : "rgba(255,255,255,.07)";
 
   return (
     <div style={{ borderRadius: 13, border: `1px solid ${accentBorder}`, background: "#0d0d10", padding: 15, display: "flex", flexDirection: "column" }}>
       <div style={{ display: "flex", alignItems: "center", gap: 9, marginBottom: 13 }}>
-        <span style={{ width: 9, height: 9, borderRadius: "50%", background: STATUS_COLOR[node.status], animation: node.status === "online" ? "cm-blink 1.6s infinite" : "none" }} />
+        <span style={{ width: 9, height: 9, borderRadius: "50%", background: STATUS_COLOR[st], animation: st === "online" ? "cm-blink 1.6s infinite" : "none" }} />
         <span style={{ fontSize: 14, fontWeight: 700, whiteSpace: "nowrap", overflow: "hidden", textOverflow: "ellipsis" }}>{node.hostname ?? node.name}</span>
         <span style={{ flex: 1 }} />
-        {node.status === "online" ? (
+        {st === "online" ? (
           <button type="button" onClick={() => setParams({ app: "terminal", node: node.id, device: "phone" })} title="Open terminal" aria-label="Open terminal" style={{ width: 26, height: 26, borderRadius: 7, border: "1px solid rgba(94,200,216,.3)", background: "rgba(94,200,216,.08)", color: "#5ec8d8", cursor: "pointer", display: "flex", alignItems: "center", justifyContent: "center" }}><Terminal size={13} /></button>
         ) : null}
-        <button type="button" onClick={remove} title="Remove" aria-label="Remove node" style={{ fontFamily: mono, fontSize: 9, color: STATUS_COLOR[node.status], background: "transparent", border: 0, cursor: "pointer", padding: "2px 4px" }}>{node.status.toUpperCase()}</button>
+        <button type="button" onClick={remove} title="Remove" aria-label="Remove node" style={{ fontFamily: mono, fontSize: 9, color: STATUS_COLOR[st], background: "transparent", border: 0, cursor: "pointer", padding: "2px 4px" }}>{st.toUpperCase()}</button>
       </div>
       <div style={{ fontFamily: mono, fontSize: 10, color: "#6a6a72", marginBottom: 14 }}>
         {node.localIp ?? "—"}{node.agentVersion ? ` · v${node.agentVersion}` : ""}
@@ -172,13 +173,13 @@ function HostCard({ node, onRemoved }: { node: FleetNode; onRemoved: () => void
         </>
       ) : null}
 
-      {node.status === "online" && sshHost ? (
+      {st === "online" && sshHost ? (
         <button type="button" onClick={copySsh} style={{ display: "flex", alignItems: "center", gap: 8, padding: "8px 11px", borderRadius: 8, background: "#070708", border: "1px solid rgba(255,255,255,.07)", cursor: "pointer", textAlign: "left" }}>
           <Server size={13} style={{ color: "#5ec8d8", flex: "none" }} />
           <span style={{ fontFamily: mono, fontSize: 11, color: "#cfcfd5", flex: 1, overflow: "hidden", textOverflow: "ellipsis", whiteSpace: "nowrap" }}>ssh {sshHost}</span>
           <span style={{ fontFamily: mono, fontSize: 9, color: copied ? "#5fd08a" : "#5ec8d8" }}>{copied ? "copied" : "copy"}</span>
         </button>
-      ) : node.status === "pending" ? (
+      ) : st === "pending" ? (
         <div style={{ display: "flex", alignItems: "center", gap: 8, padding: "8px 11px", borderRadius: 8, background: "rgba(232,196,106,.06)", border: "1px solid rgba(232,196,106,.2)" }}>
           <span style={{ width: 11, height: 11, border: "2px solid #e8b465", borderTopColor: "transparent", borderRadius: "50%", animation: "spin .8s linear infinite" }} />
           <span style={{ fontFamily: mono, fontSize: 10, color: "#e8b465" }}>waiting for daemon to dial home…</span>
@@ -201,7 +202,7 @@ function Stat({ label, value, tint }: { label: string; value: string; tint?: str
 export function FleetConsole({ view, onConnectHost }: { view: string; onConnectHost: () => void }) {
   const { nodes, refresh } = useNodes();
   const { data: ts } = useFetchJson<{ connected: boolean; tailnet: string | null; devices: TsDevice[] }>("/api/fleet/tailscale/devices");
-  const online = nodes.filter((n) => n.status === "online");
+  const online = nodes.filter(isLive);
   const totalMem = online.reduce((a, n) => a + (n.health?.memTotal ?? 0), 0);
   const totalDisk = online.reduce((a, n) => a + (n.health?.diskTotal ?? 0), 0);
   const containers = online.reduce((a, n) => a + (n.health?.containerCount ?? 0), 0);
@@ -286,7 +287,7 @@ export function FleetConsole({ view, onConnectHost }: { view: string; onConnectH
 /** Top-bar pill: live online/total host count. */
 export function FleetPill() {
   const { nodes } = useNodes();
-  const online = nodes.filter((n) => n.status === "online").length;
+  const online = nodes.filter(isLive).length;
   return (
     <div style={{ display: "flex", alignItems: "center", gap: 7, fontFamily: mono, fontSize: 11, color: "#5fd08a", padding: "5px 10px", border: "1px solid rgba(95,208,138,.25)", borderRadius: 7, background: "rgba(95,208,138,.06)" }}>
       <span style={{ width: 6, height: 6, borderRadius: "50%", background: "#5fd08a", animation: "cm-blink 1.6s infinite" }} />
@@ -298,7 +299,7 @@ export function FleetPill() {
 /** Thin always-on status bar (ambient system truth). */
 export function FleetStatusBar() {
   const { nodes } = useNodes();
-  const online = nodes.filter((n) => n.status === "online").length;
+  const online = nodes.filter(isLive).length;
   const { data: ts } = useFetchJson<{ connected: boolean }>("/api/fleet/tailscale");
   const cell = (dot: string, text: string) => (
     <span style={{ display: "inline-flex", alignItems: "center", gap: 6 }}><span style={{ width: 5, height: 5, borderRadius: "50%", background: dot }} />{text}</span>

frontend/src/components/dashboard/fleet/FleetPanels.tsx

@@ -48,6 +48,20 @@ const STATUS_COLOR: Record<FleetNode["status"], string> = {
   draining: "#ff8a7a",
 };
 
+/** A node is "live" if the channel reports online OR its last heartbeat is fresh
+ *  (<15s) — so a transient status-column flip never shows a healthy node down. */
+export function isLive(n: FleetNode): boolean {
+  if (n.status === "pending" || n.status === "draining") return false;
+  if (n.lastSeen != null && Date.now() / 1000 - n.lastSeen < 15) return true;
+  return n.status === "online";
+}
+
+/** The status to render: pending/draining pass through; otherwise online iff live. */
+export function displayStatus(n: FleetNode): FleetNode["status"] {
+  if (n.status === "pending" || n.status === "draining") return n.status;
+  return isLive(n) ? "online" : "offline";
+}
+
 function fmtBytes(b: number): string {
   if (b >= 1e12) return `${(b / 1e12).toFixed(1)} TB`;
   if (b >= 1e9) return `${(b / 1e9).toFixed(1)} GB`;
@@ -55,14 +69,32 @@ function fmtBytes(b: number): string {
   return `${b} B`;
 }
 
-/** Poll the workspace's nodes every 3s. */
+/** Stream the workspace's nodes live over SSE (status updates push instantly,
+ *  no 3s poll lag). `refresh` does a one-shot GET for an immediate nudge after
+ *  pairing a new host. */
 export function useNodes(): { nodes: FleetNode[]; refresh: () => void } {
-  const { data, refresh } = useFetchJson<{ nodes: FleetNode[] }>("/api/nodes");
+  const [nodes, setNodes] = useState<FleetNode[]>([]);
+  const refresh = useCallback(() => {
+    fetch("/api/nodes")
+      .then((r) => (r.ok ? r.json() : null))
+      .then((d: { nodes: FleetNode[] } | null) => {
+        if (d?.nodes) setNodes(d.nodes);
+      })
+      .catch(() => {});
+  }, []);
   useEffect(() => {
-    const t = setInterval(refresh, 3000);
+    refresh();
-    return () => clearInterval(t);
+    const es = new EventSource("/api/nodes/live");
+    es.addEventListener("nodes", (e) => {
+      try {
+        setNodes(JSON.parse((e as MessageEvent).data) as FleetNode[]);
+      } catch {
+        /* ignore malformed frame */
+      }
+    });
+    return () => es.close();
   }, [refresh]);
-  return { nodes: data?.nodes ?? [], refresh };
+  return { nodes, refresh };
 }
 
 function Bar({ label, pct, detail, color }: { label: string; pct: number; detail: string; color: string }) {