From 1779fa056877151901ade8bb8541a55744281d47 Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 15 Nov 2026 17:23:22 +0100 Subject: [PATCH] swarm: the contract for an agent's bash task status and output A status key per task in the bash-task-status bucket and the task's output rows in a per-agent stream, kept apart as for subagents. The controller creates the bucket and the streams, and the agent credential may publish its own keys and subjects. Nothing publishes yet; the bash MCP half follows. --- docs/4-implementation/swarm/controller.md | 13 ++ nix/host-modules/swarm-nats.nix | 8 + nix/module-eval/nats-authelia.nix | 2 + swarm-controller/Cargo.toml | 1 + swarm-controller/src/subagent_term.rs | 13 ++ swarm-queue-client/Cargo.toml | 3 + swarm-queue-client/src/bash_task.rs | 87 ++++++++ swarm-queue-client/src/lib.rs | 11 +- swarm-wire/src/bash_task.rs | 250 ++++++++++++++++++++++ swarm-wire/src/lib.rs | 1 + 10 files changed, 388 insertions(+), 1 deletion(-) create mode 100644 swarm-queue-client/src/bash_task.rs create mode 100644 swarm-wire/src/bash_task.rs diff --git a/docs/4-implementation/swarm/controller.md b/docs/4-implementation/swarm/controller.md index 75b099c95..c5dc98ba3 100644 --- a/docs/4-implementation/swarm/controller.md +++ b/docs/4-implementation/swarm/controller.md @@ -190,6 +190,19 @@ publish `$KV.subagent-status..*` and nothing else of the bucket. Keeping this apart from the rows means a list of subagents reads one small value each and never a terminal stream. +Its bash MCP will publish the agent's background tasks the same way, with the +status kept apart from the output. A task's status is one key, `.`, in the +`bash-task-status` bucket: its `state` (`pending`, `running`, `done`, `timed_out`, +`interrupted` or `killed`), the command cut to 200 characters, when it began and when it +entered its state, the exit status, and `ts`, the time of the publish. A live task republishes +every 60 seconds, and a reader treats a live value older than three of those as stale. The output +goes to `$SWARM.term..bash.`, with a `Swarm-Stream` header of `stdout` or +`stderr`, into the stream `term-bash-` that swarm-controller creates for every declared +agent, for 24 hours or 64 MiB like the subagent rows. The agent's credential may publish +`$KV.bash-task-status..*` and `$SWARM.term..bash.>` and nothing else of +JetStream. Only the names, the grants and the two stores exist so far; the bash MCP doesn't +publish yet. + The queue would refuse a row too large for its `max_payload` outright and take the connection down with it, so the harness drops such a row's body before sending and leaves a marker in its place; the summary, level and icon still diff --git a/nix/host-modules/swarm-nats.nix b/nix/host-modules/swarm-nats.nix index 82575c05e..569f2ce42 100644 --- a/nix/host-modules/swarm-nats.nix +++ b/nix/host-modules/swarm-nats.nix @@ -995,6 +995,14 @@ in # and swarm-controller creates the bucket, so no # `$$JS.API.*` subject. "--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.subagent-status.{agent}.*"}" + # Its bash tasks (`swarm_wire::bash_task`): one status key per + # task in the `bash-task-status` KV bucket and the task's output + # rows. Publish only and no `$JS.API.*` subject: + # swarm-controller creates the bucket and the + # `term-bash-{agent}` stream, so their limits are never the + # agent's to choose. + "--agent-token-publish-subject ${lib.escapeShellArg "\$\$KV.bash-task-status.{agent}.*"}" + "--agent-token-publish-subject ${lib.escapeShellArg "\$\$SWARM.term.{agent}.bash.>"}" # Reading its own key of the `agent-schedules` KV bucket # (`swarm_queue_client::agent_schedules`), and nothing else of # JetStream: a direct get names the key in its subject, so diff --git a/nix/module-eval/nats-authelia.nix b/nix/module-eval/nats-authelia.nix index ea1f6441d..7c2e04f80 100644 --- a/nix/module-eval/nats-authelia.nix +++ b/nix/module-eval/nats-authelia.nix @@ -545,6 +545,8 @@ let "'$$KV.agents-md.{agent}'" "'$$SWARM.term.{agent}.sub.>'" "'$$KV.subagent-status.{agent}.*'" + "'$$KV.bash-task-status.{agent}.*'" + "'$$SWARM.term.{agent}.bash.>'" "'$$JS.API.DIRECT.GET.KV_agent-schedules.$$KV.agent-schedules.{agent}'" "'$$JS.API.DIRECT.GET.KV_agent-inference.$$KV.agent-inference.{agent}'" "'$$JS.API.DIRECT.GET.KV_agent-permissions.$$KV.agent-permissions.{agent}'" diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 90bdbc051..6167a770e 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -117,6 +117,7 @@ swarm-queue-client = { workspace = true, features = [ "kv", "agent-permissions", "subagent-term", + "bash-task", "agent-term", "forge-events", ] } diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index b662f7986..1fe1a320d 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -79,7 +79,20 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc, hives "subagent status: opening the bucket failed; next pass" ); } + if let Err(e) = swarm_queue_client::bash_task::open_status(&client).await { + tracing::warn!( + error = %swarm_queue_client::chain(&e), + "bash task status: opening the bucket failed; next pass" + ); + } for agent in live_agents(&wanted, &hives).await { + if let Err(e) = swarm_queue_client::bash_task::open_output(&client, &agent).await { + tracing::warn!( + %agent, + error = %swarm_queue_client::chain(&e), + "bash task output: creating the agent's stream failed; next pass" + ); + } if let Err(e) = swarm_queue_client::subagent_term::open_or_create(&client, &agent).await { diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 99c744f6e..1b4a72357 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -33,6 +33,9 @@ notices = ["async-nats/jetstream"] # `subagent_term::open_or_create`, for swarm-controller, which creates every # agent's subagent stream. Explicit for the same reason as `notices`. subagent-term = ["async-nats/jetstream"] +# `bash_task::open_status` and `open_output`, for swarm-controller, which creates +# the status bucket and every agent's task output stream. Needs `kv` for the bucket. +bash-task = ["kv"] # `agent_term::open_or_create` and the replay consumer, for swarm-controller, # which creates every agent's terminal stream and replays it to swarm-ui. # Explicit for the same reason as `notices`. diff --git a/swarm-queue-client/src/bash_task.rs b/swarm-queue-client/src/bash_task.rs new file mode 100644 index 000000000..b30593a98 --- /dev/null +++ b/swarm-queue-client/src/bash_task.rs @@ -0,0 +1,87 @@ +//! The connection side of an agent's bash tasks: the status bucket and the +//! output stream. The names, keys, subjects and the status value are +//! `swarm_wire::bash_task`. +//! +//! `swarm-controller` creates the bucket once and one output stream per agent +//! with the fixed config [`output_config`] spells. + +use swarm_wire::bash_task::{OUTPUT_MAX_AGE, output_stream_name, output_stream_subjects}; + +/// How large one agent's output stream may grow before the discard policy +/// reclaims space, as for the subagent rows. +const OUTPUT_MAX_BYTES: i64 = 64 * 1024 * 1024; + +/// The config [`open_output`] creates `agent`'s output stream with. +/// +/// `max_message_size` is the queue's live `max_payload` ([`crate::max_payload`]), +/// so the stream never rejects a row the queue itself accepted. +pub(crate) fn output_config( + agent: &str, + max_message_size: i32, +) -> async_nats::jetstream::stream::Config { + async_nats::jetstream::stream::Config { + name: output_stream_name(agent), + description: Some(format!("Output of {agent}'s bash tasks")), + subjects: vec![output_stream_subjects(agent)], + max_age: OUTPUT_MAX_AGE, + max_bytes: OUTPUT_MAX_BYTES, + discard: async_nats::jetstream::stream::DiscardPolicy::Old, + max_message_size, + ..Default::default() + } +} + +/// Open the status bucket, creating it if nothing has yet. `history: 1`: a +/// reader wants each task's current state, not every state it has been in. +pub async fn open_status( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + crate::kv::open_or_create( + &js, + swarm_wire::bash_task::BUCKET, + "bash-task-status", + "The state of each bash task, as its agent's bash MCP last published it".to_owned(), + ) + .await +} + +/// Open `agent`'s output stream, creating it if it does not exist yet. +pub async fn open_output( + client: &async_nats::Client, + agent: &str, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + let name = output_stream_name(agent); + match js.get_stream(&name).await { + Ok(stream) => Ok(stream), + Err(e) => { + tracing::info!(stream = %name, reason = %e, "bash task output stream not available, creating it"); + let max_message_size = i32::try_from(crate::max_payload(client)).unwrap_or(i32::MAX); + js.create_stream(output_config(agent, max_message_size)) + .await + .map_err(|source| crate::Error::CreateAgentStream { + stream: name, + source, + }) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_output_stream_caps_bytes_and_discards_the_oldest() { + let cfg = output_config("iris", 8_388_608); + assert_eq!(cfg.name, "term-bash-iris"); + assert_eq!(cfg.subjects, vec!["$SWARM.term.iris.bash.>".to_owned()]); + assert_eq!(cfg.max_bytes, OUTPUT_MAX_BYTES); + assert_eq!( + cfg.discard, + async_nats::jetstream::stream::DiscardPolicy::Old + ); + assert_eq!(cfg.max_message_size, 8_388_608); + } +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index b06d1a1bd..0da02d876 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -138,7 +138,11 @@ pub enum Error { source: async_nats::jetstream::context::CreateStreamError, }, - #[cfg(any(feature = "subagent-term", feature = "agent-term"))] + #[cfg(any( + feature = "subagent-term", + feature = "agent-term", + feature = "bash-task" + ))] #[error("creating the {stream} stream")] CreateAgentStream { // Owned: the name carries the agent's (`subagent_term::stream_name`, @@ -200,6 +204,11 @@ pub mod wanted; /// merge, are in `swarm_wire::agent_status`. pub mod agent_status; +/// The status bucket and the output stream of an agent's bash tasks. The names, +/// keys and values are `swarm_wire::bash_task`. Behind the `bash-task` feature. +#[cfg(feature = "bash-task")] +pub mod bash_task; + /// Opening the bucket each agent's subagent daemon publishes the run state of /// its subagents into: one key per subagent. The name, keys and value are /// `swarm_wire::subagent_status`; opening the bucket needs `kv`. diff --git a/swarm-wire/src/bash_task.rs b/swarm-wire/src/bash_task.rs new file mode 100644 index 000000000..b2a5005ff --- /dev/null +++ b/swarm-wire/src/bash_task.rs @@ -0,0 +1,250 @@ +//! What an agent's bash MCP tells the swarm about its background tasks, so the +//! swarm UI can list the running ones and read their output without calling +//! the agent. It follows the subagents: a small status per task, kept apart +//! from the task's output so listing the tasks never reads an output stream. +//! +//! **Status** is one key per task in the [`BUCKET`] KV bucket, `.`, +//! written with the agent's own queue credential, which may publish +//! `$KV.bash-task-status..*` and nothing else of `JetStream`. The writer +//! publishes to [`status_subject`] directly instead of opening the bucket; +//! `swarm-controller` creates it (`swarm_queue_client::bash_task::open_status`). +//! +//! **Output** is rows on [`output_subject`], `$SWARM.term..bash.`, +//! in the stream [`output_stream_name`] that `swarm-controller` creates per +//! agent (`swarm_queue_client::bash_task::open_output`). The agent may publish +//! `$SWARM.term..bash.>` and nothing else of `JetStream`. The family is +//! disjoint from the agent's own terminal (`$SWARM.term.`) and its +//! subagents' (`.sub.`), so none of their readers receives these rows. +//! +//! A task id is a single subject token: the bash MCP's task names are +//! `[a-z0-9-]` and its generated ids are hex. + +use serde::{Deserialize, Serialize}; + +use crate::TERM_SUBJECT_PREFIX; + +/// The KV bucket the task statuses are published into. +pub const BUCKET: &str = "bash-task-status"; + +/// How often the writer republishes a `running` status while the task's +/// process lives, so a writer that died shows up as stale. +pub const HEARTBEAT_S: i64 = 60; + +/// A `running` status whose [`Status::ts`] is older than this is stale. +pub const STALE_AFTER_S: i64 = 3 * HEARTBEAT_S; + +/// The longest [`Status::cmd`], in characters; a longer command is cut. +pub const CMD_MAX_CHARS: usize = 200; + +/// The token between the agent and the task id in an output subject. +const BASH_TOKEN: &str = "bash"; + +/// How long the output stream keeps a row. +pub const OUTPUT_MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24); + +/// The message header on an output row naming the stream it came from: +/// `stdout` or `stderr`. +pub const HEADER_STREAM: &str = "Swarm-Stream"; + +/// The task's lifecycle state. The names are `hive_sh4re::bash_task::TaskStatus`'s, +/// so a writer maps one to the other without a table. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum State { + /// Accepted, not started. + Pending, + /// The process is running. + Running, + /// It exited by itself; [`Status::exit`] says how. + Done, + /// It ran past its timeout and was stopped. + TimedOut, + /// The daemon restarted while it ran, so the process is gone. + Interrupted, + /// It was killed on request. + Killed, +} + +impl State { + /// Whether the task can still change: the writer keeps republishing it. + #[must_use] + pub fn is_live(self) -> bool { + matches!(self, Self::Pending | Self::Running) + } +} + +/// The value under one task's key. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Status { + /// The task's lifecycle state. + pub state: State, + /// The command, cut to [`CMD_MAX_CHARS`] characters. + pub cmd: String, + /// Unix seconds when the task was accepted. + pub created: i64, + /// Unix seconds when the task entered [`Self::state`]. + pub since: i64, + /// The exit status once the process exited by itself; `None` otherwise, + /// and when a signal ended it. + pub exit: Option, + /// Unix seconds when the writer published this value; a live status + /// refreshes it every [`HEARTBEAT_S`]. + pub ts: i64, +} + +impl Status { + /// Whether a reader at `now_s` should show the task as running: live and + /// refreshed within [`STALE_AFTER_S`]. + #[must_use] + pub fn is_running(&self, now_s: i64) -> bool { + self.state.is_live() && now_s - self.ts <= STALE_AFTER_S + } +} + +/// `cmd` cut to [`CMD_MAX_CHARS`] characters, with `…` when it was cut. +#[must_use] +pub fn clip_cmd(cmd: &str) -> String { + if cmd.chars().count() <= CMD_MAX_CHARS { + return cmd.to_owned(); + } + let mut cut: String = cmd.chars().take(CMD_MAX_CHARS - 1).collect(); + cut.push('…'); + cut +} + +/// The key of `task` under `agent`: `.`. +#[must_use] +pub fn key(agent: &str, task: &str) -> String { + format!("{agent}.{task}") +} + +/// The subject `task`'s status is written on: `$KV..`. +#[must_use] +pub fn status_subject(agent: &str, task: &str) -> String { + format!("$KV.{BUCKET}.{}", key(agent, task)) +} + +/// The task a bucket key names under `agent`, or `None` for a key that is not +/// exactly one task token below `agent`. +#[must_use] +pub fn task_from_key<'a>(agent: &str, key: &'a str) -> Option<&'a str> { + let name = key.strip_prefix(agent)?.strip_prefix('.')?; + (!name.is_empty() && !name.contains('.')).then_some(name) +} + +/// The stream holding `agent`'s task output: `term-bash-`. +#[must_use] +pub fn output_stream_name(agent: &str) -> String { + format!("term-bash-{agent}") +} + +/// The subject `task`'s output rows are published on. +#[must_use] +pub fn output_subject(agent: &str, task: &str) -> String { + format!("{TERM_SUBJECT_PREFIX}.{agent}.{BASH_TOKEN}.{task}") +} + +/// Every [`output_subject`] of `agent` at once: the stream's subjects and the +/// publish grant's shape. +#[must_use] +pub fn output_stream_subjects(agent: &str) -> String { + format!("{TERM_SUBJECT_PREFIX}.{agent}.{BASH_TOKEN}.>") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_names_match_the_queue_grant() { + assert_eq!( + status_subject("iris", "build"), + "$KV.bash-task-status.iris.build" + ); + assert_eq!(output_stream_name("iris"), "term-bash-iris"); + assert_eq!( + output_subject("iris", "build"), + "$SWARM.term.iris.bash.build" + ); + assert_eq!(output_stream_subjects("iris"), "$SWARM.term.iris.bash.>"); + } + + /// The output family never overlaps the subagent family or the agent's own. + #[test] + fn the_output_subjects_are_disjoint_from_the_other_terminals() { + assert_ne!( + output_stream_subjects("iris"), + crate::subagent_term::stream_subjects("iris") + ); + assert!( + !output_subject("iris", "x").starts_with(&crate::subagent_term::subject("iris", "")) + ); + } + + #[test] + fn a_key_names_one_task_of_its_agent_or_nothing() { + assert_eq!(task_from_key("iris", "iris.build"), Some("build")); + assert_eq!(task_from_key("iris", "iris.a.b"), None); + assert_eq!(task_from_key("iris", "iris."), None); + assert_eq!(task_from_key("iris", "irisx.build"), None); + } + + #[test] + fn only_a_pending_or_running_task_is_live() { + assert!(State::Pending.is_live() && State::Running.is_live()); + for s in [ + State::Done, + State::TimedOut, + State::Interrupted, + State::Killed, + ] { + assert!(!s.is_live(), "{s:?}"); + } + } + + #[test] + fn a_live_status_goes_stale_after_three_missed_heartbeats() { + let status = |state, ts| Status { + state, + cmd: "x".to_owned(), + created: 0, + since: 0, + exit: None, + ts, + }; + let live = status(State::Running, 10_000); + assert!(live.is_running(10_000 + STALE_AFTER_S)); + assert!(!live.is_running(10_000 + STALE_AFTER_S + 1)); + assert!(!status(State::Done, 10_000).is_running(10_000)); + } + + #[test] + fn a_long_command_is_cut_with_an_ellipsis() { + assert_eq!(clip_cmd("ls"), "ls"); + let long = "x".repeat(CMD_MAX_CHARS + 50); + let cut = clip_cmd(&long); + assert_eq!(cut.chars().count(), CMD_MAX_CHARS); + assert!(cut.ends_with('…')); + assert_eq!( + clip_cmd(&"y".repeat(CMD_MAX_CHARS)).chars().count(), + CMD_MAX_CHARS + ); + } + + #[test] + fn the_wire_form_is_snake_case_json() { + let wire = serde_json::to_string(&Status { + state: State::TimedOut, + cmd: "sleep 9".to_owned(), + created: 1, + since: 2, + exit: None, + ts: 3, + }) + .expect("encodes"); + assert_eq!( + wire, + r#"{"state":"timed_out","cmd":"sleep 9","created":1,"since":2,"exit":null,"ts":3}"# + ); + } +} diff --git a/swarm-wire/src/lib.rs b/swarm-wire/src/lib.rs index 589d0bd61..cbcaac9bd 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -16,6 +16,7 @@ pub mod agent_permissions; pub mod agent_schedules; pub mod agent_status; pub mod agent_term; +pub mod bash_task; pub mod builder; pub mod claude_login; pub mod egress_allow;