From fc72bea1511e09074d2e14a9f215f490f79b890d Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 18 Oct 2026 21:34:58 +0200 Subject: [PATCH] swarm-wire: take notices's wire types, fmt the moved imports --- hive-agent/src/swarm_agent_state.rs | 4 +-- hive-c0re/src/swarm_notices.rs | 4 +-- swarm-controller/src/agent_renewal.rs | 2 +- swarm-controller/src/agent_status.rs | 7 ++-- swarm-controller/src/routes/agents.rs | 24 +++++-------- swarm-nats-auth/Cargo.toml | 11 +++--- swarm-nats-auth/src/policy.rs | 6 ++-- swarm-queue-client/src/archetypes.rs | 2 +- swarm-queue-client/src/lib.rs | 6 ++-- swarm-queue-client/src/notices.rs | 52 +++------------------------ swarm-wire/src/lib.rs | 3 +- swarm-wire/src/notices.rs | 37 +++++++++++++++++++ 12 files changed, 72 insertions(+), 86 deletions(-) create mode 100644 swarm-wire/src/notices.rs diff --git a/hive-agent/src/swarm_agent_state.rs b/hive-agent/src/swarm_agent_state.rs index 11344edb..0c41e852 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_wire::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_wire::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/src/swarm_notices.rs b/hive-c0re/src/swarm_notices.rs index ac663341..97c848df 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 the stream this publishes into. +//! [`swarm_queue_client::notices`] for opening 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_queue_client::notices::subject(hive), payload.into()) + js.publish(swarm_wire::notices::subject(hive), payload.into()) .await .context("publishing the notice")? .await diff --git a/swarm-controller/src/agent_renewal.rs b/swarm-controller/src/agent_renewal.rs index 9977718b..bb0f6b5f 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_wire::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; diff --git a/swarm-controller/src/agent_status.rs b/swarm-controller/src/agent_status.rs index e2e8b9c1..62bb4416 100644 --- a/swarm-controller/src/agent_status.rs +++ b/swarm-controller/src/agent_status.rs @@ -141,10 +141,9 @@ impl AgentStatusReader { else { continue; }; - let payload = match serde_json::from_slice::< - swarm_wire::agent_status::AgentStatus, - >(&entry.value) - { + let payload = match serde_json::from_slice::( + &entry.value, + ) { Ok(status) => serde_json::to_value(status).ok(), Err(e) => { tracing::warn!( diff --git a/swarm-controller/src/routes/agents.rs b/swarm-controller/src/routes/agents.rs index 637422b8..c70b62b0 100644 --- a/swarm-controller/src/routes/agents.rs +++ b/swarm-controller/src/routes/agents.rs @@ -78,9 +78,7 @@ pub(crate) struct AgentDeclaration { state: String, } -pub(crate) fn render( - declaration: &swarm_wire::wanted::HiveWanted, -) -> Vec { +pub(crate) fn render(declaration: &swarm_wire::wanted::HiveWanted) -> Vec { declaration .agents .iter() @@ -867,9 +865,7 @@ async fn placements_elsewhere( read_declared: impl FnOnce() -> Fut, ) -> Result, E> where - Fut: std::future::Future< - Output = Result, E>, - >, + Fut: std::future::Future, E>>, { let queued = snapshot_queued(); let declared = read_declared().await?; @@ -886,8 +882,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 .map_err(|(other, e)| { @@ -1709,10 +1704,9 @@ mod tests { 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 }, - ); + declaration + .agents + .insert(agent.to_owned(), swarm_wire::wanted::AgentWanted { state }); (hive.to_owned(), declaration) } @@ -2010,8 +2004,7 @@ 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_wire::wanted::AgentState::Up).await; assert_eq!( err.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), @@ -2029,8 +2022,7 @@ 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_wire::wanted::AgentState::Destroyed).await; assert_eq!( err.status, Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), diff --git a/swarm-nats-auth/Cargo.toml b/swarm-nats-auth/Cargo.toml index 0bf8e78b..d5347a23 100644 --- a/swarm-nats-auth/Cargo.toml +++ b/swarm-nats-auth/Cargo.toml @@ -23,12 +23,11 @@ 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` 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"] } +# 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 # 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 244d5eed..31ea3eae 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -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_queue_client::notices::STREAM` + // NATS stream name *is* `swarm_wire::notices::STREAM` // verbatim — the `KV_` prefix is `create_key_value`'s own // convention, not something every stream carries. - let stream = swarm_queue_client::notices::STREAM; + let stream = swarm_wire::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_queue_client::notices::subject(hive)); + subjects.push(swarm_wire::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* diff --git a/swarm-queue-client/src/archetypes.rs b/swarm-queue-client/src/archetypes.rs index acb9e085..8df55357 100644 --- a/swarm-queue-client/src/archetypes.rs +++ b/swarm-queue-client/src/archetypes.rs @@ -11,8 +11,8 @@ use std::collections::BTreeMap; -use swarm_wire::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 063b3402..63efa26f 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -469,9 +469,9 @@ pub struct AgentInput { pub body: String, } -/// 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. +/// Opening the hive-notices stream. The name and subject are +/// `swarm_wire::notices`. Behind the `notices` feature, same reason `status` +/// is behind `kv`. #[cfg(feature = "notices")] pub mod notices; diff --git a/swarm-queue-client/src/notices.rs b/swarm-queue-client/src/notices.rs index ca1a8b35..98a33fcb 100644 --- a/swarm-queue-client/src/notices.rs +++ b/swarm-queue-client/src/notices.rs @@ -1,54 +1,12 @@ -//! The hive-notices stream: its name, subject shape, and how a hive -//! opens it to publish. +//! 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`. //! -//! 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 +//! Feature-gated (`notices`) so the crate's other consumers do not compile //! against a stream shape they never touch. use crate::Error; - -/// 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}") -} +use swarm_wire::notices::STREAM; /// Open the notices stream, creating it if nothing has yet. /// diff --git a/swarm-wire/src/lib.rs b/swarm-wire/src/lib.rs index adc9713a..e9b1bdf6 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -8,10 +8,11 @@ pub mod agent_icon; pub mod agent_inference; pub mod agent_limits; -pub mod agent_schedules; 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 new file mode 100644 index 00000000..30c4db26 --- /dev/null +++ b/swarm-wire/src/notices.rs @@ -0,0 +1,37 @@ +//! 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}") +}