vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(#1540): expose per-event ts on live frame + history rows

Browse source
This commit is contained in:
damocles 2026-06-10 14:01:35 +02:00 • committed by mara
commit 50e0ef9b46
1 changed files with 72 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

85
hive-ag3nt/src/events.rs
View file

@ -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<Vec<LiveEvent>> {
fn recent(&self, limit: usize) -> rusqlite::Result<Vec<StoredEvent>> {
let (events, _, _) = self.page(None, limit)?;
Ok(events)
}
@ -293,43 +310,52 @@ impl EventStore {
&self,
before_id: Option<i64>,
limit: usize,
) -> rusqlite::Result<(Vec<LiveEvent>, Option<i64>, bool)> {
) -> rusqlite::Result<(Vec<StoredEvent>, Option<i64>, 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::<LiveEvent>(&s).ok().map(|e| (id, e)))
let ts: i64 = row.get(1)?;
let s: String = row.get(2)?;
Ok(serde_json::from_str::<LiveEvent>(&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::<LiveEvent>(&s).ok().map(|e| (id, e)))
let ts: i64 = row.get(1)?;
let s: String = row.get(2)?;
Ok(serde_json::from_str::<LiveEvent>(&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<LiveEvent> {
pub fn history(&self) -> Vec<StoredEvent> {
let Some(store) = &self.store else {
return Vec::new();
};
@ -1035,7 +1062,7 @@ impl Bus {
&self,
before_id: Option<i64>,
limit: usize,
) -> (Vec<LiveEvent>, Option<i64>, bool) {
) -> (Vec<StoredEvent>, Option<i64>, 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 {