Fleet P0: node registry + daemon + health + connect-host wizard

Users can connect their own local-hardware nodes into a fleet. Each node runs a
new Rust daemon that dials home over an outbound WebSocket, reports host health,
and runs commands we send.

Backend:
- migrations/0018_fleet_nodes.sql: nodes + node_health tables + agent_containers
  (node_id, workspace_id) index. cm-domain NodeId.
- cm-db repo/nodes.rs: create/auth/list+health/get/heartbeat/set_status/delete
  (unchecked sqlx, no .sqlx regen).
- cm-api fleet.rs NodeHub: live daemon channels (node_id→sender) + the WS channel
  runner (heartbeat→DB upsert, exec request/response framing). routes/nodes.rs:
  POST /pair, GET /nodes, SSE /nodes/live, POST /{id}/exec-test, DELETE /{id},
  WS /nodes/agent (token-auth). Wired into AppState + router.

Daemon (new crate crates/bins/clawmates-node):
- sysinfo host metrics (cpu/mem/pressure/swap/disk/load/containers), outbound WSS
  dial + reconnect, heartbeat loop, exec command handling, tailscale-ip probe.
  install.sh convenience installer.

Frontend:
- Fleet sidebar item + FleetOverview + LocalHardware node-health cards (live via
  /api/nodes, 3s poll) + ConnectHostWizard (install → verify connection →
  exec-test). InfraStage dispatches fleet/local; default selection = fleet.

Deferred: P1 (BYO Tailscale + network metrics), P2 (RemoteDriver + placement so
agents actually run on connected nodes).

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

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

@@ -12,6 +12,7 @@ pub mod files;
 pub mod gateway;
 pub mod health;
 pub mod identity;
+pub mod nodes;
 pub mod oauth;
 pub mod orgs;
 pub mod routines;

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

@@ -0,0 +1,153 @@
+//! Fleet node registry HTTP/SSE/WS routes: pair a new node, list nodes with live
+//! health, stream health updates, run a verification command, deregister, and the
+//! daemon's outbound control channel.
+
+use std::convert::Infallible;
+use std::time::Duration;
+
+use axum::extract::ws::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 serde::Deserialize;
+use serde_json::{json, Value};
+use uuid::Uuid;
+
+use crate::fleet::run_channel;
+use crate::{ApiError, AppState, Authed};
+
+/// An unguessable control-channel token (two time-ordered UUIDs).
+fn gen_token() -> String {
+    format!("{}{}", Uuid::now_v7().simple(), Uuid::now_v7().simple())
+}
+
+fn node_json(n: &nodes::NodeRow) -> Value {
+    json!({
+        "id": n.id,
+        "name": n.name,
+        "status": n.status,
+        "agentVersion": n.agent_version,
+        "tailscaleIp": n.tailscale_ip,
+        "lastSeen": n.last_seen.map(|t| t.unix_timestamp()),
+        "createdAt": n.created_at.unix_timestamp(),
+        "health": n.health.as_ref().map(|h| json!({
+            "cpuPct": h.cpu_pct,
+            "memTotal": h.mem_total,
+            "memUsed": h.mem_used,
+            "memPressure": h.mem_pressure,
+            "swapUsed": h.swap_used,
+            "diskTotal": h.disk_total,
+            "diskFree": h.disk_free,
+            "load1": h.load1,
+            "load5": h.load5,
+            "load15": h.load15,
+            "containerCount": h.container_count,
+        })),
+    })
+}
+
+#[derive(Deserialize)]
+pub struct PairReq {
+    pub name: Option<String>,
+}
+
+/// `POST /api/nodes/pair` — register a pending node + mint its control-channel
+/// token. The frontend builds the `curl … | bash` install command (it knows its
+/// own origin); we just return the token + node id.
+pub async fn pair(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Json(req): Json<PairReq>,
+) -> Result<Json<Value>, ApiError> {
+    let token = gen_token();
+    let name = req
+        .name
+        .filter(|s| !s.trim().is_empty())
+        .unwrap_or_else(|| "New node".to_owned());
+    let id = nodes::create(&state.pool, user.workspace_id, &name, &token).await?;
+    Ok(Json(json!({ "id": id, "token": token })))
+}
+
+/// `GET /api/nodes` — list the workspace's nodes with their latest health.
+pub async fn list(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+) -> Result<Json<Value>, ApiError> {
+    let rows = nodes::list(&state.pool, user.workspace_id).await?;
+    Ok(Json(json!({ "nodes": rows.iter().map(node_json).collect::<Vec<_>>() })))
+}
+
+/// `GET /api/nodes/live` — SSE stream of the node list + health (2s poll).
+pub async fn live(State(state): State<AppState>, Authed(user): Authed) -> impl IntoResponse {
+    let pool = state.pool.clone();
+    let ws = user.workspace_id;
+    let stream = async_stream::stream! {
+        loop {
+            if let Ok(rows) = nodes::list(&pool, ws).await {
+                let arr: Vec<_> = rows.iter().map(node_json).collect();
+                let data = serde_json::to_string(&arr).unwrap_or_else(|_| "[]".to_owned());
+                yield Ok::<_, Infallible>(Event::default().event("nodes").data(data));
+            }
+            tokio::time::sleep(Duration::from_secs(2)).await;
+        }
+    };
+    Sse::new(stream).keep_alive(KeepAlive::default())
+}
+
+/// `POST /api/nodes/{id}/exec-test` — run a verification command on the node and
+/// return its output (wizard step 3: "we can run commands on your behalf").
+pub async fn exec_test(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Value>, ApiError> {
+    let node_id = NodeId::from(id);
+    // Scope: the node must belong to the caller's workspace.
+    let node = nodes::get(&state.pool, node_id, user.workspace_id)
+        .await?
+        .ok_or(ApiError::NotFound)?;
+    let cmd = vec![
+        "sh".to_owned(),
+        "-lc".to_owned(),
+        "uname -a; echo '---'; docker version --format 'docker {{.Server.Version}}' 2>/dev/null || echo 'docker: not found'".to_owned(),
+    ];
+    match state.node_hub.exec(node_id, &cmd).await {
+        Ok(out) => Ok(Json(json!({ "ok": out.ok, "output": out.output, "node": node.name }))),
+        Err(e) => Ok(Json(json!({ "ok": false, "output": e, "node": node.name }))),
+    }
+}
+
+/// `DELETE /api/nodes/{id}` — deregister a node (workspace-scoped).
+pub async fn remove(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Value>, ApiError> {
+    nodes::delete(&state.pool, NodeId::from(id), user.workspace_id).await?;
+    Ok(Json(json!({ "ok": true })))
+}
+
+#[derive(Deserialize)]
+pub struct AgentQuery {
+    pub token: String,
+}
+
+/// `GET /api/nodes/agent?token=…` — the daemon's outbound control channel. The
+/// WS handshake can't carry a bearer header, so the daemon authenticates with
+/// its node token in the query string (like the Terminal WS ticket).
+pub async fn agent_ws(
+    State(state): State<AppState>,
+    Query(q): Query<AgentQuery>,
+    upgrade: WebSocketUpgrade,
+) -> Response {
+    let auth = nodes::auth(&state.pool, &q.token).await.ok().flatten();
+    let Some((node_id, _workspace_id)) = auth else {
+        return ApiError::Unauthorized.into_response();
+    };
+    let pool = state.pool.clone();
+    let hub = state.node_hub.clone();
+    upgrade.on_upgrade(move |socket| run_channel(pool, hub, node_id, socket))
+}