todo: phase 5b — mutation events for approvals/questions/transients

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.

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.

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<String> },
@@ -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<broadcast::Sender<LiveEvent>>,
+    tx: Arc<broadcast::Sender<BusEvent>>,
+    /// 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<AtomicU64>,
     /// 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<LiveEvent> {
+    pub fn subscribe(&self) -> broadcast::Receiver<BusEvent> {
         self.tx.subscribe()
     }
 

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<bool>
     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<bool>
                     }
                     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<bool>
             // renders; the tracing line is what `journalctl -M <c> -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();

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<AppState>) -> axum::Json<StateSnapshot> {
+    // 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<AppState>) -> axum::Json<StateSnapshot> {
     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<AppState>, Form(form): Form<SendForm>) ->
         },
     };
     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<AppState>,
-) -> axum::Json<Vec<crate::events::LiveEvent>> {
-    axum::Json(state.bus.history())
+async fn events_history(State(state): State<AppState>) -> axum::Json<serde_json::Value> {
+    // 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<AppState>, Form(form): Form<ModelFor
         return error_response("model: name required");
     }
     state.bus.set_model(name);
-    state.bus.emit(crate::events::LiveEvent::Note(format!(
-        "operator: /model — claude model set to '{name}' for future turns"
-    )));
+    state.bus.emit(crate::events::LiveEvent::Note {
+        text: format!("operator: /model — claude model set to '{name}' for future turns"),
+    });
     tracing::info!(%name, "operator set model");
     Redirect::to("/").into_response()
 }
@@ -450,16 +471,16 @@ async fn post_compact(State(state): State<AppState>) -> 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<AppState>) -> Response {
 /// than asking claude to forget mid-stream.
 async fn post_new_session(State(state): State<AppState>) -> 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<AppState>) -> 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()
 }
 

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 <pre>); 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 =
-        '<span class="msg-ts">' + tsFmt(m.at) + '</span>' +
-        '<span class="msg-arrow">' + kind + '</span>' +
-        '<span class="msg-from">' + esc(m.from) + '</span>' +
+    function renderMsg(ev, api, glyph) {
+      const el = api.row('msgrow ' + ev.kind, '');
+      el.innerHTML =
+        '<span class="msg-ts">' + tsFmt(ev.at) + '</span>' +
+        '<span class="msg-arrow">' + glyph + '</span>' +
+        '<span class="msg-from">' + esc(ev.from) + '</span>' +
         '<span class="msg-sep">→</span>' +
-        '<span class="msg-to">' + esc(m.to) + '</span>' +
-        '<span class="msg-body">' + esc(m.body) + '</span>';
-      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);
-    };
+        '<span class="msg-to">' + esc(ev.to) + '</span>' +
+        '<span class="msg-body">' + esc(ev.body) + '</span>';
+    }
+    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;
         }

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

hive-c0re/assets/index.html

@@ -61,13 +61,15 @@
   <h2>◆ MESS4GE FL0W ◆</h2>
   <div class="divider">══════════════════════════════════════════════════════════════</div>
   <p class="meta">live tail — newest at the top. tap on every <code>send</code> / <code>recv</code> through the broker. compose below: <code>@name</code> picks the recipient (sticky until you @ someone else); <code>tab</code> completes.</p>
-  <div id="msgflow" class="msgflow"><span class="meta">connecting…</span></div>
-  <div id="op-compose" class="op-compose">
-    <span id="op-compose-prompt" class="op-compose-prompt">@—&gt;</span>
-    <textarea id="op-compose-input" class="op-compose-input"
-              placeholder="@agent message…  (enter sends, shift+enter newline, tab completes @-mention)"
-              rows="1" autocomplete="off"></textarea>
-    <div id="op-compose-suggest" class="op-compose-suggest" hidden></div>
+  <div class="terminal-wrap">
+    <div id="msgflow" class="live"><div class="meta">connecting…</div></div>
+    <div id="op-compose" class="op-compose">
+      <span id="op-compose-prompt" class="op-compose-prompt">@—&gt;</span>
+      <textarea id="op-compose-input" class="op-compose-input"
+                placeholder="@agent message…  (enter sends, shift+enter newline, tab completes @-mention)"
+                rows="1" autocomplete="off"></textarea>
+      <div id="op-compose-suggest" class="op-compose-suggest" hidden></div>
+    </div>
   </div>
 
   <footer>
@@ -75,6 +77,7 @@
     <p>▲△▲ <a href="https://git.berlin.ccc.de/vinzenz/hyperhive">hyperhive</a> ▲△▲ hive-c0re on this host ▲△▲</p>
   </footer>
 
+  <script src="/static/hive-fr0nt.js" defer></script>
   <script src="/static/app.js" defer></script>
 </body>
 </html>

hive-c0re/src/broker.rs

@@ -46,6 +46,11 @@ const EVENT_CHANNEL: usize = 256;
 /// self-documenting.
 pub type DueReminder = (String, i64, String, Option<String>);
 
+/// 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<Vec<MessageEvent>> {
+        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::<rusqlite::Result<Vec<_>>>()
+            .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.

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<HashMap<String, TransientState>>,
+    /// 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<DashboardEvent>,
+    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<DashboardEvent> {
+        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<Self>, name: &str) -> Result<PathBuf> {
         // 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.

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<Coordinator>) -> 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<Coordinator>) -> 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<ApprovalHistoryView>,
-    /// 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<hive_sh4re::InboxRow>,
     /// 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<String>,
-    /// Pre-rendered syntax-coloured diff HTML, for `ApplyCommit` only.
-    diff_html: Option<String>,
+    /// 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 `<span>` markup) and removes the only HTML-escape
+    /// surface from the snapshot.
+    diff: Option<String>,
     /// Manager-supplied description shown on the approval card.
     #[serde(skip_serializing_if = "Option::is_none")]
     description: Option<String>,
@@ -276,6 +293,14 @@ async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> 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<AppState>) -> 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<AppState>) -> 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<Approval>) -> Vec<ApprovalView> {
                     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<Approval>) -> Vec<ApprovalView> {
                 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<AppState>) -> 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<crate::dashboard_events::DashboardEvent> = 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<AppState>,
 ) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
-    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<AppState>, Form(form): Form<OpSendForm
     }) {
         return error_response(&format!("op-send to {to} failed: {e:#}"));
     }
-    Redirect::to("/").into_response()
+    // 200 instead of 303 → the client doesn't refetch /api/state. The
+    // broker `send` already emitted a `MessageEvent` which the
+    // dashboard channel forwarder mirrors as `DashboardEvent::Sent`,
+    // and the page's terminal + inbox derive from that stream — so the
+    // operator's send shows up the same way an agent's send does, with
+    // no full-state refresh in between.
+    (axum::http::StatusCode::OK, "ok").into_response()
 }
 
 async fn post_request_spawn(
@@ -1304,29 +1377,6 @@ fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<Approval> {
         .collect()
 }
 
-/// Render a unified diff with per-line CSS classes so the dashboard can
-/// colour adds / dels / hunk headers / context. Each line becomes a
-/// `<span>` tagged by its leading character; the wrapping `<pre>` 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, "<span class=\"{cls}\">{}</span>", 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
     Ok(String::from_utf8_lossy(&out.stdout).into_owned())
 }
 
-fn html_escape(s: &str) -> String {
-    s.replace('&', "&amp;")
-        .replace('<', "&lt;")
-        .replace('>', "&gt;")
-}

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

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<Coordinator>) {
+    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 {

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 };
   }