swarm-ui: key message rows on an arrival id and follow the tail on it

At the 2000-message cap the count stops changing, so the page stopped
following the newest message, and every row's index key shifted on each
arrival, remounting all of them. Each message now gets an id when it
arrives; the rows key on it and the scroll follows the newest id. Rows
are memoised on the message, and the events of one frame join the list
in a single update.

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.<agent>` 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/<name>/cancel` sends `"cancel"` as a NATS request on
 `$SWARM.control.<name>`, and only swarm-controller may publish there. The

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() {
         <Route path="/schedules" component={SchedulesPage} />
         <Route path="/builds" component={BuildsPage} />
         <Route path="/broadcast" component={BroadcastPage} />
+        <Route path="/messages" component={MessagesPage} />
         <Route path="/issues" component={IssueReportPage} />
         <Route path="/review" component={ReviewPage} />
         <Route path="/components" component={ComponentsPage} />

frontend/packages/swarm-ui/src/pages/messages/MessagesPage.css

@@ -0,0 +1,28 @@
+/* <MessagesPage>: 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);
+}

frontend/packages/swarm-ui/src/pages/messages/MessagesPage.tsx

@@ -0,0 +1,136 @@
+// <MessagesPage> — `/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<ConnectionState, string> = {
+  connecting: "connecting…",
+  open: "live",
+  reconnecting: "reconnecting…",
+  closed: "disconnected",
+};
+
+const CONNECTION_TONE: Record<ConnectionState, BadgeTone> = {
+  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 <Row row={messageRow(m, String(m.id))} />;
+});
+
+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<HTMLDivElement>(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 (
+    <Panel
+      title="messages"
+      icon="💬"
+      actions={
+        <Badge
+          tone={CONNECTION_TONE[connection]}
+          value={CONNECTION_LABEL[connection]}
+        />
+      }
+    >
+      <div class="ui-messages-bar">
+        <SelectField
+          id="messages-window"
+          label="history"
+          value={String(sinceMinutes)}
+          onChange={(v) => setSinceMinutes(Number(v))}
+          options={WINDOWS.map((w) => ({
+            value: String(w.minutes),
+            label: w.label,
+          }))}
+        />
+        <TextField
+          id="messages-filter"
+          label="agent"
+          value={query}
+          placeholder="sender or recipient"
+          onInput={setQuery}
+        />
+        <span class="ui-messages-count">
+          {query.trim() === ""
+            ? `${messages.length} messages`
+            : `${shown.length} of ${messages.length} messages`}
+        </span>
+      </div>
+      <div class="terminal-wrap ui-messages-wrap">
+        <div class="live terminal" ref={logRef} onScroll={handleScroll}>
+          {shown.length === 0 ? (
+            <div class="row note">
+              <span class="row-content">
+                {connection === "open"
+                  ? messages.length === 0
+                    ? "(connected — no messages yet)"
+                    : "(no message matches the filter)"
+                  : CONNECTION_LABEL[connection]}
+              </span>
+            </div>
+          ) : (
+            shown.map((m) => <MessageRow key={m.id} m={m} />)
+          )}
+        </div>
+      </div>
+      <p class="ui-messages-note">
+        Shows what the agents sent each other. A message an agent drained with
+        an in-turn <code>recv</code> is missing. The history is at most 100
+        recent messages per agent.
+      </p>
+    </Panel>
+  );
+}

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<T extends SwarmMessage>(
+  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<T extends SwarmMessage>(
+  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,
+  };
+}

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<ArrivedMessage[]>([]);
+  const [connection, setConnection] = useState<ConnectionState>("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 };
+}

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)" },

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, / — /);
+});

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<TermMsg> {
         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);
             }

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")
+        );
+    }
 }

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 ` · <N>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

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;

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<String>,
+}

swarm-controller/src/agent_input.rs

@@ -205,7 +205,7 @@ fn queue(state: &AppState) -> Result<async_nats::Client, String> {
 }
 
 /// The swarm roster, or the 503 `GET /api/agents` answers when it has none.
-async fn roster(state: &AppState) -> Result<Vec<String>, ProblemDetails> {
+pub(crate) async fn roster(state: &AppState) -> Result<Vec<String>, ProblemDetails> {
     let Some(bridge) = state.auth.as_ref() else {
         return Err(error_problem(
             StatusCode::SERVICE_UNAVAILABLE,

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;

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<u64>,
+}
+
+#[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<AppState>,
+    query: Result<Query<MessagesQuery>, QueryRejection>,
+) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, 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<HashSet<String>> = Arc::new(roster.into_iter().collect());
+    let history = history(&client, &roster, minutes).await;
+    let seen: HashSet<SwarmMessage> = 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<HashSet<String>>,
+    minutes: u64,
+) -> Vec<SwarmMessage> {
+    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<SwarmMessage> = stream::iter(roster.iter().cloned().collect::<Vec<_>>())
+        .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<HashSet<String>>,
+) -> Vec<SwarmMessage> {
+    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.<agent>` subjects.
+fn live(
+    subscriber: async_nats::Subscriber,
+    roster: Arc<HashSet<String>>,
+) -> impl Stream<Item = SwarmMessage> {
+    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<String>,
+    #[serde(default)]
+    summary: String,
+    body: Option<String>,
+    body_format: Option<String>,
+}
+
+/// 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<SwarmMessage> {
+    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<HashSet<String>> {
+        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<Item = SwarmMessage> + 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<Item = SwarmMessage> + 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::<Vec<_>>(),
+            ["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);
+    }
+}

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))

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<u64>,
-    payload: bytes::Bytes,
+    pub(crate) seq: Option<u64>,
+    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,