diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index bce82c8d..4d43f5d9 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -18,6 +18,7 @@ domain, covers host-level detail for one hive. | `/jobs` | the controller's job graph β€” where agent creation and credential mints show progress | | `/schedules` | the scheduled prompts: add, edit, pause, resume and remove | | `/broadcast` | send one message to several agents at once, picked from a grid of their icons | +| `/messages` | the swarm-wide message feed: every message between agents, and to them from the operator, reminders and compaction, with a history window and a filter on sender or recipient | | `/issues` | a cross-repo issue report | | `/review` | the review classifier: its recent runs, a try-out on a real PR, and its settings | | `/builds` | the swarm builder's recent builds of agent config PRs, with what nix printed when one failed | @@ -233,6 +234,28 @@ whole request and sends nothing when `body` or `agents` is empty, when `agents` names an agent twice, names more than 64, or names one that `GET /api/agents` doesn't list. +`GET /api/messages/stream` is an SSE feed of the messages between agents, and +from senders that are no agent (the operator, `reminder`, `compact`) to an +agent. It reads the terminal rows on `$SWARM.term.` and has no hive in +the path. Each event's `data` is `{ts, from, to, body, body_format?}`. An +agent's `send` row gives one event from that agent to the recipient in the +row's summary. A turn started by a non-agent sender gives one event to the +agent whose terminal holds the row. A turn started by an agent gives none, +since that agent's `send` row already did. The query `since_minutes` (default +60, at most 1440, 0 for live only) sends recent history first, oldest first, +read from each agent's newest 100 rows. A message an agent drained with an +in-turn `recv` starts no turn and so is missing. + +The **messages** page (`/messages`) reads that feed. A **history** select picks +the window, from live only up to 24 hours, and a filter keeps the messages whose +sender or recipient contains the text you type. Each message is one row of the +same list an agent's terminal uses, a summary line with the time, the two +parties and the start of the body, and the whole body under it. The list +follows the newest message unless you scroll up, and holds the last 2000. The +feed has no event id, so a reconnect sends the history again, and the page +empties its list each time the connection opens, then fills it from the replay. +Changing the window opens a new stream. + A command has its own subject, so the harness never mistakes typed text for one. `POST /api/agents//cancel` sends `"cancel"` as a NATS request on `$SWARM.control.`, and only swarm-controller may publish there. The diff --git a/frontend/packages/swarm-ui/src/App.tsx b/frontend/packages/swarm-ui/src/App.tsx index 9002d98b..381f127d 100644 --- a/frontend/packages/swarm-ui/src/App.tsx +++ b/frontend/packages/swarm-ui/src/App.tsx @@ -7,6 +7,7 @@ import { AgentTerminalPage } from "./pages/agents/AgentTerminalPage.js"; import { AlertsPage } from "./pages/AlertsPage.js"; import { BuildsPage } from "./pages/builds/BuildsPage.js"; import { BroadcastPage } from "./pages/broadcast/BroadcastPage.js"; +import { MessagesPage } from "./pages/messages/MessagesPage.js"; import { ComponentsPage } from "./pages/ComponentsPage.js"; import { JobsPage } from "./pages/JobsPage.js"; import { HivesPage } from "./pages/HivesPage.js"; @@ -41,6 +42,7 @@ export function App() { + diff --git a/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.css b/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.css new file mode 100644 index 00000000..6fe1d758 --- /dev/null +++ b/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.css @@ -0,0 +1,28 @@ +/* : the filter bar over the terminal-style message list. */ +@import "@hive/shared/terminal.css"; + +.ui-messages-bar { + display: flex; + flex-wrap: wrap; + align-items: flex-end; + gap: 0.5em 1em; + margin-block-end: 0.75em; +} +.ui-messages-bar > .ui-form-field { + flex: 0 1 16em; +} +.ui-messages-count { + margin-inline-start: auto; + color: var(--muted); +} +.ui-messages-wrap.terminal-wrap { + max-height: 70vh; +} +.ui-messages-wrap .live.terminal { + box-sizing: border-box; + max-height: 70vh; +} +.ui-messages-note { + margin: 0.75em 0 0; + color: var(--muted); +} diff --git a/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.tsx b/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.tsx new file mode 100644 index 00000000..906a1296 --- /dev/null +++ b/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.tsx @@ -0,0 +1,136 @@ +// β€” `/messages`: every message between agents, and from the +// operator, reminders and compaction to an agent, as the swarm's controller +// streams them (`GET /api/messages/stream`). The read side of the hive +// dashboard's flow page, across all hives. +// +// The stream sends the chosen window of history first, then the live tail. A +// message an agent drained with an in-turn `recv` is missing, because it starts +// no turn; the page says so under the list. Each message is a row of the same +// terminal list an agent's own terminal uses, one summary line with the time, +// the two parties and the start of the body, and the whole body under it. +import { memo } from "preact/compat"; +import { useLayoutEffect, useRef, useState } from "preact/hooks"; +import { Badge, type BadgeTone } from "@hive/shared/badge.js"; +import { Row } from "@hive/shared/term-row.js"; +import { Panel } from "../../ui/panel/Panel.js"; +import { SelectField } from "../../ui/select-field/SelectField.js"; +import { TextField } from "../../ui/text-field/TextField.js"; +import { + type ArrivedMessage, + DEFAULT_WINDOW_MINUTES, + filterMessages, + messageRow, + WINDOWS, +} from "./messages.js"; +import { useMessageStream, type ConnectionState } from "./useMessageStream.js"; +import "./MessagesPage.css"; + +const NEAR_BOTTOM_PX = 48; + +const CONNECTION_LABEL: Record = { + connecting: "connecting…", + open: "live", + reconnecting: "reconnecting…", + closed: "disconnected", +}; + +const CONNECTION_TONE: Record = { + connecting: "neutral", + open: "positive", + reconnecting: "warning", + closed: "negative", +}; + +// One message's row. Memoised on the message, which keeps its identity while it +// is on the page, so a new arrival or a keystroke in the filter re-renders only +// the rows that changed, not every body's markdown. +const MessageRow = memo(function MessageRow({ m }: { m: ArrivedMessage }) { + return ; +}); + +export function MessagesPage() { + const [sinceMinutes, setSinceMinutes] = useState(DEFAULT_WINDOW_MINUTES); + const [query, setQuery] = useState(""); + const { messages, connection } = useMessageStream(sinceMinutes); + const shown = filterMessages(messages, query); + + const logRef = useRef(null); + const [stickToBottom, setStickToBottom] = useState(true); + + function handleScroll() { + const el = logRef.current; + if (!el) return; + setStickToBottom( + el.scrollHeight - el.scrollTop - el.clientHeight <= NEAR_BOTTOM_PX, + ); + } + + // After new rows are committed, follow the tail unless the reader has + // scrolled up. The newest id changes with every arrival, which the count + // does not once the list is full and each arrival pushes one off the front. + const newestId = shown.length > 0 ? shown[shown.length - 1].id : -1; + useLayoutEffect(() => { + const el = logRef.current; + if (el && stickToBottom) el.scrollTop = el.scrollHeight - el.clientHeight; + }, [newestId, shown.length, stickToBottom]); + + return ( + + } + > +
+ setSinceMinutes(Number(v))} + options={WINDOWS.map((w) => ({ + value: String(w.minutes), + label: w.label, + }))} + /> + + + {query.trim() === "" + ? `${messages.length} messages` + : `${shown.length} of ${messages.length} messages`} + +
+
+
+ {shown.length === 0 ? ( +
+ + {connection === "open" + ? messages.length === 0 + ? "(connected β€” no messages yet)" + : "(no message matches the filter)" + : CONNECTION_LABEL[connection]} + +
+ ) : ( + shown.map((m) => ) + )} +
+
+

+ Shows what the agents sent each other. A message an agent drained with + an in-turn recv is missing. The history is at most 100 + recent messages per agent. +

+ + ); +} diff --git a/frontend/packages/swarm-ui/src/pages/messages/messages.ts b/frontend/packages/swarm-ui/src/pages/messages/messages.ts new file mode 100644 index 00000000..ebbc8659 --- /dev/null +++ b/frontend/packages/swarm-ui/src/pages/messages/messages.ts @@ -0,0 +1,108 @@ +// The messages page's logic, kept out of the component so `npm test` can run +// it: the stream URL, which messages a filter keeps, and how one message +// becomes a terminal row. +import type { TermRow } from "@hive/shared/term-msg.js"; + +/** One event of `GET /api/messages/stream`; mirrors `SwarmMessage`. */ +export interface SwarmMessage { + /** ISO 8601 / RFC 3339 UTC. */ + ts: string; + from: string; + to: string; + body: string; + body_format?: "markdown" | "diff"; +} + +/** + * A message with the id the page gave it when it arrived. The id only grows, + * so it names a row for as long as the message is on the page, however many + * older ones drop off the front. + */ +export interface ArrivedMessage extends SwarmMessage { + id: number; +} + +/** `batch` numbered from `firstId` on, in order. */ +export function numbered( + batch: SwarmMessage[], + firstId: number, +): ArrivedMessage[] { + return batch.map((m, i) => ({ ...m, id: firstId + i })); +} + +/** The history windows the page offers, in minutes; 0 is live only. */ +export const WINDOWS: { minutes: number; label: string }[] = [ + { minutes: 0, label: "live only" }, + { minutes: 15, label: "last 15 min" }, + { minutes: 60, label: "last hour" }, + { minutes: 360, label: "last 6 hours" }, + { minutes: 1440, label: "last 24 hours" }, +]; + +export const DEFAULT_WINDOW_MINUTES = 60; + +/** The stream for `sinceMinutes` of history before the live tail. */ +export function messagesUrl(sinceMinutes: number): string { + return `/api/messages/stream?since_minutes=${sinceMinutes}`; +} + +/** + * The messages that name `query` in their sender or recipient, as a + * case-insensitive substring; all of them when `query` is blank. + */ +export function filterMessages( + messages: T[], + query: string, +): T[] { + const q = query.trim().toLowerCase(); + if (q === "") return messages; + return messages.filter( + (m) => m.from.toLowerCase().includes(q) || m.to.toLowerCase().includes(q), + ); +} + +/** At most `max` messages, the newest ones, when there are more. */ +export function newest( + messages: T[], + max: number, +): T[] { + return messages.length > max + ? messages.slice(messages.length - max) + : messages; +} + +/** `body`'s first non-blank line, cut to `max` characters with an ellipsis. */ +export function snippet(body: string, max: number): string { + const line = + body + .split("\n") + .map((l) => l.trim()) + .find((l) => l !== "") ?? ""; + return line.length > max ? line.slice(0, max - 1) + "…" : line; +} + +/** `ts` as `HH:MM:SS` in the viewer's time zone; `ts` itself when it is no date. */ +export function clock(ts: string): string { + const d = new Date(ts); + if (Number.isNaN(d.getTime())) return ts; + const p = (n: number) => String(n).padStart(2, "0"); + return `${p(d.getHours())}:${p(d.getMinutes())}:${p(d.getSeconds())}`; +} + +/** + * The terminal row for `m`: one summary line with the time, the two parties + * and the start of the body, and the whole body under it. + */ +export function messageRow(m: SwarmMessage, key: string): TermRow { + const start = snippet(m.body, 80); + return { + key, + ts: m.ts, + icon: "πŸ’¬", + level: "info", + summary: `${clock(m.ts)} ${m.from} β†’ ${m.to}${start ? ` β€” ${start}` : ""}`, + body: m.body, + body_format: m.body_format, + fromHistory: false, + }; +} diff --git a/frontend/packages/swarm-ui/src/pages/messages/useMessageStream.ts b/frontend/packages/swarm-ui/src/pages/messages/useMessageStream.ts new file mode 100644 index 00000000..663935ed --- /dev/null +++ b/frontend/packages/swarm-ui/src/pages/messages/useMessageStream.ts @@ -0,0 +1,76 @@ +// The live feed of `GET /api/messages/stream`, one `SwarmMessage` per event. +// The stream has no event id, so a reconnect sends the history again from the +// start; the list therefore empties each time the connection opens, and the +// replay refills it. Not `useSwarmTermStream`: that one keeps its rows across +// a reconnect, which here would list every message twice. +// +// Each message gets an id as it arrives, and the events that arrive within one +// frame join the list in a single update, so a long replay is one render per +// frame instead of one per message. +import { useEffect, useRef, useState } from "preact/hooks"; +import type { ArrivedMessage, SwarmMessage } from "./messages.js"; +import { messagesUrl, newest, numbered } from "./messages.js"; + +export type ConnectionState = "connecting" | "open" | "reconnecting" | "closed"; + +/** The most messages the page holds; older ones drop off. */ +export const MAX_MESSAGES = 2000; + +export function useMessageStream(sinceMinutes: number): { + messages: ArrivedMessage[]; + connection: ConnectionState; +} { + const [messages, setMessages] = useState([]); + const [connection, setConnection] = useState("connecting"); + // The next id to hand out; it never goes back, so an id is not reused after + // a reconnect either. + const nextId = useRef(0); + + useEffect(() => { + setMessages([]); + setConnection("connecting"); + const es = new EventSource(messagesUrl(sinceMinutes)); + let pending: SwarmMessage[] = []; + let frame: number | null = null; + + function drop() { + pending = []; + if (frame !== null) cancelAnimationFrame(frame); + frame = null; + } + function flush() { + frame = null; + const batch = numbered(pending, nextId.current); + pending = []; + nextId.current += batch.length; + setMessages((prev) => newest([...prev, ...batch], MAX_MESSAGES)); + } + + es.onopen = () => { + drop(); + setMessages([]); + setConnection("open"); + }; + es.onmessage = (e) => { + let m: SwarmMessage; + try { + m = JSON.parse(e.data) as SwarmMessage; + } catch { + return; + } + pending.push(m); + if (frame === null) frame = requestAnimationFrame(flush); + }; + es.onerror = () => { + setConnection( + es.readyState === 0 /* CONNECTING */ ? "reconnecting" : "closed", + ); + }; + return () => { + drop(); + es.close(); + }; + }, [sinceMinutes]); + + return { messages, connection }; +} diff --git a/frontend/packages/swarm-ui/src/shell/Shell.tsx b/frontend/packages/swarm-ui/src/shell/Shell.tsx index d79b771d..6d23763b 100644 --- a/frontend/packages/swarm-ui/src/shell/Shell.tsx +++ b/frontend/packages/swarm-ui/src/shell/Shell.tsx @@ -55,6 +55,7 @@ const NAV_ITEMS: { href: string; label: string; accent: string }[] = [ { href: "/jobs", label: "jobs", accent: "var(--pink)" }, { href: "/schedules", label: "schedules", accent: "var(--cyan)" }, { href: "/broadcast", label: "broadcast", accent: "var(--blue)" }, + { href: "/messages", label: "messages", accent: "var(--purple)" }, { href: "/issues", label: "issues", accent: "var(--yellow)" }, { href: "/review", label: "review", accent: "var(--amber)" }, { href: "/builds", label: "builds", accent: "var(--green)" }, diff --git a/frontend/packages/swarm-ui/test/messages.test.mjs b/frontend/packages/swarm-ui/test/messages.test.mjs new file mode 100644 index 00000000..56771ec3 --- /dev/null +++ b/frontend/packages/swarm-ui/test/messages.test.mjs @@ -0,0 +1,128 @@ +// The messages page's logic: the stream URL, the filter, the cap, and how a +// message becomes a terminal row. Run by `npm test`, which Node runs with its +// built-in test runner and type stripping, so the `.ts` module is imported as +// it is. +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { + clock, + filterMessages, + messageRow, + messagesUrl, + newest, + numbered, + snippet, +} from "../src/pages/messages/messages.ts"; + +const msg = (from, to, body = "hi", extra = {}) => ({ + ts: "2026-10-18T12:00:00Z", + from, + to, + body, + ...extra, +}); + +test("the stream URL carries the history window", () => { + assert.equal(messagesUrl(60), "/api/messages/stream?since_minutes=60"); + assert.equal(messagesUrl(0), "/api/messages/stream?since_minutes=0"); +}); + +test("a blank filter keeps every message", () => { + const all = [msg("a", "b"), msg("c", "d")]; + assert.deepEqual(filterMessages(all, ""), all); + assert.deepEqual(filterMessages(all, " "), all); +}); + +test("the filter matches sender or recipient, ignoring case", () => { + const all = [msg("atlas", "iris"), msg("iris", "coder"), msg("a", "b")]; + assert.deepEqual( + filterMessages(all, "IRIS").map((m) => `${m.from}>${m.to}`), + ["atlas>iris", "iris>coder"], + ); + assert.deepEqual(filterMessages(all, "cod").length, 1); + assert.deepEqual(filterMessages(all, "nobody"), []); +}); + +test("the filter does not search the body", () => { + assert.deepEqual(filterMessages([msg("a", "b", "talk to iris")], "iris"), []); +}); + +test("the cap keeps the newest messages and leaves a short list alone", () => { + const all = [1, 2, 3, 4, 5].map((n) => msg("a", "b", String(n))); + assert.deepEqual( + newest(all, 3).map((m) => m.body), + ["3", "4", "5"], + ); + assert.equal(newest(all, 5), all); + assert.equal(newest(all, 10), all); +}); + +test("arrivals are numbered in order from the id given", () => { + const batch = numbered([msg("a", "b", "x"), msg("c", "d", "y")], 7); + assert.deepEqual( + batch.map((m) => [m.id, m.body]), + [ + [7, "x"], + [8, "y"], + ], + ); + assert.deepEqual(numbered([], 3), []); +}); + +test("in a full list the newest id still changes with every arrival", () => { + // The page follows the tail on the newest id, because the count stops + // moving once the cap drops one message per arrival. + let list = numbered( + [1, 2, 3].map((n) => msg("a", "b", String(n))), + 0, + ); + const seen = new Set(); + for (let id = 3; id < 8; id++) { + list = newest([...list, ...numbered([msg("a", "b", "n")], id)], 3); + assert.equal(list.length, 3); + seen.add(list[list.length - 1].id); + } + assert.equal(seen.size, 5); +}); + +test("a number-tagged message keeps its id through the filter and the cap", () => { + const all = numbered([msg("atlas", "iris"), msg("a", "b")], 40); + assert.deepEqual( + filterMessages(all, "atlas").map((m) => m.id), + [40], + ); + assert.deepEqual( + newest(all, 1).map((m) => m.id), + [41], + ); +}); + +test("a snippet is the first non-blank line, cut with an ellipsis", () => { + assert.equal(snippet("\n\n hello there \nsecond", 80), "hello there"); + assert.equal(snippet("abcdefghij", 5), "abcd…"); + assert.equal(snippet("abcde", 5), "abcde"); + assert.equal(snippet(" \n ", 5), ""); +}); + +test("the clock reads hours, minutes and seconds, and passes a non-date through", () => { + assert.match(clock("2026-10-18T12:00:05Z"), /^\d\d:\d\d:\d\d$/); + assert.equal(clock("not a date"), "not a date"); +}); + +test("a message becomes an info row with its parties in the summary", () => { + const row = messageRow(msg("atlas", "iris", "ready\nmore"), "k1"); + assert.equal(row.key, "k1"); + assert.equal(row.level, "info"); + assert.equal(row.body, "ready\nmore"); + assert.equal(row.fromHistory, false); + assert.match(row.summary, /atlas β†’ iris β€” ready$/); +}); + +test("the body format rides on the row, and an empty body leaves no dash", () => { + assert.equal( + messageRow(msg("a", "b", "x", { body_format: "diff" }), "k").body_format, + "diff", + ); + assert.equal(messageRow(msg("a", "b", "x"), "k").body_format, undefined); + assert.doesNotMatch(messageRow(msg("a", "b", ""), "k").summary, / β€” /); +}); diff --git a/hive-agent/src/term_msg.rs b/hive-agent/src/term_msg.rs index caeddcc8..55553c95 100644 --- a/hive-agent/src/term_msg.rs +++ b/hive-agent/src/term_msg.rs @@ -11,6 +11,7 @@ //! both a text block and a `tool_use` block produces two rows. pub use hive_sh4re::term_msg::{ClassifyCtx, Level, TermMsg, iso8601_utc}; +use hive_sh4re::term_msg::{TURN_ICON, TURN_SUMMARY_PREFIX}; use crate::events::LiveEvent; @@ -33,7 +34,8 @@ fn classify_rows(ev: &LiveEvent, ctx: &mut ClassifyCtx) -> Vec { LiveEvent::TurnStart { from, body, .. } => { // `unread` (the third field) was dropped β€” mara: "i can see // the todo/inbox count in toolbar and dont need the N unread". - let mut m = TermMsg::new(Level::Info, format!("TURN ← {from}")).icon("β—†"); + let mut m = + TermMsg::new(Level::Info, format!("{TURN_SUMMARY_PREFIX}{from}")).icon(TURN_ICON); if !body.trim().is_empty() { m = m.body(body.clone(), None); } diff --git a/hive-sh4re/src/stream_enrich.rs b/hive-sh4re/src/stream_enrich.rs index 0cb709e3..67ac545d 100644 --- a/hive-sh4re/src/stream_enrich.rs +++ b/hive-sh4re/src/stream_enrich.rs @@ -15,7 +15,7 @@ //! `mcp__hyperhive__send`'s row say, which tools get an expandable body) //! didn't change; only the shape it gets packed into did. -use crate::term_msg::{BodyFormat, ClassifyCtx, Level, TermMsg}; +use crate::term_msg::{BodyFormat, ClassifyCtx, Level, SEND_ICON, SEND_SUMMARY_PREFIX, TermMsg}; use serde_json::{Value, json}; /// Classify one raw claude stream-json line into zero or more terminal rows. @@ -490,7 +490,7 @@ fn rich_tool_body(name: &str, input: &Value) -> Option<(String, &'static str)> { fn tool_icon(name: &str) -> &'static str { // Exact-match table first, then prefix/contains fallbacks. match name { - "mcp__hyperhive__send" => "πŸ“€", + "mcp__hyperhive__send" => SEND_ICON, "mcp__hyperhive__recv" => "πŸ“₯", "mcp__hyperhive__remind" => "⏰", "mcp__hyperhive__set_status" => "🏷️", @@ -610,9 +610,9 @@ fn fmt_hyperhive_message_tool(name: &str, short: &str, input: &Value) -> String let to = sv(input, "to"); let lines = sv(input, "body").lines().count(); if lines > 1 { - format!("{short} β†’ {to} Β· {lines}L") + format!("{SEND_SUMMARY_PREFIX}{to} Β· {lines}L") } else { - format!("{short} β†’ {to}") + format!("{SEND_SUMMARY_PREFIX}{to}") } } _ => fmt_args_generic(short, input), @@ -1331,4 +1331,29 @@ mod tests { assert_eq!(fmt_room("!Γ„Γ–Γœ:server"), "!Γ„Γ–Γœ"); assert_eq!(fmt_room("!Γ„Γ–Γœ"), "!Γ„Γ–Γœ"); } + + /// The send row the classifier builds is the shape `send_recipient` + /// reads back, with and without the line-count suffix. + #[test] + fn a_send_row_names_its_recipient_through_the_shared_parser() { + for (body, lines) in [("hi", false), ("one\ntwo\nthree", true)] { + let row = one(&json!({ + "type": "assistant", + "message": { "content": [{ + "type": "tool_use", "id": "s1", "name": "mcp__hyperhive__send", + "input": { "to": "iris", "body": body }, + }]}, + })); + assert_eq!(row.icon.as_deref(), Some(SEND_ICON)); + assert_eq!(row.summary.contains(" Β· 3L"), lines, "{}", row.summary); + assert_eq!(crate::term_msg::send_recipient(&row.summary), Some("iris")); + } + assert_eq!(crate::term_msg::send_recipient("send* β†’"), None); + assert_eq!(crate::term_msg::send_recipient("Read(x)"), None); + // A suffix lookalike in the name is not stripped. + assert_eq!( + crate::term_msg::send_recipient("send* β†’ a Β· xL"), + Some("a Β· xL") + ); + } } diff --git a/hive-sh4re/src/term_msg.rs b/hive-sh4re/src/term_msg.rs index d896934c..429e45be 100644 --- a/hive-sh4re/src/term_msg.rs +++ b/hive-sh4re/src/term_msg.rs @@ -128,6 +128,41 @@ pub fn iso8601_utc(unix_seconds: i64) -> String { .to_rfc3339_opts(SecondsFormat::Secs, true) } +/// Icon of the row an agent's `send` tool call makes. +pub const SEND_ICON: &str = "πŸ“€"; + +/// What a `send` row's summary starts with; the recipient follows. A +/// multi-line body appends ` Β· L`, which [`send_recipient`] drops. +pub const SEND_SUMMARY_PREFIX: &str = "send* β†’ "; + +/// Icon of the row a turn started by a message makes. +pub const TURN_ICON: &str = "β—†"; + +/// What a turn-start row's summary starts with; the sender follows. +pub const TURN_SUMMARY_PREFIX: &str = "TURN ← "; + +/// The recipient a `send` row's summary names, without the line-count suffix. +#[must_use] +pub fn send_recipient(summary: &str) -> Option<&str> { + let rest = summary.strip_prefix(SEND_SUMMARY_PREFIX)?; + Some(match rest.rsplit_once(" Β· ") { + Some((to, lines)) + if lines + .strip_suffix('L') + .is_some_and(|n| !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit())) => + { + to + } + _ => rest, + }) +} + +/// The sender a turn-start row's summary names. +#[must_use] +pub fn turn_sender(summary: &str) -> Option<&str> { + summary.strip_prefix(TURN_SUMMARY_PREFIX) +} + /// Per-connection/per-request classification state. A live SSE stream keeps /// one of these alive for the connection's lifetime β€” `tool_use` id β†’ name /// correlation, so a `tool_result` can tell it's answering a `recv` call and diff --git a/swarm-controller-api/src/lib.rs b/swarm-controller-api/src/lib.rs index fc9c47d1..5c515656 100644 --- a/swarm-controller-api/src/lib.rs +++ b/swarm-controller-api/src/lib.rs @@ -7,4 +7,5 @@ pub mod agents; pub mod agents_md; pub mod forge; +pub mod messages; pub mod schedules; diff --git a/swarm-controller-api/src/messages.rs b/swarm-controller-api/src/messages.rs new file mode 100644 index 00000000..56661b37 --- /dev/null +++ b/swarm-controller-api/src/messages.rs @@ -0,0 +1,20 @@ +//! `GET /api/messages/stream`: the swarm-wide message feed. + +use serde::{Deserialize, Serialize}; + +/// One message between two parties, as one event of `GET /api/messages/stream`. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))] +pub struct SwarmMessage { + /// When the sender's terminal row was stamped, ISO 8601 / RFC 3339 UTC. + pub ts: String, + /// An agent, or a non-agent sender such as `reminder`, `compact` or the + /// operator's name. + pub from: String, + /// The receiving agent. + pub to: String, + pub body: String, + /// `markdown` or `diff`; absent for plain text. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub body_format: Option, +} diff --git a/swarm-controller/src/agent_input.rs b/swarm-controller/src/agent_input.rs index aeaf8f7c..b371c428 100644 --- a/swarm-controller/src/agent_input.rs +++ b/swarm-controller/src/agent_input.rs @@ -205,7 +205,7 @@ fn queue(state: &AppState) -> Result { } /// The swarm roster, or the 503 `GET /api/agents` answers when it has none. -async fn roster(state: &AppState) -> Result, ProblemDetails> { +pub(crate) async fn roster(state: &AppState) -> Result, ProblemDetails> { let Some(bridge) = state.auth.as_ref() else { return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 1e927f79..545bf774 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -48,6 +48,7 @@ mod issue_report; mod jobq; mod linked_accounts; mod matrix_account; +mod message_stream; mod otel_http_client; mod pulls; mod queue_identity; diff --git a/swarm-controller/src/message_stream.rs b/swarm-controller/src/message_stream.rs new file mode 100644 index 00000000..c130686b --- /dev/null +++ b/swarm-controller/src/message_stream.rs @@ -0,0 +1,411 @@ +//! `GET /api/messages/stream` β€” an SSE feed of every message between agents, +//! and from non-agent senders to agents, read from the agents' terminal rows +//! on the swarm queue. No hive is involved. +//! +//! **Two kinds of row make an event.** An agent's `send` call leaves a +//! [`SEND_ICON`] row on its own terminal; the sender is that terminal's +//! agent, the recipient is in the summary. A turn started by a message leaves +//! a [`TURN_ICON`] row on the receiver's terminal; it makes an event only when +//! the sender is not in the roster (`reminder`, `compact`, the operator), since +//! an agent sender's `send` row already made one. +//! +//! **Not every message appears.** Messages an agent drains with an in-turn +//! `recv` start no turn and leave no `TURN ←` row, and a `send` row exists +//! only while the sender's terminal publisher was up. A sender added to the +//! roster after the connection opened counts as a non-agent until the next +//! connection, so its messages show twice. +//! +//! **History, then live, with no gap.** The live subscription on +//! `$SWARM.term.*` (one token, so subagent subjects are excluded) opens first. +//! Then each roster agent's stream gives its newest +//! [`agent_term::REPLAY_ROWS`] rows, which are filtered by `since_minutes`, +//! merged by `ts` and sent; a live event equal to a history event is dropped. +//! History is bounded by those 100 rows per agent, not only by `since_minutes`. + +use std::collections::HashSet; +use std::convert::Infallible; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use axum::extract::rejection::QueryRejection; +use axum::extract::{Query, State}; +use axum::http::StatusCode; +use axum::response::sse::{Event, KeepAlive, Sse}; +use futures_util::{Stream, StreamExt as _, stream}; +use hive_sh4re::term_msg::{SEND_ICON, TURN_ICON, iso8601_utc, send_recipient, turn_sender}; +use serde::Deserialize; +use swarm_controller_api::messages::SwarmMessage; +use swarm_queue_client::agent_term; +use utoipa::IntoParams; + +use crate::routes::{AppState, error_problem}; + +/// `since_minutes` when the query names none. +const DEFAULT_SINCE_MINUTES: u64 = 60; + +/// The largest `since_minutes`: the stream's own retention. +const MAX_SINCE_MINUTES: u64 = 24 * 60; + +/// How long a stream that has delivered its rows stays quiet before the +/// replay counts as caught up. +const HISTORY_IDLE: Duration = Duration::from_millis(500); + +/// How many agents' histories are read at once. +const HISTORY_CONCURRENCY: usize = 16; + +/// `stream_messages`'s query. +#[derive(Deserialize, IntoParams)] +pub struct MessagesQuery { + /// How many minutes of history to send before the live tail. Default + /// 60, at most 1440; 0 sends live messages only. + since_minutes: Option, +} + +#[utoipa::path( + get, + path = "/api/messages/stream", + params(MessagesQuery), + responses( + (status = 200, description = "server-sent event stream; each event's `data` is \ + one `SwarmMessage` as JSON: first the recent history, oldest first, then \ + live messages. Built from agents' terminal rows, so a message an agent \ + drained with an in-turn `recv` is missing", + body = SwarmMessage, content_type = "text/event-stream"), + (status = 400, description = "`since_minutes` is not a number (problem+json)", + body = String), + (status = 503, description = "no swarm queue or identity bridge is configured on \ + this host, the roster could not be read, or the queue could not be reached \ + (problem+json)", body = String), + (status = 500, description = "the subscribe failed (problem+json)", body = String), + ), + tag = "agents" +)] +pub(crate) async fn stream_messages( + State(state): State, + query: Result, QueryRejection>, +) -> Result>>, problem_details::ProblemDetails> { + let Query(query) = query.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; + let minutes = query + .since_minutes + .unwrap_or(DEFAULT_SINCE_MINUTES) + .min(MAX_SINCE_MINUTES); + let Some(status) = state.status.as_ref() else { + return Err(error_problem( + StatusCode::SERVICE_UNAVAILABLE, + "no swarm queue is configured on this host", + )); + }; + let roster = crate::agent_input::roster(&state).await?; + let client = status.queue_client(); + swarm_queue_client::ensure_connected(&client).map_err(|e| { + error_problem( + StatusCode::SERVICE_UNAVAILABLE, + &swarm_queue_client::chain(&e), + ) + })?; + + let subscriber = crate::term_stream::subscribe( + &client, + swarm_queue_client::term_subject_wildcard(), + "messages", + ) + .await?; + let roster: Arc> = Arc::new(roster.into_iter().collect()); + let history = history(&client, &roster, minutes).await; + let seen: HashSet = history.iter().cloned().collect(); + let live = live(subscriber, roster).filter(move |m| std::future::ready(!seen.contains(m))); + let events = stream::iter(history) + .chain(live) + .map(|m| Ok(message_event(&m))); + Ok(Sse::new(events).keep_alive(KeepAlive::default())) +} + +fn message_event(message: &SwarmMessage) -> Event { + // A struct of strings always serializes. + Event::default().data(serde_json::to_string(message).unwrap_or_default()) +} + +/// The messages in the newest rows of each of `roster`'s streams from the last +/// `minutes`, oldest first. An agent whose stream cannot be read adds none. +async fn history( + client: &async_nats::Client, + roster: &Arc>, + minutes: u64, +) -> Vec { + if minutes == 0 { + return Vec::new(); + } + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX)); + let cutoff = iso8601_utc(now.saturating_sub(i64::try_from(minutes * 60).unwrap_or(i64::MAX))); + let mut messages: Vec = stream::iter(roster.iter().cloned().collect::>()) + .map(|agent| agent_history(client.clone(), agent, roster.clone())) + .buffer_unordered(HISTORY_CONCURRENCY) + .flat_map(stream::iter) + .filter(|m| std::future::ready(m.ts >= cutoff)) + .collect() + .await; + messages.sort_by(|a, b| a.ts.cmp(&b.ts)); + messages +} + +/// The messages in `agent`'s replayed rows, read until the stream goes quiet. +async fn agent_history( + client: async_nats::Client, + agent: String, + roster: Arc>, +) -> Vec { + let replayed = crate::term_stream::replay( + &client, + &agent_term::stream_name(&agent), + swarm_queue_client::term_subject(&agent), + None, + ) + .await; + let mut rows = match replayed { + Ok(rows) => rows, + Err(error) => { + tracing::info!(%agent, %error, "messages stream: no history for this agent"); + return Vec::new(); + } + }; + let mut messages = Vec::new(); + while let Ok(Some(row)) = tokio::time::timeout(HISTORY_IDLE, rows.next()).await { + messages.extend(message_of(&agent, &row.payload, |a| roster.contains(a))); + } + messages +} + +/// The messages in the rows arriving on `$SWARM.term.` subjects. +fn live( + subscriber: async_nats::Subscriber, + roster: Arc>, +) -> impl Stream { + subscriber.filter_map(move |msg| { + let message = agent_of(msg.subject.as_str()) + .and_then(|agent| message_of(agent, &msg.payload, |a| roster.contains(a))); + std::future::ready(message) + }) +} + +/// The agent a main terminal subject belongs to. A subagent's subject has +/// more tokens and names none. +fn agent_of(subject: &str) -> Option<&str> { + let agent = subject.strip_prefix(&swarm_queue_client::term_subject(""))?; + (!agent.is_empty() && !agent.contains('.')).then_some(agent) +} + +/// The fields of a terminal row this feed reads. +#[derive(Deserialize)] +struct RowView { + #[serde(default)] + ts: String, + icon: Option, + #[serde(default)] + summary: String, + body: Option, + body_format: Option, +} + +/// The message in `agent`'s terminal row `payload`, if it holds one. +/// `is_agent` says whether a turn's sender is an agent. +fn message_of( + agent: &str, + payload: &[u8], + is_agent: impl Fn(&str) -> bool, +) -> Option { + let row: RowView = serde_json::from_slice(payload).ok()?; + let (from, to) = match row.icon.as_deref()? { + SEND_ICON => ( + agent, + send_recipient(&row.summary).filter(|to| !to.is_empty())?, + ), + TURN_ICON => { + let sender = turn_sender(&row.summary).filter(|s| !s.is_empty() && !is_agent(s))?; + (sender, agent) + } + _ => return None, + }; + Some(SwarmMessage { + ts: row.ts, + from: from.to_owned(), + to: to.to_owned(), + body: row.body.unwrap_or_default(), + body_format: row.body_format, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + use serde_json::json; + + use crate::test_nats::TestNats; + + fn row(icon: &str, summary: &str, body: &str) -> bytes::Bytes { + json!({ + "ts": "2026-10-18T12:00:00Z", "icon": icon, "level": "info", + "summary": summary, "body": body, "body_format": "markdown", + }) + .to_string() + .into() + } + + fn roster() -> Arc> { + Arc::new(["atlas", "iris"].map(String::from).into()) + } + + async fn publish(client: &async_nats::Client, subject: String, payload: bytes::Bytes) { + client.publish(subject, payload).await.expect("publish"); + client.flush().await.expect("flush"); + } + + async fn next(messages: &mut (impl Stream + Unpin)) -> SwarmMessage { + tokio::time::timeout(Duration::from_secs(5), messages.next()) + .await + .expect("a message arrived") + .expect("the feed ended") + } + + async fn nothing_more(messages: &mut (impl Stream + Unpin)) { + if let Ok(Some(m)) = tokio::time::timeout(Duration::from_millis(500), messages.next()).await + { + panic!("an extra message: {m:?}"); + } + } + + /// The last row published is a `send`, so reaching it proves the rows + /// before it were read and dropped rather than not yet delivered. + #[tokio::test] + async fn live_rows_become_messages_and_the_rest_are_dropped() { + let nats = TestNats::start(); + let client = nats.client().await; + let subscriber = client + .subscribe(swarm_queue_client::term_subject_wildcard()) + .await + .expect("subscribe"); + let mut messages = Box::pin(live(subscriber, roster())); + let atlas = swarm_queue_client::term_subject("atlas"); + let iris = swarm_queue_client::term_subject("iris"); + + publish( + &client, + atlas.clone(), + row(SEND_ICON, "send* β†’ iris Β· 3L", "hello"), + ) + .await; + let m = next(&mut messages).await; + assert_eq!( + (m.from.as_str(), m.to.as_str(), m.body.as_str()), + ("atlas", "iris", "hello") + ); + assert_eq!(m.ts, "2026-10-18T12:00:00Z"); + assert_eq!(m.body_format.as_deref(), Some("markdown")); + + // A reminder's turn is a message from `reminder` to the agent. + publish( + &client, + iris.clone(), + row(TURN_ICON, "TURN ← reminder", "stand up"), + ) + .await; + let m = next(&mut messages).await; + assert_eq!( + (m.from.as_str(), m.to.as_str(), m.body.as_str()), + ("reminder", "iris", "stand up") + ); + + // A turn from an agent, a subagent subject, and other rows: no event. + publish( + &client, + iris.clone(), + row(TURN_ICON, "TURN ← atlas", "hello"), + ) + .await; + publish( + &client, + swarm_queue_client::subagent_term::subject("atlas", "scout"), + row(SEND_ICON, "send* β†’ iris", "from a subagent"), + ) + .await; + publish(&client, iris.clone(), row("πŸ”§", "Bash", "ls")).await; + publish(&client, iris, bytes::Bytes::from_static(b"not json")).await; + publish(&client, atlas, row(SEND_ICON, "send* β†’ iris", "last")).await; + assert_eq!(next(&mut messages).await.body, "last"); + nothing_more(&mut messages).await; + } + + #[tokio::test] + async fn history_holds_only_rows_since_the_cutoff_oldest_first() { + let nats = TestNats::start(); + let client = nats.client().await; + for agent in ["atlas", "iris"] { + agent_term::open_or_create(&client, agent) + .await + .expect("the stream"); + } + let stamped = |ts: &str, summary: &str, body: &str| -> bytes::Bytes { + json!({"ts": ts, "icon": SEND_ICON, "level": "info", "summary": summary, "body": body}) + .to_string() + .into() + }; + let secs = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("after the epoch") + .as_secs(); + let now = iso8601_utc(i64::try_from(secs).expect("fits")); + let earlier = iso8601_utc(i64::try_from(secs - 600).expect("fits")); + let atlas = swarm_queue_client::term_subject("atlas"); + let iris = swarm_queue_client::term_subject("iris"); + publish(&client, atlas, stamped(&now, "send* β†’ iris", "newer")).await; + publish( + &client, + iris.clone(), + stamped(&earlier, "send* β†’ atlas", "older"), + ) + .await; + publish( + &client, + iris, + stamped("2000-01-01T00:00:00Z", "send* β†’ atlas", "ancient"), + ) + .await; + let js = async_nats::jetstream::new(client.clone()); + for (agent, want) in [("atlas", 1), ("iris", 2)] { + for _ in 0..100 { + let mut s = js + .get_stream(agent_term::stream_name(agent)) + .await + .expect("the stream"); + if s.info().await.expect("info").state.messages >= want { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + let got = history(&client, &roster(), 60).await; + assert_eq!( + got.iter().map(|m| m.body.as_str()).collect::>(), + ["older", "newer"] + ); + // No window reads no history. + assert!(history(&client, &roster(), 0).await.is_empty()); + } + + #[test] + fn a_row_without_a_party_makes_no_message() { + let none = |_: &str| false; + assert!(message_of("atlas", &row(SEND_ICON, "send* β†’ ", "x"), none).is_none()); + assert!(message_of("atlas", &row(TURN_ICON, "TURN ← ", "x"), none).is_none()); + assert!(message_of("atlas", &row(SEND_ICON, "send* β†’ iris", "x"), none).is_some()); + } + + #[test] + fn only_main_terminal_subjects_name_an_agent() { + assert_eq!(agent_of("$SWARM.term.atlas"), Some("atlas")); + assert_eq!(agent_of("$SWARM.term.atlas.sub.scout"), None); + assert_eq!(agent_of("$SWARM.term."), None); + } +} diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index ec4d5e11..29a0b418 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -31,8 +31,8 @@ use crate::{ agent_control, agent_icon, agent_input, agent_limits, agent_login, agent_permissions, agent_renewal, agent_state_stream, agent_status, agents_md, alerts, archetypes, auth, builder, claude_login, config_pr, forge, forge_account, github_account, inference, issue_report, - linked_accounts, matrix_account, pulls, queue_identity, review_classifier, schedules, status, - subagent_term, term_stream, wanted, webhook, + linked_accounts, matrix_account, message_stream, pulls, queue_identity, review_classifier, + schedules, status, subagent_term, term_stream, wanted, webhook, }; /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. @@ -602,6 +602,7 @@ pub(crate) fn build_app(state: AppState) -> axum::Router { .routes(routes!(linked_accounts::get_linked_accounts)) .routes(routes!(agents::get_hive_wanted)) .routes(routes!(term_stream::stream_agent_term)) + .routes(routes!(message_stream::stream_messages)) .routes(routes!(subagent_term::list_subagents)) .routes(routes!(subagent_term::stream_subagent_term)) .routes(routes!(agent_state_stream::stream_agent_state)) diff --git a/swarm-controller/src/term_stream.rs b/swarm-controller/src/term_stream.rs index 8a40aa9d..193cb3cc 100644 --- a/swarm-controller/src/term_stream.rs +++ b/swarm-controller/src/term_stream.rs @@ -153,10 +153,10 @@ pub(crate) async fn replay_or_live( } /// One row read from a terminal stream. -struct Row { +pub(crate) struct Row { /// Its stream sequence, when the delivery names one. - seq: Option, - payload: bytes::Bytes, + pub(crate) seq: Option, + pub(crate) payload: bytes::Bytes, } /// The rows on `subject` in the stream `stream_name`: every row after @@ -170,7 +170,7 @@ struct Row { /// subject's rows, so a fresh reader's consumer starts at the oldest row and /// the reader drops all but the newest [`agent_term::REPLAY_ROWS`] it counts /// as pending when the consumer is created. -async fn replay( +pub(crate) async fn replay( client: &async_nats::Client, stream_name: &str, subject: String,