web: replay cached snapshot on SSE connect (live indicator no longer waits a poll tick) + jobs window groups running containers one row per launcher job with its node list

specs/tech-architecture/web-dashboard.md

@@ -96,10 +96,12 @@ Newline-delimited JSON, one event per `data:`:
 {"type":"log","attach":"l2","node":"dgx1.lan","recipe":"qwen38-27b-nvfp4-unsloth","rank":0,"line":"…"}
 ```
 
-- State events: every supervisor tick (30 s) and immediately after any job
-  state change.
-- Reconnect: browser `EventSource` auto-reconnects; on open the client
-  re-fetches `GET /api/state` (SSE is fire-and-forget, not a log).
+- State events: every supervisor tick (30 s), immediately after any job
+  state change, and **once on connect** — a fresh client gets the cached
+  snapshot first (`AppState.last`), so it paints and shows `live` at once
+  instead of waiting up to a full poll interval.
+- Reconnect: browser `EventSource` auto-reconnects; the connection itself
+  flips the indicator to `live` (`onopen`), replay keeps the data fresh.
 - Log lines are split server-side on `\n`; a chunk without a trailing
   newline is buffered (vLLM banner lines are long but single-line).
 
@@ -140,7 +142,11 @@ UI outgrows ~2–3k lines.)
 - **Window kinds** (parameterized views):
   - `state` — the LED grid (default, opens at start).
   - `logs {node, recipe, rank}` — live stream + history + pause + raw-copy.
-  - `jobs` — in-flight jobs with progress lines.
+  - `jobs` — in-flight jobs with progress lines. Running containers are
+    grouped **one row per launcher job** (docker `sm121.cluster` +
+    `sm121.recipe` label) with the node list — a container is reported by
+    every cluster sharing its node, so a per-node×cluster listing made the
+    whole board look busy.
   - `raw {kind: config|recipe:name|effective}` — verbatim content, monospace,
     copy button, JSON pretty for `effective`.
 - **LED grid:** rows = clusters → nodes; each node row: power LED,

src/web/mod.rs

@@ -78,26 +78,44 @@ pub(crate) async fn push_state(app: &AppState) {
     *app.last.lock().unwrap() = Some(snap);
 }
 
-/// `GET /events` — forward every bus event as an SSE message.
+/// `GET /events` — forward every bus event as an SSE message. A fresh
+/// client gets the cached snapshot first, so it paints (and flips its
+/// live indicator) at once instead of waiting for the next poll tick.
 async fn events_stream(
     app: State<AppState>,
 ) -> Sse<impl futures::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
+    let head = app
+        .last
+        .lock()
+        .unwrap()
+        .clone()
+        .map(|snapshot| Event::State {
+            ts: snapshot.ts,
+            snapshot,
+        });
     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.
+    Sse::new(futures::stream::unfold(
+        (rx, head),
+        |(mut rx, mut head)| async move {
+            if let Some(ev) = head.take() {
+                return Some((sse_event(&ev), (rx, None)));
+            }
+            loop {
+                match rx.recv().await {
+                    Ok(ev) => return Some((sse_event(&ev), (rx, None))),
+                    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,
                 }
-                Err(broadcast::error::RecvError::Closed) => return None,
             }
-        }
-    }))
+        },
+    ))
+}
+
+fn sse_event(ev: &Event) -> Result<SseEvent, std::convert::Infallible> {
+    Ok(SseEvent::default().data(serde_json::to_string(ev).unwrap_or_default()))
 }
 
 /// Bearer-token gate for data endpoints (`/` stays public — it is static

src/web/tests.rs

@@ -70,7 +70,10 @@ pub(crate) async fn boot() -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandl
 }
 
 async fn boot_with(config: Config) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
-    let app = app_state_with(config, None);
+    boot_app(app_state_with(config, None)).await
+}
+
+async fn boot_app(app: AppState) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
     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 {
@@ -124,6 +127,34 @@ async fn token_gate_enforces_when_set() {
     );
 }
 
+#[tokio::test]
+async fn sse_replays_cached_snapshot_on_connect() {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    // Nothing is broadcast after this point: the frame must come from the
+    // cached snapshot, otherwise a fresh client waits a whole poll tick.
+    let app = app_state(None);
+    *app.last.lock().unwrap() = Some(crate::web::events::Snapshot {
+        ts: 1,
+        clusters: Vec::new(),
+        jobs: Vec::new(),
+        topo: Default::default(),
+        attaches: Vec::new(),
+    });
+    let (addr, handle) = boot_app(app).await;
+    let mut sock = tokio::net::TcpStream::connect(addr).await.unwrap();
+    sock.write_all(b"GET /events HTTP/1.1\r\nHost: x\r\n\r\n")
+        .await
+        .unwrap();
+    let mut buf = vec![0u8; 4096];
+    let n = sock.read(&mut buf).await.unwrap();
+    let text = String::from_utf8_lossy(&buf[..n]);
+    assert!(
+        text.contains("\"type\":\"state\""),
+        "no replayed frame: {text}"
+    );
+    handle.abort();
+}
+
 #[tokio::test]
 async fn bus_delivers_state_events() {
     let (tx, mut rx) = tokio::sync::broadcast::channel::<Event>(8);
@@ -242,6 +273,54 @@ console.log('ok');",
     let _ = std::fs::remove_file(&js2);
 }
 
+/// runningFromLabels (JOBS-LT block): one row per launcher job. A container
+/// is reported by every cluster sharing its node — the list must not turn
+/// one serving job into a board-wide "everything is busy".
+#[test]
+fn ui_js_running_jobs_grouping() {
+    let html = include_str!("ui.html");
+    if std::process::Command::new("node")
+        .arg("--version")
+        .output()
+        .is_err()
+    {
+        eprintln!("SKIP ui_js_running_jobs_grouping: node not available");
+        return;
+    }
+    let script = extract_script(html);
+    let s = script.find("/*JOBS-LT:START*/").unwrap() + "/*JOBS-LT:START*/".len();
+    let e = script.find("/*JOBS-LT:END*/").unwrap();
+    let body = r#"
+const c = (cluster, node) => ({ name: cluster, up: true, nodes: [{ name: node, containers: [
+  { name: "sm121-a-" + node, status: "running", cluster: "ds4", recipe: "ds4-flash" }] }] });
+const state = { snap: { clusters: [
+  c("4nodes-switch", "dgx3"), c("ds4", "dgx4"), c("1node-3", "dgx3"), c("ds4", "dgx3"),
+  { name: "down", up: false, nodes: [{ name: "dgx9", containers: [
+    { name: "x", status: "running", cluster: "z", recipe: "zz" }] }] },
+] } };
+const run = runningFromLabels();
+if (run.length !== 1) throw new Error("one row per job, got " + JSON.stringify(run));
+if (run[0].nodes.join(",") !== "dgx3,dgx4") throw new Error("nodes: " + run[0].nodes);
+if (run[0].statuses.join(",") !== "running") throw new Error("statuses: " + run[0].statuses);
+if (run[0].cluster !== "ds4" || run[0].recipe !== "ds4-flash") throw new Error("job id: " + JSON.stringify(run[0]));
+console.log("ok");
+"#;
+    let probe = format!("{}\n{}", &script[s..e], body);
+    let pid = std::process::id();
+    let js = std::env::temp_dir().join(format!("sm121-ui-jobs-{pid}.js"));
+    std::fs::write(&js, probe).unwrap();
+    let out = std::process::Command::new("node")
+        .arg(&js)
+        .output()
+        .unwrap();
+    assert!(
+        out.status.success(),
+        "runningFromLabels probe failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let _ = std::fs::remove_file(&js);
+}
+
 /// serveBody (FLAGS-LT block): the serve flags → API body mapping is the
 /// trickiest pure logic in the UI — probe it in node like sameSnapshot.
 #[test]

src/web/ui.html

@@ -552,6 +552,9 @@ function openLogsPicker() {
 function sse(cls, txt) { $("#sse").className = "ssec " + cls; $("#sse-t").textContent = txt; }
 function connectSSE() {
   const es = new EventSource("/events");
+  // connection is up and a snapshot is already on its way — don't sit on
+  // "connecting…" for a whole poll interval
+  es.onopen = () => sse("ok", "live");
   es.onmessage = (e) => {
     sse("ok", "live");
     let ev;
@@ -1082,17 +1085,28 @@ function renderTopo(snap) {
 // --- jobs ---------------------------------------------------------------------
 // "Running" entries come from the docker labels (sm121.recipe / sm121.cluster)
 // on live containers — jobs don't need to have started from this dashboard.
+// One row per launcher job, NOT per node×cluster: a container is visible on
+// every cluster sharing its node, so the board used to look entirely busy.
+/*JOBS-LT:START*/
 function runningFromLabels() {
-  const out = [];
-  if (!state.snap) return out;
+  const m = new Map();
+  if (!state.snap) return [];
   for (const c of state.snap.clusters) {
     if (!c.up) continue; // down clusters report no containers
     for (const n of c.nodes) {
-      for (const x of n.containers) out.push({ cluster: c.name, node: n.name, name: x.name, recipe: x.recipe, status: x.status });
+      for (const x of n.containers) {
+        const key = x.cluster + "|" + x.recipe;
+        let r = m.get(key);
+        if (!r) m.set(key, (r = { cluster: x.cluster, recipe: x.recipe, nodes: [], statuses: new Set() }));
+        if (!r.nodes.includes(n.name)) r.nodes.push(n.name);
+        r.statuses.add(x.status);
+      }
     }
   }
-  return out;
+  const nat = (a, b) => a.localeCompare(b, undefined, { numeric: true });
+  return [...m.values()].map((r) => ({ ...r, nodes: r.nodes.sort(nat), statuses: [...r.statuses].sort(nat) }));
 }
+/*JOBS-LT:END*/
 function renderJobs() {
   const box = $("#jobs");
   box.innerHTML = "";
@@ -1112,9 +1126,10 @@ function renderJobs() {
     h.className = "jobs-empty"; h.textContent = "running (docker labels)";
     box.appendChild(h);
     for (const r of run) {
-      row("c:" + r.cluster + ":" + r.recipe + ":" + r.name, "jjob",
+      const many = r.nodes.length > 1;
+      row("c:" + r.cluster + ":" + r.recipe, "jjob",
         `<span class="jdot"></span><span class="jkind">serve</span>
-        <span class="jmsg">${esc(r.cluster)}:${esc(r.recipe)} — ${esc(r.name)} on ${esc(r.node)} · ${esc(r.status)}</span>`);
+        <span class="jmsg">${esc(r.cluster)}:${esc(r.recipe || "?")} — ${r.nodes.length} node${many ? "s" : ""} (${esc(r.nodes.join(", "))}) · ${esc(r.statuses.join(", "))}</span>`);
     }
   }
   const js = [...state.jobs.values()].sort((a, b) => b.started - a.started);