From 748ef2da0816e5e070ec13b023586ff8eb05cac6 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 11 Nov 2026 15:41:19 +0100 Subject: [PATCH] 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. --- docs/2-handbook/swarm/ui.md | 4 +- docs/4-implementation/web-ui/swarm-ui.md | 6 ++ .../packages/swarm-ui/src/agent/AgentTypes.ts | 9 +++ .../swarm-ui/src/agent/roster/AgentCard.tsx | 8 ++ .../swarm-ui/test/agent-types.test.mjs | 17 +++++ hive-c0re/src/coordinator.rs | 41 ++++++++++ hive-c0re/src/job_queue/exec.rs | 20 ++++- hive-c0re/src/job_queue/exec/tests.rs | 75 +++++++++++++++++++ hive-c0re/src/main.rs | 2 +- hive-c0re/src/swarm_agent_status.rs | 26 +++++-- swarm-wire/src/agent_status.rs | 31 ++++++++ 11 files changed, 227 insertions(+), 12 deletions(-) diff --git a/docs/2-handbook/swarm/ui.md b/docs/2-handbook/swarm/ui.md index 1f3ee93a7..10dfa8f69 100644 --- a/docs/2-handbook/swarm/ui.md +++ b/docs/2-handbook/swarm/ui.md @@ -25,7 +25,9 @@ An agent created here or with `swarmctl agent create` starts `paused`; set it `up` from its card. A card shows **login needed** when the agent's claude has no session, with a **log in** button that runs the claude login from a dialog (see [`claude-login.md`](claude-login.md)). The **wanted** badge changes an agent's -declared state, destroy included. +declared state, destroy included. A card shows **updating** while the hive +updates the agent, from the moment the update stops its container until the +container is back up, so the stopped agent doesn't read as lost. Each page, its controls and what they send are in [the swarm UI reference](../../4-implementation/web-ui/swarm-ui.md). diff --git a/docs/4-implementation/web-ui/swarm-ui.md b/docs/4-implementation/web-ui/swarm-ui.md index 1400d11cb..900e3332c 100644 --- a/docs/4-implementation/web-ui/swarm-ui.md +++ b/docs/4-implementation/web-ui/swarm-ui.md @@ -74,6 +74,12 @@ works too. The settings section of an opened card has a **claude login** row: wh the swarm stored the agent's login, or that none is, with a **delete login** action that asks first. +A card shows **updating** when the agent's status snapshot has `updating` set. +The hive sets it when an update stops a running container and clears it when +the tail reconcile or start settles the agent, whether the container came back +or not, and it sends the change at once instead of waiting for its next status +sweep. A hive that doesn't publish the field shows no badge. + ### The agents page `/agents` is a column of separate cards: a filter card, then one diff --git a/frontend/packages/swarm-ui/src/agent/AgentTypes.ts b/frontend/packages/swarm-ui/src/agent/AgentTypes.ts index fbc093711..7122ab52d 100644 --- a/frontend/packages/swarm-ui/src/agent/AgentTypes.ts +++ b/frontend/packages/swarm-ui/src/agent/AgentTypes.ts @@ -15,6 +15,9 @@ export interface AgentStatusSnapshot { status_text: string | null; status_set_at: number | null; running: boolean; + // The hive is updating the agent: down from the update's stop until the + // container is back up. Absent from a hive that does not publish it. + updating?: boolean; } // `"up"` / `"offline"` / `"destroyed"`, wire-spelled by @@ -72,6 +75,12 @@ export function hivePauseReason( return row.paused_by_hive ? row.paused_by_hive : null; } +// Whether the hive is updating the agent right now, so it reads as stopped for +// a reason. Only a snapshot that says so counts; none shows nothing. +export function isUpdating(row: Pick): boolean { + return row.snapshot?.updating === true; +} + // Same tone pairing as HivesCard — one freshness enum shared by both // endpoints, so the same tone rule applies to both pages — but // `unknown`'s label diverges deliberately: an agent outside the diff --git a/frontend/packages/swarm-ui/src/agent/roster/AgentCard.tsx b/frontend/packages/swarm-ui/src/agent/roster/AgentCard.tsx index ab08a64cf..fff2b1eab 100644 --- a/frontend/packages/swarm-ui/src/agent/roster/AgentCard.tsx +++ b/frontend/packages/swarm-ui/src/agent/roster/AgentCard.tsx @@ -25,6 +25,7 @@ import { StatusMessage } from "../detail/StatusMessage.js"; import { termPaths } from "../terminal/term-paths.js"; import { hivePauseReason, + isUpdating, showsLoginNeeded, type AgentRow, } from "../AgentTypes.js"; @@ -66,6 +67,13 @@ export function AgentCard({ name={{row.name}} status={ <> + {isUpdating(row) ? ( + + ) : null} {hivePauseReason(row) !== null ? ( { assert.equal(hivePauseReason({ paused_by_hive: "" }), null); assert.equal(hivePauseReason({}), null); }); + +test("only a snapshot that says the hive is updating the agent shows the badge", () => { + const snapshot = (updating) => ({ + snapshot: { + status_text: null, + status_set_at: null, + running: false, + updating, + }, + }); + assert.equal(isUpdating(snapshot(true)), true); + assert.equal(isUpdating(snapshot(false)), false); + // A hive that does not publish the field, and an agent with no snapshot. + assert.equal(isUpdating(snapshot(undefined)), false); + assert.equal(isUpdating({ snapshot: null }), false); +}); diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index cc60cc0bd..cc53be05b 100644 --- a/hive-c0re/src/coordinator.rs +++ b/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>, + /// 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>, + /// 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). diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index f969f4fe5..0eaae9801 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -841,7 +841,13 @@ async fn run_reconcile( ) -> Result> { 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, 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, 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; } diff --git a/hive-c0re/src/job_queue/exec/tests.rs b/hive-c0re/src/job_queue/exec/tests.rs index 618dd8edd..12c0ee69e 100644 --- a/hive-c0re/src/job_queue/exec/tests.rs +++ b/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(); diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 8d30592f6..669c7e07d 100644 --- a/hive-c0re/src/main.rs +++ b/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 diff --git a/hive-c0re/src/swarm_agent_status.rs b/hive-c0re/src/swarm_agent_status.rs index 27bdb2535..c30a0fb12 100644 --- a/hive-c0re/src/swarm_agent_status.rs +++ b/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) { +/// 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) { + 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) { 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) { // 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) { /// 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 { +async fn publish_all( + client: &async_nats::Client, + hive: &str, + coord: &crate::coordinator::Coordinator, +) -> Result { // 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 { status_text, status_set_at, running, + updating: coord.is_updating(&spec.name), }; let bytes = match serde_json::to_vec(&payload) { Ok(b) => b, diff --git a/swarm-wire/src/agent_status.rs b/swarm-wire/src/agent_status.rs index b9c2f7e53..f23ea57de 100644 --- a/swarm-wire/src/agent_status.rs +++ b/swarm-wire/src/agent_status.rs @@ -89,6 +89,15 @@ pub struct AgentStatus { /// before the container stopped — the same "stopped containers have /// stale state" rule `read_agent_status_live` already enforces. pub running: bool, + /// Whether the hive is updating the agent: from the moment an update stops + /// its container until the container is back up (or the update gave up and + /// the hive settled the agent's power state). The agent reads as stopped + /// across that window, and this says why. Left out of the document when + /// `false`, so a snapshot of an agent not being updated is byte for byte + /// what a hive without the field publishes, and a document without it + /// reads as `false`. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub updating: bool, } #[cfg(test)] @@ -136,12 +145,34 @@ mod tests { status_text: Some("reviewing a pull request".to_owned()), status_set_at: Some(1_725_000_000), running: true, + updating: false, }; let json = serde_json::to_string(&status).expect("serialises"); let decoded: AgentStatus = serde_json::from_str(&json).expect("decodes"); assert_eq!(decoded, status); } + #[test] + fn an_updating_agent_says_so_and_a_document_without_the_field_does_not() { + let updating = AgentStatus { + running: false, + updating: true, + ..AgentStatus::default() + }; + let json = serde_json::to_string(&updating).expect("serialises"); + assert_eq!( + json, + r#"{"status_text":null,"status_set_at":null,"running":false,"updating":true}"# + ); + let decoded: AgentStatus = serde_json::from_str(&json).expect("decodes"); + assert!(decoded.updating); + // A hive that predates the field publishes none. + let old: AgentStatus = + serde_json::from_str(r#"{"status_text":null,"status_set_at":null,"running":true}"#) + .expect("decodes"); + assert!(!old.updating); + } + #[test] fn a_stopped_agent_serialises_with_no_status() { let status = AgentStatus::default();