feat(#3124): publish the agent set the swarm declares for each hive

The hive-side loop landed without anything to converge to: nothing wrote
`$KV.hive-wanted.<hive>`, so in production only the "no key" branch ran.
This is the writer.

`WantedWriter` mirrors `StatusReader` — that module reads what hives report,
this one writes what they are told, so it holds a client rather than a bucket
handle and resolves the store on first use. It shares the status reader's
connection: the controller has exactly one by design, and a second connect
would double the auth-callout traffic and give the two paths independent
reconnect state.

The value under a hive's key is the map of every agent on that hive, so a
plain `put` of a single-agent change would drop a concurrent change to a
different agent, with only one revision of history to not recover from.
Writes are read-modify-write against the entry revision, and only
`WrongLastRevision` / `AlreadyExists` count as a lost race — every other
error returns immediately rather than spinning the retry loop and then
blaming a concurrent writer that never existed.

`apply` is split out and tested because it holds the invariant: declaring
one agent preserves the rest, and a current value that will not decode is an
error rather than a fresh start. Overwriting a document nobody can read
discards every other agent's declaration.

Two routes, no swarmctl verb and no jobq node: `create_agent` needs a graph
because it is multi-step, and one CAS'd write is not.

`build_app` is extracted from `main` in the same change because `main` sat at
exactly the `too_many_lines` limit, so adding an endpoint tripped a lint
about the startup sequence. The route list is the part that grows.

swarm-controller/src/main.rs

@@ -47,6 +47,7 @@ mod issue_report;
 mod otel_http_client;
 mod status;
 mod vcs_metrics;
+mod wanted;
 mod webhook;
 
 /// Node payload for the swarm-level job graph. Named `Swarm*` rather than
@@ -429,6 +430,12 @@ struct AppState {
     /// that is merely *unreachable* still yields a reader, because
     /// `async-nats` reconnects underneath it.
     status: Option<Arc<status::StatusReader>>,
+    /// Publishes the agent set this swarm declares for each hive.
+    ///
+    /// `None` in exactly the state `status` is: no swarm queue was wired
+    /// up, so there is nowhere to publish a declaration to. Shares that
+    /// reader's connection rather than opening a second one.
+    wanted: Option<Arc<wanted::WantedWriter>>,
     /// The swarm-level job graph, wrapped in its
     /// [`hive_jobq::scheduler::Scheduler`] now that something drives it
     /// (`spawn_jobq_worker`) — the graph alone was enough for the
@@ -681,6 +688,150 @@ fn error_problem(status: axum::http::StatusCode, detail: &str) -> problem_detail
     problem_details::ProblemDetails::from_status_code(status).with_detail(detail)
 }
 
+/// The state to declare for one agent.
+#[derive(Debug, Deserialize, ToSchema)]
+struct SetAgentStateRequest {
+    /// Typed as a string in the schema only — the parse is the real enum, so
+    /// a value this build does not know is a 400 rather than a field that
+    /// silently does nothing.
+    #[schema(value_type = String, example = "up")]
+    state: swarm_queue_client::wanted::AgentState,
+}
+
+/// One agent's line in a hive's declaration.
+#[derive(Clone, Debug, Serialize, ToSchema)]
+struct AgentDeclaration {
+    agent: String,
+    state: String,
+}
+
+fn render(declaration: &swarm_queue_client::wanted::HiveWanted) -> Vec<AgentDeclaration> {
+    declaration
+        .agents
+        .iter()
+        .map(|(agent, wanted)| AgentDeclaration {
+            agent: agent.clone(),
+            state: wanted.state.as_str().to_owned(),
+        })
+        .collect()
+}
+
+/// The declaration writer, sharing the status reader's connection.
+///
+/// The controller holds exactly one queue connection by design — see
+/// `StatusReader::queue_client`. A second connect would double the
+/// auth-callout traffic and give the two paths independent reconnect state,
+/// so one could be serving while the other was still down.
+fn wanted_writer(status: Option<&Arc<status::StatusReader>>) -> Option<Arc<wanted::WantedWriter>> {
+    status.map(|s| Arc::new(wanted::WantedWriter::new(s.queue_client())))
+}
+
+/// Both handlers below take the same hive name and reject it the same way.
+///
+/// Reports the status and the detail rather than a rendered
+/// `ProblemDetails`: that type is 232 bytes, which makes every `Result` in
+/// this path pay for the error case it usually does not take. The handlers
+/// render at the boundary, where the body is actually needed.
+fn declaration_target(
+    state: &AppState,
+    hive: &str,
+) -> Result<(Arc<wanted::WantedWriter>, String), (axum::http::StatusCode, String)> {
+    let writer = state.wanted.clone().ok_or_else(|| {
+        (
+            axum::http::StatusCode::SERVICE_UNAVAILABLE,
+            "this deployment wired up no swarm queue, so there is nowhere to publish a declaration"
+                .to_owned(),
+        )
+    })?;
+    let hive = hive_types::Ident::parse(hive)
+        .map_err(|reason| (axum::http::StatusCode::BAD_REQUEST, reason.to_owned()))?
+        .into_string();
+    // Shaped like a hive name, and actually one. Same two checks
+    // `create_agent` makes, for the same reason: a typo otherwise publishes a
+    // declaration under a key no hive will ever read.
+    if !state.hives.iter().any(|h| h.name == hive) {
+        return Err((
+            axum::http::StatusCode::BAD_REQUEST,
+            format!("{hive:?} is not a hive in this swarm"),
+        ));
+    }
+    Ok((writer, hive))
+}
+
+/// Declare what this swarm wants of one agent on one hive.
+///
+/// The whole declaration is returned as published, because the value is the
+/// hive's entire agent map and a caller that changed one agent still wants to
+/// render the rest.
+#[utoipa::path(
+    put,
+    path = "/api/hives/{hive}/agents/{agent}/state",
+    params(
+        ("hive" = String, Path, description = "hive whose declaration this is"),
+        ("agent" = String, Path, description = "agent to declare"),
+    ),
+    request_body = SetAgentStateRequest,
+    responses(
+        (status = 200, description = "the declaration as now published", body = Vec<AgentDeclaration>),
+        (status = 400, description = "a name is not an identifier, the hive is not in this swarm, or the state is unknown (problem+json)", body = String),
+        (status = 503, description = "no swarm queue is wired up (problem+json)", body = String),
+        (status = 500, description = "the declaration could not be published (problem+json)", body = String),
+    ),
+    tag = "agents"
+)]
+async fn set_agent_state(
+    State(state): State<AppState>,
+    axum::extract::Path((hive, agent)): axum::extract::Path<(String, String)>,
+    Json(req): Json<SetAgentStateRequest>,
+) -> Result<Json<Vec<AgentDeclaration>>, problem_details::ProblemDetails> {
+    let (writer, hive) =
+        declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?;
+    let agent = hive_types::Ident::parse(&agent)
+        .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
+        .into_string();
+
+    let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| {
+        tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed");
+        error_problem(
+            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("{e:#}"),
+        )
+    })?;
+    Ok(Json(render(&declaration)))
+}
+
+/// What this swarm currently declares for a hive.
+///
+/// Read back from the bucket rather than from a second copy kept here: the
+/// published value is the record.
+#[utoipa::path(
+    get,
+    path = "/api/hives/{hive}/wanted",
+    params(("hive" = String, Path, description = "hive whose declaration to read")),
+    responses(
+        (status = 200, description = "the declaration, empty when nothing is published yet", body = Vec<AgentDeclaration>),
+        (status = 400, description = "not an identifier, or not a hive in this swarm (problem+json)", body = String),
+        (status = 503, description = "no swarm queue is wired up (problem+json)", body = String),
+        (status = 500, description = "the declaration could not be read (problem+json)", body = String),
+    ),
+    tag = "agents"
+)]
+async fn get_hive_wanted(
+    State(state): State<AppState>,
+    axum::extract::Path(hive): axum::extract::Path<String>,
+) -> Result<Json<Vec<AgentDeclaration>>, problem_details::ProblemDetails> {
+    let (writer, hive) =
+        declaration_target(&state, &hive).map_err(|(s, d)| error_problem(s, &d))?;
+    let declaration = writer.view(&hive).await.map_err(|e| {
+        tracing::warn!(hive = %hive, error = %format!("{e:#}"), "reading the declaration failed");
+        error_problem(
+            axum::http::StatusCode::INTERNAL_SERVER_ERROR,
+            &format!("{e:#}"),
+        )
+    })?;
+    Ok(Json(declaration.as_ref().map(render).unwrap_or_default()))
+}
+
 /// What each hive last said about itself, read from the swarm queue at
 /// request time.
 ///
@@ -1328,6 +1479,7 @@ async fn main() -> Result<()> {
     let state = AppState {
         hives: Arc::new(load_hives()),
         links: Arc::new(load_links()),
+        wanted: wanted_writer(status.as_ref()),
         status,
         jobq,
         webhook_secret,
@@ -1337,6 +1489,19 @@ async fn main() -> Result<()> {
         forge: state_forge,
     };
 
+    let app = build_app(state);
+    axum::serve(listener, app)
+        .await
+        .context("serving swarm-controller")
+}
+
+/// Every route this daemon serves, wired to its state.
+///
+/// Split out of `main` because the route list is the part that grows, and
+/// `main` sat exactly on the `too_many_lines` limit — adding an endpoint
+/// tripped a lint about the startup sequence, which is not where the change
+/// was. A new route now costs one line here and none there.
+fn build_app(state: AppState) -> axum::Router {
     let (router, api) = OpenApiRouter::<AppState>::with_openapi(ApiDoc::openapi())
         .routes(routes!(health))
         .routes(routes!(get_hives))
@@ -1349,6 +1514,8 @@ async fn main() -> Result<()> {
         .routes(routes!(get_config_prs))
         .routes(routes!(create_agent))
         .routes(routes!(get_agents))
+        .routes(routes!(set_agent_state))
+        .routes(routes!(get_hive_wanted))
         .routes(routes!(issue_report::get_repos))
         .routes(routes!(issue_report::get_issue_report_all))
         .routes(routes!(issue_report::get_issue_report))
@@ -1358,15 +1525,12 @@ async fn main() -> Result<()> {
     // the nix store (see the module doc comment above). `api` is
     // `Clone`; each request gets its own owned copy for `Json` to
     // serialize, same as `hive-c0re::dashboard::serve`.
-    let app = router
+    router
         .route(
             "/api/openapi.json",
             get(move || async move { Json(api.clone()) }),
         )
-        .with_state(state);
+        .with_state(state)
-    axum::serve(listener, app)
-        .await
-        .context("serving swarm-controller")
 }
 
 #[cfg(test)]
@@ -1455,6 +1619,9 @@ mod tests {
             }]),
             links: std::sync::Arc::new(Vec::new()),
             status: None,
+            // No queue, for the same reason as `status`: these tests drive
+            // agent creation, which publishes no declaration.
+            wanted: None,
             jobq: std::sync::Arc::clone(&sched),
             webhook_secret: None,
             config_prs: None,

swarm-controller/src/wanted.rs

@@ -0,0 +1,216 @@
+//! Writes the agent set this swarm declares for each hive.
+//!
+//! The mirror of [`crate::status`]: that module reads what hives report,
+//! this one writes what they are told, and both address the same queue.
+//! The lifecycle is deliberately identical — a NATS client rather than a
+//! bucket handle, resolved on first use and cached, so a controller that
+//! starts before the bucket exists picks it up without a restart.
+//!
+//! The bucket is the record. Nothing here keeps a second copy of the
+//! declaration to reconcile against, because the current value can be read
+//! back from the queue whenever it is needed.
+
+use anyhow::{Context, Result};
+use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind};
+use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted};
+
+/// The three outcomes of one write attempt, which the two KV verbs report
+/// through separate error types.
+enum Wrote {
+    Ok,
+    /// Another writer won the race; re-read and re-apply.
+    LostRace,
+    Failed(anyhow::Error),
+}
+
+/// How many times a losing writer re-reads and re-applies before giving up.
+///
+/// A conflict means another writer changed a *different* agent between this
+/// one's read and its write, so a retry re-reads and re-applies onto the
+/// winner. Bounded because an unbounded loop against a hot key is a spin,
+/// and a caller that gets an error can ask again with fresh intent.
+const MAX_ATTEMPTS: usize = 5;
+
+/// Apply one agent's declared state to a hive's current declaration.
+///
+/// Split out from the write loop because it holds the invariant that matters:
+/// the value under a hive's key is the map of **every** agent on that hive,
+/// so declaring one agent must preserve the rest.
+///
+/// `current` is `None` when the hive has no declaration yet. An undecodable
+/// value is an **error**, never treated as absent: overwriting a document
+/// nobody can read discards the declarations of every other agent on that
+/// hive, which is exactly what a fresh-start fallback would do quietly.
+fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(HiveWanted, Vec<u8>)> {
+    let mut declaration = match current {
+        Some(raw) => serde_json::from_slice::<HiveWanted>(raw)
+            .context("the hive's current declaration is not decodable")?,
+        None => HiveWanted::default(),
+    };
+    declaration
+        .agents
+        .insert(agent.to_owned(), AgentWanted { state });
+    let encoded = serde_json::to_vec(&declaration).context("encoding the new declaration")?;
+    Ok((declaration, encoded))
+}
+
+/// Writes the wanted-state bucket, and reads it back.
+pub struct WantedWriter {
+    client: async_nats::Client,
+    store: tokio::sync::OnceCell<async_nats::jetstream::kv::Store>,
+}
+
+impl WantedWriter {
+    #[must_use]
+    pub fn new(client: async_nats::Client) -> Self {
+        Self {
+            client,
+            store: tokio::sync::OnceCell::new(),
+        }
+    }
+
+    /// The bucket handle, created on first use if nothing has made it yet.
+    ///
+    /// Creation lives in [`swarm_queue_client::wanted`] because a bucket is
+    /// described identically by everyone who may create it. Only the
+    /// controller creates this one; a hive opens it read-only.
+    async fn store(
+        &self,
+    ) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> {
+        self.store
+            .get_or_try_init(|| swarm_queue_client::wanted::open_or_create(&self.client))
+            .await
+    }
+
+    /// The declaration currently published for `hive`, or `None`.
+    pub async fn view(&self, hive: &str) -> Result<Option<HiveWanted>> {
+        // An unconnected client does not fail a JetStream request, it hangs
+        // on it — see `swarm_queue_client::ensure_connected`.
+        swarm_queue_client::ensure_connected(&self.client)?;
+        let store = self.store().await?;
+        let Some(entry) = store
+            .entry(hive)
+            .await
+            .with_context(|| format!("reading the declaration for {hive}"))?
+        else {
+            return Ok(None);
+        };
+        serde_json::from_slice(&entry.value)
+            .map(Some)
+            .with_context(|| format!("the declaration for {hive} is not decodable"))
+    }
+
+    /// Declare `agent` on `hive` to be in `state`, and return the whole
+    /// declaration as published.
+    ///
+    /// Read-modify-write against the entry's revision rather than a plain
+    /// `put`: the value is the hive's whole agent map, so a blind write
+    /// would drop a concurrent change to a different agent. The bucket keeps
+    /// one revision of history, so a lost write is not recoverable after the
+    /// fact — the conflict has to be caught here.
+    pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result<HiveWanted> {
+        swarm_queue_client::ensure_connected(&self.client)?;
+        let store = self.store().await?;
+
+        for _ in 0..MAX_ATTEMPTS {
+            let entry = store
+                .entry(hive)
+                .await
+                .with_context(|| format!("reading the declaration for {hive}"))?;
+            let revision = entry.as_ref().map(|e| e.revision);
+            let (declaration, encoded) =
+                apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?;
+
+            // `update` and `create` have separate error types, and only one
+            // variant of each means "someone else got there first". Every
+            // other failure returns immediately: retrying a disconnect or a
+            // permission error would spin the loop and then report a
+            // conflict, blaming a concurrent writer that never existed.
+            let written = match revision {
+                Some(revision) => match store.update(hive, encoded.into(), revision).await {
+                    Ok(_) => Wrote::Ok,
+                    Err(e) if matches!(e.kind(), UpdateErrorKind::WrongLastRevision) => {
+                        Wrote::LostRace
+                    }
+                    Err(e) => Wrote::Failed(anyhow::Error::new(e)),
+                },
+                None => match store.create(hive, encoded.into()).await {
+                    Ok(_) => Wrote::Ok,
+                    Err(e) if matches!(e.kind(), CreateErrorKind::AlreadyExists) => Wrote::LostRace,
+                    Err(e) => Wrote::Failed(anyhow::Error::new(e)),
+                },
+            };
+            match written {
+                Wrote::Ok => {
+                    tracing::info!(hive, agent, ?state, "declared agent state");
+                    return Ok(declaration);
+                }
+                // Re-read and re-apply onto the winner's value, not over it.
+                Wrote::LostRace => {
+                    tracing::debug!(hive, agent, "declaration write lost a race, retrying");
+                }
+                Wrote::Failed(e) => {
+                    return Err(e).with_context(|| format!("declaring {agent} on {hive}"));
+                }
+            }
+        }
+        anyhow::bail!("gave up declaring {agent} on {hive} after {MAX_ATTEMPTS} conflicting writes")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::apply;
+    use swarm_queue_client::wanted::AgentState;
+
+    #[test]
+    fn declaring_one_agent_preserves_every_other() {
+        let current = br#"{"agents":{"iris":{"state":"up"},"argus":{"state":"offline"}}}"#;
+        let (declaration, _) = apply(Some(current), "atlas", AgentState::Up).unwrap();
+        assert_eq!(declaration.agents.len(), 3);
+        assert_eq!(declaration.agents["iris"].state, AgentState::Up);
+        assert_eq!(declaration.agents["argus"].state, AgentState::Offline);
+        assert_eq!(declaration.agents["atlas"].state, AgentState::Up);
+    }
+
+    #[test]
+    fn redeclaring_an_agent_replaces_only_its_own_state() {
+        let current = br#"{"agents":{"iris":{"state":"up"},"atlas":{"state":"up"}}}"#;
+        let (declaration, _) = apply(Some(current), "atlas", AgentState::Offline).unwrap();
+        assert_eq!(declaration.agents.len(), 2);
+        assert_eq!(declaration.agents["iris"].state, AgentState::Up);
+        assert_eq!(declaration.agents["atlas"].state, AgentState::Offline);
+    }
+
+    #[test]
+    fn a_hive_with_no_declaration_yet_gets_a_one_agent_one() {
+        let (declaration, _) = apply(None, "atlas", AgentState::Up).unwrap();
+        assert_eq!(declaration.agents.len(), 1);
+        assert_eq!(declaration.agents["atlas"].state, AgentState::Up);
+    }
+
+    // The failure this function exists to prevent: a fresh-start fallback
+    // here would publish a one-agent document over a hive's whole set.
+    #[test]
+    fn an_undecodable_declaration_is_an_error_not_a_fresh_start() {
+        let err = apply(Some(b"{not json"), "atlas", AgentState::Up).unwrap_err();
+        assert!(
+            err.to_string().contains("not decodable"),
+            "unexpected error: {err}"
+        );
+    }
+
+    #[test]
+    fn an_unknown_state_in_the_current_value_is_also_an_error() {
+        let current = br#"{"agents":{"iris":{"state":"sideways"}}}"#;
+        assert!(apply(Some(current), "atlas", AgentState::Up).is_err());
+    }
+
+    #[test]
+    fn the_encoded_form_round_trips() {
+        let (_, encoded) = apply(None, "atlas", AgentState::Offline).unwrap();
+        let (again, _) = apply(Some(&encoded), "iris", AgentState::Up).unwrap();
+        assert_eq!(again.agents["atlas"].state, AgentState::Offline);
+        assert_eq!(again.agents["iris"].state, AgentState::Up);
+    }
+}

swarm-queue-client/src/wanted.rs

@@ -63,7 +63,7 @@ pub struct AgentWanted {
 /// 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, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[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.
@@ -72,12 +72,26 @@ pub enum AgentState {
     Offline,
 }
 
+impl AgentState {
+    /// The wire spelling, for a reader that renders rather than decodes.
+    ///
+    /// Kept beside the enum so it cannot drift from the `rename_all` above;
+    /// a test pins the two together.
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        match self {
+            AgentState::Up => "up",
+            AgentState::Offline => "offline",
+        }
+    }
+}
+
 /// Open the wanted-state bucket for writing, creating it if nothing has yet.
 ///
 /// **Controller-side only.** `history: 1` because a hive converges to the
-/// current declaration and never asks what the previous one was — that
+/// current declaration and never asks what the previous one was. The
-/// question is answered by the controller's own records, not by replaying a
+/// controller keeps no second copy either: the value published here is the
-/// bucket.
+/// record, and it reads it back from this bucket when it needs it.
 #[cfg(feature = "kv")]
 pub async fn open_or_create(
     client: &async_nats::Client,
@@ -160,4 +174,22 @@ mod tests {
     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];
+        for state in every {
+            match state {
+                AgentState::Up | AgentState::Offline => {}
+            }
+            assert_eq!(
+                serde_json::to_string(&state).expect("serialises"),
+                format!("\"{}\"", state.as_str())
+            );
+        }
+    }
 }