//! Terminal-message wire shape: what the per-agent web UI's live/history //! endpoints actually 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. //! //! Seven fields, and classification belongs **here**, not in the client: the //! web UI renders what it is handed and owns no per-tool dispatch table, so //! a new tool needs no frontend change. `level` carries styling, so a row //! never names a CSS class. `kind`, `unread`, `from` and `expanded_default` //! are deliberately absent — adding one back is a design change, not an //! oversight. //! //! The seventh field is `ts`, and it is the row's own: [`classify`] stamps //! every row it returns with the time of the event it classified, not the //! time it ran. The swarm queue publishes the bare row with no envelope //! around it, so a subscriber that reads a row has nowhere else to learn //! when the thing happened; carrying the source event's time means a row //! replayed out of sqlite months later still says when it happened rather //! than when it was read. use chrono::{DateTime, SecondsFormat}; use serde::Serialize; use std::collections::HashMap; use crate::events::LiveEvent; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum Level { /// Low-signal / ambient chatter — thinking, progress ticks, harness /// housekeeping. The client's default rendering can dim/de-emphasize /// these without hiding them outright. Debug, /// Routine substantive content — turn boundaries, assistant text, tool /// calls/results, message bodies. Info, /// Heads-up, not necessarily broken — stderr lines, an unclassified /// event shape landing (the old `.sys` catch-all), API retries. Warn, /// Something actually failed — a turn ending non-ok, a tool result with /// `is_error: true`. Error, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum BodyFormat { Markdown, Diff, } /// One terminal row. `body_format: None` with `body: Some(_)` means plain /// text (the common case — no explicit tag on the wire for it, same logic /// as `body` itself being absent meaning "nothing to expand"). #[derive(Debug, Clone, Serialize)] pub struct TermMsg { /// When the classified event happened, ISO 8601 / RFC 3339 UTC. /// /// Empty on a freshly built row and filled by [`classify`], which is /// the only path a row reaches either wire by — a builder that had to /// be handed the time would repeat the same value across the several /// rows one `stream-json` line expands into. pub ts: String, #[serde(skip_serializing_if = "Option::is_none")] pub icon: Option, pub level: Level, pub summary: String, #[serde(skip_serializing_if = "Option::is_none")] pub body: Option, #[serde(skip_serializing_if = "Option::is_none")] pub body_format: Option, #[serde(skip_serializing_if = "Option::is_none")] pub coalesce_key: Option, } impl TermMsg { pub fn new(level: Level, summary: impl Into) -> Self { Self { ts: String::new(), icon: None, level, summary: summary.into(), body: None, body_format: None, coalesce_key: None, } } #[must_use] pub fn icon(mut self, icon: impl Into) -> Self { self.icon = Some(icon.into()); self } #[must_use] pub fn body(mut self, body: impl Into, format: Option) -> Self { self.body = Some(body.into()); self.body_format = format; self } #[must_use] pub fn coalesce(mut self, key: impl Into) -> Self { self.coalesce_key = Some(key.into()); self } /// Stamp the row with the time of the event it came from. #[must_use] pub fn at(mut self, ts: impl Into) -> Self { self.ts = ts.into(); self } } /// Render an event's unix-seconds stamp as ISO 8601 / RFC 3339 UTC. /// /// Seconds rather than milliseconds because that is the resolution the /// event bus and the sqlite `events.ts` column actually carry /// (`crate::events`) — a `.000` on every row would be precision the source /// does not have. UTC rather than local: the reader of a swarm-published /// row is not on the machine that wrote it, and an offset-carrying stamp /// would make two agents' rows sort by string differently than by time. /// /// A stamp outside the representable range renders as the epoch: a row /// whose only defect is an absurd clock is still worth reading. #[must_use] pub fn iso8601_utc(unix_seconds: i64) -> String { DateTime::from_timestamp(unix_seconds, 0) .unwrap_or_default() .to_rfc3339_opts(SecondsFormat::Secs, true) } /// Per-connection/per-request classification state. A live SSE stream keeps /// one of these alive for the connection's lifetime — `tool_use` id → name /// correlation, so a `tool_result` can tell it's answering a `recv` call and /// render its body as markdown instead of plain text. The history endpoint /// uses a fresh one per page: correlation only works within the page /// actually returned, not across the live/history boundary. Accepted /// degradation — the only user-visible effect is a `recv` result whose /// `tool_use` fell on the other side of a page/reconnect boundary rendering /// its body as plain text instead of markdown. #[derive(Default)] pub struct ClassifyCtx { tool_name_by_id: HashMap, } impl ClassifyCtx { pub fn record_tool_use(&mut self, id: &str, name: &str) { self.tool_name_by_id.insert(id.to_owned(), name.to_owned()); } #[must_use] pub fn tool_name(&self, id: &str) -> Option<&str> { self.tool_name_by_id.get(id).map(String::as_str) } } /// 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 reaching a wire carries one: a row built anywhere else has an /// empty `ts` and cannot get out without passing through this function. pub fn classify(ev: &LiveEvent, ts: i64, ctx: &mut ClassifyCtx) -> Vec { 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 { 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) => crate::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"); } }