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 };
}