From 50e0ef9b4695a81e52fe33acd798d0cfda5ed484 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 10 Jun 2026 14:01:35 +0200 Subject: [PATCH] feat(#1540): expose per-event ts on live frame + history rows --- hive-ag3nt/src/events.rs | 85 ++++++++++++++++++++++++++++++++++------ 1 file changed, 72 insertions(+), 13 deletions(-) diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 0acc3bd5..a4e15a07 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -169,6 +169,23 @@ CREATE INDEX IF NOT EXISTS idx_events_ts ON events (ts); #[derive(Debug, Clone, Serialize)] pub struct BusEvent { pub seq: u64, + /// Unix seconds at emit time. Serialized as a sibling of the `kind` + /// tag so the agent terminal can render turn start/end times (and + /// turn duration) on the live stream; history rows carry the same + /// `ts` field sourced from the persisted `events.ts` column, so the + /// renderer reads `ts` identically for live + scrollback. + pub ts: i64, + #[serde(flatten)] + pub event: LiveEvent, +} + +/// A persisted event paired with its stored unix-seconds timestamp. +/// Serializes with `ts` as a sibling of the `kind` tag — same wire shape +/// as a live [`BusEvent`] minus `seq` — so the agent terminal reads `ts` +/// identically whether an event arrives live or is replayed from history. +#[derive(Debug, Clone, Serialize)] +pub struct StoredEvent { + pub ts: i64, #[serde(flatten)] pub event: LiveEvent, } @@ -281,7 +298,7 @@ impl EventStore { Ok(()) } - fn recent(&self, limit: usize) -> rusqlite::Result> { + fn recent(&self, limit: usize) -> rusqlite::Result> { let (events, _, _) = self.page(None, limit)?; Ok(events) } @@ -293,43 +310,52 @@ impl EventStore { &self, before_id: Option, limit: usize, - ) -> rusqlite::Result<(Vec, Option, bool)> { + ) -> rusqlite::Result<(Vec, Option, bool)> { let limit_i = i64::try_from(limit).unwrap_or(i64::MAX); let conn = self.conn.lock().unwrap(); // Fetch one extra row so we can tell whether more exist. let fetch = limit_i.saturating_add(1); - let rows: Vec<(i64, LiveEvent)> = if let Some(bid) = before_id { + // `ts` is the persisted emit-time unix-seconds stamp; carried out + // alongside each event so history replay shows the same turn + // start/end times the live stream did. + let rows: Vec<(i64, StoredEvent)> = if let Some(bid) = before_id { let mut stmt = conn.prepare( - "SELECT id, payload_json FROM events + "SELECT id, ts, payload_json FROM events WHERE id < ?1 ORDER BY id DESC LIMIT ?2", )?; stmt.query_map(params![bid, fetch], |row| { let id: i64 = row.get(0)?; - let s: String = row.get(1)?; - Ok(serde_json::from_str::(&s).ok().map(|e| (id, e))) + let ts: i64 = row.get(1)?; + let s: String = row.get(2)?; + Ok(serde_json::from_str::(&s) + .ok() + .map(|event| (id, StoredEvent { ts, event }))) })? .flatten() .flatten() .collect() } else { let mut stmt = conn.prepare( - "SELECT id, payload_json FROM events + "SELECT id, ts, payload_json FROM events ORDER BY id DESC LIMIT ?1", )?; stmt.query_map(params![fetch], |row| { let id: i64 = row.get(0)?; - let s: String = row.get(1)?; - Ok(serde_json::from_str::(&s).ok().map(|e| (id, e))) + let ts: i64 = row.get(1)?; + let s: String = row.get(2)?; + Ok(serde_json::from_str::(&s) + .ok() + .map(|event| (id, StoredEvent { ts, event }))) })? .flatten() .flatten() .collect() }; let has_more = rows.len() > limit; - let mut rows: Vec<(i64, LiveEvent)> = rows.into_iter().take(limit).collect(); + let mut rows: Vec<(i64, StoredEvent)> = rows.into_iter().take(limit).collect(); rows.reverse(); // oldest first let min_id = rows.first().map(|(id, _)| *id); let events = rows.into_iter().map(|(_, e)| e).collect(); @@ -1003,6 +1029,7 @@ impl Bus { } let envelope = BusEvent { seq: self.next_seq(), + ts: now_unix(), event, }; // Lagged subscribers drop events — fine; the UI is a tail, not a log. @@ -1018,7 +1045,7 @@ impl Bus { /// Drives the terminal pre-fill when the operator opens the agent /// page; without a store (db open failed) this is empty. #[must_use] - pub fn history(&self) -> Vec { + pub fn history(&self) -> Vec { let Some(store) = &self.store else { return Vec::new(); }; @@ -1035,7 +1062,7 @@ impl Bus { &self, before_id: Option, limit: usize, - ) -> (Vec, Option, bool) { + ) -> (Vec, Option, bool) { let Some(store) = &self.store else { return (Vec::new(), None, false); }; @@ -1051,9 +1078,41 @@ impl Default for Bus { #[cfg(test)] mod tests { - use super::{DEFAULT_EFFORT, EFFORT_LEVELS, TokenUsage, is_valid_effort}; + use super::{ + BusEvent, DEFAULT_EFFORT, EFFORT_LEVELS, LiveEvent, StoredEvent, TokenUsage, + is_valid_effort, + }; use serde_json::json; + #[test] + fn stored_event_serializes_ts_beside_kind() { + // History-row wire shape: `ts` is a flattened sibling of `kind`, + // which is what the agent terminal reads to time turn boundaries. + let v = serde_json::to_value(StoredEvent { + ts: 1_700_000_000, + event: LiveEvent::Note { text: "hi".into() }, + }) + .unwrap(); + assert_eq!(v["ts"], 1_700_000_000_i64); + assert_eq!(v["kind"], "note"); + assert_eq!(v["text"], "hi"); + } + + #[test] + fn bus_event_serializes_ts_and_seq_beside_kind() { + // Live SSE frame: same `ts` sibling as history (plus `seq`), so the + // renderer is path-agnostic between live + scrollback. + let v = serde_json::to_value(BusEvent { + seq: 7, + ts: 1_700_000_000, + event: LiveEvent::Note { text: "yo".into() }, + }) + .unwrap(); + assert_eq!(v["seq"], 7); + assert_eq!(v["ts"], 1_700_000_000_i64); + assert_eq!(v["kind"], "note"); + } + #[test] fn effort_validation_accepts_only_known_levels() { for level in EFFORT_LEVELS {