swarm UI: send a message to an agent, without its hive

The agent view in the swarm UI gets a message box. The text reaches the
agent's harness straight from swarm-controller over the swarm queue and
becomes an inbox turn from the operator's authelia username. hive-c0re is
not in the path.

- gateway: `/api/` forwards authelia's `Remote-User`, replacing any the
  client sent.
- swarm-controller: `POST /api/agents/{name}/input` takes that header as
  the sender (401 without it) and sends `{from, body}` as a NATS request
  on `$SWARM.input.<agent>`. No subscriber is a 503.
- swarm-nats-auth: the controller is granted publish on `$SWARM.input.*`;
  no hive or agent grant names the family, so the harness can trust
  `from`.
- hive-agent: a task subscribes to the agent's own input subject and
  pushes each message into the serve loop's existing reminder channel,
  renamed `local_msg_*` since it now carries both.

The harness does not reply to the request: no agent grant covers a reply
subject. The controller therefore answers 202 once the request times out
(2s) and 503 on the server's no-responders status.

Closes #4828

hive-agent/src/main.rs

@@ -30,6 +30,7 @@ mod state_entry_watch;
 mod stats;
 mod swarm_agent_icon;
 mod swarm_agent_state;
+mod swarm_input;
 mod swarm_queue;
 mod swarm_term;
 mod term_msg;
@@ -595,7 +596,10 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
     ) {
         tracing::warn!(error = ?e, "legacy db_migrate failed — continuing with fresh/partial state db");
     }
-    let (reminder_tx, reminder_rx) = tokio::sync::mpsc::unbounded_channel();
+    // Messages made inside this container rather than by the broker:
+    // reminders and the swarm UI's operator input.
+    let (local_msg_tx, local_msg_rx) = tokio::sync::mpsc::unbounded_channel();
+    swarm_input::spawn(local_msg_tx.clone());
     let reminder_store = match reminders::Reminders::open(&paths::state_db()) {
         Ok(store) => Some(Arc::new(store)),
         Err(e) => {
@@ -603,7 +607,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
             None
         }
     };
-    tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx));
+    tokio::spawn(reminder_timer::run(reminder_store.clone(), local_msg_tx));
     let (todo_wake, todos_store) = spawn_todo_socket(reminder_store.clone(), &bus);
     if matches!(initial, LoginState::NeedsLogin) {
         login::wait_for_login(
@@ -630,7 +634,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         &files,
         todo_wake,
         todos_store,
-        reminder_rx,
+        local_msg_rx,
         interrupted,
     )
     .await
@@ -665,7 +669,7 @@ async fn serve_loop<S: Surface>(
     files: &turn::TurnFiles,
     todo_wake: Arc<tokio::sync::Notify>,
     todos_store: Option<Arc<todos::Todos>>,
-    mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::inbox::DeliveredMessage>,
+    mut local_msg_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::inbox::DeliveredMessage>,
     interrupted: Arc<std::sync::atomic::AtomicBool>,
 ) -> Result<()> {
     tracing::info!(socket = %socket.display(), "harness serve");
@@ -690,7 +694,7 @@ async fn serve_loop<S: Surface>(
         // Nothing here touches the broker: not calling `S::recv_next` is
         // exactly the "messages queue unacked, resume drains the
         // backlog" semantic, with no fencing and nothing to requeue.
-        // Reminders (unbounded channel) and todo wakes (a `Notify`
+        // Local messages (unbounded channel) and todo wakes (a `Notify`
         // permit) buffer on their own. The web UI and MCP daemons run as
         // separate tasks, so the agent stays inspectable while parked.
         //
@@ -733,7 +737,7 @@ async fn serve_loop<S: Surface>(
                 biased;
                 o = S::recv_next(socket) => o,
                 () = todo_wake.notified() => RecvOutcome::LocalTodo,
-                Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm),
+                Some(dm) = local_msg_rx.recv() => RecvOutcome::Message(dm),
             }
         } {
             RecvOutcome::Message(first) => first,

hive-agent/src/swarm_input.rs

@@ -0,0 +1,144 @@
+//! Operator input from the swarm UI, taken straight off the swarm queue.
+//!
+//! swarm-controller sends each message on
+//! [`swarm_queue_client::agent_input_subject`] for this agent. This task
+//! subscribes to it and pushes each message into the serve loop's local
+//! message channel, where it becomes a turn from the operator it names.
+//! hive-c0re is not in the path, so the message has no broker row: the hive
+//! dashboard's history and `recv` never see it, and `send` has no recipient
+//! by the sender's name.
+//!
+//! **`from` is taken as given.** The controller is the only client granted
+//! publish on the subject (`swarm-nats-auth`'s reader grant), and it names
+//! the sender from the swarm gateway's authelia answer.
+//!
+//! **Core NATS, no reply.** A message sent while this task is not subscribed
+//! is gone; the controller sends it as a request so the operator gets a 503
+//! for that instead. No agent grant covers a reply subject, so nothing here
+//! answers the request.
+
+use futures_util::{Stream, StreamExt as _};
+use hive_sh4re::inbox::DeliveredMessage;
+use tokio::sync::mpsc::UnboundedSender;
+
+use crate::swarm_queue::Connection;
+
+/// Start the subscriber, if this agent has a queue credential and a label to
+/// name its subject with.
+pub fn spawn(tx: UnboundedSender<DeliveredMessage>) {
+    if !crate::swarm_queue::configured() {
+        return;
+    }
+    let agent = crate::identity::label();
+    if agent.is_empty() {
+        tracing::warn!("this agent has no label; not taking input from the swarm queue");
+        return;
+    }
+    tokio::spawn(run(tx, agent));
+}
+
+async fn run(tx: UnboundedSender<DeliveredMessage>, agent: String) {
+    let Some(Connection { client, .. }) = crate::swarm_queue::client().await else {
+        return;
+    };
+    let subject = swarm_queue_client::agent_input_subject(&agent);
+    let subscriber = match client.subscribe(subject.clone()).await {
+        Ok(subscriber) => subscriber,
+        Err(e) => {
+            tracing::warn!(subject, error = %e, "swarm input: subscribe failed");
+            return;
+        }
+    };
+    tracing::info!(subject, "taking operator input from the swarm queue");
+    forward(subscriber.map(|msg| msg.payload), &tx).await;
+}
+
+/// Push each decodable payload into `tx` until either end closes.
+async fn forward<P: AsRef<[u8]>>(
+    payloads: impl Stream<Item = P>,
+    tx: &UnboundedSender<DeliveredMessage>,
+) {
+    let mut payloads = std::pin::pin!(payloads);
+    while let Some(payload) = payloads.next().await {
+        let Some(dm) = delivered(payload.as_ref()) else {
+            continue;
+        };
+        if tx.send(dm).is_err() {
+            // The serve loop is gone: the harness is shutting down.
+            return;
+        }
+    }
+}
+
+/// The message a payload becomes, or `None` (logged) for one that is not an
+/// [`swarm_queue_client::AgentInput`].
+fn delivered(payload: &[u8]) -> Option<DeliveredMessage> {
+    match serde_json::from_slice::<swarm_queue_client::AgentInput>(payload) {
+        Ok(input) => {
+            tracing::info!(from = %input.from, "swarm input: message received");
+            Some(DeliveredMessage {
+                from: input.from,
+                body: input.body,
+                id: 0,
+                redelivered: false,
+                in_reply_to: None,
+            })
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "swarm input: undecodable message dropped");
+            None
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::forward;
+    use swarm_queue_client::AgentInput;
+
+    fn payload(from: &str, body: &str) -> Vec<u8> {
+        serde_json::to_vec(&AgentInput {
+            from: from.to_owned(),
+            body: body.to_owned(),
+        })
+        .expect("encodes")
+    }
+
+    #[tokio::test]
+    async fn each_message_reaches_the_serve_loop_as_from_its_sender() {
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let payloads = vec![
+            payload("mara", "first"),
+            b"not json".to_vec(),
+            payload("argus", "second"),
+        ];
+        forward(futures_util::stream::iter(payloads), &tx).await;
+        drop(tx);
+
+        let mut got = Vec::new();
+        while let Some(dm) = rx.recv().await {
+            got.push(dm);
+        }
+        // The undecodable one is dropped, the ones either side of it are not.
+        assert_eq!(got.len(), 2, "{got:?}");
+        assert_eq!(
+            (got[0].from.as_str(), got[0].body.as_str()),
+            ("mara", "first")
+        );
+        assert_eq!(
+            (got[1].from.as_str(), got[1].body.as_str()),
+            ("argus", "second")
+        );
+        // No broker row behind it, so no `[msg #id]` tag in the prompt.
+        assert!(got.iter().all(|dm| dm.id == 0 && dm.in_reply_to.is_none()));
+    }
+
+    #[tokio::test]
+    async fn a_closed_serve_loop_stops_the_forwarder() {
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+        drop(rx);
+        // An endless stream: returning at all proves the closed channel ended it.
+        let payloads = futures_util::stream::repeat(payload("mara", "hi"));
+        forward(payloads, &tx).await;
+    }
+}