recipes panel: running state follows the run ledger (stale rows after stop)

Reported 2026-10-21: everything was stopped, the topology showed the nodes free,
but the recipes panel still listed three lanes as active. Two causes:

- the refresher only re-collected on report-file events (a `reports/` path or
  `recipe.toml`); `serve`/`stop` rewrite the recipe's `.lock`, which was not
  "relevant", so the panel kept the rows from its last scan. `.lock` now is.
- the flags are now recomputed every pass instead of being set once:
  `mark_stale`/`mark_fleet_up` clear their flag first, so a lane that comes back
  (or whose container is gone) stops looking stale/up. And the keepalive tick
  compares the rows' "up" set with the ledger (`ledger_open`/`ledger_differs`)
  and re-collects on any disagreement, so a stop that raced the watcher still
  lands within 30 s. The tick also re-runs the fleet cross-check whenever a new
  snapshot arrived — without a ledger (the add-on) `fleet_up` is the only
  running signal and it used to freeze after the first pass.

Moved `cross_check` + the ledger helpers into report/stale.rs (500-line gate),
tests in report/serve_tests.rs; new unit tests cover flag clearing and the
ledger diff.

Verified live: injecting an open run makes the row appear as orange "stale"
within the debounce, `sm121 runlock close` clears it on the next tick.

src/report/serve.rs

@@ -61,7 +61,8 @@ fn is_relevant(ev: &notify::Event) -> bool {
     ev.paths.iter().any(|p| {
         p.ancestors()
             .any(|a| a.file_name().is_some_and(|n| n == "reports"))
-            || p.file_name().is_some_and(|n| n == "recipe.toml")
+            || p.file_name()
+                .is_some_and(|n| n == "recipe.toml" || n == ".lock")
     })
 }
 
@@ -98,6 +99,8 @@ pub(crate) async fn refresh_loop(
     let mut last = String::new();
     let mut rows: Vec<crate::report::ReportRow> = Vec::new();
     let mut fleet: Option<crate::web::events::Snapshot> = None;
+    // ts of the fleet snapshot the rows were last reconciled against
+    let mut hull_ts: Option<u64> = None;
     let cluster_nodes: HashMap<String, Vec<String>> = cfg
         .clusters
         .iter()
@@ -120,7 +123,7 @@ pub(crate) async fn refresh_loop(
                 tokio::time::sleep(DEBOUNCE).await;
                 while ev_rx.try_recv().is_ok() {}
                 rows = collect(&cfg, archived);
-                cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
+                crate::report::stale::cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
                 let json = serde_json::to_string(&rows).unwrap_or_default();
                 if json != last {
                     last = json;
@@ -142,15 +145,23 @@ pub(crate) async fn refresh_loop(
                     fleet = Some(snapshot);
                 }
             }
-            // Keepalive safety net (missed watcher events): full collect only
-            // while uncollected (startup); afterwards only the cheap
-            // cross-check — and only when a run is open.
+            // Keepalive safety net (missed watcher events): a full collect when
+            // the RUN LEDGER disagrees with the rows — `serve`/`stop` rewrite
+            // `.lock`, and a missed event used to leave the panel showing lanes
+            // as up long after they were stopped (2026-10-21). Otherwise the
+            // cheap cross-check, re-run whenever a run is open OR a fresh fleet
+            // snapshot arrived (the add-on has no ledger: `fleet_up` is its
+            // only running signal, so it must be recomputed from the probe).
             _ = tokio::time::sleep(keepalive) => {
-                if rows.is_empty() {
+                let ledger = crate::report::stale::ledger_open(&cfg);
+                let fresh_fleet = fleet.as_ref().is_some_and(|f| Some(f.ts) != hull_ts);
+                if rows.is_empty() || crate::report::stale::ledger_differs(&rows, &ledger) {
                     rows = collect(&cfg, archived);
-                    cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
-                } else if has_open_run(&rows) {
-                    cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
+                    crate::report::stale::cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
+                    hull_ts = fleet.as_ref().map(|f| f.ts);
+                } else if has_open_run(&rows) || fresh_fleet {
+                    crate::report::stale::cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
+                    hull_ts = fleet.as_ref().map(|f| f.ts);
                 } else {
                     continue; // nothing to do — zero work, zero allocation
                 }
@@ -164,50 +175,6 @@ pub(crate) async fn refresh_loop(
     }
 }
 
-/// Reconcile the open-ledger rows against the live fleet: prefer the main
-/// dashboard's last fleet snapshot (no extra SSH); probe directly only when
-/// no snapshot has arrived yet (CLI `report --serve`, first 30 s of web).
-async fn cross_check(
-    cfg: &Config,
-    rows: &mut [crate::report::ReportRow],
-    cluster_nodes: &HashMap<String, Vec<String>>,
-    fleet: &mut Option<crate::web::events::Snapshot>,
-) {
-    let (reachable, live) = match fleet {
-        Some(snap) => {
-            let mut reachable: HashMap<String, bool> = HashMap::new();
-            let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
-            for c in &snap.clusters {
-                for n in &c.nodes {
-                    reachable.insert(n.name.clone(), n.reachable);
-                    for ct in &n.containers {
-                        live.insert(ct.recipe.clone());
-                    }
-                }
-            }
-            (reachable, live)
-        }
-        None => {
-            let f = crate::web::probe::probe(cfg).await;
-            let mut reachable: HashMap<String, bool> = HashMap::new();
-            let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
-            for (node, st) in &f.nodes {
-                reachable.insert(node.clone(), st.reachable);
-                for c in &st.containers {
-                    live.insert(c.recipe.clone());
-                }
-            }
-            (reachable, live)
-        }
-    };
-    crate::report::stale::mark_stale(rows, cluster_nodes, &reachable, &live);
-    // The inverse direction: a container in the fleet with NO local ledger
-    // run (the add-on machine) still renders as running.
-    crate::report::stale::mark_fleet_up(rows, &live);
-}
-
-/// `GET /events` — the latest rows JSON on connect, then one message per
-/// refresher update. The stream ends when the refresher (the sender) drops.
 async fn events_stream(
     axum::extract::State(s): axum::extract::State<AppState>,
 ) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
@@ -485,3 +452,7 @@ pub(crate) fn router(
             base: base.to_string(),
         })
 }
+
+#[cfg(test)]
+#[path = "serve_tests.rs"]
+mod tests;

src/report/serve_tests.rs

@@ -0,0 +1,49 @@
+//! Tests for `report::serve` (split at the 500-line cap).
+
+use crate::report::ReportRow;
+use crate::report::quality::Medals;
+use crate::report::stale::ledger_differs;
+
+fn row(name: &str, up: Option<i64>) -> ReportRow {
+    ReportRow {
+        name: name.to_string(),
+        title: name.to_string(),
+        model: "m".to_string(),
+        parent: "m".to_string(),
+        quant: "nvfp4".to_string(),
+        engine: "vllm".to_string(),
+        image: "img".to_string(),
+        nodes: None,
+        tp: 1,
+        deploy: "docker".to_string(),
+        port: 30000,
+        incomplete: false,
+        up,
+        up_cluster: None,
+        runs: 0,
+        fleet_up: false,
+        medals: Medals::default(),
+        stale: false,
+        reports: Vec::new(),
+    }
+}
+
+/// A `stop` that closed a ledger (or a `serve` that opened one) must force a
+/// full re-collect — that is how the panel learns a lane is gone.
+#[test]
+fn ledger_diff_detects_stopped_and_started_lanes() {
+    use std::collections::HashSet;
+    let rows = vec![row("a", Some(1000)), row("b", None), row("c", Some(2000))];
+    let led = HashSet::from(["a".to_string()]);
+    assert!(
+        ledger_differs(&rows, &led),
+        "a closed run must force a re-collect"
+    );
+    let led = HashSet::from(["a".to_string(), "c".to_string()]);
+    assert!(!ledger_differs(&rows, &led), "ledger matches the rows");
+    let led = HashSet::from(["a".to_string(), "b".to_string(), "c".to_string()]);
+    assert!(
+        ledger_differs(&rows, &led),
+        "a newly opened run must force a re-collect"
+    );
+}

src/report/stale.rs

@@ -12,6 +12,7 @@
 //! for a confirmed stale row is `sm121 runlock close <recipe>`.
 
 use super::ReportRow;
+use crate::config::Config;
 use std::collections::{HashMap, HashSet};
 
 /// Flag every open ledger run the live probe contradicts.
@@ -26,6 +27,11 @@ pub(crate) fn mark_stale(
     reachable: &HashMap<String, bool>,
     live_recipes: &HashSet<String>,
 ) {
+    // Recompute from scratch: a lane that came back up (or whose run was
+    // closed) must lose the orange flag, not keep it for the process lifetime.
+    for row in rows.iter_mut() {
+        row.stale = false;
+    }
     for row in rows {
         if row.up.is_none() {
             continue;
@@ -58,10 +64,73 @@ pub(crate) fn mark_stale(
 /// container present).
 pub(crate) fn mark_fleet_up(rows: &mut [ReportRow], live_recipes: &HashSet<String>) {
     for row in rows {
+        // Recomputed every pass, so a stopped container clears it again.
         row.fleet_up = row.up.is_none() && !row.stale && live_recipes.contains(&row.name);
     }
 }
 
+/// Recipes with an open run in the ledger (`.lock` files in the recipe dirs).
+pub(crate) fn ledger_open(cfg: &Config) -> HashSet<String> {
+    crate::runlock::open_runs(cfg)
+        .into_iter()
+        .map(|(name, _, _)| name)
+        .collect()
+}
+
+/// Do the rows' "up" set and the ledger's open set disagree? A `stop` that
+/// closed the ledger (or a `serve` that opened one) means the rows are stale
+/// and need a full re-collect, watcher events notwithstanding.
+pub(crate) fn ledger_differs(rows: &[ReportRow], ledger: &HashSet<String>) -> bool {
+    let shown: HashSet<&String> = rows
+        .iter()
+        .filter(|r| r.up.is_some())
+        .map(|r| &r.name)
+        .collect();
+    shown.len() != ledger.len() || !shown.iter().all(|n| ledger.contains(*n))
+}
+
+/// Reconcile the open-ledger rows against the live fleet: prefer the main
+/// dashboard's last fleet snapshot (no extra SSH); probe directly only when
+/// no snapshot has arrived yet (CLI `report --serve`, first 30 s of web).
+pub(crate) async fn cross_check(
+    cfg: &Config,
+    rows: &mut [ReportRow],
+    cluster_nodes: &HashMap<String, Vec<String>>,
+    fleet: &mut Option<crate::web::events::Snapshot>,
+) {
+    let (reachable, live) = match fleet {
+        Some(snap) => {
+            let mut reachable: HashMap<String, bool> = HashMap::new();
+            let mut live: HashSet<String> = HashSet::new();
+            for c in &snap.clusters {
+                for n in &c.nodes {
+                    reachable.insert(n.name.clone(), n.reachable);
+                    for ct in &n.containers {
+                        live.insert(ct.recipe.clone());
+                    }
+                }
+            }
+            (reachable, live)
+        }
+        None => {
+            let f = crate::web::probe::probe(cfg).await;
+            let mut reachable: HashMap<String, bool> = HashMap::new();
+            let mut live: HashSet<String> = HashSet::new();
+            for (node, st) in &f.nodes {
+                reachable.insert(node.clone(), st.reachable);
+                for c in &st.containers {
+                    live.insert(c.recipe.clone());
+                }
+            }
+            (reachable, live)
+        }
+    };
+    mark_stale(rows, cluster_nodes, &reachable, &live);
+    // The inverse direction: a container in the fleet with NO local ledger
+    // run (the add-on machine) still renders as running.
+    mark_fleet_up(rows, &live);
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -148,4 +217,28 @@ mod tests {
         mark_fleet_up(&mut rows, &HashSet::from(["r-a".to_string()]));
         assert!(!rows[0].fleet_up);
     }
+
+    /// Flags are RECOMPUTED, not accumulated: the panel re-runs the check every
+    /// 30 s, so a lane that is stopped must stop looking stale/up within that
+    /// window (reported 2026-10-21: three lanes stayed "active" after `stop`).
+    #[test]
+    fn flags_are_recomputed_not_accumulated() {
+        let mut rows = vec![row("r-a", Some(1000), Some("c1"))];
+        let cn = HashMap::from([("c1".to_string(), vec!["n1".to_string()])]);
+        let reach = HashMap::from([("n1".to_string(), true)]);
+        mark_stale(&mut rows, &cn, &reach, &HashSet::new());
+        assert!(rows[0].stale);
+        let live = HashSet::from(["r-a".to_string()]);
+        mark_stale(&mut rows, &cn, &reach, &live);
+        assert!(!rows[0].stale, "stale clears once the lane is live again");
+
+        let mut rows = vec![row("r-b", None, None)];
+        mark_fleet_up(&mut rows, &HashSet::from(["r-b".to_string()]));
+        assert!(rows[0].fleet_up);
+        mark_fleet_up(&mut rows, &HashSet::new());
+        assert!(
+            !rows[0].fleet_up,
+            "fleet_up clears when the container is gone"
+        );
+    }
 }