swarm: cancel an agent's turn from the swarm queue

A separate $SWARM.control.<agent> subject carries commands, apart from
the text subject. swarm-controller exposes POST /api/agents/{name}/cancel
as a NATS request; the harness answers whether a turn was actually
stopped, through the same cancel path as the dashboard button. The agent
credential gets subscribe on its control subject and allow_responses to
answer; only the controller may publish there.

Refs #3803

hive-agent/src/main.rs

@@ -566,7 +566,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
     // 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());
+    swarm_input::spawn(local_msg_tx.clone(), bus.clone(), interrupted.clone());
     let reminder_store = match reminders::Reminders::open(&paths::state_db()) {
         Ok(store) => Some(Arc::new(store)),
         Err(e) => {

hive-agent/src/swarm_input.rs

@@ -15,23 +15,90 @@
 //! **Own credential only.** The queue grants subscribe on this subject to the
 //! agent's own credential alone.
 //!
-//! **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.
+//! **Core NATS, no reply to input.** 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. Input is not answered: the controller reads a
+//! request that times out as "a subscriber held it".
+//!
+//! **Commands are a separate subject and are answered.** The swarm UI's
+//! cancel arrives on [`swarm_queue_client::agent_control_subject`], never on
+//! the input subject, so nothing a person types can be read as a command. The
+//! reply says whether a turn was actually stopped; the agent's grant allows
+//! answering only the request it was just sent (`allow_responses`).
 
 use futures_util::{Stream, StreamExt as _};
 use hive_sh4re::inbox::DeliveredMessage;
 use tokio::sync::mpsc::UnboundedSender;
 
-/// Start the subscriber, if this agent has a label to name its subject with.
-pub fn spawn(tx: UnboundedSender<DeliveredMessage>) {
+use crate::events::Bus;
+
+/// Start the subscribers, if this agent has a label to name its subjects with.
+pub fn spawn(
+    tx: UnboundedSender<DeliveredMessage>,
+    bus: Bus,
+    interrupted: std::sync::Arc<std::sync::atomic::AtomicBool>,
+) {
     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));
+    tokio::spawn(run(tx, agent.clone()));
+    tokio::spawn(run_control(agent, bus, interrupted));
+}
+
+/// Answer each control command for `agent` until the subscription ends.
+async fn run_control(
+    agent: String,
+    bus: Bus,
+    interrupted: std::sync::Arc<std::sync::atomic::AtomicBool>,
+) {
+    let client = crate::swarm_queue::client().await;
+    let subject = swarm_queue_client::agent_control_subject(&agent);
+    let mut subscriber = match client.subscribe(subject.clone()).await {
+        Ok(subscriber) => subscriber,
+        Err(e) => {
+            tracing::warn!(subject, error = %e, "swarm control: subscribe failed");
+            return;
+        }
+    };
+    tracing::info!(subject, "taking control commands from the swarm queue");
+    while let Some(msg) = subscriber.next().await {
+        let Some(reply) = msg.reply else {
+            // Sent without a reply subject: nobody to tell, and the
+            // controller always asks for one.
+            tracing::warn!("swarm control: command without a reply subject dropped");
+            continue;
+        };
+        let Some(answer) = act(&msg.payload, |command| match command {
+            swarm_queue_client::AgentControl::Cancel => {
+                crate::web_ui::cancel_turn(&bus, &interrupted, "swarm UI: cancel")
+            }
+        }) else {
+            continue;
+        };
+        let payload = serde_json::to_vec(&answer).expect("a reply encodes");
+        if let Err(e) = client.publish(reply, payload.into()).await {
+            tracing::warn!(error = %e, "swarm control: answering failed");
+        }
+    }
+}
+
+/// The answer to a command payload, running `cancel` for a cancel; `None`
+/// (logged) when the payload is not a command, which gets no answer.
+fn act(
+    payload: &[u8],
+    cancel: impl FnOnce(swarm_queue_client::AgentControl) -> bool,
+) -> Option<swarm_queue_client::AgentControlReply> {
+    match serde_json::from_slice::<swarm_queue_client::AgentControl>(payload) {
+        Ok(command) => Some(swarm_queue_client::AgentControlReply {
+            acted: cancel(command),
+        }),
+        Err(e) => {
+            tracing::warn!(error = %e, "swarm control: undecodable command dropped");
+            None
+        }
+    }
 }
 
 async fn run(tx: UnboundedSender<DeliveredMessage>, agent: String) {
@@ -88,8 +155,8 @@ fn delivered(payload: &[u8]) -> Option<DeliveredMessage> {
 
 #[cfg(test)]
 mod tests {
-    use super::forward;
-    use swarm_queue_client::AgentInput;
+    use super::{act, forward};
+    use swarm_queue_client::{AgentControl, AgentControlReply, AgentInput};
 
     fn payload(from: &str, body: &str) -> Vec<u8> {
         serde_json::to_vec(&AgentInput {
@@ -136,4 +203,21 @@ mod tests {
         let payloads = futures_util::stream::repeat(payload("mara", "hi"));
         forward(payloads, &tx).await;
     }
+
+    #[test]
+    fn a_cancel_is_answered_with_whether_it_acted() {
+        for acted in [true, false] {
+            let got = act(br#""cancel""#, |c| {
+                assert_eq!(c, AgentControl::Cancel);
+                acted
+            });
+            assert_eq!(got, Some(AgentControlReply { acted }));
+        }
+    }
+
+    #[test]
+    fn a_payload_that_is_not_a_command_is_not_answered_or_acted_on() {
+        let got = act(b"pause", |_| panic!("ran a command that was not one"));
+        assert_eq!(got, None);
+    }
 }

hive-agent/src/web_ui/actions.rs

@@ -10,7 +10,7 @@ use axum::{
 use serde::{Deserialize, Serialize};
 use std::future::Future;
 
-use super::{AppState, SigintOutcome, error_response};
+use super::{AppState, error_response};
 
 #[derive(Deserialize)]
 pub(super) struct SendForm {
@@ -53,41 +53,7 @@ pub(super) async fn post_send(
 }
 
 pub(super) async fn post_cancel_turn(State(state): State<AppState>) -> Response {
-    if let Some(canceller) = state.bus.turn_canceller() {
-        let note = if canceller.cancel() {
-            // Same as a signalled claude turn: the next wake prompt says it
-            // was cut off.
-            state
-                .interrupted
-                .store(true, std::sync::atomic::Ordering::Relaxed);
-            "operator: /cancel — sent session/cancel to the ACP agent"
-        } else {
-            "operator: /cancel — no ACP turn to cancel"
-        };
-        state
-            .bus
-            .emit(crate::events::LiveEvent::Note { text: note.into() });
-        return (axum::http::StatusCode::OK, "ok").into_response();
-    }
-    let out = super::sigint_claude();
-    let note = match out {
-        SigintOutcome::Signalled => {
-            // A process actually got signalled — the *next* turn's wake
-            // prompt should tell the agent it was cut off mid-work. Only
-            // set on an actual signal: `NoProcess` means /cancel raced an
-            // already-finished turn, so there's nothing to flag as
-            // interrupted.
-            state
-                .interrupted
-                .store(true, std::sync::atomic::Ordering::Relaxed);
-            "operator: /cancel — sent SIGINT to claude".to_owned()
-        }
-        SigintOutcome::NoProcess => "operator: /cancel — no claude process to interrupt".to_owned(),
-        SigintOutcome::Failed(e) => format!("operator: /cancel — kill failed: {e}"),
-    };
-    state
-        .bus
-        .emit(crate::events::LiveEvent::Note { text: note });
+    super::cancel_turn(&state.bus, &state.interrupted, "operator: /cancel");
     (axum::http::StatusCode::OK, "ok").into_response()
 }
 

hive-agent/src/web_ui/mod.rs

@@ -339,6 +339,35 @@ fn find_claude_child() -> Option<u32> {
     None
 }
 
+/// Stop the turn in flight and say so in the live panel as `by`. Returns
+/// whether a turn was actually stopped: a cancel that raced an already
+/// finished turn is `false`, and only a `true` marks the next wake prompt as
+/// cut off. Shared by the dashboard's `/cancel` and the swarm queue's cancel
+/// command, so the two cannot disagree about what cancelling is.
+pub(crate) fn cancel_turn(bus: &Bus, interrupted: &AtomicBool, by: &str) -> bool {
+    let (acted, what) = if let Some(canceller) = bus.turn_canceller() {
+        if canceller.cancel() {
+            (true, "sent session/cancel to the ACP agent".to_owned())
+        } else {
+            (false, "no ACP turn to cancel".to_owned())
+        }
+    } else {
+        match sigint_claude() {
+            SigintOutcome::Signalled => (true, "sent SIGINT to claude".to_owned()),
+            SigintOutcome::NoProcess => (false, "no claude process to interrupt".to_owned()),
+            SigintOutcome::Failed(e) => (false, format!("kill failed: {e}")),
+        }
+    };
+    if acted {
+        // The next turn's wake prompt tells the agent it was cut off mid-work.
+        interrupted.store(true, std::sync::atomic::Ordering::Relaxed);
+    }
+    bus.emit(crate::events::LiveEvent::Note {
+        text: format!("{by} - {what}"),
+    });
+    acted
+}
+
 /// SIGINT this container's harness-spawned claude child, if any
 /// (best-effort). See [`find_claude_child`] for how it's identified —
 /// **not** a name-based `pkill claude`, which would also hit an unrelated