hives report an agent being updated: its status shows updating from the update's stop until the container is back up

A hive marks an agent updating when StopForUpdate stops its running container and clears it when the tail Reconcile or Start settles it, success or failure. The flag rides in the agent status snapshot (absent when false, so older readers and documents are unchanged), the status publisher sends a change at once instead of at the next sweep, and the swarm UI card shows an updating badge.

hive-c0re/src/coordinator.rs

@@ -101,6 +101,17 @@ pub struct Coordinator {
     /// orchestrations that can be in flight for different agents
     /// simultaneously.
     pause_pending: Mutex<HashSet<String>>,
+    /// Agents an update has taken down and not yet brought back: set when
+    /// `StopForUpdate` stops a running container, cleared when the tail
+    /// `Reconcile` / `Start` settles the agent. Published in the agent's
+    /// status snapshot so the swarm reads the gap as an update and not as a
+    /// stopped or lost agent. In memory only: a hive restart mid-update loses
+    /// the flag, and the status sweep then reads the agent as stopped until it
+    /// is back, as it did before the flag existed.
+    updating: Mutex<HashSet<String>>,
+    /// Woken when [`Self::set_updating`] changes the set, so the status
+    /// publisher sends the change at once instead of at its next sweep.
+    updating_changed: tokio::sync::Notify,
     /// Unified wire-facing event channel feeding the dashboard SSE
     /// stream. Carries broker messages (mirrored from `broker.subscribe`
     /// by the forwarder task in `main.rs`) and dashboard-only mutation
@@ -365,6 +376,8 @@ impl Coordinator {
             recent_transient: Mutex::new(HashMap::new()),
             graceful_stop_pending: Mutex::new(HashSet::new()),
             pause_pending: Mutex::new(HashSet::new()),
+            updating: Mutex::new(HashSet::new()),
+            updating_changed: tokio::sync::Notify::new(),
             dashboard_events,
             event_seq: AtomicU64::new(0),
 
@@ -618,6 +631,34 @@ impl Coordinator {
         });
     }
 
+    /// Mark `name` as being updated (`true`) or settled (`false`), and wake the
+    /// status publisher when that changed something.
+    pub fn set_updating(&self, name: &str, updating: bool) {
+        let changed = {
+            let mut set = self.updating.lock().unwrap();
+            if updating {
+                set.insert(name.to_owned())
+            } else {
+                set.remove(name)
+            }
+        };
+        if changed {
+            self.updating_changed.notify_one();
+        }
+    }
+
+    /// Whether an update has `name` down and not yet back up.
+    #[must_use]
+    pub fn is_updating(&self, name: &str) -> bool {
+        self.updating.lock().unwrap().contains(name)
+    }
+
+    /// Resolves when the set of updating agents changed since it was last
+    /// awaited, for the status publisher.
+    pub async fn updating_changed(&self) {
+        self.updating_changed.notified().await;
+    }
+
     /// 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).

hive-c0re/src/job_queue/exec.rs

@@ -841,7 +841,13 @@ async fn run_reconcile(
 ) -> Result<Option<NodeKind>> {
     let running = host.is_running(name).await;
     let wanted = coord.power.get_or_seed(name, running);
-    Ok(match reconcile_action(wanted, running) {
+    let action = reconcile_action(wanted, running);
+    // The update ends here unless this plans the start that brings the agent
+    // back; `run_start` ends it then, once the container is up or has failed.
+    if !matches!(action, ReconcileAction::Start) {
+        coord.set_updating(name, false);
+    }
+    Ok(match action {
         ReconcileAction::Start => Some(NodeKind::Start {
             agent: name.to_owned(),
         }),
@@ -870,7 +876,12 @@ async fn run_start(host: &impl Host, coord: &Arc<Coordinator>, name: &str) -> Re
     let agent_dir = crate::paths::agent_runtime_dir(name);
     let hive = coord.hive_env();
     let paths = Coordinator::agent_paths(name, agent_dir)?;
-    host.start(name, &hive, &paths).await?;
+    // Whatever the outcome, the update this start belongs to is over: the
+    // container is back up, or the node fails and the agent is as stopped as
+    // it looks.
+    let started = host.start(name, &hive, &paths).await;
+    coord.set_updating(name, false);
+    started?;
     // Bind the MCP listener immediately after starting the container.
     // The preamble created the runtime dir; the container is now coming
     // up and will connect to this socket on its first turn. Event-driven
@@ -906,6 +917,11 @@ async fn run_stop_for_update(host: &impl Host, coord: &Arc<Coordinator>, name: &
         // stopped state and would otherwise seed a running-but-unknown
         // agent as `Offline`, stranding it down after its own rebuild.
         coord.power.get_or_seed(name, true);
+        // The agent is down for the update from here until the tail
+        // `Reconcile` / `Start` settles it. Set before the kill so the status
+        // never reads as plain stopped, and so a failed kill still gets cleared
+        // by that tail.
+        coord.set_updating(name, true);
         host.kill(name).await?;
         coord.rescan_containers_and_emit().await;
     }

hive-c0re/src/job_queue/exec/tests.rs

@@ -609,8 +609,68 @@ async fn stop_for_update_failure_fails_the_node_after_seeding() {
     assert_eq!(coord.power.get("alice"), Some(Wanted::Up));
 }
 
+#[tokio::test]
+async fn an_update_is_marked_from_the_stop_until_the_agent_is_settled() {
+    let (_dir, coord) = coordinator();
+    // A stopped container is no update: nothing was taken down.
+    run_stop_for_update(&FakeHost::down(), &coord, "alice")
+        .await
+        .expect("noop");
+    assert!(!coord.is_updating("alice"));
+    // Marked by the stop, and still marked when the kill fails: the tail
+    // Reconcile is what ends it.
+    run_stop_for_update(&FakeHost::up().failing("kill"), &coord, "alice")
+        .await
+        .expect_err("kill fails");
+    assert!(coord.is_updating("alice"));
+    coord.set_updating("alice", false);
+    run_stop_for_update(&FakeHost::up(), &coord, "alice")
+        .await
+        .expect("stop ok");
+    assert!(coord.is_updating("alice"));
+    assert!(!coord.is_updating("bob"), "per agent");
+}
+
+#[tokio::test]
+async fn a_change_of_the_updating_set_wakes_the_status_publisher_once() {
+    let (_dir, coord) = coordinator();
+    coord.set_updating("alice", true);
+    tokio::time::timeout(std::time::Duration::from_secs(1), coord.updating_changed())
+        .await
+        .expect("woken by the change");
+    // The same state again is no change, so there is nothing to wake for.
+    coord.set_updating("alice", true);
+    assert!(
+        tokio::time::timeout(
+            std::time::Duration::from_millis(50),
+            coord.updating_changed()
+        )
+        .await
+        .is_err()
+    );
+}
+
 // --- Reconcile --------------------------------------------------------------
 
+#[tokio::test]
+async fn reconcile_ends_the_update_unless_it_plans_the_start_that_does() {
+    let (_dir, coord) = coordinator();
+    // Wanted up and down: the start node ends the update, not this.
+    coord.power.set("alice", Wanted::Up);
+    coord.set_updating("alice", true);
+    run_reconcile(&FakeHost::down(), &coord, "alice")
+        .await
+        .expect("ok");
+    assert!(coord.is_updating("alice"));
+    // Wanted offline (a deliberately stopped agent's rebuild): nothing comes
+    // back, so the update is over.
+    coord.power.set("alice", Wanted::Offline);
+    run_reconcile(&FakeHost::down(), &coord, "alice")
+        .await
+        .expect("ok");
+    assert!(!coord.is_updating("alice"));
+}
+
 #[tokio::test]
 async fn reconcile_fans_out_start_for_wanted_up_and_down() {
     let (_dir, coord) = coordinator();
@@ -934,6 +994,21 @@ async fn start_starts_then_registers_the_listener() {
     assert_eq!(host.calls(), ["start alice", "register_agent alice"]);
 }
 
+#[tokio::test]
+async fn a_start_ends_the_update_whether_it_works_or_not() {
+    let (_dir, coord) = coordinator();
+    coord.set_updating("alice", true);
+    run_start(&FakeHost::down(), &coord, "alice")
+        .await
+        .expect("ok");
+    assert!(!coord.is_updating("alice"));
+    coord.set_updating("alice", true);
+    run_start(&FakeHost::down().failing("start"), &coord, "alice")
+        .await
+        .expect_err("start");
+    assert!(!coord.is_updating("alice"));
+}
+
 #[tokio::test]
 async fn a_failed_start_registers_no_listener() {
     let (_dir, coord) = coordinator();

hive-c0re/src/main.rs

@@ -318,7 +318,7 @@ async fn cmd_serve(
     // Offer each locally-hosted agent's status upward too — a sibling task,
     // not a part of the above; see swarm_agent_status's module doc for why
     // the two buckets stay separate.
-    swarm_agent_status::spawn(coord.shutdown_rx());
+    swarm_agent_status::spawn(std::sync::Arc::clone(&coord));
     // Squid's users file: the hashes of this hive's agents' proxy credentials.
     workers::squid_users::spawn(coord.shutdown_rx());
     // Per-agent events.sqlite + bash-tasks file cleanup now runs

hive-c0re/src/swarm_agent_status.rs

@@ -37,12 +37,14 @@ const FAILURES_BEFORE_WARNING: u32 = 3;
 /// `crate::swarm_queue::wait_for_client()`, one connection per process — so a
 /// hive with no queue coordinates pays for this decision once, not twice.
 ///
-/// Takes no `Coordinator` handle, unlike its sibling: this task only reads
-/// agent state that already exists on disk / in the container runtime
-/// (`lifecycle::agents_for_meta_listing`, `container_view::read_agent_status_bounded`),
-/// it never touches the job queue or broker the way `swarm_status::spawn`'s
-/// deploy-event listener does.
-pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
+/// Takes the `Coordinator` only for the set of agents an update has down
+/// (`Coordinator::is_updating`), which it publishes with the rest and wakes on.
+/// Everything else it reads is agent state that already exists on disk / in the
+/// container runtime (`lifecycle::agents_for_meta_listing`,
+/// `container_view::read_agent_status_bounded`); it never touches the job queue
+/// or broker the way `swarm_status::spawn`'s deploy-event listener does.
+pub fn spawn(coord: std::sync::Arc<crate::coordinator::Coordinator>) {
+    let mut shutdown = coord.shutdown_rx();
     let Some(hive) = crate::container_view::hive_swarm_names().0 else {
         // Already warned once by `swarm_status::spawn`, which runs the
         // identical check — a second warning for the same missing env var
@@ -61,7 +63,7 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
             FAILURES_BEFORE_WARNING,
         );
         loop {
-            match publish_all(&client, &hive).await {
+            match publish_all(&client, &hive, &coord).await {
                 Ok(published) => {
                     health.record_ok();
                     tracing::debug!(published, "agent status: sweep complete");
@@ -88,6 +90,9 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
             // one whose agents someone is looking at.
             tokio::select! {
                 () = tokio::time::sleep(PUBLISH_INTERVAL) => {}
+                // An agent entering or leaving an update is the one change worth
+                // sending ahead of the sweep: the window can be shorter than it.
+                () = coord.updating_changed() => {}
                 _ = shutdown.changed() => {
                     tracing::info!("agent status: shutdown signal received");
                     break;
@@ -104,7 +109,11 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
 /// all) is the one thing that fails the whole sweep; a single agent's
 /// status read or publish failing is logged and skipped; skipping one
 /// agent should not report every *other* agent on this hive as stale too.
-async fn publish_all(client: &async_nats::Client, hive: &str) -> Result<usize> {
+async fn publish_all(
+    client: &async_nats::Client,
+    hive: &str,
+    coord: &crate::coordinator::Coordinator,
+) -> Result<usize> {
     // Same hang risk `swarm_status::publish` guards against: an unconnected
     // client does not fail a JetStream request, it hangs on it.
     swarm_queue_client::ensure_connected(client)?;
@@ -134,6 +143,7 @@ async fn publish_all(client: &async_nats::Client, hive: &str) -> Result<usize> {
             status_text,
             status_set_at,
             running,
+            updating: coord.is_updating(&spec.name),
         };
         let bytes = match serde_json::to_vec(&payload) {
             Ok(b) => b,