hive-c0re: converge when the controller republishes, not only at boot

The wanted-state read was a boot-time DAG node, so a swarm-level change sat
unapplied until the next restart. This watches the hive's own bucket and
converges on each update.

It does not replace the boot read: a watch hears only what is published while
it is listening, so a hive that was down still learns the current declaration
from `pull`. The watch is the fast path, `pull` stays the repair path.

Rides the connection swarm-status already opens, as a third consumer — a
second connect would double the auth-callout traffic and give the two paths
independent reconnect state, which is the reason the deploy-event drain is
spawned there too.

A delete is not a deletion order. `carries_a_declaration` is pure and tested
so that rule is enforced rather than asserted: converging on a withdrawn key
would tear down exactly the agents "absence is not a deletion order" protects.

hive-c0re/src/workers/wanted.rs

@@ -81,6 +81,105 @@ pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
     converge(coord, &declared).await
 }
 
+/// Converge again each time the controller republishes this hive's declaration.
+///
+/// The **fast** path, and it does not replace [`pull`]: this hears only what is
+/// published while it is listening, so a hive that was down still learns the
+/// current declaration from the boot read. Both, not either.
+///
+/// ⚠️ A refused watch and a quiet one are told apart here, unlike the core-NATS
+/// subscriptions in [`crate::swarm_status`]: a watch is a `JetStream` consumer,
+/// so the request is answered, and a hive lacking the grant gets `None` rather
+/// than silence. That is why this warns instead of returning quietly.
+pub async fn watch_declarations(
+    client: async_nats::Client,
+    coord: Arc<Coordinator>,
+    hive: String,
+    mut shutdown: tokio::sync::watch::Receiver<bool>,
+) {
+    use futures_util::StreamExt as _;
+
+    let Some(mut updates) = swarm_queue_client::wanted::watch(&client, &hive).await else {
+        tracing::warn!(
+            %hive,
+            "wanted state: cannot watch this hive's bucket; changes will be \
+             picked up at the next boot instead"
+        );
+        return;
+    };
+    tracing::info!(%hive, "wanted state: watching for declarations");
+
+    loop {
+        tokio::select! {
+            entry = updates.next() => {
+                let Some(entry) = entry else {
+                    // Same reading as the sibling subscriptions: `async-nats`
+                    // reconnects underneath a live watch, so an ended stream is
+                    // the connection going away for good rather than a blip to
+                    // spin on.
+                    tracing::warn!(%hive, "wanted state: watch closed");
+                    return;
+                };
+                match entry {
+                    Ok(entry) => apply_entry(&coord, &hive, &entry).await,
+                    // The watch survives one bad entry; the stream ending is
+                    // the case above.
+                    Err(e) => tracing::warn!(%hive, error = %e, "wanted state: watch error"),
+                }
+            }
+            _ = shutdown.changed() => {
+                tracing::info!("wanted state: shutdown signal received");
+                return;
+            }
+        }
+    }
+}
+
+/// Converge one watched update, or decline to.
+///
+/// A delete is **not** a deletion order — the module docs' rule, and the reason
+/// this is not simply "decode and converge": the controller removing the key
+/// says nothing about the agents this hive runs, so acting on it would tear
+/// down the very set that absence is defined not to touch.
+async fn apply_entry(
+    coord: &Arc<Coordinator>,
+    hive: &str,
+    entry: &async_nats::jetstream::kv::Entry,
+) {
+    if !carries_a_declaration(entry.operation) {
+        tracing::info!(%hive, "wanted state: declaration withdrawn; nothing to converge");
+        return;
+    }
+    let declared: HiveWanted = match serde_json::from_slice(&entry.value) {
+        Ok(declared) => declared,
+        // Warned rather than propagated: this end and the controller share one
+        // type, so a decode failure means they disagree about it — and the
+        // watch must keep running to pick up the next, possibly good, value.
+        Err(e) => {
+            tracing::warn!(%hive, error = %e, "wanted state: undecodable declaration");
+            return;
+        }
+    };
+    if let Err(e) = converge(coord, &declared).await {
+        tracing::warn!(%hive, error = ?e, "wanted state: converging a watched update failed");
+    }
+}
+
+/// Whether a watched operation carries a declaration to converge to.
+///
+/// Pure, and separate from [`apply_entry`], so the module's "absence is not a
+/// deletion order" rule is enforced by a test rather than only asserted in
+/// prose — converging on a removed key is the one mistake here that would tear
+/// down agents nobody asked to stop.
+fn carries_a_declaration(operation: async_nats::jetstream::kv::Operation) -> bool {
+    use async_nats::jetstream::kv::Operation;
+
+    match operation {
+        Operation::Put => true,
+        Operation::Delete | Operation::Purge => false,
+    }
+}
+
 /// Queue whatever the declaration asks for and this hive is not already doing.
 async fn converge(coord: &Arc<Coordinator>, declared: &HiveWanted) -> Result<()> {
     // Fail closed, for the reason the deploy event's own arm gives: without
@@ -223,7 +322,7 @@ fn decide(state: AgentState, present: bool, intent: Option<Wanted>) -> Converge
 mod tests {
     use std::collections::{BTreeMap, BTreeSet};
 
-    use super::{Converge, Plan, decide, plan};
+    use super::{Converge, Plan, carries_a_declaration, decide, plan};
     use crate::power::Wanted;
     use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted};
 
@@ -333,6 +432,22 @@ mod tests {
         assert_eq!(decide(AgentState::Offline, false, None), Converge::Nothing);
     }
 
+    /// The watch's half of "absence is not a deletion order". `Put` is the
+    /// control: without it this would pass on a function that refused
+    /// everything, which would silently stop the fast path converging at all.
+    #[test]
+    fn only_a_put_carries_a_declaration_to_converge_to() {
+        use async_nats::jetstream::kv::Operation;
+
+        assert!(carries_a_declaration(Operation::Put));
+        for withdrawn in [Operation::Delete, Operation::Purge] {
+            assert!(
+                !carries_a_declaration(withdrawn),
+                "{withdrawn:?} must not converge"
+            );
+        }
+    }
+
     /// Version skew is handled one layer up, at the decode: `AgentState` is
     /// closed, so an unknown value never reaches [`decide`] — it fails the
     /// whole declaration in `swarm-queue-client`, which owns that test