hive-c0re: publish each agent's status upward to the swarm queue (#3341 item 1)

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

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<bool>) {
+    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<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)?;
+
+    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)
+}

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<String>,
+    /// Unix timestamp the status was last set. `None` alongside
+    /// `status_text: None` — the two are never independently absent.
+    pub status_set_at: Option<i64>,
+    /// 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<async_nats::jetstream::kv::Store, Error> {
+    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}"#
+        );
+    }
+}

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.
 ///