diff --git a/frontend/packages/shared/src/terminal.css b/frontend/packages/shared/src/terminal.css index bdbc9903..d7cc5b64 100644 --- a/frontend/packages/shared/src/terminal.css +++ b/frontend/packages/shared/src/terminal.css @@ -152,6 +152,33 @@ pointer-events: auto; } .tail-pill:hover { filter: brightness(1.1); } +/* "↑ load older" pill: sits inline at the top of the log (not + absolutely positioned) so it scrolls with the content. Appears + when `has_more` is true after initial history load. */ +.load-more-pill { + display: block; + width: 100%; + background: transparent; + color: var(--muted); + font-family: inherit; + font-size: 0.8em; + font-weight: bold; + letter-spacing: 0.08em; + border: 0; + border-bottom: 1px dashed var(--purple-dim); + padding: 0.4em 1em; + cursor: pointer; + text-align: left; + transition: color 120ms ease, background 120ms ease; +} +.load-more-pill:hover:not(:disabled) { + color: var(--fg); + background: var(--border); +} +.load-more-pill:disabled { + cursor: default; + opacity: 0.6; +} /* Expandable rows reuse the flat-row prefix metrics (padding-left + negative text-indent) so the disclosure glyph (`▸ / ▾`) lands in exactly the same column as flat-row prefix glyphs (`→ ← · ◆ ✓ ✗`). diff --git a/frontend/packages/shared/src/terminal.js b/frontend/packages/shared/src/terminal.js index 2ce101ce..e22a6abc 100644 --- a/frontend/packages/shared/src/terminal.js +++ b/frontend/packages/shared/src/terminal.js @@ -57,8 +57,12 @@ const SCROLL_ANIM_MS = 140; const SCROLL_SNAP_PX = 24; export function create(opts) { - const log = opts.logEl; + // `log` is `let` not `const` so loadMore() can temporarily redirect + // row/details/etc. into a detached temp element while prepending older + // history (restored before any scrollTop adjustments). + let log = opts.logEl; if (!log) throw new Error('HiveTerminal.create: logEl is required'); + const rootLog = log; // always the real DOM element — never reassigned const renderers = opts.renderers || {}; const defaultRender = renderers._default || ((ev, api) => api.row('note', JSON.stringify(ev))); @@ -68,6 +72,11 @@ export function create(opts) { let pill = null; let unseen = 0; let currentNoAnim = false; + // Pagination state for the "load older" feature. + let histMinId = null; + let histHasMore = false; + let histLoading = false; + let loadMoreBtn = null; // Sticky-bottom intent. True means "keep snapping to bottom on // any mutation"; false means "the operator scrolled up — leave // them alone". Updated synchronously from the scroll event @@ -257,6 +266,80 @@ export function create(opts) { } } + // ── Load-older machinery ─────────────────────────────────────── + // + // When the initial backfill response includes `has_more: true`, a + // "↑ load older" button appears at the top of the log. Clicking it + // fetches the next page (`?before=`) and prepends the events + // while holding the viewport steady so the operator's reading position + // doesn't jump. + + function updateLoadMoreBtn() { + if (!histHasMore || !opts.historyUrl) { + if (loadMoreBtn && loadMoreBtn.parentElement) { + loadMoreBtn.parentElement.removeChild(loadMoreBtn); + } + loadMoreBtn = null; + return; + } + if (!loadMoreBtn) { + loadMoreBtn = document.createElement('button'); + loadMoreBtn.type = 'button'; + loadMoreBtn.className = 'load-more-pill'; + loadMoreBtn.addEventListener('click', loadMore); + rootLog.prepend(loadMoreBtn); + } + loadMoreBtn.textContent = '↑ load older'; + loadMoreBtn.disabled = false; + } + + async function loadMore() { + if (!histHasMore || histLoading || !opts.historyUrl || histMinId === null) return; + histLoading = true; + if (loadMoreBtn) { loadMoreBtn.textContent = '↑ loading…'; loadMoreBtn.disabled = true; } + try { + const sep = opts.historyUrl.includes('?') ? '&' : '?'; + const url = opts.historyUrl + sep + 'before=' + histMinId; + const resp = await fetch(url); + if (!resp.ok) { updateLoadMoreBtn(); return; } + const body = await resp.json(); + const events = Array.isArray(body) ? body : (body.events || []); + histHasMore = body.has_more || false; + if (typeof body.min_id === 'number') histMinId = body.min_id; + + if (events.length > 0) { + // Render into a detached element; `log` is temporarily redirected + // so that row/details/etc. append there instead of rootLog. + const tempEl = document.createElement('div'); + log = tempEl; + currentNoAnim = true; + for (const ev of events) dispatch(ev, true); + currentNoAnim = false; + log = rootLog; + + // Separator to mark the boundary between loaded-older and newer. + const sepEl = document.createElement('div'); + sepEl.className = 'row note no-anim'; + sepEl.textContent = '─── older above ───'; + tempEl.appendChild(sepEl); + + // Insert before the "live" divider (i.e. right after the load-more + // button if present, else at the very top of rootLog). + const anchor = loadMoreBtn ? loadMoreBtn.nextSibling : rootLog.firstChild; + const beforeH = rootLog.scrollHeight; + while (tempEl.firstChild) rootLog.insertBefore(tempEl.firstChild, anchor); + // Compensate scroll so the viewport stays on the same content. + rootLog.scrollTop += rootLog.scrollHeight - beforeH; + } + updateLoadMoreBtn(); + } catch (err) { + console.warn('loadMore failed', err); + updateLoadMoreBtn(); + } finally { + histLoading = false; + } + } + // Subscribe → buffer → fetch history → dedupe → apply. // // Race the SSE subscription opens before the history fetch starts. @@ -359,11 +442,16 @@ export function create(opts) { return; } const body = await resp.json(); - // Accept the envelope `{ seq, events }`. A bare array means - // the server hasn't been updated to include seq yet — treat - // it as "no dedupe possible." + // Accept the envelope `{ seq, events, min_id?, has_more? }`. + // A bare array means the server hasn't been updated — treat it + // as "no dedupe possible, no pagination." const events = Array.isArray(body) ? body : (body.events || []); const boundarySeq = Array.isArray(body) ? null : (body.seq ?? null); + // Pagination cursors — set on the outer load-more state. + if (!Array.isArray(body)) { + histHasMore = body.has_more || false; + if (typeof body.min_id === 'number') histMinId = body.min_id; + } // Kinds present in the history replay — the only kinds that // can double and therefore the only ones to seq-dedupe. const historyKinds = new Set(events.map((ev) => ev.kind)); @@ -373,6 +461,8 @@ export function create(opts) { if (events.length) row('note', '─── live (older above) ───'); else placeholder('(connected — waiting for events)'); flushBuffered(boundarySeq, historyKinds); + // Show load-older button if the server reports more history. + updateLoadMoreBtn(); if (opts.onBackfillDone) opts.onBackfillDone(events.length); } catch (err) { console.warn('history backfill failed', err); diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 700c7fcc..ff7fafd9 100644 --- a/hive-ag3nt/src/events.rs +++ b/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> { + 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, + limit: usize, + ) -> rusqlite::Result<(Vec, Option, 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::(&s).ok()) - })?; - let mut out: Vec = 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::(&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::(&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, + limit: usize, + ) -> (Vec, Option, bool) { + let Some(store) = &self.store else { + return (Vec::new(), None, false); + }; + store.page(before_id, limit).unwrap_or_default() + } } impl Default for Bus { diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 2fae46e5..fd4b8bd9 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -717,15 +717,41 @@ async fn post_send(State(state): State, Form(form): Form) -> } } -async fn events_history(State(state): State) -> axum::Json { - // 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, + /// Page size (default 100, capped at HISTORY_CAPACITY). + limit: Option, +} + +async fn events_history( + State(state): State, + axum::extract::Query(params): axum::extract::Query, +) -> axum::Json { + 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(