diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index a1525d37..58749694 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -34,6 +34,7 @@ mod snapshot_push; mod socket_server; mod stats; mod stores; +mod swarm_agent_status; mod swarm_notices; mod swarm_queue; mod swarm_status; @@ -498,6 +499,10 @@ async fn cmd_serve( // swarm_status, which owns the whole task including its own decision // not to start. swarm_status::spawn(std::sync::Arc::clone(&coord), coord.shutdown_rx()); + // 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()); // Per-agent events.sqlite + bash-tasks file cleanup now runs // agent-side in the harness (`hive_agent::vacuum`): the files are // agent-owned, so host-side deletes hit PermissionDenied / readonly-db diff --git a/hive-c0re/src/swarm_agent_status.rs b/hive-c0re/src/swarm_agent_status.rs new file mode 100644 index 00000000..2514927c --- /dev/null +++ b/hive-c0re/src/swarm_agent_status.rs @@ -0,0 +1,146 @@ +//! Offering each locally-hosted agent's status upward to the swarm. +//! +//! Sibling task to [`crate::swarm_status`], not a field folded into it — see +//! `swarm_queue_client::agent_status`'s module doc for why hive status and +//! agent status live in separate buckets. Same direction, though: a hive +//! **offers** what it already knows about its own agents (via +//! [`crate::container_view::read_agent_status_live`], the same read +//! `GetAgentMeta` serves to the dashboard), and the swarm controller never +//! reaches down to collect it. +//! +//! **One publish per agent, not one document for the hive.** Each agent +//! gets its own key (`swarm_queue_client::agent_status::key`), so one +//! agent's status changing does not touch any other agent's key, and a +//! consumer watching one agent reads exactly its own revision history. + +use std::time::Duration; + +use anyhow::{Context, Result}; + +use crate::stats::sweep_health::{self, SweepHealth}; + +/// How often this hive offers a snapshot of every agent it hosts. +/// +/// Same value as [`crate::swarm_status::PUBLISH_INTERVAL`] on purpose: both +/// feed the same controller-side `staleAfterSeconds` freshness window, and +/// there is no reason for an agent's status to go stale on a different +/// cadence than the hive's own. +pub const PUBLISH_INTERVAL: Duration = crate::swarm_status::PUBLISH_INTERVAL; + +/// Consecutive failed sweeps before the dashboard banners. Mirrors +/// [`crate::swarm_status`]'s constant of the same name and purpose. +const FAILURES_BEFORE_BANNER: u32 = 3; + +/// Start the publish loop, if this deployment wired up a swarm queue. +/// +/// Shares its connect gate with [`crate::swarm_status::spawn`] — both read +/// `crate::swarm_queue::client()`, which connects once per process — so a +/// hive with no queue configured 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_live`), +/// 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) { + let Some(hive) = crate::container_view::hive_swarm_names().0 else { + // Already bannered once by `swarm_status::spawn`, which runs the + // identical check — a second banner for the same missing env var + // would just be the same fact told twice. + return; + }; + + tokio::spawn(async move { + let Some(client) = crate::swarm_queue::client().await else { + return; + }; + + let mut health = + SweepHealth::new("swarm_agent_status_publish", "warn", FAILURES_BEFORE_BANNER); + loop { + match publish_all(&client, &hive).await { + Ok(published) => { + health.record_ok(); + tracing::debug!(published, "agent status: sweep complete"); + } + Err(e) => { + tracing::warn!(error = ?e, "agent status: sweep failed"); + let err = format!("{e:#}"); + health.record_err(|ctx| { + let age = ctx.since_last_ok.map_or_else( + || "no success this session".to_owned(), + |d| format!("last ok {} ago", sweep_health::fmt_age(d)), + ); + format!( + "agent status publishing is failing ({} consecutive, {age}) \ + — the swarm sees this hive's agents as stale, the agents \ + themselves are unaffected: {err}", + ctx.consecutive + ) + }); + } + } + // Publish first, then wait — same reasoning as + // `swarm_status::spawn`: a hive that just came up is exactly the + // one whose agents someone is looking at. + tokio::select! { + () = tokio::time::sleep(PUBLISH_INTERVAL) => {} + _ = shutdown.changed() => { + tracing::info!("agent status: shutdown signal received"); + break; + } + } + } + }); +} + +/// Offer one snapshot of every agent this hive currently hosts. +/// +/// Returns the count published, purely so the caller can log it — nothing +/// downstream reads the number. Enumeration failure (cannot list agents at +/// 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 { + // 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)?; + + let agents = crate::lifecycle::agents_for_meta_listing() + .await + .context("enumerating this hive's agents")?; + + let store = swarm_queue_client::agent_status::open_or_create(client).await?; + let mut published = 0usize; + for spec in &agents { + let Ok(ident) = hive_types::Ident::parse(&spec.name) else { + // An agent name that fails `Ident::parse` here would mean + // something already on disk violates the naming charset every + // other path enforces at creation — worth a warning, not a + // reason to abandon everyone else's publish. + tracing::warn!(agent = %spec.name, "agent status: name failed Ident::parse, skipping"); + continue; + }; + let (status_text, status_set_at, running) = + crate::container_view::read_agent_status_live(&ident).await; + let payload = swarm_queue_client::agent_status::AgentStatus { + status_text, + status_set_at, + running, + }; + let bytes = match serde_json::to_vec(&payload) { + Ok(b) => b, + Err(e) => { + tracing::warn!(agent = %spec.name, error = %e, "agent status: serialising failed, skipping"); + continue; + } + }; + let key = swarm_queue_client::agent_status::key(hive, &spec.name); + if let Err(e) = store.put(key, bytes.into()).await { + tracing::warn!(agent = %spec.name, error = %e, "agent status: publish failed, skipping"); + continue; + } + published += 1; + } + Ok(published) +} diff --git a/swarm-queue-client/src/agent_status.rs b/swarm-queue-client/src/agent_status.rs new file mode 100644 index 00000000..9174237c --- /dev/null +++ b/swarm-queue-client/src/agent_status.rs @@ -0,0 +1,130 @@ +//! The per-agent status KV bucket: one key per `(hive, agent)` pair. +//! +//! Deliberately **not** a field inside [`crate::status::BUCKET`]'s hive +//! document, even though a hive already publishes its own status there. +//! Three reasons, none of which show up until the second write: +//! +//! - **Cardinality and write rate.** Hive status is one value per hive. +//! Agent status is N values per hive, changing independently and far +//! more often — folded into the hive document, every agent's change +//! rewrites the whole thing, and two agents changing at once race on +//! the same key. +//! - **A KV key is the unit of update and of watch.** A consumer that +//! cares about one agent wants a per-key read with its own revision, +//! not a hive-wide document to diff. +//! - **Staleness means different things.** A hive can be healthy while +//! one agent is wedged, and the reverse; one timestamp cannot answer +//! both questions honestly. +//! +//! The transport *is* shared — this rides the same [`async_nats::Client`] +//! and the same auth callout as [`crate::status`] and [`crate::wanted`], +//! just a different bucket. + +#[cfg(feature = "kv")] +use crate::Error; + +/// The KV bucket per-agent status snapshots are published into, keyed by +/// [`key`]. +/// +/// A constant and not an option, for the reason [`crate::status::BUCKET`] +/// gives: writer (a hive) and reader (the controller) must name the same +/// bucket, and an option is a way for the two to disagree about which one +/// that is. +pub const BUCKET: &str = "agent-status"; + +/// The bucket key for one agent on one hive. +/// +/// `/`-joined rather than NATS's usual `.`-joined subject style: this is a +/// KV key, not a subject, and neither `hive` nor `agent` names can contain +/// `/` ([`hive_types::Ident`]'s charset is `[a-z0-9-]`), so the join is +/// unambiguous to split back apart if a consumer ever needs to. +#[must_use] +pub fn key(hive: &str, agent: &str) -> String { + format!("{hive}/{agent}") +} + +/// One agent's status snapshot, as published under [`key`]. +/// +/// Mirrors the tuple `container_view::read_agent_status_live` returns on +/// the publishing side — this is that value, not a recomputation of it. +/// Nothing here stamps a time: same rule as [`crate::status`], freshness is +/// derived by the reader from when the value landed in the bucket, so a +/// hive with a wrong clock only skews its own payload. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentStatus { + /// The agent's free-text status (`set_status`), or `None` if it has + /// never set one, or if the container isn't running. + pub status_text: Option, + /// Unix timestamp the status was last set. `None` alongside + /// `status_text: None` — the two are never independently absent. + pub status_set_at: Option, + /// Whether the container is currently running. `false` forces the two + /// fields above to `None` even if an on-disk status file is stale from + /// before the container stopped — the same "stopped containers have + /// stale state" rule `read_agent_status_live` already enforces. + pub running: bool, +} + +/// Open the agent-status bucket, creating it if nothing has yet. +/// +/// `history: 1`, same rationale as [`crate::status::open_or_create`]: a +/// consumer wants the last thing each agent said, not a log of everything +/// it has ever said. +#[cfg(feature = "kv")] +pub async fn open_or_create( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + match js.get_key_value(BUCKET).await { + Ok(store) => Ok(store), + Err(e) => { + tracing::info!( + bucket = BUCKET, + reason = %e, + "agent-status bucket not available, creating it" + ); + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: BUCKET.to_owned(), + description: "Last status snapshot offered by each agent".to_owned(), + history: 1, + ..Default::default() + }) + .await + .map_err(|source| Error::CreateBucket { + bucket: BUCKET, + source, + }) + } + } +} + +#[cfg(test)] +mod tests { + use super::{AgentStatus, key}; + + #[test] + fn key_joins_hive_and_agent_with_a_slash() { + assert_eq!(key("prod", "iris"), "prod/iris"); + } + + #[test] + fn agent_status_round_trips() { + let status = AgentStatus { + status_text: Some("reviewing a pull request".to_owned()), + status_set_at: Some(1_725_000_000), + running: true, + }; + 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 a_stopped_agent_serialises_with_no_status() { + let status = AgentStatus::default(); + assert_eq!( + serde_json::to_string(&status).expect("serialises"), + r#"{"status_text":null,"status_set_at":null,"running":false}"# + ); + } +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index d2acc1c3..ebe213d3 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -171,6 +171,11 @@ pub mod status; /// "absence is not a deletion order" rule rather than inheriting either. pub mod wanted; +/// The bucket per-agent status snapshots are published into — one key per +/// `(hive, agent)`, deliberately a sibling of [`status`] rather than a field +/// inside it. See the module doc for why the two must not merge. +pub mod agent_status; + /// The subject the swarm controller publishes on when the hive-wide knowledge /// repository has changed. One writer, many readers — every hive subscribes. ///