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

// Backfill + live SSE for the agent's event stream, reduced to a plain
// `TermRow[]` — 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 envelope landing between EventSource-
// open and the history response resolving is buffered, not dropped or
// double-counted (`envelope.seq <= history.seq` → already covered by the
// initial history page, drop it from the buffer). `seq` alone is the
// whole dedup signal, see `TermEnvelope`'s doc in hive-agent's
// `web_ui/stream.rs`.
import { useEffect, useRef, useState } from 'preact/hooks';
import type { TermEnvelope, TermRow } from '../lib/termMsg.js';

export interface UseLiveStreamOptions {
  historyUrl?: string;
  streamUrl?: string;
}

export interface UseLiveStreamResult {
  rows: TermRow[];
  hasMore: boolean;
  loadingMore: boolean;
  loadMore: () => void;
  /** Append a local-only note row (TermInput's `/help` output, error
   * feedback on a failed slash-command POST) — never sent anywhere,
   * purely a client-side echo. */
  pushLocalNote: (text: string) => void;
  /** Wipe the local view only (server-side history is untouched) —
   * TermInput's `/clear`. */
  clearLocal: () => void;
}

function appendRow(rows: TermRow[], row: TermRow): TermRow[] {
  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.coalesce_key && last && last.coalesce_key === row.coalesce_key) {
    const next = rows.slice(0, -1);
    next.push({ ...row, key: last.key });
    return next;
  }
  return [...rows, row];
}

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

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

  const keySeqRef = useRef(0);
  function nextKey(): string {
    keySeqRef.current += 1;
    return 'r' + keySeqRef.current;
  }
  function toRows(env: TermEnvelope, fromHistory: boolean): TermRow[] {
    return env.msgs.map((m) => ({ ...m, key: nextKey(), fromHistory }));
  }
  function appendEnvelope(rows: TermRow[], env: TermEnvelope, fromHistory: boolean): TermRow[] {
    let next = rows;
    for (const row of toRows(env, fromHistory)) next = appendRow(next, row);
    return next;
  }

  const minIdRef = useRef<number | null>(null);
  const liveRef = useRef(false);
  const bufferedRef = useRef<TermEnvelope[]>([]);

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

    function pushLive(env: TermEnvelope) {
      if (env.msgs.length) setRows((prev) => appendEnvelope(prev, env, false));
    }

    const es = new EventSource(streamUrl);
    es.onmessage = (e) => {
      let env: TermEnvelope;
      try {
        env = JSON.parse(e.data);
      } catch {
        setRows((prev) => appendRow(prev, {
          key: 'parse-err-' + Date.now(), level: 'warn', summary: '[parse err] ' + e.data, fromHistory: false,
        }));
        return;
      }
      if (!liveRef.current) {
        bufferedRef.current.push(env);
        return;
      }
      pushLive(env);
    };
    es.onerror = () => {
      setRows((prev) => appendRow(prev, {
        key: 'conn-note', level: 'warn', fromHistory: false, coalesce_key: 'conn-status',
        summary: 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: TermEnvelope[] = 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;
        }
        if (cancelled) return;

        let initial: TermRow[] = [];
        for (const env of events) initial = appendEnvelope(initial, env, true);
        initial = events.length
          ? appendRow(initial, { key: 'live-sep', level: 'debug', fromHistory: true, summary: '─── live (older above) ───' })
          : [{ key: 'placeholder', level: 'debug', fromHistory: true, summary: '(connected — waiting for events)' }];
        setRows(initial);

        const drained = bufferedRef.current;
        bufferedRef.current = [];
        liveRef.current = true;
        for (const env of drained) {
          // Already covered by the initial history page — drop it from
          // the buffer rather than rendering it twice.
          if (boundarySeq != null && typeof env.seq === 'number' && env.seq <= boundarySeq) continue;
          pushLive(env);
        }
      } catch (err) {
        console.warn('history backfill failed', err);
        if (cancelled) return;
        const drained = bufferedRef.current;
        bufferedRef.current = [];
        liveRef.current = true;
        for (const env of drained) pushLive(env);
      }
    }
    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: TermEnvelope[] = 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: TermRow[] = [];
        for (const env of events) older = appendEnvelope(older, env, true);
        older = appendRow(older, {
          key: 'older-sep-' + minIdRef.current, level: 'debug', fromHistory: true, summary: '─── older above ───',
        });
        setRows((prev) => [...older, ...prev]);
      }
    } catch (err) {
      console.warn('loadMore failed', err);
    } finally {
      setLoadingMore(false);
    }
  }

  function pushLocalNote(text: string) {
    setRows((prev) => appendRow(prev, { key: 'local-' + nextKey(), level: 'info', fromHistory: false, summary: text }));
  }
  function clearLocal() {
    setRows([]);
  }

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