refactor(#2815): derive the transient snapshot, don't mirror it

mara on !2910: "why is set_transient still a thing if it completely
derives from nodes?"

It was still a thing because the scheduler mirrored the derived set into
a stored map that every consumer read — derived state computed once and
then cached, with the reconciliation loop existing only to keep the cache
honest. `transient_snapshot()` now derives: `running_transients()` off the
live graph, with the handful of entries that have no node behind them
(destroy, migration) overlaid on top. There is no cached copy left to go
stale or disagree with what is running.

`set_transient` / `clear_transient` split by what they actually do:
`set_manual_transient` / `clear_manual_transient` own the stored map for
the no-node callers, and `emit_transient_set` / `emit_transient_cleared`
publish the edges both paths need.

Two things had to survive, and both are edges rather than state:

- The dashboard's `TransientSet` / `TransientCleared` events. The
  scheduler carries the previous derived value and emits the diff.
- The crash watcher's grace window. `recent_transient_within` answers
  "was a transient cleared just now?", which is what stops a deliberate
  stop from reading as a crash on the next 10s poll — a derived read of
  current state cannot answer it, so the clear still stamps. The
  scheduler keeps `deliberate_stop` alongside the label precisely so it
  is available at clear time: the node it came from is, by definition, no
  longer running to be asked.

`TransientState::since` becomes wall-clock and, for derived entries, is
the node's own `started_at` — the true start of the operation rather than
the moment a watcher first noticed it, which is what the old
guard-creation timestamp actually measured.

`running_transients` returns a named `RunningTransient` rather than a
4-tuple; two of its fields are strings and one is a bool whose meaning is
not guessable at a call site.

Note for anyone reaching for a timestamp here: chrono is vendored with
`default-features = false`, so there is no `Utc::now()`. The workspace
convention is `wire_time::now_unix()` / `from_secs()`.

Checked with clippy (`--all-targets -D warnings`), `cargo test -p
hive-c0re -p hive-jobq` (322 + 41 passed) and `nix fmt`.

hive-c0re/src/coordinator.rs

@@ -8,6 +8,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
 use anyhow::{Context, Result};
+use chrono::{DateTime, Utc};
 use tokio::sync::{broadcast, watch};
 
 use crate::approvals::Approvals;
@@ -118,7 +119,9 @@ pub struct Coordinator {
     /// Agents whose lifecycle action (currently just spawn) is in flight.
     /// Read by the dashboard to render a spinner; cleared when the action
     /// resolves (success or failure).
-    transient: Mutex<HashMap<String, TransientState>>,
+    /// Transients for work with NO queue node behind it (destroy, migration).
+    /// Queue-driven pills are derived, not stored — see `transient_snapshot`.
+    manual_transient: Mutex<HashMap<String, TransientState>>,
     /// Tombstone for transients that have JUST been cleared. The
     /// crash watcher polls every 10s and would race the
     /// drop-clears-immediately path of `TransientGuard`: an operator
@@ -381,7 +384,10 @@ pub struct TransientState {
     /// Set by whoever creates the transient, which is the only place that
     /// actually knows — it is not recoverable from `label`.
     pub deliberate_stop: bool,
-    pub since: std::time::Instant,
+    /// When the operation started. Wall-clock rather than `Instant` because a
+    /// derived entry takes it from the node's own `started_at` — the true start
+    /// of the work, not the moment a watcher first noticed it.
+    pub since: DateTime<Utc>,
 }
 
 /// RAII handle returned by `Coordinator::transient_guard`. Cleared on
@@ -398,7 +404,7 @@ pub struct TransientGuard {
 
 impl Drop for TransientGuard {
     fn drop(&mut self) {
-        self.coord.clear_transient(&self.name);
+        self.coord.clear_manual_transient(&self.name);
     }
 }
 
@@ -537,7 +543,7 @@ impl Coordinator {
             agent_io_weight,
             model_prices,
             agents: Mutex::new(HashMap::new()),
-            transient: Mutex::new(HashMap::new()),
+            manual_transient: Mutex::new(HashMap::new()),
             recent_transient: Mutex::new(HashMap::new()),
             recent_crashes: Mutex::new(HashMap::new()),
             graceful_stop_pending: Mutex::new(HashSet::new()),
@@ -1077,25 +1083,31 @@ impl Coordinator {
 
     /// Mark an agent as in-progress (only one state per agent for now).
     ///
-    /// Two callers, for two different reasons:
+    /// Record a transient for work that has **no queue node behind it** —
-    /// - **Work with a queue node behind it** — the job-queue scheduler, which
+    /// destroy and migration. Reached only through the RAII [`TransientGuard`],
-    ///   publishes the edges of a *derived* set
+    /// so the paired clear still runs if the surrounding future is cancelled
-    ///   ([`crate::job_queue::JobQueue::running_transients`]). It needs no guard:
+    /// (HTTP request aborted, shutdown mid-rebuild, panic); a bare set here
-    ///   nothing is owned, and a node that stops running stops appearing.
+    /// really would leak the transient and pin the dashboard on "rebuilding…".
-    /// - **Work with no node** (destroy, migration) — via the RAII
+    ///
-    ///   [`TransientGuard`], so the paired `clear_transient` still runs if the
+    /// Queue-driven work does **not** come through here. Its pills are derived
-    ///   surrounding future is cancelled (HTTP request aborted, shutdown
+    /// from the running graph ([`crate::job_queue::JobQueue::running_transients`])
-    ///   mid-rebuild, panic). There a bare set really would leak the transient
+    /// and merged in by [`Coordinator::transient_snapshot`] — nothing is stored,
-    ///   and pin the dashboard on "rebuilding…" forever.
+    /// so nothing can go stale or leak.
-    pub(crate) fn set_transient(&self, name: &str, label: String, deliberate_stop: bool) {
+    fn set_manual_transient(&self, name: &str, label: String, deliberate_stop: bool) {
-        self.transient.lock().unwrap().insert(
+        self.manual_transient.lock().unwrap().insert(
             name.to_owned(),
             TransientState {
                 label: label.clone(),
                 deliberate_stop,
-                since: std::time::Instant::now(),
+                since: hive_sh4re::wire_time::from_secs(hive_sh4re::wire_time::now_unix()),
             },
         );
+        self.emit_transient_set(name, label);
+    }
+
+    /// Emit the "a pill appeared" edge. Shared by the manual path and the
+    /// job-queue scheduler, which publishes the transitions of its derived set.
+    pub(crate) fn emit_transient_set(&self, name: &str, label: String) {
         // Live-update dashboards. `since_unix` is wall-clock so the
         // browser can tick "Ns spawning…" without polling. The
         // intra-process map keeps using `Instant` for monotonicity.
@@ -1112,30 +1124,37 @@ impl Coordinator {
         });
     }
 
-    /// Clear an agent's transient state. Reached either from
+    /// Drop a manual transient (see [`Coordinator::set_manual_transient`]).
-    /// [`TransientGuard`]'s `Drop` (the no-node callers) or from the scheduler
+    /// Reached only from [`TransientGuard`]'s `Drop`, which guarantees it runs.
-    /// when a derived pill stops being current — see
+    fn clear_manual_transient(&self, name: &str) {
-    /// [`Coordinator::set_transient`].
+        let removed = self.manual_transient.lock().unwrap().remove(name);
-    pub(crate) fn clear_transient(&self, name: &str) {
-        let removed = self.transient.lock().unwrap().remove(name);
         if let Some(state) = removed {
-            // Stamp the tombstone so the crash watcher can still see
+            self.emit_transient_cleared(name, state.deliberate_stop);
-            // "operator kicked this off recently" on its next 10s poll
-            // — without this, the clear-then-poll race produced a
-            // spurious ContainerCrash on every operator stop/restart.
-            // Old entries get reaped lazily on read so the map doesn't
-            // grow unbounded.
-            self.recent_transient.lock().unwrap().insert(
-                name.to_owned(),
-                (state.deliberate_stop, std::time::Instant::now()),
-            );
-            self.emit_dashboard_event(DashboardEvent::TransientCleared {
-                seq: self.next_seq(),
-                name: name.to_owned(),
-            });
         }
     }
 
+    /// Emit the "a pill went away" edge **and stamp the tombstone the crash
+    /// watcher reads**. Shared by the manual path and the job-queue scheduler.
+    ///
+    /// 🚨 The stamp is not bookkeeping. Without it the clear-then-poll race
+    /// produced a spurious `ContainerCrash` on **every** operator stop/restart:
+    /// the transient is gone by the time the 10s poll looks, so a deliberate
+    /// stop is indistinguishable from a crash. `recent_transient_within` is what
+    /// closes that window, which is why the clear has to be an *event* — a
+    /// derived read of current state cannot answer "was one here a moment ago?".
+    ///
+    /// Old entries are reaped lazily on read, so the map stays bounded.
+    pub(crate) fn emit_transient_cleared(&self, name: &str, deliberate_stop: bool) {
+        self.recent_transient.lock().unwrap().insert(
+            name.to_owned(),
+            (deliberate_stop, std::time::Instant::now()),
+        );
+        self.emit_dashboard_event(DashboardEvent::TransientCleared {
+            seq: self.next_seq(),
+            name: name.to_owned(),
+        });
+    }
+
     /// Mark `name` as having a graceful stop in progress. While set,
     /// `socket_server::handle_recv` returns `Response::GracefulStop` for
     /// this agent instead of polling the broker (the inbound fence).
@@ -1259,15 +1278,46 @@ impl Coordinator {
         label: impl Into<String>,
         deliberate_stop: bool,
     ) -> TransientGuard {
-        self.set_transient(name, label.into(), deliberate_stop);
+        self.set_manual_transient(name, label.into(), deliberate_stop);
         TransientGuard {
             coord: self.clone(),
             name: name.to_owned(),
         }
     }
 
+    /// Every live transient, keyed by agent.
+    ///
+    /// **Derived on read**, not stored: the queue-driven pills come straight
+    /// from the running graph, so there is no cached copy to go stale, leak, or
+    /// disagree with what is actually running. The only stored entries are the
+    /// handful with no node behind them (destroy, migration), overlaid on top —
+    /// they win, since an agent being destroyed is the more urgent truth than
+    /// whatever node was mid-flight when it started.
+    #[must_use]
     pub fn transient_snapshot(&self) -> HashMap<String, TransientState> {
-        self.transient.lock().unwrap().clone()
+        let mut out: HashMap<String, TransientState> = self
+            .job_queue
+            .running_transients()
+            .into_iter()
+            .map(|t| {
+                (
+                    t.agent,
+                    TransientState {
+                        label: t.label,
+                        deliberate_stop: t.takes_container_down,
+                        since: t.since,
+                    },
+                )
+            })
+            .collect();
+        out.extend(
+            self.manual_transient
+                .lock()
+                .unwrap()
+                .iter()
+                .map(|(k, v)| (k.clone(), v.clone())),
+        );
+        out
     }
 
     /// Drop a system message into the given agent's inbox. Wakes the

hive-c0re/src/dashboard/state_snapshot.rs

@@ -535,7 +535,13 @@ fn build_transient_views(
         .map(|(name, st)| TransientView {
             name: name.clone(),
             kind: st.label.clone(),
-            secs: st.since.elapsed().as_secs(),
+            // Clamped at 0: `since` is wall-clock now (the node's own
+            // `started_at`), so a backwards clock adjustment could otherwise
+            // render a negative age.
+            secs: (hive_sh4re::wire_time::from_secs(hive_sh4re::wire_time::now_unix()) - st.since)
+                .num_seconds()
+                .max(0)
+                .cast_unsigned(),
         })
         .collect()
 }

hive-c0re/src/job_queue/mod.rs

@@ -59,6 +59,27 @@ const MAX_HISTORY_DAGS: usize = 50;
 /// Cap on stored node error strings.
 const MAX_ERROR_LEN: usize = 2_000;
 
+/// One live transient pill, derived from a running node.
+///
+/// A named struct rather than a tuple because three of its four fields are
+/// easy to confuse at a call site: two are strings and two answer questions
+/// nobody should have to guess at ("is this the agent or the label?", "does
+/// this bool mean deliberate or running?").
+#[derive(Debug, Clone)]
+pub struct RunningTransient {
+    /// The agent whose lease the node declared.
+    pub agent: String,
+    /// The node's own wire tag, rendered as the pill.
+    pub label: String,
+    /// Whether this operation is expected to take the container down — the
+    /// crash watcher's input. See [`NodeKind::takes_container_down`].
+    pub takes_container_down: bool,
+    /// When the node started running, so the dashboard can tick elapsed
+    /// seconds. Taken from the node itself, which is the true start of the
+    /// operation rather than the moment a watcher noticed it.
+    pub since: DateTime<Utc>,
+}
+
 /// A node claimed for execution — everything the executor needs, snapshotted at
 /// claim time.
 #[derive(Debug, Clone)]
@@ -422,7 +443,7 @@ impl JobQueue {
     /// the resources-where-constructed work, not this function. An agent's lease
     /// is cap-1, so at most one entry per agent.
     #[must_use]
-    pub fn running_transients(&self) -> Vec<(String, String, bool)> {
+    pub fn running_transients(&self) -> Vec<RunningTransient> {
         let inner = self.lock();
         inner
             .sched
@@ -441,11 +462,17 @@ impl JobQueue {
                         } => Some(a),
                         _ => None,
                     })?;
-                Some((
+                Some(RunningTransient {
                     agent,
-                    n.payload.as_str().to_owned(),
+                    label: n.payload.as_str().to_owned(),
-                    n.payload.takes_container_down(),
+                    takes_container_down: n.payload.takes_container_down(),
-                ))
+                    // `started_at` is set when a node enters `Running`, and this
+                    // only sees `Running` nodes — the fallback is unreachable in
+                    // practice, and "just now" is the honest answer if it isn't.
+                    since: n
+                        .started_at
+                        .unwrap_or_else(|| hive_sh4re::wire_time::from_secs(now_unix())),
+                })
             })
             .collect()
     }

hive-c0re/src/job_queue/scheduler.rs

@@ -59,7 +59,7 @@ pub async fn run_worker(coord: Arc<Coordinator>) {
     // it exists to spot transitions, since the dashboard wants edges
     // (`TransientSet` / `TransientCleared`) and the crash watcher wants the
     // moment of the clear. Nothing owns a pill; nothing can leak one.
-    let mut transients: HashMap<String, String> = HashMap::new();
+    let mut transients: HashMap<String, (String, bool)> = HashMap::new();
     loop {
         // Checked every iteration, not just in the `select!` below — a
         // continuous stream of ready claims never reaches the `select!`, so
@@ -174,25 +174,29 @@ fn handle_completion(coord: &Arc<Coordinator>, done: NodeDone) {
 /// starting* is a real crash that must keep reporting as one.
 ///
 /// [`NodeKind::takes_container_down`]: super::NodeKind::takes_container_down
-fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut HashMap<String, String>) {
+fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut HashMap<String, (String, bool)>) {
     let running = coord.job_queue.running_transients();
 
     // Cleared: in `prev`, gone (or relabelled) now. Emitted before the sets
     // below so a same-agent label change reads as clear-then-set rather than
-    // two overlapping pills.
+    // two overlapping pills. `deliberate_stop` is carried in `prev` precisely
-    prev.retain(|agent, label| {
+    // so it is still available *here* — the node it came from is, by
-        let still = running.iter().any(|(a, l, _)| a == agent && l == label);
+    // definition, no longer running to be asked.
+    prev.retain(|agent, (label, deliberate)| {
+        let still = running
+            .iter()
+            .any(|t| &t.agent == agent && &t.label == label);
         if !still {
-            coord.clear_transient(agent);
+            coord.emit_transient_cleared(agent, *deliberate);
         }
         still
     });
 
-    for (agent, label, takes_down) in running {
+    for t in running {
-        if prev.get(&agent) == Some(&label) {
+        if prev.get(&t.agent).map(|(l, _)| l) == Some(&t.label) {
             continue;
         }
-        coord.set_transient(&agent, label.clone(), takes_down);
+        coord.emit_transient_set(&t.agent, t.label.clone());
-        prev.insert(agent, label);
+        prev.insert(t.agent, (t.label, t.takes_container_down));
     }
 }