feat(#1187): dynamic terminal scrollback with paginated history

hive-ag3nt/src/events.rs

@@ -19,7 +19,7 @@ use tokio::sync::broadcast;
 const CHANNEL_CAPACITY: usize = 256;
 /// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
 /// sqlite. Older rows are vacuumed on a periodic sweep.
-const HISTORY_CAPACITY: usize = 2000;
+pub const HISTORY_CAPACITY: usize = 2000;
 /// Path to the persisted event db. Overridable via `HYPERHIVE_EVENTS_DB`
 /// for dev / tests; otherwise derived from the agent's harness dir.
 fn events_db_path() -> PathBuf {
@@ -249,20 +249,61 @@ impl EventStore {
     }
 
     fn recent(&self, limit: usize) -> rusqlite::Result<Vec<LiveEvent>> {
+        let (events, _, _) = self.page(None, limit)?;
+        Ok(events)
+    }
+
+    /// Fetch up to `limit` events with id < `before_id` (or the most recent
+    /// `limit` events when `before_id` is `None`). Returns
+    /// `(events_oldest_first, min_row_id, has_more)`.
+    fn page(
+        &self,
+        before_id: Option<i64>,
+        limit: usize,
+    ) -> rusqlite::Result<(Vec<LiveEvent>, Option<i64>, bool)> {
         let limit_i = i64::try_from(limit).unwrap_or(i64::MAX);
         let conn = self.conn.lock().unwrap();
-        let mut stmt = conn.prepare(
-            "SELECT payload_json FROM events
-              ORDER BY id DESC
-              LIMIT ?1",
-        )?;
-        let rows = stmt.query_map(params![limit_i], |row| {
-            let s: String = row.get(0)?;
-            Ok(serde_json::from_str::<LiveEvent>(&s).ok())
-        })?;
-        let mut out: Vec<LiveEvent> = rows.flatten().flatten().collect();
-        out.reverse();
-        Ok(out)
+        // Fetch one extra row so we can tell whether more exist.
+        let fetch = limit_i.saturating_add(1);
+        let rows: Vec<(i64, LiveEvent)> = match before_id {
+            Some(bid) => {
+                let mut stmt = conn.prepare(
+                    "SELECT id, 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)))
+                })?
+                .flatten()
+                .flatten()
+                .collect()
+            }
+            None => {
+                let mut stmt = conn.prepare(
+                    "SELECT id, 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)))
+                })?
+                .flatten()
+                .flatten()
+                .collect()
+            }
+        };
+        let has_more = rows.len() > limit;
+        let mut rows: Vec<(i64, LiveEvent)> = 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();
+        Ok((events, min_id, has_more))
     }
 }
 
@@ -834,6 +875,23 @@ impl Bus {
         };
         store.recent(HISTORY_CAPACITY).unwrap_or_default()
     }
+
+    /// Paginated history: up to `limit` events before `before_id`
+    /// (or the most recent `limit` when `before_id` is `None`).
+    /// Returns `(events_oldest_first, min_row_id, has_more)`.
+    /// `min_row_id` is the cursor for the next page; pass it as
+    /// `before_id` on the next call.
+    #[must_use]
+    pub fn history_page(
+        &self,
+        before_id: Option<i64>,
+        limit: usize,
+    ) -> (Vec<LiveEvent>, Option<i64>, bool) {
+        let Some(store) = &self.store else {
+            return (Vec::new(), None, false);
+        };
+        store.page(before_id, limit).unwrap_or_default()
+    }
 }
 
 impl Default for Bus {

hive-ag3nt/src/web_ui.rs

@@ -717,15 +717,41 @@ async fn post_send(State(state): State<AppState>, Form(form): Form<SendForm>) ->
     }
 }
 
-async fn events_history(State(state): State<AppState>) -> axum::Json<serde_json::Value> {
-    // Capture seq *before* the read so dedupe is "drop buffered events
-    // you've already seen in history", never "lose an event that fired
-    // between the read and the timestamp." Historical rows have no
-    // per-row seq; only the high-water mark matters for the dedupe
-    // window.
-    let seq = state.bus.current_seq();
-    let events = state.bus.history();
-    axum::Json(serde_json::json!({ "seq": seq, "events": events }))
+/// Query params for the paginated history endpoint.
+#[derive(Debug, Deserialize)]
+struct HistoryParams {
+    /// Cursor: only return events with sqlite row id < `before`.
+    /// Omit for the initial (most-recent) page.
+    before: Option<i64>,
+    /// Page size (default 100, capped at HISTORY_CAPACITY).
+    limit: Option<usize>,
+}
+
+async fn events_history(
+    State(state): State<AppState>,
+    axum::extract::Query(params): axum::extract::Query<HistoryParams>,
+) -> axum::Json<serde_json::Value> {
+    use crate::events::HISTORY_CAPACITY;
+    let limit = params.limit.unwrap_or(100).min(HISTORY_CAPACITY);
+    let before = params.before;
+    let is_initial = before.is_none();
+
+    // Capture seq *before* the read on initial loads so the SSE dedupe
+    // window is "drop buffered events you've already seen in history",
+    // never "lose an event that fired between the read and the seq."
+    // On paginated loads (`before` is set) seq is not needed.
+    let seq = if is_initial { Some(state.bus.current_seq()) } else { None };
+
+    let (events, min_id, has_more) = state.bus.history_page(before, limit);
+    let mut resp = serde_json::json!({
+        "events": events,
+        "min_id": min_id,
+        "has_more": has_more,
+    });
+    if let Some(s) = seq {
+        resp["seq"] = serde_json::json!(s);
+    }
+    axum::Json(resp)
 }
 
 async fn events_stream(