From 48ac4ac42301459d916c068030cc4008e9a13737 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 4 Oct 2026 03:00:05 +0200 Subject: [PATCH] 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.`. 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 --- docs/swarm/ui.md | 32 +- .../src/pages/agents/AgentInputForm.css | 17 + .../src/pages/agents/AgentInputForm.tsx | 70 ++++ .../src/pages/agents/AgentTermPreview.tsx | 8 +- .../src/pages/agents/AgentTerminalPage.tsx | 6 +- .../swarm-ui/src/pages/agents/AgentsPage.tsx | 4 +- hive-agent/src/main.rs | 16 +- hive-agent/src/swarm_input.rs | 144 +++++++++ nix/host-modules/swarm-ui.nix | 17 +- swarm-controller/src/agent_input.rs | 300 ++++++++++++++++++ swarm-controller/src/main.rs | 2 + swarm-nats-auth/src/policy.rs | 42 +++ swarm-queue-client/src/lib.rs | 30 ++ 13 files changed, 664 insertions(+), 24 deletions(-) create mode 100644 frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.css create mode 100644 frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.tsx create mode 100644 hive-agent/src/swarm_input.rs create mode 100644 swarm-controller/src/agent_input.rs diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 0481ebfd..e731ea35 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -29,9 +29,35 @@ terminal in the same preview, and **expand** opens it in its own tab, as it does for the agent. The list refreshes every 10 seconds, so a subagent the agent spawns shows up once it writes output. -Both terminals are live views with no input box: a preview shows rows -published while it's open. Subagents take no input from the swarm. A subagent -terminal has no turn-state badges. +A preview shows rows published while it's open. A subagent terminal has no +turn-state badges. + +### Sending a message to an agent + +Under an agent's terminal, in the preview and in its own tab, is a **message** +box. **send** hands the text to the agent's harness, and the agent takes it as +an inbox message from your authelia username. **sent** means the harness was +listening and holds the message. If the harness isn't running, or isn't +connected to the swarm queue, the box shows an error and the agent gets nothing. +Subagents have no message box. + +The message doesn't go through the agent's hive, so the hive dashboard's +message history doesn't list it. The agent's answer shows in its terminal; an +agent that calls `send` to your username gets an unknown-recipient error. A +harness that restarts before taking its turn loses the message. + +
How a message reaches the agent + +The gateway sets `Remote-User` from authelia's answer on every `/api/` +request, replacing any the browser sent. `POST /api/agents//input` +takes that header as the sender and sends `{from, body}` as a NATS request on +`$SWARM.input.`. swarm-controller is the only queue client allowed to +publish there. The agent's harness subscribes to its own subject and queues +each message for its turn loop with its reminders. 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. + +
Where the subagent list and rows come from diff --git a/frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.css b/frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.css new file mode 100644 index 00000000..0c8acb2b --- /dev/null +++ b/frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.css @@ -0,0 +1,17 @@ +/* — one row: the text field takes the width, the send + button sits at its baseline. */ +.ui-agent-input { + margin-block-start: 0.5rem; +} +.ui-agent-input-form { + display: flex; + align-items: flex-end; + gap: 0.5em; +} +.ui-agent-input-form > :first-child { + flex: 1; +} +.ui-agent-input-sent { + margin-block-start: 0.25rem; + color: var(--green); +} diff --git a/frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.tsx b/frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.tsx new file mode 100644 index 00000000..90284649 --- /dev/null +++ b/frontend/packages/swarm-ui/src/pages/agents/AgentInputForm.tsx @@ -0,0 +1,70 @@ +// — text for one agent, under its swarm terminal. +// POSTs `/api/agents/{name}/input`: 202 once the agent's harness holds the +// message, 503 when it is not listening, problem+json otherwise. The agent +// sees it as an inbox turn from the logged-in user (the controller takes the +// sender from the gateway, not from this request). +import { useState } from "preact/hooks"; +import { ApiErrorPanel } from "@hive/shared/api-error-panel.js"; +import { readApiError, type ProblemDetails } from "@hive/shared/api-error.js"; +import { TextField } from "../../ui/text-field/TextField.js"; +import { Button } from "../../ui/button/Button.js"; +import "./AgentInputForm.css"; + +type SubmitState = + | { status: "idle" } + | { status: "submitting" } + | { status: "sent" } + | { status: "error"; problem: ProblemDetails }; + +export function AgentInputForm({ agentName }: { agentName: string }) { + const [body, setBody] = useState(""); + const [result, setResult] = useState({ status: "idle" }); + + async function submit(e: Event) { + e.preventDefault(); + setResult({ status: "submitting" }); + try { + const r = await fetch( + `/api/agents/${encodeURIComponent(agentName)}/input`, + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ body }), + }, + ); + if (!r.ok) { + setResult({ status: "error", problem: await readApiError(r) }); + return; + } + setResult({ status: "sent" }); + setBody(""); + } catch (err) { + setResult({ status: "error", problem: { detail: String(err) } }); + } + } + + return ( +
+
+ + + + {result.status === "sent" &&

sent

} + {result.status === "error" && ( + + )} +
+ ); +} diff --git a/frontend/packages/swarm-ui/src/pages/agents/AgentTermPreview.tsx b/frontend/packages/swarm-ui/src/pages/agents/AgentTermPreview.tsx index bcf0e43f..7d0f54e1 100644 --- a/frontend/packages/swarm-ui/src/pages/agents/AgentTermPreview.tsx +++ b/frontend/packages/swarm-ui/src/pages/agents/AgentTermPreview.tsx @@ -1,7 +1,7 @@ // — the MVP mara's own ruling asked for: not a separate -// page, a small read-only preview embedded below `AgentsPage`'s detail -// panel fields. No input — sending input back to the agent is explicitly -// follow-up scope, not this issue's. Consumes swarm-controller's +// page, a small preview embedded below `AgentsPage`'s detail panel fields, with +// an `AgentInputForm` under an agent's terminal (a subagent's has none: it +// takes no input of its own). Consumes swarm-controller's // `GET /api/agents/{name}/term/stream`, or for a subagent // `GET /api/agents/{name}/subagents/{subagent}/term/stream`, via // `useSwarmTermStream`, rendering through the same `@hive/shared` `Row` @@ -49,6 +49,7 @@ import { useSwarmAgentStateStream, type TokenUsage, } from "./useSwarmAgentStateStream.js"; +import { AgentInputForm } from "./AgentInputForm.js"; import "./AgentTermPreview.css"; const NEAR_BOTTOM_PX = 48; @@ -256,6 +257,7 @@ export function AgentTermPreview({ )} + {subagent === undefined && } ); } diff --git a/frontend/packages/swarm-ui/src/pages/agents/AgentTerminalPage.tsx b/frontend/packages/swarm-ui/src/pages/agents/AgentTerminalPage.tsx index 149b6a4f..cebaaad6 100644 --- a/frontend/packages/swarm-ui/src/pages/agents/AgentTerminalPage.tsx +++ b/frontend/packages/swarm-ui/src/pages/agents/AgentTerminalPage.tsx @@ -1,10 +1,8 @@ // — the swarm terminal tab's full page, mounted at // `/agents/:name/terminal`. The MVP `AgentTermPreview` (below `AgentsPage`'s -// detail panel) is explicitly capped and read-only; this is that same +// detail panel) is height-capped; this is that same // component's `fullHeight` mount — no 12em cap, floating header badges -// shown (there's no detail-panel `
` here to duplicate them). Sending -// input back to the agent is still explicit follow-up scope, same as the -// preview's own file comment says — this page doesn't change that. +// shown (there's no detail-panel `
` here to duplicate them). // // Registers itself in the dynamic-tabs row on mount (not just when opened // via the preview's expand trigger) — a bookmark, a shared link, or a diff --git a/frontend/packages/swarm-ui/src/pages/agents/AgentsPage.tsx b/frontend/packages/swarm-ui/src/pages/agents/AgentsPage.tsx index 911cada4..a6c70284 100644 --- a/frontend/packages/swarm-ui/src/pages/agents/AgentsPage.tsx +++ b/frontend/packages/swarm-ui/src/pages/agents/AgentsPage.tsx @@ -651,9 +651,7 @@ export function AgentsPage() { } /> - {/* MVP scope per mara's own ruling: a small read-only - preview, no header/no input — the full terminal - (+ sending input back) is explicit follow-up scope. */} + {/* The agent's live terminal and its input box. */} (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(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(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( files: &turn::TurnFiles, todo_wake: Arc, todos_store: Option>, - mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver, + mut local_msg_rx: tokio::sync::mpsc::UnboundedReceiver, interrupted: Arc, ) -> Result<()> { tracing::info!(socket = %socket.display(), "harness serve"); @@ -690,7 +694,7 @@ async fn serve_loop( // 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( 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, diff --git a/hive-agent/src/swarm_input.rs b/hive-agent/src/swarm_input.rs new file mode 100644 index 00000000..c77b1702 --- /dev/null +++ b/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) { + 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, 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>( + payloads: impl Stream, + tx: &UnboundedSender, +) { + 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 { + match serde_json::from_slice::(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 { + 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; + } +} diff --git a/nix/host-modules/swarm-ui.nix b/nix/host-modules/swarm-ui.nix index 90443482..14bd2889 100644 --- a/nix/host-modules/swarm-ui.nix +++ b/nix/host-modules/swarm-ui.nix @@ -215,7 +215,14 @@ in # nginx forwards, no prefix-stripping to keep in sync by hand. "/api/" = { proxyPass = "http://unix:${deployCfg.swarm-controller.socketPath}:"; - extraConfig = swarmAuthRequest; + # `Remote-User` from authelia's answer, set here so it replaces + # any the client sent: swarm-controller takes it as the sender of + # the text the operator types to an agent (`agent_input.rs`). + extraConfig = '' + ${swarmAuthRequest} + auth_request_set $authelia_user $upstream_http_remote_user; + proxy_set_header Remote-User $authelia_user; + ''; }; # ⚠️ THE ONE LOCATION ON THIS VHOST WITH NO `swarmAuthRequest`, # and that is deliberate rather than an omission. @@ -247,10 +254,10 @@ in ''; }; # Session identity for the header's profile menu: initials - # avatar + "who is this" line. `auth_request` above - # only ever answers yes/no — it never forwards *who* — so the - # frontend has no other way to learn this. Same-origin proxy - # straight to authelia's own `GET /api/user/info` + # avatar + "who is this" line. `auth_request` tells the + # browser nothing about *who* (only `/api/` passes it on, to the + # controller), so the frontend has no other way to learn this. + # Same-origin proxy straight to authelia's own `GET /api/user/info` # (session-cookie authenticated) rather than new # swarm-controller code: the cookie is already valid here (the # session cookie's domain is the swarm's, shared across every diff --git a/swarm-controller/src/agent_input.rs b/swarm-controller/src/agent_input.rs new file mode 100644 index 00000000..3e6bfef5 --- /dev/null +++ b/swarm-controller/src/agent_input.rs @@ -0,0 +1,300 @@ +//! `POST /api/agents/{name}/input` — text the operator types in the swarm UI, +//! 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. +//! +//! **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 +//! refused rather than sent anonymously. +//! +//! **Sent as a NATS request** so that an agent with no subscriber is a 503 +//! instead of text that silently goes nowhere. The agent does not reply: no +//! agent grant covers a reply subject. A request that times out therefore +//! means a subscriber held the message, and that is all "sent" means. Core +//! NATS, no stream: an agent that restarts before its turn loses the text. + +use std::time::Duration; + +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::{HeaderMap, StatusCode}; +use problem_details::ProblemDetails; +use serde::Deserialize; +use swarm_queue_client::AgentInput; +use utoipa::ToSchema; + +use crate::{AppState, error_problem}; + +/// The header the gateway names the session's user in. +const SENDER_HEADER: &str = "remote-user"; + +/// How long to wait for the server's no-responders answer. It comes back +/// within a round trip when nothing subscribes; a subscriber sends nothing, +/// so a send to a live agent always waits this long. +const NO_RESPONDERS_WINDOW: Duration = Duration::from_secs(2); + +/// Text for one agent. +#[derive(Deserialize, ToSchema)] +pub struct PostAgentInputRequest { + /// What the agent's turn shows as the message body, verbatim. Not logged. + body: String, +} + +/// What became of one message. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Delivery { + /// A subscriber held it. + Sent, + /// Nothing subscribes to the agent's input subject. + NotRunning, + /// The queue refused or failed the request. + Failed(String), +} + +/// Where a message goes: the queue in production, a fake in the tests. +pub(crate) trait InputSink { + fn deliver(&self, subject: String, payload: Vec) -> impl Future + Send; +} + +impl InputSink for async_nats::Client { + async fn deliver(&self, subject: String, payload: Vec) -> Delivery { + use async_nats::RequestErrorKind; + + let request = async_nats::Request::new() + .payload(payload.into()) + .timeout(Some(NO_RESPONDERS_WINDOW)); + match self.send_request(subject, request).await { + Ok(_) => Delivery::Sent, + Err(e) => match e.kind() { + RequestErrorKind::TimedOut => Delivery::Sent, + RequestErrorKind::NoResponders => Delivery::NotRunning, + _ => Delivery::Failed(e.to_string()), + }, + } + } +} + +/// Send text to an agent as the logged-in operator. +#[utoipa::path( + post, + path = "/api/agents/{name}/input", + params(("name" = String, Path, description = "agent to send the text to")), + request_body = PostAgentInputRequest, + responses( + (status = 202, description = "the agent's harness is subscribed and was handed the \ + message; it becomes a turn from the sender when the harness next picks one up"), + (status = 400, description = "the agent name is not shaped like an identifier, or \ + the body is empty (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 agent's harness is not subscribed — not running, \ + or not connected to the swarm queue — or no swarm queue is configured on this \ + host or it could not be reached (problem+json)", body = String), + (status = 500, description = "the queue refused or failed the request (problem+json)", + body = String), + ), + tag = "agents" +)] +pub(crate) async fn post_agent_input( + State(state): State, + Path(agent): Path, + headers: HeaderMap, + Json(req): Json, +) -> Result { + let (agent, input) = + message(&headers, &agent, req.body).map_err(|(s, d)| error_problem(s, d))?; + let Some(status) = state.status.as_ref() else { + return Err(error_problem( + 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| { + error_problem( + StatusCode::SERVICE_UNAVAILABLE, + &swarm_queue_client::chain(&e), + ) + })?; + send(&client, &agent, &input).await +} + +/// The agent and the message, or the refusal. Everything decided before the +/// queue is touched. +fn message( + headers: &HeaderMap, + agent: &str, + body: String, +) -> Result<(String, AgentInput), (StatusCode, &'static str)> { + let from = headers + .get(SENDER_HEADER) + .and_then(|v| v.to_str().ok()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .ok_or(( + StatusCode::UNAUTHORIZED, + "no Remote-User header: the gateway names the sender, and this request \ + did not come through it", + ))? + .to_owned(); + let agent = hive_types::Ident::parse(agent) + .map_err(|reason| (StatusCode::BAD_REQUEST, reason))? + .into_string(); + if body.trim().is_empty() { + return Err((StatusCode::BAD_REQUEST, "body is empty")); + } + Ok((agent, AgentInput { from, body })) +} + +async fn send( + sink: &impl InputSink, + agent: &str, + input: &AgentInput, +) -> Result { + 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}"), + ) + })?; + match sink.deliver(subject.clone(), payload).await { + Delivery::Sent => { + tracing::info!(%agent, from = %input.from, "agent input sent"); + Ok(StatusCode::ACCEPTED) + } + 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::Failed(e) => { + tracing::warn!(%agent, %subject, error = %e, "agent input failed"); + Err(error_problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("sending to {subject} failed: {e}"), + )) + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use axum::http::{HeaderMap, HeaderValue, StatusCode}; + use swarm_queue_client::AgentInput; + + use super::{Delivery, InputSink, message, send}; + + /// Answers every message with one fixed outcome and records what it got. + struct FakeSink { + answer: fn() -> Delivery, + got: Mutex)>>, + } + + impl FakeSink { + fn new(answer: fn() -> Delivery) -> Self { + Self { + answer, + got: Mutex::default(), + } + } + } + + impl InputSink for FakeSink { + async fn deliver(&self, subject: String, payload: Vec) -> Delivery { + self.got + .lock() + .expect("unpoisoned") + .push((subject, payload)); + (self.answer)() + } + } + + fn from(user: &str) -> HeaderMap { + let mut h = HeaderMap::new(); + h.insert("Remote-User", HeaderValue::from_str(user).expect("ascii")); + h + } + + fn status_of(r: Result<(String, AgentInput), (StatusCode, &'static str)>) -> StatusCode { + r.expect_err("refused").0 + } + + #[test] + fn the_sender_is_the_remote_user_header() { + let (agent, input) = message(&from("mara"), "atlas", "hi".to_owned()).expect("valid"); + assert_eq!(agent, "atlas"); + assert_eq!( + input, + AgentInput { + from: "mara".to_owned(), + body: "hi".to_owned() + } + ); + } + + #[test] + fn no_sender_is_refused_before_anything_else() { + // A bad agent and an empty body as well: the missing sender is the + // answer, not whichever check happens to run first. + assert_eq!( + status_of(message(&HeaderMap::new(), "../x", String::new())), + StatusCode::UNAUTHORIZED + ); + assert_eq!( + status_of(message(&from(" "), "atlas", "hi".to_owned())), + StatusCode::UNAUTHORIZED + ); + } + + #[test] + fn a_bad_agent_or_an_empty_body_is_refused() { + assert_eq!( + status_of(message(&from("mara"), "../x", "hi".to_owned())), + StatusCode::BAD_REQUEST + ); + assert_eq!( + status_of(message(&from("mara"), "atlas", " \n".to_owned())), + StatusCode::BAD_REQUEST + ); + } + + #[tokio::test] + async fn a_held_message_is_202_on_the_agent_s_own_subject() { + let sink = FakeSink::new(|| Delivery::Sent); + let input = AgentInput { + from: "mara".to_owned(), + body: "hi".to_owned(), + }; + let status = send(&sink, "atlas", &input).await.expect("sent"); + assert_eq!(status, StatusCode::ACCEPTED); + let got = sink.got.lock().expect("unpoisoned"); + assert_eq!(got.len(), 1); + assert_eq!(got[0].0, "$SWARM.input.atlas"); + let decoded: AgentInput = serde_json::from_slice(&got[0].1).expect("an AgentInput"); + assert_eq!(decoded, input); + } + + #[tokio::test] + async fn no_subscriber_is_503_and_a_queue_failure_is_500() { + let input = AgentInput { + from: "mara".to_owned(), + body: "hi".to_owned(), + }; + let not_running = send(&FakeSink::new(|| Delivery::NotRunning), "atlas", &input) + .await + .expect_err("nobody listens"); + assert_eq!(not_running.status, Some(StatusCode::SERVICE_UNAVAILABLE)); + let failed = send( + &FakeSink::new(|| Delivery::Failed("refused".to_owned())), + "atlas", + &input, + ) + .await + .expect_err("the queue failed"); + assert_eq!(failed.status, Some(StatusCode::INTERNAL_SERVER_ERROR)); + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 047175ce..64209c94 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -42,6 +42,7 @@ use utoipa_axum::{router::OpenApiRouter, routes}; mod agent_icon; mod agent_identity; +mod agent_input; mod agent_renewal; mod agent_state_stream; mod agent_status; @@ -3007,6 +3008,7 @@ fn build_app(state: AppState) -> axum::Router { .routes(routes!(subagent_term::list_subagents)) .routes(routes!(subagent_term::stream_subagent_term)) .routes(routes!(agent_state_stream::stream_agent_state)) + .routes(routes!(agent_input::post_agent_input)) .routes(routes!(issue_report::get_repos)) .routes(routes!(issue_report::get_issue_report_all)) .routes(routes!(issue_report::get_issue_report)) diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 74f20eb8..dbface27 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -498,6 +498,11 @@ impl Policy { // admission). Same failure mode as the knowledge event above: a // refused publish reaches the client as a timeout. swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(), + // Operator input to an agent, one subject per agent, wildcard for + // the same no-roster reason. The harness takes the message's + // `from` as given, which holds only while this client is the + // family's sole publisher: no hive or agent grant may name it. + swarm_queue_client::AGENT_INPUT_SUBJECT_WILDCARD.to_owned(), // The wanted-state buckets — one per hive, all created and written // by this single client. // @@ -774,6 +779,43 @@ mod tests { } } + #[test] + fn a_reader_may_publish_input_to_any_agent() { + let p = policy().permissions("swarm-controller").expect("a reader"); + assert!( + p.publish + .contains(&swarm_queue_client::AGENT_INPUT_SUBJECT_WILDCARD.to_owned()), + "the controller is the only publisher of operator input; without \ + the grant its request is refused, and a refusal arrives as a timeout" + ); + } + + #[test] + fn neither_a_hive_nor_its_agents_may_publish_input_to_an_agent() { + // The harness turns this message into a turn from the operator named in + // it. A hive or agent able to publish here could speak as the operator + // to any agent in the swarm. + let policy = policy_with_agent_subject(); + for client in ["hive-alpha", "hive-alpha-agent"] { + let p = policy.permissions(client).expect("admitted"); + for forbidden in [ + swarm_queue_client::agent_input_subject("alpha"), + swarm_queue_client::AGENT_INPUT_SUBJECT_WILDCARD.to_owned(), + ] { + assert!( + !p.publish.contains(&forbidden), + "{client} must not publish {forbidden}: {:?}", + p.publish + ); + } + assert!( + !p.publish.iter().any(|s| s.starts_with("$SWARM.input")), + "{client}: {:?}", + p.publish + ); + } + } + #[test] fn a_hive_grant_never_includes_the_jetstream_wildcard() { // `$JS.API.>` also covers `$JS.API.STREAM.DELETE.KV_hive-status`, with diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 2316a6a9..b3483509 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -251,6 +251,36 @@ pub struct DeployRequest { pub rev: Option, } +/// The subject the swarm controller sends an operator's input for `agent` +/// on. That agent's harness subscribes to its own and turns each message into +/// an inbox turn. +/// +/// Its own family rather than a leaf under `$SWARM.term.`: every agent on a +/// hive's shared credential may publish under `$SWARM.term..>`, and the +/// harness trusts [`AgentInput::from`] only because the controller is the one +/// client granted publish here. +#[must_use] +pub fn agent_input_subject(agent: &str) -> String { + format!("{AGENT_INPUT_SUBJECT_PREFIX}.{agent}") +} + +/// The publish grant covering every [`agent_input_subject`], for the one +/// client that may send them. A wildcard for the same no-roster reason as +/// [`DEPLOY_SUBJECT_WILDCARD`]. +pub const AGENT_INPUT_SUBJECT_WILDCARD: &str = "$SWARM.input.*"; + +/// Shared by [`agent_input_subject`] and [`AGENT_INPUT_SUBJECT_WILDCARD`]. +const AGENT_INPUT_SUBJECT_PREFIX: &str = "$SWARM.input"; + +/// What an [`agent_input_subject`] message carries. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentInput { + /// The operator's authelia username, as the swarm gateway reported it. + pub from: String, + /// The text, verbatim. + pub body: String, +} + /// The hive-notices stream, shared by the hive that publishes and /// whatever eventually consumes it. Behind the `notices` feature, same /// reason `status` is behind `kv` — see the module doc.