diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 62882145..79d88353 100644 --- a/docs/swarm/ui.md +++ b/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//cancel` sends `"cancel"` as a NATS request on `$SWARM.control.`, and only swarm-controller may publish there. The diff --git a/swarm-controller/src/agent_input.rs b/swarm-controller/src/agent_input.rs index 455e56a2..7fac03a0 100644 --- a/swarm-controller/src/agent_input.rs +++ b/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, + /// 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, +} + +/// 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, +} + +/// 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 { 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, + headers: HeaderMap, + Json(req): Json, +) -> Result, 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 { let Some(status) = state.status.as_ref() else { - return Err(error_problem( - StatusCode::SERVICE_UNAVAILABLE, - "no swarm queue is configured on this host", - )); + return Err("no swarm queue is configured on this host".to_owned()); }; let client = status.queue_client(); - swarm_queue_client::ensure_connected(&client).map_err(|e| { - error_problem( + swarm_queue_client::ensure_connected(&client).map_err(|e| swarm_queue_client::chain(&e))?; + Ok(client) +} + +/// The swarm roster, or the 503 `GET /api/agents` answers when it has none. +async fn roster(state: &AppState) -> Result, ProblemDetails> { + let Some(bridge) = state.auth.as_ref() else { + return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, - &swarm_queue_client::chain(&e), - ) - })?; - send(&client, &agent, &input).await + "no identity bridge is configured on this host", + )); + }; + 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) - .map_err(|reason| (StatusCode::BAD_REQUEST, reason))? - .into_string(); + let agent = agent_name(agent).map_err(|reason| (StatusCode::BAD_REQUEST, reason))?; + Ok((agent, input(from, body)?)) +} + +/// 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, + body: String, +) -> Result<(Vec, 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 { + Ok(hive_types::Ident::parse(agent)?.into_string()) +} + +fn input(from: String, body: String) -> Result { 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 { + 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( + sink: &S, + agents: Vec, + input: &AgentInput, +) -> Vec { + 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| { - error_problem( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("encoding the input failed: {e}"), - ) - })?; + let payload = serde_json::to_vec(input) + .map_err(|e| NotSent::Failed(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( - StatusCode::SERVICE_UNAVAILABLE, - &format!( - "{agent} is not listening for input: its harness is not running or not connected to the swarm queue" - ), - )), + Delivery::NotRunning => Err(NotSent::NotRunning(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( - StatusCode::INTERNAL_SERVER_ERROR, - &format!("sending to {subject} failed: {e}"), - )) + Err(NotSent::Failed(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 { + 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>, + } + + impl InputSink for SomeListen { + async fn deliver(&self, subject: String, _payload: Vec) -> 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 = (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); diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 90c6266e..9b8eb672 100644 --- a/swarm-controller/src/main.rs +++ b/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,