web: SSE streams can no longer die silently or stay dead — 15s keepalive comments on both endpoints, live-chip watchdog rebuilds a stale stream / a gave-up EventSource, bounce on visibility/online; supervisor isolates ticks and survives poisoned locks

1

@@ -0,0 +1,27 @@
+[general]
+import = [
+  "~/.alacritty/alacritty-theme/themes/alabaster.toml"
+]
+[font]
+#normal.family = 'Hack Nerd Font'
+normal.family = 'DMMono Nerd Font'
+#normal.family = 'DM Mono Regular Nerd Font'
+#normal.family = 'BlexMono Nerd Font Light'
+size = 14
+offset.x = 0  # For letter spacing (default: 0)
+offset.y = 0  # For line spacing (default: 0)
+
+[window]
+option_as_alt = "Both"
+dynamic_title = true
+
+[keyboard]
+bindings = [
+  { key = "C", mods = "Command", action = "Copy" },
+  { key = "V", mods = "Command", action = "Paste" },
+  { key = "с", mods = "Command", action = "Copy" },   # RU: C → «с»
+  { key = "м", mods = "Command", action = "Paste" },  # RU: V → «м»
+]
+
+[env]
+TERM = "xterm-256color"

specs/tech-architecture/web-dashboard.md

@@ -100,6 +100,18 @@ Newline-delimited JSON, one event per `data:`:
   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.
+- **Streams must not be able to die silently or stay dead.** Both SSE
+  endpoints send keepalive comments every 15 s (axum `Sse::keep_alive`) —
+  with the fleet poll that means bytes flow at least every 15 s, so NAT/VPN
+  idle timers and a frozen tab are noticed in seconds. The client
+  (`connectSSE`) treats the chip as a claim to verify: a stream silent past
+  3 polls is rebuilt (`bounce` — close + new `EventSource`; the replayed
+  snapshot repaints at once), an EventSource that gave up (`readyState
+  CLOSED` never retries by itself) schedules a rebuild, and
+  `visibilitychange`/`online` re-check immediately. Same CLOSED-rebuild on
+  the report page. The supervisor isolates each tick in its own task and
+  takes locks with `PoisonError::into_inner` — a panicking probe or a
+  poisoned mutex must not kill the live feed for good.
 - 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

src/report/page.html

@@ -272,23 +272,32 @@ notifBox.addEventListener("click", e => {
   render();
 });
 
-const es = new EventSource("/events");
-es.addEventListener("open", () => { liveEl.textContent = "● live"; liveEl.classList.remove("dim"); });
-es.addEventListener("error", () => { liveEl.textContent = "● off"; liveEl.classList.add("dim"); });
-es.onmessage = ev => {
-  let rows; try { rows = JSON.parse(ev.data); } catch { return; }
-  if (JSON.stringify(rows) === JSON.stringify(DATA)) return;
-  const upd = notifDiff(DATA, rows);
-  DATA = rows; syncTp(); render();
-  liveEl.title = "updated " + new Date().toLocaleTimeString();
-  if (upd.length) {
-    NOTIFS.unshift(...upd);
-    if (NOTIFS.length > 100) NOTIFS.length = 100;
-    UNREAD += upd.length;
-    renderBell(); renderNotifs();
-    bellEl.classList.remove("ring"); void bellEl.offsetWidth; bellEl.classList.add("ring");
-  }
-};
+let es = null;
+function connect() {
+  if (es) es.close();
+  es = new EventSource("/events");
+  es.addEventListener("open", () => { liveEl.textContent = "● live"; liveEl.classList.remove("dim"); });
+  es.addEventListener("error", () => {
+    liveEl.textContent = "● off"; liveEl.classList.add("dim");
+    // readyState CLOSED = EventSource gave up for good — recreate it
+    if (es.readyState === 2) setTimeout(connect, 1000);
+  });
+  es.onmessage = ev => {
+    let rows; try { rows = JSON.parse(ev.data); } catch { return; }
+    if (JSON.stringify(rows) === JSON.stringify(DATA)) return;
+    const upd = notifDiff(DATA, rows);
+    DATA = rows; syncTp(); render();
+    liveEl.title = "updated " + new Date().toLocaleTimeString();
+    if (upd.length) {
+      NOTIFS.unshift(...upd);
+      if (NOTIFS.length > 100) NOTIFS.length = 100;
+      UNREAD += upd.length;
+      renderBell(); renderNotifs();
+      bellEl.classList.remove("ring"); void bellEl.offsetWidth; bellEl.classList.add("ring");
+    }
+  };
+}
+connect();
 renderBell(); renderNotifs();
 const P = new URLSearchParams(location.search);
 // Persistence: URL params (deep links) win; without them the last view is

src/report/serve.rs

@@ -149,6 +149,10 @@ async fn events_stream(
         Some((Ok(SseEvent::default().data(json)), rx))
     });
     Sse::new(first.chain(rest))
+        // the rows only change when a bench report changes — a silent
+        // stream would be indistinguishable from a dead one for anything
+        // between the server and the browser
+        .keep_alive(axum::response::sse::KeepAlive::new().interval(Duration::from_secs(15)))
 }
 
 /// `GET /r/<recipe>/<file>` — full file text (report, log, json, or the

src/web/mod.rs

@@ -25,7 +25,7 @@ use axum::{
     middleware::{self, Next},
     response::{
         Html, Response,
-        sse::{Event as SseEvent, Sse},
+        sse::{Event as SseEvent, KeepAlive, Sse},
     },
     routing::{get, post},
 };
@@ -75,7 +75,7 @@ pub(crate) async fn push_state(app: &AppState) {
         ts: snap.ts,
         snapshot: snap.clone(),
     });
-    *app.last.lock().unwrap() = Some(snap);
+    *lock_last(app) = Some(snap);
 }
 
 /// `GET /events` — forward every bus event as an SSE message. A fresh
@@ -112,6 +112,10 @@ async fn events_stream(
             }
         },
     ))
+    // comment lines between events: NAT/VPN/proxy idle timers (and frozen
+    // tabs on reconnect) see traffic, so a dead stream is noticed in seconds,
+    // not minutes
+    .keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
 }
 
 fn sse_event(ev: &Event) -> Result<SseEvent, std::convert::Infallible> {
@@ -151,11 +155,24 @@ async fn token_guard(
 /// Supervisor: poll the fleet and broadcast a state snapshot every tick.
 async fn supervisor(app: AppState) {
     loop {
-        push_state(&app).await;
+        // isolate each tick: a probe bug or a poisoned lock must not kill
+        // the poll loop for good (a dead supervisor means a dead live feed)
+        let tick = app.clone();
+        if let Err(e) = tokio::spawn(async move { push_state(&tick).await }).await {
+            tracing::warn!("sm121 web: supervisor tick panicked: {e}");
+        }
         tokio::time::sleep(Duration::from_secs(POLL_SECS)).await;
     }
 }
 
+/// `AppState.last` guard that survives a poisoned mutex (a panic while
+/// holding the lock would otherwise take the whole live feed down).
+fn lock_last(app: &AppState) -> std::sync::MutexGuard<'_, Option<events::Snapshot>> {
+    app.last
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
 fn is_loopback(addr: SocketAddr) -> bool {
     addr.ip().is_loopback()
 }
@@ -244,3 +261,5 @@ pub async fn run(
 mod api_tests;
 #[cfg(test)]
 mod tests;
+#[cfg(test)]
+mod ui_js_tests;

src/web/tests.rs

@@ -238,195 +238,6 @@ async fn snapshot_serializes_with_type_tag() {
 
 // (P2/P3/P4 API round-trip tests moved to api_tests.rs — 500-line cap)
 
-// --- UI (ui.html) -------------------------------------------------------------
-// The embedded page is vanilla JS with no build step; these tests run the
-// real file through node: (1) syntax gate, (2) behavior probe of the pure
-// snapshot-compare logic (the anti-blink rule), extracted between the
-// SNAP-LT markers in ui.html. Skipped when node is unavailable.
-
-fn extract_script(html: &str) -> String {
-    let start = html.find("<script>").unwrap() + "<script>".len();
-    let end = html.rfind("</script>").unwrap();
-    html[start..end].to_string()
-}
-
-#[test]
-fn ui_js_syntax_and_snapshot_logic() {
-    let html = include_str!("ui.html");
-    if std::process::Command::new("node")
-        .arg("--version")
-        .output()
-        .is_err()
-    {
-        eprintln!("SKIP ui_js_syntax_and_snapshot_logic: node not available");
-        return;
-    }
-    let script = extract_script(html);
-    let pid = std::process::id();
-    let js = std::env::temp_dir().join(format!("sm121-ui-{pid}.js"));
-    std::fs::write(&js, &script).unwrap();
-    let out = std::process::Command::new("node")
-        .arg("--check")
-        .arg(&js)
-        .output()
-        .unwrap();
-    assert!(
-        out.status.success(),
-        "ui.html JS must parse: {}",
-        String::from_utf8_lossy(&out.stderr)
-    );
-
-    // sameSnapshot: ts must be ignored, real changes must not be equal.
-    let s = script.find("/*SNAP-LT:START*/").unwrap() + "/*SNAP-LT:START*/".len();
-    let e = script.find("/*SNAP-LT:END*/").unwrap();
-    let probe = format!(
-        "{fn}
-const a = {{ ts: 1, clusters: [{{ name: 'c', up: true, free: 1, nodes: [] }}], jobs: [] }};
-const b = {{ ts: 99, clusters: [{{ name: 'c', up: true, free: 1, nodes: [] }}], jobs: [] }};
-const d = {{ ts: 99, clusters: [{{ name: 'c', up: true, free: 0, nodes: [] }}], jobs: [] }};
-if (sameSnapshot(a, b) !== true) throw new Error('ts must be ignored');
-if (sameSnapshot(a, d) !== false) throw new Error('changed fleet must differ');
-if (sameSnapshot(a, a) !== true) throw new Error('identity');
-console.log('ok');",
-        fn = &script[s..e]
-    );
-    let js2 = std::env::temp_dir().join(format!("sm121-ui-probe-{pid}.js"));
-    std::fs::write(&js2, probe).unwrap();
-    let out = std::process::Command::new("node")
-        .arg(&js2)
-        .output()
-        .unwrap();
-    assert!(
-        out.status.success(),
-        "sameSnapshot probe failed: {}",
-        String::from_utf8_lossy(&out.stderr)
-    );
-    let _ = std::fs::remove_file(&js);
-    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]
-fn ui_js_serve_body_logic() {
-    let html = include_str!("ui.html");
-    if std::process::Command::new("node")
-        .arg("--version")
-        .output()
-        .is_err()
-    {
-        eprintln!("SKIP ui_js_serve_body_logic: node not available");
-        return;
-    }
-    let script = extract_script(html);
-    let s = script.find("/*FLAGS-LT:START*/").unwrap() + "/*FLAGS-LT:START*/".len();
-    let e = script.find("/*FLAGS-LT:END*/").unwrap();
-    let pid = std::process::id();
-    let probe = format!(
-        "const state = {{ flags: {{}} }};\n{fn}\n\n\
-         // all flags set: everything passes through\n\
-         let b = serveBody('c', 'r', true, {{ node: 'n9', solo: true, force: true, clear: true, keep: true, fit: '2', fitAll: false, mirror: '/m', mirrorAll: true }});\n\
-         if (b.node !== 'n9' || b.solo !== true || b.force !== true || b.clear !== true || b.keep !== true || b.dry_run !== true) throw new Error('flag passthrough: ' + JSON.stringify(b));\n\
-         if (b.fit !== '2') throw new Error('fit N must win over bare: ' + b.fit);\n\
-         if (b.mirror !== '/m') throw new Error('mirror path must win over bare: ' + b.mirror);\n\
-         // bare fit / bare mirror (empty strings = whole cluster / default path)\n\
-         b = serveBody('c', 'r', false, {{ fit: '', fitAll: true }});\n\
-         if (b.fit !== '') throw new Error('bare fit');\n\
-         b = serveBody('c', 'r', false, {{ mirror: '', mirrorAll: true }});\n\
-         if (b.mirror !== '') throw new Error('bare mirror');\n\
-         // no flags: only cluster/recipe/dry_run — server defaults win\n\
-         b = serveBody('c', 'r', true, {{}});\n\
-         for (const k of ['node', 'solo', 'force', 'clear', 'keep', 'fit', 'mirror']) {{\n\
-           if (k in b) throw new Error('absent flag must be omitted: ' + k);\n\
-         }}\n\
-         // flagsOf caches per cluster\n\
-         if (flagsOf('x') !== flagsOf('x')) throw new Error('flagsOf must cache');\n\
-         console.log('ok');",
-        fn = &script[s..e]
-    );
-    let js2 = std::env::temp_dir().join(format!("sm121-ui-flags-{pid}.js"));
-    std::fs::write(&js2, &probe).unwrap();
-    let out = std::process::Command::new("node")
-        .arg(&js2)
-        .output()
-        .unwrap();
-    assert!(
-        out.status.success(),
-        "serveBody probe failed: {}",
-        String::from_utf8_lossy(&out.stderr)
-    );
-    let _ = std::fs::remove_file(&js2);
-}
-
-/// Two offline clusters (both nodes on refused loopback ports) — the
-/// fleet the Playwright UI checks render: chip strip with two chips,
-/// down-cluster action bar, type-the-name power flow.
-fn playwright_config() -> Config {
-    let mut cfg = fake_config();
-    cfg.clusters.push(Cluster {
-        name: "fake2".into(),
-        master_ib_ip: None,
-        nodes: vec![Node {
-            name: "n1".into(),
-            ssh_ip: "127.0.0.1:2223".into(),
-            ib_ip: None,
-            rank: 0,
-            host_key: None,
-            entity: None,
-        }],
-        extra_entities: vec![],
-    });
-    cfg
-}
-
 /// Click-through UI tests: boots the real server on an ephemeral port and
 /// drives the embedded ui.html with Playwright (deno, npm-free).
 /// Skipped when deno or the chromium binary is unavailable.
@@ -463,3 +274,24 @@ async fn ui_playwright_checks() {
     );
     eprintln!("{stdout}");
 }
+
+/// Two offline clusters (both nodes on refused loopback ports) — the
+/// fleet the Playwright UI checks render: chip strip with two chips,
+/// down-cluster action bar, type-the-name power flow.
+fn playwright_config() -> Config {
+    let mut cfg = fake_config();
+    cfg.clusters.push(Cluster {
+        name: "fake2".into(),
+        master_ib_ip: None,
+        nodes: vec![Node {
+            name: "n1".into(),
+            ssh_ip: "127.0.0.1:2223".into(),
+            ib_ip: None,
+            rank: 0,
+            host_key: None,
+            entity: None,
+        }],
+        extra_entities: vec![],
+    });
+    cfg
+}

src/web/ui.html

@@ -715,21 +715,48 @@ function openLogsPicker() {
 
 // --- SSE -------------------------------------------------------------------
 function sse(cls, txt) { $("#sse").className = "ssec " + cls; $("#sse-t").textContent = txt; }
+/*SSE-LT:START*/
 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;
-    try { ev = JSON.parse(e.data); } catch { return; }
-    if (ev.type === "state") onState(ev.snapshot);
-    else if (ev.type === "job") onJob(ev);
-    else if (ev.type === "log") onLog(ev);
+  // the fleet polls every 30s, so a working stream is never silent for
+  // long (plus 15s server keepalive comments). If it IS silent, the chip
+  // is lying — rebuild the stream instead of trusting "open".
+  const POLL_MS = 30000, STALE_MS = 3 * POLL_MS + 5000;
+  let es = null, last = Date.now();
+  const bounce = (txt) => {
+    if (es) es.close();
+    sse("re", txt);
+    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 = () => { last = Date.now(); sse("ok", "live"); };
+    es.onmessage = (e) => {
+      last = Date.now();
+      sse("ok", "live");
+      let ev;
+      try { ev = JSON.parse(e.data); } catch { return; }
+      if (ev.type === "state") onState(ev.snapshot);
+      else if (ev.type === "job") onJob(ev);
+      else if (ev.type === "log") onLog(ev);
+    };
+    es.onerror = () => {
+      sse("re", "reconnecting…");
+      // readyState CLOSED = EventSource gave up for good — only a fresh
+      // object reconnects (the server replays the snapshot, so it heals
+      // without a manual page reload)
+      if (es.readyState === 2) setTimeout(() => bounce("reconnecting…"), 1000);
+    };
   };
-  es.onerror = () => { sse("re", "reconnecting…"); };
+  bounce("connecting…");
+  setInterval(() => {
+    if (document.visibilityState === "visible" && Date.now() - last > STALE_MS) bounce("resyncing…");
+  }, POLL_MS);
+  // back from a frozen/sleeping tab or a network hop: check immediately
+  document.addEventListener("visibilitychange", () => {
+    if (document.visibilityState === "visible" && Date.now() - last > 40000) bounce("resyncing…");
+  });
+  addEventListener("online", () => bounce("reconnecting…"));
 }
+/*SSE-LT:END*/
 
 // --- state render ------------------------------------------------------------
 function ledState(n) { return n.reachable ? (n.occupied ? "b" : "on") : "off"; }

src/web/ui_js_tests.rs

@@ -0,0 +1,243 @@
+//! Pure-logic probes of the embedded ui.html, run through node (the file
+//! is a no-build vanilla page): syntax gate + the extracted marker blocks.
+//! Skipped when node is unavailable. Split from tests.rs for the 500-line cap.
+
+// --- UI (ui.html) -------------------------------------------------------------
+// The embedded page is vanilla JS with no build step; these tests run the
+// real file through node: (1) syntax gate, (2) behavior probe of the pure
+// snapshot-compare logic (the anti-blink rule), extracted between the
+// SNAP-LT markers in ui.html. Skipped when node is unavailable.
+
+fn extract_script(html: &str) -> String {
+    let start = html.find("<script>").unwrap() + "<script>".len();
+    let end = html.rfind("</script>").unwrap();
+    html[start..end].to_string()
+}
+
+#[test]
+fn ui_js_syntax_and_snapshot_logic() {
+    let html = include_str!("ui.html");
+    if std::process::Command::new("node")
+        .arg("--version")
+        .output()
+        .is_err()
+    {
+        eprintln!("SKIP ui_js_syntax_and_snapshot_logic: node not available");
+        return;
+    }
+    let script = extract_script(html);
+    let pid = std::process::id();
+    let js = std::env::temp_dir().join(format!("sm121-ui-{pid}.js"));
+    std::fs::write(&js, &script).unwrap();
+    let out = std::process::Command::new("node")
+        .arg("--check")
+        .arg(&js)
+        .output()
+        .unwrap();
+    assert!(
+        out.status.success(),
+        "ui.html JS must parse: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+
+    // sameSnapshot: ts must be ignored, real changes must not be equal.
+    let s = script.find("/*SNAP-LT:START*/").unwrap() + "/*SNAP-LT:START*/".len();
+    let e = script.find("/*SNAP-LT:END*/").unwrap();
+    let probe = format!(
+        "{fn}
+const a = {{ ts: 1, clusters: [{{ name: 'c', up: true, free: 1, nodes: [] }}], jobs: [] }};
+const b = {{ ts: 99, clusters: [{{ name: 'c', up: true, free: 1, nodes: [] }}], jobs: [] }};
+const d = {{ ts: 99, clusters: [{{ name: 'c', up: true, free: 0, nodes: [] }}], jobs: [] }};
+if (sameSnapshot(a, b) !== true) throw new Error('ts must be ignored');
+if (sameSnapshot(a, d) !== false) throw new Error('changed fleet must differ');
+if (sameSnapshot(a, a) !== true) throw new Error('identity');
+console.log('ok');",
+        fn = &script[s..e]
+    );
+    let js2 = std::env::temp_dir().join(format!("sm121-ui-probe-{pid}.js"));
+    std::fs::write(&js2, probe).unwrap();
+    let out = std::process::Command::new("node")
+        .arg(&js2)
+        .output()
+        .unwrap();
+    assert!(
+        out.status.success(),
+        "sameSnapshot probe failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let _ = std::fs::remove_file(&js);
+    let _ = std::fs::remove_file(&js2);
+}
+
+/// connectSSE (SSE-LT block): the live chip must never stick in a broken
+/// state — a CLOSED EventSource (gave up retrying) schedules a rebuild, and
+/// a stream silent past three poll ticks is rebuilt by the watchdog; fresh
+/// data must not trigger rebuilds.
+#[test]
+fn ui_js_sse_recovery_logic() {
+    let html = include_str!("ui.html");
+    if std::process::Command::new("node")
+        .arg("--version")
+        .output()
+        .is_err()
+    {
+        eprintln!("SKIP ui_js_sse_recovery_logic: node not available");
+        return;
+    }
+    let script = extract_script(html);
+    let s = script.find("/*SSE-LT:START*/").unwrap() + "/*SSE-LT:START*/".len();
+    let e = script.find("/*SSE-LT:END*/").unwrap();
+    let body = r#"
+let created = 0, now = 1000000;
+class FakeES { constructor() { this.readyState = 0; created++; FakeES.last = this; } close() {} }
+const handlers = {}, timers = [], sseLog = [];
+global.document = { visibilityState: "visible", addEventListener: (t, f) => { handlers[t] = f; } };
+global.addEventListener = (t, f) => { handlers["w:" + t] = f; };
+global.EventSource = FakeES;
+global.setTimeout = (f, ms) => { timers.push({ f, ms }); return 0; };
+global.setInterval = (f) => { timers.push({ f, every: true }); return 0; };
+global.$ = () => ({ set className(v) {}, set textContent(v) { sseLog.push(v); } });
+global.sse = (cls, txt) => sseLog.push(txt);
+global.onState = global.onJob = global.onLog = () => {};
+global.Date = class extends Date { static now() { return now; } };
+const chip = () => sseLog[sseLog.length - 1];
+connectSSE();
+if (created !== 1 || chip() !== "connecting…") throw new Error("no initial connect");
+FakeES.last.readyState = 1; FakeES.last.onopen();
+if (chip() !== "live") throw new Error("open must set live");
+const tick = timers.find((t) => t.every);
+now += 100000; // silent past 3 poll ticks (95s)
+tick.f();
+if (created !== 2 || chip() !== "resyncing…") throw new Error("stale stream not rebuilt");
+FakeES.last.readyState = 2; FakeES.last.onerror();
+const scheduled = timers.filter((t) => !t.every);
+if (scheduled.length !== 1 || scheduled[0].ms !== 1000) throw new Error("CLOSED not scheduled");
+scheduled[0].f();
+if (created !== 3) throw new Error("CLOSED did not re-bounce");
+FakeES.last.readyState = 1; FakeES.last.onopen();
+now += 1000;
+tick.f();
+if (created !== 3 || chip() !== "live") throw new Error("fresh stream must not rebuild");
+console.log("ok");
+"#;
+    let probe = format!(
+        "{}
+{}",
+        &script[s..e],
+        body
+    );
+    let pid = std::process::id();
+    let js = std::env::temp_dir().join(format!("sm121-ui-sse-{pid}.js"));
+    std::fs::write(&js, probe).unwrap();
+    let out = std::process::Command::new("node")
+        .arg(&js)
+        .output()
+        .unwrap();
+    assert!(
+        out.status.success(),
+        "connectSSE probe failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let _ = std::fs::remove_file(&js);
+}
+
+/// 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]
+fn ui_js_serve_body_logic() {
+    let html = include_str!("ui.html");
+    if std::process::Command::new("node")
+        .arg("--version")
+        .output()
+        .is_err()
+    {
+        eprintln!("SKIP ui_js_serve_body_logic: node not available");
+        return;
+    }
+    let script = extract_script(html);
+    let s = script.find("/*FLAGS-LT:START*/").unwrap() + "/*FLAGS-LT:START*/".len();
+    let e = script.find("/*FLAGS-LT:END*/").unwrap();
+    let pid = std::process::id();
+    let probe = format!(
+        "const state = {{ flags: {{}} }};\n{fn}\n\n\
+         // all flags set: everything passes through\n\
+         let b = serveBody('c', 'r', true, {{ node: 'n9', solo: true, force: true, clear: true, keep: true, fit: '2', fitAll: false, mirror: '/m', mirrorAll: true }});\n\
+         if (b.node !== 'n9' || b.solo !== true || b.force !== true || b.clear !== true || b.keep !== true || b.dry_run !== true) throw new Error('flag passthrough: ' + JSON.stringify(b));\n\
+         if (b.fit !== '2') throw new Error('fit N must win over bare: ' + b.fit);\n\
+         if (b.mirror !== '/m') throw new Error('mirror path must win over bare: ' + b.mirror);\n\
+         // bare fit / bare mirror (empty strings = whole cluster / default path)\n\
+         b = serveBody('c', 'r', false, {{ fit: '', fitAll: true }});\n\
+         if (b.fit !== '') throw new Error('bare fit');\n\
+         b = serveBody('c', 'r', false, {{ mirror: '', mirrorAll: true }});\n\
+         if (b.mirror !== '') throw new Error('bare mirror');\n\
+         // no flags: only cluster/recipe/dry_run — server defaults win\n\
+         b = serveBody('c', 'r', true, {{}});\n\
+         for (const k of ['node', 'solo', 'force', 'clear', 'keep', 'fit', 'mirror']) {{\n\
+           if (k in b) throw new Error('absent flag must be omitted: ' + k);\n\
+         }}\n\
+         // flagsOf caches per cluster\n\
+         if (flagsOf('x') !== flagsOf('x')) throw new Error('flagsOf must cache');\n\
+         console.log('ok');",
+        fn = &script[s..e]
+    );
+    let js2 = std::env::temp_dir().join(format!("sm121-ui-flags-{pid}.js"));
+    std::fs::write(&js2, &probe).unwrap();
+    let out = std::process::Command::new("node")
+        .arg(&js2)
+        .output()
+        .unwrap();
+    assert!(
+        out.status.success(),
+        "serveBody probe failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let _ = std::fs::remove_file(&js2);
+}