diff --git a/Cargo.lock b/Cargo.lock index a705ad8a..f6252caa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1802,7 +1802,6 @@ dependencies = [ "swarm-matrix-client", "swarm-queue-client", "swarm-secret-client", - "swarm-wire", "tempfile", "tokio", "tokio-stream", @@ -5036,6 +5035,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "strum", "swarm-secret-client", "swarm-wire", "thiserror 2.0.18", diff --git a/frontend/packages/swarm-ui/src/agent/AgentTypes.ts b/frontend/packages/swarm-ui/src/agent/AgentTypes.ts index dc82f7d1..0f78b82a 100644 --- a/frontend/packages/swarm-ui/src/agent/AgentTypes.ts +++ b/frontend/packages/swarm-ui/src/agent/AgentTypes.ts @@ -18,7 +18,7 @@ export interface AgentStatusSnapshot { } // `"up"` / `"offline"` / `"destroyed"`, wire-spelled by -// `swarm_wire::wanted::AgentState::as_str` — kept as a bare +// `swarm_queue_client::wanted::AgentState::as_str` — kept as a bare // `string | null` rather than a union, same reason `config_pr`'s shape // isn't re-derived here: this renders whatever the wire sends, it // doesn't validate the enum client-side. diff --git a/hive-agent/src/schedule_timer.rs b/hive-agent/src/schedule_timer.rs index 5fac9a6d..a6f4f4c2 100644 --- a/hive-agent/src/schedule_timer.rs +++ b/hive-agent/src/schedule_timer.rs @@ -31,8 +31,7 @@ use std::time::Duration; use anyhow::{Context as _, Result}; use rusqlite::{Connection, OptionalExtension as _, params}; -use swarm_queue_client::agent_schedules::{Read, read_own}; -use swarm_wire::agent_schedules::{AgentSchedules, Schedule}; +use swarm_queue_client::agent_schedules::{AgentSchedules, Read, Schedule, read_own}; use tokio::sync::Notify; use crate::todos::Todos; diff --git a/hive-agent/src/swarm_agent_icon.rs b/hive-agent/src/swarm_agent_icon.rs index 3427fe10..a752fda8 100644 --- a/hive-agent/src/swarm_agent_icon.rs +++ b/hive-agent/src/swarm_agent_icon.rs @@ -72,7 +72,7 @@ pub fn spawn() { async fn run(agent: String) { let client = crate::swarm_queue::client().await; - let subject = swarm_wire::agent_icon::subject(&agent); + let subject = swarm_queue_client::agent_icon::subject(&agent); let icon = match read_icon(Path::new(ICON_PATH)) { Ok(icon) => icon, Err(e) => { diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index 0c41e852..845826c1 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -33,9 +33,9 @@ use serde::Serialize; use tokio::sync::broadcast; use tokio::time::{Interval, MissedTickBehavior}; +use swarm_queue_client::wanted::AgentState; use swarm_wire::agent_inference::Effective; use swarm_wire::agent_login::Login; -use swarm_wire::wanted::AgentState; use crate::events::{Bus, BusEvent, LiveEvent, TurnState}; use crate::login::LoginState; @@ -337,8 +337,8 @@ mod tests { use crate::events::{BusEvent, LiveEvent, TurnState}; use crate::login::LoginState; use std::sync::Mutex; + use swarm_queue_client::wanted::AgentState; use swarm_wire::agent_inference::{Effective, Source}; - use swarm_wire::wanted::AgentState; use tokio::sync::broadcast; use tokio::time::{Duration, Instant}; diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 6bbb7cf2..578a267c 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -63,7 +63,6 @@ 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 04c5fb28..9b0bd7df 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_wire::agent_status`'s module doc for why hive status and +//! `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_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_wire::agent_status::key`), so one +//! 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. @@ -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_wire::agent_status::AgentStatus { + let payload = swarm_queue_client::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_wire::agent_status::key(hive, &spec.name); + 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; diff --git a/hive-c0re/src/swarm_notices.rs b/hive-c0re/src/swarm_notices.rs index 97c848df..ac663341 100644 --- a/hive-c0re/src/swarm_notices.rs +++ b/hive-c0re/src/swarm_notices.rs @@ -8,7 +8,7 @@ //! a status snapshot has a current value a late reader can always ask //! for; a lifecycle notice ("container crashed at 04:12") does not — miss //! it and there is nothing left to read later that says it happened. See -//! [`swarm_queue_client::notices`] for opening the stream this publishes into. +//! [`swarm_queue_client::notices`] for the stream this publishes into. //! //! **Best-effort, never fatal to the caller.** A hive with no queue //! coordinates is a silent no-op (the ordinary case). A hive that cannot @@ -129,7 +129,7 @@ async fn publish( .context("serialising the notice")?; let js = async_nats::jetstream::new(client.clone()); - js.publish(swarm_wire::notices::subject(hive), payload.into()) + js.publish(swarm_queue_client::notices::subject(hive), payload.into()) .await .context("publishing the notice")? .await diff --git a/hive-c0re/src/workers/swarm_limits.rs b/hive-c0re/src/workers/swarm_limits.rs index d0a41d7d..a362b356 100644 --- a/hive-c0re/src/workers/swarm_limits.rs +++ b/hive-c0re/src/workers/swarm_limits.rs @@ -1,7 +1,7 @@ //! Reading the caps the swarm sets for this hive's agents. //! //! swarm-controller keeps one key per agent in the `agent-limits` bucket -//! ([`swarm_wire::agent_limits`]). Every [`READ_INTERVAL`] this reads +//! ([`swarm_queue_client::agent_limits`]). Every [`READ_INTERVAL`] this reads //! the key of each agent this hive has and stores what it finds in //! `swarm-resource-limits.json`, where the drop-in reads it. //! A running agent gets a changed cap at its first spawn or rebuild after @@ -14,8 +14,7 @@ use std::collections::BTreeMap; use std::time::Duration; use anyhow::{Context, Result}; -use swarm_queue_client::agent_limits::{self, Read}; -use swarm_wire::agent_limits::AgentCaps; +use swarm_queue_client::agent_limits::{self, AgentCaps, Read}; use crate::resource_limits::AgentLimits; use crate::stats::sweep_health::{self, SweepHealth}; @@ -128,7 +127,7 @@ mod tests { use std::collections::BTreeMap; use super::validated; - use swarm_wire::agent_limits::AgentCaps; + use swarm_queue_client::agent_limits::AgentCaps; /// `atlas`'s answer with `atlas_key` as the value read for it, over the /// hive-wide defaults of 200% / 4G. diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index af7bd5ca..a488c5ee 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -1,7 +1,7 @@ //! Converging this hive onto the agent set the swarm controller declares. //! //! The controller gives each hive its own `hive-wanted-` bucket (see -//! [`swarm_wire::wanted`]); this reads its own and acts on it. It is +//! [`swarm_queue_client::wanted`]); this reads its own and acts on it. It is //! the **repair** path, not the fast one: a deploy event //! ([`crate::swarm_status`]) is core NATS, so a hive that was down never hears //! it, and until this existed that left the hive on its old shape with nothing @@ -33,7 +33,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use anyhow::{Context, Result}; -use swarm_wire::wanted::{AgentState, HiveWanted}; +use swarm_queue_client::wanted::{AgentState, HiveWanted}; use crate::coordinator::Coordinator; use crate::power::Wanted; @@ -469,7 +469,7 @@ mod tests { use super::{Converge, PauseConverge, Plan, carries_a_declaration, decide, decide_pause, plan}; use crate::power::Wanted; - use swarm_wire::wanted::{AgentState, AgentWanted, HiveWanted}; + use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; fn declaration(agents: &[(&str, AgentState)]) -> HiveWanted { HiveWanted { diff --git a/swarm-controller/src/agent_icon.rs b/swarm-controller/src/agent_icon.rs index ee871925..bf4f019d 100644 --- a/swarm-controller/src/agent_icon.rs +++ b/swarm-controller/src/agent_icon.rs @@ -7,7 +7,7 @@ //! property of it, so there is nothing for a timestamp to mean here. //! //! **No hive is named on this path**, because the key does not carry one — -//! see `swarm_wire::agent_icon` for why, and for who writes it. +//! see `swarm_queue_client::agent_icon` for why, and for who writes it. //! //! This reader is also what creates the bucket: the agents that write it are //! granted their own key and nothing that could create a stream. diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index 151cb9a4..9c4bcfa9 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -2,7 +2,7 @@ //! one agent. //! //! Each agent's caps are one value under its own key of `agent-limits` -//! (`swarm_wire::agent_limits`), which this daemon alone writes. The +//! (`swarm_queue_client::agent_limits`), which this daemon alone writes. The //! hive running the agent reads the key; on the hive, a cap replaces the //! hive's own answer for that field, and a field left unset stays with the //! hive. @@ -16,8 +16,7 @@ use axum::extract::{Path, State}; use axum::http::StatusCode; use problem_details::ProblemDetails; use serde::{Deserialize, Serialize}; -use swarm_queue_client::agent_limits; -use swarm_wire::agent_limits::AgentCaps; +use swarm_queue_client::agent_limits::{self, AgentCaps}; use utoipa::ToSchema; use crate::routes::{AppState, error_problem, wanted_error_status}; @@ -359,7 +358,7 @@ pub(crate) mod tests { use axum::extract::{Path, State}; use axum::http::StatusCode; use std::sync::Arc; - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; /// Of the `bad` values, systemd parses `""` (no quota) and `+5%`, and the /// last adds a directive on a second line. It ignores every other one diff --git a/swarm-controller/src/agent_permissions.rs b/swarm-controller/src/agent_permissions.rs index f20ace7d..7aacea6a 100644 --- a/swarm-controller/src/agent_permissions.rs +++ b/swarm-controller/src/agent_permissions.rs @@ -860,7 +860,7 @@ mod tests { State(state.clone()), Path(("pr1ma".to_owned(), "atlas".to_owned())), axum::Json(crate::routes::agents::SetAgentStateRequest { - state: swarm_wire::wanted::AgentState::Destroyed, + state: swarm_queue_client::wanted::AgentState::Destroyed, }), ) .await diff --git a/swarm-controller/src/agent_renewal.rs b/swarm-controller/src/agent_renewal.rs index bb0f6b5f..a3e38bd9 100644 --- a/swarm-controller/src/agent_renewal.rs +++ b/swarm-controller/src/agent_renewal.rs @@ -33,8 +33,8 @@ use std::collections::BTreeSet; use std::sync::Arc; use anyhow::{Context, Result, bail}; +use swarm_queue_client::wanted::{AgentState, HiveWanted}; use swarm_secret_client::{SecretStore, mtls, queue}; -use swarm_wire::wanted::{AgentState, HiveWanted}; use x509_cert::der::DecodePem as _; use crate::wanted::WantedWriter; @@ -429,7 +429,7 @@ pub fn spawn( #[cfg(test)] mod tests { use super::*; - use swarm_wire::wanted::AgentWanted; + use swarm_queue_client::wanted::AgentWanted; const NOW: i64 = 1_790_000_000; const DAY: i64 = 86_400; diff --git a/swarm-controller/src/agent_status.rs b/swarm-controller/src/agent_status.rs index 62bb4416..5d51c336 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_wire::agent_status`'s module doc for why the two buckets +//! `swarm_queue_client::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_wire::agent_status::split_key`), and a +//! its bucket key (`swarm_queue_client::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_wire::agent_status::AgentStatus`] + /// failed to decode as [`swarm_queue_client::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_wire::agent_status::split_key(&key) else { + let Some((hive, agent)) = swarm_queue_client::agent_status::split_key(&key) else { tracing::warn!(%key, "agent-status bucket key doesn't split into hive/agent, skipping"); continue; }; @@ -141,9 +141,10 @@ impl AgentStatusReader { else { continue; }; - let payload = match serde_json::from_slice::( - &entry.value, - ) { + let payload = match serde_json::from_slice::< + swarm_queue_client::agent_status::AgentStatus, + >(&entry.value) + { Ok(status) => serde_json::to_value(status).ok(), Err(e) => { tracing::warn!( diff --git a/swarm-controller/src/agents_md_initial.rs b/swarm-controller/src/agents_md_initial.rs index 54df8a51..235ddc2c 100644 --- a/swarm-controller/src/agents_md_initial.rs +++ b/swarm-controller/src/agents_md_initial.rs @@ -242,7 +242,7 @@ mod tests { State(state.clone()), Path(("pr1ma".to_owned(), "atlas".to_owned())), axum::Json(crate::routes::agents::SetAgentStateRequest { - state: swarm_wire::wanted::AgentState::Destroyed, + state: swarm_queue_client::wanted::AgentState::Destroyed, }), ) .await diff --git a/swarm-controller/src/archetypes.rs b/swarm-controller/src/archetypes.rs index 531ef8ea..e2f31e81 100644 --- a/swarm-controller/src/archetypes.rs +++ b/swarm-controller/src/archetypes.rs @@ -191,7 +191,7 @@ pub(crate) struct ArchetypeView { /// As `PUT /api/agents/{name}/limits` takes them. #[serde(skip_serializing_if = "Option::is_none")] #[schema(value_type = Option)] - limits: Option, + limits: Option, /// As `PUT /api/agents/{name}/inference` takes them. #[serde(skip_serializing_if = "Option::is_none")] #[schema(value_type = Option)] diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 6934b907..54cb4ee6 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -259,8 +259,8 @@ pub(crate) struct WorkerDeps { /// decision the node exists to carry — writing a declaration at all is the /// mechanism, `Paused` is the policy — and a test can pin a name where it /// cannot pin a value written into a KV bucket no unit test has a server for. -const NEW_AGENT_WANTED_STATE: swarm_wire::wanted::AgentState = - swarm_wire::wanted::AgentState::Paused; +const NEW_AGENT_WANTED_STATE: swarm_queue_client::wanted::AgentState = + swarm_queue_client::wanted::AgentState::Paused; /// Run a claimed node's actual work. Mirrors `hive-c0re/src/job_queue/ /// exec.rs::run_node`'s role exactly — the one place a `SwarmNodeKind` @@ -665,7 +665,7 @@ async fn declare_new_agent( agent: &str, ) -> hive_jobq::scheduler::Outcome { use hive_jobq::scheduler::Outcome; - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; let existing = match writer.view(hive).await { Ok(declaration) => declaration.and_then(|d| d.agents.get(agent).map(|a| a.state)), @@ -1645,7 +1645,7 @@ mod tests { fn a_new_agent_is_declared_paused() { assert_eq!( super::NEW_AGENT_WANTED_STATE, - swarm_wire::wanted::AgentState::Paused + swarm_queue_client::wanted::AgentState::Paused ); } diff --git a/swarm-controller/src/routes/agents.rs b/swarm-controller/src/routes/agents.rs index c70b62b0..7414b380 100644 --- a/swarm-controller/src/routes/agents.rs +++ b/swarm-controller/src/routes/agents.rs @@ -68,7 +68,7 @@ pub(crate) struct SetAgentStateRequest { /// a value this build does not know is a 400 rather than a field that /// silently does nothing. #[schema(value_type = String, example = "up")] - pub(crate) state: swarm_wire::wanted::AgentState, + pub(crate) state: swarm_queue_client::wanted::AgentState, } /// One agent's line in a hive's declaration. @@ -78,7 +78,9 @@ pub(crate) struct AgentDeclaration { state: String, } -pub(crate) fn render(declaration: &swarm_wire::wanted::HiveWanted) -> Vec { +pub(crate) fn render( + declaration: &swarm_queue_client::wanted::HiveWanted, +) -> Vec { declaration .agents .iter() @@ -221,8 +223,8 @@ pub(crate) async fn set_agent_state( /// /// Exhaustive for the reason [`revokes_queue_credential`] is. A `Destroyed` /// agent is gone from that hive, and the hive deploys nothing for it. -fn places_agent(state: swarm_wire::wanted::AgentState) -> bool { - use swarm_wire::wanted::AgentState; +fn places_agent(state: swarm_queue_client::wanted::AgentState) -> bool { + use swarm_queue_client::wanted::AgentState; match state { AgentState::Up | AgentState::Offline | AgentState::Paused => true, AgentState::Destroyed => false, @@ -241,7 +243,7 @@ fn places_agent(state: swarm_wire::wanted::AgentState) -> bool { async fn refuse_placement( agent: &str, hive: &str, - here: Option<&swarm_wire::wanted::HiveWanted>, + here: Option<&swarm_queue_client::wanted::HiveWanted>, reserved: Option<&str>, read_roster: impl FnOnce() -> R, snapshot_queued: impl FnOnce() -> Vec<(String, String)>, @@ -251,7 +253,7 @@ where R: std::future::Future>, D: std::future::Future< Output = Result< - Vec<(String, swarm_wire::wanted::HiveWanted)>, + Vec<(String, swarm_queue_client::wanted::HiveWanted)>, problem_details::ProblemDetails, >, >, @@ -293,13 +295,13 @@ where /// Whether declaring an agent into `state` ends its queue credential's life. /// /// Exhaustive on purpose, like every other match on -/// [`AgentState`][swarm_wire::wanted::AgentState] in this +/// [`AgentState`][swarm_queue_client::wanted::AgentState] in this /// tree: a state added later has to say whether it revokes rather than /// inheriting "no" from a catch-all. `Offline` and `Paused` deliberately do /// not — both are states an agent comes back from, and revoking on either /// would mean an agent that could be stopped but never restarted. -fn revokes_queue_credential(state: swarm_wire::wanted::AgentState) -> bool { - use swarm_wire::wanted::AgentState; +fn revokes_queue_credential(state: swarm_queue_client::wanted::AgentState) -> bool { + use swarm_queue_client::wanted::AgentState; match state { AgentState::Destroyed => true, AgentState::Up | AgentState::Offline | AgentState::Paused => false, @@ -423,7 +425,7 @@ pub(crate) async fn get_agents_status( rows.iter().filter_map(|r| r.hive.clone()).collect(); let mut declarations: std::collections::HashMap< String, - swarm_wire::wanted::HiveWanted, + swarm_queue_client::wanted::HiveWanted, > = std::collections::HashMap::new(); for hive in hives { match writer.view(&hive).await { @@ -709,7 +711,7 @@ pub(crate) async fn create_agent( /// The settings a `POST /api/agents` body gives, as they are stored. struct CreateSettings { - limits: Option, + limits: Option, inference: Option, permissions: Option, archetype: Option, @@ -792,7 +794,7 @@ impl StoredSettings { async fn store( client: &async_nats::Client, agent: &str, - limits: Option<&swarm_wire::agent_limits::AgentCaps>, + limits: Option<&swarm_queue_client::agent_limits::AgentCaps>, inference: Option<&swarm_wire::agent_inference::AgentInference>, permissions: Option<&swarm_queue_client::agent_permissions::AgentPermissions>, archetype: Option<&swarm_queue_client::agent_archetype::ArchetypeRef>, @@ -865,7 +867,9 @@ async fn placements_elsewhere( read_declared: impl FnOnce() -> Fut, ) -> Result, E> where - Fut: std::future::Future, E>>, + Fut: std::future::Future< + Output = Result, E>, + >, { let queued = snapshot_queued(); let declared = read_declared().await?; @@ -882,7 +886,8 @@ async fn declarations_elsewhere( state: &AppState, agent: &str, hive: &str, -) -> Result, problem_details::ProblemDetails> { +) -> Result, problem_details::ProblemDetails> +{ read_declarations(state, state.hives.iter().filter(|h| h.name != hive)) .await .map_err(|(other, e)| { @@ -900,7 +905,7 @@ async fn declarations_elsewhere( pub(crate) async fn read_declarations<'a>( state: &AppState, hives: impl Iterator, -) -> Result, (String, anyhow::Error)> { +) -> Result, (String, anyhow::Error)> { let Some(writer) = state.wanted.as_deref() else { return Ok(Vec::new()); }; @@ -918,7 +923,7 @@ pub(crate) async fn read_declarations<'a>( /// Every hive whose declaration places `agent` there. pub(crate) fn claiming_hives( agent: &str, - declared: &[(String, swarm_wire::wanted::HiveWanted)], + declared: &[(String, swarm_queue_client::wanted::HiveWanted)], ) -> Vec { declared .iter() @@ -966,7 +971,7 @@ fn queued_placements( fn placed_elsewhere( agent: &str, hive: &str, - declared: &[(String, swarm_wire::wanted::HiveWanted)], + declared: &[(String, swarm_queue_client::wanted::HiveWanted)], queued: &[(String, String)], ) -> Vec { let declared = declared.iter().filter(|(_, declaration)| { @@ -1160,7 +1165,7 @@ pub(crate) async fn get_agent_icon( [ ( header::CONTENT_TYPE, - swarm_wire::agent_icon::MEDIA_TYPE.to_owned(), + swarm_queue_client::agent_icon::MEDIA_TYPE.to_owned(), ), ( header::CACHE_CONTROL, @@ -1701,12 +1706,13 @@ mod tests { pub(crate) fn declaring( hive: &str, agent: &str, - state: swarm_wire::wanted::AgentState, - ) -> (String, swarm_wire::wanted::HiveWanted) { - let mut declaration = swarm_wire::wanted::HiveWanted::default(); - declaration - .agents - .insert(agent.to_owned(), swarm_wire::wanted::AgentWanted { state }); + state: swarm_queue_client::wanted::AgentState, + ) -> (String, swarm_queue_client::wanted::HiveWanted) { + let mut declaration = swarm_queue_client::wanted::HiveWanted::default(); + declaration.agents.insert( + agent.to_owned(), + swarm_queue_client::wanted::AgentWanted { state }, + ); (hive.to_owned(), declaration) } @@ -1715,7 +1721,7 @@ mod tests { /// terminal and declared. Only the queue-first order catches it. #[tokio::test] async fn a_placement_finishing_between_the_two_reads_is_still_caught() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; let reads = std::cell::Cell::new(0); let snapshot_queued = || { reads.set(reads.get() + 1); @@ -1744,7 +1750,7 @@ mod tests { #[test] fn a_live_declaration_on_another_hive_is_a_placement() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] { let declared = [declaring("sec0nd", "atlas", state)]; assert_eq!( @@ -1761,14 +1767,14 @@ mod tests { let declared = [declaring( "sec0nd", "atlas", - swarm_wire::wanted::AgentState::Destroyed, + swarm_queue_client::wanted::AgentState::Destroyed, )]; assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &[]).is_empty()); } #[test] fn placements_on_the_requested_hive_or_of_other_names_do_not_count() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; let declared = [ declaring("pr1ma", "atlas", AgentState::Up), declaring("sec0nd", "argus", AgentState::Up), @@ -1782,7 +1788,7 @@ mod tests { #[test] fn the_hive_placing_an_agent_claims_it_and_no_other_does() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; let declared = [ declaring("pr1ma", "atlas", AgentState::Paused), declaring("sec0nd", "argus", AgentState::Up), @@ -1794,7 +1800,7 @@ mod tests { #[test] fn two_hives_placing_one_agent_both_claim_it() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; let declared = [ declaring("pr1ma", "atlas", AgentState::Up), declaring("sec0nd", "atlas", AgentState::Up), @@ -1810,7 +1816,7 @@ mod tests { /// state, `pr1ma`'s included. async fn refuse_on_pr1ma_with( agent: &str, - declared: Vec<(String, swarm_wire::wanted::HiveWanted)>, + declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>, roster: Result, ) -> Result<(), problem_details::ProblemDetails> { let here = declared @@ -1832,15 +1838,15 @@ mod tests { /// [`refuse_on_pr1ma_with`] for an `agent` the roster does not hold. async fn refuse_on_pr1ma( agent: &str, - declared: Vec<(String, swarm_wire::wanted::HiveWanted)>, + declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>, ) -> Result<(), problem_details::ProblemDetails> { refuse_on_pr1ma_with(agent, declared, Ok(false)).await } - const PLACING: [swarm_wire::wanted::AgentState; 3] = [ - swarm_wire::wanted::AgentState::Up, - swarm_wire::wanted::AgentState::Offline, - swarm_wire::wanted::AgentState::Paused, + const PLACING: [swarm_queue_client::wanted::AgentState; 3] = [ + swarm_queue_client::wanted::AgentState::Up, + swarm_queue_client::wanted::AgentState::Offline, + swarm_queue_client::wanted::AgentState::Paused, ]; /// The handler checks exactly the placing states; a destroy is never @@ -1851,7 +1857,7 @@ mod tests { assert!(super::places_agent(state), "{state:?}"); } assert!(!super::places_agent( - swarm_wire::wanted::AgentState::Destroyed + swarm_queue_client::wanted::AgentState::Destroyed )); } @@ -1914,7 +1920,7 @@ mod tests { /// first declaration and its name is checked. #[tokio::test] async fn a_destroyed_declaration_here_does_not_exempt_the_name() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; let err = refuse_on_pr1ma( "admin", vec![declaring("pr1ma", "admin", AgentState::Destroyed)], @@ -1947,7 +1953,7 @@ mod tests { #[tokio::test] async fn a_first_declaration_of_a_name_placed_on_another_hive_is_refused() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; let err = refuse_on_pr1ma("atlas", vec![declaring("sec0nd", "atlas", AgentState::Up)]) .await .expect_err("a name placed on another hive must be refused"); @@ -1970,7 +1976,7 @@ mod tests { #[tokio::test] async fn changing_the_state_of_an_agent_on_its_own_hive_is_accepted() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; refuse_on_pr1ma("atlas", vec![declaring("pr1ma", "atlas", AgentState::Up)]) .await .expect("an agent's own hive must accept a new state for it"); @@ -1980,7 +1986,7 @@ mod tests { /// connected and with no identity bridge. async fn set_on_disconnected_pr1ma( agent: &str, - wanted_state: swarm_wire::wanted::AgentState, + wanted_state: swarm_queue_client::wanted::AgentState, ) -> problem_details::ProblemDetails { let (state, _sched) = state_with_two_hives(); let state = crate::routes::AppState { @@ -2004,7 +2010,8 @@ mod tests { /// declaration before it is written, as it refuses creation. #[tokio::test] async fn set_agent_state_refuses_when_the_wanted_state_is_unreadable() { - let err = set_on_disconnected_pr1ma("atlas", swarm_wire::wanted::AgentState::Up).await; + let err = + set_on_disconnected_pr1ma("atlas", swarm_queue_client::wanted::AgentState::Up).await; assert_eq!( err.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), @@ -2022,7 +2029,8 @@ mod tests { #[tokio::test] async fn set_agent_state_does_not_check_a_destroy() { let err = - set_on_disconnected_pr1ma("admin", swarm_wire::wanted::AgentState::Destroyed).await; + set_on_disconnected_pr1ma("admin", swarm_queue_client::wanted::AgentState::Destroyed) + .await; assert_eq!( err.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), @@ -2189,7 +2197,7 @@ mod tests { #[test] fn declaring_an_agent_destroyed_revokes_its_queue_credential() { assert!(super::revokes_queue_credential( - swarm_wire::wanted::AgentState::Destroyed + swarm_queue_client::wanted::AgentState::Destroyed )); } @@ -2199,7 +2207,7 @@ mod tests { /// a state added later shows up here as well as in the match itself. #[test] fn no_state_an_agent_comes_back_from_revokes() { - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] { assert!( !super::revokes_queue_credential(state), @@ -2342,7 +2350,7 @@ mod tests { axum::extract::State(state), axum::extract::Path(("pr1ma".to_owned(), "atlas".to_owned())), axum::Json(SetAgentStateRequest { - state: swarm_wire::wanted::AgentState::Paused, + state: swarm_queue_client::wanted::AgentState::Paused, }), ) .await; @@ -2387,7 +2395,7 @@ mod tests { axum::extract::State(state), axum::extract::Path(("not-a-hive".to_owned(), "atlas".to_owned())), axum::Json(SetAgentStateRequest { - state: swarm_wire::wanted::AgentState::Paused, + state: swarm_queue_client::wanted::AgentState::Paused, }), ) .await; diff --git a/swarm-controller/src/schedules.rs b/swarm-controller/src/schedules.rs index 0d014417..a1299682 100644 --- a/swarm-controller/src/schedules.rs +++ b/swarm-controller/src/schedules.rs @@ -7,7 +7,7 @@ //! schedule exists for all of its agents or for none of them. //! //! Each agent reads only its own key of `agent-schedules` -//! (`swarm_wire::agent_schedules`), which holds the schedules naming +//! (`swarm_queue_client::agent_schedules`), which holds the schedules naming //! it, without their names or other agents. That key is a projection of the //! definitions: after a change this daemon rewrites the key of each agent the //! change touched, and [`spawn_reconcile`] rewrites every agent's key every @@ -30,8 +30,7 @@ use serde::{Deserialize, Serialize}; use swarm_controller_api::schedules::{ AddScheduleRequest, EditScheduleRequest, IntervalEdit, ScheduleView, }; -use swarm_queue_client::agent_schedules; -use swarm_wire::agent_schedules::{AgentSchedules, Schedule}; +use swarm_queue_client::agent_schedules::{self, AgentSchedules, Schedule}; use crate::agent_renewal::unix_now; use crate::routes::{AppState, error_problem}; diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index c33bf4e5..b6a9ffb6 100644 --- a/swarm-controller/src/wanted.rs +++ b/swarm-controller/src/wanted.rs @@ -21,7 +21,7 @@ use anyhow::{Context, Result}; use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind}; -use swarm_wire::wanted::{AgentState, AgentWanted, HiveWanted}; +use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; /// The three outcomes of one write attempt, which the two KV verbs report /// through separate error types. @@ -167,7 +167,7 @@ impl WantedWriter { #[cfg(test)] mod tests { use super::apply; - use swarm_wire::wanted::AgentState; + use swarm_queue_client::wanted::AgentState; #[test] fn declaring_one_agent_preserves_every_other() { diff --git a/swarm-nats-auth/Cargo.toml b/swarm-nats-auth/Cargo.toml index d5347a23..0bf8e78b 100644 --- a/swarm-nats-auth/Cargo.toml +++ b/swarm-nats-auth/Cargo.toml @@ -23,11 +23,12 @@ serde.workspace = true serde_json.workspace = true # The jti digest: base32hex(sha256(claims)) over every JWT this crate signs. sha2.workspace = true -# For `status::BUCKET`: the subjects a hive may publish to are derived from -# the name, and every end that touches it must agree on the same one. -# Deliberately WITHOUT the `kv` feature: this crate derives subject strings, -# it never opens the bucket. -swarm-queue-client.workspace = true +# For `status::BUCKET` and `notices::STREAM` - the subjects a hive may +# publish to are derived from these names, and every end that touches them +# must agree on the same one. Deliberately WITHOUT the `kv` feature: this +# crate derives subject strings, it never opens the bucket. `notices` +# is name-only too (no `jetstream`/`kv` surface), same reason. +swarm-queue-client = { workspace = true, features = ["notices"] } # The store reads: the callout keys, its own OIDC secret, and the agent # credentials `swarm-agent-verify` checks a presented token against. swarm-secret-client.workspace = true diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 31ea3eae..e6550dc5 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -404,7 +404,7 @@ impl Policy { /// reader both call, so a flag here would let two deployments disagree /// about a name they must share. fn wanted_stream(hive: &str) -> String { - format!("KV_{}", swarm_wire::wanted::bucket(hive)) + format!("KV_{}", swarm_queue_client::wanted::bucket(hive)) } /// The stream backing the per-agent status bucket. @@ -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_wire::agent_status::BUCKET) + format!("KV_{}", swarm_queue_client::agent_status::BUCKET) } /// What *any* `JetStream` client must be able to ask before it can do @@ -477,10 +477,10 @@ impl Policy { fn notices_subjects() -> [String; 2] { // Unlike `Self::stream()` above, no `KV_` prefix: the notices // stream is a plain `JetStream` stream, not a KV bucket, so its - // NATS stream name *is* `swarm_wire::notices::STREAM` + // NATS stream name *is* `swarm_queue_client::notices::STREAM` // verbatim — the `KV_` prefix is `create_key_value`'s own // convention, not something every stream carries. - let stream = swarm_wire::notices::STREAM; + let stream = swarm_queue_client::notices::STREAM; [ format!("$JS.API.STREAM.INFO.{stream}"), format!("$JS.API.STREAM.CREATE.{stream}"), @@ -514,7 +514,7 @@ impl Policy { // Named from the defining crate so a rename cannot leave the grant // pointing at a subject nobody publishes to. Not `extra_hive_subjects`: // that is for streams this crate does not know about. - subjects.push(swarm_wire::notices::subject(hive)); + subjects.push(swarm_queue_client::notices::subject(hive)); // Reading the wanted-state bucket, which is **this hive's own**. // // One bucket per hive rather than one keyed by hive, so that a *watch* @@ -530,7 +530,7 @@ impl Policy { format!("$JS.API.STREAM.INFO.{wanted}"), format!( "$JS.API.DIRECT.GET.{wanted}.$KV.{}.{hive}", - swarm_wire::wanted::bucket(hive) + swarm_queue_client::wanted::bucket(hive) ), // The watch the paragraph above buys: a KV watch is a consumer, so // `DIRECT.GET` covers the boot-time read and nothing after it. @@ -563,10 +563,10 @@ impl Policy { subjects.extend([ format!("$JS.API.STREAM.INFO.{agent_status}"), format!("$JS.API.STREAM.CREATE.{agent_status}"), - format!("$KV.{}.{hive}.*", swarm_wire::agent_status::BUCKET), + format!("$KV.{}.{hive}.*", swarm_queue_client::agent_status::BUCKET), ]); // Reading the caps the swarm sets for the agents this hive runs - // (`swarm_wire::agent_limits`): a direct get, so read-only, and + // (`swarm_queue_client::agent_limits`): a direct get, so read-only, and // no `STREAM.INFO` because the reader skips the lookup. // // Every agent's key, not just this hive's: the bucket is keyed by agent @@ -574,7 +574,7 @@ impl Policy { // `hive_name`'s doc argues against. subjects.push(format!( "$JS.API.DIRECT.GET.{}.>", - swarm_wire::agent_limits::STREAM + swarm_queue_client::agent_limits::STREAM )); // The permissions the swarm grants those agents // (`swarm_queue_client::agent_permissions`), read the same way 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_wire::agent_status::BUCKET; + let bucket = swarm_queue_client::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_wire::agent_status::BUCKET; + let bucket = swarm_queue_client::agent_status::BUCKET; let subjects = policy().hive_subjects("alpha"); let s = Policy::agent_status_stream(); for want in [ @@ -847,7 +847,7 @@ mod tests { /// grant reaches no other bucket. #[test] fn a_hive_may_read_every_agents_caps_and_write_none() { - use swarm_wire::agent_limits::{STREAM, direct_get_subject, key_subject}; + use swarm_queue_client::agent_limits::{STREAM, direct_get_subject, key_subject}; let p = policy() .permissions("hive-alpha") @@ -881,7 +881,7 @@ mod tests { // The read reaches only this bucket. for other in [ swarm_wire::agent_inference::direct_get_subject("atlas"), - swarm_wire::agent_schedules::direct_get_subject("atlas"), + swarm_queue_client::agent_schedules::direct_get_subject("atlas"), "$JS.API.DIRECT.GET.KV_hive-wanted-beta.$KV.hive-wanted-beta.beta".to_owned(), ] { assert!(!may(&other), "hive alpha may publish {other}"); @@ -1964,7 +1964,7 @@ mod tests { /// A verified agent may write its own icon key and no other agent's. #[test] fn an_agent_may_write_only_its_own_icon_key() { - use swarm_wire::agent_icon::subject; + use swarm_queue_client::agent_icon::subject; let p = policy() .with_agent_token_subjects(vec!["$KV.agent-icons.{agent}".to_owned()]) @@ -2012,7 +2012,7 @@ mod tests { /// and no way to open, watch or write the bucket. #[test] fn an_agent_may_read_only_its_own_schedule_key() { - use swarm_wire::agent_schedules::{STREAM, direct_get_subject, key_subject}; + use swarm_queue_client::agent_schedules::{STREAM, direct_get_subject, key_subject}; let p = policy() .with_agent_token_subjects(vec![ @@ -2078,7 +2078,7 @@ mod tests { format!("$JS.API.STREAM.INFO.{STREAM}"), format!("$JS.API.CONSUMER.CREATE.{STREAM}"), key_subject("atlas"), - swarm_wire::agent_schedules::direct_get_subject("atlas"), + swarm_queue_client::agent_schedules::direct_get_subject("atlas"), ] { assert!( !atlas.publish.iter().any(|g| covers(g, &refused)), diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 75685c95..7c8de72c 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -81,6 +81,7 @@ hive-types = { workspace = true, optional = true } reqwest = { workspace = true, features = ["blocking"] } serde.workspace = true serde_json.workspace = true +strum.workspace = true # `stored_secret::StoredSecret`, the held client secret, and the rustls # provider `connect` installs. swarm-secret-client.workspace = true diff --git a/swarm-queue-client/src/agent_archetype.rs b/swarm-queue-client/src/agent_archetype.rs index 51563b43..9c0aa739 100644 --- a/swarm-queue-client/src/agent_archetype.rs +++ b/swarm-queue-client/src/agent_archetype.rs @@ -9,9 +9,9 @@ //! **The hive running the agent reads the key** when it provisions or //! rebuilds the agent, and renders the archetype's module into its meta flake //! in place of a config repo. Its grant covers every key of this bucket, -//! read-only, for the reason `swarm_wire::agent_limits` gives. +//! read-only, for the reason [`crate::agent_limits`] gives. //! -//! Keyed by the agent alone, like `swarm_wire::agent_limits`: an agent can move +//! Keyed by the agent alone, like [`crate::agent_limits`]: an agent can move //! between hives, and its archetype goes with it. /// The KV bucket archetype choices are published into, keyed by agent name. diff --git a/swarm-queue-client/src/agent_icon.rs b/swarm-queue-client/src/agent_icon.rs index 98a65de8..1805b03d 100644 --- a/swarm-queue-client/src/agent_icon.rs +++ b/swarm-queue-client/src/agent_icon.rs @@ -1,5 +1,56 @@ -//! The connection side of the per-agent icon KV bucket: opening it. The -//! bucket's name, media type and subject are `swarm_wire::agent_icon`. +//! The per-agent icon KV bucket: one key per agent, holding the icon the +//! swarm shows for it. +//! +//! **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 +//! 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. +//! Operator ruling: *"the bucket is per agent, not per hive. agents can +//! move hives."* +//! +//! That is also what makes the icon answerable for an agent whose +//! container is stopped: the value's lifetime is the agent's, not its +//! placement's. +//! +//! **Each agent writes its own key** with its own queue credential. The auth +//! callout grants that credential `$KV.agent-icons.` and no other key. +//! +//! The writer publishes to [`crate::agent_icon::subject`] directly instead of +//! opening the bucket, so that one subject is its whole grant. Creating the +//! bucket is left to the reader, with `open_or_create`. A write that arrives +//! before the bucket exists is refused with "no stream", and the writer +//! retries. + +#[cfg(feature = "kv")] +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 +/// same bucket, and an option is a way for the two to disagree. +pub const BUCKET: &str = "agent-icons"; + +/// The media type of every value in this bucket. +/// +/// The value is the icon's bytes **verbatim, not a JSON envelope**: an SVG +/// carries no metadata this bucket would have to describe, and the one +/// consumer serves the bytes straight back out. So the type is fixed here +/// rather than stored per entry — a publisher that has something other +/// than an SVG does not have an agent icon. +pub const MEDIA_TYPE: &str = "image/svg+xml"; + +/// The subject `agent`'s entry is written on: `$KV..`, where the +/// key is the agent name and nothing else. +/// +/// A KV put is a `JetStream` publish to this subject, so a writer that +/// publishes directly has to use the same subject the bucket's `get` reads. +#[must_use] +pub fn subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} /// Open the agent-icon bucket, creating it if nothing has yet. /// @@ -9,13 +60,26 @@ #[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, - swarm_wire::agent_icon::BUCKET, + BUCKET, "agent-icon", "Current icon published by each agent".to_owned(), ) .await } + +#[cfg(test)] +mod tests { + use super::subject; + + /// One token after the bucket, the agent name alone. That is what the + /// per-agent grant `$KV.agent-icons.{agent}` expands to, and what a + /// hive-scoped grant (`$KV...*`, two tokens) cannot match. + #[test] + fn the_published_subject_carries_the_agent_and_no_placement() { + assert_eq!(subject("iris"), "$KV.agent-icons.iris"); + } +} diff --git a/swarm-queue-client/src/agent_limits.rs b/swarm-queue-client/src/agent_limits.rs index 25b78097..77cf3b7f 100644 --- a/swarm-queue-client/src/agent_limits.rs +++ b/swarm-queue-client/src/agent_limits.rs @@ -1,6 +1,58 @@ -//! The connection side of the per-agent resource-cap KV bucket: opening it for -//! writing or reading, and the hive's direct read. The bucket's names, subjects -//! and stored type are `swarm_wire::agent_limits`. +//! The per-agent resource-cap KV bucket: one key per agent, holding the CPU +//! and memory caps the swarm sets for it. +//! +//! **swarm-controller is the only writer**, and creates the bucket on its +//! first write. Every field is optional: an absent field leaves the agent on +//! its hive's own answer, so an agent with no key runs exactly as it would +//! with no swarm. +//! +//! **The hive running the agent reads the key**, not the agent: the caps go +//! into the container's systemd drop-in, which the host writes. A hive reads +//! with one `DIRECT.GET` per agent +//! ([`crate::agent_limits::direct_get_subject`]), and its grant covers every +//! key of this bucket, read-only. Scoping it to the hive's own agents would +//! need a roster in the auth responder. +//! +//! Keyed by the agent alone, like [`crate::agent_inference`]: an agent can +//! move between hives, and its caps go with it. + +/// The KV bucket caps are published into, keyed by agent name. +pub const BUCKET: &str = "agent-limits"; + +/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. +pub const STREAM: &str = "KV_agent-limits"; + +/// The subject `agent`'s entry is stored under. +#[must_use] +pub fn key_subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} + +/// The request subject that reads `agent`'s current entry. +#[must_use] +pub fn direct_get_subject(agent: &str) -> String { + format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent)) +} + +/// One agent's caps: the value under `BUCKET`/``, as systemd values +/// carried verbatim. `None` is "not capped by the swarm", never "uncapped". +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentCaps { + /// systemd `CPUQuota=`, e.g. `"400%"`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cpu_quota: Option, + /// systemd `MemoryMax=`, e.g. `"8G"`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memory_max: Option, +} + +impl AgentCaps { + /// True when no field is set, which a hive reads the same as no key. + #[must_use] + pub fn is_empty(&self) -> bool { + self.cpu_quota.is_none() && self.memory_max.is_none() + } +} /// Open the caps bucket for writing, creating it if nothing has yet. /// @@ -12,7 +64,7 @@ pub async fn open_or_create( let js = async_nats::jetstream::new(client.clone()); crate::kv::open_or_create( &js, - swarm_wire::agent_limits::BUCKET, + BUCKET, "agent-limits", "Resource caps the swarm controller sets for each agent".to_owned(), ) @@ -33,11 +85,11 @@ pub async fn open( client: &async_nats::Client, ) -> Result, crate::Error> { let js = async_nats::jetstream::new(client.clone()); - match js.get_key_value(swarm_wire::agent_limits::BUCKET).await { + match js.get_key_value(BUCKET).await { Ok(store) => Ok(Some(store)), Err(e) if crate::wanted::is_bucket_not_found(&e) => Ok(None), Err(source) => Err(crate::Error::OpenBucket { - bucket: swarm_wire::agent_limits::BUCKET.to_owned(), + bucket: BUCKET.to_owned(), source, }), } @@ -48,7 +100,7 @@ pub async fn open( #[derive(Debug)] pub enum Read { /// The bucket answered; an absent or deleted key has every field unset. - Current(swarm_wire::agent_limits::AgentCaps), + Current(AgentCaps), /// Nothing answers for the bucket's stream: no cap has been written yet, /// or `JetStream` is not serving it right now. Not an unset value, since /// the second case says nothing about the agent's caps. @@ -67,7 +119,6 @@ pub async fn read( agent: &str, ) -> Result> { use async_nats::jetstream::stream::DirectGetErrorKind; - use swarm_wire::agent_limits::{AgentCaps, STREAM, key_subject}; let js = async_nats::jetstream::new(client.clone()); let stream = js.get_stream_no_info(STREAM).await?; @@ -98,3 +149,42 @@ fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool { if status == async_nats::StatusCode::NO_RESPONDERS ) } + +#[cfg(test)] +mod tests { + use super::*; + + /// The names the hive's grant in `swarm-nats-auth` is built from. + #[test] + fn names_match_the_queue_grant() { + assert_eq!(STREAM, format!("KV_{BUCKET}")); + assert_eq!(key_subject("iris"), "$KV.agent-limits.iris"); + assert_eq!( + direct_get_subject("iris"), + "$JS.API.DIRECT.GET.KV_agent-limits.$KV.agent-limits.iris" + ); + } + + /// An unset field is left out on the wire, and `{}` is every field unset. + #[test] + fn unset_fields_are_absent() { + let value = AgentCaps { + memory_max: Some("8G".to_owned()), + ..AgentCaps::default() + }; + assert_eq!( + serde_json::to_string(&value).unwrap(), + r#"{"memory_max":"8G"}"# + ); + let empty: AgentCaps = serde_json::from_str("{}").unwrap(); + assert!(empty.is_empty()); + } + + /// Fields a newer controller adds must not cost an older hive its caps. + #[test] + fn decoding_tolerates_unknown_fields() { + let decoded: AgentCaps = + serde_json::from_str(r#"{"cpu_quota":"400%","io_weight":"10"}"#).unwrap(); + assert_eq!(decoded.cpu_quota.as_deref(), Some("400%")); + } +} diff --git a/swarm-queue-client/src/agent_permissions.rs b/swarm-queue-client/src/agent_permissions.rs index d0e86dcd..306fe6df 100644 --- a/swarm-queue-client/src/agent_permissions.rs +++ b/swarm-queue-client/src/agent_permissions.rs @@ -9,12 +9,12 @@ //! **Two readers, each enforcing its own part.** The hive running the agent //! applies `manage_root_agent` as bind mounts at container start; its grant //! covers every key of this bucket, read-only, for the reason -//! `swarm_wire::agent_limits` gives. The agent itself (its harness and subagent +//! [`crate::agent_limits`] gives. The agent itself (its harness and subagent //! daemon) applies the tool permissions; its grant covers only its own key. //! Both read with one `DIRECT.GET` on the agent's key //! ([`crate::agent_permissions::key_subject`]). //! -//! Keyed by the agent alone, like `swarm_wire::agent_limits`: an agent can move +//! Keyed by the agent alone, like [`crate::agent_limits`]: an agent can move //! between hives, and its permissions go with it. //! //! Also carries the one-shot import a hive sends of the grants its meta repo diff --git a/swarm-queue-client/src/agent_schedules.rs b/swarm-queue-client/src/agent_schedules.rs index f55d9e7b..13537e8e 100644 --- a/swarm-queue-client/src/agent_schedules.rs +++ b/swarm-queue-client/src/agent_schedules.rs @@ -1,6 +1,93 @@ -//! The connection side of the per-agent schedule KV bucket: opening it for -//! writing and the agent's own direct read. The bucket's names, subjects and -//! stored types are `swarm_wire::agent_schedules`. +//! The per-agent schedule KV bucket: one key per agent, holding the scheduled +//! prompts that fire in that agent. +//! +//! **swarm-controller is the only writer**, and creates the bucket on its +//! first write. It keeps each prompt once, with its name and every agent it +//! goes to, and writes each agent's key as the prompts naming that agent, +//! without the name or the other agents. Each agent's harness reads its own +//! key and fires the schedules itself; the fire times are a pure function of +//! each schedule ([`crate::agent_schedules::Schedule::latest_slot`]), so +//! rewriting the list never moves them. +//! +//! **An agent reads with one `DIRECT.GET` on its own key, not a watch.** A KV +//! watch is an unnamed consumer, whose create subject carries no filter, so +//! granting it would grant every agent's key. The direct get carries the key +//! in its subject ([`crate::agent_schedules::direct_get_subject`]), which is +//! what lets the queue grant scope the read to one agent. It is also the +//! agent's whole `JetStream` grant: `read_own` builds the stream handle +//! without a `STREAM.INFO`. +//! +//! Keyed by the agent alone, like [`crate::agent_icon`], and for the same +//! reason: an agent can move between hives. + +/// The KV bucket schedules are published into, keyed by agent name. +pub const BUCKET: &str = "agent-schedules"; + +/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. +pub const STREAM: &str = "KV_agent-schedules"; + +/// The subject `agent`'s entry is stored under. +#[must_use] +pub fn key_subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} + +/// The request subject that reads `agent`'s current entry, and so the one +/// subject the queue grants that agent in this bucket. +#[must_use] +pub fn direct_get_subject(agent: &str) -> String { + format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent)) +} + +/// The prompts that fire in one agent: the value under `BUCKET`/``. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentSchedules { + #[serde(default)] + pub schedules: Vec, +} + +/// One scheduled prompt. +/// +/// Its fire times are `anchor_unix + k * interval_seconds` for `k >= 0`, or +/// `anchor_unix` alone when there is no interval. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct Schedule { + /// Minted by the controller, the same in every agent the prompt names, + /// and never reused, so an agent's record of what it last fired for an + /// id cannot apply to a different schedule. + pub id: String, + /// The prompt, delivered verbatim as the todo's summary. + pub body: String, + /// The first fire time, unix seconds. + pub anchor_unix: i64, + /// Seconds between fires; `None` is a one-shot. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub interval_seconds: Option, + /// A paused schedule does not fire, and the slots that pass while it is + /// paused are not fired when it resumes. + #[serde(default)] + pub paused: bool, +} + +impl Schedule { + /// The latest fire time at or before `now`, or `None` while the first is + /// still ahead. + #[must_use] + pub fn latest_slot(&self, now: i64) -> Option { + if now < self.anchor_unix { + return None; + } + let Some(interval) = self + .interval_seconds + .filter(|i| *i > 0) + .and_then(|i| i64::try_from(i).ok()) + else { + return Some(self.anchor_unix); + }; + let elapsed = now - self.anchor_unix; + Some(self.anchor_unix + elapsed - elapsed % interval) + } +} /// Open the schedule bucket for writing, creating it if nothing has yet. /// @@ -12,7 +99,7 @@ pub async fn open_or_create( let js = async_nats::jetstream::new(client.clone()); crate::kv::open_or_create( &js, - swarm_wire::agent_schedules::BUCKET, + BUCKET, "agent-schedule", "Scheduled prompts the swarm controller declares for each agent".to_owned(), ) @@ -24,7 +111,7 @@ pub async fn open_or_create( #[derive(Debug)] pub enum Read { /// The bucket answered; an absent or deleted key is an empty list. - Current(swarm_wire::agent_schedules::AgentSchedules), + Current(AgentSchedules), /// Nothing answers for the bucket's stream: no schedule has been written /// yet, or `JetStream` is not serving it right now. Not an empty list, /// since the second case says nothing about the agent's schedules. @@ -43,7 +130,6 @@ pub async fn read_own( agent: &str, ) -> Result> { use async_nats::jetstream::stream::DirectGetErrorKind; - use swarm_wire::agent_schedules::{AgentSchedules, STREAM, key_subject}; let js = async_nats::jetstream::new(client.clone()); let stream = js.get_stream_no_info(STREAM).await?; @@ -74,3 +160,60 @@ fn is_no_responders(e: &async_nats::jetstream::stream::DirectGetError) -> bool { if status == async_nats::StatusCode::NO_RESPONDERS ) } + +#[cfg(test)] +mod tests { + use super::*; + + fn every(anchor_unix: i64, interval: Option) -> Schedule { + Schedule { + id: "abc".to_owned(), + body: "hi".to_owned(), + anchor_unix, + interval_seconds: interval, + paused: false, + } + } + + /// The names the grant in `swarm-nats.nix` spells with `{agent}`. + #[test] + fn names_match_the_queue_grant() { + assert_eq!(STREAM, format!("KV_{BUCKET}")); + assert_eq!(key_subject("iris"), "$KV.agent-schedules.iris"); + assert_eq!( + direct_get_subject("iris"), + "$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.iris" + ); + } + + #[test] + fn slots_sit_on_the_anchor_grid() { + let s = every(1000, Some(60)); + assert_eq!(s.latest_slot(999), None); + assert_eq!(s.latest_slot(1000), Some(1000)); + assert_eq!(s.latest_slot(1059), Some(1000)); + assert_eq!(s.latest_slot(1060), Some(1060)); + assert_eq!(s.latest_slot(1000 + 60 * 7 + 5), Some(1000 + 60 * 7)); + } + + #[test] + fn a_one_shot_has_one_slot() { + let s = every(1000, None); + assert_eq!(s.latest_slot(999), None); + assert_eq!(s.latest_slot(1000), Some(1000)); + assert_eq!(s.latest_slot(1_000_000), Some(1000)); + } + + /// Fields a newer controller adds must not cost an older agent its list, + /// and a list written without the optional fields decodes to a recurring + /// or one-shot schedule that is not paused. + #[test] + fn decoding_tolerates_missing_and_unknown_fields() { + let doc = r#"{"schedules":[{"id":"a","body":"b","anchor_unix":5,"note":"x"}],"rev":2}"#; + let decoded: AgentSchedules = serde_json::from_str(doc).expect("decodes"); + let mut want = every(5, None); + "a".clone_into(&mut want.id); + "b".clone_into(&mut want.body); + assert_eq!(decoded.schedules, vec![want]); + } +} diff --git a/swarm-queue-client/src/agent_status.rs b/swarm-queue-client/src/agent_status.rs index 8a8696ec..52ba59ff 100644 --- a/swarm-queue-client/src/agent_status.rs +++ b/swarm-queue-client/src/agent_status.rs @@ -1,6 +1,103 @@ -//! 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`. +//! 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, +} /// Open the agent-status bucket, creating it if nothing has yet. /// @@ -10,13 +107,74 @@ #[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, - swarm_wire::agent_status::BUCKET, + 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/archetypes.rs b/swarm-queue-client/src/archetypes.rs index 8df55357..48e09abe 100644 --- a/swarm-queue-client/src/archetypes.rs +++ b/swarm-queue-client/src/archetypes.rs @@ -11,8 +11,8 @@ use std::collections::BTreeMap; +use crate::agent_limits::AgentCaps; use swarm_wire::agent_inference::AgentInference; -use swarm_wire::agent_limits::AgentCaps; /// The KV bucket archetype sources are published into, keyed by source name. pub const BUCKET: &str = "agent-archetypes"; diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 63efa26f..f7e782cf 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -108,7 +108,7 @@ pub enum Error { #[error("creating the {bucket} bucket")] CreateBucket { // `String` rather than `&'static str`: a per-hive bucket name is built - // at runtime (`swarm_wire::wanted::bucket`), so the error has to own it. + // at runtime (`wanted::bucket`), so the error has to own it. bucket: String, #[source] source: async_nats::jetstream::context::CreateKeyValueError, @@ -188,21 +188,21 @@ pub fn chain(error: &dyn std::error::Error) -> String { /// which is the disagreement this module exists to prevent. pub mod status; -/// Opening and watching the bucket the controller declares each hive's wanted -/// agent set in — the opposite direction from [`status`], and deliberately not a -/// mirror of it. The bucket names and stored types, and the durability and -/// "absence is not a deletion order" rules, are in `swarm_wire::wanted`. +/// The bucket the controller declares each hive's wanted agent set in — the +/// opposite direction from [`status`], and deliberately not a mirror of it. +/// Status is *observed* and republishes itself; this is *declared* and nothing +/// recreates it, which is why the module documents its own durability and its +/// "absence is not a deletion order" rule rather than inheriting either. pub mod wanted; -/// 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`. +/// 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; -/// Opening the bucket agent icons are published into: one key per agent, with -/// no hive in it, unlike [`agent_status`]. The name, media type and subject, -/// and why the placement stays out of the key, are in `swarm_wire::agent_icon`. +/// The bucket agent icons are published into — one key per agent, with no +/// hive in it, unlike [`agent_status`]. See the module doc for why the +/// placement stays out of the key, and for who may write it. pub mod agent_icon; /// The bucket each agent mirrors its `AGENTS.md` into, and the edit command @@ -215,10 +215,9 @@ pub mod agents_md; /// `agents-md-initial`. pub mod agents_md_initial; -/// Opening and reading the bucket swarm-controller declares each agent's -/// scheduled prompts in — one key per agent, read by that agent alone. The -/// names and stored types are `swarm_wire::agent_schedules`; writing needs `kv` -/// and reading needs `agent-schedules`. +/// The bucket swarm-controller declares each agent's scheduled prompts in — +/// one key per agent, read by that agent alone. The names are unconditional; +/// writing needs `kv` and reading needs `agent-schedules`. pub mod agent_schedules; /// The bucket swarm-controller sets each agent's model, effort and subagent @@ -226,10 +225,9 @@ pub mod agent_schedules; /// unconditional; writing needs `kv` and reading needs `agent-inference`. pub mod agent_inference; -/// Opening and reading the bucket swarm-controller sets each agent's CPU and -/// memory caps in — one key per agent, read by the hive that runs it. The -/// names and the stored type are `swarm_wire::agent_limits`; writing needs `kv` -/// and reading needs `agent-limits`. +/// The bucket swarm-controller sets each agent's CPU and memory caps in — one +/// key per agent, read by the hive that runs it. The names are unconditional; +/// writing needs `kv` and reading needs `agent-limits`. pub mod agent_limits; /// The bucket swarm-controller sets each agent's permissions in — one key @@ -469,9 +467,9 @@ pub struct AgentInput { pub body: String, } -/// Opening the hive-notices stream. The name and subject are -/// `swarm_wire::notices`. Behind the `notices` feature, same reason `status` -/// is behind `kv`. +/// The hive-notices stream, shared by the hive that publishes and +/// whatever eventually consumes it. Behind the `notices` feature, same +/// reason `status` is behind `kv` — see the module doc. #[cfg(feature = "notices")] pub mod notices; diff --git a/swarm-queue-client/src/notices.rs b/swarm-queue-client/src/notices.rs index 98a33fcb..ca1a8b35 100644 --- a/swarm-queue-client/src/notices.rs +++ b/swarm-queue-client/src/notices.rs @@ -1,12 +1,54 @@ -//! Opening the hive-notices stream a hive publishes lifecycle notices into. The -//! stream name, subject shape and why it is a stream rather than a bucket are in -//! `swarm_wire::notices`. +//! The hive-notices stream: its name, subject shape, and how a hive +//! opens it to publish. //! -//! Feature-gated (`notices`) so the crate's other consumers do not compile +//! Same reason [`crate::status`] exists rather than a bare `const` on +//! whichever side happens to need one first: a hive that publishes and a +//! swarm-level reader that eventually consumes live in different crates, +//! and a literal name repeated across both is an agreement nothing +//! checks. +//! +//! **This is a stream, not a bucket, and that is a real design choice — +//! not the same shape as [`crate::status`] wearing a different name.** +//! [`crate::status`]'s hive-status snapshot has a current value: a +//! reconnecting reader can always ask "what does this hive say *now*" +//! and get the true answer, so a KV bucket (last value per key) is the +//! right shape. A lifecycle notice ("container crashed at 04:12") has no +//! such steady state — miss the message and there is nothing left to +//! read later that would tell you it happened. That needs durable +//! delivery (a JetStream stream a consumer acks against), which is what +//! this module opens instead. +//! +//! Feature-gated (`notices`) for the same reason [`crate::status`] is +//! gated behind `kv`: the crate's other consumers (the auth-callout +//! responder, a hive that only publishes status) should not compile //! against a stream shape they never touch. use crate::Error; -use swarm_wire::notices::STREAM; + +/// The stream a hive publishes lifecycle notices into. +/// +/// A constant and not an option, matching [`crate::status::BUCKET`]: +/// reader and writer must name the same stream, and letting either side +/// pick its own name is how two deployments end up disagreeing about +/// which stream a notice actually landed in. +pub const STREAM: &str = "hive-notices"; + +/// Build the subject a given hive's notices publish to. +/// +/// Every hive's notices land under `STREAM`, one subject per hive: +/// `hive-notices.`. +/// +/// Not one subject per notice *kind* — a consumer that wants a specific +/// hive's notices subscribes to `subject(hive)`; one that wants the whole +/// swarm's subscribes to `{STREAM}.>`. The kind travels inside the message +/// payload instead, so adding a new notice kind is never a subject-design +/// change. Deliberately reuses `STREAM` rather than a second `PREFIX` +/// constant with the same value — one name for one fact, same reasoning +/// the module doc above gives for a shared `const` over a repeated literal. +#[must_use] +pub fn subject(hive: &str) -> String { + format!("{STREAM}.{hive}") +} /// Open the notices stream, creating it if nothing has yet. /// diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index 597021e4..16681441 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -1,18 +1,116 @@ -//! The connection side of the hive-wanted KV buckets: opening and watching a -//! hive's bucket. The bucket names and the stored types, and why absence is -//! not a deletion order, are `swarm_wire::wanted`. +//! The hive-wanted KV buckets: the agent set the controller declares for each +//! hive, **one bucket per hive** (see [`crate::wanted::bucket`] for why the +//! split is a grant boundary rather than a data-modelling choice), keyed inside +//! it by `hiveName`. +//! +//! Sibling of [`crate::status`] and deliberately **not** a mirror of it: that +//! one is **observed** — each hive republishes what it is, so its store losing +//! everything "degrades to honesty" (`swarm-nats.nix`). This one is +//! **declared**, and nothing regenerates it; lose it and a reconcile loop has +//! nothing to converge to, which is the failure such a loop exists to remove. +//! +//! # Absence is not a deletion order +//! +//! A hive that finds no key for itself — or no bucket at all — has learned +//! nothing about the agents it is running, not that it should have none. +//! Swarm-side lifecycle does not yet cover agents that predate it, so callers +//! converge the agents a value NAMES and leave the rest alone. //! //! # Only the controller creates it //! //! Unlike [`crate::status::open_or_create`], where either end may legitimately //! arrive first, the writer here is single and known. A hive opens read-only -//! and treats absence as "not published yet", so that stays quiet rather than -//! looking like an error a hive could fix. +//! and treats absence as the case above, so "not published yet" stays quiet +//! rather than looking like an error a hive could fix. #[cfg(feature = "kv")] use crate::Error; -#[cfg(feature = "kv")] -use swarm_wire::wanted::bucket; + +/// The prefix every hive's wanted-state bucket name starts with. +/// +/// Derived from, and not configurable, for the reason [`crate::status::BUCKET`] +/// gives: writer and reader must name the same bucket, and an option is a way +/// for two deployments to disagree about which one that is. +pub const BUCKET_PREFIX: &str = "hive-wanted-"; + +/// The KV bucket one hive's declaration lives in — **one bucket per hive**, not +/// one bucket keyed by hive. +/// +/// The split is a grant-scoping decision, not a data-modelling one. A KV read +/// scopes per key, because `DIRECT.GET` carries the key in the subject; a +/// *watch* does not, because a consumer's filter travels in the request payload +/// and `$JS.API.CONSUMER.CREATE.` therefore grants the whole stream. So +/// with every hive in one bucket, letting a hive watch its own declaration +/// means letting it read every other hive's. One stream per hive makes the +/// grant a hive can hold exactly as wide as what it is allowed to see. +/// +/// A bucket name may contain `[a-zA-Z0-9_-]`, and a hive name is an `Ident` +/// (`[a-z0-9-]`), so this composition is always a legal bucket name — plain +/// code span rather than an intra-doc link because this crate does not depend +/// on `hive_types`. +#[must_use] +pub fn bucket(hive: &str) -> String { + format!("{BUCKET_PREFIX}{hive}") +} + +/// One hive's whole declaration: the agents the controller names, and what it +/// wants of each. The value under `BUCKET`/``. +/// +/// The agents it does **not** name are not part of the declaration at all — +/// see the module docs. A hive converges the ones listed here and leaves the +/// rest alone. +#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct HiveWanted { + /// Agent name → what the controller wants of it. + #[serde(default)] + pub agents: std::collections::BTreeMap, +} + +/// One agent's entry in a [`HiveWanted`]. +/// +/// A struct around one field rather than the bare state, because this is where +/// a later declaration grows — a config revision, a schedule — and a hive +/// built before that field existed keeps deserialising the whole document. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentWanted { + /// The state to converge this agent to. + pub state: AgentState, +} + +/// What the controller wants an agent to be. +/// +/// **Closed on purpose.** A value this build does not know fails the whole +/// document's decode, so a hive running older code converges **nothing** rather +/// than part of a declaration it only half understands. Adding a state means +/// adding a variant here and shipping it to both ends — which is the intended +/// workflow, not an obstacle to route around with a catch-all variant. +#[derive( + Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, strum::IntoStaticStr, +)] +#[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] +pub enum AgentState { + /// Exists on the hive and is running. + Up, + /// Exists on the hive and is not running. + Offline, + /// Exists on the hive, is running, and its turn loop is parked — the + /// swarm-declared counterpart of the hive-local `hivectl agent pause` + /// marker. Orthogonal to the power axis `Up`/`Offline` express: a + /// paused agent still has a running container (the web UI and MCP + /// daemons stay reachable), it just drives no turns. See + /// `hive-c0re::workers::wanted::decide_pause` for the convergence side + /// of that split. + Paused, + /// Torn down entirely — the hive runs the destroy template once, then + /// treats absence as agreement rather than re-running it. The key + /// stays in the declared set with this state forever, on purpose: it's + /// what lets a redeclare of an already-destroyed agent be rejected as + /// a terminal-state transition instead of silently reviving it. + /// Irreversible: there is no state that brings a destroyed agent back + /// short of a fresh deploy. + Destroyed, +} /// Open the wanted-state bucket for writing, creating it if nothing has yet. /// @@ -35,7 +133,7 @@ pub async fn open_or_create( /// /// **Hive-side.** `None` is the ordinary pre-publication state, not a failure: /// a hive holds no grant to create this bucket and must not treat its absence -/// as a reason to converge to an empty agent set — see `swarm_wire::wanted`. +/// as a reason to converge to an empty agent set — see the module docs. /// /// # Errors /// @@ -107,3 +205,107 @@ pub async fn watch( source, }) } + +#[cfg(test)] +mod tests { + use super::{AgentState, BUCKET_PREFIX, HiveWanted, bucket}; + + #[test] + fn each_hive_gets_its_own_bucket_name() { + assert_eq!(bucket("alpha"), "hive-wanted-alpha"); + assert_ne!(bucket("alpha"), bucket("beta")); + } + + #[test] + fn a_bucket_name_is_legal_for_every_legal_hive_name() { + // The client rejects a bucket name outside `[a-zA-Z0-9_-]`, and a hive + // name is an `Ident` — lowercase, digits, hyphen. Pinned here because + // the grant scoping in `swarm-nats-auth` names this string, so a bucket + // the client refuses to open would surface as a permissions problem + // rather than as the naming problem it is. + let legal = |s: &str| { + s.chars() + .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') + }; + assert!(legal(&bucket("alpha"))); + assert!(legal(&bucket("a-hive-with-hyphens"))); + assert!(legal(&bucket("h9"))); + // Control: the predicate can fail, so the assertions above are not + // vacuously true of any string. + assert!(!legal("hive.wanted")); + } + + #[test] + fn the_prefix_is_what_every_bucket_starts_with() { + assert!(bucket("alpha").starts_with(BUCKET_PREFIX)); + } + + #[test] + fn the_known_states_decode_and_round_trip() { + let doc = r#"{"agents":{"a":{"state":"up"},"b":{"state":"offline"},"c":{"state":"destroyed"},"d":{"state":"paused"}}}"#; + let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes"); + assert_eq!(decoded.agents["a"].state, AgentState::Up); + assert_eq!(decoded.agents["b"].state, AgentState::Offline); + assert_eq!(decoded.agents["c"].state, AgentState::Destroyed); + assert_eq!(decoded.agents["d"].state, AgentState::Paused); + assert_eq!(serde_json::to_string(&decoded).expect("serialises"), doc); + } + + /// The enum is closed, so a state this build does not know takes the + /// **whole document** down rather than one agent. That is the point: a hive + /// converges nothing rather than part of a declaration it half understands. + /// The `up` beside it is the control — a valid entry in the same document + /// does not rescue it, and this test would pass on a broken decoder without + /// the round-trip test above proving the happy path still works. + #[test] + fn an_unknown_state_fails_the_whole_declaration() { + // `"paused"` used to be this test's unknown example; it is a real + // variant now, so `"sleeping"` takes its place as one this build + // still does not know. + let doc = r#"{"agents":{"a":{"state":"up"},"c":{"state":"sleeping"}}}"#; + assert!(serde_json::from_str::(doc).is_err()); + } + + /// Closed *values*, tolerant *fields*: a field a newer controller adds must + /// not cost an older hive the document, since adding one is not a semantic + /// a hive has to understand to obey the rest. + #[test] + fn unknown_fields_are_ignored() { + let doc = r#"{"agents":{"a":{"state":"up","config_rev":"deadbeef"}},"epoch":3}"#; + let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes"); + assert_eq!(decoded.agents["a"].state, AgentState::Up); + } + + /// A declaration missing the one required field is an error, so a malformed + /// document cannot read as "no agents". + #[test] + fn a_missing_state_is_an_error() { + assert!(serde_json::from_str::(r#"{"agents":{"a":{}}}"#).is_err()); + } + + /// `as_str` and the `rename_all` are two spellings of one fact, and a + /// renderer that disagrees with the wire is worse than one that does not + /// exist. The `match` is what makes this exhaustive: a new variant fails + /// to compile here rather than quietly going untested. + #[test] + fn as_str_matches_the_serde_spelling() { + let every = [ + AgentState::Up, + AgentState::Offline, + AgentState::Paused, + AgentState::Destroyed, + ]; + for state in every { + match state { + AgentState::Up + | AgentState::Offline + | AgentState::Paused + | AgentState::Destroyed => {} + } + assert_eq!( + serde_json::to_string(&state).expect("serialises"), + format!("\"{}\"", <&str>::from(state)) + ); + } + } +} diff --git a/swarm-wire/src/agent_icon.rs b/swarm-wire/src/agent_icon.rs deleted file mode 100644 index 2f1703f5..00000000 --- a/swarm-wire/src/agent_icon.rs +++ /dev/null @@ -1,63 +0,0 @@ -//! The per-agent icon KV bucket: one key per agent, holding the icon the -//! swarm shows for it. -//! -//! **Keyed by the agent alone — no hive token**, which is the one way this -//! bucket deliberately differs from its otherwise-identical sibling -//! `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. -//! Operator ruling: *"the bucket is per agent, not per hive. agents can -//! move hives."* -//! -//! That is also what makes the icon answerable for an agent whose -//! container is stopped: the value's lifetime is the agent's, not its -//! placement's. -//! -//! **Each agent writes its own key** with its own queue credential. The auth -//! callout grants that credential `$KV.agent-icons.` and no other key. -//! -//! The writer publishes to [`subject`] directly instead of opening the -//! bucket, so that one subject is its whole grant. Creating the bucket is left -//! to the reader (`swarm_queue_client::agent_icon::open_or_create`). A write -//! that arrives before the bucket exists is refused with "no stream", and the -//! writer retries. - -/// 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 -/// same bucket, and an option is a way for the two to disagree. -pub const BUCKET: &str = "agent-icons"; - -/// The media type of every value in this bucket. -/// -/// The value is the icon's bytes **verbatim, not a JSON envelope**: an SVG -/// carries no metadata this bucket would have to describe, and the one -/// consumer serves the bytes straight back out. So the type is fixed here -/// rather than stored per entry — a publisher that has something other -/// than an SVG does not have an agent icon. -pub const MEDIA_TYPE: &str = "image/svg+xml"; - -/// The subject `agent`'s entry is written on: `$KV..`, where the -/// key is the agent name and nothing else. -/// -/// A KV put is a `JetStream` publish to this subject, so a writer that -/// publishes directly has to use the same subject the bucket's `get` reads. -#[must_use] -pub fn subject(agent: &str) -> String { - format!("$KV.{BUCKET}.{agent}") -} - -#[cfg(test)] -mod tests { - use super::subject; - - /// One token after the bucket, the agent name alone. That is what the - /// per-agent grant `$KV.agent-icons.{agent}` expands to, and what a - /// hive-scoped grant (`$KV...*`, two tokens) cannot match. - #[test] - fn the_published_subject_carries_the_agent_and_no_placement() { - assert_eq!(subject("iris"), "$KV.agent-icons.iris"); - } -} diff --git a/swarm-wire/src/agent_inference.rs b/swarm-wire/src/agent_inference.rs index b5586aea..43aeaf65 100644 --- a/swarm-wire/src/agent_inference.rs +++ b/swarm-wire/src/agent_inference.rs @@ -7,11 +7,11 @@ //! no swarm. //! //! **An agent reads with one `DIRECT.GET` on its own key, not a watch**, for -//! the reason [`crate::agent_schedules`] gives: the direct get names the key -//! in its subject ([`direct_get_subject`]), which is what lets the queue grant -//! scope the read to one agent. +//! the reason `swarm_queue_client::agent_schedules` gives: the direct get names the key +//! in its subject ([`direct_get_subject`]), which is +//! what lets the queue grant scope the read to one agent. //! -//! Keyed by the agent alone, like [`crate::agent_schedules`]: an agent can +//! Keyed by the agent alone, like `swarm_queue_client::agent_schedules`: an agent can //! move between hives. /// The KV bucket inference settings are published into, keyed by agent name. diff --git a/swarm-wire/src/agent_limits.rs b/swarm-wire/src/agent_limits.rs deleted file mode 100644 index 0d8a7e4f..00000000 --- a/swarm-wire/src/agent_limits.rs +++ /dev/null @@ -1,93 +0,0 @@ -//! The per-agent resource-cap KV bucket: one key per agent, holding the CPU -//! and memory caps the swarm sets for it. -//! -//! **swarm-controller is the only writer**, and creates the bucket on its -//! first write. Every field is optional: an absent field leaves the agent on -//! its hive's own answer, so an agent with no key runs exactly as it would -//! with no swarm. -//! -//! **The hive running the agent reads the key**, not the agent: the caps go -//! into the container's systemd drop-in, which the host writes. A hive reads -//! with one `DIRECT.GET` per agent ([`direct_get_subject`]), and its grant -//! covers every key of this bucket, read-only. Scoping it to the hive's own -//! agents would need a roster in the auth responder. -//! -//! Keyed by the agent alone, like [`crate::agent_inference`]: an agent can -//! move between hives, and its caps go with it. - -/// The KV bucket caps are published into, keyed by agent name. -pub const BUCKET: &str = "agent-limits"; - -/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. -pub const STREAM: &str = "KV_agent-limits"; - -/// The subject `agent`'s entry is stored under. -#[must_use] -pub fn key_subject(agent: &str) -> String { - format!("$KV.{BUCKET}.{agent}") -} - -/// The request subject that reads `agent`'s current entry. -#[must_use] -pub fn direct_get_subject(agent: &str) -> String { - format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent)) -} - -/// One agent's caps: the value under `BUCKET`/``, as systemd values -/// carried verbatim. `None` is "not capped by the swarm", never "uncapped". -#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct AgentCaps { - /// systemd `CPUQuota=`, e.g. `"400%"`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub cpu_quota: Option, - /// systemd `MemoryMax=`, e.g. `"8G"`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub memory_max: Option, -} - -impl AgentCaps { - /// True when no field is set, which a hive reads the same as no key. - #[must_use] - pub fn is_empty(&self) -> bool { - self.cpu_quota.is_none() && self.memory_max.is_none() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - /// The names the hive's grant in `swarm-nats-auth` is built from. - #[test] - fn names_match_the_queue_grant() { - assert_eq!(STREAM, format!("KV_{BUCKET}")); - assert_eq!(key_subject("iris"), "$KV.agent-limits.iris"); - assert_eq!( - direct_get_subject("iris"), - "$JS.API.DIRECT.GET.KV_agent-limits.$KV.agent-limits.iris" - ); - } - - /// An unset field is left out on the wire, and `{}` is every field unset. - #[test] - fn unset_fields_are_absent() { - let value = AgentCaps { - memory_max: Some("8G".to_owned()), - ..AgentCaps::default() - }; - assert_eq!( - serde_json::to_string(&value).unwrap(), - r#"{"memory_max":"8G"}"# - ); - let empty: AgentCaps = serde_json::from_str("{}").unwrap(); - assert!(empty.is_empty()); - } - - /// Fields a newer controller adds must not cost an older hive its caps. - #[test] - fn decoding_tolerates_unknown_fields() { - let decoded: AgentCaps = - serde_json::from_str(r#"{"cpu_quota":"400%","io_weight":"10"}"#).unwrap(); - assert_eq!(decoded.cpu_quota.as_deref(), Some("400%")); - } -} diff --git a/swarm-wire/src/agent_schedules.rs b/swarm-wire/src/agent_schedules.rs deleted file mode 100644 index 12fb7ad1..00000000 --- a/swarm-wire/src/agent_schedules.rs +++ /dev/null @@ -1,147 +0,0 @@ -//! The per-agent schedule KV bucket: one key per agent, holding the scheduled -//! prompts that fire in that agent. -//! -//! **swarm-controller is the only writer**, and creates the bucket on its -//! first write. It keeps each prompt once, with its name and every agent it -//! goes to, and writes each agent's key as the prompts naming that agent, -//! without the name or the other agents. Each agent's harness reads its own -//! key and fires the schedules itself; the fire times are a pure function of -//! each schedule ([`Schedule::latest_slot`]), so rewriting the list never moves -//! them. -//! -//! **An agent reads with one `DIRECT.GET` on its own key, not a watch.** A KV -//! watch is an unnamed consumer, whose create subject carries no filter, so -//! granting it would grant every agent's key. The direct get carries the key -//! in its subject ([`direct_get_subject`]), which is what lets the queue grant -//! scope the read to one agent. It is also the agent's whole `JetStream` grant: -//! `swarm_queue_client::agent_schedules::read_own` builds the stream handle -//! without a `STREAM.INFO`. -//! -//! Keyed by the agent alone, like [`crate::agent_icon`], and for the same -//! reason: an agent can move between hives. - -/// The KV bucket schedules are published into, keyed by agent name. -pub const BUCKET: &str = "agent-schedules"; - -/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. -pub const STREAM: &str = "KV_agent-schedules"; - -/// The subject `agent`'s entry is stored under. -#[must_use] -pub fn key_subject(agent: &str) -> String { - format!("$KV.{BUCKET}.{agent}") -} - -/// The request subject that reads `agent`'s current entry, and so the one -/// subject the queue grants that agent in this bucket. -#[must_use] -pub fn direct_get_subject(agent: &str) -> String { - format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject(agent)) -} - -/// The prompts that fire in one agent: the value under `BUCKET`/``. -#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct AgentSchedules { - #[serde(default)] - pub schedules: Vec, -} - -/// One scheduled prompt. -/// -/// Its fire times are `anchor_unix + k * interval_seconds` for `k >= 0`, or -/// `anchor_unix` alone when there is no interval. -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct Schedule { - /// Minted by the controller, the same in every agent the prompt names, - /// and never reused, so an agent's record of what it last fired for an - /// id cannot apply to a different schedule. - pub id: String, - /// The prompt, delivered verbatim as the todo's summary. - pub body: String, - /// The first fire time, unix seconds. - pub anchor_unix: i64, - /// Seconds between fires; `None` is a one-shot. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub interval_seconds: Option, - /// A paused schedule does not fire, and the slots that pass while it is - /// paused are not fired when it resumes. - #[serde(default)] - pub paused: bool, -} - -impl Schedule { - /// The latest fire time at or before `now`, or `None` while the first is - /// still ahead. - #[must_use] - pub fn latest_slot(&self, now: i64) -> Option { - if now < self.anchor_unix { - return None; - } - let Some(interval) = self - .interval_seconds - .filter(|i| *i > 0) - .and_then(|i| i64::try_from(i).ok()) - else { - return Some(self.anchor_unix); - }; - let elapsed = now - self.anchor_unix; - Some(self.anchor_unix + elapsed - elapsed % interval) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn every(anchor_unix: i64, interval: Option) -> Schedule { - Schedule { - id: "abc".to_owned(), - body: "hi".to_owned(), - anchor_unix, - interval_seconds: interval, - paused: false, - } - } - - /// The names the grant in `swarm-nats.nix` spells with `{agent}`. - #[test] - fn names_match_the_queue_grant() { - assert_eq!(STREAM, format!("KV_{BUCKET}")); - assert_eq!(key_subject("iris"), "$KV.agent-schedules.iris"); - assert_eq!( - direct_get_subject("iris"), - "$JS.API.DIRECT.GET.KV_agent-schedules.$KV.agent-schedules.iris" - ); - } - - #[test] - fn slots_sit_on_the_anchor_grid() { - let s = every(1000, Some(60)); - assert_eq!(s.latest_slot(999), None); - assert_eq!(s.latest_slot(1000), Some(1000)); - assert_eq!(s.latest_slot(1059), Some(1000)); - assert_eq!(s.latest_slot(1060), Some(1060)); - assert_eq!(s.latest_slot(1000 + 60 * 7 + 5), Some(1000 + 60 * 7)); - } - - #[test] - fn a_one_shot_has_one_slot() { - let s = every(1000, None); - assert_eq!(s.latest_slot(999), None); - assert_eq!(s.latest_slot(1000), Some(1000)); - assert_eq!(s.latest_slot(1_000_000), Some(1000)); - } - - /// Fields a newer controller adds must not cost an older agent its list, - /// and a list written without the optional fields decodes to a recurring - /// or one-shot schedule that is not paused. - #[test] - fn decoding_tolerates_missing_and_unknown_fields() { - let doc = r#"{"schedules":[{"id":"a","body":"b","anchor_unix":5,"note":"x"}],"rev":2}"#; - let decoded: AgentSchedules = serde_json::from_str(doc).expect("decodes"); - let mut want = every(5, None); - "a".clone_into(&mut want.id); - "b".clone_into(&mut want.body); - assert_eq!(decoded.schedules, vec![want]); - } -} diff --git a/swarm-wire/src/agent_status.rs b/swarm-wire/src/agent_status.rs deleted file mode 100644 index b9c2f7e5..00000000 --- a/swarm-wire/src/agent_status.rs +++ /dev/null @@ -1,153 +0,0 @@ -//! 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 e9b1bdf6..8f95cf72 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -5,14 +5,8 @@ //! names a contract (the sender, the answerer, the auth-callout responder that //! grants the subject) depends on this one without linking either. -pub mod agent_icon; pub mod agent_inference; -pub mod agent_limits; pub mod agent_login; -pub mod agent_schedules; -pub mod agent_status; pub mod builder; pub mod claude_login; -pub mod notices; pub mod review_classifier; -pub mod wanted; diff --git a/swarm-wire/src/notices.rs b/swarm-wire/src/notices.rs deleted file mode 100644 index 30c4db26..00000000 --- a/swarm-wire/src/notices.rs +++ /dev/null @@ -1,37 +0,0 @@ -//! The hive-notices stream: its name and subject shape. -//! -//! A hive that publishes and a swarm-level reader that consumes live in -//! different crates, and a literal name repeated across both is an agreement -//! nothing checks. -//! -//! **This is a stream, not a bucket.** The hive-status snapshot has a current -//! value, so a KV bucket (last value per key) is the right shape. A lifecycle -//! notice ("container crashed at 04:12") has no such steady state: miss the -//! message and nothing left to read later says it happened. That needs durable -//! delivery (a `JetStream` stream a consumer acks against), which -//! `swarm_queue_client::notices::open_or_create` opens. - -/// The stream a hive publishes lifecycle notices into. -/// -/// A constant and not an option, like every bucket name here: -/// reader and writer must name the same stream, and letting either side -/// pick its own name is how two deployments end up disagreeing about -/// which stream a notice actually landed in. -pub const STREAM: &str = "hive-notices"; - -/// Build the subject a given hive's notices publish to. -/// -/// Every hive's notices land under `STREAM`, one subject per hive: -/// `hive-notices.`. -/// -/// Not one subject per notice *kind* — a consumer that wants a specific -/// hive's notices subscribes to `subject(hive)`; one that wants the whole -/// swarm's subscribes to `{STREAM}.>`. The kind travels inside the message -/// payload instead, so adding a new notice kind is never a subject-design -/// change. Deliberately reuses `STREAM` rather than a second `PREFIX` -/// constant with the same value — one name for one fact, same reasoning -/// the module doc above gives for a shared `const` over a repeated literal. -#[must_use] -pub fn subject(hive: &str) -> String { - format!("{STREAM}.{hive}") -} diff --git a/swarm-wire/src/wanted.rs b/swarm-wire/src/wanted.rs deleted file mode 100644 index 2bf2b3f0..00000000 --- a/swarm-wire/src/wanted.rs +++ /dev/null @@ -1,221 +0,0 @@ -//! The hive-wanted KV buckets: the agent set the controller declares for each -//! hive, **one bucket per hive** (see [`bucket`] for why the -//! split is a grant boundary rather than a data-modelling choice), keyed inside -//! it by `hiveName`. -//! -//! Sibling of `swarm_queue_client::status` and deliberately **not** a mirror of it: that -//! one is **observed** — each hive republishes what it is, so its store losing -//! everything "degrades to honesty" (`swarm-nats.nix`). This one is -//! **declared**, and nothing regenerates it; lose it and a reconcile loop has -//! nothing to converge to, which is the failure such a loop exists to remove. -//! -//! # Absence is not a deletion order -//! -//! A hive that finds no key for itself — or no bucket at all — has learned -//! nothing about the agents it is running, not that it should have none. -//! Swarm-side lifecycle does not yet cover agents that predate it, so callers -//! converge the agents a value NAMES and leave the rest alone. -//! -//! # Only the controller creates it -//! -//! Unlike `swarm_queue_client::status::open_or_create`, where either end may -//! legitimately arrive first, the writer here is single and known. A hive -//! opens read-only and treats absence as the case above, so "not published -//! yet" stays quiet rather than looking like an error a hive could fix. - -/// The prefix every hive's wanted-state bucket name starts with. -/// -/// Derived from, and not configurable, for the reason -/// `swarm_queue_client::status::BUCKET` gives: writer and reader must name the -/// same bucket, and an option is a way for two deployments to disagree about -/// which one that is. -/// for two deployments to disagree about which one that is. -pub const BUCKET_PREFIX: &str = "hive-wanted-"; - -/// The KV bucket one hive's declaration lives in — **one bucket per hive**, not -/// one bucket keyed by hive. -/// -/// The split is a grant-scoping decision, not a data-modelling one. A KV read -/// scopes per key, because `DIRECT.GET` carries the key in the subject; a -/// *watch* does not, because a consumer's filter travels in the request payload -/// and `$JS.API.CONSUMER.CREATE.` therefore grants the whole stream. So -/// with every hive in one bucket, letting a hive watch its own declaration -/// means letting it read every other hive's. One stream per hive makes the -/// grant a hive can hold exactly as wide as what it is allowed to see. -/// -/// A bucket name may contain `[a-zA-Z0-9_-]`, and a hive name is an `Ident` -/// (`[a-z0-9-]`), so this composition is always a legal bucket name. -#[must_use] -pub fn bucket(hive: &str) -> String { - format!("{BUCKET_PREFIX}{hive}") -} - -/// One hive's whole declaration: the agents the controller names, and what it -/// wants of each. The value under `BUCKET`/``. -/// -/// The agents it does **not** name are not part of the declaration at all — -/// see the module docs. A hive converges the ones listed here and leaves the -/// rest alone. -#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct HiveWanted { - /// Agent name → what the controller wants of it. - #[serde(default)] - pub agents: std::collections::BTreeMap, -} - -/// One agent's entry in a [`HiveWanted`]. -/// -/// A struct around one field rather than the bare state, because this is where -/// a later declaration grows — a config revision, a schedule — and a hive -/// built before that field existed keeps deserialising the whole document. -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub struct AgentWanted { - /// The state to converge this agent to. - pub state: AgentState, -} - -/// What the controller wants an agent to be. -/// -/// **Closed on purpose.** A value this build does not know fails the whole -/// document's decode, so a hive running older code converges **nothing** rather -/// than part of a declaration it only half understands. Adding a state means -/// adding a variant here and shipping it to both ends — which is the intended -/// workflow, not an obstacle to route around with a catch-all variant. -#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum AgentState { - /// Exists on the hive and is running. - Up, - /// Exists on the hive and is not running. - Offline, - /// Exists on the hive, is running, and its turn loop is parked — the - /// swarm-declared counterpart of the hive-local `hivectl agent pause` - /// marker. Orthogonal to the power axis `Up`/`Offline` express: a - /// paused agent still has a running container (the web UI and MCP - /// daemons stay reachable), it just drives no turns. See - /// `hive-c0re::workers::wanted::decide_pause` for the convergence side - /// of that split. - Paused, - /// Torn down entirely — the hive runs the destroy template once, then - /// treats absence as agreement rather than re-running it. The key - /// stays in the declared set with this state forever, on purpose: it's - /// what lets a redeclare of an already-destroyed agent be rejected as - /// a terminal-state transition instead of silently reviving it. - /// Irreversible: there is no state that brings a destroyed agent back - /// short of a fresh deploy. - Destroyed, -} - -impl From for &'static str { - /// The state's wire spelling, the same string serde writes. - fn from(state: AgentState) -> Self { - match state { - AgentState::Up => "up", - AgentState::Offline => "offline", - AgentState::Paused => "paused", - AgentState::Destroyed => "destroyed", - } - } -} - -#[cfg(test)] -mod tests { - use super::{AgentState, BUCKET_PREFIX, HiveWanted, bucket}; - - #[test] - fn each_hive_gets_its_own_bucket_name() { - assert_eq!(bucket("alpha"), "hive-wanted-alpha"); - assert_ne!(bucket("alpha"), bucket("beta")); - } - - #[test] - fn a_bucket_name_is_legal_for_every_legal_hive_name() { - // The client rejects a bucket name outside `[a-zA-Z0-9_-]`, and a hive - // name is an `Ident` — lowercase, digits, hyphen. Pinned here because - // the grant scoping in `swarm-nats-auth` names this string, so a bucket - // the client refuses to open would surface as a permissions problem - // rather than as the naming problem it is. - let legal = |s: &str| { - s.chars() - .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') - }; - assert!(legal(&bucket("alpha"))); - assert!(legal(&bucket("a-hive-with-hyphens"))); - assert!(legal(&bucket("h9"))); - // Control: the predicate can fail, so the assertions above are not - // vacuously true of any string. - assert!(!legal("hive.wanted")); - } - - #[test] - fn the_prefix_is_what_every_bucket_starts_with() { - assert!(bucket("alpha").starts_with(BUCKET_PREFIX)); - } - - #[test] - fn the_known_states_decode_and_round_trip() { - let doc = r#"{"agents":{"a":{"state":"up"},"b":{"state":"offline"},"c":{"state":"destroyed"},"d":{"state":"paused"}}}"#; - let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes"); - assert_eq!(decoded.agents["a"].state, AgentState::Up); - assert_eq!(decoded.agents["b"].state, AgentState::Offline); - assert_eq!(decoded.agents["c"].state, AgentState::Destroyed); - assert_eq!(decoded.agents["d"].state, AgentState::Paused); - assert_eq!(serde_json::to_string(&decoded).expect("serialises"), doc); - } - - /// The enum is closed, so a state this build does not know takes the - /// **whole document** down rather than one agent. That is the point: a hive - /// converges nothing rather than part of a declaration it half understands. - /// The `up` beside it is the control — a valid entry in the same document - /// does not rescue it, and this test would pass on a broken decoder without - /// the round-trip test above proving the happy path still works. - #[test] - fn an_unknown_state_fails_the_whole_declaration() { - // `sleeping` is not a variant, so the document is refused. - let doc = r#"{"agents":{"a":{"state":"up"},"c":{"state":"sleeping"}}}"#; - assert!(serde_json::from_str::(doc).is_err()); - } - - /// Closed *values*, tolerant *fields*: a field a newer controller adds must - /// not cost an older hive the document, since adding one is not a semantic - /// a hive has to understand to obey the rest. - #[test] - fn unknown_fields_are_ignored() { - let doc = r#"{"agents":{"a":{"state":"up","config_rev":"deadbeef"}},"epoch":3}"#; - let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes"); - assert_eq!(decoded.agents["a"].state, AgentState::Up); - } - - /// A declaration missing the one required field is an error, so a malformed - /// document cannot read as "no agents". - #[test] - fn a_missing_state_is_an_error() { - assert!(serde_json::from_str::(r#"{"agents":{"a":{}}}"#).is_err()); - } - - /// The `&str` conversion and the `rename_all` are two spellings of one fact, - /// and a renderer that disagrees with the wire is worse than one that does - /// not exist. The `match` is what makes this exhaustive: a new variant fails - /// to compile here rather than quietly going untested. - #[test] - fn the_str_conversion_matches_the_serde_spelling() { - let every = [ - AgentState::Up, - AgentState::Offline, - AgentState::Paused, - AgentState::Destroyed, - ]; - for state in every { - match state { - AgentState::Up - | AgentState::Offline - | AgentState::Paused - | AgentState::Destroyed => {} - } - assert_eq!( - serde_json::to_string(&state).expect("serialises"), - format!("\"{}\"", <&str>::from(state)) - ); - } - } -}