vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-wire: take wanted's wire types

Browse source
This commit is contained in:
atlas 2026-10-18 12:22:36 +02:00 • committed by mara
commit 266be0a5db
17 changed files with 298 additions and 282 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -5036,7 +5036,6 @@ dependencies = [
"reqwest",
"serde",
"serde_json",
"strum",
"swarm-secret-client",
"swarm-wire",
"thiserror 2.0.18",

2
frontend/packages/swarm-ui/src/agent/AgentTypes.ts
View file

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

4
hive-agent/src/swarm_agent_state.rs
View file

@ -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};

6
hive-c0re/src/workers/wanted.rs
View file

@ -1,7 +1,7 @@
//! Converging this hive onto the agent set the swarm controller declares.
//!
//! The controller gives each hive its own `hive-wanted-<hive>` 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 {

2
swarm-controller/src/agent_limits.rs
View file

@ -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

2
swarm-controller/src/agent_permissions.rs
View file

@ -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

4
swarm-controller/src/agent_renewal.rs
View file

@ -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;

2
swarm-controller/src/agents_md_initial.rs
View file

@ -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

8
swarm-controller/src/jobq.rs
View file

@ -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
);
}

88
swarm-controller/src/routes/agents.rs
View file

@ -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<AgentDeclaration> {
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<R, D>(
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<Output = Result<bool, String>>,
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<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>,
@ -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<E, Fut>(
) -> Result<Vec<String>, E>
where
Fut: std::future::Future<
Output = Result<Vec<(String, swarm_queue_client::wanted::HiveWanted)>, E>,
Output = Result<Vec<(String, swarm_wire::wanted::HiveWanted)>, E>,
>,
{
let queued = snapshot_queued();
@ -886,7 +886,7 @@ async fn declarations_elsewhere(
state: &AppState,
agent: &str,
hive: &str,
) -> Result<Vec<(String, swarm_queue_client::wanted::HiveWanted)>, problem_details::ProblemDetails>
) -> Result<Vec<(String, swarm_wire::wanted::HiveWanted)>, 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<Item = &'a HiveEntry>,
) -> Result<Vec<(String, swarm_queue_client::wanted::HiveWanted)>, (String, anyhow::Error)> {
) -> Result<Vec<(String, swarm_wire::wanted::HiveWanted)>, (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<String> {
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<String> {
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<bool, String>,
) -> 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;

4
swarm-controller/src/wanted.rs
View file

@ -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() {

4
swarm-nats-auth/src/policy.rs
View file

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

1
swarm-queue-client/Cargo.toml
View file

@ -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

12
swarm-queue-client/src/lib.rs
View file

@ -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`.

218
swarm-queue-client/src/wanted.rs
View file

@ -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.<stream>` 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`/`<hiveName>`.
///
/// 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<String, AgentWanted>,
}
/// 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::<HiveWanted>(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::<HiveWanted>(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))
);
}
}
}

1
swarm-wire/src/lib.rs
View file

@ -12,3 +12,4 @@ pub mod agent_status;
pub mod builder;
pub mod claude_login;
pub mod review_classifier;
pub mod wanted;

221
swarm-wire/src/wanted.rs Normal file
View file

@ -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.<stream>` 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`/`<hiveName>`.
///
/// 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<String, AgentWanted>,
}
/// 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<AgentState> 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::<HiveWanted>(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::<HiveWanted>(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))
);
}
}
}