feat(#3255): hives subscribe to their own knowledge event

A hive learned the knowledge repository had changed only by registering
its own forge webhook. This subscribes to the per-hive subject the
controller now publishes on and calls the pull this daemon already runs
at boot.

Shares the hive's ONE queue connection rather than opening a second: a
second connect would double the auth-callout traffic against authelia and
give the two paths independent reconnect state, so one could be serving
while the other was still down. Same argument as the controller side.

No payload is read, because there is none to read — the webhook handler
this replaces took two fields from Forgejo and used neither, then ran
`git pull`, which re-derives everything from the repository.

At-most-once, and that is not a regression: a webhook delivery to a hive
that is down is lost identically today, and the boot pull covers it.
JetStream would require this end to publish to
`$JS.API.CONSUMER.CREATE.<stream>`, which the callout policy does not
grant, so durability would cost grants on both sides to remove a failure
the boot pull already handles.

⚠️ Documented at the call site rather than left implicit: a refused
subscription is indistinguishable from a quiet one, because NATS reports
authorization violations asynchronously on the connection. If hives stop
hearing events, the server log is the thing that knows.

futures-util comes from the workspace (same version swarm-controller
already uses), not a new dependency version.

Cargo.lock

@@ -1673,6 +1673,7 @@ dependencies = [
  "clap-markdown",
  "clap_complete",
  "forgejo-api",
+ "futures-util",
  "hive-agent-sock",
  "hive-core-agent-sock",
  "hive-host-sock",

hive-c0re/Cargo.toml

@@ -8,6 +8,9 @@ readme = "README.md"
 workspace = true
 
 [dependencies]
+# For `StreamExt::next` on the swarm-event subscription in `swarm_status`.
+# Workspace-level, same version swarm-controller already uses — not a second copy.
+futures-util.workspace = true
 anyhow.workspace = true
 # Named directly only for the client type the swarm status publisher passes
 # around; the connect itself lives in `swarm-queue-client` below.

hive-c0re/src/main.rs

@@ -483,7 +483,7 @@ async fn cmd_serve(
     // A no-op on a standalone hive (no queue env, logged once) — see
     // swarm_status, which owns the whole task including its own decision
     // not to start.
-    swarm_status::spawn(coord.shutdown_rx());
+    swarm_status::spawn(std::sync::Arc::clone(&coord), coord.shutdown_rx());
     // Per-agent events.sqlite + bash-tasks file cleanup now runs
     // agent-side in the harness (`hive_agent::vacuum`): the files are
     // agent-owned, so host-side deletes hit PermissionDenied / readonly-db

hive-c0re/src/swarm_status.rs

@@ -31,6 +31,9 @@
 use std::time::Duration;
 
 use anyhow::{Context, Result};
+// `Subscriber` is a `Stream`, so reading the next event needs the extension
+// trait — there is no inherent `next()` on it.
+use futures_util::StreamExt as _;
 
 use crate::stats::sweep_health::{self, SweepHealth};
 
@@ -62,7 +65,10 @@ const FAILURES_BEFORE_BANNER: u32 = 3;
 /// makes it a hard error; it is bannered here rather than swallowed,
 /// because the failure it otherwise produces is a hive that looks fine
 /// and silently never reports.
-pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
+pub fn spawn(
+    coord: std::sync::Arc<crate::coordinator::Coordinator>,
+    mut shutdown: tokio::sync::watch::Receiver<bool>,
+) {
     let cfg = match swarm_queue_client::QueueConfig::from_env(ENV_PREFIX) {
         Ok(Some(cfg)) => cfg,
         Ok(None) => {
@@ -123,6 +129,19 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
             }
         };
 
+        // The hive's ONE queue connection, now serving both directions:
+        // status goes up, swarm events come down. A second `connect` would
+        // double the auth-callout traffic against authelia and give the two
+        // 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(),
+            hive.clone(),
+            coord,
+            shutdown.clone(),
+        ));
+
         let mut health = SweepHealth::new("swarm_status_publish", "warn", FAILURES_BEFORE_BANNER);
         loop {
             match publish(&client, &hive).await {
@@ -160,6 +179,82 @@ pub fn spawn(mut shutdown: tokio::sync::watch::Receiver<bool>) {
     });
 }
 
+/// Listen on this hive's swarm-event subject 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
+/// **the knowledge repository changed**, and the response is the pull this
+/// daemon already runs at boot.
+///
+/// # There is no payload, and that is deliberate
+///
+/// The event carries nothing. The webhook handler this replaces read two
+/// 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: a hive that is down when the
+/// controller publishes never hears it, and its knowledge stays as of its last
+/// pull until it next boots. **That is not a regression** — a webhook delivery
+/// to a hive that is down is lost identically, and this daemon pulls at startup
+/// regardless. `JetStream` would require this end to *publish* to
+/// `$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
+/// changes, the server log is the thing that knows why — nothing here will say.
+async fn drain_swarm_events(
+    client: async_nats::Client,
+    hive: String,
+    coord: std::sync::Arc<crate::coordinator::Coordinator>,
+    mut shutdown: tokio::sync::watch::Receiver<bool>,
+) {
+    let subject = swarm_queue_client::events::knowledge(&hive);
+    let mut sub = match client.subscribe(subject.clone()).await {
+        Ok(sub) => sub,
+        Err(e) => {
+            // Warn rather than a boot banner: the hive is fully functional
+            // without this, it just falls back to learning about knowledge
+            // changes at its next boot.
+            tracing::warn!(
+                %subject, error = %e,
+                "swarm events: subscribe failed; this hive will not hear knowledge changes"
+            );
+            return;
+        }
+    };
+    tracing::info!(%subject, "swarm events: listening");
+
+    loop {
+        tokio::select! {
+            msg = sub.next() => {
+                if msg.is_none() {
+                    // The subscription ended — the connection went away for
+                    // good. Returning is right: `async-nats` reconnects
+                    // underneath a live subscription, so a closed stream is
+                    // not a blip this should spin on.
+                    tracing::warn!(%subject, "swarm events: subscription closed");
+                    return;
+                }
+                tracing::info!(%subject, "swarm events: knowledge change announced, pulling");
+                if let Err(e) = crate::workers::knowledge::pull(&coord).await {
+                    tracing::warn!(error = ?e, "swarm events: knowledge pull failed");
+                }
+            }
+            _ = shutdown.changed() => {
+                tracing::info!("swarm events: shutdown signal received");
+                return;
+            }
+        }
+    }
+}
+
 /// 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