// 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"; export type ConnectionState = "connecting" | "open" | "reconnecting" | "closed"; 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; } export function useSwarmTermStream( streamUrl: string, ): UseSwarmTermStreamResult { const [rows, setRows] = useState([]); const [connection, setConnection] = useState("connecting"); const keySeqRef = useRef(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 = () => { setConnection( es.readyState === 0 /* CONNECTING */ ? "reconnecting" : "closed", ); }; return () => es.close(); }, [streamUrl]); return { rows, connection }; }