feat(#3124): converge the hive onto the agent set the swarm declares

The deploy event is a nudge with no second path: core NATS is
at-most-once, so a hive that was down when the controller published
simply never learns that an agent is meant to exist here. This adds the
repair path — one boot-time DAG node that reads this hive's own key in
the `hive-wanted` bucket and converges the agents it names.

Two semantics settled on the issue thread, and both are places where a
plausible implementation is the wrong one:

- **Absence is not a deletion order.** No bucket, no key, or an agent
  the value does not name all mean the controller has said nothing.
  Swarm-side lifecycle does not yet cover agents that predate it, so
  "converge to exactly this set" would tear down every agent the swarm
  has not adopted. `plan` only ever inspects the agents a declaration
  names.
- **An unrecognised state is inert.** `AgentState` is an open enum: a
  value this build cannot read deserialises into `Unrecognised` and is
  left alone. A closed enum would force "not `Up`" onto a state like
  `paused`, so a controller that learned a new value would take agents
  down on every hive not yet updated.

Divergence is measured against the hive's **stored power intent**, not
the container's observed running state — an agent that is down while its
intent says `Up` is already the boot reconcile's work, and a loop reading
`is_running` would insert a start DAG behind that reconcile's back on
every boot. A hive that already agrees with its declaration queues
nothing at all.

`queue_first_deploy` is extracted from the deploy-event path rather than
open-coded here, for the power-intent seed: without it `first_deploy`'s
tail `Reconcile` seeds `Wanted` from a container that exists but has not
started yet, which locks the agent to `Offline` on its first reconcile.

The read is authorised as-is: `store.get` takes async-nats' direct-get
arm (the KV bucket is created with `allow_direct`), which is exactly the
`$JS.API.DIRECT.GET.KV_hive-wanted.$KV.hive-wanted.<hive>` subject
`swarm-nats-auth` grants a hive. The fallback subject is not granted, and
a refused NATS request surfaces as a timeout rather than an error.

Nothing writes the bucket yet — the controller-side writer is the other
half of #3124, so this does not close it.

swarm-queue-client/src/wanted.rs

@@ -32,6 +32,51 @@ use crate::Error;
 /// for two deployments to disagree about which one that is.
 pub const BUCKET: &str = "hive-wanted";
 
+/// 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 — an **open** enum.
+///
+/// A value this build does not know deserialises into
+/// [`AgentState::Unrecognised`], carried verbatim, rather than failing the
+/// whole document or collapsing into a known variant. Both alternatives break
+/// the same way: a closed enum forces "not `Up`" onto a state like `paused`,
+/// so a controller that learns a new value would take agents down on every
+/// hive that has not been updated yet.
+#[derive(Debug, Clone, 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,
+    /// Anything else. Kept as written so a hive can say what it declined to
+    /// act on instead of logging that something unspecified was skipped.
+    #[serde(untagged)]
+    Unrecognised(String),
+}
+
 /// Open the wanted-state bucket for writing, creating it if nothing has yet.
 ///
 /// **Controller-side only.** `history: 1` because a hive converges to the
@@ -78,3 +123,50 @@ pub async fn open_read_only(
     let js = async_nats::jetstream::new(client.clone());
     js.get_key_value(BUCKET).await.ok()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{AgentState, HiveWanted};
+
+    #[test]
+    fn known_states_decode_and_an_unknown_one_stays_itself() {
+        let doc = r#"{"agents":{
+            "a":{"state":"up"},
+            "b":{"state":"offline"},
+            "c":{"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::Unrecognised("paused".to_owned())
+        );
+    }
+
+    #[test]
+    fn an_unrecognised_state_survives_a_round_trip() {
+        let doc = r#"{"agents":{"c":{"state":"paused"}}}"#;
+        let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes");
+        let reserialised = serde_json::to_string(&decoded).expect("serialises");
+        assert_eq!(reserialised, r#"{"agents":{"c":{"state":"paused"}}}"#);
+    }
+
+    /// A field a newer controller adds must not cost an older hive the whole
+    /// document — that is the version skew the open enum exists for, one level
+    /// up.
+    #[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);
+    }
+
+    /// The control for the two above: openness is about *values and fields*,
+    /// not about accepting anything. 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());
+    }
+}