jobq: key transient tombstones per pill, not per agent

Deletes `TransientState`, adds `transient_kind` to `TransientCleared`, and
fixes a crash misreport — three changes over the same functions.

`TransientState` was `RunningTransient` with `agent` dropped and
`takes_container_down` renamed; that rename was the only thing it did, and
its three consumers each read a disjoint subset. `transient_snapshot` now
returns `RunningTransient` directly.

`recent_transient` was keyed by agent alone and overwritten on each clear.
An agent can clear several pills in one grace window, so a `Prebuild`
(`takes_container_down = false`) landing after a `StopForUpdate` (`true`)
left the tombstone reading `false` and the crash watcher reported a
deliberate stop as a container crash. Keyed by `(agent, label)` now, with
`recent_transient_within` folding back per agent by OR — the same question
`crash_watch` asks of the active set.

`TransientCleared` gains the label for the same reason: a client holding
two open pills for one agent could not tell which one a clear referred to.

The out-of-band suppression guard has no node and so no label; it uses
`NO_NODE_LABEL`, angle-bracketed to stay out of the `NodeKind::as_str`
namespace.

hive-c0re/src/coordinator.rs

@@ -8,13 +8,13 @@ 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;
 use crate::broker::Broker;
 use crate::container_view::{self, ContainerView};
 use crate::dashboard_events::DashboardEvent;
+use crate::job_queue::RunningTransient;
 use crate::operator_questions::OperatorQuestions;
 use crate::socket_server::{self, AgentSocket};
 
@@ -137,7 +137,17 @@ pub struct Coordinator {
     /// agents whose tombstone is still inside the grace window. Crash
     /// watcher consults both this and the active map before declaring
     /// a stop deliberate.
-    recent_transient: Mutex<HashMap<String, (bool, std::time::Instant)>>,
+    ///
+    /// 🚨 **Keyed by `(agent, label)`, not by agent.** An agent can have several
+    /// pills clearing in the same window — the transient set tests status alone,
+    /// so a lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both
+    /// live and both clear. Keyed by agent, the last clear *overwrites* the
+    /// others: a `Prebuild` (`deliberate_stop = false`) landing after a
+    /// `StopForUpdate` (`true`) leaves the tombstone reading `false`, and the
+    /// crash watcher then reports an intentional stop as a **crash**. The
+    /// out-of-band suppression guard, which has no node behind it, uses
+    /// [`NO_NODE_LABEL`].
+    recent_transient: Mutex<HashMap<(String, String), (bool, std::time::Instant)>>,
     /// Timestamps of recent unexpected container crashes, keyed by agent.
     /// Fed by `crash_watch` each time it classifies a stop as a crash (so
     /// a crash-looping container — which `Restart=on-failure` flips back
@@ -359,38 +369,39 @@ pub struct AgentPaths {
     pub notes: PathBuf,
 }
 
-/// Per-agent in-progress state that the dashboard surfaces between approve
-/// click and container ready.
+/// Collapse per-pill tombstones to one answer per agent: **was any recently
+/// cleared pill for this agent a deliberate stop?**
 ///
-/// The two fields answer genuinely different questions and are set
-/// independently on purpose. There used to be a single `TransientKind` enum
-/// serving both, which meant a display concern and a safety decision shared one
-/// vocabulary and moved together.
-#[derive(Debug, Clone)]
-pub struct TransientState {
-    /// What the dashboard pill renders. For queue-driven work this is the
-    /// running node's own wire tag ([`crate::job_queue::NodeKind::as_str`]) —
-    /// the same vocabulary the DAG view ships, so a pill and a node name an
-    /// operation identically. Work with no node behind it (destroy, migration)
-    /// supplies its own.
-    ///
-    /// Display only. Nothing branches on it — match on a string and this
-    /// becomes a taxonomy again, silently.
-    pub label: String,
-    /// Whether the container going down is **expected**, i.e. this operation
-    /// takes it down on purpose. Read by the crash watcher to tell a
-    /// deliberate stop from a crash, so a wrong value here either raises a
-    /// false alarm or swallows a real one.
-    ///
-    /// Set by whoever creates the transient, which is the only place that
-    /// actually knows — it is not recoverable from `label`.
-    pub deliberate_stop: bool,
-    /// 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>,
+/// `OR`, not last-write-wins — that is the whole fix. An agent can clear several
+/// pills inside one grace window (the transient set tests status alone, so a
+/// lease-exempt `Prebuild` and a lease-holding `StopForUpdate` are both live),
+/// and taking the last one means an incidental `false` erases a real `true`,
+/// which the crash watcher then reads as a container **crash**.
+///
+/// Asks the same question of the cleared set that `crash_watch` asks of the
+/// active set with `.any(…)`, so the two agree by construction. A free function
+/// so it is testable without a `Coordinator` fixture — same reason
+/// `crash_watch::is_deliberate_stop` is one.
+fn fold_tombstones_by_agent<'a>(
+    entries: impl Iterator<Item = (&'a str, bool)>,
+) -> HashMap<String, bool> {
+    let mut out: HashMap<String, bool> = HashMap::new();
+    for (agent, deliberate) in entries {
+        *out.entry(agent.to_owned()).or_default() |= deliberate;
+    }
+    out
 }
 
+/// Tombstone label for work with **no queue node behind it** — the
+/// out-of-band operations (destroy, migration) that hold a
+/// [`CrashWatchGuard`] instead of appearing in the derived transient set.
+///
+/// [`Coordinator::recent_transient`] is keyed by `(agent, label)` so concurrent
+/// pills can't overwrite each other's `deliberate_stop`; a guard has no node and
+/// therefore no node label, so it needs one of its own. Angle-bracketed to keep
+/// it out of the `NodeKind::as_str` namespace — no node can ever render this.
+const NO_NODE_LABEL: &str = "<no-node>";
+
 /// RAII handle returned by [`Coordinator::suppress_crash_watch`]. While held,
 /// the crash watcher treats this container disappearing as **expected**.
 ///
@@ -425,11 +436,10 @@ impl Drop for CrashWatchSuppression {
         // Tombstone the release so the next poll — which may land in the
         // window between the container going away and this guard dropping —
         // still reads the stop as deliberate.
-        self.coord
-            .recent_transient
-            .lock()
-            .unwrap()
-            .insert(self.name.clone(), (true, std::time::Instant::now()));
+        self.coord.recent_transient.lock().unwrap().insert(
+            (self.name.clone(), NO_NODE_LABEL.to_owned()),
+            (true, std::time::Instant::now()),
+        );
     }
 }
 
@@ -1136,14 +1146,20 @@ impl Coordinator {
     /// 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) {
+    /// `label` identifies *which* pill cleared. An agent can hold several at
+    /// once, so both the tombstone key and the wire event need it — without it
+    /// a client that received two `TransientSet`s cannot tell which one this
+    /// clears, and the tombstone silently overwrites a concurrent pill's
+    /// `deliberate_stop`.
+    pub(crate) fn emit_transient_cleared(&self, name: &str, label: &str, deliberate_stop: bool) {
         self.recent_transient.lock().unwrap().insert(
-            name.to_owned(),
+            (name.to_owned(), label.to_owned()),
             (deliberate_stop, std::time::Instant::now()),
         );
         self.emit_dashboard_event(DashboardEvent::TransientCleared {
             seq: self.next_seq(),
             name: name.to_owned(),
+            transient_kind: label.to_owned(),
         });
     }
 
@@ -1222,9 +1238,7 @@ impl Coordinator {
         let now = std::time::Instant::now();
         let mut map = self.recent_transient.lock().unwrap();
         map.retain(|_, (_, ts)| now.duration_since(*ts) <= grace);
-        map.iter()
-            .map(|(k, (deliberate, _))| (k.clone(), *deliberate))
-            .collect()
+        fold_tombstones_by_agent(map.iter().map(|((a, _), (d, _))| (a.as_str(), *d)))
     }
 
     /// Record an unexpected crash for `agent`. Called by the crash
@@ -1301,14 +1315,10 @@ impl Coordinator {
     /// `deliberate_stop = true` behind a `false` — reporting an intentional
     /// stop as a crash.
     #[must_use]
-    pub fn transient_snapshot(&self) -> HashMap<String, Vec<TransientState>> {
-        let mut out: HashMap<String, Vec<TransientState>> = HashMap::new();
+    pub fn transient_snapshot(&self) -> HashMap<String, Vec<RunningTransient>> {
+        let mut out: HashMap<String, Vec<RunningTransient>> = HashMap::new();
         for t in self.job_queue.running_transients() {
-            out.entry(t.agent).or_default().push(TransientState {
-                label: t.label,
-                deliberate_stop: t.takes_container_down,
-                since: t.since,
-            });
+            out.entry(t.agent.clone()).or_default().push(t);
         }
         out
     }
@@ -1659,6 +1669,41 @@ pub fn rebuilt_todo_summary(
     summary
 }
 
+#[cfg(test)]
+mod tombstone_fold_tests {
+    use super::fold_tombstones_by_agent;
+
+    #[test]
+    fn a_deliberate_stop_survives_an_incidental_clear_on_the_same_agent() {
+        // The regression this fold exists for. `a` clears two pills in one
+        // grace window: a `StopForUpdate` that really does take the container
+        // down, and a lease-exempt `Prebuild` that doesn't. Keyed by agent
+        // alone the second overwrote the first and the crash watcher reported
+        // a deliberate stop as a crash — order must not matter.
+        let out = fold_tombstones_by_agent([("a", true), ("a", false)].into_iter());
+        assert_eq!(out.get("a"), Some(&true));
+
+        let reversed = fold_tombstones_by_agent([("a", false), ("a", true)].into_iter());
+        assert_eq!(reversed.get("a"), Some(&true));
+    }
+
+    #[test]
+    fn all_incidental_clears_stay_false() {
+        // The other direction has to keep working: nothing deliberate cleared,
+        // so a container going down now really is a crash.
+        let out = fold_tombstones_by_agent([("a", false), ("a", false)].into_iter());
+        assert_eq!(out.get("a"), Some(&false));
+    }
+
+    #[test]
+    fn agents_do_not_bleed_into_each_other() {
+        let out = fold_tombstones_by_agent([("a", true), ("b", false)].into_iter());
+        assert_eq!(out.get("a"), Some(&true));
+        assert_eq!(out.get("b"), Some(&false));
+        assert_eq!(out.get("c"), None);
+    }
+}
+
 #[cfg(test)]
 mod rebuilt_todo_summary_tests {
     use super::rebuilt_todo_summary;

hive-c0re/src/dashboard/state_snapshot.rs

@@ -533,7 +533,7 @@ fn build_port_conflicts(containers: &[ContainerView]) -> Vec<PortConflict> {
 /// `ContainerView.pending` inline; this list only catches pre-creation.
 fn build_transient_views(
     containers: &[ContainerView],
-    transient_snapshot: &std::collections::HashMap<String, Vec<crate::coordinator::TransientState>>,
+    transient_snapshot: &std::collections::HashMap<String, Vec<crate::job_queue::RunningTransient>>,
 ) -> Vec<TransientView> {
     transient_snapshot
         .iter()

hive-c0re/src/dashboard_events.rs

@@ -157,7 +157,18 @@ pub enum DashboardEvent {
     },
     /// The matching lifecycle action resolved (success or failure).
     /// Clients drop the spinner row.
-    TransientCleared { seq: u64, name: String },
+    ///
+    /// `transient_kind` matches the `TransientSet` that opened this pill, and is
+    /// what makes the pair addressable: an agent can hold **several** pills at
+    /// once (the transient set tests status alone, so a lease-exempt `Prebuild`
+    /// and a lease-holding `StopForUpdate` are both live). Without it a client
+    /// holding two open pills for one agent cannot tell which one cleared, and
+    /// has to drop both or guess.
+    TransientCleared {
+        seq: u64,
+        name: String,
+        transient_kind: String,
+    },
     /// One container row changed — new container appeared (post-spawn
     /// finalise), an existing one flipped `running` / `needs_update` /
     /// `sha`, etc. Clients upsert by `container.name`. Payload carries
@@ -392,6 +403,7 @@ mod tests {
             DashboardEvent::TransientCleared {
                 seq: 1,
                 name: "x".into(),
+                transient_kind: "rebuilding".into(),
             },
             DashboardEvent::ContainerRemoved {
                 seq: 1,

hive-c0re/src/job_queue/model.rs

@@ -389,7 +389,7 @@ impl NodeKind {
     }
 
     /// Whether running this node is *expected* to take the agent's container
-    /// down. Feeds `TransientState::deliberate_stop`, which the crash watcher
+    /// down. Feeds `RunningTransient::takes_container_down`, which the crash watcher
     /// reads to tell an intentional stop from a crash.
     ///
     /// This is a **safety** question, not a display one — it decides whether a

hive-c0re/src/job_queue/scheduler.rs

@@ -190,7 +190,7 @@ fn reconcile_transients(coord: &Arc<Coordinator>, prev: &mut TransientSeen) {
             .iter()
             .any(|t| &t.agent == agent && &t.label == label);
         if !still {
-            coord.emit_transient_cleared(agent, *deliberate);
+            coord.emit_transient_cleared(agent, label, *deliberate);
         }
         still
     });

hive-c0re/src/workers/crash_watch.rs

@@ -97,7 +97,7 @@ fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet<String>, current:
         // a deliberate stop rather than a crash.
         let active = transients
             .get(stopped)
-            .map(|sts| sts.iter().any(|st| st.deliberate_stop))
+            .map(|sts| sts.iter().any(|st| st.takes_container_down))
             .or_else(|| coord.crash_watch_suppressed(stopped).then_some(true));
         let recently_cleared = recent.get(stopped).copied();
         if is_deliberate_stop(active, recently_cleared) {