Fleet P2a: in-dashboard remote terminal (PTY over the WSS channel)

You can now open a real shell on any connected node from the dashboard — the
daemon spawns a host PTY and streams it over the existing outbound control
channel (no inbound port, no Tailscale brokering needed).

Daemon:
- portable-pty host shell sessions: pty_open/pty_in/pty_resize/pty_close ops; a
  reader thread streams base64 pty_out frames. Outbound frames now funnel through
  one mpsc channel so PTY output and heartbeats interleave.

cm-api NodeHub:
- per-connection pty_sinks + sid multiplexing; open_terminal/terminal_input/
  terminal_resize/terminal_close; in-memory single-use terminal tickets (the
  browser WS can't carry a bearer, and the session is instance-local anyway).
- routes/nodes.rs: POST /api/nodes/{id}/terminal/ticket + GET .../terminal/ws
  (bridges browser xterm <-> node PTY: binary = keystrokes, text = resize).

Frontend:
- NodeTerminal xterm modal (reuses the agent Terminal's xterm setup); a Terminal
  button on each online node card opens a shell.

This proves the bidirectional streaming-over-channel mechanism the RemoteDriver
will reuse. Remaining P2: RemoteDriver + placement (run agent workloads on nodes).

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

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

@@ -3,20 +3,22 @@
 //! daemon's outbound control channel.
 
 use std::convert::Infallible;
+use std::sync::Arc;
 use std::time::Duration;
 
-use axum::extract::ws::WebSocketUpgrade;
+use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
 use axum::extract::{Path, Query, State};
 use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::{IntoResponse, Response};
 use axum::Json;
 use cm_db::repo::nodes;
 use cm_domain::NodeId;
+use futures::{SinkExt, StreamExt};
 use serde::Deserialize;
 use serde_json::{json, Value};
 use uuid::Uuid;
 
-use crate::fleet::run_channel;
+use crate::fleet::{run_channel, NodeHub};
 use crate::{ApiError, AppState, Authed};
 
 /// An unguessable control-channel token (two time-ordered UUIDs).
@@ -146,3 +148,80 @@ pub async fn agent_ws(
     let hub = state.node_hub.clone();
     upgrade.on_upgrade(move |socket| run_channel(pool, hub, node_id, socket))
 }
+
+/// `POST /api/nodes/{id}/terminal/ticket` — mint a single-use terminal ticket
+/// (the browser WS handshake can't carry a bearer header).
+pub async fn terminal_ticket(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Value>, ApiError> {
+    let node_id = NodeId::from(id);
+    nodes::get(&state.pool, node_id, user.workspace_id)
+        .await?
+        .ok_or(ApiError::NotFound)?;
+    if !state.node_hub.is_online(node_id).await {
+        return Ok(Json(json!({ "error": "node is offline" })));
+    }
+    Ok(Json(json!({ "ticket": state.node_hub.mint_ticket(node_id).await })))
+}
+
+/// `GET /api/nodes/{id}/terminal/ws?ticket=…` — bridge a browser xterm to a host
+/// shell on the node (PTY proxied over the daemon's control channel).
+pub async fn terminal_ws(
+    State(state): State<AppState>,
+    Path(id): Path<Uuid>,
+    Query(q): Query<AgentQuery>,
+    upgrade: WebSocketUpgrade,
+) -> Response {
+    let node_id = NodeId::from(id);
+    match state.node_hub.redeem_ticket(&q.token).await {
+        Some(t) if t == node_id => {}
+        _ => return ApiError::Unauthorized.into_response(),
+    }
+    let hub = state.node_hub.clone();
+    upgrade.on_upgrade(move |socket| bridge_terminal(hub, node_id, socket))
+}
+
+#[derive(Deserialize)]
+struct TermCtrl {
+    #[serde(rename = "type")]
+    kind: String,
+    cols: u16,
+    rows: u16,
+}
+
+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 {
+        return;
+    };
+    let (mut ws_tx, mut ws_rx) = socket.split();
+    let to_browser = async {
+        while let Some(bytes) = rx.recv().await {
+            if ws_tx.send(Message::Binary(bytes.into())).await.is_err() {
+                break;
+            }
+        }
+    };
+    let to_node = async {
+        while let Some(Ok(msg)) = ws_rx.next().await {
+            match msg {
+                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()) {
+                        if c.kind == "resize" {
+                            hub.terminal_resize(node_id, sid, c.cols, c.rows).await;
+                        }
+                    }
+                }
+                Message::Close(_) => break,
+                _ => {}
+            }
+        }
+    };
+    tokio::select! {
+        _ = to_browser => {},
+        _ = to_node => {},
+    }
+    hub.terminal_close(node_id, sid).await;
+}