diff --git a/docs/swarm/README.md b/docs/swarm/README.md index be229c74..0186b996 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -281,12 +281,14 @@ one, as the hive dashboard does. The queue also limits what an agent may **subscribe** to, because agents are the least trusted clients on the queue. An agent's own queue credential, which its subagent daemon also presents, may subscribe to its reply inbox -`_INBOX.agent..>` and its operator-input subject `$SWARM.input.`, -and nothing else. The harness and the subagent daemon connect with -`_INBOX.agent.` as their inbox prefix, so the replies to the icon write -and the schedule read arrive. An agent can't read another agent's terminal, -state, input or replies. The queue refuses the hive's shared agent client -outright. +`_INBOX.agent..>`, its operator-input subject `$SWARM.input.` and +its command subject `$SWARM.control.`, and nothing else. Besides the +subjects it publishes to, the credential may answer the one request it was just +sent, which is how a command gets its reply. The harness and the subagent daemon +connect with `_INBOX.agent.` as their inbox prefix, so the replies to the +icon write and the schedule read arrive. An agent can't read another agent's +terminal, state, input, commands or replies. The queue refuses the hive's shared +agent client outright. Hives and the swarm controller get a subscribe allow-list too, each with its own reply inbox `_INBOX.client..>`, which `swarm_queue_client::connect` diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 7052a663..a6634db5 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -209,6 +209,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. +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 +harness answers `{acted}`, which says whether a turn was running when the +cancel stopped it. It runs the same code as the hive dashboard's cancel button. +Unlike input, the harness answers a command, so swarm-controller returns that +answer as 200, 503 for no subscriber, and 504 when the harness held the command +and didn't answer within ten seconds. A 504 means unknown, not cancelled. +
Where the subagent list and rows come from diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 6f605d4f..012bdbfd 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -566,7 +566,7 @@ async fn serve_main(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) => { diff --git a/hive-agent/src/swarm_input.rs b/hive-agent/src/swarm_input.rs index 75acbe04..858f355c 100644 --- a/hive-agent/src/swarm_input.rs +++ b/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) { +use crate::events::Bus; + +/// Start the subscribers, if this agent has a label to name its subjects with. +pub fn spawn( + tx: UnboundedSender, + bus: Bus, + interrupted: std::sync::Arc, +) { 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, +) { + 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 { + match serde_json::from_slice::(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, agent: String) { @@ -88,8 +155,8 @@ fn delivered(payload: &[u8]) -> Option { #[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 { 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); + } } diff --git a/hive-agent/src/web_ui/actions.rs b/hive-agent/src/web_ui/actions.rs index cf399f57..c438933b 100644 --- a/hive-agent/src/web_ui/actions.rs +++ b/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) -> 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() } diff --git a/hive-agent/src/web_ui/mod.rs b/hive-agent/src/web_ui/mod.rs index ba036b23..a9fc5e7f 100644 --- a/hive-agent/src/web_ui/mod.rs +++ b/hive-agent/src/web_ui/mod.rs @@ -339,6 +339,35 @@ fn find_claude_child() -> Option { 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 diff --git a/swarm-controller/src/agent_control.rs b/swarm-controller/src/agent_control.rs new file mode 100644 index 00000000..5a3af000 --- /dev/null +++ b/swarm-controller/src/agent_control.rs @@ -0,0 +1,280 @@ +//! `POST /api/agents/{name}/cancel` — stop the turn an agent is in the middle +//! of. A command, not text: it travels on its own subject +//! ([`swarm_queue_client::agent_control_subject`]), never the one a turn's +//! message body does, and no hive is in the path - the agent's harness +//! subscribes itself. +//! +//! **Sent as a NATS request, and unlike input it is answered**: the harness +//! replies with whether a turn was actually running. That reply is the only +//! thing that distinguishes "cancelled" from "was idle", and it is the only +//! confirmation a command can have - so a request nobody answers within +//! [`swarm_queue_client::AGENT_CONTROL_REPLY_DEADLINE`] is a 504, never a +//! success. No subscriber at all is a 503, as for input. +//! +//! The sender is the `Remote-User` header, for the same reasons as +//! [`crate::agent_input`]; a command is refused without one. + +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::{HeaderMap, StatusCode}; +use problem_details::ProblemDetails; +use serde::Serialize; +use swarm_queue_client::{AgentControl, AgentControlReply}; +use utoipa::ToSchema; + +use crate::{AppState, error_problem}; + +/// What became of one command. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Outcome { + /// The agent answered. + Answered(AgentControlReply), + /// Nothing subscribes to the agent's control subject. + NotRunning, + /// A subscriber held the command and did not answer in time. + Unconfirmed, + /// The queue refused or failed the request, or the answer was unreadable. + Failed(String), +} + +/// Where a command goes: the queue in production, a fake in the tests. +pub(crate) trait ControlSink { + fn request(&self, subject: String, payload: Vec) -> impl Future + Send; +} + +impl ControlSink for async_nats::Client { + async fn request(&self, subject: String, payload: Vec) -> Outcome { + use async_nats::RequestErrorKind; + + let request = async_nats::Request::new() + .payload(payload.into()) + .timeout(Some(swarm_queue_client::AGENT_CONTROL_REPLY_DEADLINE)); + match self.send_request(subject, request).await { + Ok(reply) => match serde_json::from_slice(&reply.payload) { + Ok(reply) => Outcome::Answered(reply), + Err(e) => Outcome::Failed(format!("the agent's answer was unreadable: {e}")), + }, + Err(e) => match e.kind() { + RequestErrorKind::TimedOut => Outcome::Unconfirmed, + RequestErrorKind::NoResponders => Outcome::NotRunning, + _ => Outcome::Failed(e.to_string()), + }, + } + } +} + +/// The agent's answer to a cancel. +#[derive(Serialize, ToSchema)] +pub struct CancelResponse { + /// Whether a turn was running and was stopped. `false` means the agent + /// heard the command and had nothing to stop. + acted: bool, +} + +/// Stop the turn an agent is running. +#[utoipa::path( + post, + path = "/api/agents/{name}/cancel", + params(("name" = String, Path, description = "agent whose turn to cancel")), + responses( + (status = 200, description = "the agent answered: `acted` says whether a turn was \ + running and was stopped", body = CancelResponse), + (status = 400, description = "the agent name is not shaped like an identifier \ + (problem+json)", body = String), + (status = 401, description = "the request carries no `Remote-User` (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 = 504, description = "the harness held the command and did not answer in \ + time, so whether anything was cancelled is unknown (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_cancel( + State(state): State, + Path(agent): Path, + headers: HeaderMap, +) -> Result, ProblemDetails> { + let (agent, from) = target(&headers, &agent).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, &from, AgentControl::Cancel).await +} + +/// The agent and who is asking, or the refusal. Everything decided before the +/// queue is touched. +fn target( + headers: &HeaderMap, + agent: &str, +) -> Result<(String, String), (StatusCode, &'static str)> { + let from = crate::agent_input::sender(headers)?; + let agent = hive_types::Ident::parse(agent) + .map_err(|reason| (StatusCode::BAD_REQUEST, reason))? + .into_string(); + Ok((agent, from)) +} + +async fn send( + sink: &impl ControlSink, + agent: &str, + from: &str, + command: AgentControl, +) -> Result, ProblemDetails> { + let subject = swarm_queue_client::agent_control_subject(agent); + let payload = serde_json::to_vec(&command).map_err(|e| { + error_problem( + StatusCode::INTERNAL_SERVER_ERROR, + &format!("encoding the command failed: {e}"), + ) + })?; + match sink.request(subject.clone(), payload).await { + Outcome::Answered(reply) => { + tracing::info!(%agent, %from, ?command, acted = reply.acted, "agent command answered"); + Ok(Json(CancelResponse { acted: reply.acted })) + } + Outcome::NotRunning => Err(error_problem( + StatusCode::SERVICE_UNAVAILABLE, + &format!( + "{agent} is not listening for commands: its harness is not running or not connected to the swarm queue" + ), + )), + Outcome::Unconfirmed => { + tracing::warn!(%agent, %from, ?command, "agent command unanswered"); + Err(error_problem( + StatusCode::GATEWAY_TIMEOUT, + &format!( + "{agent} did not answer in time: whether anything was cancelled is unknown" + ), + )) + } + Outcome::Failed(e) => { + tracing::warn!(%agent, %subject, error = %e, "agent command 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::{AgentControl, AgentControlReply}; + + use super::{ControlSink, Outcome, send, target}; + + /// Answers every command with one fixed outcome and records what it got. + struct FakeSink { + answer: fn() -> Outcome, + got: Mutex)>>, + } + + impl FakeSink { + fn new(answer: fn() -> Outcome) -> Self { + Self { + answer, + got: Mutex::default(), + } + } + } + + impl ControlSink for FakeSink { + async fn request(&self, subject: String, payload: Vec) -> Outcome { + 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 + } + + #[test] + fn a_command_needs_a_sender_and_an_identifier() { + assert_eq!( + target(&from("mara"), "atlas").expect("valid"), + ("atlas".to_owned(), "mara".to_owned()) + ); + assert_eq!( + target(&HeaderMap::new(), "../x").expect_err("no sender").0, + StatusCode::UNAUTHORIZED + ); + assert_eq!( + target(&from("mara"), "../x").expect_err("bad agent").0, + StatusCode::BAD_REQUEST + ); + } + + #[tokio::test] + async fn the_answer_is_relayed_from_the_agent_s_own_control_subject() { + let answers: [(bool, fn() -> Outcome); 2] = [ + (true, || { + Outcome::Answered(AgentControlReply { acted: true }) + }), + (false, || { + Outcome::Answered(AgentControlReply { acted: false }) + }), + ]; + for (acted, answer) in answers { + let sink = FakeSink::new(answer); + let body = send(&sink, "atlas", "mara", AgentControl::Cancel) + .await + .expect("answered"); + assert_eq!(body.0.acted, acted); + let got = sink.got.lock().expect("unpoisoned"); + assert_eq!(got[0].0, "$SWARM.control.atlas"); + let sent: AgentControl = serde_json::from_slice(&got[0].1).expect("a command"); + assert_eq!(sent, AgentControl::Cancel); + } + } + + #[tokio::test] + async fn unanswered_is_504_not_success_and_the_rest_are_503_and_500() { + let status = |answer: fn() -> Outcome| async move { + send( + &FakeSink::new(answer), + "atlas", + "mara", + AgentControl::Cancel, + ) + .await + .err() + .expect("not a success") + .status + }; + assert_eq!( + status(|| Outcome::Unconfirmed).await, + Some(StatusCode::GATEWAY_TIMEOUT) + ); + assert_eq!( + status(|| Outcome::NotRunning).await, + Some(StatusCode::SERVICE_UNAVAILABLE) + ); + assert_eq!( + status(|| Outcome::Failed("refused".to_owned())).await, + Some(StatusCode::INTERNAL_SERVER_ERROR) + ); + } +} diff --git a/swarm-controller/src/agent_input.rs b/swarm-controller/src/agent_input.rs index 3e6bfef5..455e56a2 100644 --- a/swarm-controller/src/agent_input.rs +++ b/swarm-controller/src/agent_input.rs @@ -8,10 +8,11 @@ //! 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. +//! instead of text that silently goes nowhere. The agent does not reply to +//! input (it does to a command, see [`crate::agent_control`]). 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; @@ -119,6 +120,22 @@ pub(crate) async fn post_agent_input( send(&client, &agent, &input).await } +/// Who the gateway says is asking, or the 401 for a request that did not come +/// through it. Shared with [`crate::agent_control`]. +pub(crate) fn sender(headers: &HeaderMap) -> Result { + headers + .get(SENDER_HEADER) + .and_then(|v| v.to_str().ok()) + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_owned) + .ok_or(( + StatusCode::UNAUTHORIZED, + "no Remote-User header: the gateway names the sender, and this request \ + did not come through it", + )) +} + /// The agent and the message, or the refusal. Everything decided before the /// queue is touched. fn message( @@ -126,17 +143,7 @@ fn message( 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 from = sender(headers)?; let agent = hive_types::Ident::parse(agent) .map_err(|reason| (StatusCode::BAD_REQUEST, reason))? .into_string(); diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 23d22559..9f628bae 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -41,6 +41,7 @@ use swarm_authelia_bridge_sock::BridgeResponse; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; +mod agent_control; mod agent_icon; mod agent_identity; mod agent_input; @@ -3424,6 +3425,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_control::post_agent_cancel)) .routes(routes!(schedules::list_schedules, schedules::add_schedule)) .routes(routes!( schedules::edit_schedule, diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 0d773209..dc3c6e31 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -174,8 +174,16 @@ impl Policy { subscribe: Some(vec![ format!("{}.>", swarm_queue_client::agent_inbox_prefix(agent)), swarm_queue_client::agent_input_subject(agent), + swarm_queue_client::agent_control_subject(agent), ]), - responses: None, + // The controller sends a control command as a request, and the + // harness answers it. `allow_responses` lets it reply to that one + // request without a publish grant that would name the + // controller's inbox, which the agent has no business reaching. + // Operator text input is sent the same way but never answered + // (see `hive-agent`'s `swarm_input`), so the reply right is + // unused there. + responses: Some(swarm_queue_client::AGENT_CONTROL_REPLY_DEADLINE), }) } @@ -585,6 +593,11 @@ impl Policy { // `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(), + // Operator control commands to an agent (cancel its turn). A + // family of its own so the text-input grant never implies the + // power to stop a turn; same sole-publisher property, so a + // command needs no sender field to be trusted. + swarm_queue_client::AGENT_CONTROL_SUBJECT_WILDCARD.to_owned(), // A review classifier try-out, sent as a request: the reply comes // back on this client's inbox, which it already subscribes to. swarm_queue_client::review_classifier::SUBJECT.to_owned(), @@ -922,6 +935,49 @@ mod tests { ); } + #[test] + fn a_reader_may_publish_control_to_any_agent() { + let p = policy().permissions("swarm-controller").expect("a reader"); + assert!( + p.publish + .contains(&swarm_queue_client::AGENT_CONTROL_SUBJECT_WILDCARD.to_owned()), + "the controller is the only publisher of control commands; without \ + the grant its request is refused, and a refusal arrives as a timeout" + ); + } + + #[test] + fn nothing_but_the_controller_may_publish_control() { + // A command carries no sender: the harness acts on it because only + // the controller can publish it. A hive or an agent able to publish + // here could cancel any agent's turn. + let hive = policy().permissions("hive-alpha").expect("admitted"); + let agent = policy() + .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()]) + .expect("a per-agent template is valid") + .agent_token_permissions("alpha") + .expect("configured"); + for (who, p) in [("a hive", &hive), ("an agent", &agent)] { + assert!( + !p.publish.iter().any(|s| s.starts_with("$SWARM.control")), + "{who} must not publish control: {:?}", + p.publish + ); + } + } + + #[test] + fn an_agent_may_answer_the_command_it_was_sent() { + let p = policy() + .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()]) + .expect("a per-agent template is valid"); + let atlas = p.agent_token_permissions("atlas").expect("configured"); + assert_eq!( + atlas.responses, + Some(swarm_queue_client::AGENT_CONTROL_REPLY_DEADLINE) + ); + } + #[test] fn a_hive_may_not_publish_input_to_an_agent() { // The harness turns this message into a turn from the operator named in @@ -1082,8 +1138,8 @@ mod tests { } #[test] - fn an_agent_token_may_subscribe_to_its_own_inbox_and_input_only() { - use swarm_queue_client::{agent_inbox_prefix, agent_input_subject}; + fn an_agent_token_may_subscribe_to_its_own_inbox_input_and_control_only() { + use swarm_queue_client::{agent_control_subject, agent_inbox_prefix, agent_input_subject}; let p = policy() .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()]) @@ -1091,13 +1147,16 @@ mod tests { let atlas = p.agent_token_permissions("atlas").expect("configured"); assert!(may_subscribe(&atlas, &reply(&agent_inbox_prefix("atlas")))); assert!(may_subscribe(&atlas, &agent_input_subject("atlas"))); + assert!(may_subscribe(&atlas, &agent_control_subject("atlas"))); for refused in [ ">".to_owned(), "$SWARM.term.>".to_owned(), "$SWARM.term.argus".to_owned(), "$SWARM.term.atlas".to_owned(), "$SWARM.input.*".to_owned(), + "$SWARM.control.*".to_owned(), agent_input_subject("argus"), + agent_control_subject("argus"), reply(&agent_inbox_prefix("argus")), "_INBOX.nuid123.token456".to_owned(), ] { diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index bb932c50..9dce4ec2 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -299,6 +299,26 @@ 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"; +/// The subject the swarm controller sends an operator's control command for +/// `agent` on (see [`AgentControl`]). +/// +/// A family of its own, apart from [`agent_input_subject`]: a grant to send +/// text never covers a command that stops a turn, and the other way round. +/// Like the input subject it runs controller to agent over the swarm queue +/// with no hive in the path. +#[must_use] +pub fn agent_control_subject(agent: &str) -> String { + format!("{AGENT_CONTROL_SUBJECT_PREFIX}.{agent}") +} + +/// The publish grant covering every [`agent_control_subject`], for the one +/// client that may send them. A wildcard for the same no-roster reason as +/// [`AGENT_INPUT_SUBJECT_WILDCARD`]. +pub const AGENT_CONTROL_SUBJECT_WILDCARD: &str = "$SWARM.control.*"; + +/// Shared by [`agent_control_subject`] and [`AGENT_CONTROL_SUBJECT_WILDCARD`]. +const AGENT_CONTROL_SUBJECT_PREFIX: &str = "$SWARM.control"; + /// The subject `agent` publishes its terminal rows on. One leaf per agent, so /// a subscriber can follow one agent without filtering the swarm's whole /// terminal traffic. @@ -376,6 +396,37 @@ pub fn client_inbox_prefix(client_id: &str) -> String { format!("_INBOX.client.{client_id}") } +/// What an [`agent_control_subject`] message carries: one command for the +/// agent's harness. +/// +/// The harness acts on a command only because the controller is the one +/// client granted publish on the subject, so there is no sender field to +/// trust or forge here, unlike [`AgentInput::from`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AgentControl { + /// Interrupt the turn in flight, as the agent's own cancel button does. + Cancel, +} + +/// What an agent answers an [`AgentControl`] request with. +/// +/// The controller sends a command as a request, so a reply is the only +/// confirmation the operator gets that the harness took it. No reply within +/// [`AGENT_CONTROL_REPLY_DEADLINE`] means "not confirmed", which is not the +/// same as "refused". +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentControlReply { + /// Whether the command changed anything: `false` for a cancel that + /// found no turn in flight. + pub acted: bool, +} + +/// How long the queue lets an agent answer an [`AgentControl`] request, and +/// the controller waits for it. The harness answers as soon as it has acted, +/// so this only bounds a harness that is down or wedged. +pub const AGENT_CONTROL_REPLY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(10); + /// What an [`agent_input_subject`] message carries. #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct AgentInput {