diff --git a/TODO.md b/TODO.md index 257282a0..3fa31eb4 100644 --- a/TODO.md +++ b/TODO.md @@ -29,8 +29,10 @@ - Per-agent reminder status (pending, delivered) - Reminder query interface for debugging - Display reminder delivery errors (failed sends, mark failures) +- **Phase 5b: per-domain mutation event types + client derived state.** Foundation already in place (`DashboardEvent` channel on Coordinator, broker→dashboard forwarder, `/dashboard/{stream,history}`, snapshot+SSE seq dedupe). Remaining work: add `ApprovalAdded` / `ApprovalResolved`, `QuestionAdded` / `QuestionAnswered`, `TransientChanged` variants to `DashboardEvent`; emit each at the corresponding mutation site (`actions::approve`/`deny`/`finish_approval`, `approvals.submit_kind`, `OperatorQuestions::{submit,answer,cancel}`, `Coordinator::{set_transient,clear_transient}`); have the client maintain derived `approvals` / `questions` / `transients` arrays applied from events and drop those fields from `/api/state`. Unblocks dropping the redirect-and-refetch on every remaining action endpoint (`/approve`, `/deny`, `/restart`, `/destroy`, `/kill`, `/rebuild`, `/api/cancel`, `/api/compact`, `/api/model`, `/api/new-session`, `/request-spawn`, `/answer-question`, `/cancel-question`, `/meta-update`, `/purge-tombstone`). Container-list events deferred until `ContainerView` becomes event-derivable (currently sourced from external `nixos-container list`). ## Bugs - ~~**Pending message wake-up**~~ ✓ fixed (e423d57) — subscribe-before-check race in `broker.recv_blocking` meant a send landing between the initial `recv()` and `subscribe()` was missed; agent then sat on the 180s long-poll until another, unrelated message woke it. Now subscribe first. - **Post-rebuild system-message missed wake**: at 09:13:14 the dashboard showed `system → damocles container rebuilt` as ✓ delivered, but the agent harness never ran a turn for it (no claude invocation, no operator-visible activity). A subsequent `recv()` from inside the agent returned `(empty)`, confirming the message was popped + marked delivered server-side — yet drove no turn. Most likely cause: the agent_server `serve_agent_stdio` task is up and answering MCP/socket calls, but the `hive-ag3nt::serve` long-poll loop that drives `drive_turn` either died silently during rebuild or never restarted. Investigate: (a) does hive-ag3nt's serve loop survive `nixos-container update` cleanly, or does its tokio runtime get torn down mid-loop? (b) is there an early-exit path on a transient socket error during rebuild that drops the serve task without notifying the manager? (c) compare timeline with manager's own post-rebuild wake to see if this is rebuilt-agents-only or universal. Could be related to the `recv_blocking` fix in `e423d57` if the rebuild restarts the broker mid-subscribe. +- ~~**`LiveEvent::Note(String)` never reaches the browser**~~ ✓ fixed — converted to struct variant `Note { text: String }`; wire shape `{"kind":"note","text":"..."}` matches what the JS already reads via `ev.text`. Historical sqlite rows persisted as the literal string `"null"` (from when serialization silently failed) get filtered out by the `rows.flatten().flatten()` pipeline in `EventStore::recent`, so replay tolerates them. diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs index d4bdd72f..4e88813e 100644 --- a/hive-ag3nt/src/bin/hive-m1nd.rs +++ b/hive-ag3nt/src/bin/hive-m1nd.rs @@ -136,7 +136,9 @@ async fn serve( } else { tracing::info!(%from, %body, "system message"); } - bus.emit(LiveEvent::Note(format!("[system] {body}"))); + bus.emit(LiveEvent::Note { + text: format!("[system] {body}"), + }); // Fall through: drive a turn with the event in the wake // prompt body so claude sees it. Sender stays "system" // so the wake prompt can label it as such. diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 0588c820..8af3e049 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -9,7 +9,7 @@ //! showing "connecting…" until the first event arrives. use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use rusqlite::{Connection, params}; @@ -74,6 +74,18 @@ CREATE TABLE IF NOT EXISTS events ( CREATE INDEX IF NOT EXISTS idx_events_ts ON events (ts); "; +/// Envelope carried over the broadcast channel: the `LiveEvent` itself +/// plus a monotonic per-process seq stamped by `Bus::emit`. SSE consumers +/// serialize this directly (seq becomes a sibling of the `kind` tag); +/// clients use seq to dedupe their buffered live traffic against the +/// snapshot/history responses (drop anything with `seq <= snapshot.seq`). +#[derive(Debug, Clone, Serialize)] +pub struct BusEvent { + pub seq: u64, + #[serde(flatten)] + pub event: LiveEvent, +} + /// One row of the agent's live stream. Serialised to JSON for SSE delivery. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] @@ -93,7 +105,16 @@ pub enum LiveEvent { /// Free-form note from the harness (e.g. "claude exited 0", /// "stream-json parse error: ..."). Useful when stream-json itself /// fails so the UI doesn't just go silent. - Note(String), + /// + /// Must be a struct variant (not `Note(String)`): internally-tagged + /// enums can't flatten a tag onto a primitive newtype, and serde + /// fails serialization at runtime — silently, because the SSE + /// handler's `filter_map(... .ok()? ...)` swallows the error. From + /// 2025-08 through 2026-05 every `Note` emission was a no-op + the + /// sqlite history persisted them as the literal string `"null"`. + /// The web UI's `note` renderer already reads `ev.text`, so the + /// wire shape matches without a JS change. + Note { text: String }, /// Turn finished. `ok=false` means claude exited non-zero or the /// harness hit a transport error. TurnEnd { ok: bool, note: Option }, @@ -126,7 +147,7 @@ impl EventStore { let kind = match event { LiveEvent::TurnStart { .. } => "turn_start", LiveEvent::Stream(_) => "stream", - LiveEvent::Note(_) => "note", + LiveEvent::Note { .. } => "note", LiveEvent::TurnEnd { .. } => "turn_end", }; let payload = serde_json::to_string(event).unwrap_or_else(|_| "null".into()); @@ -216,7 +237,13 @@ pub const DEFAULT_MODEL: &str = "haiku"; #[derive(Clone)] pub struct Bus { - tx: Arc>, + tx: Arc>, + /// Monotonic per-process counter stamped onto every `BusEvent`. + /// Persisted nowhere — a harness restart resets seq to 0; clients + /// always treat reconnect as "fresh state, fresh stream of seqs." + /// Historical events served from sqlite carry no seq (they predate + /// the live channel the seq is meant to dedupe against). + event_seq: Arc, /// Persistent event log. `None` only if opening the sqlite db failed /// at construction — we keep going so the harness doesn't die on a /// missing state dir mount in dev / test scenarios. @@ -258,6 +285,7 @@ impl Bus { let initial_model = load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()); Self { tx: Arc::new(tx), + event_seq: Arc::new(AtomicU64::new(0)), store, state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))), model: Arc::new(Mutex::new(initial_model)), @@ -266,6 +294,20 @@ impl Bus { } } + /// Current high-water seq. Snapshot endpoints read this before + /// gathering state so the resulting (snapshot.seq, snapshot) pair + /// satisfies: any live event with seq > snapshot.seq is post-snapshot + /// (not yet reflected). Clients dedupe buffered SSE traffic against + /// this value. + #[must_use] + pub fn current_seq(&self) -> u64 { + self.event_seq.load(Ordering::SeqCst) + } + + fn next_seq(&self) -> u64 { + self.event_seq.fetch_add(1, Ordering::SeqCst) + 1 + } + /// Arm the one-shot: the next claude invocation will run without /// `--continue`, dropping any prior session context. Idempotent /// — calling twice in a row before the next turn still consumes @@ -333,11 +375,15 @@ impl Bus { { tracing::warn!(error = ?e, "events: append failed"); } + let envelope = BusEvent { + seq: self.next_seq(), + event, + }; // Lagged subscribers drop events — fine; the UI is a tail, not a log. - let _ = self.tx.send(event); + let _ = self.tx.send(envelope); } - pub fn subscribe(&self) -> broadcast::Receiver { + pub fn subscribe(&self) -> broadcast::Receiver { self.tx.subscribe() } diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 19923179..a07dc75b 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -206,11 +206,13 @@ pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome /// compact state matches a normal turn's. Only the prompt over stdin /// differs (`/compact` vs the wake-up payload). pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> Result<()> { - bus.emit(LiveEvent::Note( - "context overflow — running /compact on the persistent session".into(), - )); + bus.emit(LiveEvent::Note { + text: "context overflow — running /compact on the persistent session".into(), + }); let _ = run_claude("/compact", files, bus).await?; - bus.emit(LiveEvent::Note("/compact done".into())); + bus.emit(LiveEvent::Note { + text: "/compact done".into(), + }); Ok(()) } @@ -218,9 +220,9 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result let model = bus.model(); let resume = !bus.take_skip_continue(); if !resume { - bus.emit(LiveEvent::Note( - "fresh session (--continue suppressed for this turn)".into(), - )); + bus.emit(LiveEvent::Note { + text: "fresh session (--continue suppressed for this turn)".into(), + }); } let mut cmd = Command::new("claude"); // Spawn inside the agent's state dir so relative paths in tool calls @@ -282,7 +284,9 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result } bus_out.emit(LiveEvent::Stream(v)); } - Err(_) => bus_out.emit(LiveEvent::Note(format!("(non-json) {line}"))), + Err(_) => bus_out.emit(LiveEvent::Note { + text: format!("(non-json) {line}"), + }), } } }); @@ -304,7 +308,9 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result // renders; the tracing line is what `journalctl -M -b` // surfaces when claude exits non-zero. tracing::warn!(line = %line, "claude stderr"); - bus_err.emit(LiveEvent::Note(format!("stderr: {line}"))); + bus_err.emit(LiveEvent::Note { + text: format!("stderr: {line}"), + }); let mut t = tail_clone.lock().unwrap(); if t.len() >= STDERR_TAIL_LINES { t.pop_front(); diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 5eb13fce..44243abe 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -191,6 +191,12 @@ async fn serve_shared_js() -> impl IntoResponse { #[derive(Serialize)] struct StateSnapshot { + /// Bus seq at the moment this snapshot was assembled. Clients dedupe + /// their buffered SSE traffic against this value: events with + /// `seq <= snapshot.seq` are already reflected (or pre-date the + /// snapshot); `seq > snapshot.seq` is post-snapshot. Reset to 0 on + /// harness restart — clients treat reconnect as a fresh world. + seq: u64, label: String, dashboard_port: u16, /// `"online"` | `"needs_login_idle"` | `"needs_login_in_progress"`. @@ -226,6 +232,9 @@ struct SessionView { } async fn api_state(State(state): State) -> axum::Json { + // Capture seq *before* any reads so the dedupe contract is + // "events with seq > snapshot.seq are post-snapshot, never missed." + let seq = state.bus.current_seq(); drop_if_finished(&state.session); let login = *state.login.lock().unwrap(); let session_snapshot = state.session.lock().unwrap().clone(); @@ -251,6 +260,7 @@ async fn api_state(State(state): State) -> axum::Json { let model = state.bus.model(); let token_usage = state.bus.last_usage(); axum::Json(StateSnapshot { + seq, label: state.label.clone(), dashboard_port, status, @@ -333,15 +343,26 @@ async fn post_send(State(state): State, Form(form): Form) -> }, }; match result { - Ok(()) => Redirect::to("/").into_response(), + // 200 instead of 303 → the client doesn't refetch /api/state. + // The operator message becomes a broker `Sent` (already shown + // server-side in the dashboard); on the agent side, the + // resulting `TurnStart` SSE event drives the terminal + the + // inbox row gets consumed by the time `TurnEnd` fires the + // existing turn-end refresh. + Ok(()) => (axum::http::StatusCode::OK, "ok").into_response(), Err(e) => error_response(&format!("send failed: {e}")), } } -async fn events_history( - State(state): State, -) -> axum::Json> { - axum::Json(state.bus.history()) +async fn events_history(State(state): State) -> axum::Json { + // Capture seq *before* the read so dedupe is "drop buffered events + // you've already seen in history", never "lose an event that fired + // between the read and the timestamp." Historical rows have no + // per-row seq; only the high-water mark matters for the dedupe + // window. + let seq = state.bus.current_seq(); + let events = state.bus.history(); + axum::Json(serde_json::json!({ "seq": seq, "events": events })) } async fn events_stream( @@ -351,9 +372,9 @@ async fn events_stream( let rx = state.bus.subscribe(); // Drop a "hello" note into the bus so every new subscriber sees at // least one event immediately and can clear the connecting placeholder. - state.bus.emit(crate::events::LiveEvent::Note( - "live stream attached".into(), - )); + state.bus.emit(crate::events::LiveEvent::Note { + text: "live stream attached".into(), + }); let stream = BroadcastStream::new(rx).filter_map(|res| { let ev = res.ok()?; let json = serde_json::to_string(&ev).ok()?; @@ -427,9 +448,9 @@ async fn post_set_model(State(state): State, Form(form): Form) -> Response { let files = state.files.clone(); tokio::spawn(async move { let _guard = guard; // keep lock alive for the duration of compaction - bus.emit(crate::events::LiveEvent::Note( - "operator: /compact — running on persistent session".into(), - )); + bus.emit(crate::events::LiveEvent::Note { + text: "operator: /compact — running on persistent session".into(), + }); bus.set_state(crate::events::TurnState::Compacting); let r = crate::turn::compact_session(&files, &bus).await; bus.set_state(crate::events::TurnState::Idle); if let Err(e) = r { - bus.emit(crate::events::LiveEvent::Note(format!( - "/compact failed: {e:#}" - ))); + bus.emit(crate::events::LiveEvent::Note { + text: format!("/compact failed: {e:#}"), + }); } }); Redirect::to("/").into_response() @@ -480,9 +501,9 @@ async fn post_compact(State(state): State) -> Response { /// than asking claude to forget mid-stream. async fn post_new_session(State(state): State) -> Response { state.bus.request_new_session(); - state.bus.emit(crate::events::LiveEvent::Note( - "operator: new session armed — next turn runs without --continue".into(), - )); + state.bus.emit(crate::events::LiveEvent::Note { + text: "operator: new session armed — next turn runs without --continue".into(), + }); Redirect::to("/").into_response() } @@ -503,7 +524,7 @@ async fn post_cancel_turn(State(state): State) -> Response { ), Err(e) => format!("operator: /cancel — pkill failed: {e}"), }; - state.bus.emit(crate::events::LiveEvent::Note(note)); + state.bus.emit(crate::events::LiveEvent::Note { text: note }); Redirect::to("/").into_response() } diff --git a/hive-c0re/assets/app.js b/hive-c0re/assets/app.js index 600d6729..06f5e108 100644 --- a/hive-c0re/assets/app.js +++ b/hive-c0re/assets/app.js @@ -1,6 +1,6 @@ // Dashboard SPA. Renders containers + approvals from `/api/state`, wires // up async-form submission (URL-encoded POST + spinner + state refresh), -// and tails the broker over `/messages/stream` SSE. +// and tails the unified dashboard event channel over `/dashboard/stream`. (() => { // ─── helpers ──────────────────────────────────────────────────────────── @@ -118,20 +118,20 @@ // when the page reloads. const seenApprovals = new Set(); const seenQuestions = new Set(); - const seenInboxIds = new Set(); let seededNotify = false; function notifyDeltas(s) { const approvals = s.approvals || []; const questions = s.questions || []; - const inbox = s.operator_inbox || []; if (!seededNotify) { // First render after page load — fill the "seen" sets without // firing notifications. We only want to notify on NEW items - // that arrived while the page is open. + // that arrived while the page is open. The inbox no longer + // needs seeding here: it's derived from the broker stream which + // does its own per-event notification on live arrival, and + // history-replayed events are silent by virtue of `fromHistory`. for (const a of approvals) seenApprovals.add(a.id); for (const q of questions) seenQuestions.add(q.id); - for (const m of inbox) seenInboxIds.add(m.id); seededNotify = true; return; } @@ -148,14 +148,6 @@ NOTIF.show('◆ manager asks', q.question.slice(0, 120), 'hyperhive:question:' + q.id); } - // operator_inbox: only notify on truly new ids — sse already - // handles single-message notifications, but if the operator - // missed an SSE event (page reloaded), this catches up. - for (const m of inbox) { - if (seenInboxIds.has(m.id)) continue; - seenInboxIds.add(m.id); - // suppress here; SSE path handles the live notification. - } } // ─── async forms ──────────────────────────────────────────────────────── @@ -605,16 +597,30 @@ } } - function renderInbox(s) { + // ─── operator inbox (derived from the broker message stream) ─────────── + // No longer shipped on `/api/state.operator_inbox`. The dashboard + // terminal's HiveTerminal feeds this via `onAnyEvent` — backfill from + // `/dashboard/history` populates on load, live SSE keeps it current. + // Newest-first to match the previous behaviour. + const INBOX_LIMIT = 50; + const operatorInbox = []; + function inboxAppendFromEvent(ev) { + if (ev.kind !== 'sent' || ev.to !== 'operator') return false; + operatorInbox.unshift({ from: ev.from, body: ev.body, at: ev.at }); + if (operatorInbox.length > INBOX_LIMIT) operatorInbox.length = INBOX_LIMIT; + return true; + } + function renderInbox() { const root = $('inbox-section'); + if (!root) return; root.innerHTML = ''; - if (!s.operator_inbox || !s.operator_inbox.length) { + if (!operatorInbox.length) { root.append(el('p', { class: 'empty' }, 'no messages')); return; } const fmt = (n) => new Date(n * 1000).toISOString().replace('T', ' ').slice(0, 19); const ul = el('ul', { class: 'inbox' }); - for (const m of s.operator_inbox) { + for (const m of operatorInbox) { const li = el('li'); li.append( el('span', { class: 'msg-ts' }, fmt(m.at)), ' ', @@ -730,14 +736,29 @@ denyForm, ); li.append(row); - if (a.diff_html) { + if (a.diff) { const details = el('details', { 'data-restore-key': 'approval-diff:' + a.id, }); details.append(el('summary', {}, 'diff vs applied')); - // diff_html is pre-rendered server-side (per-line class spans inside - // a 
); inject as innerHTML.
-        const pre = el('pre', { class: 'diff', html: a.diff_html });
+        // Server ships the raw unified diff; classify each line by its
+        // leading char so `.diff-add` / `.diff-del` / `.diff-hunk` /
+        // `.diff-file` / `.diff-ctx` colour the output. Building spans
+        // here (instead of innerHTML-ing pre-rendered markup) keeps
+        // the snapshot wire format text-only and one less HTML-escape
+        // surface server-side.
+        const pre = el('pre', { class: 'diff' });
+        for (const raw of a.diff.split('\n')) {
+          let cls = 'diff-ctx';
+          if (raw.startsWith('--- ') || raw.startsWith('+++ ')) cls = 'diff-file';
+          else if (raw.startsWith('@')) cls = 'diff-hunk';
+          else if (raw.startsWith('+'))  cls = 'diff-add';
+          else if (raw.startsWith('-'))  cls = 'diff-del';
+          const span = document.createElement('span');
+          span.className = cls;
+          span.textContent = raw + '\n';
+          pre.appendChild(span);
+        }
         details.append(pre);
         li.append(details);
       }
@@ -932,7 +953,7 @@
       renderContainers(s);
       renderTombstones(s);
       renderQuestions(s);
-      renderInbox(s);
+      renderInbox();
       renderApprovals(s);
       renderMetaInputs(s);
       restoreOpenDetails(openDetails);
@@ -955,17 +976,19 @@
   refreshState();
   NOTIF.bind();
 
-  // ─── message flow SSE ───────────────────────────────────────────────────
+  // ─── message flow: shared terminal pane ────────────────────────────────
+  // Scroll, pill, backfill + SSE plumbing live in hive-fr0nt::TERMINAL_JS
+  // (window.HiveTerminal). What stays here is the broker-message
+  // renderer + the page-local side effects (banner pulse, inbox refresh
+  // on operator-bound traffic, OS notifications).
   (() => {
     const flow = $('msgflow');
-    if (!flow) return;
+    if (!flow || !window.HiveTerminal) return;
     flow.innerHTML = '';
-    const es = new EventSource('/messages/stream');
-    const MAX_ROWS = 200;
     const tsFmt = (n) => new Date(n * 1000).toISOString().slice(11, 19);
-    // Animate the banner whenever a broker event lands. Each event nudges
-    // the shimmer window; if traffic stops, the class falls off after the
-    // grace timer.
+    // Pulse the page banner whenever a broker event lands. Each event
+    // nudges the shimmer window; if traffic stops, the class falls off
+    // after the grace timer.
     const banner = document.querySelector('.banner');
     let bannerOffTimer = null;
     function pulseBanner() {
@@ -974,40 +997,45 @@
       if (bannerOffTimer) clearTimeout(bannerOffTimer);
       bannerOffTimer = setTimeout(() => banner.classList.remove('active'), 4000);
     }
-    es.onmessage = (e) => {
-      let m;
-      try { m = JSON.parse(e.data); } catch { return; }
-      pulseBanner();
-      // Live-update the inbox when claude sends to operator + ping
-      // the OS notification center.
-      if (m.kind === 'sent' && m.to === 'operator') {
-        refreshState();
-        NOTIF.show(
-          '◆ ' + m.from + ' → operator',
-          String(m.body || '').slice(0, 200),
-          // Unique-per-arrival tag so a burst stacks instead of
-          // overwriting itself in the OS notification center.
-          'hyperhive:msg:' + m.at + ':' + Math.random().toString(36).slice(2, 6),
-        );
-      }
-      const row = document.createElement('div');
-      row.className = 'msgrow ' + m.kind;
-      const kind = m.kind === 'sent' ? '→' : '✓';
-      row.innerHTML =
-        '' + tsFmt(m.at) + '' +
-        '' + kind + '' +
-        '' + esc(m.from) + '' +
+    function renderMsg(ev, api, glyph) {
+      const el = api.row('msgrow ' + ev.kind, '');
+      el.innerHTML =
+        '' + tsFmt(ev.at) + '' +
+        '' + glyph + '' +
+        '' + esc(ev.from) + '' +
         '→' +
-        '' + esc(m.to) + '' +
-        '' + esc(m.body) + '';
-      flow.insertBefore(row, flow.firstChild);
-      while (flow.childNodes.length > MAX_ROWS) flow.removeChild(flow.lastChild);
-    };
-    es.onerror = () => {
-      flow.insertBefore(Object.assign(document.createElement('div'), {
-        className: 'msgrow meta', textContent: '[connection lost — retrying]',
-      }), flow.firstChild);
-    };
+        '' + esc(ev.to) + '' +
+        '' + esc(ev.body) + '';
+    }
+    HiveTerminal.create({
+      logEl: flow,
+      historyUrl: '/dashboard/history',
+      streamUrl: '/dashboard/stream',
+      renderers: {
+        sent:      (ev, api) => renderMsg(ev, api, '→'),
+        delivered: (ev, api) => renderMsg(ev, api, '✓'),
+      },
+      // Both history backfill and live frames flow through here, so the
+      // inbox section ends up populated correctly on first paint and
+      // updated thereafter — no /api/state refetch needed for inbox
+      // freshness (which used to be the workaround for the
+      // double-render bug).
+      onAnyEvent: (ev /* , { fromHistory } */) => {
+        if (inboxAppendFromEvent(ev)) renderInbox();
+      },
+      onLiveEvent: (ev) => {
+        pulseBanner();
+        if (ev.kind === 'sent' && ev.to === 'operator') {
+          NOTIF.show(
+            '◆ ' + ev.from + ' → operator',
+            String(ev.body || '').slice(0, 200),
+            // Unique-per-arrival tag so a burst stacks instead of
+            // overwriting itself in the OS notification center.
+            'hyperhive:msg:' + ev.at + ':' + Math.random().toString(36).slice(2, 6),
+          );
+        }
+      },
+    });
   })();
 
   // ─── compose: @-mention with sticky recipient ───────────────────────────
@@ -1115,14 +1143,15 @@
       fd.append('body', body);
       input.disabled = true;
       try {
+        // /op-send now returns 200 (no more 303-to-/). The SSE channel
+        // carries the resulting MessageEvent → the terminal renders the
+        // sent row + the inbox updates on its own; no /api/state
+        // refetch needed.
         const resp = await fetch('/op-send', {
           method: 'POST',
           body: new URLSearchParams(fd),
-          redirect: 'manual',
         });
-        const ok = resp.ok || resp.type === 'opaqueredirect'
-          || (resp.status >= 200 && resp.status < 400);
-        if (!ok) {
+        if (!resp.ok) {
           flashError(`send failed: http ${resp.status}`);
           return;
         }
diff --git a/hive-c0re/assets/dashboard.css b/hive-c0re/assets/dashboard.css
index dd893997..beba286f 100644
--- a/hive-c0re/assets/dashboard.css
+++ b/hive-c0re/assets/dashboard.css
@@ -537,43 +537,28 @@ summary:hover { color: var(--purple); }
 .inbox .msg-from { color: var(--amber); }
 .inbox .msg-sep  { color: var(--muted); }
 .inbox .msg-body { color: var(--fg); white-space: pre-wrap; word-break: break-word; }
-.msgflow {
-  background: rgba(24, 24, 37, 0.78);
-  -webkit-backdrop-filter: blur(8px) saturate(120%);
-  backdrop-filter: blur(8px) saturate(120%);
-  border: 1px solid var(--border);
-  padding: 0.8em;
-  font-size: 0.85em;
-  line-height: 1.5;
-  max-height: 32em;
-  overflow-y: auto;
-}
-.msgflow .msgrow {
-  animation: row-fade-in 220ms ease-out both;
-}
-@keyframes row-fade-in {
-  from { opacity: 0; transform: translateY(4px); }
-  to   { opacity: 1; transform: translateY(0); }
-}
-.msgrow { display: grid; grid-template-columns: auto auto auto auto auto 1fr; gap: 0.6em; align-items: baseline; padding: 0.1em 0; }
-.msgrow.sent .msg-arrow { color: var(--cyan); }
-.msgrow.delivered .msg-arrow { color: var(--green); }
+/* `#msgflow` is a shared `.live` pane inside `.terminal-wrap` (see
+   hive-fr0nt::TERMINAL_CSS). The msgrow / msg-* rules below are
+   dashboard-specific: each broker event becomes a grid of timestamp +
+   arrow + from/sep/to + body inside the `.row` shell. */
+.live .msgrow { display: grid; grid-template-columns: auto auto auto auto auto 1fr; gap: 0.6em; align-items: baseline; padding: 0.1em 0; }
+.live .msgrow.sent .msg-arrow { color: var(--cyan); }
+.live .msgrow.delivered .msg-arrow { color: var(--green); }
 .msg-ts { color: var(--muted); font-size: 0.85em; }
 .msg-arrow { font-weight: bold; }
 .msg-from { color: var(--amber); }
 .msg-sep { color: var(--muted); }
 .msg-to { color: var(--pink); }
 .msg-body { color: var(--fg); white-space: pre-wrap; word-break: break-word; }
+/* Compose box sits inside `.terminal-wrap`, below the `.live` log. The
+   dashed separator mirrors the agent terminal's prompt divider. */
 .op-compose {
   position: relative;
   display: flex;
   align-items: flex-start;
   gap: 0.6em;
-  margin-top: 0.4em;
   padding: 0.55em 0.8em;
-  background: rgba(24, 24, 37, 0.85);
-  border: 1px solid var(--border);
-  border-top: none;
+  border-top: 1px dashed var(--purple-dim);
 }
 .op-compose-prompt {
   color: var(--purple);
diff --git a/hive-c0re/assets/index.html b/hive-c0re/assets/index.html
index 6af8ef76..257ecb91 100644
--- a/hive-c0re/assets/index.html
+++ b/hive-c0re/assets/index.html
@@ -61,13 +61,15 @@
   
◆ MESS4GE FL0W ◆

   
══════════════════════════════════════════════════════════════

   
live tail — newest at the top. tap on every send / recv through the broker. compose below: @name picks the recipient (sticky until you @ someone else); tab completes.

-  
connecting…

-  

-    @—>
-    
-    
+  

+    
connecting…

+    

+      @—>
+      
+      
+    

   

 
   

@@ -75,6 +77,7 @@
     
▲△▲ hyperhive ▲△▲ hive-c0re on this host ▲△▲

   

 
+  
   
 
 
diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs
index b9d6a20b..c25d2c1b 100644
--- a/hive-c0re/src/broker.rs
+++ b/hive-c0re/src/broker.rs
@@ -46,6 +46,11 @@ const EVENT_CHANNEL: usize = 256;
 /// self-documenting.
 pub type DueReminder = (String, i64, String, Option);
 
+/// Intra-process broker event. `recv_blocking` listens on the same
+/// channel as the dashboard forwarder; the forwarder re-emits each
+/// event as a `DashboardEvent` with a freshly-stamped seq from the
+/// Coordinator. The broker itself doesn't stamp seqs — that's a wire
+/// concern, not a storage concern.
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "snake_case", tag = "kind")]
 pub enum MessageEvent {
@@ -129,6 +134,36 @@ impl Broker {
             .map_err(Into::into)
     }
 
+    /// Latest `limit` messages across every recipient, newest-first.
+    /// Backs the dashboard's message-flow backfill so a reload doesn't
+    /// blank the operator's view of recent traffic. Returns each row as
+    /// a [`MessageEvent::Sent`] so the dashboard's live renderer (which
+    /// already speaks `MessageEvent`) can replay history through the
+    /// same code path. We don't synthesise `Delivered` events here —
+    /// the recv-side acks live in a different table column and would
+    /// double-render on backfill; the live stream picks them up
+    /// immediately on the first new `recv`.
+    pub fn recent_all(&self, limit: u64) -> Result> {
+        let conn = self.conn.lock().unwrap();
+        let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX);
+        let mut stmt = conn.prepare(
+            "SELECT sender, recipient, body, sent_at
+               FROM messages
+              ORDER BY id DESC
+              LIMIT ?1",
+        )?;
+        let rows = stmt.query_map(params![limit_i], |row| {
+            Ok(MessageEvent::Sent {
+                from: row.get(0)?,
+                to: row.get(1)?,
+                body: row.get(2)?,
+                at: row.get(3)?,
+            })
+        })?;
+        rows.collect::>>()
+            .map_err(Into::into)
+    }
+
     /// Number of undelivered messages addressed to `recipient`. Non-mutating
     /// — used by the harness to surface "N unread" in tool-result status
     /// lines without popping the queue.
diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs
index bd8fd07e..4df990ec 100644
--- a/hive-c0re/src/coordinator.rs
+++ b/hive-c0re/src/coordinator.rs
@@ -4,15 +4,23 @@
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 
 use anyhow::{Context, Result};
+use tokio::sync::broadcast;
 
 use crate::agent_server::{self, AgentSocket};
 use crate::approvals::Approvals;
 use crate::broker::Broker;
+use crate::dashboard_events::DashboardEvent;
 use crate::operator_questions::OperatorQuestions;
 
+/// Capacity of the dashboard event channel. Slow browser subscribers
+/// (idle tab, throttled connection) drop frames past this — that's
+/// fine, the seq dedupe makes a reconnect resync safe.
+const DASHBOARD_CHANNEL: usize = 256;
+
 const AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents";
 const MANAGER_RUNTIME_ROOT: &str = "/run/hyperhive/manager";
 /// Manager-editable per-agent config repos. Bind-mounted RW into the manager
@@ -47,6 +55,15 @@ pub struct Coordinator {
     /// Read by the dashboard to render a spinner; cleared when the action
     /// resolves (success or failure).
     transient: Mutex>,
+    /// Unified wire-facing event channel feeding the dashboard SSE
+    /// stream. Carries broker messages (mirrored from `broker.subscribe`
+    /// by the forwarder task in `main.rs`) and dashboard-only mutation
+    /// events (approval added/resolved, question added/answered, etc.).
+    /// Snapshot endpoints capture `event_seq` before reading state so
+    /// the client can dedupe its buffered live traffic against the
+    /// snapshot.
+    dashboard_events: broadcast::Sender,
+    event_seq: AtomicU64,
 }
 
 /// Per-agent in-progress state that the dashboard surfaces between approve
@@ -98,6 +115,7 @@ impl Coordinator {
         let broker = Broker::open(db_path).context("open broker")?;
         let approvals = Approvals::open(db_path).context("open approvals")?;
         let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
+        let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
         Ok(Self {
             broker: Arc::new(broker),
             approvals: Arc::new(approvals),
@@ -107,9 +125,42 @@ impl Coordinator {
             operator_pronouns,
             agents: Mutex::new(HashMap::new()),
             transient: Mutex::new(HashMap::new()),
+            dashboard_events,
+            event_seq: AtomicU64::new(0),
         })
     }
 
+    /// Subscribe to the unified dashboard event channel. Used by the
+    /// `/dashboard/stream` SSE handler and by the broker-to-dashboard
+    /// forwarder task.
+    pub fn dashboard_subscribe(&self) -> broadcast::Receiver {
+        self.dashboard_events.subscribe()
+    }
+
+    /// Stamp the next sequence number. Each emission of a
+    /// `DashboardEvent` should fill its `seq` with `next_seq()` so the
+    /// frame the wire carries is the one the client uses to dedupe.
+    pub fn next_seq(&self) -> u64 {
+        self.event_seq.fetch_add(1, Ordering::SeqCst) + 1
+    }
+
+    /// Current high-water seq. Snapshot endpoints read this *before*
+    /// gathering state so the (snapshot.seq, snapshot) pair satisfies:
+    /// any frame with `seq > snapshot.seq` is post-snapshot. The seq
+    /// captured here may grow during snapshot construction — clients
+    /// may double-apply such events, which renderers must tolerate.
+    pub fn current_seq(&self) -> u64 {
+        self.event_seq.load(Ordering::SeqCst)
+    }
+
+    /// Broadcast a freshly-built `DashboardEvent` (caller fills `seq`
+    /// via `next_seq()`). Returns silently when there are no
+    /// subscribers — the dashboard channel is best-effort presentation
+    /// plumbing, not a delivery guarantee.
+    pub fn emit_dashboard_event(&self, event: DashboardEvent) {
+        let _ = self.dashboard_events.send(event);
+    }
+
     pub fn register_agent(self: &Arc, name: &str) -> Result {
         // Idempotent: drop any existing listener so re-registration (e.g. on rebuild,
         // or after a hive-c0re restart cleared /run/hyperhive) gets a fresh socket.
diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs
index cf0c626e..4e78acde 100644
--- a/hive-c0re/src/dashboard.rs
+++ b/hive-c0re/src/dashboard.rs
@@ -3,7 +3,6 @@
 //! repo, plus approve/deny buttons), and the manager.
 
 use std::convert::Infallible;
-use std::fmt::Write as _;
 use std::net::SocketAddr;
 use std::path::Path;
 use std::sync::Arc;
@@ -58,7 +57,9 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> {
         .route("/request-spawn", post(post_request_spawn))
         .route("/op-send", post(post_op_send))
         .route("/meta-update", post(post_meta_update))
-        .route("/messages/stream", get(messages_stream))
+        .route("/dashboard/stream", get(dashboard_stream))
+        .route("/dashboard/history", get(dashboard_history))
+        .route("/static/hive-fr0nt.js", get(serve_shared_js))
         .with_state(AppState { coord });
     let addr = SocketAddr::from(([0, 0, 0, 0], port));
     let listener = bind_with_retry(addr).await?;
@@ -72,7 +73,7 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> {
 // (static) shell; `GET /static/*` serves the CSS + JS app; `GET /api/state`
 // returns the current snapshot as JSON. The JS app fetches state on load,
 // re-fetches after every async-form submit, and listens on
-// `/messages/stream` for broker traffic.
+// `/dashboard/stream` for the unified live event channel.
 // ---------------------------------------------------------------------------
 
 /// `SO_REUSEADDR` bind with retry. Mirrors the per-agent variant —
@@ -133,8 +134,23 @@ async fn serve_app_js() -> impl IntoResponse {
     )
 }
 
+async fn serve_shared_js() -> impl IntoResponse {
+    (
+        [("content-type", "application/javascript")],
+        hive_fr0nt::TERMINAL_JS,
+    )
+}
+
 #[derive(Serialize)]
 struct StateSnapshot {
+    /// Broker seq at the moment this snapshot was assembled. Clients
+    /// dedupe their buffered SSE traffic against this value: any
+    /// `MessageEvent` with `seq <= snapshot.seq` is already reflected in
+    /// the snapshot (or pre-dates it); anything with `seq > snapshot.seq`
+    /// is post-snapshot and should be applied. Set to 0 in the
+    /// pre-emit case (no events ever fired) — clients treat that as
+    /// "apply everything you've buffered".
+    seq: u64,
     hostname: String,
     manager_port: u16,
     any_stale: bool,
@@ -144,10 +160,6 @@ struct StateSnapshot {
     /// Last 30 resolved approvals (approved / denied / failed), newest-
     /// first. Drives the "history" tab on the approvals section.
     approval_history: Vec,
-    /// Latest messages addressed to `operator` — surfaces agent replies
-    /// asynchronously so the operator can see them without watching the
-    /// live panel during a turn.
-    operator_inbox: Vec,
     /// Pending operator questions (currently only from the manager).
     /// `ask_operator` returns immediately with the id; on `/answer-question`
     /// we mark the row answered and fire `HelperEvent::OperatorAnswered`
@@ -243,8 +255,13 @@ struct ApprovalView {
     kind: &'static str,
     /// First 12 chars of the `commit_ref`, for `ApplyCommit` only.
     sha_short: Option,
-    /// Pre-rendered syntax-coloured diff HTML, for `ApplyCommit` only.
-    diff_html: Option,
+    /// Raw unified diff text, for `ApplyCommit` only. The client splits
+    /// on `\n` and per-line classifies (`+` / `-` / `@@` / `--- ` / `+++ `
+    /// → diff-add / diff-del / diff-hunk / diff-file). Shipping raw
+    /// instead of pre-rendered HTML saves bytes on the wire (no
+    /// per-line `` markup) and removes the only HTML-escape
+    /// surface from the snapshot.
+    diff: Option,
     /// Manager-supplied description shown on the approval card.
     #[serde(skip_serializing_if = "Option::is_none")]
     description: Option,
@@ -276,6 +293,14 @@ async fn api_state(headers: HeaderMap, State(state): State) -> axum::J
         .unwrap_or("localhost");
     let hostname = host.split(':').next().unwrap_or(host).to_owned();
 
+    // Capture the unified dashboard-channel seq *before* any read so the
+    // dedupe contract is "events with seq > snapshot.seq are
+    // post-snapshot, never missed." An event landing during snapshot
+    // construction may be doubly applied (snapshot caught the write +
+    // client also applies the SSE frame) — that's a renderer's problem
+    // to make idempotent, not ours to avoid here.
+    let seq = state.coord.current_seq();
+
     let raw_containers = log_default("nixos-container list", lifecycle::list().await);
     let current_rev = crate::auto_update::current_flake_rev(&state.coord.hyperhive_flake);
     let transient_snapshot = state.coord.transient_snapshot();
@@ -298,18 +323,15 @@ async fn api_state(headers: HeaderMap, State(state): State) -> axum::J
     let tombstones = build_tombstone_views(&state.coord, &containers, &transient_snapshot);
     let port_conflicts = build_port_conflicts(&containers);
 
-    let operator_inbox = log_default(
-        "broker.recent_for(operator)",
-        state
-            .coord
-            .broker
-            .recent_for(hive_sh4re::OPERATOR_RECIPIENT, 50),
-    );
+    // operator_inbox used to be served here as a 50-row array; the
+    // dashboard now derives it client-side from the message stream
+    // (terminal backfill + live SSE), so the snapshot stops shipping it.
     let questions = log_default("questions.pending", state.coord.questions.pending());
     let question_history =
         log_default("questions.recent_answered", state.coord.questions.recent_answered(20));
 
     axum::Json(StateSnapshot {
+        seq,
         hostname,
         manager_port: MANAGER_PORT,
         any_stale,
@@ -318,7 +340,6 @@ async fn api_state(headers: HeaderMap, State(state): State) -> axum::J
         approvals,
         approval_history,
         meta_inputs: read_meta_inputs(),
-        operator_inbox,
         questions,
         question_history,
         tombstones,
@@ -622,7 +643,7 @@ async fn build_approval_views(approvals: Vec) -> Vec {
                     agent: a.agent.clone(),
                     kind: "apply_commit",
                     sha_short: Some(sha),
-                    diff_html: Some(render_diff_lines(&diff)),
+                    diff: Some(diff),
                     description: a.description,
                 }
             }
@@ -631,7 +652,7 @@ async fn build_approval_views(approvals: Vec) -> Vec {
                 agent: a.agent,
                 kind: "spawn",
                 sha_short: None,
-                diff_html: None,
+                diff: None,
                 description: a.description,
             },
         });
@@ -699,12 +720,58 @@ fn dir_size_bytes(root: &Path) -> u64 {
     total
 }
 
-async fn messages_stream(
+async fn dashboard_history(State(state): State) -> Response {
+    // Backfill source for the dashboard terminal. Returns up to ~200
+    // historical broker messages (no other event kinds are persisted)
+    // converted to `DashboardEvent::Sent` JSON so the client can replay
+    // through the same dispatch path as live frames. Wrapped in
+    // `{ seq, events }`: the seq is the dashboard channel's high-water
+    // mark at fetch time. Clients use it to dedupe their buffered live
+    // SSE traffic (drop anything with `seq <= history_seq`) so a frame
+    // that lands between SSE-subscribe and history-fetch isn't shown
+    // twice and isn't lost. Historical rows carry `seq = 0`; the
+    // boundary seq is what closes the dedupe window.
+    const HISTORY_LIMIT: u64 = 200;
+    let seq = state.coord.current_seq();
+    match state.coord.broker.recent_all(HISTORY_LIMIT) {
+        Ok(mut messages) => {
+            messages.reverse();
+            let events: Vec = messages
+                .into_iter()
+                .map(|m| match m {
+                    crate::broker::MessageEvent::Sent { from, to, body, at } => {
+                        crate::dashboard_events::DashboardEvent::Sent {
+                            seq: 0,
+                            from,
+                            to,
+                            body,
+                            at,
+                        }
+                    }
+                    crate::broker::MessageEvent::Delivered { from, to, body, at } => {
+                        crate::dashboard_events::DashboardEvent::Delivered {
+                            seq: 0,
+                            from,
+                            to,
+                            body,
+                            at,
+                        }
+                    }
+                })
+                .collect();
+            axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response()
+        }
+        Err(e) => error_response(&format!("dashboard/history failed: {e:#}")),
+    }
+}
+
+async fn dashboard_stream(
     State(state): State,
 ) -> Sse>> {
-    let rx = state.coord.broker.subscribe();
+    let rx = state.coord.dashboard_subscribe();
     let stream = BroadcastStream::new(rx).filter_map(|res| {
-        // Drop lagged events. Browsers reconnect; nothing to do here.
+        // Drop lagged frames. Browsers reconnect; the seq dedupe on
+        // reconnect skips any frame already reflected in the snapshot.
         let event = res.ok()?;
         let json = serde_json::to_string(&event).ok()?;
         Some(Ok(Event::default().data(json)))
@@ -1074,7 +1141,13 @@ async fn post_op_send(State(state): State, Form(form): Form) -> Vec {
         .collect()
 }
 
-/// Render a unified diff with per-line CSS classes so the dashboard can
-/// colour adds / dels / hunk headers / context. Each line becomes a
-/// `` tagged by its leading character; the wrapping `
` keeps
-/// whitespace intact.
-fn render_diff_lines(diff: &str) -> String {
-    let mut out = String::new();
-    for raw in diff.lines() {
-        let cls = match raw.as_bytes().first() {
-            // file headers (`--- a/...` / `+++ b/...`) come before any
-            // line starting with a single `+`/`-`. similar-rs emits them
-            // with the doubled prefix.
-            _ if raw.starts_with("--- ") => "diff-file",
-            _ if raw.starts_with("+++ ") => "diff-file",
-            Some(b'@') => "diff-hunk",
-            Some(b'+') => "diff-add",
-            Some(b'-') => "diff-del",
-            _ => "diff-ctx",
-        };
-        let _ = writeln!(out, "{}", html_escape(raw),);
-    }
-    out
-}
-
 /// Host-side mirror of `hive_ag3nt::login::has_session`. Returns true if the
 /// agent's bound `~/.claude/` dir on disk contains any regular file. The
 /// dashboard reads this each render so logins driven from the agent web UI
@@ -1374,8 +1424,3 @@ async fn git_diff_main_to(applied_dir: &Path, target_ref: &str) -> Result String {
-    s.replace('&', "&")
-        .replace('<', "<")
-        .replace('>', ">")
-}
diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs
new file mode 100644
index 00000000..fc8305fe
--- /dev/null
+++ b/hive-c0re/src/dashboard_events.rs
@@ -0,0 +1,47 @@
+//! Unified dashboard event channel.
+//!
+//! Anything the browser wants to react to in near-real-time flows through
+//! `Coordinator.dashboard_events`. Each event is stamped with a monotonic
+//! per-process `seq` so the client can dedupe its buffered live traffic
+//! against snapshot/history responses (drop frames with
+//! `seq <= snapshot.seq`).
+//!
+//! Why one channel instead of one-per-domain: browsers cap concurrent
+//! SSE connections per origin (~6 in chrome) and dispatch-by-kind on the
+//! client is a one-liner. Splits get reserved for high-volume sub-streams
+//! that most consumers don't care about (none yet).
+//!
+//! Message-broker traffic (`Sent` / `Delivered`) lives on this channel
+//! too. A background forwarder task in `main.rs` subscribes to the broker
+//! and re-emits each `MessageEvent` as a `DashboardEvent::Sent` /
+//! `DashboardEvent::Delivered` with a freshly-stamped seq. Keeping the
+//! broker's intra-process channel separate avoids coupling the broker
+//! (used by `recv_blocking` inside the harness loop) to dashboard
+//! presentation concerns.
+//!
+//! New mutation kinds (approval added/resolved, question added/answered,
+//! transient changed, etc.) land here as additional variants. The client
+//! dispatches by `kind` and updates the relevant section.
+
+use serde::Serialize;
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "snake_case", tag = "kind")]
+pub enum DashboardEvent {
+    /// Broker `Sent` event mirrored onto the dashboard channel.
+    Sent {
+        seq: u64,
+        from: String,
+        to: String,
+        body: String,
+        at: i64,
+    },
+    /// Broker `Delivered` event mirrored onto the dashboard channel.
+    Delivered {
+        seq: u64,
+        from: String,
+        to: String,
+        body: String,
+        at: i64,
+    },
+}
diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs
index 489dbacf..97c5adf4 100644
--- a/hive-c0re/src/main.rs
+++ b/hive-c0re/src/main.rs
@@ -14,6 +14,7 @@ mod client;
 mod coordinator;
 mod crash_watch;
 mod dashboard;
+mod dashboard_events;
 mod events_vacuum;
 mod forge;
 mod lifecycle;
@@ -170,6 +171,12 @@ async fn main() -> Result<()> {
             // Reminder scheduler: drains due reminders + handles
             // file_path payload persistence. See reminder_scheduler.rs.
             reminder_scheduler::spawn(coord.clone());
+            // Forward every broker event onto the unified dashboard
+            // channel with a freshly-stamped seq, so the dashboard SSE
+            // sees broker messages + future mutation events on one
+            // stream with one monotonic seq. The broker's intra-process
+            // channel (used by `recv_blocking`) stays untouched.
+            spawn_broker_to_dashboard_forwarder(coord.clone());
             let dash_coord = coord.clone();
             tokio::spawn(async move {
                 if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {
@@ -202,6 +209,46 @@ async fn main() -> Result<()> {
     }
 }
 
+/// Re-emit every broker `MessageEvent` onto the dashboard channel as
+/// a `DashboardEvent::Sent` / `Delivered` with a freshly-stamped seq.
+/// Background task; runs for the life of the process. On a lagged
+/// broker subscription we just keep going — the dashboard channel is
+/// best-effort presentation plumbing, the broker keeps its own sqlite
+/// log for replay.
+fn spawn_broker_to_dashboard_forwarder(coord: Arc) {
+    use broker::MessageEvent;
+    use dashboard_events::DashboardEvent;
+    let mut rx = coord.broker.subscribe();
+    tokio::spawn(async move {
+        loop {
+            match rx.recv().await {
+                Ok(MessageEvent::Sent { from, to, body, at }) => {
+                    coord.emit_dashboard_event(DashboardEvent::Sent {
+                        seq: coord.next_seq(),
+                        from,
+                        to,
+                        body,
+                        at,
+                    });
+                }
+                Ok(MessageEvent::Delivered { from, to, body, at }) => {
+                    coord.emit_dashboard_event(DashboardEvent::Delivered {
+                        seq: coord.next_seq(),
+                        from,
+                        to,
+                        body,
+                        at,
+                    });
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                    tracing::warn!(skipped = n, "broker-to-dashboard forwarder lagged");
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+            }
+        }
+    });
+}
+
 fn render(resp: HostResponse) -> Result<()> {
     println!("{}", serde_json::to_string_pretty(&resp)?);
     if !resp.ok {
diff --git a/hive-fr0nt/assets/terminal.js b/hive-fr0nt/assets/terminal.js
index 625af28a..56f7dc6b 100644
--- a/hive-fr0nt/assets/terminal.js
+++ b/hive-fr0nt/assets/terminal.js
@@ -14,7 +14,11 @@
 //       delivered: (ev, api) => api.row('msgrow delivered', ...),
 //       _default:  (ev, api) => api.row('note', JSON.stringify(ev)),
 //     },
-//     onLiveEvent: (ev) => { /* side effects: notifications, state pokes */ },
+//     onLiveEvent: (ev) => { /* live-only side effects (notif, state pokes) */ },
+//     onAnyEvent:  (ev, { fromHistory }) => { /* runs for every event in
+//       both backfill replay and live — use for derived views that need
+//       the full picture (e.g. a per-recipient inbox built from broker
+//       events) */ },
 //     onBackfillDone: (count) => { /* one-shot after history replay */ },
 //     pillAnchor: document.getElementById('msgflow').parentElement,
 //   });
@@ -164,38 +168,41 @@
         console.error('terminal renderer threw', ev, err);
         row('note', '[render err] ' + (err && err.message ? err.message : err));
       }
-    }
-
-    async function backfill() {
-      if (!opts.historyUrl) {
-        if (opts.onBackfillDone) opts.onBackfillDone(0);
-        return;
-      }
-      try {
-        const resp = await fetch(opts.historyUrl);
-        if (!resp.ok) {
-          if (opts.onBackfillDone) opts.onBackfillDone(0);
-          return;
-        }
-        const events = await resp.json();
-        currentNoAnim = true;
-        for (const ev of events) dispatch(ev, true);
-        currentNoAnim = false;
-        if (events.length) row('note', '─── live (older above) ───');
-        else placeholder('(connected — waiting for events)');
-        if (opts.onBackfillDone) opts.onBackfillDone(events.length);
-      } catch (err) {
-        console.warn('history backfill failed', err);
-        if (opts.onBackfillDone) opts.onBackfillDone(0);
+      if (opts.onAnyEvent) {
+        try { opts.onAnyEvent(ev, { fromHistory }); }
+        catch (err) { console.error('onAnyEvent threw', err); }
       }
     }
 
-    function subscribe() {
+    // Subscribe → buffer → fetch history → dedupe → apply.
+    //
+    // Race the SSE subscription opens before the history fetch starts.
+    // Live events that land before history resolves are buffered, not
+    // rendered. Once the history response (`{ seq, events }`) arrives we:
+    //   1. Replay `events` (fromHistory=true).
+    //   2. Drop buffered events with `seq <= history.seq` — they're
+    //      already reflected in the history rows above.
+    //   3. Apply remaining buffered events (fromHistory=false).
+    //   4. Switch to live mode: each new SSE event dispatches immediately.
+    //
+    // Without this dance an event that fires between history-fetch and
+    // SSE-subscribe goes missing; without seq dedupe the same event
+    // shows twice (once via history, once via live buffer). Both bugs
+    // were latent before.
+    //
+    // If `historyUrl` is unset we skip the dance: buffered events apply
+    // as live the moment the buffer flushes (no dedupe possible without
+    // a boundary seq).
+    function start() {
+      let live = false;
+      let buffered = [];
+
       const es = new EventSource(opts.streamUrl);
       es.onmessage = (e) => {
         let ev;
         try { ev = JSON.parse(e.data); }
         catch (err) { row('note', '[parse err] ' + e.data); return; }
+        if (!live) { buffered.push(ev); return; }
         dispatch(ev, false);
         if (opts.onLiveEvent) {
           try { opts.onLiveEvent(ev); }
@@ -206,10 +213,62 @@
         if (es.readyState === EventSource.CONNECTING) row('note', '[reconnecting…]');
         else row('note', '[disconnected]');
       };
-      return es;
+
+      function flushBuffered(boundarySeq) {
+        const drained = buffered;
+        buffered = [];
+        live = true;
+        for (const ev of drained) {
+          // ev.seq is set by the server on live frames; absent/0 means
+          // "no dedupe possible, apply." Historical replays via the
+          // history endpoint carry no seq either way.
+          if (boundarySeq != null && typeof ev.seq === 'number' && ev.seq <= boundarySeq) {
+            continue;
+          }
+          dispatch(ev, false);
+          if (opts.onLiveEvent) {
+            try { opts.onLiveEvent(ev); }
+            catch (err) { console.error('onLiveEvent threw', err); }
+          }
+        }
+      }
+
+      async function backfill() {
+        if (!opts.historyUrl) {
+          flushBuffered(null);
+          if (opts.onBackfillDone) opts.onBackfillDone(0);
+          return;
+        }
+        try {
+          const resp = await fetch(opts.historyUrl);
+          if (!resp.ok) {
+            flushBuffered(null);
+            if (opts.onBackfillDone) opts.onBackfillDone(0);
+            return;
+          }
+          const body = await resp.json();
+          // Accept the envelope `{ seq, events }`. A bare array means
+          // the server hasn't been updated to include seq yet — treat
+          // it as "no dedupe possible."
+          const events = Array.isArray(body) ? body : (body.events || []);
+          const boundarySeq = Array.isArray(body) ? null : (body.seq ?? null);
+          currentNoAnim = true;
+          for (const ev of events) dispatch(ev, true);
+          currentNoAnim = false;
+          if (events.length) row('note', '─── live (older above) ───');
+          else placeholder('(connected — waiting for events)');
+          flushBuffered(boundarySeq);
+          if (opts.onBackfillDone) opts.onBackfillDone(events.length);
+        } catch (err) {
+          console.warn('history backfill failed', err);
+          flushBuffered(null);
+          if (opts.onBackfillDone) opts.onBackfillDone(0);
+        }
+      }
+      return backfill();
     }
 
-    const ready = backfill().then(subscribe);
+    const ready = start();
     return { row, details, detailsDiff, placeholder, ready };
   }