From 266be0a5dbeb8d42e1919005666bc437de01c1d0 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 18 Oct 2026 12:22:36 +0200 Subject: [PATCH] swarm-wire: take wanted's wire types --- Cargo.lock | 1 - .../packages/swarm-ui/src/agent/AgentTypes.ts | 2 +- hive-agent/src/swarm_agent_state.rs | 4 +- hive-c0re/src/workers/wanted.rs | 6 +- swarm-controller/src/agent_limits.rs | 2 +- swarm-controller/src/agent_permissions.rs | 2 +- swarm-controller/src/agent_renewal.rs | 4 +- swarm-controller/src/agents_md_initial.rs | 2 +- swarm-controller/src/jobq.rs | 8 +- swarm-controller/src/routes/agents.rs | 88 +++---- swarm-controller/src/wanted.rs | 4 +- swarm-nats-auth/src/policy.rs | 4 +- swarm-queue-client/Cargo.toml | 1 - swarm-queue-client/src/lib.rs | 12 +- swarm-queue-client/src/wanted.rs | 218 +---------------- swarm-wire/src/lib.rs | 1 + swarm-wire/src/wanted.rs | 221 ++++++++++++++++++ 17 files changed, 298 insertions(+), 282 deletions(-) create mode 100644 swarm-wire/src/wanted.rs diff --git a/Cargo.lock b/Cargo.lock index 4f6fa732..a705ad8a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5036,7 +5036,6 @@ 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 0f78b82a..dc82f7d1 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_queue_client::wanted::AgentState::as_str` — kept as a bare +// `swarm_wire::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/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index 845826c1..11344edb 100644 --- a/hive-agent/src/swarm_agent_state.rs +++ b/hive-agent/src/swarm_agent_state.rs @@ -33,7 +33,7 @@ use serde::Serialize; use tokio::sync::broadcast; use tokio::time::{Interval, MissedTickBehavior}; -use swarm_queue_client::wanted::AgentState; +use swarm_wire::wanted::AgentState; use swarm_wire::agent_inference::Effective; use swarm_wire::agent_login::Login; @@ -337,7 +337,7 @@ 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::wanted::AgentState; use swarm_wire::agent_inference::{Effective, Source}; use tokio::sync::broadcast; use tokio::time::{Duration, Instant}; diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index a488c5ee..af7bd5ca 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_queue_client::wanted`]); this reads its own and acts on it. It is +//! [`swarm_wire::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_queue_client::wanted::{AgentState, HiveWanted}; +use swarm_wire::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_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; + use swarm_wire::wanted::{AgentState, AgentWanted, HiveWanted}; fn declaration(agents: &[(&str, AgentState)]) -> HiveWanted { HiveWanted { diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index 9c4bcfa9..bc463b71 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -358,7 +358,7 @@ pub(crate) mod tests { use axum::extract::{Path, State}; use axum::http::StatusCode; use std::sync::Arc; - use swarm_queue_client::wanted::AgentState; + use swarm_wire::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 7aacea6a..f20ace7d 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_queue_client::wanted::AgentState::Destroyed, + state: swarm_wire::wanted::AgentState::Destroyed, }), ) .await diff --git a/swarm-controller/src/agent_renewal.rs b/swarm-controller/src/agent_renewal.rs index a3e38bd9..9977718b 100644 --- a/swarm-controller/src/agent_renewal.rs +++ b/swarm-controller/src/agent_renewal.rs @@ -33,7 +33,7 @@ use std::collections::BTreeSet; use std::sync::Arc; use anyhow::{Context, Result, bail}; -use swarm_queue_client::wanted::{AgentState, HiveWanted}; +use swarm_wire::wanted::{AgentState, HiveWanted}; use swarm_secret_client::{SecretStore, mtls, queue}; use x509_cert::der::DecodePem as _; @@ -429,7 +429,7 @@ pub fn spawn( #[cfg(test)] mod tests { use super::*; - use swarm_queue_client::wanted::AgentWanted; + use swarm_wire::wanted::AgentWanted; const NOW: i64 = 1_790_000_000; const DAY: i64 = 86_400; diff --git a/swarm-controller/src/agents_md_initial.rs b/swarm-controller/src/agents_md_initial.rs index 235ddc2c..54df8a51 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_queue_client::wanted::AgentState::Destroyed, + state: swarm_wire::wanted::AgentState::Destroyed, }), ) .await diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 54cb4ee6..6934b907 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_queue_client::wanted::AgentState = - swarm_queue_client::wanted::AgentState::Paused; +const NEW_AGENT_WANTED_STATE: swarm_wire::wanted::AgentState = + swarm_wire::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_queue_client::wanted::AgentState; + use swarm_wire::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_queue_client::wanted::AgentState::Paused + swarm_wire::wanted::AgentState::Paused ); } diff --git a/swarm-controller/src/routes/agents.rs b/swarm-controller/src/routes/agents.rs index a348f437..637422b8 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_queue_client::wanted::AgentState, + pub(crate) state: swarm_wire::wanted::AgentState, } /// One agent's line in a hive's declaration. @@ -79,7 +79,7 @@ pub(crate) struct AgentDeclaration { } pub(crate) fn render( - declaration: &swarm_queue_client::wanted::HiveWanted, + declaration: &swarm_wire::wanted::HiveWanted, ) -> Vec { declaration .agents @@ -223,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_queue_client::wanted::AgentState) -> bool { - use swarm_queue_client::wanted::AgentState; +fn places_agent(state: swarm_wire::wanted::AgentState) -> bool { + use swarm_wire::wanted::AgentState; match state { AgentState::Up | AgentState::Offline | AgentState::Paused => true, AgentState::Destroyed => false, @@ -243,7 +243,7 @@ fn places_agent(state: swarm_queue_client::wanted::AgentState) -> bool { async fn refuse_placement( agent: &str, hive: &str, - here: Option<&swarm_queue_client::wanted::HiveWanted>, + here: Option<&swarm_wire::wanted::HiveWanted>, reserved: Option<&str>, read_roster: impl FnOnce() -> R, snapshot_queued: impl FnOnce() -> Vec<(String, String)>, @@ -253,7 +253,7 @@ where R: std::future::Future>, D: std::future::Future< Output = Result< - Vec<(String, swarm_queue_client::wanted::HiveWanted)>, + Vec<(String, swarm_wire::wanted::HiveWanted)>, problem_details::ProblemDetails, >, >, @@ -295,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_queue_client::wanted::AgentState] in this +/// [`AgentState`][swarm_wire::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_queue_client::wanted::AgentState) -> bool { - use swarm_queue_client::wanted::AgentState; +fn revokes_queue_credential(state: swarm_wire::wanted::AgentState) -> bool { + use swarm_wire::wanted::AgentState; match state { AgentState::Destroyed => true, AgentState::Up | AgentState::Offline | AgentState::Paused => false, @@ -425,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_queue_client::wanted::HiveWanted, + swarm_wire::wanted::HiveWanted, > = std::collections::HashMap::new(); for hive in hives { match writer.view(&hive).await { @@ -711,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, @@ -794,7 +794,7 @@ impl StoredSettings { async fn store( client: &async_nats::Client, agent: &str, - limits: Option<&swarm_queue_client::agent_limits::AgentCaps>, + limits: Option<&swarm_wire::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>, @@ -868,7 +868,7 @@ async fn placements_elsewhere( ) -> Result, E> where Fut: std::future::Future< - Output = Result, E>, + Output = Result, E>, >, { let queued = snapshot_queued(); @@ -886,7 +886,7 @@ 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 @@ -905,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()); }; @@ -923,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_queue_client::wanted::HiveWanted)], + declared: &[(String, swarm_wire::wanted::HiveWanted)], ) -> Vec { declared .iter() @@ -971,7 +971,7 @@ fn queued_placements( fn placed_elsewhere( agent: &str, hive: &str, - declared: &[(String, swarm_queue_client::wanted::HiveWanted)], + declared: &[(String, swarm_wire::wanted::HiveWanted)], queued: &[(String, String)], ) -> Vec { let declared = declared.iter().filter(|(_, declaration)| { @@ -1706,12 +1706,12 @@ mod tests { pub(crate) fn declaring( hive: &str, agent: &str, - state: swarm_queue_client::wanted::AgentState, - ) -> (String, swarm_queue_client::wanted::HiveWanted) { - let mut declaration = swarm_queue_client::wanted::HiveWanted::default(); + 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_queue_client::wanted::AgentWanted { state }, + swarm_wire::wanted::AgentWanted { state }, ); (hive.to_owned(), declaration) } @@ -1721,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_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; let reads = std::cell::Cell::new(0); let snapshot_queued = || { reads.set(reads.get() + 1); @@ -1750,7 +1750,7 @@ mod tests { #[test] fn a_live_declaration_on_another_hive_is_a_placement() { - use swarm_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] { let declared = [declaring("sec0nd", "atlas", state)]; assert_eq!( @@ -1767,14 +1767,14 @@ mod tests { let declared = [declaring( "sec0nd", "atlas", - swarm_queue_client::wanted::AgentState::Destroyed, + swarm_wire::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_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; let declared = [ declaring("pr1ma", "atlas", AgentState::Up), declaring("sec0nd", "argus", AgentState::Up), @@ -1788,7 +1788,7 @@ mod tests { #[test] fn the_hive_placing_an_agent_claims_it_and_no_other_does() { - use swarm_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; let declared = [ declaring("pr1ma", "atlas", AgentState::Paused), declaring("sec0nd", "argus", AgentState::Up), @@ -1800,7 +1800,7 @@ mod tests { #[test] fn two_hives_placing_one_agent_both_claim_it() { - use swarm_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; let declared = [ declaring("pr1ma", "atlas", AgentState::Up), declaring("sec0nd", "atlas", AgentState::Up), @@ -1816,7 +1816,7 @@ mod tests { /// state, `pr1ma`'s included. async fn refuse_on_pr1ma_with( agent: &str, - declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>, + declared: Vec<(String, swarm_wire::wanted::HiveWanted)>, roster: Result, ) -> Result<(), problem_details::ProblemDetails> { let here = declared @@ -1838,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_queue_client::wanted::HiveWanted)>, + declared: Vec<(String, swarm_wire::wanted::HiveWanted)>, ) -> Result<(), problem_details::ProblemDetails> { refuse_on_pr1ma_with(agent, declared, Ok(false)).await } - 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, + const PLACING: [swarm_wire::wanted::AgentState; 3] = [ + swarm_wire::wanted::AgentState::Up, + swarm_wire::wanted::AgentState::Offline, + swarm_wire::wanted::AgentState::Paused, ]; /// The handler checks exactly the placing states; a destroy is never @@ -1857,7 +1857,7 @@ mod tests { assert!(super::places_agent(state), "{state:?}"); } assert!(!super::places_agent( - swarm_queue_client::wanted::AgentState::Destroyed + swarm_wire::wanted::AgentState::Destroyed )); } @@ -1920,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_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; let err = refuse_on_pr1ma( "admin", vec![declaring("pr1ma", "admin", AgentState::Destroyed)], @@ -1953,7 +1953,7 @@ mod tests { #[tokio::test] async fn a_first_declaration_of_a_name_placed_on_another_hive_is_refused() { - use swarm_queue_client::wanted::AgentState; + use swarm_wire::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"); @@ -1976,7 +1976,7 @@ mod tests { #[tokio::test] async fn changing_the_state_of_an_agent_on_its_own_hive_is_accepted() { - use swarm_queue_client::wanted::AgentState; + use swarm_wire::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"); @@ -1986,7 +1986,7 @@ mod tests { /// connected and with no identity bridge. async fn set_on_disconnected_pr1ma( agent: &str, - wanted_state: swarm_queue_client::wanted::AgentState, + wanted_state: swarm_wire::wanted::AgentState, ) -> problem_details::ProblemDetails { let (state, _sched) = state_with_two_hives(); let state = crate::routes::AppState { @@ -2011,7 +2011,7 @@ mod tests { #[tokio::test] async fn set_agent_state_refuses_when_the_wanted_state_is_unreadable() { let err = - set_on_disconnected_pr1ma("atlas", swarm_queue_client::wanted::AgentState::Up).await; + set_on_disconnected_pr1ma("atlas", swarm_wire::wanted::AgentState::Up).await; assert_eq!( err.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), @@ -2029,7 +2029,7 @@ mod tests { #[tokio::test] async fn set_agent_state_does_not_check_a_destroy() { let err = - set_on_disconnected_pr1ma("admin", swarm_queue_client::wanted::AgentState::Destroyed) + set_on_disconnected_pr1ma("admin", swarm_wire::wanted::AgentState::Destroyed) .await; assert_eq!( err.status, @@ -2197,7 +2197,7 @@ mod tests { #[test] fn declaring_an_agent_destroyed_revokes_its_queue_credential() { assert!(super::revokes_queue_credential( - swarm_queue_client::wanted::AgentState::Destroyed + swarm_wire::wanted::AgentState::Destroyed )); } @@ -2207,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_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; for state in [AgentState::Up, AgentState::Offline, AgentState::Paused] { assert!( !super::revokes_queue_credential(state), @@ -2350,7 +2350,7 @@ mod tests { axum::extract::State(state), axum::extract::Path(("pr1ma".to_owned(), "atlas".to_owned())), axum::Json(SetAgentStateRequest { - state: swarm_queue_client::wanted::AgentState::Paused, + state: swarm_wire::wanted::AgentState::Paused, }), ) .await; @@ -2395,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_queue_client::wanted::AgentState::Paused, + state: swarm_wire::wanted::AgentState::Paused, }), ) .await; diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index b6a9ffb6..c33bf4e5 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_queue_client::wanted::{AgentState, AgentWanted, HiveWanted}; +use swarm_wire::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_queue_client::wanted::AgentState; + use swarm_wire::wanted::AgentState; #[test] fn declaring_one_agent_preserves_every_other() { diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 0ba68a71..4229a70a 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_queue_client::wanted::bucket(hive)) + format!("KV_{}", swarm_wire::wanted::bucket(hive)) } /// The stream backing the per-agent status bucket. @@ -530,7 +530,7 @@ impl Policy { format!("$JS.API.STREAM.INFO.{wanted}"), format!( "$JS.API.DIRECT.GET.{wanted}.$KV.{}.{hive}", - swarm_queue_client::wanted::bucket(hive) + swarm_wire::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. diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 7c8de72c..75685c95 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -81,7 +81,6 @@ 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/lib.rs b/swarm-queue-client/src/lib.rs index e7b22d55..85fa5a2e 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 (`wanted::bucket`), so the error has to own it. + // at runtime (`swarm_wire::wanted::bucket`), so the error has to own it. bucket: String, #[source] source: async_nats::jetstream::context::CreateKeyValueError, @@ -188,11 +188,10 @@ pub fn chain(error: &dyn std::error::Error) -> String { /// which is the disagreement this module exists to prevent. pub mod status; -/// 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. +/// 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`. pub mod wanted; /// Opening the bucket per-agent status snapshots are published into: one key @@ -201,7 +200,6 @@ pub mod wanted; /// merge, are in `swarm_wire::agent_status`. 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`. diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index 16681441..597021e4 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -1,116 +1,18 @@ -//! 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. +//! 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`. //! //! # 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 the case above, so "not published yet" stays quiet -//! rather than looking like an error a hive could fix. +//! and treats absence as "not published yet", so that stays quiet rather than +//! looking like an error a hive could fix. #[cfg(feature = "kv")] use crate::Error; - -/// 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, -} +#[cfg(feature = "kv")] +use swarm_wire::wanted::bucket; /// Open the wanted-state bucket for writing, creating it if nothing has yet. /// @@ -133,7 +35,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 the module docs. +/// as a reason to converge to an empty agent set — see `swarm_wire::wanted`. /// /// # Errors /// @@ -205,107 +107,3 @@ 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/lib.rs b/swarm-wire/src/lib.rs index 0c6e9440..c8a3602c 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -12,3 +12,4 @@ pub mod agent_status; pub mod builder; pub mod claude_login; pub mod review_classifier; +pub mod wanted; diff --git a/swarm-wire/src/wanted.rs b/swarm-wire/src/wanted.rs new file mode 100644 index 00000000..2bf2b3f0 --- /dev/null +++ b/swarm-wire/src/wanted.rs @@ -0,0 +1,221 @@ +//! 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)) + ); + } + } +}