From 9222a051f544c2f5a08a43144e9a1b3ca458e54f Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 18 Oct 2026 12:17:46 +0200 Subject: [PATCH] swarm-wire: take agent_status's wire types --- Cargo.lock | 1 + hive-c0re/Cargo.toml | 1 + hive-c0re/src/swarm_agent_status.rs | 8 +- swarm-controller/src/agent_status.rs | 10 +- swarm-nats-auth/src/policy.rs | 8 +- swarm-queue-client/src/agent_icon.rs | 4 +- swarm-queue-client/src/agent_status.rs | 168 +------------------------ swarm-queue-client/src/lib.rs | 7 +- swarm-wire/src/agent_status.rs | 153 ++++++++++++++++++++++ swarm-wire/src/lib.rs | 1 + 10 files changed, 180 insertions(+), 181 deletions(-) create mode 100644 swarm-wire/src/agent_status.rs diff --git a/Cargo.lock b/Cargo.lock index f6252caa..4f6fa732 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1802,6 +1802,7 @@ dependencies = [ "swarm-matrix-client", "swarm-queue-client", "swarm-secret-client", + "swarm-wire", "tempfile", "tokio", "tokio-stream", diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 578a267c..6bbb7cf2 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -63,6 +63,7 @@ swarm-queue-client = { workspace = true, features = [ ] } swarm-matrix-client.workspace = true swarm-secret-client.workspace = true +swarm-wire.workspace = true tokio.workspace = true tokio-stream.workspace = true tracing.workspace = true diff --git a/hive-c0re/src/swarm_agent_status.rs b/hive-c0re/src/swarm_agent_status.rs index 9b0bd7df..04c5fb28 100644 --- a/hive-c0re/src/swarm_agent_status.rs +++ b/hive-c0re/src/swarm_agent_status.rs @@ -1,7 +1,7 @@ //! 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 +//! `swarm_wire::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_bounded`], the bounded form of @@ -9,7 +9,7 @@ //! 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 +//! gets its own key (`swarm_wire::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. @@ -127,7 +127,7 @@ async fn publish_all(client: &async_nats::Client, hive: &str) -> Result { tracing::warn!(agent = %spec.name, "agent status: unit state unreadable, skipping"); continue; }; - let payload = swarm_queue_client::agent_status::AgentStatus { + let payload = swarm_wire::agent_status::AgentStatus { status_text, status_set_at, running, @@ -139,7 +139,7 @@ async fn publish_all(client: &async_nats::Client, hive: &str) -> Result { continue; } }; - let key = swarm_queue_client::agent_status::key(hive, &spec.name); + let key = swarm_wire::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; diff --git a/swarm-controller/src/agent_status.rs b/swarm-controller/src/agent_status.rs index 5d51c336..e2e8b9c1 100644 --- a/swarm-controller/src/agent_status.rs +++ b/swarm-controller/src/agent_status.rs @@ -1,7 +1,7 @@ //! Swarm-wide view of what each agent last said about itself. //! //! Sibling of [`crate::status`], not a mode of it — see -//! `swarm_queue_client::agent_status`'s module doc for why the two buckets +//! `swarm_wire::agent_status`'s module doc for why the two buckets //! stay separate. Everything about *how* a snapshot is read (queue is the //! store, no cache, freshness derived at read time, rows come from the //! roster so an empty bucket cannot render as "every agent is fine") is @@ -11,7 +11,7 @@ //! **The roster here is agent *names* only** ([`crate::auth::AuthBridge::list_agent_identities`]) — //! it does not say which hive an agent lives on. That is not a gap this //! module needs to close: a *reporting* agent's hive comes straight out of -//! its bucket key (`swarm_queue_client::agent_status::split_key`), and a +//! its bucket key (`swarm_wire::agent_status::split_key`), and a //! *never-reported* agent's hive is unknowable regardless of what any //! roster might claim, so [`AgentStatusRow::hive`] is `None` in exactly //! that one case rather than a second lookup this module would have to @@ -42,7 +42,7 @@ pub struct AgentStatusRow { pub age_seconds: Option, /// The agent's status text + running flag, unopened. `None` exactly /// when `freshness` is `NeverReported`, or when the stored bytes - /// failed to decode as [`swarm_queue_client::agent_status::AgentStatus`] + /// failed to decode as [`swarm_wire::agent_status::AgentStatus`] /// (logged as a warning naming the agent). pub snapshot: Option, /// The agent's open config PR, if any. Always `None` coming out of @@ -130,7 +130,7 @@ impl AgentStatusReader { .await .context("reading the agent-status bucket's key list")? { - let Some((hive, agent)) = swarm_queue_client::agent_status::split_key(&key) else { + let Some((hive, agent)) = swarm_wire::agent_status::split_key(&key) else { tracing::warn!(%key, "agent-status bucket key doesn't split into hive/agent, skipping"); continue; }; @@ -142,7 +142,7 @@ impl AgentStatusReader { continue; }; let payload = match serde_json::from_slice::< - swarm_queue_client::agent_status::AgentStatus, + swarm_wire::agent_status::AgentStatus, >(&entry.value) { Ok(status) => serde_json::to_value(status).ok(), diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index e6550dc5..5eb39daf 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -412,7 +412,7 @@ impl Policy { /// Same shape and same reason as [`Self::wanted_stream`]: derived from /// the crate constant rather than a third configurable bucket name. fn agent_status_stream() -> String { - format!("KV_{}", swarm_queue_client::agent_status::BUCKET) + format!("KV_{}", swarm_wire::agent_status::BUCKET) } /// What *any* `JetStream` client must be able to ask before it can do @@ -563,7 +563,7 @@ impl Policy { subjects.extend([ format!("$JS.API.STREAM.INFO.{agent_status}"), format!("$JS.API.STREAM.CREATE.{agent_status}"), - format!("$KV.{}.{hive}.*", swarm_queue_client::agent_status::BUCKET), + format!("$KV.{}.{hive}.*", swarm_wire::agent_status::BUCKET), ]); // Reading the caps the swarm sets for the agents this hive runs // (`swarm_queue_client::agent_limits`): a direct get, so read-only, and @@ -782,7 +782,7 @@ mod tests { // A hive publishes its *own* agents' status under // `$KV.agent-status..*` and must reach no further. This is the // arm of the old reader-side guard that still holds. - let bucket = swarm_queue_client::agent_status::BUCKET; + let bucket = swarm_wire::agent_status::BUCKET; let subjects = policy().hive_subjects("alpha"); for forbidden in [ format!("$KV.{bucket}.>"), @@ -814,7 +814,7 @@ mod tests { #[test] fn a_hive_may_write_only_its_own_agents_status_keys() { - let bucket = swarm_queue_client::agent_status::BUCKET; + let bucket = swarm_wire::agent_status::BUCKET; let subjects = policy().hive_subjects("alpha"); let s = Policy::agent_status_stream(); for want in [ diff --git a/swarm-queue-client/src/agent_icon.rs b/swarm-queue-client/src/agent_icon.rs index 1805b03d..e6a9c163 100644 --- a/swarm-queue-client/src/agent_icon.rs +++ b/swarm-queue-client/src/agent_icon.rs @@ -3,7 +3,7 @@ //! //! **Keyed by the agent alone — no hive token**, which is the one way this //! bucket deliberately differs from its otherwise-identical sibling -//! [`crate::agent_status`] (`{hive}.{agent}`). An agent is not fixed to a +//! `swarm_wire::agent_status` (`{hive}.{agent}`). An agent is not fixed to a //! hive and can move between them; an icon keyed by where the agent //! currently runs would be stranded under the old key by a migration, and //! a reader would have to know the placement to ask the question at all. @@ -29,7 +29,7 @@ use crate::Error; /// The KV bucket agent icons are published into, keyed by agent name. /// /// A constant and not an option, for the reason -/// [`crate::agent_status::BUCKET`] gives: writer and reader must name the +/// [`swarm_wire::agent_status::BUCKET`] gives: writer and reader must name the /// same bucket, and an option is a way for the two to disagree. pub const BUCKET: &str = "agent-icons"; diff --git a/swarm-queue-client/src/agent_status.rs b/swarm-queue-client/src/agent_status.rs index 52ba59ff..8a8696ec 100644 --- a/swarm-queue-client/src/agent_status.rs +++ b/swarm-queue-client/src/agent_status.rs @@ -1,103 +1,6 @@ -//! 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, and the separator is load-bearing rather than cosmetic: a KV -/// entry is published to `$KV..`, and NATS wildcards match -/// whole `.`-delimited tokens. Two tokens make a hive's write grant -/// expressible as `$KV.agent-status..*` — every agent of one hive and -/// nothing else. Joined by any character that is not `.`, the key is a -/// single token, and the only grants available are one exact subject per -/// agent or a bucket-wide wildcard that lets any hive overwrite any other -/// hive's agents. -/// -/// Neither name can be empty or contain a `.` (`hive_types::Ident` is -/// `[a-z0-9-]`, non-empty), so the join stays unambiguous to split back -/// apart — and a leading or trailing `.` would be rejected outright by the -/// client's own KV key validation. -/// -/// 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}") -} - -/// The inverse of [`key`]: split a bucket key back into `(hive, agent)`. -/// -/// `None` for a key with zero or more than one `.` — a hive publishing -/// under [`key`] never produces one, so a malformed key means something -/// else wrote this bucket. Splitting on the *first* `.` would be equally -/// valid today (neither name can contain one), but this rejects rather -/// than guesses, so a future name-charset change can't silently start -/// misreading old keys. -#[must_use] -pub fn split_key(key: &str) -> Option<(&str, &str)> { - let mut parts = key.splitn(3, '.'); - let hive = parts.next()?; - let agent = parts.next()?; - if parts.next().is_some() { - return None; - } - Some((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, -} +//! The connection side of the per-agent status KV bucket: opening it for +//! writing. The bucket's name, keys and stored type are +//! `swarm_wire::agent_status`. /// Open the agent-status bucket, creating it if nothing has yet. /// @@ -107,74 +10,13 @@ pub struct AgentStatus { #[cfg(feature = "kv")] pub async fn open_or_create( client: &async_nats::Client, -) -> Result { +) -> Result { let js = async_nats::jetstream::new(client.clone()); crate::kv::open_or_create( &js, - BUCKET, + swarm_wire::agent_status::BUCKET, "agent-status", "Last status snapshot offered by each agent".to_owned(), ) .await } - -#[cfg(test)] -mod tests { - use super::{AgentStatus, BUCKET, key, split_key}; - - #[test] - fn key_joins_hive_and_agent_with_a_dot() { - assert_eq!(key("prod", "iris"), "prod.iris"); - } - - #[test] - fn the_published_subject_carries_the_hive_as_its_own_token() { - // This is what the hive's write grant `$KV.agent-status..*` - // matches on, so the separator is pinned here rather than left as a - // property of `key`'s formatting string. - assert_eq!( - format!("$KV.{BUCKET}.{}", key("prod", "iris")), - "$KV.agent-status.prod.iris" - ); - // Control: a single-token join lands in the same bucket and would - // pass any assertion that only checked the prefix — it is the hive - // token that the grant needs, and only the `.` produces one. - assert!(!format!("$KV.{BUCKET}.prod/iris").starts_with("$KV.agent-status.prod.")); - } - - #[test] - fn split_key_is_the_inverse_of_key() { - assert_eq!(split_key(&key("prod", "iris")), Some(("prod", "iris"))); - } - - #[test] - fn split_key_rejects_a_key_with_no_dot() { - assert_eq!(split_key("iris"), None); - } - - #[test] - fn split_key_rejects_a_key_with_two_dots() { - assert_eq!(split_key("prod.iris.extra"), None); - } - - #[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 f7e782cf..e4afef89 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -195,9 +195,10 @@ 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. +/// Opening 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. The name, keys and stored type, and why the two must not +/// merge, are in `swarm_wire::agent_status`. pub mod agent_status; /// The bucket agent icons are published into — one key per agent, with no diff --git a/swarm-wire/src/agent_status.rs b/swarm-wire/src/agent_status.rs new file mode 100644 index 00000000..b9c2f7e5 --- /dev/null +++ b/swarm-wire/src/agent_status.rs @@ -0,0 +1,153 @@ +//! The per-agent status KV bucket: one key per `(hive, agent)` pair. +//! +//! Deliberately **not** a field inside `swarm_queue_client::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 `swarm_queue_client::status` and +//! `swarm_queue_client::wanted`, +//! just a different bucket. + +/// The KV bucket per-agent status snapshots are published into, keyed by +/// [`key`]. +/// +/// A constant and not an option, for the reason `swarm_queue_client::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, and the separator is load-bearing rather than cosmetic: a KV +/// entry is published to `$KV..`, and NATS wildcards match +/// whole `.`-delimited tokens. Two tokens make a hive's write grant +/// expressible as `$KV.agent-status..*` — every agent of one hive and +/// nothing else. Joined by any character that is not `.`, the key is a +/// single token, and the only grants available are one exact subject per +/// agent or a bucket-wide wildcard that lets any hive overwrite any other +/// hive's agents. +/// +/// Neither name can be empty or contain a `.` (`hive_types::Ident` is +/// `[a-z0-9-]`, non-empty), so the join stays unambiguous to split back +/// apart — and a leading or trailing `.` would be rejected outright by the +/// client's own KV key validation. +#[must_use] +pub fn key(hive: &str, agent: &str) -> String { + format!("{hive}.{agent}") +} + +/// The inverse of [`key`]: split a bucket key back into `(hive, agent)`. +/// +/// `None` for a key with zero or more than one `.` — a hive publishing +/// under [`key`] never produces one, so a malformed key means something +/// else wrote this bucket. Splitting on the *first* `.` would be equally +/// valid today (neither name can contain one), but this rejects rather +/// than guesses, so a future name-charset change can't silently start +/// misreading old keys. +#[must_use] +pub fn split_key(key: &str) -> Option<(&str, &str)> { + let mut parts = key.splitn(3, '.'); + let hive = parts.next()?; + let agent = parts.next()?; + if parts.next().is_some() { + return None; + } + Some((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 `swarm_queue_client::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, +} + +#[cfg(test)] +mod tests { + use super::{AgentStatus, BUCKET, key, split_key}; + + #[test] + fn key_joins_hive_and_agent_with_a_dot() { + assert_eq!(key("prod", "iris"), "prod.iris"); + } + + #[test] + fn the_published_subject_carries_the_hive_as_its_own_token() { + // This is what the hive's write grant `$KV.agent-status..*` + // matches on, so the separator is pinned here rather than left as a + // property of `key`'s formatting string. + assert_eq!( + format!("$KV.{BUCKET}.{}", key("prod", "iris")), + "$KV.agent-status.prod.iris" + ); + // Control: a single-token join lands in the same bucket and would + // pass any assertion that only checked the prefix — it is the hive + // token that the grant needs, and only the `.` produces one. + assert!(!format!("$KV.{BUCKET}.prod/iris").starts_with("$KV.agent-status.prod.")); + } + + #[test] + fn split_key_is_the_inverse_of_key() { + assert_eq!(split_key(&key("prod", "iris")), Some(("prod", "iris"))); + } + + #[test] + fn split_key_rejects_a_key_with_no_dot() { + assert_eq!(split_key("iris"), None); + } + + #[test] + fn split_key_rejects_a_key_with_two_dots() { + assert_eq!(split_key("prod.iris.extra"), None); + } + + #[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-wire/src/lib.rs b/swarm-wire/src/lib.rs index 8f95cf72..c48a7429 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -7,6 +7,7 @@ pub mod agent_inference; pub mod agent_login; +pub mod agent_status; pub mod builder; pub mod claude_login; pub mod review_classifier;