hyperhive/frontend/packages/agent/src/hooks/useLiveStream.ts

// Backfill + live SSE for the agent's event stream, reduced to a plain
// `StreamRow[]` — the Preact-data sibling of
// @hive/shared/terminal.js's `create()`. Scroll behaviour is
// deliberately NOT this hook's job (see components/LiveStream.tsx):
// mara flagged the old page's scroll-while-streaming bug explicitly as
// something not to copy 1:1, and keeping "what rows exist" separate
// from "where the viewport is" is what makes that fixable — this hook
// only ever appends/prepends to `rows`, the DOM-owning component
// decides whether that should move the scroll position.
//
// Same subscribe → buffer → fetch-history → seq-dedupe → flush dance as
// terminal.js's `start()`: a live event landing between EventSource-open
// and the history response resolving is buffered, not dropped or
// double-counted (seq <= history.seq AND the event's kind appeared in
// the history replay → already covered, drop it from the buffer).
import { useEffect, useRef, useState } from 'preact/hooks';
import { classifyEvent, createClassifyCtx, type ClassifyCtx } from '../lib/classifyEvent.js';
import type { StreamRow } from '../lib/streamRow.js';

// eslint-disable-next-line @typescript-eslint/no-explicit-any -- raw SSE payload, dynamically shaped
type AnyEvent = any;

export interface UseLiveStreamOptions {
  historyUrl?: string;
  streamUrl?: string;
  /** Fired on every LIVE (not history-replay) turn_start/turn_end, so the
   * caller can trigger a snappier `/api/state` refresh than the plain
   * poll interval — mirrors app.js's turn_end → refreshState()/
   * refreshTodos(). */
  onLiveTurnBoundary?: () => void;
}

export interface UseLiveStreamResult {
  rows: StreamRow[];
  hasMore: boolean;
  loadingMore: boolean;
  loadMore: () => void;
}

function appendRow(rows: StreamRow[], row: StreamRow): StreamRow[] {
  const last = rows[rows.length - 1];
  // Coalesce in place only while the coalescible row is still the last
  // one in the list — any other row landing in between starts a fresh
  // one, same rule as terminal.js's makeCoalescer.
  if (row.coalesceKey && last && last.coalesceKey === row.coalesceKey) {
    const next = rows.slice(0, -1);
    next.push({ ...row, key: last.key });
    return next;
  }
  return [...rows, row];
}

function appendMany(rows: StreamRow[], newRows: StreamRow[]): StreamRow[] {
  let next = rows;
  for (const r of newRows) next = appendRow(next, r);
  return next;
}

export function useLiveStream(opts: UseLiveStreamOptions = {}): UseLiveStreamResult {
  const historyUrl = opts.historyUrl ?? 'events/history';
  const streamUrl = opts.streamUrl ?? 'events/stream';

  const [rows, setRows] = useState<StreamRow[]>([]);
  const [hasMore, setHasMore] = useState(false);
  const [loadingMore, setLoadingMore] = useState(false);

  const ctxRef = useRef<ClassifyCtx | null>(null);
  if (!ctxRef.current) ctxRef.current = createClassifyCtx();
  const minIdRef = useRef<number | null>(null);
  const liveRef = useRef(false);
  const bufferedRef = useRef<AnyEvent[]>([]);
  const onBoundaryRef = useRef(opts.onLiveTurnBoundary);
  onBoundaryRef.current = opts.onLiveTurnBoundary;

  useEffect(() => {
    let cancelled = false;

    function pushLive(ev: AnyEvent) {
      const newRows = classifyEvent(ev, false, ctxRef.current!);
      if (newRows.length) setRows((prev) => appendMany(prev, newRows));
      if (ev.kind === 'turn_start' || ev.kind === 'turn_end') onBoundaryRef.current?.();
    }

    const es = new EventSource(streamUrl);
    es.onmessage = (e) => {
      let ev: AnyEvent;
      try {
        ev = JSON.parse(e.data);
      } catch {
        setRows((prev) => appendRow(prev, {
          key: 'parse-err-' + Date.now(), cssClass: 'note', fromHistory: false,
          text: '[parse err] ' + e.data,
        }));
        return;
      }
      if (!liveRef.current) {
        bufferedRef.current.push(ev);
        return;
      }
      pushLive(ev);
    };
    es.onerror = () => {
      setRows((prev) => appendRow(prev, {
        key: 'conn-note', cssClass: 'note', fromHistory: false, coalesceKey: 'conn-status',
        text: es.readyState === 0 /* CONNECTING */ ? '[reconnecting…]' : '[disconnected]',
      }));
    };

    async function backfill() {
      try {
        const resp = await fetch(historyUrl);
        if (!resp.ok) throw new Error('http ' + resp.status);
        const body = await resp.json();
        const events: AnyEvent[] = Array.isArray(body) ? body : body.events || [];
        const boundarySeq: number | null = Array.isArray(body) ? null : (body.seq ?? null);
        if (!Array.isArray(body)) {
          setHasMore(!!body.has_more);
          if (typeof body.min_id === 'number') minIdRef.current = body.min_id;
        }
        const historyKinds = new Set(events.map((e) => e.kind));
        if (cancelled) return;

        let initial: StreamRow[] = [];
        for (const ev of events) initial = appendMany(initial, classifyEvent(ev, true, ctxRef.current!));
        initial = events.length
          ? appendRow(initial, { key: 'live-sep', cssClass: 'note', fromHistory: true, text: '─── live (older above) ───' })
          : [{ key: 'placeholder', cssClass: 'note', fromHistory: true, text: '(connected — waiting for events)' }];
        setRows(initial);

        const drained = bufferedRef.current;
        bufferedRef.current = [];
        liveRef.current = true;
        for (const ev of drained) {
          if (boundarySeq != null && typeof ev.seq === 'number' && ev.seq <= boundarySeq && historyKinds.has(ev.kind)) {
            continue;
          }
          pushLive(ev);
        }
      } catch (err) {
        console.warn('history backfill failed', err);
        if (cancelled) return;
        const drained = bufferedRef.current;
        bufferedRef.current = [];
        liveRef.current = true;
        for (const ev of drained) pushLive(ev);
      }
    }
    backfill();

    return () => {
      cancelled = true;
      es.close();
    };
  }, [historyUrl, streamUrl]);

  async function loadMore() {
    if (!hasMore || loadingMore || minIdRef.current == null) return;
    setLoadingMore(true);
    try {
      const sep = historyUrl.includes('?') ? '&' : '?';
      const resp = await fetch(historyUrl + sep + 'before=' + minIdRef.current);
      if (!resp.ok) return;
      const body = await resp.json();
      const events: AnyEvent[] = Array.isArray(body) ? body : body.events || [];
      setHasMore(!!body.has_more);
      if (typeof body.min_id === 'number') minIdRef.current = body.min_id;
      if (events.length) {
        let older: StreamRow[] = [];
        for (const ev of events) older = appendMany(older, classifyEvent(ev, true, ctxRef.current!));
        older = appendRow(older, {
          key: 'older-sep-' + minIdRef.current, cssClass: 'note', fromHistory: true, text: '─── older above ───',
        });
        setRows((prev) => [...older, ...prev]);
      }
    } catch (err) {
      console.warn('loadMore failed', err);
    } finally {
      setLoadingMore(false);
    }
  }

  return { rows, hasMore, loadingMore, loadMore };
}