diff --git a/crates/bins/clawmates-server/src/main.rs b/crates/bins/clawmates-server/src/main.rs index cc50454..bb92c44 100644 --- a/crates/bins/clawmates-server/src/main.rs +++ b/crates/bins/clawmates-server/src/main.rs @@ -185,14 +185,21 @@ async fn run() -> Result<(), String> { } else { None }; - let terminals = std::sync::Arc::new(cm_runtime::TerminalManager::new( - driver, - pool.clone(), - "local", - &config.sandbox.terminal_image, - config.sandbox.terminal_egress, - drives, - )); + let terminals = std::sync::Arc::new( + cm_runtime::TerminalManager::new( + driver, + pool.clone(), + "local", + &config.sandbox.terminal_image, + config.sandbox.terminal_egress, + drives, + ) + // An agent placed on a fleet node runs its terminal there too + // (beside its sandbox), sharing the agent's node-local drives. + .with_node_provider(std::sync::Arc::new( + cm_api::fleet::HubDriverProvider::new(node_hub.clone()), + )), + ); // Boot reconciliation: any sandbox the engine still holds is an // orphan from a dead process (we track none yet) — remove them // before warming so a crash/redeploy can't leak containers. diff --git a/crates/cm-api/src/fleet.rs b/crates/cm-api/src/fleet.rs index f2d33e4..ac1462e 100644 --- a/crates/cm-api/src/fleet.rs +++ b/crates/cm-api/src/fleet.rs @@ -168,20 +168,48 @@ impl NodeHub { } /// 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) { + /// `container` (+ `session`) targets `docker exec` into an agent container on + /// the node (the node-placed agent terminal); both `None` ⇒ the host shell. + pub async fn open_pty( + &self, + id: NodeId, + sid: u64, + cols: u16, + rows: u16, + container: Option<&str>, + session: Option<&str>, + ) { if let Some(conn) = self.get(id).await { - let _ = conn.tx.send( - json!({ "t": "pty_open", "sid": sid, "cols": cols, "rows": rows }).to_string(), - ); + let mut frame = json!({ "t": "pty_open", "sid": sid, "cols": cols, "rows": rows }); + if let Some(c) = container { + frame["container"] = json!(c); + } + if let Some(s) = session { + frame["session"] = json!(s); + } + let _ = conn.tx.send(frame.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) { + /// `container`/`session` target a node-placed agent container as in `open_pty`. + pub async fn webrtc_offer( + &self, + id: NodeId, + sid: u64, + sdp: &str, + container: Option<&str>, + session: Option<&str>, + ) { if let Some(conn) = self.get(id).await { - let _ = conn - .tx - .send(json!({ "t": "webrtc_offer", "sid": sid, "sdp": sdp }).to_string()); + let mut frame = json!({ "t": "webrtc_offer", "sid": sid, "sdp": sdp }); + if let Some(c) = container { + frame["container"] = json!(c); + } + if let Some(s) = session { + frame["session"] = json!(s); + } + let _ = conn.tx.send(frame.to_string()); } } diff --git a/crates/cm-api/src/routes/nodes.rs b/crates/cm-api/src/routes/nodes.rs index 33f22e0..5e76cf8 100644 --- a/crates/cm-api/src/routes/nodes.rs +++ b/crates/cm-api/src/routes/nodes.rs @@ -258,9 +258,11 @@ async fn bridge_terminal(hub: Arc, node_id: NodeId, socket: WebSocket) 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, + // Host shell (no container) — the Infra node terminal. + "fallback" => hub.open_pty(node_id, sid, cols, rows, None, None).await, "webrtc_offer" => { - hub.webrtc_offer(node_id, sid, c.sdp.as_deref().unwrap_or("")).await + hub.webrtc_offer(node_id, sid, c.sdp.as_deref().unwrap_or(""), None, None) + .await } "webrtc_ice" => { hub.webrtc_ice( diff --git a/crates/cm-api/src/routes/terminal.rs b/crates/cm-api/src/routes/terminal.rs index c5d62c6..e07e15b 100644 --- a/crates/cm-api/src/routes/terminal.rs +++ b/crates/cm-api/src/routes/terminal.rs @@ -13,13 +13,14 @@ use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::Json; use cm_db::repo::audit::Actor; -use cm_domain::{AgentId, WorkspaceId}; +use cm_domain::{AgentId, NodeId, WorkspaceId}; use futures::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use serde_json::json; use sqlx::{PgPool, Row}; use tokio::io::AsyncWriteExt; +use crate::fleet::NodeHub; use crate::routes::claws::workspace_agent; use crate::{ApiError, AppState, Authed}; @@ -83,6 +84,9 @@ async fn redeem_ticket( #[derive(Serialize)] pub struct TicketResponse { pub ticket: String, + /// The fleet-node UUID this agent's terminal is placed on, or null = gateway- + /// local. The browser uses the WebRTC transport for a node, plain WS for local. + pub node: Option, } /// `POST /api/terminal/{id}/ticket` — owner-gated; mints a short-lived ticket. @@ -115,6 +119,14 @@ pub async fn ticket( Err(_) => "there".to_string(), }; let ticket = issue_ticket(&state.pool, agent_id, user.workspace_id, &label).await?; + // Where the terminal is (or would be) placed — null = gateway-local. + let node = match state.runtime.terminals() { + Some(tm) => match tm.placement_node_for(agent_id).await { + n if n == "local" => None, + n => Some(n), + }, + None => None, + }; cm_db::repo::audit::append( &state.pool, user.workspace_id, @@ -125,7 +137,7 @@ pub async fn ticket( json!({}), ) .await?; - Ok(Json(TicketResponse { ticket })) + Ok(Json(TicketResponse { ticket, node })) } /// GET /api/terminal/{id}/tabs — the user's saved tab layout (or null). @@ -233,6 +245,19 @@ pub async fn ws( // MOTD greeting injected into the PTY session (read by the container .zshrc). let env = vec![format!("CLAWMATES_USER={label}")]; let session = sanitize_session(q.session.as_deref()); + // Node-placed agent → bridge to the node's daemon (WebRTC direct + WS relay), + // execing into the agent's terminal container there. Local stays unchanged. + if tm.placement_node_for(agent_id).await != "local" { + if let Ok((node_id, container)) = tm.placement_for(workspace_id, agent_id).await { + if let Ok(uuid) = node_id.parse::() { + let hub = state.node_hub.clone(); + let nid = NodeId::from(uuid); + return upgrade + .on_upgrade(move |socket| bridge_node(socket, hub, nid, container, session)); + } + } + // Couldn't place remotely (node dropped, etc.) — fall through to local. + } upgrade.on_upgrade(move |socket| bridge(socket, tm, workspace_id, agent_id, env, session)) } @@ -316,3 +341,102 @@ async fn bridge( } tm.detach(agent_id).await; } + +/// Browser→server control frames on the node-placed terminal WS: resize, the +/// `fallback` to open the WS-relay PTY, or WebRTC signaling (offer/ICE/close). +#[derive(Deserialize)] +struct NodeCtrl { + #[serde(rename = "type")] + kind: String, + cols: Option, + rows: Option, + sdp: Option, + candidate: Option, + sdp_mid: Option, + sdp_mline_index: Option, +} + +/// Bridge the browser terminal to a NODE-placed agent terminal container: relays +/// PTY bytes + WebRTC signaling to/from the node's daemon, which `docker exec`s a +/// tmux into `container`. Mirrors the Infra node terminal bridge, but targets the +/// agent's container (so the shell shares the agent's node-local ~/drives) and +/// carries the tmux `session` (one per tab). The same WS is the WebRTC signaling +/// channel and the WS-relay fallback path. +async fn bridge_node( + socket: WebSocket, + hub: Arc, + node_id: NodeId, + container: String, + session: String, +) { + 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 { + loop { + tokio::select! { + 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 (the DataChannel carries + // its own input when direct). + Message::Binary(b) => hub.terminal_input(node_id, sid, b.as_ref()).await, + Message::Text(t) => { + let Ok(c) = serde_json::from_str::(t.as_str()) else { + continue; + }; + 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, Some(&container), Some(&session)) + .await + } + "webrtc_offer" => { + hub.webrtc_offer( + node_id, + sid, + c.sdp.as_deref().unwrap_or(""), + Some(&container), + Some(&session), + ) + .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, + _ => {} + } + } + }; + tokio::select! { + _ = to_browser => {}, + _ = to_node => {}, + } + hub.terminal_close(node_id, sid).await; +} diff --git a/crates/cm-runtime/src/sandboxes.rs b/crates/cm-runtime/src/sandboxes.rs index 450db0b..0fec0fd 100644 --- a/crates/cm-runtime/src/sandboxes.rs +++ b/crates/cm-runtime/src/sandboxes.rs @@ -217,9 +217,16 @@ impl SandboxManager { (handle, self.node_id.clone()) } else { // Remote fleet node: provision directly via its driver (no warm pool). + // Mount the agent's per-agent node-local drive volume so the sandbox + // shares files with the agent's terminal (+ other containers) on that + // node. The egress browser sandbox stays driveless. + let mut spec = self.spec(); + if !self.egress { + spec.mounts = vec![crate::terminals::node_local_drive_mount(agent_id)]; + } let handle = self .driver_for(&node) - .provision(&self.spec()) + .provision(&spec) .await .map_err(|e| format!("remote sandbox provision failed: {e}"))?; (handle, node) diff --git a/crates/cm-runtime/src/terminals.rs b/crates/cm-runtime/src/terminals.rs index 7ff7c70..0cd73d7 100644 --- a/crates/cm-runtime/src/terminals.rs +++ b/crates/cm-runtime/src/terminals.rs @@ -18,9 +18,24 @@ use cm_domain::{AgentId, WorkspaceId}; use cm_sandbox::{DriveMount, PtySession, SandboxDriver, SandboxHandle, SandboxKind, SandboxSpec}; use sqlx::PgPool; +use crate::NodeDriverProvider; + /// The `agent_containers.kind` discriminator for terminal containers. const KIND: &str = "terminal"; +/// The per-agent node-local drive volume mounted at `~/drives` for a node-placed +/// agent — auto-created by Docker on first mount (no subpath). All of the agent's +/// containers on that node mount the SAME volume, so they share files. Both the +/// terminal and the sandbox use this exact mount (see `SandboxManager`). +pub(crate) fn node_local_drive_mount(agent_id: AgentId) -> DriveMount { + DriveMount { + volume: format!("clawmates_agent_{}", agent_id.as_uuid().simple()), + subpath: String::new(), + target: "/home/agent/drives".to_string(), + read_only: false, + } +} + /// Where the agent's Files drives live, so the Terminal can mount them. #[derive(Clone)] pub struct DriveConfig { @@ -34,12 +49,16 @@ pub struct DriveConfig { pub struct TerminalManager { driver: Arc, pool: PgPool, - /// Placement node this manager's driver provisions onto ("local" today). + /// This manager's LOCAL placement id ("local"); remote placements come from + /// `node_provider` keyed by a fleet node's id. node_id: String, image: String, egress: bool, /// Drive mounts; None disables the ~/drives mapping. drives: Option, + /// Resolves drivers for remote fleet nodes (None ⇒ local-only deployment), so + /// an agent placed on a node runs its terminal there too. + node_provider: Option>, } impl std::fmt::Debug for TerminalManager { @@ -67,9 +86,74 @@ impl TerminalManager { image: image.to_owned(), egress, drives, + node_provider: None, } } + /// Wire a fleet-node driver provider so an agent placed on a connected node + /// runs its terminal container there (sharing the agent's node-local drives). + /// Default (unset) = local-only, unchanged. + pub fn with_node_provider(mut self, provider: Arc) -> TerminalManager { + self.node_provider = Some(provider); + self + } + + /// The driver that owns containers on `node_id`: the local driver for "local" + /// (or when the node isn't connected), else the node's remote driver. + fn driver_for(&self, node_id: &str) -> Arc { + if node_id != self.node_id { + if let Some(d) = self.node_provider.as_ref().and_then(|p| p.driver(node_id)) { + return d; + } + } + self.driver.clone() + } + + /// Where a NEW terminal container for this agent should run: the agent's + /// workspace placement node if it's connected (and not draining), else "local" + /// — mirroring `SandboxManager` so the terminal lands beside the sandbox. + async fn placement_node(&self, agent_id: AgentId) -> String { + match cm_db::repo::workspace_placement::for_agent(&self.pool, agent_id).await { + Ok(Some(node)) if node != self.node_id => { + if let Ok(nid) = node.parse::() { + if matches!( + cm_db::repo::nodes::status_of(&self.pool, cm_domain::NodeId::from(nid)).await, + Ok(Some(ref s)) if s == "draining" + ) { + return self.node_id.clone(); + } + } + if self.node_provider.as_ref().and_then(|p| p.driver(&node)).is_some() { + return node; + } + self.node_id.clone() + } + _ => self.node_id.clone(), + } + } + + /// Where the agent's terminal is / would run, WITHOUT provisioning: the node + /// of an existing container, else the computed placement. For the ticket's + /// `node` hint (the browser uses the WebRTC-vs-WS transport accordingly). + pub async fn placement_node_for(&self, agent_id: AgentId) -> String { + if let Ok(Some(row)) = cm_db::repo::agent_containers::get(&self.pool, agent_id, KIND).await { + return row.node_id; + } + self.placement_node(agent_id).await + } + + /// Ensure the agent's terminal container exists (on its placement node) and + /// return `(node_id, container_name)` — what the node-bridge route needs to + /// `docker exec` into it. `node_id` is "local" or a fleet-node UUID. + pub async fn placement_for( + &self, + workspace_id: WorkspaceId, + agent_id: AgentId, + ) -> Result<(String, String), String> { + let (handle, node_id) = self.ensure(workspace_id, agent_id).await?; + Ok((node_id, handle.name)) + } + /// The three Files drives mounted read-write at `~/drives/*`, each a per-agent /// subpath of the shared volume (subpath = isolation). Creates the subdirs /// first so the mount doesn't fail on an empty drive. @@ -106,8 +190,17 @@ impl TerminalManager { &self, workspace_id: WorkspaceId, agent_id: AgentId, + node: &str, ) -> Result { let short = uuid::Uuid::now_v7().simple().to_string(); + // Local: per-agent subpaths of the shared gateway volume (today's path). + // Remote: a single per-agent node-local volume at ~/drives, shared with + // the agent's sandbox on that node. + let mounts = if node == self.node_id { + self.drive_mounts(workspace_id, agent_id).await + } else { + vec![node_local_drive_mount(agent_id)] + }; let spec = SandboxSpec { name: format!("tc-term-{}", &short[short.len() - 12..]), image: self.image.clone(), @@ -117,9 +210,9 @@ impl TerminalManager { pids_limit: 512, egress: self.egress, kind: SandboxKind::Terminal, - mounts: self.drive_mounts(workspace_id, agent_id).await, + mounts, }; - self.driver + self.driver_for(node) .provision(&spec) .await .map_err(|e| format!("terminal provision failed: {e}")) @@ -132,33 +225,35 @@ impl TerminalManager { &self, workspace_id: WorkspaceId, agent_id: AgentId, - ) -> Result { + ) -> Result<(SandboxHandle, String), String> { if let Ok(Some(row)) = cm_db::repo::agent_containers::get(&self.pool, agent_id, KIND).await { + let driver = self.driver_for(&row.node_id); let handle = SandboxHandle { id: row.container_id.clone(), name: row.name.clone(), }; - if self.driver.health(&handle).await.unwrap_or(false) { - return Ok(handle); + if driver.health(&handle).await.unwrap_or(false) { + return Ok((handle, row.node_id)); } // Recorded but dead: clean both the container and the stale row. - let _ = self.driver.destroy(&handle).await; + let _ = driver.destroy(&handle).await; let _ = cm_db::repo::agent_containers::delete(&self.pool, agent_id, KIND).await; } - let handle = self.provision_one(workspace_id, agent_id).await?; + let node = self.placement_node(agent_id).await; + let handle = self.provision_one(workspace_id, agent_id, &node).await?; cm_db::repo::agent_containers::upsert( &self.pool, agent_id, KIND, - &self.node_id, + &node, &handle.id, &handle.name, ) .await .map_err(|e| format!("registry upsert failed: {e}"))?; - Ok(handle) + Ok((handle, node)) } /// Open an interactive login zsh in the agent's terminal container. `env` @@ -173,12 +268,14 @@ impl TerminalManager { env: &[String], tmux_session: &str, ) -> Result { - let handle = self.ensure(workspace_id, agent_id).await?; + let (handle, node_id) = self.ensure(workspace_id, agent_id).await?; // tmux attach-or-create: the named session + its processes survive a WS // disconnect and resume on reconnect. Different session names are // independent tabs sharing the agent's one container (+ its ~/drives). + // Local attach uses the local driver (unchanged); a remote terminal is + // served via the node WebRTC/WS bridge, not here. let session = self - .driver + .driver_for(&node_id) .attach_pty( &handle, // -A attach-or-create; -D detaches any stale client on reattach so diff --git a/crates/cm-sandbox/src/docker.rs b/crates/cm-sandbox/src/docker.rs index 4f89533..970259c 100644 --- a/crates/cm-sandbox/src/docker.rs +++ b/crates/cm-sandbox/src/docker.rs @@ -97,10 +97,18 @@ impl SandboxDriver for DockerDriver { source: Some(m.volume.clone()), typ: Some(MountTypeEnum::VOLUME), read_only: Some(m.read_only), - volume_options: Some(MountVolumeOptions { - subpath: Some(m.subpath.clone()), - ..Default::default() - }), + // An empty subpath ⇒ mount the whole volume at the target + // (the per-agent node-local drive volume, auto-created by + // Docker). A non-empty subpath is per-agent isolation within + // the shared gateway volume. + volume_options: if m.subpath.is_empty() { + None + } else { + Some(MountVolumeOptions { + subpath: Some(m.subpath.clone()), + ..Default::default() + }) + }, ..Default::default() }) .collect(), diff --git a/frontend/src/components/computer/apps/TerminalApp.tsx b/frontend/src/components/computer/apps/TerminalApp.tsx index 4318b5f..5b1939e 100644 --- a/frontend/src/components/computer/apps/TerminalApp.tsx +++ b/frontend/src/components/computer/apps/TerminalApp.tsx @@ -13,7 +13,7 @@ import type { Agent } from "@/lib/api/schemas"; import { panelParsers } from "@/lib/url/panel-params"; import { useSubHeader } from "./AppShell"; -import { useResilientTerminal, wsConnector } from "./terminal/core"; +import { agentTerminalConnector, useResilientTerminal, type TermMode } from "./terminal/core"; import "@xterm/xterm/css/xterm.css"; @@ -279,15 +279,10 @@ function TerminalTab({ useEffect(() => { activeRef.current = active; }); + const [mode, setMode] = useState("connecting"); const { hostRef, refit } = useResilientTerminal( { - connect: wsConnector(async () => { - const res = await fetch(`/api/terminal/${agent.id}/ticket`, { method: "POST" }); - if (!res.ok) return null; - const { ticket } = (await res.json()) as { ticket: string }; - const proto = location.protocol === "https:" ? "wss:" : "ws:"; - return `${proto}//${location.host}/api/terminal/${agent.id}/ws?ticket=${encodeURIComponent(ticket)}&session=${encodeURIComponent(session)}`; - }), + connect: agentTerminalConnector(agent.id, session, setMode), visible: () => activeRef.current, autoFocus: true, }, @@ -302,10 +297,21 @@ function TerminalTab({ }, [active, refit]); return ( -
+
+ {/* Transport badge — shown only when the container is node-placed. */} + {mode === "direct" || mode === "relayed" ? ( +
+ {mode === "direct" ? "⚡ direct" : "relayed"} +
+ ) : null} +
+
); } diff --git a/frontend/src/components/computer/apps/infra/NodeTerminalApp.tsx b/frontend/src/components/computer/apps/infra/NodeTerminalApp.tsx index 33c9a4c..b41f979 100644 --- a/frontend/src/components/computer/apps/infra/NodeTerminalApp.tsx +++ b/frontend/src/components/computer/apps/infra/NodeTerminalApp.tsx @@ -13,11 +13,11 @@ import { useFetchJson } from "@/lib/api/use-fetch"; import { panelParsers } from "@/lib/url/panel-params"; import type { FleetNode } from "@/components/dashboard/fleet/FleetPanels"; -import { nodeWebrtcConnector, useResilientTerminal } from "../terminal/core"; +import { nodeWebrtcConnector, useResilientTerminal, type TermMode } from "../terminal/core"; import "@xterm/xterm/css/xterm.css"; -type Transport = "connecting" | "direct" | "relayed"; +type Transport = TermMode; /** xterm bridged to a node's host shell, preferring a direct WebRTC DataChannel. * Transport + xterm lifecycle live in the shared resilient-terminal core. */ diff --git a/frontend/src/components/computer/apps/terminal/core.ts b/frontend/src/components/computer/apps/terminal/core.ts index fe16bc5..aaadd45 100644 --- a/frontend/src/components/computer/apps/terminal/core.ts +++ b/frontend/src/components/computer/apps/terminal/core.ts @@ -213,7 +213,13 @@ const ICE_SERVERS: RTCIceServer[] = [{ urls: ["stun:stun.l.google.com:19302"] }] * (browser↔node, LAN speed) and falls back to the gateway WS relay if no direct * path forms within ~2.5s. The same WS carries the WebRTC signaling. `onMode` * reports the live transport so callers can show a direct/relayed indicator. */ -export function nodeWebrtcConnector(nodeId: string, onMode?: (m: "connecting" | "direct" | "relayed") => void): TermConnector { +export type TermMode = "connecting" | "direct" | "relayed" | "local"; + +/** WebRTC transport over a signaling WebSocket at `getUrl()` (which mints the + * ticket + builds the wss URL): direct DataChannel browser↔node, with the + * gateway WS relay as automatic fallback. Endpoint-agnostic — works for the node + * host shell and the node-placed agent container alike. */ +export function webrtcConnector(getUrl: () => Promise, onMode?: (m: TermMode) => void): TermConnector { return ({ term, onClosed }) => new Promise((resolve) => { let settled = false; @@ -337,15 +343,13 @@ export function nodeWebrtcConnector(nodeId: string, onMode?: (m: "connecting" | goRelayed(); } }; - void fetch(`/api/nodes/${nodeId}/terminal/ticket`, { method: "POST" }) - .then((r) => (r.ok ? r.json() : {})) - .then((body: { ticket?: string }) => { - if (!body.ticket) { + void getUrl() + .then((url) => { + if (!url) { resolve(null); return; } - const proto = location.protocol === "https:" ? "wss:" : "ws:"; - ws = new WebSocket(`${proto}//${location.host}/api/nodes/${nodeId}/terminal/ws?token=${encodeURIComponent(body.ticket)}`); + ws = new WebSocket(url); ws.binaryType = "arraybuffer"; ws.onopen = () => { ready(); @@ -371,3 +375,33 @@ export function nodeWebrtcConnector(nodeId: string, onMode?: (m: "connecting" | .catch(() => resolve(null)); }); } + +/** WebRTC shell on a fleet node's HOST (the Infra node terminal). */ +export function nodeWebrtcConnector(nodeId: string, onMode?: (m: TermMode) => void): TermConnector { + return webrtcConnector(async () => { + const res = await fetch(`/api/nodes/${nodeId}/terminal/ticket`, { method: "POST" }); + if (!res.ok) return null; + const { ticket } = (await res.json()) as { ticket?: string }; + if (!ticket) return null; + const proto = location.protocol === "https:" ? "wss:" : "ws:"; + return `${proto}//${location.host}/api/nodes/${nodeId}/terminal/ws?token=${encodeURIComponent(ticket)}`; + }, onMode); +} + +/** The agent terminal: mint the ticket, then pick the transport from the + * response — a node-placed container → WebRTC (LAN speed, signaling over the + * agent WS); a gateway-local container → plain WS (today's path). */ +export function agentTerminalConnector(agentId: string, session: string, onMode?: (m: TermMode) => void): TermConnector { + return (ctx) => + fetch(`/api/terminal/${agentId}/ticket`, { method: "POST" }) + .then((r) => (r.ok ? r.json() : null)) + .then((body: { ticket?: string; node?: string | null } | null) => { + if (!body?.ticket) return null; + const proto = location.protocol === "https:" ? "wss:" : "ws:"; + const url = `${proto}//${location.host}/api/terminal/${agentId}/ws?ticket=${encodeURIComponent(body.ticket)}&session=${encodeURIComponent(session)}`; + if (body.node) return webrtcConnector(async () => url, onMode)(ctx); + onMode?.("local"); + return wsConnector(async () => url)(ctx); + }) + .catch(() => null); +}