e9bc28fe2f
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.
504 lines
21 KiB
Rust
504 lines
21 KiB
Rust
//! An agent's subagents: the stream that keeps their rows, which ones the
|
|
//! swarm has rows for, and an SSE relay of one subagent's terminal.
|
|
//!
|
|
//! The agent's subagent daemon publishes each subagent's classified rows on
|
|
//! `$SWARM.term.{agent}.sub.{subagent}` under the agent's own queue credential,
|
|
//! which may publish there and nothing else of these streams. [`spawn`]
|
|
//! creates the stream `term-sub-{agent}` for every live agent, with the config
|
|
//! `swarm_queue_client::subagent_term` spells, and in the same pass the
|
|
//! agent's own terminal stream (`swarm_queue_client::agent_term`). Subagents are spawned on demand
|
|
//! under caller-chosen names, so the list is the set of subjects that stream
|
|
//! holds rows for, read with `STREAM.INFO`. A missing stream lists nothing.
|
|
//! Each subagent carries the stream's own publish time of its newest row, so
|
|
//! a reader can tell a subagent that is running from one that has gone quiet.
|
|
//! The same row's `Swarm-Model` and `Swarm-Effort` headers
|
|
//! (`swarm_wire::subagent_term`) give the model and effort it runs with.
|
|
//!
|
|
//! **No agent-status lookup.** Unlike [`crate::term_stream`], an agent that
|
|
//! has never reported gets an empty stream, not a 404.
|
|
//!
|
|
//! **Replay, then the live tail, output only.** The relay is
|
|
//! [`crate::term_stream::replay_or_live`], the agent's own terminal route's,
|
|
//! on the subagent's subject in `term-sub-{agent}`: the subagent's newest
|
|
//! `agent_term::REPLAY_ROWS` rows, then each new one, with the same
|
|
//! `Last-Event-ID` resume and the same live-only fallback while the stream
|
|
//! cannot be read. Nothing is sent toward a subagent: subagents take no
|
|
//! input from the swarm.
|
|
|
|
use std::collections::BTreeSet;
|
|
use std::convert::Infallible;
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use axum::Json;
|
|
use axum::extract::{Path, State};
|
|
use axum::http::{HeaderMap, StatusCode};
|
|
use axum::response::sse::{Event, KeepAlive, Sse};
|
|
use futures_util::{Stream, StreamExt as _, TryStreamExt as _};
|
|
use serde::Serialize;
|
|
use swarm_wire::subagent_term;
|
|
use utoipa::ToSchema;
|
|
|
|
use crate::routes::AppState;
|
|
use crate::wanted::WantedWriter;
|
|
|
|
/// How often [`spawn`] checks every live agent's stream. Also how long a newly
|
|
/// declared agent's subagents can go unlisted.
|
|
const ENSURE_INTERVAL: Duration = Duration::from_mins(1);
|
|
|
|
/// How long [`list_subagents`] waits for one subagent's newest row before
|
|
/// listing it without a time. The lookups run concurrently, so a stream that
|
|
/// does not answer costs the list this long per [`LOOKUP_CONCURRENCY`]
|
|
/// subagents.
|
|
const LOOKUP_TIMEOUT: Duration = Duration::from_secs(2);
|
|
|
|
/// How many newest-row lookups [`list_subagents`] has in flight at once.
|
|
const LOOKUP_CONCURRENCY: usize = 16;
|
|
|
|
/// Create the subagent stream and the terminal stream (`agent_term`) of every
|
|
/// live agent that lacks one, now and every [`ENSURE_INTERVAL`] after. An
|
|
/// existing stream is left as it is.
|
|
///
|
|
/// The live agents are the ones some hive's wanted-state declaration names in
|
|
/// a state other than `Destroyed` (`agent_renewal::live_agents`). A pass skips
|
|
/// a hive whose declaration cannot be read and an agent whose stream cannot be
|
|
/// created, logging each; a pass while the queue is down does nothing. It
|
|
/// never stops the daemon.
|
|
pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, hives: Vec<String>) {
|
|
tokio::spawn(async move {
|
|
let mut ticker = tokio::time::interval(ENSURE_INTERVAL);
|
|
loop {
|
|
ticker.tick().await;
|
|
if swarm_queue_client::ensure_connected(&client).is_err() {
|
|
tracing::debug!("terminal streams: queue not connected; next pass");
|
|
continue;
|
|
}
|
|
if let Err(e) = swarm_queue_client::subagent_status::open_or_create(&client).await {
|
|
tracing::warn!(
|
|
error = %swarm_queue_client::chain(&e),
|
|
"subagent status: opening the bucket failed; next pass"
|
|
);
|
|
}
|
|
for agent in live_agents(&wanted, &hives).await {
|
|
if let Err(e) =
|
|
swarm_queue_client::subagent_term::open_or_create(&client, &agent).await
|
|
{
|
|
tracing::warn!(
|
|
%agent,
|
|
error = %swarm_queue_client::chain(&e),
|
|
"subagent streams: creating the agent's stream failed; next pass"
|
|
);
|
|
}
|
|
if let Err(e) =
|
|
swarm_queue_client::agent_term::open_or_create(&client, &agent).await
|
|
{
|
|
tracing::warn!(
|
|
%agent,
|
|
error = %swarm_queue_client::chain(&e),
|
|
"terminal streams: creating the agent's stream failed; next pass"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Every agent the readable declarations of `hives` keep alive.
|
|
pub(crate) async fn live_agents(wanted: &WantedWriter, hives: &[String]) -> BTreeSet<String> {
|
|
let mut declarations = Vec::with_capacity(hives.len());
|
|
for hive in hives {
|
|
match wanted.view(hive).await {
|
|
Ok(Some(d)) => declarations.push(d),
|
|
Ok(None) => {}
|
|
Err(e) => tracing::warn!(
|
|
%hive,
|
|
error = %format!("{e:#}"),
|
|
"subagent streams: reading the hive's wanted-state declaration failed"
|
|
),
|
|
}
|
|
}
|
|
crate::agent_renewal::live_agents(&declarations)
|
|
}
|
|
|
|
/// One subagent of an agent, as the list endpoint reports it.
|
|
#[derive(Debug, PartialEq, Eq, Serialize, ToSchema)]
|
|
pub(crate) struct SubagentEntry {
|
|
/// The subagent's name, as it appears in its stream subject.
|
|
pub name: String,
|
|
/// When its newest row in the stream was published, unix seconds, by the
|
|
/// stream's own clock. `None` when that row could not be read.
|
|
pub last_row_unix: Option<i64>,
|
|
/// The model it runs, as its newest row's `Swarm-Model` header says. `None`
|
|
/// when the row has none (the harness's default model) or could not be read.
|
|
pub model: Option<String>,
|
|
/// Its reasoning effort, from the `Swarm-Effort` header, with the same `None`.
|
|
pub effort: Option<String>,
|
|
/// 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<SubagentState>,
|
|
/// When the subagent entered its published state, unix seconds. `None`
|
|
/// with `state`.
|
|
pub since_unix: Option<i64>,
|
|
/// The process's exit status once it finished, when it exited by itself.
|
|
pub exit: Option<i32>,
|
|
}
|
|
|
|
/// 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(
|
|
get,
|
|
path = "/api/agents/{name}/subagents",
|
|
params(("name" = String, Path, description = "agent whose subagents to list")),
|
|
responses(
|
|
(status = 200, description = "the agent's subagents with terminal rows in the \
|
|
swarm queue, sorted by name, each with the publish time of its newest row; \
|
|
empty when the agent has published none",
|
|
body = Vec<SubagentEntry>),
|
|
(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 agent's subagent stream failed \
|
|
(problem+json)", body = String),
|
|
),
|
|
tag = "agents"
|
|
)]
|
|
pub(crate) async fn list_subagents(
|
|
State(state): State<AppState>,
|
|
Path(agent): Path<String>,
|
|
) -> Result<Json<Vec<SubagentEntry>>, 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.clone());
|
|
let stream_name = subagent_term::stream_name(&agent);
|
|
let stream = match js.get_stream(&stream_name).await {
|
|
Ok(stream) => stream,
|
|
Err(e) if is_stream_not_found(&e) => return Ok(Json(Vec::new())),
|
|
Err(e) => return Err(read_failed(&agent, &e)),
|
|
};
|
|
let subjects: Vec<(String, usize)> = stream
|
|
.info_with_subjects(subagent_term::stream_subjects(&agent))
|
|
.await
|
|
.map_err(|e| read_failed(&agent, &e))?
|
|
.try_collect()
|
|
.await
|
|
.map_err(|e| read_failed(&agent, &e))?;
|
|
let names = subagent_term::subagent_names(&agent, subjects.iter().map(|(s, _)| s.as_str()));
|
|
// 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);
|
|
// 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<Option<Result<NewestRow, String>>> = 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,
|
|
subagents = names.len(),
|
|
%error,
|
|
"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, 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<swarm_wire::subagent_status::Status>,
|
|
lookup: Option<Result<NewestRow, String>>,
|
|
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<Option<swarm_wire::subagent_status::Status>> {
|
|
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<swarm_wire::subagent_status::Status> {
|
|
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)]
|
|
struct NewestRow {
|
|
unix: i64,
|
|
model: Option<String>,
|
|
effort: Option<String>,
|
|
}
|
|
|
|
/// One header's value, or `None` when the row has none or an empty one.
|
|
fn header(headers: &async_nats::HeaderMap, name: &str) -> Option<String> {
|
|
headers
|
|
.get(name)
|
|
.map(ToString::to_string)
|
|
.filter(|v| !v.is_empty())
|
|
}
|
|
|
|
/// A subagent's newest row, or why it could not be read. A row that cannot be
|
|
/// read, aged out between the listing and this lookup say, is a missing time
|
|
/// and not a failed list.
|
|
async fn newest_row(
|
|
stream: &async_nats::jetstream::stream::Stream,
|
|
agent: &str,
|
|
subagent: &str,
|
|
) -> Result<NewestRow, String> {
|
|
let subject = subagent_term::subject(agent, subagent);
|
|
let lookup = stream.get_last_raw_message_by_subject(&subject);
|
|
match tokio::time::timeout(LOOKUP_TIMEOUT, lookup).await {
|
|
Ok(Ok(row)) => Ok(NewestRow {
|
|
unix: row.time.unix_timestamp(),
|
|
model: header(&row.headers, subagent_term::HEADER_MODEL),
|
|
effort: header(&row.headers, subagent_term::HEADER_EFFORT),
|
|
}),
|
|
Ok(Err(e)) => Err(format!("{subagent}: {}", swarm_queue_client::chain(&e))),
|
|
Err(_) => Err(format!("{subagent}: no answer in {LOOKUP_TIMEOUT:?}")),
|
|
}
|
|
}
|
|
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/agents/{name}/subagents/{subagent}/term/stream",
|
|
params(
|
|
("name" = String, Path, description = "agent the subagent belongs to"),
|
|
("subagent" = String, Path, description = "subagent whose terminal to stream"),
|
|
),
|
|
responses(
|
|
(status = 200, description = "server-sent event stream; each event's `data` is \
|
|
one already-classified TermMsg row, JSON as the agent's subagent daemon \
|
|
published it (opaque to this daemon); the subagent's newest retained rows \
|
|
first, then live rows. Each event's `id` is the row's stream sequence, and a \
|
|
`Last-Event-ID` request header resumes right after that row",
|
|
body = String, content_type = "text/event-stream"),
|
|
(status = 400, description = "the agent or subagent 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 = "the subscribe itself failed (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "agents"
|
|
)]
|
|
pub(crate) async fn stream_subagent_term(
|
|
State(state): State<AppState>,
|
|
Path((agent, subagent)): Path<(String, String)>,
|
|
headers: HeaderMap,
|
|
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> {
|
|
let agent = ident(&agent).map_err(bad_request)?;
|
|
let subagent = ident(&subagent).map_err(bad_request)?;
|
|
let client = connected_client(&state).map_err(|e| unavailable(&e))?;
|
|
let stream = crate::term_stream::replay_or_live(
|
|
&client,
|
|
&subagent_term::stream_name(&agent),
|
|
subagent_term::subject(&agent, &subagent),
|
|
crate::term_stream::last_event_id(&headers),
|
|
"subagent term",
|
|
)
|
|
.await?;
|
|
Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
|
|
}
|
|
|
|
/// The queue client, or why there is none: the 503 every queue-backed route
|
|
/// answers.
|
|
fn connected_client(state: &AppState) -> Result<async_nats::Client, String> {
|
|
let status = state
|
|
.status
|
|
.as_ref()
|
|
.ok_or_else(|| "no swarm queue is configured on this host".to_owned())?;
|
|
let client = status.queue_client();
|
|
swarm_queue_client::ensure_connected(&client).map_err(|e| swarm_queue_client::chain(&e))?;
|
|
Ok(client)
|
|
}
|
|
|
|
/// A path segment as a single subject token, or why it is not one.
|
|
fn ident(name: &str) -> Result<String, &'static str> {
|
|
hive_types::Ident::parse(name).map(hive_types::Ident::into_string)
|
|
}
|
|
|
|
fn bad_request(reason: &str) -> problem_details::ProblemDetails {
|
|
crate::routes::error_problem(StatusCode::BAD_REQUEST, reason)
|
|
}
|
|
|
|
fn unavailable(detail: &str) -> problem_details::ProblemDetails {
|
|
crate::routes::error_problem(StatusCode::SERVICE_UNAVAILABLE, detail)
|
|
}
|
|
|
|
fn is_stream_not_found(e: &async_nats::jetstream::context::GetStreamError) -> bool {
|
|
matches!(
|
|
e.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 subagents: reading the stream failed");
|
|
crate::routes::error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::header;
|
|
use swarm_wire::subagent_term::{HEADER_EFFORT, HEADER_MODEL, subagent_names};
|
|
|
|
/// The list is what `STREAM.INFO` reports for the stream's subjects, by
|
|
/// name: a subagent that published once and one that published many rows
|
|
/// list the same way, and the order is the name's, not the stream's.
|
|
#[test]
|
|
fn the_list_is_the_streams_subjects_by_name() {
|
|
let reported = [
|
|
("$SWARM.term.atlas.sub.scout".to_owned(), 12),
|
|
("$SWARM.term.atlas.sub.builder".to_owned(), 1),
|
|
];
|
|
assert_eq!(
|
|
subagent_names("atlas", reported.iter().map(|(s, _)| s.as_str())),
|
|
["builder", "scout"]
|
|
);
|
|
}
|
|
|
|
/// The model and effort are read off the newest row's headers; a row
|
|
/// without them, or with an empty one, names nothing.
|
|
#[test]
|
|
fn the_model_and_effort_come_from_the_rows_headers() {
|
|
let mut headers = async_nats::HeaderMap::new();
|
|
headers.insert(HEADER_MODEL, "opus");
|
|
headers.insert(HEADER_EFFORT, "");
|
|
assert_eq!(header(&headers, HEADER_MODEL), Some("opus".to_owned()));
|
|
assert_eq!(header(&headers, HEADER_EFFORT), None);
|
|
assert_eq!(header(&async_nats::HeaderMap::new(), HEADER_MODEL), None);
|
|
}
|
|
}
|