diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 58749694..a1525d37 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -34,7 +34,6 @@ mod snapshot_push; mod socket_server; mod stats; mod stores; -mod swarm_agent_status; mod swarm_notices; mod swarm_queue; mod swarm_status; @@ -499,10 +498,6 @@ 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 deleted file mode 100644 index 2514927c..00000000 --- a/hive-c0re/src/swarm_agent_status.rs +++ /dev/null @@ -1,146 +0,0 @@ -//! 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 deleted file mode 100644 index 98381465..00000000 --- a/swarm-queue-client/src/agent_status.rs +++ /dev/null @@ -1,136 +0,0 @@ -//! 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. -/// -/// Plain code span above, deliberately not an intra-doc link — this crate -/// doesn't depend on `hive_types`, and a bracketed reference to its `Ident` -/// type fails `cargo doc` with "no item named `hive_types` in scope" under -/// this workspace's `docs-rustdoc` CI job, which treats broken doc links as -/// errors. -#[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 ebe213d3..d2acc1c3 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -171,11 +171,6 @@ 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. ///