swarm: list an agent's bash tasks from their published status

GET /api/agents/{name}/bash-tasks reads the agent's keys of the bash-task-status
bucket and reports each task's state, command and times, with a live task whose
daemon stopped reporting as stale. The swarm UI shows them in a background tasks
section of the agent card, running ones first. The task output stays unread.

swarm-controller/src/bash_task.rs

@@ -0,0 +1,320 @@
+//! An agent's bash tasks, as its bash MCP published them: the list the swarm
+//! UI shows.
+//!
+//! The bash MCP writes one key per task, `<agent>.<task>`, into the
+//! `bash-task-status` bucket (`swarm_wire::bash_task`), as the agent. This
+//! reads the keys of one agent and reports each task's state. The task's
+//! output is kept apart, in `term-bash-<agent>`, and is not read here.
+//!
+//! A key expires `STATUS_MAX_AGE` after its last write, so the bucket holds
+//! the tasks of the last few days, and listing one agent's means going through
+//! every key of the bucket for the ones with its prefix. That is fine at this
+//! size; a filtered watch would avoid it if the bucket grows.
+
+use std::time::Duration;
+
+use axum::Json;
+use axum::extract::{Path, State};
+use axum::http::StatusCode;
+use futures_util::{StreamExt as _, TryStreamExt as _};
+use serde::Serialize;
+use swarm_wire::bash_task::{self, State as TaskState, Status};
+use utoipa::ToSchema;
+
+use crate::routes::AppState;
+
+/// How long one read of a key may take.
+const READ_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// How many keys are read at once.
+const READ_CONCURRENCY: usize = 16;
+
+/// The most tasks one list returns, newest first. An agent that ran more in
+/// the window shows its newest.
+const MAX_TASKS: usize = 200;
+
+/// A task's state as of the request: the published state, with a live one
+/// whose writer stopped reporting called `stale`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum TaskView {
+    /// Accepted, not started, and the writer is reporting.
+    Pending,
+    /// The process is running and the writer is reporting.
+    Running,
+    /// A live state, but the writer has not reported for three heartbeats:
+    /// the daemon died or lost the queue.
+    Stale,
+    /// It exited by itself.
+    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 TaskView {
+    fn of(status: &Status, now_s: i64) -> Self {
+        if status.state.is_live() && !status.is_running(now_s) {
+            return Self::Stale;
+        }
+        match status.state {
+            TaskState::Pending => Self::Pending,
+            TaskState::Running => Self::Running,
+            TaskState::Done => Self::Done,
+            TaskState::TimedOut => Self::TimedOut,
+            TaskState::Interrupted => Self::Interrupted,
+            TaskState::Killed => Self::Killed,
+        }
+    }
+}
+
+/// One bash task of an agent, as the list endpoint reports it.
+#[derive(Debug, PartialEq, Eq, Serialize, ToSchema)]
+pub(crate) struct BashTaskEntry {
+    /// The task's id.
+    pub task: String,
+    /// What the task is doing, as of the request.
+    pub state: TaskView,
+    /// The command, cut to 200 characters.
+    pub cmd: String,
+    /// Unix seconds when the task was accepted.
+    pub created_unix: i64,
+    /// Unix seconds when the task entered its published state.
+    pub since_unix: i64,
+    /// The exit status once the process exited by itself.
+    pub exit: Option<i32>,
+}
+
+/// The entry for `task`, or `None` for a value that is not a status.
+fn entry_of(task: &str, status: &Status, now_s: i64) -> BashTaskEntry {
+    BashTaskEntry {
+        task: task.to_owned(),
+        state: TaskView::of(status, now_s),
+        cmd: status.cmd.clone(),
+        created_unix: status.created,
+        since_unix: status.since,
+        exit: status.exit,
+    }
+}
+
+/// Newest first, by when the task was accepted, ties by id; cut to
+/// [`MAX_TASKS`].
+fn newest_first(mut entries: Vec<BashTaskEntry>) -> Vec<BashTaskEntry> {
+    entries.sort_by(|a, b| {
+        b.created_unix
+            .cmp(&a.created_unix)
+            .then_with(|| a.task.cmp(&b.task))
+    });
+    entries.truncate(MAX_TASKS);
+    entries
+}
+
+#[utoipa::path(
+    get,
+    path = "/api/agents/{name}/bash-tasks",
+    params(("name" = String, Path, description = "agent whose bash tasks to list")),
+    responses(
+        (status = 200, description = "the agent's bash tasks of the last few days, newest \
+            first, each with its state as its bash MCP last published it; empty when the \
+            agent has published none",
+            body = Vec<BashTaskEntry>),
+        (status = 400, description = "the agent name is not shaped like an identifier \
+            (problem+json)", body = String),
+        (status = 503, description = "no swarm queue is configured on this host, or the \
+            queue could not be reached (problem+json)", body = String),
+        (status = 500, description = "reading the status bucket failed (problem+json)",
+            body = String),
+    ),
+    tag = "agents"
+)]
+pub(crate) async fn list_bash_tasks(
+    State(state): State<AppState>,
+    Path(agent): Path<String>,
+) -> Result<Json<Vec<BashTaskEntry>>, problem_details::ProblemDetails> {
+    let agent = hive_types::Ident::parse(&agent)
+        .map(hive_types::Ident::into_string)
+        .map_err(|reason| crate::routes::error_problem(StatusCode::BAD_REQUEST, reason))?;
+    let status = state.status.as_ref().ok_or_else(|| {
+        crate::routes::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| {
+        crate::routes::error_problem(
+            StatusCode::SERVICE_UNAVAILABLE,
+            &swarm_queue_client::chain(&e),
+        )
+    })?;
+    let js = async_nats::jetstream::new(client);
+    let store = match js.get_key_value(bash_task::BUCKET).await {
+        Ok(store) => store,
+        // The controller creates the bucket on its first pass; until then
+        // nothing has published, so the list is empty.
+        Err(e) if is_bucket_not_found(&e) => return Ok(Json(Vec::new())),
+        Err(e) => return Err(read_failed(&agent, &e)),
+    };
+    let tasks: Vec<String> = store
+        .keys()
+        .await
+        .map_err(|e| read_failed(&agent, &e))?
+        .try_filter_map(|key| {
+            std::future::ready(Ok(bash_task::task_from_key(&agent, &key).map(str::to_owned)))
+        })
+        .try_collect()
+        .await
+        .map_err(|e| read_failed(&agent, &e))?;
+    let now_s = unix_now();
+    // Owned items, as `list_subagents` does: a borrowed item type trips a
+    // "`FnOnce` is not general enough" error where `routes!` checks the handler.
+    let items: Vec<(async_nats::jetstream::kv::Store, String, String)> = tasks
+        .iter()
+        .map(|task| (store.clone(), agent.clone(), task.clone()))
+        .collect();
+    let entries: Vec<BashTaskEntry> = futures_util::stream::iter(items)
+        .map(|(store, agent, task)| async move { read_status(&store, &agent, &task, now_s).await })
+        .buffered(READ_CONCURRENCY)
+        .filter_map(std::future::ready)
+        .collect()
+        .await;
+    Ok(Json(newest_first(entries)))
+}
+
+/// One task's entry, or `None` when its key is gone (it expired between the
+/// listing and this read), cannot be read, or is not a status.
+async fn read_status(
+    store: &async_nats::jetstream::kv::Store,
+    agent: &str,
+    task: &str,
+    now_s: i64,
+) -> Option<BashTaskEntry> {
+    let key = bash_task::key(agent, task);
+    match tokio::time::timeout(READ_TIMEOUT, store.get(&key)).await {
+        Ok(Ok(Some(value))) => match serde_json::from_slice::<Status>(&value) {
+            Ok(status) => Some(entry_of(task, &status, now_s)),
+            Err(e) => {
+                tracing::warn!(%agent, %task, error = %e, "listing bash tasks: the status is not readable");
+                None
+            }
+        },
+        Ok(Ok(None)) => None,
+        Ok(Err(e)) => {
+            tracing::warn!(%agent, %task, error = %swarm_queue_client::chain(&e), "listing bash tasks: reading the status failed");
+            None
+        }
+        Err(_) => {
+            tracing::warn!(%agent, %task, "listing bash tasks: no answer reading the status");
+            None
+        }
+    }
+}
+
+/// Seconds since the unix epoch, by this host's clock.
+fn unix_now() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX))
+}
+
+/// Whether the bucket's stream does not exist yet, as against a read that
+/// failed: the one case that lists nothing instead of failing.
+fn is_bucket_not_found(e: &async_nats::jetstream::context::KeyValueError) -> bool {
+    use std::error::Error as _;
+    e.source()
+        .and_then(|source| source.downcast_ref::<async_nats::jetstream::context::GetStreamError>())
+        .is_some_and(|err| {
+            matches!(
+                err.kind(),
+                async_nats::jetstream::context::GetStreamErrorKind::JetStream(js)
+                    if js.error_code() == async_nats::jetstream::ErrorCode::STREAM_NOT_FOUND
+            )
+        })
+}
+
+fn read_failed(agent: &str, e: &dyn std::error::Error) -> problem_details::ProblemDetails {
+    let detail = swarm_queue_client::chain(e);
+    tracing::warn!(%agent, error = %detail, "listing bash tasks: reading the bucket failed");
+    crate::routes::error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn status(state: TaskState, created: i64, ts: i64) -> Status {
+        Status {
+            state,
+            cmd: "make".to_owned(),
+            created,
+            since: created + 1,
+            exit: None,
+            ts,
+        }
+    }
+
+    const NOW: i64 = 1_000_000;
+
+    /// A live state whose writer keeps reporting is shown as it is; one that
+    /// stopped is stale; a finished state is shown as it is, however old.
+    #[test]
+    fn a_live_task_goes_stale_and_a_finished_one_does_not() {
+        let fresh = status(TaskState::Running, NOW - 500, NOW - 10);
+        assert_eq!(TaskView::of(&fresh, NOW), TaskView::Running);
+        let queued = status(TaskState::Pending, NOW - 500, NOW - 10);
+        assert_eq!(TaskView::of(&queued, NOW), TaskView::Pending);
+        let silent = status(TaskState::Running, NOW - 5000, NOW - 4000);
+        assert_eq!(TaskView::of(&silent, NOW), TaskView::Stale);
+        for (state, view) in [
+            (TaskState::Done, TaskView::Done),
+            (TaskState::TimedOut, TaskView::TimedOut),
+            (TaskState::Interrupted, TaskView::Interrupted),
+            (TaskState::Killed, TaskView::Killed),
+        ] {
+            assert_eq!(TaskView::of(&status(state, 0, 0), NOW), view);
+        }
+    }
+
+    #[test]
+    fn an_entry_carries_the_published_fields() {
+        let mut s = status(TaskState::Done, 100, 200);
+        s.exit = Some(3);
+        let e = entry_of("build", &s, NOW);
+        assert_eq!(
+            e,
+            BashTaskEntry {
+                task: "build".to_owned(),
+                state: TaskView::Done,
+                cmd: "make".to_owned(),
+                created_unix: 100,
+                since_unix: 101,
+                exit: Some(3),
+            }
+        );
+    }
+
+    /// Newest first by when the task was accepted, ties by id, and no more than
+    /// the cap.
+    #[test]
+    fn the_list_is_newest_first_and_capped() {
+        let mut entries: Vec<BashTaskEntry> = (0..MAX_TASKS + 5)
+            .map(|i| {
+                entry_of(
+                    &format!("t{i:04}"),
+                    &status(TaskState::Done, i64::try_from(i).expect("small"), 0),
+                    NOW,
+                )
+            })
+            .collect();
+        entries.push(entry_of("same-a", &status(TaskState::Done, 1000, 0), NOW));
+        entries.push(entry_of("same-b", &status(TaskState::Done, 1000, 0), NOW));
+        let list = newest_first(entries);
+        assert_eq!(list.len(), MAX_TASKS);
+        assert_eq!(list[0].task, "same-a");
+        assert_eq!(list[1].task, "same-b");
+        assert_eq!(list[2].task, format!("t{:04}", MAX_TASKS + 4));
+    }
+}

swarm-controller/src/main.rs

@@ -45,6 +45,7 @@ mod attic_copy;
 mod auth;
 mod backend_egress;
 mod backend_env;
+mod bash_task;
 mod builder;
 mod cache_miss;
 mod claude_login;

swarm-controller/src/startup.rs

@@ -28,11 +28,11 @@ use crate::routes::{self, AppState, agents, config_prs, forge_admin, hives, jobq
 use crate::{
     agent_control, agent_custom, agent_icon, agent_input, agent_limits, agent_login,
     agent_permissions, agent_renewal, agent_state_stream, agent_status, agents_md, alerts,
-    archetypes, auth, backend_env, builder, claude_login, config_drift, config_pr, egress_allow,
-    egress_deny, forge, forge_account, github_account, gitlab_account, inference, issue_report,
-    linked_accounts, matrix_account, message_stream, notify_flows, notify_router_api, pause_hold,
-    pr_report, pulls, queue_identity, review_classifier, schedules, status, subagent_term,
-    term_stream, wanted, webhook,
+    archetypes, auth, backend_env, bash_task, builder, claude_login, config_drift, config_pr,
+    egress_allow, egress_deny, forge, forge_account, github_account, gitlab_account, inference,
+    issue_report, linked_accounts, matrix_account, message_stream, notify_flows, notify_router_api,
+    pause_hold, pr_report, pulls, queue_identity, review_classifier, schedules, status,
+    subagent_term, term_stream, wanted, webhook,
 };
 
 /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`.
@@ -665,6 +665,7 @@ pub(crate) fn build_app(state: AppState) -> axum::Router {
             .routes(routes!(term_stream::stream_agent_term))
             .routes(routes!(message_stream::stream_messages))
             .routes(routes!(subagent_term::list_subagents))
+            .routes(routes!(bash_task::list_bash_tasks))
             .routes(routes!(subagent_term::stream_subagent_term))
             .routes(routes!(agent_state_stream::stream_agent_state))
             .routes(routes!(agent_input::post_agent_input))