From 5de20f31dd1802d75308587da250f2bd7c4090d3 Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 15 Nov 2026 18:36:44 +0100 Subject: [PATCH] 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. --- docs/4-implementation/swarm/controller.md | 3 + docs/4-implementation/web-ui/swarm-ui.md | 14 + .../swarm-ui/src/agent/detail/AgentDetail.tsx | 5 +- .../swarm-ui/src/agent/detail/sections.ts | 12 +- .../swarm-ui/src/agent/terminal/BashTasks.css | 38 +++ .../swarm-ui/src/agent/terminal/BashTasks.tsx | 60 ++++ .../swarm-ui/src/agent/terminal/bash-tasks.ts | 75 ++++ .../src/agent/terminal/useBashTasks.ts | 23 ++ .../swarm-ui/test/bash-tasks.test.mjs | 69 ++++ .../swarm-ui/test/detail-sections.test.mjs | 5 +- swarm-controller/src/bash_task.rs | 320 ++++++++++++++++++ swarm-controller/src/main.rs | 1 + swarm-controller/src/startup.rs | 11 +- 13 files changed, 625 insertions(+), 11 deletions(-) create mode 100644 frontend/packages/swarm-ui/src/agent/terminal/BashTasks.css create mode 100644 frontend/packages/swarm-ui/src/agent/terminal/BashTasks.tsx create mode 100644 frontend/packages/swarm-ui/src/agent/terminal/bash-tasks.ts create mode 100644 frontend/packages/swarm-ui/src/agent/terminal/useBashTasks.ts create mode 100644 frontend/packages/swarm-ui/test/bash-tasks.test.mjs create mode 100644 swarm-controller/src/bash_task.rs diff --git a/docs/4-implementation/swarm/controller.md b/docs/4-implementation/swarm/controller.md index bb5c45194..7238e96e8 100644 --- a/docs/4-implementation/swarm/controller.md +++ b/docs/4-implementation/swarm/controller.md @@ -203,6 +203,9 @@ goes to `$SWARM.term..bash.`, with a `Swarm-Stream` header of `stdo 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. The bash daemon publishes both (see [`bash.md`](../agent-tools/bash.md)). +`GET /api/agents//bash-tasks` lists an agent's tasks from the status keys, newest first and at +most 200, each with its `state` (the published one, or `stale` when a live task's daemon stopped +reporting), `cmd`, `created_unix`, `since_unix` and `exit`. The output stream isn't read 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 diff --git a/docs/4-implementation/web-ui/swarm-ui.md b/docs/4-implementation/web-ui/swarm-ui.md index 022132aac..ac9254def 100644 --- a/docs/4-implementation/web-ui/swarm-ui.md +++ b/docs/4-implementation/web-ui/swarm-ui.md @@ -595,6 +595,20 @@ store identity, or no queue address, publishes no subagent terminals. +
Where the background task list comes from + +An opened agent card has a **background tasks** section, below the subagents, listing the +tasks the agent started through the bash MCP. Its bash MCP publishes one status per task +to the `bash-task-status` bucket, and swarm-controller serves the list as +`GET /api/agents//bash-tasks`: each task's id, command, state, the time the agent queued it and +the time it entered its state, and the exit status. The state is the published +one, or `stale` for a live task whose daemon stopped reporting. The section lists the running +tasks first, then the rest, each newest first. Before the age it says `started`, +`finished`, `exited N`, `timed out`, `killed` or `stopped reporting`. It refetches +every ten seconds and shows no output, which is a separate stream. + +
+
Where an agent's terminal history comes from The harness publishes the agent's terminal rows on `$SWARM.term.`, into diff --git a/frontend/packages/swarm-ui/src/agent/detail/AgentDetail.tsx b/frontend/packages/swarm-ui/src/agent/detail/AgentDetail.tsx index 8ae96db6c..68df78a21 100644 --- a/frontend/packages/swarm-ui/src/agent/detail/AgentDetail.tsx +++ b/frontend/packages/swarm-ui/src/agent/detail/AgentDetail.tsx @@ -42,6 +42,7 @@ import { startsOpen, type DetailSection, } from "./sections.js"; +import { BashTasks } from "../terminal/BashTasks.js"; import { SubagentTerms } from "../terminal/SubagentTerms.js"; import { WantedMenu } from "../roster/WantedMenu.js"; import "./AgentDetail.css"; @@ -313,6 +314,8 @@ function SectionBody({ return ; case "subagents": return ; + case "tasks": + return ; } } @@ -324,7 +327,7 @@ export function AgentDetail({ }: { row: AgentRow; actions: AgentActions; - /** Add the live terminal and subagents sections. */ + /** Add the live terminal, subagents and background tasks sections. */ withTerminal?: boolean; /** * An opened card, whose header already shows the state rows. Without it diff --git a/frontend/packages/swarm-ui/src/agent/detail/sections.ts b/frontend/packages/swarm-ui/src/agent/detail/sections.ts index 39e17476d..50ce94b07 100644 --- a/frontend/packages/swarm-ui/src/agent/detail/sections.ts +++ b/frontend/packages/swarm-ui/src/agent/detail/sections.ts @@ -3,24 +3,30 @@ // agent's state, message and wanted state, so the sections hold the rest. /** One folded block of an opened agent card. */ -export type DetailSection = "status" | "terminal" | "subagents" | "settings"; +export type DetailSection = + | "status" + | "terminal" + | "subagents" + | "tasks" + | "settings"; /** What each section's header says. */ export const SECTION_TITLE: Record = { status: "status", terminal: "terminal", subagents: "subagents", + tasks: "background tasks", settings: "settings", }; /** - * The sections a card shows, top to bottom. The terminal and subagents are + * The sections a card shows, top to bottom. The terminal, subagents and tasks are * left out where the page is already beside a terminal (the terminal page's * details popover). */ export function sectionsFor(withTerminal: boolean): DetailSection[] { return withTerminal - ? ["status", "terminal", "subagents", "settings"] + ? ["status", "terminal", "subagents", "tasks", "settings"] : ["status", "settings"]; } diff --git a/frontend/packages/swarm-ui/src/agent/terminal/BashTasks.css b/frontend/packages/swarm-ui/src/agent/terminal/BashTasks.css new file mode 100644 index 000000000..714ac457e --- /dev/null +++ b/frontend/packages/swarm-ui/src/agent/terminal/BashTasks.css @@ -0,0 +1,38 @@ +/* — the rows of an agent's background bash tasks. */ +.ui-bash-tasks-count { + margin: 0 0 0.5em; + color: var(--muted); +} + +.ui-bash-tasks-list { + display: flex; + flex-direction: column; + gap: 0.35em; + margin: 0; + padding: 0; + list-style: none; +} + +.ui-bash-tasks-item { + display: grid; + grid-template-columns: minmax(0, 1fr) auto auto; + align-items: baseline; + gap: 0.75em; +} + +.ui-bash-tasks-item-over { + opacity: 0.6; +} + +.ui-bash-tasks-cmd { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.ui-bash-tasks-id, +.ui-bash-tasks-age { + color: var(--muted); + font-size: 0.85em; + white-space: nowrap; +} diff --git a/frontend/packages/swarm-ui/src/agent/terminal/BashTasks.tsx b/frontend/packages/swarm-ui/src/agent/terminal/BashTasks.tsx new file mode 100644 index 000000000..ef3ddf86c --- /dev/null +++ b/frontend/packages/swarm-ui/src/agent/terminal/BashTasks.tsx @@ -0,0 +1,60 @@ +// — an agent's background bash tasks, in its card's "background +// tasks" section: the running ones first, each with its command, what state it +// is in and since when. The list is `GET /api/agents/{name}/bash-tasks` +// (`useBashTasks`), which swarm-controller builds from the status each task's +// bash MCP publishes. Read-only; a task's output is not shown here. +// +// Mount it keyed by agent name: the refresh timer is armed once per mount. +import { ApiErrorPanel } from "@hive/shared/api-error-panel.js"; +import { Hint } from "../../ui/hint/Hint.js"; +import { RelativeTime } from "../../ui/relative-time/RelativeTime.js"; +import { isLive, runningTasksLabel, stateWord } from "./bash-tasks.js"; +import { useBashTasks } from "./useBashTasks.js"; +import "./BashTasks.css"; + +export function BashTasks({ agentName }: { agentName: string }) { + const { tasks, error } = useBashTasks(agentName); + return ( +
+ {error && ( + + )} + {tasks === null ? ( + error ? null : ( + … + ) + ) : tasks.length === 0 ? ( + no background tasks + ) : ( + <> +

{runningTasksLabel(tasks)}

+
    + {tasks.map((t) => ( +
  • + + {t.cmd} + + {t.task} + + {stateWord(t)} + {t.state === "stale" ? null : ( + <> + {" "} + + + )} + +
    • + ))} +
+ + )} +
+ ); +} diff --git a/frontend/packages/swarm-ui/src/agent/terminal/bash-tasks.ts b/frontend/packages/swarm-ui/src/agent/terminal/bash-tasks.ts new file mode 100644 index 000000000..f2a7c73f7 --- /dev/null +++ b/frontend/packages/swarm-ui/src/agent/terminal/bash-tasks.ts @@ -0,0 +1,75 @@ +// What an agent's bash task list means, without a DOM: the wire row, the words +// for a state, which tasks count as running, and the count's wording. The list +// is swarm-controller's `GET /api/agents/{name}/bash-tasks`, from the status +// each task's bash MCP published. + +export type TaskState = + | "pending" + | "running" + | "stale" + | "done" + | "timed_out" + | "interrupted" + | "killed"; + +/** One row of `GET /api/agents/{name}/bash-tasks`. */ +export interface BashTask { + task: string; + /** As of the request: a live task whose daemon stopped reporting is `stale`. */ + state: TaskState; + /** The command, cut to 200 characters. */ + cmd: string; + /** Unix seconds when the task was accepted. */ + created_unix: number; + /** Unix seconds when it entered `state`. */ + since_unix: number; + /** The exit status once the process exited by itself. */ + exit: number | null; +} + +/** Whether the task is still going: queued or running, and reporting. */ +export function isLive(task: Pick): boolean { + return task.state === "pending" || task.state === "running"; +} + +/** + * The word before a task's age, saying what the age counts from. `done` shows + * the exit status when there is one and it is not 0. + */ +export function stateWord(task: BashTask): string { + switch (task.state) { + case "pending": + return "queued"; + case "running": + return "started"; + case "stale": + return "stopped reporting"; + case "done": + return task.exit === null || task.exit === 0 + ? "finished" + : `exited ${task.exit}`; + case "timed_out": + return "timed out"; + case "interrupted": + return "interrupted"; + case "killed": + return "killed"; + } +} + +/** Live tasks first, then the rest, each newest first; ties by id. */ +export function sortTasks(tasks: BashTask[]): BashTask[] { + return [...tasks].sort((a, b) => { + const live = Number(isLive(b)) - Number(isLive(a)); + if (live !== 0) return live; + if (a.created_unix !== b.created_unix) + return b.created_unix - a.created_unix; + return a.task.localeCompare(b.task); + }); +} + +/** The count of live tasks in words: `1 running task`, `3 running tasks`. */ +export function runningTasksLabel(tasks: BashTask[]): string { + const n = tasks.filter(isLive).length; + return n === 1 ? "1 running task" : `${n} running tasks`; +} diff --git a/frontend/packages/swarm-ui/src/agent/terminal/useBashTasks.ts b/frontend/packages/swarm-ui/src/agent/terminal/useBashTasks.ts new file mode 100644 index 000000000..7f3a82186 --- /dev/null +++ b/frontend/packages/swarm-ui/src/agent/terminal/useBashTasks.ts @@ -0,0 +1,23 @@ +// An agent's bash tasks, refetched on a timer: `GET /api/agents/{name}/bash-tasks`. +// `null` until the first answer; a failed fetch sets `error` and keeps the last +// list. Mount per agent name: the poll belongs to one agent. +import type { ProblemDetails } from "@hive/shared/api-error.js"; +import { fetchJson, usePolled } from "../../ui/polled/usePolled.js"; +import { sortTasks, type BashTask } from "./bash-tasks.js"; + +const REFRESH_MS = 10_000; + +export function useBashTasks(agentName: string): { + tasks: BashTask[] | null; + error: ProblemDetails | null; +} { + const { data, error } = usePolled(REFRESH_MS, async (signal) => + sortTasks( + await fetchJson( + `/api/agents/${encodeURIComponent(agentName)}/bash-tasks`, + signal, + ), + ), + ); + return { tasks: data, error }; +} diff --git a/frontend/packages/swarm-ui/test/bash-tasks.test.mjs b/frontend/packages/swarm-ui/test/bash-tasks.test.mjs new file mode 100644 index 000000000..ea3c7ab3e --- /dev/null +++ b/frontend/packages/swarm-ui/test/bash-tasks.test.mjs @@ -0,0 +1,69 @@ +// What the bash task list means: which tasks are live, the words for a state, +// the order, and the count. Run by `npm test`, which imports the `.ts` module +// as it is. +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { + isLive, + runningTasksLabel, + sortTasks, + stateWord, +} from "../src/agent/terminal/bash-tasks.ts"; + +const task = (name, state, created, exit = null) => ({ + task: name, + state, + cmd: "make", + created_unix: created, + since_unix: created + 1, + exit, +}); + +test("only a queued or running task, still reporting, is live", () => { + assert.equal(isLive(task("a", "pending", 1)), true); + assert.equal(isLive(task("a", "running", 1)), true); + for (const s of ["stale", "done", "timed_out", "interrupted", "killed"]) { + assert.equal(isLive(task("a", s, 1)), false, s); + } +}); + +test("the state word says what the age counts from", () => { + assert.equal(stateWord(task("a", "pending", 1)), "queued"); + assert.equal(stateWord(task("a", "running", 1)), "started"); + assert.equal(stateWord(task("a", "stale", 1)), "stopped reporting"); + assert.equal(stateWord(task("a", "done", 1, 0)), "finished"); + assert.equal(stateWord(task("a", "done", 1)), "finished"); + assert.equal(stateWord(task("a", "done", 1, 2)), "exited 2"); + assert.equal(stateWord(task("a", "timed_out", 1)), "timed out"); + assert.equal(stateWord(task("a", "interrupted", 1)), "interrupted"); + assert.equal(stateWord(task("a", "killed", 1)), "killed"); +}); + +test("live tasks come first, each group newest first, ties by id", () => { + const list = [ + task("old-done", "done", 10), + task("new-done", "done", 50), + task("old-run", "running", 20), + task("new-run", "running", 60), + task("b", "running", 40), + task("a", "running", 40), + ]; + assert.deepEqual( + sortTasks(list).map((t) => t.task), + ["new-run", "a", "b", "old-run", "new-done", "old-done"], + ); +}); + +test("the count names the live tasks only", () => { + assert.equal(runningTasksLabel([]), "0 running tasks"); + assert.equal(runningTasksLabel([task("a", "running", 1)]), "1 running task"); + assert.equal( + runningTasksLabel([ + task("a", "running", 1), + task("b", "pending", 1), + task("c", "done", 1), + task("d", "stale", 1), + ]), + "2 running tasks", + ); +}); diff --git a/frontend/packages/swarm-ui/test/detail-sections.test.mjs b/frontend/packages/swarm-ui/test/detail-sections.test.mjs index 3ef7fcba8..63b5d88ba 100644 --- a/frontend/packages/swarm-ui/test/detail-sections.test.mjs +++ b/frontend/packages/swarm-ui/test/detail-sections.test.mjs @@ -8,16 +8,17 @@ import { startsOpen, } from "../src/agent/detail/sections.ts"; -test("a card holds status, terminal, subagents, then settings", () => { +test("a card holds status, terminal, subagents, background tasks, then settings", () => { assert.deepEqual(sectionsFor(true), [ "status", "terminal", "subagents", + "tasks", "settings", ]); }); -test("a page beside a terminal leaves the terminal and subagents out", () => { +test("a page beside a terminal leaves the terminal, subagents and tasks out", () => { assert.deepEqual(sectionsFor(false), ["status", "settings"]); }); diff --git a/swarm-controller/src/bash_task.rs b/swarm-controller/src/bash_task.rs new file mode 100644 index 000000000..db8a8e587 --- /dev/null +++ b/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, `.`, 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-`, 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, +} + +/// 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) -> Vec { + 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), + (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, + Path(agent): Path, +) -> Result>, 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 = 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 = 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 { + 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::(&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::()) + .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 = (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)); + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 42bdf3204..341cbdf74 100644 --- a/swarm-controller/src/main.rs +++ b/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; diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index 54c0a5d0a..cbccdb859 100644 --- a/swarm-controller/src/startup.rs +++ b/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))