From 053340128bb275c10729ada1b4f47f94d9fbba62 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 13 Sep 2026 14:01:14 +0200 Subject: [PATCH] term_msg: carry the source event's time on every row MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A terminal row published on `$SWARM.term..` goes out bare, with no envelope around it and no server-side stamp, so a subscriber had nothing to place the row in time with beyond its own receipt clock — wrong by the queue's latency and meaningless for anything read later than live. `TermMsg` gains `ts`, ISO 8601 UTC. `classify` takes the event's own unix-seconds stamp and applies it to every row that event expands into, so a row replayed out of sqlite says when it happened rather than when it was read, and a row that sat in a lagging subscriber's buffer does not lie about its time. The oversize degrade keeps it; only the body is ever spent. `TermEnvelope` stops duplicating `ts` and keeps `seq`: the dedup counter is a real transport concern, the event's time is not, now that it rides on the row. Nothing in the frontend read `envelope.ts` — only the type declared it. Refs #4321 --- docs/web-ui/terminal-rendering.md | 12 +- frontend/packages/agent/src/lib/termMsg.ts | 15 ++- hive-agent/src/swarm_term.rs | 39 +++++- hive-agent/src/term_msg.rs | 143 +++++++++++++++++++-- hive-agent/src/web_ui/stream.rs | 35 ++--- 5 files changed, 206 insertions(+), 38 deletions(-) diff --git a/docs/web-ui/terminal-rendering.md b/docs/web-ui/terminal-rendering.md index 980451b5..e9bf8b53 100644 --- a/docs/web-ui/terminal-rendering.md +++ b/docs/web-ui/terminal-rendering.md @@ -1,14 +1,22 @@ # Per-agent terminal: row taxonomy (as built) The per-agent web UI's live pane renders one row per `TermMsg` -(`hive-agent/src/term_msg.rs`): `{icon?, level: debug|info|warn|error, +(`hive-agent/src/term_msg.rs`): `{ts, icon?, level: debug|info|warn|error, summary, body?, body_format?: markdown|diff, coalesce_key?}`. Classification (icon, summary text, whether a tool call gets an expandable body) happens server-side, once — `term_msg.rs` + `stream_enrich.rs` — and both `GET /api/events/history` and `GET /api/events/stream` serve it identically -as `TermEnvelope { ts, seq?, msgs: TermMsg[] }` frames +as `TermEnvelope { seq?, msgs: TermMsg[] }` frames (`hive-agent/src/web_ui/stream.rs`). +`ts` is ISO 8601 UTC (`2026-09-13T12:35:03Z`) and is the **source event's** +time, stamped at classify time from the event's own stamp — so a row +replayed from sqlite says when it happened, not when it was fetched. It +lives on the row rather than on the envelope because the swarm queue +publishes rows bare, with no envelope around them +(`hive-agent/src/swarm_term.rs`); a subscriber there has nowhere else to +read the time from. + The frontend renders a `TermMsg` close to as-is (`frontend/packages/agent/src/components/Row.tsx`): `level` picks the CSS colour (`terminal.css`'s `.live .level-*`), an empty `summary` + diff --git a/frontend/packages/agent/src/lib/termMsg.ts b/frontend/packages/agent/src/lib/termMsg.ts index daf74db2..f1a0aa11 100644 --- a/frontend/packages/agent/src/lib/termMsg.ts +++ b/frontend/packages/agent/src/lib/termMsg.ts @@ -6,6 +6,10 @@ export type Level = "debug" | "info" | "warn" | "error"; export interface TermMsg { + /** When the event this row came from happened — ISO 8601 UTC, e.g. + * "2026-09-13T12:35:03Z". The source event's own time, so a row replayed + * from history reads as when it happened rather than when it was fetched. */ + ts: string; icon?: string; level: Level; summary: string; @@ -16,9 +20,10 @@ export interface TermMsg { /** One SSE frame / history array entry — `hive-agent`'s `TermEnvelope`. * `seq` is the live per-event dedup counter (`BusEvent::seq`); absent on - * history-replayed envelopes, see useLiveStream.ts's backfill dance. */ + * history-replayed envelopes, see useLiveStream.ts's backfill dance. It is + * the only field left: the event's time rides on each row's own `ts`, where + * a consumer reading a bare row (the swarm queue's) can also see it. */ export interface TermEnvelope { - ts: number; seq?: number; msgs: TermMsg[]; } @@ -27,7 +32,11 @@ export interface TermEnvelope { * stable identity for coalescing/keys, and whether it came from history * replay vs. the live tail. Not a separate model: everything content-wise * is still exactly the wire shape. */ -export interface TermRow extends TermMsg { +export interface TermRow extends Omit { + /** Absent on the rows the client invents for itself — the history/live + * separators, the connection note, a locally echoed prompt — which answer + * to no server event and so have no event time to carry. */ + ts?: string; key: string; fromHistory: boolean; } diff --git a/hive-agent/src/swarm_term.rs b/hive-agent/src/swarm_term.rs index 187eabaa..249e2b57 100644 --- a/hive-agent/src/swarm_term.rs +++ b/hive-agent/src/swarm_term.rs @@ -81,8 +81,9 @@ fn hive_from_client_id(client_id: &str) -> Option<&str> { /// /// The body is the only field that carries arbitrary length — a diff, a whole /// tool result — so it is the only one worth spending: replacing it keeps the -/// row's identity, level and summary, which is what makes the row readable at -/// all, and a reader sees that something was there rather than seeing nothing. +/// row's identity, level, summary and time, which is what makes the row +/// readable at all, and a reader sees that something was there rather than +/// seeing nothing. /// /// `None` means even the degraded row does not fit, so the caller reports it /// instead of publishing. That matters more than it looks: an oversize publish @@ -162,7 +163,7 @@ async fn run(mut rx: broadcast::Receiver, subject: String) { // Every sender is gone, so the harness is shutting down. Err(broadcast::error::RecvError::Closed) => return, }; - for msg in classify(&event.event, &mut ctx) { + for msg in classify(&event.event, event.ts, &mut ctx) { publish(&client, &subject, msg).await; } } @@ -203,7 +204,12 @@ async fn publish(client: &async_nats::Client, subject: &str, msg: TermMsg) { #[cfg(test)] mod tests { use super::{DROPPED_BODY, fit, hive_from_client_id}; - use crate::term_msg::{BodyFormat, Level, TermMsg}; + use crate::events::LiveEvent; + use crate::term_msg::{BodyFormat, ClassifyCtx, Level, TermMsg, classify}; + + /// A row's time as `classify` renders it, for the tests below that need + /// one without classifying an event to get it. + const ROW_TS: &str = "2026-09-13T12:35:03Z"; #[test] fn a_client_id_names_the_hive_between_the_prefix_and_the_suffix() { @@ -256,7 +262,8 @@ mod tests { let msg = TermMsg::new(Level::Info, "Edit(src/main.rs)") .icon("🔧") .body("x".repeat(limit * 4), Some(BodyFormat::Diff)) - .coalesce("tool-1"); + .coalesce("tool-1") + .at(ROW_TS); let fitted = fit(msg, limit).expect("dropping the body brings this under the limit"); assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY)); // The tag describes a body that is no longer there; left set, a reader @@ -266,6 +273,9 @@ mod tests { assert_eq!(fitted.summary, "Edit(src/main.rs)"); assert_eq!(fitted.icon.as_deref(), Some("🔧")); assert_eq!(fitted.coalesce_key.as_deref(), Some("tool-1")); + // Including the time: a degraded row that lost it would be a row a + // subscriber cannot place, and nothing downstream could tell. + assert_eq!(fitted.ts, ROW_TS); assert!( serde_json::to_vec(&fitted).expect("serialises").len() <= limit, "the degraded row must be under the limit, not merely smaller" @@ -308,4 +318,23 @@ mod tests { let fitted = fit(msg, limit).expect("dropping the body fits"); assert_eq!(fitted.body.as_deref(), Some(DROPPED_BODY)); } + + /// What a subscriber actually receives: the payload this module hands + /// `publish` is the bare row, so the time has to be *in* it. A `ts` that + /// existed in Rust but never serialized would leave the queue exactly as + /// timeless as it was before. + #[test] + fn the_published_payload_carries_the_events_own_time() { + let ev = LiveEvent::TurnEnd { + ok: true, + note: None, + }; + let mut ctx = ClassifyCtx::default(); + // 2026-09-13T12:35:03Z, a time that is not the time of this run. + let msgs = classify(&ev, 1_789_302_903, &mut ctx); + let payload = serde_json::to_vec(&fit(msgs[0].clone(), 4096).expect("a small row fits")) + .expect("serialises"); + let value: serde_json::Value = serde_json::from_slice(&payload).expect("valid json"); + assert_eq!(value["ts"], serde_json::json!(ROW_TS)); + } } diff --git a/hive-agent/src/term_msg.rs b/hive-agent/src/term_msg.rs index 193d2d75..1965a0fe 100644 --- a/hive-agent/src/term_msg.rs +++ b/hive-agent/src/term_msg.rs @@ -10,13 +10,22 @@ //! `stream-json` line) can expand to several: an `assistant` message with //! both a text block and a `tool_use` block produces two rows. //! -//! Six fields, and classification belongs **here**, not in the client: the +//! 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; @@ -52,6 +61,13 @@ pub enum BodyFormat { /// 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, @@ -67,6 +83,7 @@ pub struct TermMsg { impl TermMsg { pub fn new(level: Level, summary: impl Into) -> Self { Self { + ts: String::new(), icon: None, level, summary: summary.into(), @@ -94,6 +111,31 @@ impl TermMsg { 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 @@ -121,8 +163,22 @@ impl ClassifyCtx { } } -/// Classify one [`LiveEvent`] into zero or more terminal rows. -pub fn classify(ev: &LiveEvent, ctx: &mut ClassifyCtx) -> Vec { +/// 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 @@ -172,9 +228,16 @@ fn classify_note(text: &str) -> TermMsg { #[cfg(test)] mod tests { - use super::{ClassifyCtx, Level, classify}; + 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 { @@ -183,7 +246,7 @@ mod tests { unread: 3, }; let mut ctx = ClassifyCtx::default(); - let msgs = classify(&ev, &mut ctx); + 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")); @@ -198,7 +261,7 @@ mod tests { unread: 0, }; let mut ctx = ClassifyCtx::default(); - let msgs = classify(&ev, &mut ctx); + let msgs = classify(&ev, EVENT_TS, &mut ctx); assert!(msgs[0].body.is_none()); } @@ -209,7 +272,7 @@ mod tests { note: None, }; let mut ctx = ClassifyCtx::default(); - let msgs = classify(&ev, &mut ctx); + let msgs = classify(&ev, EVENT_TS, &mut ctx); assert_eq!(msgs[0].level, Level::Info); assert_eq!(msgs[0].summary, "turn ok"); } @@ -221,7 +284,7 @@ mod tests { note: Some("rate limited".into()), }; let mut ctx = ClassifyCtx::default(); - let msgs = classify(&ev, &mut ctx); + let msgs = classify(&ev, EVENT_TS, &mut ctx); assert_eq!(msgs[0].level, Level::Error); assert_eq!(msgs[0].summary, "turn fail — rate limited"); } @@ -232,7 +295,7 @@ mod tests { text: "stderr: warning: deprecated flag".into(), }; let mut ctx = ClassifyCtx::default(); - let msgs = classify(&ev, &mut ctx); + let msgs = classify(&ev, EVENT_TS, &mut ctx); assert_eq!(msgs[0].level, Level::Warn); } @@ -242,7 +305,7 @@ mod tests { text: "operator: /compact requested".into(), }; let mut ctx = ClassifyCtx::default(); - let msgs = classify(&ev, &mut ctx); + let msgs = classify(&ev, EVENT_TS, &mut ctx); assert_eq!(msgs[0].level, Level::Info); } @@ -252,7 +315,7 @@ mod tests { text: "created fresh session".into(), }; let mut ctx = ClassifyCtx::default(); - let msgs = classify(&ev, &mut ctx); + let msgs = classify(&ev, EVENT_TS, &mut ctx); assert_eq!(msgs[0].level, Level::Debug); } @@ -264,6 +327,7 @@ mod tests { &LiveEvent::StatusChanged { status: "online".into() }, + EVENT_TS, &mut ctx ) .is_empty() @@ -273,9 +337,66 @@ mod tests { &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"); + } } diff --git a/hive-agent/src/web_ui/stream.rs b/hive-agent/src/web_ui/stream.rs index 99402942..074adbd9 100644 --- a/hive-agent/src/web_ui/stream.rs +++ b/hive-agent/src/web_ui/stream.rs @@ -9,7 +9,7 @@ use serde::{Deserialize, Serialize}; use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream}; use super::AppState; -use crate::term_msg::{ClassifyCtx, Level, TermMsg, classify}; +use crate::term_msg::{ClassifyCtx, Level, TermMsg, classify, iso8601_utc}; /// One classified envelope on the wire: transport-level metadata (a sibling /// of the terminal-row payload, not part of it) plus the zero-or-more rows @@ -20,14 +20,15 @@ use crate::term_msg::{ClassifyCtx, Level, TermMsg, classify}; /// /// `seq` is the live per-event dedup counter (`BusEvent::seq`) — `Some` on /// the SSE path, `None` on history replay (a stored row has no live seq). -/// Same category of plumbing as `ts`: the client already used it to drop -/// buffered live traffic it's about to see again in the initial history -/// page, and that need didn't go away just because rows lost their `kind` -/// tag — dropping it here would silently reintroduce duplicate rows across -/// the live/history boundary. +/// The client uses it to drop buffered live traffic it's about to see again +/// in the initial history page, and that need didn't go away just because +/// rows lost their `kind` tag — dropping it here would silently reintroduce +/// duplicate rows across the live/history boundary. It is the one piece of +/// transport plumbing left: the event's time now rides on each row's own +/// `ts` (`crate::term_msg`), which is where a consumer with no envelope +/// around it — the swarm queue's — can also read it. #[derive(Serialize)] pub(super) struct TermEnvelope { - ts: i64, #[serde(skip_serializing_if = "Option::is_none")] seq: Option, msgs: Vec, @@ -87,15 +88,11 @@ pub(super) async fn events_history( let events: Vec = events .into_iter() .filter_map(|se| { - let msgs = classify(&se.event, &mut ctx); + let msgs = classify(&se.event, se.ts, &mut ctx); if msgs.is_empty() { None } else { - Some(TermEnvelope { - ts: se.ts, - seq: None, - msgs, - }) + Some(TermEnvelope { seq: None, msgs }) } }) .collect(); @@ -117,10 +114,15 @@ pub(super) async fn events_stream( // stream rather than emitted to the bus — a bus emit would spam every // already-connected client with a spurious note each time anyone opens // the stream. + // Synthesised here rather than classified from a bus event, so this is + // the one row that stamps itself: its "source event" is the subscribe + // that just happened. let hello_envelope = TermEnvelope { - ts: chrono::Utc::now().timestamp(), seq: None, - msgs: vec![TermMsg::new(Level::Debug, "live stream attached")], + msgs: vec![ + TermMsg::new(Level::Debug, "live stream attached") + .at(iso8601_utc(chrono::Utc::now().timestamp())), + ], }; let hello = Event::default().data(serde_json::to_string(&hello_envelope).unwrap_or_default()); // One `ClassifyCtx` per connection, moved into the closure — tool_use→ @@ -130,12 +132,11 @@ pub(super) async fn events_stream( let mut ctx = ClassifyCtx::default(); let live = BroadcastStream::new(rx).filter_map(move |res| { let ev = res.ok()?; - let msgs = classify(&ev.event, &mut ctx); + let msgs = classify(&ev.event, ev.ts, &mut ctx); if msgs.is_empty() { return None; } let envelope = TermEnvelope { - ts: ev.ts, seq: Some(ev.seq), msgs, };