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

@@ -190,6 +190,19 @@ publish `$KV.subagent-status.<agent>.*` 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, `<agent>.<task>`, 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.<agent>.bash.<task>`, with a `Swarm-Stream` header of `stdout` or
+`stderr`, into the stream `term-bash-<agent>` 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.<agent>.*` and `$SWARM.term.<agent>.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

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

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}'"

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",
 ] }

swarm-controller/src/subagent_term.rs

@@ -79,7 +79,20 @@ pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, 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
                 {

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`.

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<async_nats::jetstream::kv::Store, crate::Error> {
+    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<async_nats::jetstream::stream::Stream, crate::Error> {
+    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);
+    }
+}

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`.

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, `<agent>.<task>`,
+//! written with the agent's own queue credential, which may publish
+//! `$KV.bash-task-status.<agent>.*` 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.<agent>.bash.<task>`,
+//! 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.<agent>.bash.>` and nothing else of `JetStream`. The family is
+//! disjoint from the agent's own terminal (`$SWARM.term.<agent>`) 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<i32>,
+    /// 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`: `<agent>.<task>`.
+#[must_use]
+pub fn key(agent: &str, task: &str) -> String {
+    format!("{agent}.{task}")
+}
+
+/// The subject `task`'s status is written on: `$KV.<bucket>.<key>`.
+#[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-<agent>`.
+#[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}"#
+        );
+    }
+}

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;