hyperhive/hive-agent/src/web_ui/stream.rs

//! Live SSE event stream + history endpoints.

use std::convert::Infallible;

use axum::Json;
use axum::extract::{Query, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use serde::{Deserialize, Serialize};
use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};

use super::AppState;
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
/// the raw event classified into. An event that classifies to zero rows (an
/// agent-state change — `StatusChanged`/`ModelChanged`/etc. — or
/// drop-category noise) never reaches the wire at all; see
/// `crate::term_msg` for why.
///
/// `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).
/// 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 {
    #[serde(skip_serializing_if = "Option::is_none")]
    seq: Option<u64>,
    msgs: Vec<TermMsg>,
}

/// Response body for `GET /api/events/history`. `seq` is omitted from the
/// wire entirely on a paginated (non-initial) load — matches the old
/// `json!` shape, which only ever set the `"seq"` key when `Some`.
#[derive(Serialize)]
pub(super) struct EventsHistoryBody {
    events: Vec<TermEnvelope>,
    min_id: Option<i64>,
    has_more: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    seq: Option<u64>,
}

/// Query params for the paginated history endpoint.
#[derive(Debug, Deserialize)]
pub(super) 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>,
}

pub(super) async fn events_history(
    State(state): State<AppState>,
    Query(params): Query<HistoryParams>,
) -> Json<EventsHistoryBody> {
    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);
    // Classify with the same function the live SSE path uses so history
    // replay and live tail deliver identical shapes. The DB stores raw
    // events; classification is applied at read time here (see
    // `crate::term_msg`). One `ClassifyCtx` for the whole page — tool_use→
    // name correlation (for markdown-vs-plain `recv` result bodies) only
    // works within a single page/connection, not across the live/history
    // boundary; see that module's doc for why that's an accepted
    // degradation.
    let mut ctx = ClassifyCtx::default();
    let events: Vec<TermEnvelope> = events
        .into_iter()
        .filter_map(|se| {
            let msgs = classify(&se.event, se.ts, &mut ctx);
            if msgs.is_empty() {
                None
            } else {
                Some(TermEnvelope { seq: None, msgs })
            }
        })
        .collect();
    Json(EventsHistoryBody {
        events,
        min_id,
        has_more,
        seq,
    })
}

pub(super) async fn events_stream(
    State(state): State<AppState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    tracing::info!("sse: client subscribed");
    let rx = state.bus.subscribe();
    // Prime THIS connection with a one-off "hello" so it can clear the
    // connecting placeholder immediately. Injected into this subscriber's own
    // 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 {
        seq: None,
        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→
    // name correlation persists for the connection's lifetime (see
    // `crate::term_msg::ClassifyCtx`'s doc for the history-page boundary
    // this doesn't cross).
    let mut ctx = ClassifyCtx::default();
    let live = BroadcastStream::new(rx).filter_map(move |res| {
        let ev = res.ok()?;
        let msgs = classify(&ev.event, ev.ts, &mut ctx);
        if msgs.is_empty() {
            return None;
        }
        let envelope = TermEnvelope {
            seq: Some(ev.seq),
            msgs,
        };
        let json = serde_json::to_string(&envelope).ok()?;
        Some(Ok(Event::default().data(json)))
    });
    let stream = tokio_stream::once(Ok(hello)).chain(live);
    Sse::new(stream).keep_alive(KeepAlive::default())
}