term_msg: carry the source event's time on every row
A terminal row published on `$SWARM.term.<hive>.<agent>` 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
This commit is contained in:
atlas
mara
parent
349f7096ba
commit
053340128b
5 changed files with 206 additions and 38 deletions
|
|
@ -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<BusEvent>, 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));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String>,
|
||||
pub level: Level,
|
||||
|
|
@ -67,6 +83,7 @@ pub struct TermMsg {
|
|||
impl TermMsg {
|
||||
pub fn new(level: Level, summary: impl Into<String>) -> 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<String>) -> 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<TermMsg> {
|
||||
/// 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<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
|
||||
|
|
@ -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");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<u64>,
|
||||
msgs: Vec<TermMsg>,
|
||||
|
|
@ -87,15 +88,11 @@ pub(super) async fn events_history(
|
|||
let events: Vec<TermEnvelope> = 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,
|
||||
};
|
||||
|
|
|
|||
Loading…
Reference in a new issue