web P1: read-only dashboard (axum + SSE)

`sm121 web [--port 3110] [--bind] [--token] [--open]` — local dashboard
over the existing cluster machinery (design:
specs/tech-architecture/web-dashboard.md):

- Supervisor polls ClusterState::probe (the health logic) every 4 s and
  broadcasts state snapshots; `/events` forwards them over SSE;
  `/api/state` serves the latest (probes once before the first tick).
- Embedded single-file UI (src/web/ui.html, no build step): LED grid
  (reachability + occupancy per node, per cluster with free count),
  light/dark themes (persisted, prefers-color-scheme default), SSE
  status indicator, win-in/pulse animations.
- Token gate on /api + /events (Bearer header or ?token= for SSE
  clients); binding off-loopback without --token is refused.
- axum 0.8 added as the only new dependency.

Tests: snapshot builder, bus semantics, SSE type-tag wire format,
token gate (401/401/200), full server boot on an ephemeral port with a
fake unreachable node. Smoke-tested against the real fleet (dgx1
up+occupied, dgx2-4 offline).

P2 (actions/jobs), P3 (log windows + window manager), P4 (raw views)
remain per the plan.

AGENTS.md

@@ -17,11 +17,12 @@ See [`DEVELOPMENT.md`](DEVELOPMENT.md) for architecture, module layout, and dev
 ## Commands
 
 ```
-serve/stop/logs/health/doctor/exec/power/tmux/init-host-keys/config
+serve/stop/logs/health/doctor/exec/power/tmux/init-host-keys/config/web
 ```
 
-`sm121 web` (planned) — local axum + SSE dashboard (LEDs, toggles, live log
-windows, light/dark themes). Design: `specs/tech-architecture/web-dashboard.md`.
+`sm121 web` — local axum + SSE dashboard (P1: read-only LED board +
+themes; actions/logs/raw land in P2–P4). Design/progress:
+`specs/tech-architecture/web-dashboard.md`.
 
 Cluster: exact name, prefix (`4`→`4nodes-switch`), or `auto` (smallest with NCCL + matching `required_nodes`).
 

Cargo.lock

@@ -224,6 +224,58 @@ dependencies = [
  "fs_extra",
 ]
 
+[[package]]
+name = "axum"
+version = "0.8.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
+dependencies = [
+ "axum-core",
+ "bytes",
+ "form_urlencoded",
+ "futures-util",
+ "http",
+ "http-body",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
+ "itoa",
+ "matchit",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "serde_core",
+ "serde_json",
+ "serde_path_to_error",
+ "serde_urlencoded",
+ "sync_wrapper",
+ "tokio",
+ "tower",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "axum-core"
+version = "0.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "http",
+ "http-body",
+ "http-body-util",
+ "mime",
+ "pin-project-lite",
+ "sync_wrapper",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
 [[package]]
 name = "base16ct"
 version = "0.2.0"
@@ -1494,6 +1546,12 @@ version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
 
+[[package]]
+name = "httpdate"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
+
 [[package]]
 name = "hybrid-array"
 version = "0.4.12"
@@ -1519,6 +1577,7 @@ dependencies = [
  "http",
  "http-body",
  "httparse",
+ "httpdate",
  "itoa",
  "pin-project-lite",
  "smallvec",
@@ -1970,6 +2029,12 @@ dependencies = [
  "regex-automata",
 ]
 
+[[package]]
+name = "matchit"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
+
 [[package]]
 name = "md5"
 version = "0.7.0"
@@ -1988,6 +2053,12 @@ version = "2.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4"
 
+[[package]]
+name = "mime"
+version = "0.3.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
+
 [[package]]
 name = "miniz_oxide"
 version = "0.8.9"
@@ -3116,6 +3187,12 @@ version = "1.0.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
 
+[[package]]
+name = "ryu"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
+
 [[package]]
 name = "salsa20"
 version = "0.10.2"
@@ -3288,6 +3365,17 @@ dependencies = [
  "zmij",
 ]
 
+[[package]]
+name = "serde_path_to_error"
+version = "0.1.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457"
+dependencies = [
+ "itoa",
+ "serde",
+ "serde_core",
+]
+
 [[package]]
 name = "serde_spanned"
 version = "1.1.1"
@@ -3297,6 +3385,18 @@ dependencies = [
  "serde_core",
 ]
 
+[[package]]
+name = "serde_urlencoded"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
+dependencies = [
+ "form_urlencoded",
+ "itoa",
+ "ryu",
+ "serde",
+]
+
 [[package]]
 name = "serdect"
 version = "0.4.3"
@@ -3439,6 +3539,7 @@ name = "sm121"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "axum",
  "base64",
  "clap",
  "dirs",
@@ -3865,6 +3966,7 @@ dependencies = [
  "tokio",
  "tower-layer",
  "tower-service",
+ "tracing",
 ]
 
 [[package]]
@@ -3903,6 +4005,7 @@ version = "0.1.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
 dependencies = [
+ "log",
  "pin-project-lite",
  "tracing-attributes",
  "tracing-core",

Cargo.toml

@@ -22,3 +22,4 @@ walkdir = "2"
 dirs = "6"
 reqwest = { version = "0.13", default-features = false, features = ["json", "rustls"] }
 fluent-i18n = "0.1.0"
+axum = "0.8"

specs/tech-architecture/web-dashboard.md

@@ -199,7 +199,9 @@ toggle is one click away:
 1. **P1 — read-only board.** axum server, `/api/state`, supervisor poll,
    `/events` SSE (state only), UI shell: theme toggle + LED grid in a
    window. *Ship test: dashboard reflects a live serve/stop done from the
-   CLI within ~4 s.*
+   CLI within ~4 s.* — **DONE 2026-08-22** (commit pending; smoke-tested
+   against the real fleet: dgx1 up+occupied, dgx2-4 offline; SSE + token
+   gate + non-loopback refusal verified).
 2. **P2 — actions.** jobs registry, serve/stop/power POSTs, confirm flows,
    jobs window, job LEDs, dry-run preview.
 3. **P3 — logs & windows.** log attaches + ring buffers + SSE log events,
@@ -208,6 +210,11 @@ toggle is one click away:
 4. **P4 — raw & polish.** raw content windows (config/recipe/effective),
    copy buttons, animation pass, `--open`.
 
+P1 deviations from the plan: `Event` is a one-variant enum for now (job/log
+variants added with P2/P3); per-container rows are deferred to P2/P3 (the
+supervisor reuses `ClusterState::probe` as-is — node reachability +
+occupancy only); `--open` already implemented (trivial).
+
 ## Testing
 
 - Unit: event bus broadcast/late-joiner semantics; job single-flight

src/cli.rs

@@ -193,6 +193,23 @@ pub enum Commands {
     #[command(subcommand)]
     Config(ConfigCommands),
 
+    /// Launch the web dashboard (axum + SSE) on a local port
+    Web {
+        #[arg(long, default_value_t = 3100)]
+        port: u16,
+
+        #[arg(long, default_value = "127.0.0.1")]
+        bind: String,
+
+        /// Bearer token for data endpoints (required when --bind is not loopback)
+        #[arg(long)]
+        token: Option<String>,
+
+        /// Open the browser once the server is up
+        #[arg(long)]
+        open: bool,
+    },
+
     /// Build Docker image for a recipe on a remote node
     Build {
         #[arg(long, required = true, help = "Recipe name")]

src/main.rs

@@ -27,6 +27,7 @@ mod power;
 mod pre_serve;
 mod ssh;
 mod tmux;
+mod web;
 
 use crate::cli::{Cli, Commands, PowerAction};
 use crate::config::{
@@ -164,6 +165,14 @@ async fn main() -> Result<()> {
             crate::cli::ConfigCommands::Path => crate::gconfig::show_paths(&cli.config),
             crate::cli::ConfigCommands::InitGlobal => crate::gconfig::init_global(dry_run)?,
         },
+        Commands::Web {
+            port,
+            bind,
+            token,
+            open,
+        } => {
+            crate::web::run(config, port, bind, token, open).await?;
+        }
         Commands::Exec {
             cluster,
             recipe,

src/web/api.rs

@@ -0,0 +1,30 @@
+//! REST endpoints. P1: `GET /api/state` (latest snapshot; probes once if
+//! the supervisor hasn't ticked yet).
+
+use crate::cluster_state::ClusterState;
+use axum::{extract::State, response::Json};
+
+use super::AppState;
+use super::events::{Event, Snapshot, build_snapshot};
+
+pub(crate) async fn state(app: State<AppState>) -> Json<Snapshot> {
+    if app.last.lock().unwrap().is_none() {
+        // First request before the first supervisor tick: probe now so the
+        // UI gets data immediately (probe is read-only `docker ps` per node
+        // and degrades to `reachable: false` on failure). Two racing first
+        // requests may both probe — harmless (idempotent, last write wins).
+        let st = ClusterState::probe(&app.config).await;
+        let snap = build_snapshot(&app.config, &st);
+        let _ = app.bus.send(Event::State {
+            ts: snap.ts,
+            snapshot: snap.clone(),
+        });
+        *app.last.lock().unwrap() = Some(snap);
+    }
+    app.last
+        .lock()
+        .unwrap()
+        .as_ref()
+        .map(|s| Json(s.clone()))
+        .unwrap()
+}

src/web/events.rs

@@ -0,0 +1,153 @@
+//! SSE event bus + state snapshot for the dashboard.
+//!
+//! One multiplexed `broadcast` channel; `/events` forwards every event and
+//! the UI routes by `type`. P1 emits `state` only — `job` and `log`
+//! variants are added in P2/P3 when their producers exist.
+
+use crate::cluster_state::ClusterState;
+use crate::config::Config;
+use serde::Serialize;
+
+/// One SSE event (serialized as `{"type":"state", ...}` on the wire).
+#[derive(Serialize, Clone, Debug)]
+#[serde(tag = "type", rename_all = "lowercase")]
+pub enum Event {
+    State { ts: u64, snapshot: Snapshot },
+}
+
+/// Per-node view for the LED grid.
+#[derive(Serialize, Clone, Debug)]
+pub struct NodeSnap {
+    pub name: String,
+    pub ssh_ip: String,
+    pub reachable: bool,
+    pub occupied: bool,
+}
+
+/// Per-cluster view.
+#[derive(Serialize, Clone, Debug)]
+pub struct ClusterSnap {
+    pub name: String,
+    pub nodes: Vec<NodeSnap>,
+    pub free: u32,
+}
+
+/// The full dashboard snapshot (also the `GET /api/state` body).
+#[derive(Serialize, Clone, Debug)]
+pub struct Snapshot {
+    pub ts: u64,
+    pub clusters: Vec<ClusterSnap>,
+}
+
+/// Pure snapshot builder — unit-testable without SSH.
+pub fn build_snapshot(config: &Config, st: &ClusterState) -> Snapshot {
+    let mut clusters = Vec::with_capacity(config.clusters.len());
+    for c in &config.clusters {
+        let mut nodes = Vec::with_capacity(c.nodes.len());
+        let mut free = 0u32;
+        for n in &c.nodes {
+            let s = st.nodes.get(&n.name);
+            let reachable = s.is_some_and(|x| x.reachable);
+            let occupied = s.is_some_and(|x| x.occupied);
+            if reachable && !occupied {
+                free += 1;
+            }
+            nodes.push(NodeSnap {
+                name: n.name.clone(),
+                ssh_ip: n.ssh_ip.clone(),
+                reachable,
+                occupied,
+            });
+        }
+        clusters.push(ClusterSnap {
+            name: c.name.clone(),
+            nodes,
+            free,
+        });
+    }
+    Snapshot {
+        ts: std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_secs())
+            .unwrap_or(0),
+        clusters,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::{Cluster, Node};
+    use std::collections::HashMap;
+
+    fn node(name: &str, ip: &str, rank: u32) -> Node {
+        Node {
+            name: name.into(),
+            ssh_ip: ip.into(),
+            ib_ip: None,
+            rank,
+            host_key: None,
+            entity: None,
+        }
+    }
+
+    #[test]
+    fn snapshot_reflects_probe_state() {
+        let config = Config {
+            ssh_user: "u".into(),
+            ssh_key_path: "/k".into(),
+            clusters: vec![
+                Cluster {
+                    name: "a".into(),
+                    master_ib_ip: None,
+                    nodes: vec![node("n0", "10.0.0.1", 0), node("n1", "10.0.0.2", 1)],
+                    extra_entities: vec![],
+                },
+                Cluster {
+                    name: "b".into(),
+                    master_ib_ip: None,
+                    nodes: vec![node("n2", "10.0.0.3", 0)],
+                    extra_entities: vec![],
+                },
+            ],
+            nccl: HashMap::new(),
+            hf_cache_dir: None,
+            init_port: 25000,
+            before_serve: HashMap::new(),
+            exec_aliases: HashMap::new(),
+            home_assistant: None,
+            global_mods_dir: None,
+            drop_caches: true,
+            mount_aliases: HashMap::new(),
+            recipe_sources: Vec::new(),
+            cluster_sources: Vec::new(),
+        };
+        let mut st = ClusterState::default();
+        st.nodes.insert(
+            "n0".into(),
+            crate::cluster_state::NodeState {
+                reachable: true,
+                occupied: true,
+            },
+        );
+        st.nodes.insert(
+            "n1".into(),
+            crate::cluster_state::NodeState {
+                reachable: true,
+                occupied: false,
+            },
+        );
+        // n2 absent from the probe → treated as unreachable.
+        let snap = build_snapshot(&config, &st);
+        assert_eq!(snap.clusters.len(), 2);
+        let a = &snap.clusters[0];
+        assert_eq!(a.name, "a");
+        assert!(a.nodes[0].reachable);
+        assert!(a.nodes[0].occupied);
+        assert!(a.nodes[1].reachable && !a.nodes[1].occupied);
+        assert_eq!(a.free, 1);
+        let b = &snap.clusters[1];
+        assert!(!b.nodes[0].reachable);
+        assert_eq!(b.free, 0);
+    }
+}

src/web/mod.rs

@@ -0,0 +1,169 @@
+//! `sm121 web` — axum + SSE dashboard.
+//!
+//! P1 (read-only): `/` serves the embedded UI, `/api/state` the snapshot,
+//! `/events` a multiplexed SSE stream fed by a supervisor task that polls
+//! `cluster_state::probe` every 4 s. Data endpoints are token-gated when
+//! `--token` is set (required when binding off-loopback).
+//!
+//! Design: `specs/tech-architecture/web-dashboard.md`.
+
+mod api;
+pub mod events;
+
+use crate::cluster_state::ClusterState;
+use crate::config::Config;
+use crate::web::events::{Event, build_snapshot};
+use axum::{
+    Router,
+    extract::State,
+    http::{HeaderMap, Request, StatusCode},
+    middleware::{self, Next},
+    response::{
+        Response,
+        sse::{Event as SseEvent, Sse},
+    },
+    routing::get,
+};
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
+use tokio::sync::broadcast;
+use tokio::time::Duration;
+
+const UI_HTML: &str = include_str!("ui.html");
+const POLL_SECS: u64 = 4;
+const BUS_CAP: usize = 1024;
+
+#[derive(Clone)]
+pub struct AppState {
+    pub config: Arc<Config>,
+    pub bus: broadcast::Sender<Event>,
+    pub token: Option<String>,
+    /// Latest snapshot (written by the supervisor; `/api/state` serves it).
+    pub last: Arc<Mutex<Option<events::Snapshot>>>,
+}
+
+/// `GET /events` — forward every bus event as an SSE message.
+async fn events_stream(
+    app: State<AppState>,
+) -> Sse<impl futures::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
+    let rx = app.bus.subscribe();
+    Sse::new(futures::stream::unfold(rx, |mut rx| async move {
+        loop {
+            match rx.recv().await {
+                Ok(ev) => {
+                    let json = serde_json::to_string(&ev).unwrap_or_default();
+                    return Some((Ok(SseEvent::default().data(json)), rx));
+                }
+                Err(broadcast::error::RecvError::Lagged(n)) => {
+                    tracing::debug!("SSE client lagged {n} events");
+                    // Same receiver is still valid — keep reading.
+                }
+                Err(broadcast::error::RecvError::Closed) => return None,
+            }
+        }
+    }))
+}
+
+/// Bearer-token gate for data endpoints (`/` stays public — it is static
+/// HTML with no secrets). Accepts `Authorization: Bearer <t>` or `?token=<t>`
+/// (SSE `EventSource` cannot set headers, so the UI uses the query form).
+async fn token_guard(
+    State(app): State<AppState>,
+    headers: HeaderMap,
+    req: Request<axum::body::Body>,
+    next: Next,
+) -> Result<Response, (StatusCode, String)> {
+    if let Some(expected) = &app.token {
+        // ponytail: raw query param, no percent-decoding — tokens are
+        // operator-generated; decode if exotic tokens are ever used.
+        let from_query = req.uri().query().and_then(|q| {
+            q.split('&').find_map(|kv| {
+                kv.split_once('=')
+                    .and_then(|(k, v)| (k == "token").then_some(v))
+            })
+        });
+        let got = headers
+            .get("authorization")
+            .and_then(|v| v.to_str().ok())
+            .and_then(|v| v.strip_prefix("Bearer "))
+            .or(from_query);
+        if got != Some(expected.as_str()) {
+            return Err((StatusCode::UNAUTHORIZED, "unauthorized".into()));
+        }
+    }
+    Ok(next.run(req).await)
+}
+
+/// Supervisor: poll the fleet and broadcast a state snapshot every tick.
+async fn supervisor(app: AppState) {
+    loop {
+        let st = ClusterState::probe(&app.config).await;
+        let snap = build_snapshot(&app.config, &st);
+        let _ = app.bus.send(Event::State {
+            ts: snap.ts,
+            snapshot: snap.clone(),
+        });
+        *app.last.lock().unwrap() = Some(snap);
+        tokio::time::sleep(Duration::from_secs(POLL_SECS)).await;
+    }
+}
+
+fn is_loopback(addr: SocketAddr) -> bool {
+    addr.ip().is_loopback()
+}
+
+/// Build the router (separate from `run` so tests can drive it without
+/// binding a port).
+pub fn router(app: AppState) -> Router {
+    let protected = Router::new()
+        .route("/api/state", get(api::state))
+        .route("/events", get(events_stream))
+        .layer(middleware::from_fn_with_state(app.clone(), token_guard));
+    Router::new()
+        .route("/", get(|| async { UI_HTML }))
+        .merge(protected)
+        .with_state(app)
+}
+
+/// Run the dashboard until the process exits.
+pub async fn run(
+    config: Config,
+    port: u16,
+    bind: String,
+    token: Option<String>,
+    open: bool,
+) -> anyhow::Result<()> {
+    let addr: SocketAddr = format!("{bind}:{port}").parse()?;
+    if !is_loopback(addr) && token.is_none() {
+        anyhow::bail!(
+            "refusing to bind {bind} (non-loopback) without --token — the dashboard can act on the cluster"
+        );
+    }
+    let (tx, _) = broadcast::channel(BUS_CAP);
+    let app = AppState {
+        config: Arc::new(config),
+        bus: tx,
+        token,
+        last: Arc::new(Mutex::new(None)),
+    };
+    tokio::spawn(supervisor(app.clone()));
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+    if open {
+        let url = format!("http://{addr}/");
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(400)).await;
+            let prog = if cfg!(target_os = "macos") {
+                "open"
+            } else {
+                "xdg-open"
+            };
+            let _ = std::process::Command::new(prog).arg(&url).spawn();
+        });
+    }
+    tracing::info!("sm121 web: http://{addr}/ (poll every {POLL_SECS}s)");
+    axum::serve(listener, router(app)).await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests;

src/web/tests.rs

@@ -0,0 +1,142 @@
+//! Web module tests: bus semantics, token gate, and a full server boot on an
+//! ephemeral port with a fake (unreachable) node.
+
+use crate::config::{Cluster, Config, Node};
+use crate::web::events::{Event, build_snapshot};
+use crate::web::{AppState, router};
+use std::collections::HashMap;
+use std::net::SocketAddr;
+
+/// Config with one node at 127.0.0.1:2222 — connection refused instantly,
+/// machine-independent (no real SSH, no hardware).
+fn fake_config() -> Config {
+    Config {
+        ssh_user: "u".into(),
+        ssh_key_path: "/nonexistent/key".into(),
+        clusters: vec![Cluster {
+            name: "fake".into(),
+            master_ib_ip: None,
+            nodes: vec![Node {
+                name: "n0".into(),
+                ssh_ip: "127.0.0.1:2222".into(),
+                ib_ip: None,
+                rank: 0,
+                host_key: None,
+                entity: None,
+            }],
+            extra_entities: vec![],
+        }],
+        nccl: HashMap::new(),
+        hf_cache_dir: None,
+        init_port: 25000,
+        before_serve: HashMap::new(),
+        exec_aliases: HashMap::new(),
+        home_assistant: None,
+        global_mods_dir: None,
+        drop_caches: true,
+        mount_aliases: HashMap::new(),
+        recipe_sources: Vec::new(),
+        cluster_sources: Vec::new(),
+    }
+}
+
+fn app_state(token: Option<&str>) -> AppState {
+    let (tx, _) = tokio::sync::broadcast::channel(16);
+    AppState {
+        config: std::sync::Arc::new(fake_config()),
+        bus: tx,
+        token: token.map(String::from),
+        last: std::sync::Arc::new(std::sync::Mutex::new(None)),
+    }
+}
+
+async fn boot() -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
+    let app = app_state(None);
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let handle = tokio::spawn(async move {
+        axum::serve(listener, router(app)).await.unwrap();
+    });
+    (addr, std::sync::Arc::new(handle))
+}
+
+#[tokio::test]
+async fn api_state_reports_unreachable_node() {
+    let (addr, handle) = boot().await;
+    let body: serde_json::Value = reqwest::get(format!("http://{addr}/api/state"))
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(body["clusters"].as_array().unwrap().len(), 1);
+    assert_eq!(body["clusters"][0]["name"], "fake");
+    assert_eq!(body["clusters"][0]["nodes"][0]["reachable"], false);
+    assert_eq!(body["clusters"][0]["free"], 0);
+    handle.abort();
+}
+
+#[tokio::test]
+async fn token_gate_enforces_when_set() {
+    let app = app_state(Some("s3cret"));
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, router(app)).await.unwrap();
+    });
+    let base = format!("http://{addr}/api/state");
+    // No token → 401.
+    assert_eq!(reqwest::get(&base).await.unwrap().status(), 401);
+    // Wrong token → 401.
+    assert_eq!(
+        reqwest::get(format!("{base}?token=wrong"))
+            .await
+            .unwrap()
+            .status(),
+        401
+    );
+    // Right token (query form — what SSE clients use) → 200.
+    assert_eq!(
+        reqwest::get(format!("{base}?token=s3cret"))
+            .await
+            .unwrap()
+            .status(),
+        200
+    );
+}
+
+#[tokio::test]
+async fn bus_delivers_state_events() {
+    let (tx, mut rx) = tokio::sync::broadcast::channel::<Event>(8);
+    let snap = crate::web::events::Snapshot {
+        ts: 1,
+        clusters: Vec::new(),
+    };
+    tx.send(Event::State {
+        ts: snap.ts,
+        snapshot: snap.clone(),
+    })
+    .unwrap();
+    let got = rx.recv().await.unwrap();
+    match got {
+        Event::State { ts, snapshot } => {
+            assert_eq!(ts, 1);
+            assert!(snapshot.clusters.is_empty());
+        }
+    }
+}
+
+#[tokio::test]
+async fn snapshot_serializes_with_type_tag() {
+    let ev = Event::State {
+        ts: 7,
+        snapshot: build_snapshot(
+            &fake_config(),
+            &crate::cluster_state::ClusterState::default(),
+        ),
+    };
+    let json = serde_json::to_value(&ev).unwrap();
+    assert_eq!(json["type"], "state");
+    assert_eq!(json["ts"], 7);
+    assert!(json["snapshot"].is_object());
+}

src/web/ui.html

@@ -0,0 +1,164 @@
+<!doctype html>
+<html data-theme="dark">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>sm121</title>
+<style>
+  :root[data-theme="dark"] {
+    --bg: #0d1117; --panel: #161b22; --fg: #e6edf3; --muted: #8b949e;
+    --border: #30363d; --accent: #58a6ff;
+    --led-on: #3fb950; --led-off: #f85149; --led-busy: #d29922; --led-free: #30363d;
+    --glow-on: 0 0 8px 2px rgba(63,185,80,.55); --glow-busy: 0 0 8px 2px rgba(210,153,34,.55);
+  }
+  :root[data-theme="light"] {
+    --bg: #f6f8fa; --panel: #ffffff; --fg: #1f2328; --muted: #656d76;
+    --border: #d0d7de; --accent: #0969da;
+    --led-on: #1a7f37; --led-off: #cf222e; --led-busy: #9a6700; --led-free: #d0d7de;
+    --glow-on: 0 0 8px 2px rgba(26,127,55,.35); --glow-busy: 0 0 8px 2px rgba(154,103,0,.35);
+  }
+  * { box-sizing: border-box; }
+  body {
+    margin: 0; background: var(--bg); color: var(--fg);
+    font: 14px/1.45 ui-sans-serif, system-ui, "SF Pro", sans-serif;
+    transition: background .25s, color .25s;
+  }
+  header {
+    display: flex; align-items: center; gap: 12px;
+    padding: 10px 16px; border-bottom: 1px solid var(--border);
+    background: var(--panel);
+  }
+  header h1 { font-size: 15px; font-weight: 600; margin: 0; letter-spacing: .3px; }
+  #sse {
+    display: inline-flex; align-items: center; gap: 6px;
+    font-size: 12px; color: var(--muted);
+  }
+  #sse .dot { width: 8px; height: 8px; border-radius: 50%; background: var(--led-free); transition: background .3s; }
+  #sse.ok .dot { background: var(--led-on); box-shadow: var(--glow-on); }
+  #sse.err .dot { background: var(--led-off); }
+  #theme {
+    margin-left: auto; cursor: pointer; border: 1px solid var(--border);
+    background: var(--panel); color: var(--fg); border-radius: 6px;
+    padding: 4px 10px; font-size: 12px; transition: transform .12s;
+  }
+  #theme:active { transform: scale(.95); }
+
+  .stage { padding: 20px; }
+  .win {
+    background: var(--panel); border: 1px solid var(--border); border-radius: 10px;
+    box-shadow: 0 8px 28px rgba(0,0,0,.25); overflow: hidden;
+    animation: win-in .28s cubic-bezier(.2,.9,.3,1.2);
+  }
+  @keyframes win-in { from { opacity: 0; transform: scale(.94) translateY(6px); } to { opacity: 1; transform: none; } }
+  .titlebar {
+    padding: 7px 12px; font-size: 12px; color: var(--muted);
+    border-bottom: 1px solid var(--border); display: flex; gap: 8px; align-items: center;
+    user-select: none; cursor: grab;
+  }
+  .titlebar b { color: var(--fg); font-weight: 600; }
+  .grid { padding: 10px 12px; }
+  .cluster { margin-bottom: 14px; }
+  .cluster-head {
+    display: flex; gap: 10px; align-items: baseline; font-size: 13px;
+    font-weight: 600; margin: 0 0 6px 2px;
+  }
+  .cluster-head .free { font-weight: 400; font-size: 12px; color: var(--muted); }
+  .node {
+    display: grid; grid-template-columns: 14px 14px 190px 1fr; gap: 10px; align-items: center;
+    padding: 6px 8px; border-radius: 8px; margin-bottom: 2px;
+    transition: background .15s;
+  }
+  .node:hover { background: color-mix(in srgb, var(--fg) 6%, transparent); }
+  .led { width: 10px; height: 10px; border-radius: 50%; background: var(--led-free); transition: background .35s, box-shadow .35s; }
+  .led.on  { background: var(--led-on);  box-shadow: var(--glow-on); }
+  .led.off { background: var(--led-off); }
+  .led.busy{ background: var(--led-busy); box-shadow: var(--glow-busy); }
+  .led.pulse { animation: pulse 1.1s ease-in-out infinite; }
+  @keyframes pulse { 50% { opacity: .35; } }
+  .nname { font-weight: 500; font-size: 13px; }
+  .nmeta { font-size: 12px; color: var(--muted); font-family: ui-monospace, monospace; }
+  .nstate { font-size: 12px; }
+  .nstate .up { color: var(--led-on); } .nstate .down { color: var(--led-off); }
+  .nstate .occupied { color: var(--led-busy); }
+  .empty { color: var(--muted); padding: 18px; text-align: center; }
+</style>
+</head>
+<body>
+<header>
+  <h1>sm121</h1>
+  <span id="sse"><span class="dot"></span><span id="sse-text">connecting…</span></span>
+  <button id="theme">🌙 dark</button>
+</header>
+<div class="stage">
+  <div class="win" style="max-width: 760px">
+    <div class="titlebar"><b>state</b><span id="updated"></span></div>
+    <div class="grid" id="grid"><div class="empty">waiting for state…</div></div>
+  </div>
+</div>
+<script>
+"use strict";
+// ---- theme -------------------------------------------------------------
+const root = document.documentElement;
+const saved = localStorage.getItem("theme");
+const prefersLight = matchMedia("(prefers-color-scheme: light)").matches;
+setTheme(saved || (prefersLight ? "light" : "dark"));
+document.getElementById("theme").onclick = () => setTheme(root.dataset.theme === "dark" ? "light" : "dark");
+function setTheme(t) {
+  root.dataset.theme = t;
+  localStorage.setItem("theme", t);
+  document.getElementById("theme").textContent = t === "dark" ? "🌙 dark" : "☀️ light";
+}
+// ---- state grid ----------------------------------------------------------
+const grid = document.getElementById("grid");
+const updated = document.getElementById("updated");
+function esc(s) { const d = document.createElement("i"); d.textContent = s; return d.innerHTML; }
+function renderState(snap) {
+  if (!snap.clusters || !snap.clusters.length) {
+    grid.innerHTML = '<div class="empty">no clusters in config</div>';
+    return;
+  }
+  grid.innerHTML = snap.clusters.map(c => `
+    <div class="cluster">
+      <div class="cluster-head">${esc(c.name)}<span class="free">${c.free}/${c.nodes.length} free</span></div>
+      ${c.nodes.map(n => `
+        <div class="node">
+          <span class="led ${n.reachable ? "on" : "off"} ${!n.reachable ? "pulse" : ""}" title="reachability"></span>
+          <span class="led ${n.occupied ? "busy" : "free"}" title="occupancy"></span>
+          <span class="nname">${esc(n.name)}</span>
+          <span class="nmeta">${esc(n.ssh_ip)}</span>
+        </div>
+        <div class="node" style="padding-top:0;margin-left:38px">
+          <span></span><span></span>
+          <span class="nstate">
+            ${n.reachable
+              ? (n.occupied ? '<span class="occupied">● occupied</span>' : '<span class="up">○ free</span>')
+              : '<span class="down">● unreachable</span>'}
+          </span><span></span>
+        </div>`).join("")}
+    </div>`).join("");
+  updated.textContent = "updated " + new Date(snap.ts * 1000).toLocaleTimeString();
+}
+// ---- SSE -----------------------------------------------------------------
+const TOKEN = new URLSearchParams(location.search).get("token") || "";
+const sseEl = document.getElementById("sse"), sseText = document.getElementById("sse-text");
+function connect() {
+  const url = "/events" + (TOKEN ? `?token=${encodeURIComponent(TOKEN)}` : "");
+  const es = new EventSource(url);
+  es.onopen = () => { sseEl.classList.add("ok"); sseEl.classList.remove("err"); sseText.textContent = "live"; };
+  es.onerror = () => { sseEl.classList.add("err"); sseEl.classList.remove("ok"); sseText.textContent = "reconnecting…"; };
+  es.onmessage = (e) => {
+    try {
+      const ev = JSON.parse(e.data);
+      if (ev.type === "state") renderState(ev.snapshot);
+    } catch (_) {}
+  };
+}
+// Initial snapshot, then live updates.
+fetch("/api/state" + (TOKEN ? `?token=${encodeURIComponent(TOKEN)}` : ""))
+  .then(r => r.ok ? r.json() : null)
+  .then(snap => snap && renderState(snap))
+  .catch(() => {});
+connect();
+</script>
+</body>
+</html>