An agent's subagent daemon publishes each subagent's output as terminal
rows on `$SWARM.term.<agent>.sub.<subagent>`, as the agent, into a
per-agent stream it creates itself; swarm-controller lists an agent's
subagents from that stream's subjects and relays one subagent's rows as
SSE; the swarm UI lists them under the agent's terminal preview and
reuses AgentTermPreview, full-screen tab included, with no input.
- swarm-nats.nix: the agent token may also publish
`$SWARM.term.{agent}.sub.>` and `$JS.API.STREAM.CREATE|INFO` on
`term-sub-{agent}`, and nothing else of JetStream. A module-eval arm
pins the agent-token grant as an exact list.
- mcp.nix: hive-subagent-daemon loads the agent's store identity
(`hive-agent-bao-cert/-key/-server-ca`, the ones hive-agent loads)
whenever the agent has a store, not only on the opencode preset. The
agent's own queue secret lives in the store, so this is the credential
the harness connects with.
- hive-subagent-mcp: `swarm_term` reads the agent's queue secret under
that identity, connects with the agent token, opens or creates
`term-sub-<agent>` (max_age 24h), and publishes classified rows from
the sink every subagent line already passes through. The sink only
queues (bounded, drop-and-count); a missing store, refused credential,
failed stream create or failed publish is a log line.
- The stream-json classifier (`stream_enrich`) and the `TermMsg` row
types plus `fit` move from the hive-agent binary into hive-sh4re, so
the subagent daemon publishes the rows AgentTermPreview already
renders. hive-agent keeps its LiveEvent classifier on top.
- swarm-controller: `GET /api/agents/{name}/subagents` and
`GET /api/agents/{name}/subagents/{subagent}/term/stream`.
- docs/swarm: what the UI shows and what the queue carries.
Closes #4827
252 lines
9.1 KiB
Rust
252 lines
9.1 KiB
Rust
//! The harness's events as terminal rows: what the per-agent web UI's
|
|
//! live/history endpoints serve for the "terminal" event stream, as opposed to
|
|
//! agent-state changes (`StatusChanged`/`ModelChanged`/`EffortChanged`/
|
|
//! `TokenUsageChanged`/`TurnStateChanged`), which have never rendered as
|
|
//! terminal rows (the header/badges poll `/api/state`, not this stream) and
|
|
//! produce zero [`TermMsg`]s here.
|
|
//!
|
|
//! One [`crate::events::LiveEvent`] maps to zero or more `TermMsg`s — most
|
|
//! map to exactly one, but `LiveEvent::Stream` (one raw claude
|
|
//! `stream-json` line) can expand to several: an `assistant` message with
|
|
//! both a text block and a `tool_use` block produces two rows.
|
|
|
|
pub use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, iso8601_utc};
|
|
|
|
use crate::events::LiveEvent;
|
|
|
|
/// Classify one [`LiveEvent`] into zero or more terminal rows, each
|
|
/// stamped with `ts` — the event's own unix-seconds time, from the bus on
|
|
/// the live path and from the `events` table on replay.
|
|
///
|
|
/// The stamp is applied here, over the rows the classifiers return, so that
|
|
/// every row the harness puts on a wire carries one.
|
|
pub fn classify(ev: &LiveEvent, ts: i64, ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
|
|
let ts = iso8601_utc(ts);
|
|
classify_rows(ev, ctx)
|
|
.into_iter()
|
|
.map(|m| m.at(ts.as_str()))
|
|
.collect()
|
|
}
|
|
|
|
fn classify_rows(ev: &LiveEvent, ctx: &mut ClassifyCtx) -> Vec<TermMsg> {
|
|
match ev {
|
|
LiveEvent::TurnStart { from, body, .. } => {
|
|
// `unread` (the third field) was dropped — mara: "i can see
|
|
// the todo/inbox count in toolbar and dont need the N unread".
|
|
let mut m = TermMsg::new(Level::Info, format!("TURN ← {from}")).icon("◆");
|
|
if !body.trim().is_empty() {
|
|
m = m.body(body.clone(), None);
|
|
}
|
|
vec![m]
|
|
}
|
|
LiveEvent::TurnEnd { ok, note } => {
|
|
let msg = if *ok {
|
|
TermMsg::new(Level::Info, "turn ok").icon("✅")
|
|
} else {
|
|
let summary = note
|
|
.as_deref()
|
|
.filter(|n| !n.is_empty())
|
|
.map_or_else(|| "turn fail".to_owned(), |n| format!("turn fail — {n}"));
|
|
TermMsg::new(Level::Error, summary).icon("❌")
|
|
};
|
|
vec![msg]
|
|
}
|
|
LiveEvent::Note { text } => vec![classify_note(text)],
|
|
LiveEvent::Stream(v) => hive_sh4re::stream_enrich::classify_stream_value(v, ctx),
|
|
// Agent-state transitions never render as terminal rows — the
|
|
// header/badges read `/api/state`, not this stream (see module doc).
|
|
LiveEvent::StatusChanged { .. }
|
|
| LiveEvent::ModelChanged { .. }
|
|
| LiveEvent::EffortChanged { .. }
|
|
| LiveEvent::TokenUsageChanged { .. }
|
|
| LiveEvent::TurnStateChanged { .. } => vec![],
|
|
}
|
|
}
|
|
|
|
fn classify_note(text: &str) -> TermMsg {
|
|
if let Some(rest) = text.strip_prefix("stderr:") {
|
|
TermMsg::new(Level::Warn, format!("stderr:{rest}"))
|
|
} else if let Some(rest) = text.strip_prefix("operator:") {
|
|
TermMsg::new(Level::Info, format!("operator:{rest}"))
|
|
} else {
|
|
// Ambient harness chatter (session archived, plugin loaded, etc.) —
|
|
// routine, not worth the same visual weight as a tool call or
|
|
// assistant text.
|
|
TermMsg::new(Level::Debug, text.to_owned())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{ClassifyCtx, Level, classify, iso8601_utc};
|
|
use crate::events::LiveEvent;
|
|
|
|
/// A fixed event time and the exact string it must render as. Fixed on
|
|
/// both sides on purpose: a stamp taken from the clock would render to
|
|
/// a different string on every run, and a test that can only assert
|
|
/// "some string" passes just as happily against `now()`.
|
|
const EVENT_TS: i64 = 1_789_302_903;
|
|
const EVENT_TS_ISO: &str = "2026-09-13T12:35:03Z";
|
|
|
|
#[test]
|
|
fn turn_start_carries_from_and_body_no_unread() {
|
|
let ev = LiveEvent::TurnStart {
|
|
from: "operator".into(),
|
|
body: "sweep the backlog".into(),
|
|
unread: 3,
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert_eq!(msgs.len(), 1);
|
|
assert_eq!(msgs[0].summary, "TURN ← operator");
|
|
assert_eq!(msgs[0].body.as_deref(), Some("sweep the backlog"));
|
|
assert_eq!(msgs[0].level, Level::Info);
|
|
}
|
|
|
|
#[test]
|
|
fn turn_start_empty_body_has_no_body() {
|
|
let ev = LiveEvent::TurnStart {
|
|
from: "reminder".into(),
|
|
body: String::new(),
|
|
unread: 0,
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert!(msgs[0].body.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn turn_end_ok_is_info() {
|
|
let ev = LiveEvent::TurnEnd {
|
|
ok: true,
|
|
note: None,
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert_eq!(msgs[0].level, Level::Info);
|
|
assert_eq!(msgs[0].summary, "turn ok");
|
|
}
|
|
|
|
#[test]
|
|
fn turn_end_fail_is_error_with_note() {
|
|
let ev = LiveEvent::TurnEnd {
|
|
ok: false,
|
|
note: Some("rate limited".into()),
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert_eq!(msgs[0].level, Level::Error);
|
|
assert_eq!(msgs[0].summary, "turn fail — rate limited");
|
|
}
|
|
|
|
#[test]
|
|
fn note_stderr_is_warn() {
|
|
let ev = LiveEvent::Note {
|
|
text: "stderr: warning: deprecated flag".into(),
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert_eq!(msgs[0].level, Level::Warn);
|
|
}
|
|
|
|
#[test]
|
|
fn note_operator_is_info() {
|
|
let ev = LiveEvent::Note {
|
|
text: "operator: /compact requested".into(),
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert_eq!(msgs[0].level, Level::Info);
|
|
}
|
|
|
|
#[test]
|
|
fn note_plain_is_debug() {
|
|
let ev = LiveEvent::Note {
|
|
text: "created fresh session".into(),
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert_eq!(msgs[0].level, Level::Debug);
|
|
}
|
|
|
|
#[test]
|
|
fn agent_state_events_produce_no_rows() {
|
|
let mut ctx = ClassifyCtx::default();
|
|
assert!(
|
|
classify(
|
|
&LiveEvent::StatusChanged {
|
|
status: "online".into()
|
|
},
|
|
EVENT_TS,
|
|
&mut ctx
|
|
)
|
|
.is_empty()
|
|
);
|
|
assert!(
|
|
classify(
|
|
&LiveEvent::ModelChanged {
|
|
model: "opus".into()
|
|
},
|
|
EVENT_TS,
|
|
&mut ctx
|
|
)
|
|
.is_empty()
|
|
);
|
|
}
|
|
|
|
/// The point of the field: a row replayed out of history says when the
|
|
/// thing happened, not when it was read. Asserting the exact rendering
|
|
/// of a fixed input is what makes this a test — stamping `now()`
|
|
/// instead produces today's date and fails here.
|
|
#[test]
|
|
fn a_row_carries_the_time_of_the_event_it_came_from() {
|
|
let ev = LiveEvent::Note {
|
|
text: "created fresh session".into(),
|
|
};
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert_eq!(msgs[0].ts, EVENT_TS_ISO);
|
|
}
|
|
|
|
/// One `stream-json` line expanding into several rows gives all of them
|
|
/// the same time: they are one event, and rows that disagreed about
|
|
/// when they happened would sort against each other downstream.
|
|
#[test]
|
|
fn every_row_one_event_expands_into_shares_that_time() {
|
|
let ev = LiveEvent::Stream(serde_json::json!({
|
|
"type": "assistant",
|
|
"message": {
|
|
"content": [
|
|
{ "type": "text", "text": "looking at it" },
|
|
{
|
|
"type": "tool_use",
|
|
"id": "toolu_1",
|
|
"name": "Read",
|
|
"input": { "file_path": "/src/main.rs" }
|
|
}
|
|
]
|
|
}
|
|
}));
|
|
let mut ctx = ClassifyCtx::default();
|
|
let msgs = classify(&ev, EVENT_TS, &mut ctx);
|
|
assert!(msgs.len() > 1, "this fixture must expand to several rows");
|
|
for m in &msgs {
|
|
assert_eq!(
|
|
m.ts, EVENT_TS_ISO,
|
|
"row {:?} lost the event's time",
|
|
m.summary
|
|
);
|
|
}
|
|
}
|
|
|
|
/// UTC, with no dependence on the machine's zone: the rendering of a
|
|
/// fixed stamp is the same everywhere, and the `Z` says so on the wire.
|
|
#[test]
|
|
fn the_stamp_renders_as_utc_whatever_the_local_zone_is() {
|
|
assert_eq!(iso8601_utc(EVENT_TS), EVENT_TS_ISO);
|
|
assert_eq!(iso8601_utc(0), "1970-01-01T00:00:00Z");
|
|
// Out of range for a representable date; the epoch stands in rather
|
|
// than the row losing its stamp entirely.
|
|
assert_eq!(iso8601_utc(i64::MAX), "1970-01-01T00:00:00Z");
|
|
}
|
|
}
|