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();