swarm-controller: send one operator message to several agents

The swarm UI's broadcast page (#5524) needs the controller to do the
fan-out, so POST /api/agents/broadcast takes {agents, body} and answers
200 with {results: [{agent, sent, reason?}]} in request order.

- agent_input.rs: the single-agent send is now send_one, which both
  routes call; the single route maps its outcome to the same statuses
  and texts as before. The broadcast validates the whole request first
  (sender, empty body, empty or duplicated agents, more than 64, a name
  outside the roster), then sends up to 16 at once with the single
  route's no-responders window.
- main.rs: wires the route.
- docs/swarm/ui.md: describes the route next to the single-agent one.

docs/swarm/ui.md

@@ -223,6 +223,15 @@ credential may subscribe there. The harness doesn't reply, so swarm-controller
 answers 202 when the request times out after two seconds, and 503 when the
 queue reports no subscriber.
 
+`POST /api/agents/broadcast` takes `{agents, body}` and sends the same
+message to each agent the same way, up to 16 at once. It answers 200 with
+`{results: [{agent, sent, reason?}]}`, one entry per agent in request order,
+and `reason` holds the text the single-agent route would have answered with.
+One agent not listening doesn't stop the others. swarm-controller refuses the
+whole request and sends nothing when `body` or `agents` is empty, when
+`agents` names an agent twice, names more than 64, or names one that
+`GET /api/agents` doesn't list.
+
 A command has its own subject, so the harness never mistakes typed text for
 one. `POST /api/agents/<name>/cancel` sends `"cancel"` as a NATS request on
 `$SWARM.control.<name>`, and only swarm-controller may publish there. The

swarm-controller/src/agent_input.rs

@@ -2,6 +2,11 @@
 //! handed to one agent's harness as an inbox turn. No hive is in the path: the
 //! agent subscribes to [`swarm_queue_client::agent_input_subject`] itself.
 //!
+//! `POST /api/agents/broadcast` sends one text to several agents through the
+//! same send and answers per agent, in request order: one agent not getting it
+//! does not fail the rest. A name must be in the swarm roster (what
+//! `GET /api/agents` lists), or nothing is sent.
+//!
 //! **The sender is the `Remote-User` header**, which the gateway sets on
 //! `/api/` from authelia's auth-request answer and so overwrites whatever the
 //! client sent (`nix/host-modules/swarm-ui.nix`). A request without one is
@@ -14,13 +19,15 @@
 //! "sent" means. Core NATS, no stream: an agent that restarts before its turn
 //! loses the text.
 
+use std::collections::HashSet;
 use std::time::Duration;
 
 use axum::Json;
 use axum::extract::{Path, State};
 use axum::http::{HeaderMap, StatusCode};
+use futures_util::StreamExt as _;
 use problem_details::ProblemDetails;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use swarm_queue_client::AgentInput;
 use utoipa::ToSchema;
 
@@ -34,6 +41,14 @@ const SENDER_HEADER: &str = "remote-user";
 /// so a send to a live agent always waits this long.
 const NO_RESPONDERS_WINDOW: Duration = Duration::from_secs(2);
 
+/// The most agents one `POST /api/agents/broadcast` may name.
+const MAX_BROADCAST_AGENTS: usize = 64;
+
+/// How many sends of one broadcast are in flight at once. A send to a live
+/// agent takes [`NO_RESPONDERS_WINDOW`], so a broadcast to
+/// [`MAX_BROADCAST_AGENTS`] live agents answers after four windows.
+const BROADCAST_CONCURRENCY: usize = 16;
+
 /// Text for one agent.
 #[derive(Deserialize, ToSchema)]
 pub struct PostAgentInputRequest {
@@ -41,6 +56,41 @@ pub struct PostAgentInputRequest {
     body: String,
 }
 
+/// One text for several agents.
+#[derive(Deserialize, ToSchema)]
+pub struct BroadcastRequest {
+    /// The agents to send to, each named once.
+    agents: Vec<String>,
+    /// What each agent's turn shows as the message body, verbatim. Not logged.
+    body: String,
+}
+
+/// What became of a broadcast: one entry per requested agent, in request
+/// order.
+#[derive(Debug, Serialize, ToSchema)]
+pub struct BroadcastResponse {
+    results: Vec<BroadcastResult>,
+}
+
+/// What became of the text for one agent of a broadcast.
+#[derive(Debug, PartialEq, Eq, Serialize, ToSchema)]
+pub struct BroadcastResult {
+    agent: String,
+    /// Whether the agent's harness was subscribed and handed the message.
+    sent: bool,
+    /// Why not, in the words `POST /api/agents/{name}/input` answers with.
+    /// Present only when `sent` is false.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    reason: Option<String>,
+}
+
+/// Why [`send_one`] did not hand a message over, with the text to show for it.
+#[derive(Debug, PartialEq, Eq)]
+enum NotSent {
+    NotRunning(String),
+    Failed(String),
+}
+
 /// What became of one message.
 #[derive(Debug, PartialEq, Eq)]
 pub(crate) enum Delivery {
@@ -104,20 +154,69 @@ pub(crate) async fn post_agent_input(
 ) -> Result<StatusCode, ProblemDetails> {
     let (agent, input) =
         message(&headers, &agent, req.body).map_err(|(s, d)| error_problem(s, d))?;
+    let client = queue(&state).map_err(|d| error_problem(StatusCode::SERVICE_UNAVAILABLE, &d))?;
+    send(&client, &agent, &input).await
+}
+
+/// Send one text to several agents as the logged-in operator.
+#[utoipa::path(
+    post,
+    path = "/api/agents/broadcast",
+    request_body = BroadcastRequest,
+    responses(
+        (status = 200, description = "one entry per requested agent, in request order: \
+            `sent`, and when false a `reason` in the words the single-agent input route \
+            answers with. One agent not getting it does not stop the others",
+            body = BroadcastResponse),
+        (status = 400, description = "`body` is empty, `agents` is empty, names an agent \
+            twice, names more than 64 agents, or names one not in the swarm roster or not \
+            shaped like an identifier; nothing was sent (problem+json)", body = String),
+        (status = 401, description = "the request carries no `Remote-User`, so there is no \
+            sender to name (problem+json)", body = String),
+        (status = 503, description = "the swarm roster could not be read, or no swarm queue \
+            is configured on this host or it could not be reached; nothing was sent \
+            (problem+json)", body = String),
+    ),
+    tag = "agents"
+)]
+pub(crate) async fn post_agents_broadcast(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+    Json(req): Json<BroadcastRequest>,
+) -> Result<Json<BroadcastResponse>, ProblemDetails> {
+    let (agents, input) =
+        broadcast(&headers, req.agents, req.body).map_err(|(s, d)| error_problem(s, &d))?;
+    let roster = roster(&state).await?;
+    known(&agents, &roster).map_err(|(s, d)| error_problem(s, &d))?;
+    let client = queue(&state).map_err(|d| error_problem(StatusCode::SERVICE_UNAVAILABLE, &d))?;
+    Ok(Json(BroadcastResponse {
+        results: send_all(&client, agents, &input).await,
+    }))
+}
+
+/// The connected queue client, or why there is none: the caller's 503.
+fn queue(state: &AppState) -> Result<async_nats::Client, String> {
     let Some(status) = state.status.as_ref() else {
-        return Err(error_problem(
+        return Err("no swarm queue is configured on this host".to_owned());
-            StatusCode::SERVICE_UNAVAILABLE,
-            "no swarm queue is configured on this host",
-        ));
     };
     let client = status.queue_client();
-    swarm_queue_client::ensure_connected(&client).map_err(|e| {
+    swarm_queue_client::ensure_connected(&client).map_err(|e| swarm_queue_client::chain(&e))?;
-        error_problem(
+    Ok(client)
+}
+
+/// The swarm roster, or the 503 `GET /api/agents` answers when it has none.
+async fn roster(state: &AppState) -> Result<Vec<String>, ProblemDetails> {
+    let Some(bridge) = state.auth.as_ref() else {
+        return Err(error_problem(
             StatusCode::SERVICE_UNAVAILABLE,
-            &swarm_queue_client::chain(&e),
+            "no identity bridge is configured on this host",
-        )
+        ));
-    })?;
+    };
-    send(&client, &agent, &input).await
+    bridge.list_agent_identities().await.map_err(|e| {
+        let detail = format!("reading the agent roster failed: {e:#}");
+        tracing::warn!(error = %detail, "broadcast refused");
+        error_problem(StatusCode::SERVICE_UNAVAILABLE, &detail)
+    })
 }
 
 /// Who the gateway says is asking, or the 401 for a request that did not come
@@ -144,44 +243,121 @@ fn message(
     body: String,
 ) -> Result<(String, AgentInput), (StatusCode, &'static str)> {
     let from = sender(headers)?;
-    let agent = hive_types::Ident::parse(agent)
+    let agent = agent_name(agent).map_err(|reason| (StatusCode::BAD_REQUEST, reason))?;
-        .map_err(|reason| (StatusCode::BAD_REQUEST, reason))?
+    Ok((agent, input(from, body)?))
-        .into_string();
+}
+
+/// The agents and the one message for all of them, or the refusal of the
+/// whole request. Everything decided before the roster and the queue are
+/// touched.
+fn broadcast(
+    headers: &HeaderMap,
+    agents: Vec<String>,
+    body: String,
+) -> Result<(Vec<String>, AgentInput), (StatusCode, String)> {
+    let refused = |detail: String| (StatusCode::BAD_REQUEST, detail);
+    let from = sender(headers).map_err(|(s, d)| (s, d.to_owned()))?;
+    let input = input(from, body).map_err(|(s, d)| (s, d.to_owned()))?;
+    if agents.is_empty() {
+        return Err(refused("agents is empty".to_owned()));
+    }
+    if agents.len() > MAX_BROADCAST_AGENTS {
+        return Err(refused(format!(
+            "{} agents named; one broadcast may name at most {MAX_BROADCAST_AGENTS}",
+            agents.len()
+        )));
+    }
+    let mut seen = HashSet::new();
+    let mut names = Vec::with_capacity(agents.len());
+    for agent in agents {
+        let name = agent_name(&agent).map_err(|reason| refused(format!("{agent:?}: {reason}")))?;
+        if !seen.insert(name.clone()) {
+            return Err(refused(format!("{name} is named twice")));
+        }
+        names.push(name);
+    }
+    Ok((names, input))
+}
+
+/// Refuses the first of `agents` not in `roster`.
+fn known(agents: &[String], roster: &[String]) -> Result<(), (StatusCode, String)> {
+    match agents.iter().find(|a| !roster.contains(a)) {
+        Some(agent) => Err((
+            StatusCode::BAD_REQUEST,
+            format!("{agent} is not an agent of this swarm"),
+        )),
+        None => Ok(()),
+    }
+}
+
+fn agent_name(agent: &str) -> Result<String, &'static str> {
+    Ok(hive_types::Ident::parse(agent)?.into_string())
+}
+
+fn input(from: String, body: String) -> Result<AgentInput, (StatusCode, &'static str)> {
     if body.trim().is_empty() {
         return Err((StatusCode::BAD_REQUEST, "body is empty"));
     }
-    Ok((agent, AgentInput { from, body }))
+    Ok(AgentInput { from, body })
 }
 
+/// `POST /api/agents/{name}/input`'s answer for one send.
 async fn send(
     sink: &impl InputSink,
     agent: &str,
     input: &AgentInput,
 ) -> Result<StatusCode, ProblemDetails> {
+    match send_one(sink, agent, input).await {
+        Ok(()) => Ok(StatusCode::ACCEPTED),
+        Err(NotSent::NotRunning(detail)) => {
+            Err(error_problem(StatusCode::SERVICE_UNAVAILABLE, &detail))
+        }
+        Err(NotSent::Failed(detail)) => {
+            Err(error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail))
+        }
+    }
+}
+
+/// Every agent's result, in the order of `agents`; at most
+/// [`BROADCAST_CONCURRENCY`] sends are in flight.
+async fn send_all<S: InputSink + Sync>(
+    sink: &S,
+    agents: Vec<String>,
+    input: &AgentInput,
+) -> Vec<BroadcastResult> {
+    futures_util::stream::iter(agents)
+        .map(|agent| async move {
+            let reason = match send_one(sink, &agent, input).await {
+                Ok(()) => None,
+                Err(NotSent::NotRunning(detail) | NotSent::Failed(detail)) => Some(detail),
+            };
+            BroadcastResult {
+                agent,
+                sent: reason.is_none(),
+                reason,
+            }
+        })
+        .buffered(BROADCAST_CONCURRENCY)
+        .collect()
+        .await
+}
+
+/// Hand `input` to `agent`'s harness. The one send both routes make.
+async fn send_one(sink: &impl InputSink, agent: &str, input: &AgentInput) -> Result<(), NotSent> {
     let subject = swarm_queue_client::agent_input_subject(agent);
-    let payload = serde_json::to_vec(input).map_err(|e| {
+    let payload = serde_json::to_vec(input)
-        error_problem(
+        .map_err(|e| NotSent::Failed(format!("encoding the input failed: {e}")))?;
-            StatusCode::INTERNAL_SERVER_ERROR,
-            &format!("encoding the input failed: {e}"),
-        )
-    })?;
     match sink.deliver(subject.clone(), payload).await {
         Delivery::Sent => {
             tracing::info!(%agent, from = %input.from, "agent input sent");
-            Ok(StatusCode::ACCEPTED)
+            Ok(())
         }
-        Delivery::NotRunning => Err(error_problem(
+        Delivery::NotRunning => Err(NotSent::NotRunning(format!(
-            StatusCode::SERVICE_UNAVAILABLE,
-            &format!(
             "{agent} is not listening for input: its harness is not running or not connected to the swarm queue"
-            ),
+        ))),
-        )),
         Delivery::Failed(e) => {
             tracing::warn!(%agent, %subject, error = %e, "agent input failed");
-            Err(error_problem(
+            Err(NotSent::Failed(format!("sending to {subject} failed: {e}")))
-                StatusCode::INTERNAL_SERVER_ERROR,
-                &format!("sending to {subject} failed: {e}"),
-            ))
         }
     }
 }
@@ -193,7 +369,10 @@ mod tests {
     use axum::http::{HeaderMap, HeaderValue, StatusCode};
     use swarm_queue_client::AgentInput;
 
-    use super::{Delivery, InputSink, message, send};
+    use super::{
+        BroadcastResult, Delivery, InputSink, MAX_BROADCAST_AGENTS, broadcast, known, message,
+        send, send_all,
+    };
 
     /// Answers every message with one fixed outcome and records what it got.
     struct FakeSink {
@@ -230,6 +409,31 @@ mod tests {
         r.expect_err("refused").0
     }
 
+    fn names(n: &[&str]) -> Vec<String> {
+        n.iter().map(|&a| a.to_owned()).collect()
+    }
+
+    fn broadcast_refusal(agents: &[&str], body: &str) -> (StatusCode, String) {
+        broadcast(&from("mara"), names(agents), body.to_owned()).expect_err("refused")
+    }
+
+    /// Sends to `atlas` and `iris` succeed, everyone else has no subscriber.
+    struct SomeListen {
+        got: Mutex<Vec<String>>,
+    }
+
+    impl InputSink for SomeListen {
+        async fn deliver(&self, subject: String, _payload: Vec<u8>) -> Delivery {
+            let agent = subject.trim_start_matches("$SWARM.input.").to_owned();
+            self.got.lock().expect("unpoisoned").push(agent.clone());
+            if agent == "atlas" || agent == "iris" {
+                Delivery::Sent
+            } else {
+                Delivery::NotRunning
+            }
+        }
+    }
+
     #[test]
     fn the_sender_is_the_remote_user_header() {
         let (agent, input) = message(&from("mara"), "atlas", "hi".to_owned()).expect("valid");
@@ -269,6 +473,116 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_broadcast_names_its_agents_in_request_order_with_one_message() {
+        let (agents, input) =
+            broadcast(&from("mara"), names(&["iris", "atlas"]), "hi".to_owned()).expect("valid");
+        assert_eq!(agents, names(&["iris", "atlas"]));
+        assert_eq!(
+            input,
+            AgentInput {
+                from: "mara".to_owned(),
+                body: "hi".to_owned()
+            }
+        );
+    }
+
+    #[test]
+    fn a_broadcast_is_refused_whole_with_the_single_route_s_words() {
+        assert_eq!(
+            broadcast(&HeaderMap::new(), names(&["atlas"]), "hi".to_owned())
+                .expect_err("no sender")
+                .0,
+            StatusCode::UNAUTHORIZED
+        );
+        assert_eq!(
+            broadcast_refusal(&["atlas"], " \n"),
+            (StatusCode::BAD_REQUEST, "body is empty".to_owned())
+        );
+        assert_eq!(
+            broadcast_refusal(&[], "hi"),
+            (StatusCode::BAD_REQUEST, "agents is empty".to_owned())
+        );
+        let (status, detail) = broadcast_refusal(&["atlas", "../x"], "hi");
+        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert!(detail.starts_with("\"../x\": "), "{detail}");
+        assert_eq!(
+            broadcast_refusal(&["atlas", "iris", "atlas"], "hi"),
+            (StatusCode::BAD_REQUEST, "atlas is named twice".to_owned())
+        );
+    }
+
+    #[test]
+    fn a_broadcast_may_name_up_to_the_cap_and_no_more() {
+        let at_cap: Vec<String> = (0..MAX_BROADCAST_AGENTS).map(|i| format!("a{i}")).collect();
+        let mut over = at_cap.clone();
+        over.push("one-more".to_owned());
+        broadcast(&from("mara"), at_cap, "hi".to_owned()).expect("the cap itself is allowed");
+        let (status, detail) =
+            broadcast(&from("mara"), over, "hi".to_owned()).expect_err("over the cap");
+        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert!(detail.contains("at most 64"), "{detail}");
+    }
+
+    #[test]
+    fn a_name_outside_the_roster_refuses_the_broadcast() {
+        let roster = names(&["atlas", "iris"]);
+        known(&names(&["iris", "atlas"]), &roster).expect("both in the roster");
+        assert_eq!(
+            known(&names(&["atlas", "ghost"]), &roster).expect_err("ghost is unknown"),
+            (
+                StatusCode::BAD_REQUEST,
+                "ghost is not an agent of this swarm".to_owned()
+            )
+        );
+    }
+
+    #[tokio::test]
+    async fn one_agent_not_listening_does_not_stop_the_others() {
+        let sink = SomeListen {
+            got: Mutex::default(),
+        };
+        let input = AgentInput {
+            from: "mara".to_owned(),
+            body: "hi".to_owned(),
+        };
+        let results = send_all(&sink, names(&["iris", "argus", "atlas"]), &input).await;
+        assert_eq!(
+            results,
+            vec![
+                BroadcastResult {
+                    agent: "iris".to_owned(),
+                    sent: true,
+                    reason: None
+                },
+                BroadcastResult {
+                    agent: "argus".to_owned(),
+                    sent: false,
+                    reason: Some(
+                        "argus is not listening for input: its harness is not running or not \
+                         connected to the swarm queue"
+                            .to_owned()
+                    )
+                },
+                BroadcastResult {
+                    agent: "atlas".to_owned(),
+                    sent: true,
+                    reason: None
+                },
+            ]
+        );
+        let mut got = sink.got.lock().expect("unpoisoned").clone();
+        got.sort();
+        assert_eq!(got, names(&["argus", "atlas", "iris"]));
+        let json = serde_json::to_value(&results).expect("serializes");
+        assert_eq!(
+            json[0],
+            serde_json::json!({ "agent": "iris", "sent": true })
+        );
+        assert_eq!(json[1]["sent"], false);
+        assert!(json[1]["reason"].is_string());
+    }
+
     #[tokio::test]
     async fn a_held_message_is_202_on_the_agent_s_own_subject() {
         let sink = FakeSink::new(|| Delivery::Sent);

swarm-controller/src/main.rs

@@ -3697,6 +3697,7 @@ fn build_app(state: AppState) -> axum::Router {
         .routes(routes!(subagent_term::stream_subagent_term))
         .routes(routes!(agent_state_stream::stream_agent_state))
         .routes(routes!(agent_input::post_agent_input))
+        .routes(routes!(agent_input::post_agents_broadcast))
         .routes(routes!(agent_control::post_agent_cancel))
         .routes(routes!(
             claude_login::post_claude_login,