// Live SSE consumer for the swarm-relayed agent terminal // (`GET /api/agents/{name}/term/stream`, swarm-controller's // `term_stream.rs`). Deliberately NOT `@hive/agent`'s `useLiveStream` — // that hook's wire shape is `TermEnvelope` (`{seq, msgs: []}`) plus a // history/backfill dance; the swarm relay forwards one bare `TermMsg` // per SSE event, its replayed rows first and in order on the same // stream, so there's nothing to buffer/dedupe/backfill against. The // browser's own `Last-Event-ID` resumes a reconnect, so rows are kept // across one. Same row-coalescing // rule as `useLiveStream`'s `appendRow` (ported, not imported — one // small helper isn't worth a cross-package hop for). import { useEffect, useRef, useState } from "preact/hooks"; import type { TermMsg, TermRow } from "@hive/shared/term-msg.js"; import { connectionAfterError, type ConnectionState, } from "../../ui/stream/connection.js"; import { probeSession } from "../../ui/session/session.js"; export type { ConnectionState }; function appendRow(rows: TermRow[], row: TermRow): TermRow[] { const last = rows[rows.length - 1]; if (row.coalesce_key && last && last.coalesce_key === row.coalesce_key) { return [...rows.slice(0, -1), { ...row, key: last.key }]; } return [...rows, row]; } export interface UseSwarmTermStreamResult { rows: TermRow[]; connection: ConnectionState; /** Opens the stream again, for one the browser gave up on. */ reconnect: () => void; } export function useSwarmTermStream( streamUrl: string, ): UseSwarmTermStreamResult { const [rows, setRows] = useState([]); const [connection, setConnection] = useState("connecting"); const keySeqRef = useRef(0); // Bumped by `reconnect`, which reruns the effect below. const [attempt, setAttempt] = useState(0); useEffect(() => { setRows([]); setConnection("connecting"); keySeqRef.current = 0; const es = new EventSource(streamUrl); es.onopen = () => setConnection("open"); es.onmessage = (e) => { keySeqRef.current += 1; const key = "r" + keySeqRef.current; let msg: TermMsg; try { msg = JSON.parse(e.data); } catch { setRows((prev) => appendRow(prev, { key, level: "warn", fromHistory: false, summary: "[parse err] " + e.data, }), ); return; } setRows((prev) => appendRow(prev, { ...msg, key, fromHistory: false })); }; es.onerror = () => { const state = connectionAfterError(es.readyState); setConnection(state); // The browser gave up: say whether the session ended, which `EventSource` does not tell. if (state === "closed") probeSession(); }; return () => es.close(); }, [streamUrl, attempt]); return { rows, connection, reconnect: () => setAttempt((n) => n + 1) }; }