hive-c0re: act on a deploy request addressed to this hive

The second subject on the connection this task already drains. Every
hive receives every message — that is what a swarm-wide subject buys —
so not being the addressee is the ordinary case and logs at `debug`. A
payload that will not decode is not: both ends share one type, so a
decode failure means they disagree about it.

The rebuild is the same insert the operator's own `rebuild` verb makes,
relock and all. "Deploy this agent" already means something here, and a
swarm-triggered rebuild that quietly did something narrower would be a
second definition of the word.

⚠️ The at-most-once argument in this function's docs does NOT transfer
to the new subject, and the docs now say so. A missed knowledge event is
repaired by the pull this daemon does at startup regardless; a missed
deploy event has no second path — nothing else would ever tell this hive
to build that agent. Closing that is the hive-side reconcile loop the
issue's other half calls for; until it exists this is a nudge with no
safety net. Not papered over with `JetStream` here: durability on one
subject would look like a fix while the desired state still lived only
in a message.

Swept the surrounding prose rather than only the lines I touched. Two
sections had gone quietly false: the summary said this listens on "the
knowledge-event subject" and named one event, and a whole section
argued "there is no payload, and that is deliberate" — true of the
knowledge event and the opposite of true for a deploy request, which is
addressed and has nowhere but the body to say so.

hive-c0re/src/swarm_status.rs

@@ -88,7 +88,12 @@ pub fn spawn(
         // paths independent reconnect state, so one could be serving while
         // the other was still down. `async_nats::Client` is a handle, so the
         // clone is cheap.
-        tokio::spawn(drain_swarm_events(client.clone(), coord, shutdown.clone()));
+        tokio::spawn(drain_swarm_events(
+            client.clone(),
+            coord,
+            hive.clone(),
+            shutdown.clone(),
+        ));
 
         let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER);
         loop {
@@ -127,39 +132,40 @@ pub fn spawn(
     });
 }
 
-/// Listen on the swarm's knowledge-event subject and act on what arrives.
+/// Listen on the swarm's event subjects and act on what arrives.
 ///
 /// The controller decides *what a forge delivery means* and addresses the
-/// result here; this end does not know a forge exists. Today the one event is
+/// result here; this end does not know a forge exists. Two events today:
-/// **the knowledge repository changed**, and the response is the pull this
+/// **the knowledge repository changed**, answered by the pull this daemon
-/// daemon already runs at boot.
+/// already runs at boot, and **deploy this agent**, answered by the same
+/// rebuild insert the operator's own verb makes.
 ///
-/// # There is no payload, and that is deliberate
+/// Only the deploy event carries a payload, because only it is *addressed*:
+/// a swarm-wide subject leaves the addressing nowhere to live but the body.
+/// Even then it is a trigger — which hive, which agent — never the config,
+/// which git owns.
 ///
-/// The event carries nothing. The webhook handler this replaces read two
+/// # A missed message costs the two events very differently
-/// fields from Forgejo and used neither — both were filters — then ran
-/// `git pull`, which re-derives everything from the repository. So it is an
-/// edge trigger, and reading a body here would invent a contract nobody owes.
 ///
-/// # What a missed message costs
+/// Core NATS, so delivery is at-most-once. Harmless for knowledge: this
+/// daemon pulls at startup regardless, and the webhook it replaced was lost
+/// identically when a hive was down.
 ///
-/// Core NATS, so delivery is at-most-once: a hive that is down when the
+/// ⚠️ **The deploy event has no such second path** — nothing else would ever
-/// controller publishes never hears it, and its knowledge stays as of its last
+/// tell this hive to build that agent. Its backstop is the hive-side reconcile
-/// pull until it next boots. **That is not a regression** — a webhook delivery
+/// loop ("hives pull and self-update"); until that exists this is a nudge with
-/// to a hive that is down is lost identically, and this daemon pulls at startup
+/// no safety net. Not papered over with `JetStream`: durability on one subject
-/// regardless. `JetStream` would require this end to *publish* to
+/// looks like a fix while the desired state still lives only in a message.
-/// `$JS.API.CONSUMER.CREATE.<stream>`, which the callout policy does not grant,
-/// so durability would cost a grant on both sides to remove a failure the boot
-/// pull already covers.
 ///
 /// ⚠️ **A refused subscription is indistinguishable from a quiet one.** NATS
 /// reports an authorization violation asynchronously on the connection, not as
 /// an error from `subscribe`, so this task cannot tell "no events published"
-/// from "not allowed to hear them". If a hive stops picking up knowledge
+/// from "not allowed to hear them". If a hive stops picking up either event,
-/// changes, the server log is the thing that knows why — nothing here will say.
+/// the server log is the thing that knows why — nothing here will say.
 async fn drain_swarm_events(
     client: async_nats::Client,
     coord: std::sync::Arc<crate::coordinator::Coordinator>,
+    hive: String,
     mut shutdown: tokio::sync::watch::Receiver<bool>,
 ) {
     // One subject for the whole swarm, so this hive's own name never enters
@@ -179,7 +185,18 @@ async fn drain_swarm_events(
             return;
         }
     };
-    tracing::info!(%subject, "swarm events: listening");
+    let deploy_subject = swarm_queue_client::DEPLOY_SUBJECT;
+    let mut deploy_sub = match client.subscribe(deploy_subject).await {
+        Ok(sub) => sub,
+        Err(e) => {
+            tracing::warn!(
+                subject = %deploy_subject, error = %e,
+                "swarm events: subscribe failed; this hive will not hear deploy requests"
+            );
+            return;
+        }
+    };
+    tracing::info!(%subject, %deploy_subject, "swarm events: listening");
 
     loop {
         tokio::select! {
@@ -197,6 +214,13 @@ async fn drain_swarm_events(
                     tracing::warn!(error = ?e, "swarm events: knowledge pull failed");
                 }
             }
+            msg = deploy_sub.next() => {
+                let Some(msg) = msg else {
+                    tracing::warn!(subject = %deploy_subject, "swarm events: deploy subscription closed");
+                    return;
+                };
+                handle_deploy_request(&coord, &hive, &msg.payload);
+            }
             _ = shutdown.changed() => {
                 tracing::info!("swarm events: shutdown signal received");
                 return;
@@ -205,6 +229,46 @@ async fn drain_swarm_events(
     }
 }
 
+/// Act on one deploy request, if it is addressed to this hive.
+///
+/// Every hive receives every message — that is what the swarm-wide subject
+/// buys — so *not* being the addressee is the ordinary case and logs at
+/// `debug`, not `warn`. A malformed payload is different: the controller and
+/// this end share one type, so a decode failure means they disagree about it,
+/// and that is worth saying out loud.
+fn handle_deploy_request(
+    coord: &std::sync::Arc<crate::coordinator::Coordinator>,
+    hive: &str,
+    payload: &[u8],
+) {
+    let request: swarm_queue_client::DeployRequest = match serde_json::from_slice(payload) {
+        Ok(request) => request,
+        Err(e) => {
+            tracing::warn!(error = %e, "swarm events: undecodable deploy request");
+            return;
+        }
+    };
+    if request.hive != hive {
+        tracing::debug!(target_hive = %request.hive, agent = %request.agent, "swarm events: deploy for another hive");
+        return;
+    }
+    // The same insert the operator's own `rebuild` verb makes, relock and
+    // all: "deploy this agent" means here exactly what it already meant, and
+    // a swarm-triggered rebuild that quietly did something narrower would be
+    // a second definition of the word.
+    let agent = request.agent;
+    let inserted = coord
+        .job_queue
+        .insert_job(|b| crate::job_queue::templates::rebuild(b, &agent, true));
+    match inserted {
+        Ok(_) => {
+            tracing::info!(%agent, "swarm events: deploy requested by the swarm, rebuild queued");
+            coord.emit_rebuild_queue_snapshot();
+        }
+        Err(e) => tracing::warn!(%agent, error = %e, "swarm events: queueing the deploy failed"),
+    }
+}
+
 /// Offer one snapshot: this hive's current readiness, under its own key.
 async fn publish(client: &async_nats::Client, hive: &str) -> Result<()> {
     // An unconnected client does not fail a JetStream request, it hangs