From e9bc28fe2ffd7498a9f93728362a9b659606944f Mon Sep 17 00:00:00 2001 From: iris Date: Sun, 15 Nov 2026 14:51:37 +0100 Subject: [PATCH] swarm: list subagents from their published run state The subagents list reads each subagent's status from the subagent-status bucket and reports its state, when it began and the exit status. A subagent with no status keeps the newest-row lookup, so an older daemon lists as before. The swarm UI uses the state when there is one and the row-age guess otherwise. --- docs/4-implementation/web-ui/swarm-ui.md | 22 ++- .../src/agent/terminal/SubagentTerms.tsx | 25 ++- .../src/agent/terminal/SubagentsMenu.tsx | 13 +- .../src/agent/terminal/subagent-list.ts | 63 ++++++- .../swarm-ui/test/subagent-list.test.mjs | 54 ++++++ swarm-controller/src/subagent_term.rs | 176 ++++++++++++++++-- 6 files changed, 309 insertions(+), 44 deletions(-) diff --git a/docs/4-implementation/web-ui/swarm-ui.md b/docs/4-implementation/web-ui/swarm-ui.md index 305a47005..b7a240856 100644 --- a/docs/4-implementation/web-ui/swarm-ui.md +++ b/docs/4-implementation/web-ui/swarm-ui.md @@ -565,13 +565,21 @@ The agent's subagent daemon publishes each subagent's terminal rows on into a stream named `term-sub-` that swarm-controller creates for every declared agent. The stream keeps rows for 24 hours. swarm-controller serves the list as `GET /api/agents//subagents`, the subagents named by the subjects in that -stream, each with `last_row_unix`, the stream's publish time of its newest row, -or `null` when that row isn't read within two seconds. Each row also carries -the `Swarm-Model` and `Swarm-Effort` message headers, the model and effort the -subagent runs with, and the list returns them from the newest row as -`model` and `effort`. A subagent on the default model has no `Swarm-Model` -header and lists `model: null`; every row has the headers, so a subagent that -runs past the 24-hour window still shows them. The lookups run +stream. The daemon also publishes whether each subagent is running, one key per +subagent in the `subagent-status` bucket. A subagent with such a status gets its +`state` (`running`, `finished`, or `stale` once the daemon hasn't refreshed a +running status for three minutes), `since_unix`, `exit`, and the `model` and +`effort` from the status. The swarm UI shows the state as it is. + +A subagent with no status, as from an older daemon, gets +`last_row_unix`, the stream's publish time of its newest row, or `null` when +that row isn't read within two seconds, and `state: null`. The swarm UI then +guesses from that time: a row within 15 minutes counts as running. Each row +also carries the `Swarm-Model` and `Swarm-Effort` message headers, the model +and effort the subagent runs with, and the list returns them from the newest +row as `model` and `effort`. A subagent on the default model has no +`Swarm-Model` header and lists `model: null`; every row has the headers, so a +subagent that runs past the 24-hour window still shows them. The lookups run concurrently, sixteen at a time, so the list waits at most two seconds for each sixteen subagents. swarm-controller also relays one subagent's rows as SSE on diff --git a/frontend/packages/swarm-ui/src/agent/terminal/SubagentTerms.tsx b/frontend/packages/swarm-ui/src/agent/terminal/SubagentTerms.tsx index 0867f76e3..07d9bfdb1 100644 --- a/frontend/packages/swarm-ui/src/agent/terminal/SubagentTerms.tsx +++ b/frontend/packages/swarm-ui/src/agent/terminal/SubagentTerms.tsx @@ -1,8 +1,9 @@ // — an agent's subagents, under its own terminal preview in // an opened agent card on `AgentsPage`. The list is `GET /api/agents/{name}/subagents`: // the subagents the swarm queue holds terminal rows for, each with the -// publish time of its newest row. Newest first, with a ticking age beside -// the name; one quiet for `STALE_S` is dimmed. Subagents are spawned on +// state its daemon published, or else the publish time of its newest row. +// Newest first, with a ticking age beside the name; one that is not running +// is dimmed (`isDimmed`, `subagent-list.ts`). Subagents are spawned on // demand, so `useSubagentList` refetches it. Picking one shows its // terminal in `AgentTermPreview`, expand-to-tab included. Read-only like the // agent's own: subagents take no input. @@ -12,7 +13,13 @@ import { useEffect, useState } from "preact/hooks"; import { ApiErrorPanel } from "@hive/shared/api-error-panel.js"; import { Badge } from "@hive/shared/badge.js"; import { AgentTermPreview } from "./AgentTermPreview.js"; -import { STALE_S, runLabel, stillListed } from "./subagent-list.js"; +import { + isDimmed, + lastActive, + runLabel, + stateLabel, + stillListed, +} from "./subagent-list.js"; import { useSubagentList } from "./useSubagentList.js"; import { RelativeTime } from "../../ui/relative-time/RelativeTime.js"; import "./SubagentTerms.css"; @@ -42,10 +49,11 @@ export function SubagentTerms({ agentName }: { agentName: string }) {
subagents {subagents?.map((sub) => { - const { name, last_row_unix } = sub; + const { name } = sub; const runs = runLabel(sub); - const stale = - last_row_unix !== null && nowS - last_row_unix > STALE_S; + const active = lastActive(sub); + const status = stateLabel(sub); + const stale = isDimmed(sub, nowS); return ( setPicked(name === selected ? null : name)} /> {runs && {runs}} - {last_row_unix !== null && ( + {status && {status}} + {active !== null && ( - + )} diff --git a/frontend/packages/swarm-ui/src/agent/terminal/SubagentsMenu.tsx b/frontend/packages/swarm-ui/src/agent/terminal/SubagentsMenu.tsx index a343a4211..d01e82a96 100644 --- a/frontend/packages/swarm-ui/src/agent/terminal/SubagentsMenu.tsx +++ b/frontend/packages/swarm-ui/src/agent/terminal/SubagentsMenu.tsx @@ -3,8 +3,8 @@ // its own terminal page. On a subagent's page the current one is marked and the // list starts with a way back to the agent's own terminal. The list is the // same `GET /api/agents/{name}/subagents` as the one under the preview, and -// "running" means the same too: a row published within `STALE_S` -// (`subagent-list.ts`). Quiet subagents are listed below, dimmed, so a finished +// "running" means the same too: what the subagent's daemon says, or without +// that a row published within `STALE_S` (`subagent-list.ts`). Quiet subagents are listed below, dimmed, so a finished // one can still be opened. // // With `onSplit` the entries are buttons that put the subagent's terminal @@ -22,9 +22,11 @@ import { Popover } from "../../ui/popover/Popover.js"; import { RelativeTime } from "../../ui/relative-time/RelativeTime.js"; import { termPaths } from "./term-paths.js"; import { + lastActive, runLabel, runningLabel, splitRunning, + stateLabel, type Subagent, } from "./subagent-list.js"; import { useSubagentList } from "./useSubagentList.js"; @@ -51,9 +53,12 @@ function Entry({ {runLabel(sub)} )} - {sub.last_row_unix !== null && ( + {stateLabel(sub) && ( + {stateLabel(sub)} + )} + {lastActive(sub) !== null && ( - + )} diff --git a/frontend/packages/swarm-ui/src/agent/terminal/subagent-list.ts b/frontend/packages/swarm-ui/src/agent/terminal/subagent-list.ts index 0df1bf6cf..9f871e960 100644 --- a/frontend/packages/swarm-ui/src/agent/terminal/subagent-list.ts +++ b/frontend/packages/swarm-ui/src/agent/terminal/subagent-list.ts @@ -12,6 +12,21 @@ export interface Subagent { model: string | null; /** Its reasoning effort; `null` when not readable. */ effort: string | null; + /** + * Whether it runs, as its daemon published it: `stale` is a running status + * the daemon stopped refreshing. `null` (or absent) for a subagent whose + * daemon publishes no status, which is judged by its newest row instead. + */ + state?: "running" | "finished" | "stale" | null; + /** Unix seconds when it entered `state`; `null` with `state`. */ + since_unix?: number | null; + /** The process's exit status once it finished by itself. */ + exit?: number | null; +} + +/** When a subagent was last heard from: its newest row, else its status. */ +export function lastActive(sub: Subagent): number | null { + return sub.last_row_unix ?? sub.since_unix ?? null; } /** @@ -23,23 +38,55 @@ export function runLabel(sub: Pick): string { return [sub.model, sub.effort].filter((v) => v).join(" · "); } -/** A subagent quiet for longer than this is no longer counted as running. */ +/** + * A subagent whose daemon publishes no status, quiet for longer than this, is + * no longer counted as running. + */ export const STALE_S = 15 * 60; /** - * Whether a subagent counts as running: it published a row within - * [`STALE_S`]. The controller reports only when the newest row was published, - * so this is a guess. A subagent that is thinking quietly drops out, and one - * that finished stays in until the window passes. + * Whether a subagent counts as running. The daemon's own status decides when + * there is one. Without it this is a guess from the newest row: published + * within [`STALE_S`], so a subagent thinking quietly drops out and one that + * finished stays in until the window passes. */ export function isRunning(sub: Subagent, nowS: number): boolean { + const state = sub.state ?? null; + if (state !== null) return state === "running"; return sub.last_row_unix !== null && nowS - sub.last_row_unix <= STALE_S; } -/** Newest row first; a subagent whose time could not be read sorts last. */ +/** + * Whether the list dims a subagent: its daemon says it is not running, or, + * with no status, its newest row is older than [`STALE_S`]. A subagent whose + * time could not be read is not dimmed, as nothing says it has stopped. + */ +export function isDimmed(sub: Subagent, nowS: number): boolean { + const state = sub.state ?? null; + if (state !== null) return state !== "running"; + return sub.last_row_unix !== null && nowS - sub.last_row_unix > STALE_S; +} + +/** + * A word for a state that is neither running nor unknown, to show beside the + * subagent: `finished`, or `stopped reporting` for a stale status. Empty + * when running or when the daemon published no status. + */ +export function stateLabel(sub: Subagent): string { + switch (sub.state ?? null) { + case "finished": + return "finished"; + case "stale": + return "stopped reporting"; + default: + return ""; + } +} + +/** Newest first; a subagent whose time could not be read sorts last. */ export function byRecency(a: Subagent, b: Subagent): number { - const ta = a.last_row_unix ?? -Infinity; - const tb = b.last_row_unix ?? -Infinity; + const ta = lastActive(a) ?? -Infinity; + const tb = lastActive(b) ?? -Infinity; if (ta !== tb) return tb > ta ? 1 : -1; return a.name.localeCompare(b.name); } diff --git a/frontend/packages/swarm-ui/test/subagent-list.test.mjs b/frontend/packages/swarm-ui/test/subagent-list.test.mjs index da1207cc7..d43ece4dd 100644 --- a/frontend/packages/swarm-ui/test/subagent-list.test.mjs +++ b/frontend/packages/swarm-ui/test/subagent-list.test.mjs @@ -6,10 +6,12 @@ import { test } from "node:test"; import { STALE_S, byRecency, + isDimmed, isRunning, runLabel, runningLabel, splitRunning, + stateLabel, stillListed, } from "../src/agent/terminal/subagent-list.ts"; @@ -73,6 +75,58 @@ test("a pick holds only while the refreshed list still has it", () => { assert.equal(stillListed(null, subs), null); }); +const withState = (name, state, sinceAgo, rowAgo = null) => ({ + ...sub(name, rowAgo), + state, + since_unix: NOW - sinceAgo, + exit: state === "finished" ? 0 : null, +}); + +test("a published state decides, whatever the newest row's age says", () => { + // Quiet for an hour but the daemon says running: thinking, not stopped. + assert.equal(isRunning(withState("a", "running", 3600, 3600), NOW), true); + // A row a second ago but the daemon says finished. + assert.equal(isRunning(withState("a", "finished", 5, 1), NOW), false); + assert.equal(isRunning(withState("a", "stale", 900), NOW), false); +}); + +test("only a subagent without a status falls back to the row's age", () => { + assert.equal(isDimmed(sub("a", STALE_S + 1), NOW), true); + assert.equal(isDimmed(sub("a", 5), NOW), false); + assert.equal(isDimmed(sub("a", null), NOW), false); + assert.equal(isDimmed(withState("a", "running", 3600, 3600), NOW), false); + assert.equal(isDimmed(withState("a", "finished", 5, 1), NOW), true); + assert.equal(isDimmed(withState("a", "stale", 5), NOW), true); +}); + +test("the state word names finished and stale subagents only", () => { + assert.equal(stateLabel(withState("a", "finished", 5)), "finished"); + assert.equal(stateLabel(withState("a", "stale", 5)), "stopped reporting"); + assert.equal(stateLabel(withState("a", "running", 5)), ""); + assert.equal(stateLabel(sub("a", 5)), ""); +}); + +test("a subagent with only a status sorts by when its state began", () => { + const list = [ + withState("old", "finished", 500), + sub("row", 100), + withState("new", "running", 5), + ]; + assert.deepEqual( + [...list].sort(byRecency).map((s) => s.name), + ["new", "row", "old"], + ); + const { running, quiet } = splitRunning(list, NOW); + assert.deepEqual( + running.map((s) => s.name), + ["new", "row"], + ); + assert.deepEqual( + quiet.map((s) => s.name), + ["old"], + ); +}); + test("the run label names the model and effort, whichever are known", () => { assert.equal(runLabel({ model: "opus", effort: "high" }), "opus · high"); assert.equal(runLabel({ model: "opus", effort: null }), "opus"); diff --git a/swarm-controller/src/subagent_term.rs b/swarm-controller/src/subagent_term.rs index d540465ac..54a9a9e7a 100644 --- a/swarm-controller/src/subagent_term.rs +++ b/swarm-controller/src/subagent_term.rs @@ -133,6 +133,40 @@ pub(crate) struct SubagentEntry { pub model: Option, /// Its reasoning effort, from the `Swarm-Effort` header, with the same `None`. pub effort: Option, + /// Whether it is running, from the status its daemon published. `None` + /// for a subagent that has published none (an older daemon), which a + /// reader has to judge from `last_row_unix`. + pub state: Option, + /// When the subagent entered its published state, unix seconds. `None` + /// with `state`. + pub since_unix: Option, + /// The process's exit status once it finished, when it exited by itself. + pub exit: Option, +} + +/// What the status a subagent's daemon published says about it, as of the +/// request. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub(crate) enum SubagentState { + /// A run is in progress and the daemon reported in recently. + Running, + /// The last run ended. + Finished, + /// The status says running, but the daemon has not reported in for three + /// heartbeats: it died or lost the queue. + Stale, +} + +impl SubagentState { + fn of(status: &swarm_wire::subagent_status::Status, now_s: i64) -> Self { + use swarm_wire::subagent_status::State; + match status.state { + State::Finished => Self::Finished, + State::Running if status.is_running(now_s) => Self::Running, + State::Running => Self::Stale, + } + } } #[utoipa::path( @@ -159,7 +193,7 @@ pub(crate) async fn list_subagents( ) -> Result>, problem_details::ProblemDetails> { let agent = ident(&agent).map_err(bad_request)?; let client = connected_client(&state).map_err(|e| unavailable(&e))?; - let js = async_nats::jetstream::new(client); + let js = async_nats::jetstream::new(client.clone()); let stream_name = subagent_term::stream_name(&agent); let stream = match js.get_stream(&stream_name).await { Ok(stream) => stream, @@ -177,14 +211,28 @@ pub(crate) async fn list_subagents( // Owned names and `async move`, not a closure over `&names`: a borrowed // item type trips a "`FnOnce` is not general enough" inference error where // `routes!` checks this fn as a handler. + let statuses = read_statuses(&client, &agent, &names).await; let (stream_ref, agent_ref) = (&stream, &agent); - let lookups: Vec> = futures_util::stream::iter(names.clone()) - .map(|name| async move { newest_row(stream_ref, agent_ref, &name).await }) - .buffered(LOOKUP_CONCURRENCY) - .collect() - .await; - let failed = lookups.iter().filter(|l| l.is_err()).count(); - if let Some(Err(error)) = lookups.iter().find(|l| l.is_err()) { + // A subagent with a published status needs no newest-row lookup: the + // status carries what the row's time and headers were read for. + let lookups: Vec>> = futures_util::stream::iter( + names + .iter() + .cloned() + .zip(statuses.iter().map(Option::is_some)), + ) + .map(|(name, has_status)| async move { + if has_status { + None + } else { + Some(newest_row(stream_ref, agent_ref, &name).await) + } + }) + .buffered(LOOKUP_CONCURRENCY) + .collect() + .await; + let failed = lookups.iter().flatten().filter(|l| l.is_err()).count(); + if let Some(Err(error)) = lookups.iter().flatten().find(|l| l.is_err()) { tracing::warn!( %agent, failed, @@ -193,22 +241,116 @@ pub(crate) async fn list_subagents( "listing subagents: reading the newest row failed (first error shown)" ); } + let now_s = unix_now(); let entries = names .into_iter() + .zip(statuses) .zip(lookups) - .map(|(name, lookup)| { - let row = lookup.ok(); - SubagentEntry { - name, - last_row_unix: row.as_ref().map(|r| r.unix), - model: row.as_ref().and_then(|r| r.model.clone()), - effort: row.and_then(|r| r.effort), - } - }) + .map(|((name, status), lookup)| entry_of(name, status, lookup, now_s)) .collect(); Ok(Json(entries)) } +/// One list entry. A published status gives the state, when it began and the +/// model and effort; without one, the newest row gives the time and the +/// headers give the model and effort. +fn entry_of( + name: String, + status: Option, + lookup: Option>, + now_s: i64, +) -> SubagentEntry { + if let Some(status) = status { + return SubagentEntry { + name, + last_row_unix: None, + state: Some(SubagentState::of(&status, now_s)), + since_unix: Some(status.since), + exit: status.exit, + model: status.model, + effort: status.effort, + }; + } + let row = lookup.and_then(Result::ok); + SubagentEntry { + name, + last_row_unix: row.as_ref().map(|r| r.unix), + model: row.as_ref().and_then(|r| r.model.clone()), + effort: row.and_then(|r| r.effort), + state: None, + since_unix: None, + exit: 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)) +} + +/// The status each of `names` last published, in the same order: `None` for a +/// subagent with none, and for every subagent when the bucket cannot be read +/// (an older swarm, or the queue failing), so the list falls back to the rows. +async fn read_statuses( + client: &async_nats::Client, + agent: &str, + names: &[String], +) -> Vec> { + let js = async_nats::jetstream::new(client.clone()); + let store = match js.get_key_value(swarm_wire::subagent_status::BUCKET).await { + Ok(store) => store, + Err(e) => { + tracing::debug!( + %agent, + error = %swarm_queue_client::chain(&e), + "listing subagents: the status bucket is not readable; using the rows" + ); + return vec![None; names.len()]; + } + }; + // Owned items and `async move`, not a closure over borrows: a borrowed item + // type trips the same "`FnOnce` is not general enough" error as above. + let items: Vec<(async_nats::jetstream::kv::Store, String, String)> = names + .iter() + .map(|name| (store.clone(), agent.to_owned(), name.clone())) + .collect(); + futures_util::stream::iter(items) + .map(|(store, agent, name)| async move { read_status(&store, &agent, &name).await }) + .buffered(LOOKUP_CONCURRENCY) + .collect() + .await +} + +/// One subagent's published status, or `None` when it has none or it could not +/// be read (each failure but a missing key is a log line). +async fn read_status( + store: &async_nats::jetstream::kv::Store, + agent: &str, + name: &str, +) -> Option { + let key = swarm_wire::subagent_status::key(agent, name); + match tokio::time::timeout(LOOKUP_TIMEOUT, store.get(&key)).await { + Ok(Ok(Some(value))) => match serde_json::from_slice(&value) { + Ok(status) => Some(status), + Err(e) => { + tracing::warn!(%agent, subagent = %name, error = %e, "listing subagents: the status is not readable"); + None + } + }, + Ok(Ok(None)) => None, + Ok(Err(e)) => { + tracing::warn!(%agent, subagent = %name, error = %swarm_queue_client::chain(&e), "listing subagents: reading the status failed"); + None + } + Err(_) => { + tracing::warn!(%agent, subagent = %name, "listing subagents: no answer reading the status"); + None + } + } +} + /// What a subagent's newest row says: when it was published, and the model and /// effort its headers name. #[derive(Debug, PartialEq, Eq)]