ae9ca4331e
swarm-controller now creates `term-main-<agent>` for every live agent, in the same one-minute pass and with the same config style as `term-sub-<agent>` (#4897): subjects `$SWARM.term.<agent>` only, max_age 24h, max_bytes 64 MiB, discard Old, max_message_size from the queue's max_payload. An existing stream is opened as it is. The agent keeps its plain publish on `$SWARM.term.<agent>` and gets no JetStream subject. `GET /api/agents/{name}/term/stream` creates an ephemeral push consumer on that stream starting 100 rows back, so a swarm-ui terminal opens with recent history and then follows new rows, in order, from one subscription. Each SSE event carries the row's stream sequence as its id, and a reconnect's Last-Event-ID resumes after it. Without the stream the route falls back to the live core subscription it used before. The stream is `term-main-<agent>`, not `term-<agent>`: an agent may be named `sub-x`, whose `term-<agent>` would be agent `x`'s subagent stream. Its one literal subject cannot match `$SWARM.term.<agent>.sub.>`, which needs more tokens, so the two streams never overlap. The reader gains `$JS.API.CONSUMER.CREATE.*.*.$SWARM.term.*`, the extended create form whose last tokens are the consumer's filter. Checked against nats-server 2.14.4, the trailing `*` also admits a literal `*` or `>` filter, so a consumer under it reaches any terminal row, main or subagent; those are rows the reader already subscribes to and reads with `STREAM.MSG.GET.*`. Closes #4826
71 lines
2.4 KiB
TypeScript
71 lines
2.4 KiB
TypeScript
// 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<TermRow[]>([]);
|
|
const [connection, setConnection] = useState<ConnectionState>("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 };
|
|
}
|