subagents publish their model and effort as headers on every row, and the swarm shows them
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.
This commit is contained in:
parent
ae9c3444bc
commit
da5fcbfd9b
10 changed files with 219 additions and 25 deletions
|
|
@ -63,6 +63,9 @@ does for the agent. The list refreshes every 10 seconds, so a subagent the
|
||||||
agent spawns shows up once it writes output. A subagent that drops out of the
|
agent spawns shows up once it writes output. A subagent that drops out of the
|
||||||
list closes its terminal, and a later subagent with the same name starts
|
list closes its terminal, and a later subagent with the same name starts
|
||||||
unpicked.
|
unpicked.
|
||||||
|
Each subagent in the list shows the model and effort it runs, like
|
||||||
|
`opus · high`, beside its name; one on the default model shows only the effort.
|
||||||
|
The `/agents/<name>/terminal` page's subagent list shows them too.
|
||||||
|
|
||||||
The preview carries only the connection badge and **expand**, so it stays clear
|
The preview carries only the connection badge and **expand**, so it stays clear
|
||||||
of the text. What the agent is doing shows in the details table above it, in
|
of the text. What the agent is doing shows in the details table above it, in
|
||||||
|
|
@ -433,7 +436,12 @@ into a stream named `term-sub-<agent>` that swarm-controller creates for every
|
||||||
declared agent. The stream keeps rows for 24 hours. swarm-controller serves the list as
|
declared agent. The stream keeps rows for 24 hours. swarm-controller serves the list as
|
||||||
`GET /api/agents/<name>/subagents`, the subagents named by the subjects in that
|
`GET /api/agents/<name>/subagents`, the subagents named by the subjects in that
|
||||||
stream, each with `last_row_unix`, the stream's publish time of its newest row,
|
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. The lookups run
|
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
|
||||||
concurrently, sixteen at a time, so the list waits at most two seconds for
|
concurrently, sixteen at a time, so the list waits at most two seconds for
|
||||||
each sixteen subagents.
|
each sixteen subagents.
|
||||||
swarm-controller also relays one subagent's rows as SSE on
|
swarm-controller also relays one subagent's rows as SSE on
|
||||||
|
|
|
||||||
|
|
@ -12,7 +12,7 @@ import { useEffect, useState } from "preact/hooks";
|
||||||
import { ApiErrorPanel } from "@hive/shared/api-error-panel.js";
|
import { ApiErrorPanel } from "@hive/shared/api-error-panel.js";
|
||||||
import { Badge } from "@hive/shared/badge.js";
|
import { Badge } from "@hive/shared/badge.js";
|
||||||
import { AgentTermPreview } from "./AgentTermPreview.js";
|
import { AgentTermPreview } from "./AgentTermPreview.js";
|
||||||
import { STALE_S, stillListed } from "./subagent-list.js";
|
import { STALE_S, runLabel, stillListed } from "./subagent-list.js";
|
||||||
import { useSubagentList } from "./useSubagentList.js";
|
import { useSubagentList } from "./useSubagentList.js";
|
||||||
import { RelativeTime } from "../../ui/relative-time/RelativeTime.js";
|
import { RelativeTime } from "../../ui/relative-time/RelativeTime.js";
|
||||||
import "./SubagentTerms.css";
|
import "./SubagentTerms.css";
|
||||||
|
|
@ -41,7 +41,9 @@ export function SubagentTerms({ agentName }: { agentName: string }) {
|
||||||
<section class="ui-subagent-terms" aria-label={`${agentName}'s subagents`}>
|
<section class="ui-subagent-terms" aria-label={`${agentName}'s subagents`}>
|
||||||
<div class="ui-subagent-terms-list">
|
<div class="ui-subagent-terms-list">
|
||||||
<span class="ui-subagent-terms-heading">subagents</span>
|
<span class="ui-subagent-terms-heading">subagents</span>
|
||||||
{subagents?.map(({ name, last_row_unix }) => {
|
{subagents?.map((sub) => {
|
||||||
|
const { name, last_row_unix } = sub;
|
||||||
|
const runs = runLabel(sub);
|
||||||
const stale =
|
const stale =
|
||||||
last_row_unix !== null && nowS - last_row_unix > STALE_S;
|
last_row_unix !== null && nowS - last_row_unix > STALE_S;
|
||||||
return (
|
return (
|
||||||
|
|
@ -60,6 +62,7 @@ export function SubagentTerms({ agentName }: { agentName: string }) {
|
||||||
}
|
}
|
||||||
onClick={() => setPicked(name === selected ? null : name)}
|
onClick={() => setPicked(name === selected ? null : name)}
|
||||||
/>
|
/>
|
||||||
|
{runs && <span class="ui-subagent-terms-age">{runs}</span>}
|
||||||
{last_row_unix !== null && (
|
{last_row_unix !== null && (
|
||||||
<span class="ui-subagent-terms-age">
|
<span class="ui-subagent-terms-age">
|
||||||
<RelativeTime epochMs={last_row_unix * 1000} />
|
<RelativeTime epochMs={last_row_unix * 1000} />
|
||||||
|
|
|
||||||
|
|
@ -45,3 +45,9 @@
|
||||||
.ui-subagents-menu-back {
|
.ui-subagents-menu-back {
|
||||||
color: var(--blue);
|
color: var(--blue);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* What it runs sits after the name, not centred between name and age. */
|
||||||
|
.ui-subagents-menu-run {
|
||||||
|
margin-right: auto;
|
||||||
|
margin-left: 0.6em;
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -17,7 +17,12 @@ import { Panel } from "../../ui/panel/Panel.js";
|
||||||
import { Popover } from "../../ui/popover/Popover.js";
|
import { Popover } from "../../ui/popover/Popover.js";
|
||||||
import { RelativeTime } from "../../ui/relative-time/RelativeTime.js";
|
import { RelativeTime } from "../../ui/relative-time/RelativeTime.js";
|
||||||
import { termPaths } from "./AgentTermPreview.js";
|
import { termPaths } from "./AgentTermPreview.js";
|
||||||
import { runningLabel, splitRunning, type Subagent } from "./subagent-list.js";
|
import {
|
||||||
|
runLabel,
|
||||||
|
runningLabel,
|
||||||
|
splitRunning,
|
||||||
|
type Subagent,
|
||||||
|
} from "./subagent-list.js";
|
||||||
import { useSubagentList } from "./useSubagentList.js";
|
import { useSubagentList } from "./useSubagentList.js";
|
||||||
import "./SubagentsMenu.css";
|
import "./SubagentsMenu.css";
|
||||||
|
|
||||||
|
|
@ -41,6 +46,11 @@ function Entry({
|
||||||
aria-current={current ? "page" : undefined}
|
aria-current={current ? "page" : undefined}
|
||||||
>
|
>
|
||||||
<span class="ui-subagents-menu-name">{sub.name}</span>
|
<span class="ui-subagents-menu-name">{sub.name}</span>
|
||||||
|
{runLabel(sub) && (
|
||||||
|
<span class="ui-subagents-menu-age ui-subagents-menu-run">
|
||||||
|
{runLabel(sub)}
|
||||||
|
</span>
|
||||||
|
)}
|
||||||
{sub.last_row_unix !== null && (
|
{sub.last_row_unix !== null && (
|
||||||
<span class="ui-subagents-menu-age">
|
<span class="ui-subagents-menu-age">
|
||||||
<RelativeTime epochMs={sub.last_row_unix * 1000} />
|
<RelativeTime epochMs={sub.last_row_unix * 1000} />
|
||||||
|
|
|
||||||
|
|
@ -8,6 +8,19 @@ export interface Subagent {
|
||||||
name: string;
|
name: string;
|
||||||
/** Unix seconds of its newest terminal row, `null` when not readable. */
|
/** Unix seconds of its newest terminal row, `null` when not readable. */
|
||||||
last_row_unix: number | null;
|
last_row_unix: number | null;
|
||||||
|
/** The model it runs; `null` on the harness's default or when not readable. */
|
||||||
|
model: string | null;
|
||||||
|
/** Its reasoning effort; `null` when not readable. */
|
||||||
|
effort: string | null;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* What a subagent runs, as one short label: `opus · high`. Whichever of the
|
||||||
|
* two the controller could not read is left out, and an empty string means
|
||||||
|
* there is nothing to show.
|
||||||
|
*/
|
||||||
|
export function runLabel(sub: Pick<Subagent, "model" | "effort">): 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 quiet for longer than this is no longer counted as running. */
|
||||||
|
|
|
||||||
|
|
@ -7,6 +7,7 @@ import {
|
||||||
STALE_S,
|
STALE_S,
|
||||||
byRecency,
|
byRecency,
|
||||||
isRunning,
|
isRunning,
|
||||||
|
runLabel,
|
||||||
runningLabel,
|
runningLabel,
|
||||||
splitRunning,
|
splitRunning,
|
||||||
stillListed,
|
stillListed,
|
||||||
|
|
@ -16,6 +17,8 @@ const NOW = 1_000_000;
|
||||||
const sub = (name, ago) => ({
|
const sub = (name, ago) => ({
|
||||||
name,
|
name,
|
||||||
last_row_unix: ago === null ? null : NOW - ago,
|
last_row_unix: ago === null ? null : NOW - ago,
|
||||||
|
model: null,
|
||||||
|
effort: null,
|
||||||
});
|
});
|
||||||
|
|
||||||
test("a subagent with a recent row is running, a quiet one is not", () => {
|
test("a subagent with a recent row is running, a quiet one is not", () => {
|
||||||
|
|
@ -69,3 +72,11 @@ test("a pick holds only while the refreshed list still has it", () => {
|
||||||
assert.equal(stillListed("a", null), null);
|
assert.equal(stillListed("a", null), null);
|
||||||
assert.equal(stillListed(null, subs), null);
|
assert.equal(stillListed(null, subs), null);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
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");
|
||||||
|
assert.equal(runLabel({ model: null, effort: "medium" }), "medium");
|
||||||
|
assert.equal(runLabel({ model: null, effort: null }), "");
|
||||||
|
assert.equal(runLabel({ model: "", effort: "" }), "");
|
||||||
|
});
|
||||||
|
|
|
||||||
|
|
@ -1814,6 +1814,9 @@ fn continue_on_runtime(
|
||||||
struct LivenessSink {
|
struct LivenessSink {
|
||||||
state: Arc<State>,
|
state: Arc<State>,
|
||||||
name: String,
|
name: String,
|
||||||
|
/// The model and effort this run was started with, stamped on every row
|
||||||
|
/// it publishes.
|
||||||
|
info: Arc<crate::swarm_term::RunInfo>,
|
||||||
/// Where to report the first non-terminal event, on a `resume` whose
|
/// Where to report the first non-terminal event, on a `resume` whose
|
||||||
/// caller is still waiting to hear whether it landed. `None` on a
|
/// caller is still waiting to hear whether it landed. `None` on a
|
||||||
/// `start`, which has nobody waiting.
|
/// `start`, which has nobody waiting.
|
||||||
|
|
@ -1845,11 +1848,20 @@ impl LivenessSink {
|
||||||
/// one to take it.
|
/// one to take it.
|
||||||
fn publish(&self, line: impl FnOnce() -> Output) {
|
fn publish(&self, line: impl FnOnce() -> Output) {
|
||||||
if let Some(term) = &self.state.term {
|
if let Some(term) = &self.state.term {
|
||||||
term.offer(&self.name, line());
|
term.offer(&self.name, &self.info, line());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// What a run's rows say about how it was started: its `config`'s model and
|
||||||
|
/// effort, which the swarm shows beside the subagent's name.
|
||||||
|
fn run_info(config: &Config) -> Arc<crate::swarm_term::RunInfo> {
|
||||||
|
Arc::new(crate::swarm_term::RunInfo {
|
||||||
|
model: config.model.clone(),
|
||||||
|
effort: config.effort.clone(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
/// Spawn the child (synchronous — `Claude::spawn` returns with a real pid
|
/// Spawn the child (synchronous — `Claude::spawn` returns with a real pid
|
||||||
/// the instant the process exists), track it in `running`, and hand the
|
/// the instant the process exists), track it in `running`, and hand the
|
||||||
/// actual turn off to a background task so nobody blocks on the whole turn.
|
/// actual turn off to a background task so nobody blocks on the whole turn.
|
||||||
|
|
@ -1911,6 +1923,7 @@ fn spawn_and_track(
|
||||||
None
|
None
|
||||||
};
|
};
|
||||||
|
|
||||||
|
let info = run_info(config);
|
||||||
let config = config.clone();
|
let config = config.clone();
|
||||||
let state = Arc::clone(state);
|
let state = Arc::clone(state);
|
||||||
let task_name = name.to_owned();
|
let task_name = name.to_owned();
|
||||||
|
|
@ -1922,6 +1935,7 @@ fn spawn_and_track(
|
||||||
let sink = LivenessSink {
|
let sink = LivenessSink {
|
||||||
state: Arc::clone(&state),
|
state: Arc::clone(&state),
|
||||||
name: task_name.clone(),
|
name: task_name.clone(),
|
||||||
|
info: Arc::clone(&info),
|
||||||
verdict: verdict.clone(),
|
verdict: verdict.clone(),
|
||||||
};
|
};
|
||||||
let end = classify_end(running.wait(&prompt, &sink).await, searched.as_deref());
|
let end = classify_end(running.wait(&prompt, &sink).await, searched.as_deref());
|
||||||
|
|
@ -2025,6 +2039,7 @@ fn spawn_and_track_acp(
|
||||||
state.clear_kill(name);
|
state.clear_kill(name);
|
||||||
state.note_event(name);
|
state.note_event(name);
|
||||||
|
|
||||||
|
let info = run_info(&config);
|
||||||
let state = Arc::clone(state);
|
let state = Arc::clone(state);
|
||||||
let task_name = name.to_owned();
|
let task_name = name.to_owned();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
|
|
@ -2033,6 +2048,7 @@ fn spawn_and_track_acp(
|
||||||
let sink = LivenessSink {
|
let sink = LivenessSink {
|
||||||
state: Arc::clone(&state),
|
state: Arc::clone(&state),
|
||||||
name: task_name.clone(),
|
name: task_name.clone(),
|
||||||
|
info: Arc::clone(&info),
|
||||||
verdict: verdict.clone(),
|
verdict: verdict.clone(),
|
||||||
};
|
};
|
||||||
let end = match runtime.run(&config, &prompt, &sink).await {
|
let end = match runtime.run(&config, &prompt, &sink).await {
|
||||||
|
|
@ -3455,6 +3471,7 @@ mod tests {
|
||||||
let sink = LivenessSink {
|
let sink = LivenessSink {
|
||||||
state: Arc::clone(&state),
|
state: Arc::clone(&state),
|
||||||
name: "n".to_owned(),
|
name: "n".to_owned(),
|
||||||
|
info: Arc::default(),
|
||||||
verdict: None,
|
verdict: None,
|
||||||
};
|
};
|
||||||
(state, sink)
|
(state, sink)
|
||||||
|
|
@ -3745,6 +3762,7 @@ mod tests {
|
||||||
let sink = LivenessSink {
|
let sink = LivenessSink {
|
||||||
state: Arc::new(State::new(PathBuf::from("/dev/null"), signal_url())),
|
state: Arc::new(State::new(PathBuf::from("/dev/null"), signal_url())),
|
||||||
name: "n".to_owned(),
|
name: "n".to_owned(),
|
||||||
|
info: Arc::default(),
|
||||||
verdict: Some(Arc::new(Mutex::new(Some(tx)))),
|
verdict: Some(Arc::new(Mutex::new(Some(tx)))),
|
||||||
};
|
};
|
||||||
(rx, sink)
|
(rx, sink)
|
||||||
|
|
|
||||||
|
|
@ -74,8 +74,32 @@ pub enum Output {
|
||||||
Stderr(String),
|
Stderr(String),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The model and effort a subagent was started with, published as headers on
|
||||||
|
/// each of its rows (`swarm_wire::subagent_term`). `None` is the harness's
|
||||||
|
/// default, which has no name to publish.
|
||||||
|
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||||
|
pub struct RunInfo {
|
||||||
|
pub model: Option<String>,
|
||||||
|
pub effort: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl RunInfo {
|
||||||
|
/// The headers for one row: only the values there are.
|
||||||
|
fn headers(&self) -> async_nats::HeaderMap {
|
||||||
|
let mut headers = async_nats::HeaderMap::new();
|
||||||
|
if let Some(model) = self.model.as_deref().filter(|v| !v.is_empty()) {
|
||||||
|
headers.insert(subagent_term::HEADER_MODEL, model);
|
||||||
|
}
|
||||||
|
if let Some(effort) = self.effort.as_deref().filter(|v| !v.is_empty()) {
|
||||||
|
headers.insert(subagent_term::HEADER_EFFORT, effort);
|
||||||
|
}
|
||||||
|
headers
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
struct Line {
|
struct Line {
|
||||||
subagent: String,
|
subagent: String,
|
||||||
|
info: Arc<RunInfo>,
|
||||||
output: Output,
|
output: Output,
|
||||||
/// Unix seconds when the sink saw the line.
|
/// Unix seconds when the sink saw the line.
|
||||||
ts: i64,
|
ts: i64,
|
||||||
|
|
@ -131,14 +155,16 @@ impl Publisher {
|
||||||
(publisher, rx)
|
(publisher, rx)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Queue `output` from `subagent` for publishing. Never blocks: a full
|
/// Queue `output` from `subagent` for publishing, with the `info` its rows
|
||||||
/// queue drops the line and counts it.
|
/// carry as headers. Never blocks: a full queue drops the line and counts
|
||||||
pub fn offer(&self, subagent: &str, output: Output) {
|
/// it.
|
||||||
|
pub fn offer(&self, subagent: &str, info: &Arc<RunInfo>, output: Output) {
|
||||||
let ts = SystemTime::now()
|
let ts = SystemTime::now()
|
||||||
.duration_since(UNIX_EPOCH)
|
.duration_since(UNIX_EPOCH)
|
||||||
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));
|
.map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX));
|
||||||
let line = Line {
|
let line = Line {
|
||||||
subagent: subagent.to_owned(),
|
subagent: subagent.to_owned(),
|
||||||
|
info: Arc::clone(info),
|
||||||
output,
|
output,
|
||||||
ts,
|
ts,
|
||||||
};
|
};
|
||||||
|
|
@ -296,8 +322,9 @@ async fn run(mut rx: mpsc::Receiver<Line>, target: Target, dropped: Arc<AtomicU6
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
let subject = target.subject(&line.subagent);
|
let subject = target.subject(&line.subagent);
|
||||||
|
let headers = line.info.headers();
|
||||||
for row in rows {
|
for row in rows {
|
||||||
publish(&client, &subject, row).await;
|
publish(&client, &subject, &headers, row).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -314,7 +341,12 @@ fn classify(output: &Output, ts: i64, ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Offer one row. Every failure is terminal for that row and for nothing else.
|
/// Offer one row. Every failure is terminal for that row and for nothing else.
|
||||||
async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) {
|
async fn publish(
|
||||||
|
client: &async_nats::Client,
|
||||||
|
subject: &str,
|
||||||
|
headers: &async_nats::HeaderMap,
|
||||||
|
msg: TermMsg,
|
||||||
|
) {
|
||||||
// Unconnected, the client buffers rather than fails, and reports the
|
// Unconnected, the client buffers rather than fails, and reports the
|
||||||
// library's default payload limit rather than the server's.
|
// library's default payload limit rather than the server's.
|
||||||
if let Err(e) = swarm_queue_client::ensure_connected(client) {
|
if let Err(e) = swarm_queue_client::ensure_connected(client) {
|
||||||
|
|
@ -338,7 +370,10 @@ async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if let Err(e) = client.publish(subject.to_owned(), payload.into()).await {
|
let sent = client
|
||||||
|
.publish_with_headers(subject.to_owned(), headers.clone(), payload.into())
|
||||||
|
.await;
|
||||||
|
if let Err(e) = sent {
|
||||||
tracing::warn!(error = %e, "subagent terminal: publish failed, row dropped");
|
tracing::warn!(error = %e, "subagent terminal: publish failed, row dropped");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -412,11 +447,45 @@ mod tests {
|
||||||
#[test]
|
#[test]
|
||||||
fn a_full_queue_drops_and_counts_rather_than_blocks() {
|
fn a_full_queue_drops_and_counts_rather_than_blocks() {
|
||||||
let (publisher, _rx) = Publisher::channel(1);
|
let (publisher, _rx) = Publisher::channel(1);
|
||||||
publisher.offer("scout", Output::Stdout("one".to_owned()));
|
let info = Arc::new(RunInfo::default());
|
||||||
publisher.offer("scout", Output::Stdout("two".to_owned()));
|
publisher.offer("scout", &info, Output::Stdout("one".to_owned()));
|
||||||
|
publisher.offer("scout", &info, Output::Stdout("two".to_owned()));
|
||||||
assert_eq!(publisher.dropped.load(Ordering::Relaxed), 1);
|
assert_eq!(publisher.dropped.load(Ordering::Relaxed), 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The list endpoint reads these two headers off a subagent's newest row.
|
||||||
|
#[test]
|
||||||
|
fn a_row_carries_the_model_and_effort_the_subagent_runs() {
|
||||||
|
let info = RunInfo {
|
||||||
|
model: Some("opus".to_owned()),
|
||||||
|
effort: Some("high".to_owned()),
|
||||||
|
};
|
||||||
|
let headers = info.headers();
|
||||||
|
assert_eq!(
|
||||||
|
headers
|
||||||
|
.get(subagent_term::HEADER_MODEL)
|
||||||
|
.map(ToString::to_string),
|
||||||
|
Some("opus".to_owned())
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
headers
|
||||||
|
.get(subagent_term::HEADER_EFFORT)
|
||||||
|
.map(ToString::to_string),
|
||||||
|
Some("high".to_owned())
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A subagent on the default model names none, and an empty header would
|
||||||
|
/// read as a model called "".
|
||||||
|
#[test]
|
||||||
|
fn a_value_that_is_not_set_is_not_published() {
|
||||||
|
let info = RunInfo {
|
||||||
|
model: None,
|
||||||
|
effort: Some(String::new()),
|
||||||
|
};
|
||||||
|
assert!(info.headers().is_empty());
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn every_kind_of_line_renders_as_stamped_rows() {
|
fn every_kind_of_line_renders_as_stamped_rows() {
|
||||||
let mut ctx = ClassifyCtx::default();
|
let mut ctx = ClassifyCtx::default();
|
||||||
|
|
|
||||||
|
|
@ -11,6 +11,8 @@
|
||||||
//! holds rows for, read with `STREAM.INFO`. A missing stream lists nothing.
|
//! 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
|
//! 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.
|
//! 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
|
//! **No agent-status lookup.** Unlike [`crate::term_stream`], an agent that
|
||||||
//! has never reported gets an empty stream, not a 404.
|
//! has never reported gets an empty stream, not a 404.
|
||||||
|
|
@ -120,6 +122,11 @@ pub(crate) struct SubagentEntry {
|
||||||
/// When its newest row in the stream was published, unix seconds, by the
|
/// 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.
|
/// stream's own clock. `None` when that row could not be read.
|
||||||
pub last_row_unix: Option<i64>,
|
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(
|
#[utoipa::path(
|
||||||
|
|
@ -165,8 +172,8 @@ pub(crate) async fn list_subagents(
|
||||||
// item type trips a "`FnOnce` is not general enough" inference error where
|
// item type trips a "`FnOnce` is not general enough" inference error where
|
||||||
// `routes!` checks this fn as a handler.
|
// `routes!` checks this fn as a handler.
|
||||||
let (stream_ref, agent_ref) = (&stream, &agent);
|
let (stream_ref, agent_ref) = (&stream, &agent);
|
||||||
let lookups: Vec<Result<i64, String>> = futures_util::stream::iter(names.clone())
|
let lookups: Vec<Result<NewestRow, String>> = futures_util::stream::iter(names.clone())
|
||||||
.map(|name| async move { last_row_unix(stream_ref, agent_ref, &name).await })
|
.map(|name| async move { newest_row(stream_ref, agent_ref, &name).await })
|
||||||
.buffered(LOOKUP_CONCURRENCY)
|
.buffered(LOOKUP_CONCURRENCY)
|
||||||
.collect()
|
.collect()
|
||||||
.await;
|
.await;
|
||||||
|
|
@ -183,26 +190,52 @@ pub(crate) async fn list_subagents(
|
||||||
let entries = names
|
let entries = names
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.zip(lookups)
|
.zip(lookups)
|
||||||
.map(|(name, lookup)| SubagentEntry {
|
.map(|(name, lookup)| {
|
||||||
name,
|
let row = lookup.ok();
|
||||||
last_row_unix: 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();
|
.collect();
|
||||||
Ok(Json(entries))
|
Ok(Json(entries))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The publish time of a subagent's newest row, or why it could not be read.
|
/// What a subagent's newest row says: when it was published, and the model and
|
||||||
/// A row that cannot be read, aged out between the listing and this lookup
|
/// effort its headers name.
|
||||||
/// say, is a missing time and not a failed list.
|
#[derive(Debug, PartialEq, Eq)]
|
||||||
async fn last_row_unix(
|
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,
|
stream: &async_nats::jetstream::stream::Stream,
|
||||||
agent: &str,
|
agent: &str,
|
||||||
subagent: &str,
|
subagent: &str,
|
||||||
) -> Result<i64, String> {
|
) -> Result<NewestRow, String> {
|
||||||
let subject = subagent_term::subject(agent, subagent);
|
let subject = subagent_term::subject(agent, subagent);
|
||||||
let lookup = stream.get_last_raw_message_by_subject(&subject);
|
let lookup = stream.get_last_raw_message_by_subject(&subject);
|
||||||
match tokio::time::timeout(LOOKUP_TIMEOUT, lookup).await {
|
match tokio::time::timeout(LOOKUP_TIMEOUT, lookup).await {
|
||||||
Ok(Ok(row)) => Ok(row.time.unix_timestamp()),
|
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))),
|
Ok(Err(e)) => Err(format!("{subagent}: {}", swarm_queue_client::chain(&e))),
|
||||||
Err(_) => Err(format!("{subagent}: no answer in {LOOKUP_TIMEOUT:?}")),
|
Err(_) => Err(format!("{subagent}: no answer in {LOOKUP_TIMEOUT:?}")),
|
||||||
}
|
}
|
||||||
|
|
@ -291,7 +324,8 @@ fn read_failed(agent: &str, e: &dyn std::error::Error) -> problem_details::Probl
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use swarm_wire::subagent_term::subagent_names;
|
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
|
/// 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
|
/// name: a subagent that published once and one that published many rows
|
||||||
|
|
@ -307,4 +341,16 @@ mod tests {
|
||||||
["builder", "scout"]
|
["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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -32,6 +32,16 @@ const SUBAGENT_TOKEN: &str = "sub";
|
||||||
/// its last output.
|
/// its last output.
|
||||||
pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24);
|
pub const MAX_AGE: std::time::Duration = std::time::Duration::from_hours(24);
|
||||||
|
|
||||||
|
/// The message header carrying the model a subagent runs, on every row it
|
||||||
|
/// publishes. Absent when the subagent runs the harness's default model, which
|
||||||
|
/// is no value to name. The list endpoint reads it off the newest row, so a
|
||||||
|
/// subagent that runs for longer than [`MAX_AGE`] still shows it.
|
||||||
|
pub const HEADER_MODEL: &str = "Swarm-Model";
|
||||||
|
|
||||||
|
/// The message header carrying a subagent's reasoning effort, published with
|
||||||
|
/// [`HEADER_MODEL`].
|
||||||
|
pub const HEADER_EFFORT: &str = "Swarm-Effort";
|
||||||
|
|
||||||
/// The stream holding `agent`'s subagent rows: `term-sub-<agent>`.
|
/// The stream holding `agent`'s subagent rows: `term-sub-<agent>`.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn stream_name(agent: &str) -> String {
|
pub fn stream_name(agent: &str) -> String {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue