diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 4d43f5d9..bce82c8d 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -18,7 +18,6 @@ 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 | @@ -234,28 +233,6 @@ 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 381f127d..9002d98b 100644 --- a/frontend/packages/swarm-ui/src/App.tsx +++ b/frontend/packages/swarm-ui/src/App.tsx @@ -7,7 +7,6 @@ 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"; @@ -42,7 +41,6 @@ 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 deleted file mode 100644 index 6fe1d758..00000000 --- a/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.css +++ /dev/null @@ -1,28 +0,0 @@ -/* : 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 deleted file mode 100644 index 906a1296..00000000 --- a/frontend/packages/swarm-ui/src/pages/messages/MessagesPage.tsx +++ /dev/null @@ -1,136 +0,0 @@ -// β€” `/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 deleted file mode 100644 index ebbc8659..00000000 --- a/frontend/packages/swarm-ui/src/pages/messages/messages.ts +++ /dev/null @@ -1,108 +0,0 @@ -// 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 deleted file mode 100644 index 663935ed..00000000 --- a/frontend/packages/swarm-ui/src/pages/messages/useMessageStream.ts +++ /dev/null @@ -1,76 +0,0 @@ -// 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 6d23763b..d79b771d 100644 --- a/frontend/packages/swarm-ui/src/shell/Shell.tsx +++ b/frontend/packages/swarm-ui/src/shell/Shell.tsx @@ -55,7 +55,6 @@ 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 deleted file mode 100644 index 56771ec3..00000000 --- a/frontend/packages/swarm-ui/test/messages.test.mjs +++ /dev/null @@ -1,128 +0,0 @@ -// 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 55553c95..caeddcc8 100644 --- a/hive-agent/src/term_msg.rs +++ b/hive-agent/src/term_msg.rs @@ -11,7 +11,6 @@ //! 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; @@ -34,8 +33,7 @@ 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_SUMMARY_PREFIX}{from}")).icon(TURN_ICON); + let mut m = TermMsg::new(Level::Info, format!("TURN ← {from}")).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 67ac545d..0cb709e3 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, SEND_ICON, SEND_SUMMARY_PREFIX, TermMsg}; +use crate::term_msg::{BodyFormat, ClassifyCtx, Level, 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" => SEND_ICON, + "mcp__hyperhive__send" => "πŸ“€", "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!("{SEND_SUMMARY_PREFIX}{to} Β· {lines}L") + format!("{short} β†’ {to} Β· {lines}L") } else { - format!("{SEND_SUMMARY_PREFIX}{to}") + format!("{short} β†’ {to}") } } _ => fmt_args_generic(short, input), @@ -1331,29 +1331,4 @@ 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 429e45be..d896934c 100644 --- a/hive-sh4re/src/term_msg.rs +++ b/hive-sh4re/src/term_msg.rs @@ -128,41 +128,6 @@ 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 5c515656..fc9c47d1 100644 --- a/swarm-controller-api/src/lib.rs +++ b/swarm-controller-api/src/lib.rs @@ -7,5 +7,4 @@ 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 deleted file mode 100644 index 56661b37..00000000 --- a/swarm-controller-api/src/messages.rs +++ /dev/null @@ -1,20 +0,0 @@ -//! `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 b371c428..aeaf8f7c 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. -pub(crate) async fn roster(state: &AppState) -> Result, ProblemDetails> { +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 545bf774..1e927f79 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -48,7 +48,6 @@ 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 deleted file mode 100644 index c130686b..00000000 --- a/swarm-controller/src/message_stream.rs +++ /dev/null @@ -1,411 +0,0 @@ -//! `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 29a0b418..ec4d5e11 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, message_stream, pulls, queue_identity, review_classifier, - schedules, status, subagent_term, term_stream, wanted, webhook, + linked_accounts, matrix_account, pulls, queue_identity, review_classifier, schedules, status, + subagent_term, term_stream, wanted, webhook, }; /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. @@ -602,7 +602,6 @@ 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 193cb3cc..8a40aa9d 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. -pub(crate) struct Row { +struct Row { /// Its stream sequence, when the delivery names one. - pub(crate) seq: Option, - pub(crate) payload: bytes::Bytes, + seq: Option, + payload: bytes::Bytes, } /// The rows on `subject` in the stream `stream_name`: every row after @@ -170,7 +170,7 @@ pub(crate) 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. -pub(crate) async fn replay( +async fn replay( client: &async_nats::Client, stream_name: &str, subject: String,