da5fcbfd9b
The subagent daemon stamps Swarm-Model and Swarm-Effort on each row it publishes. The list endpoint reads them off the newest row it already fetches, so a long-running subagent keeps them past the stream's 24 hour window, and the terminal page's menu and the agent panel's list show them beside the name.
356 lines
15 KiB
Rust
356 lines
15 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;
|
|
}
|
|
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>,
|
|
}
|
|
|
|
#[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);
|
|
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 (stream_ref, agent_ref) = (&stream, &agent);
|
|
let lookups: Vec<Result<NewestRow, String>> = 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()) {
|
|
tracing::warn!(
|
|
%agent,
|
|
failed,
|
|
subagents = names.len(),
|
|
%error,
|
|
"listing subagents: reading the newest row failed (first error shown)"
|
|
);
|
|
}
|
|
let entries = names
|
|
.into_iter()
|
|
.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),
|
|
}
|
|
})
|
|
.collect();
|
|
Ok(Json(entries))
|
|
}
|
|
|
|
/// 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);
|
|
}
|
|
}
|