From 8c186d4fb7c07f5f3d9c6a25fff3f709ecf6363c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 17 May 2026 11:56:29 +0200 Subject: [PATCH 1/9] dashboard: msgflow uses shared terminal + backfill via /messages/history --- hive-c0re/assets/app.js | 80 +++++++++++++++++----------------- hive-c0re/assets/dashboard.css | 35 +++++---------- hive-c0re/assets/index.html | 17 +++++--- hive-c0re/src/broker.rs | 30 +++++++++++++ hive-c0re/src/dashboard.rs | 26 +++++++++++ 5 files changed, 116 insertions(+), 72 deletions(-) diff --git a/hive-c0re/assets/app.js b/hive-c0re/assets/app.js index 600d6729..d9eafbf2 100644 --- a/hive-c0re/assets/app.js +++ b/hive-c0re/assets/app.js @@ -955,17 +955,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 +976,38 @@ 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: '/messages/history', + streamUrl: '/messages/stream', + renderers: { + sent: (ev, api) => renderMsg(ev, api, '→'), + delivered: (ev, api) => renderMsg(ev, api, '✓'), + }, + onLiveEvent: (ev) => { + pulseBanner(); + if (ev.kind === 'sent' && ev.to === 'operator') { + refreshState(); + 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 ─────────────────────────── 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..93931d38 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -129,6 +129,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/dashboard.rs b/hive-c0re/src/dashboard.rs index cf0c626e..6da925cc 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -59,6 +59,8 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { .route("/op-send", post(post_op_send)) .route("/meta-update", post(post_meta_update)) .route("/messages/stream", get(messages_stream)) + .route("/messages/history", get(messages_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?; @@ -133,6 +135,13 @@ 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 { hostname: String, @@ -699,6 +708,23 @@ fn dir_size_bytes(root: &Path) -> u64 { total } +async fn messages_history(State(state): State) -> Response { + // Backfill source for the dashboard message-flow terminal. Returns + // up to ~200 historical broker messages as `MessageEvent::Sent` JSON + // — same shape as the live `/messages/stream`, so the renderer + // doesn't branch on history vs. live. + const HISTORY_LIMIT: u64 = 200; + match state.coord.broker.recent_all(HISTORY_LIMIT) { + Ok(mut events) => { + // recent_all returns newest-first; reverse so the replay + // builds chronologically (matches the agent /events/history). + events.reverse(); + axum::Json(events).into_response() + } + Err(e) => error_response(&format!("messages/history failed: {e:#}")), + } +} + async fn messages_stream( State(state): State, ) -> Sse>> { From 1340a654e72a38cd984c984581f5c862f373a855 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 17 May 2026 12:26:00 +0200 Subject: [PATCH 2/9] sse: seq plumbing + subscribe-first dedupe dance --- hive-ag3nt/src/events.rs | 45 +++++++++++++-- hive-ag3nt/src/web_ui.rs | 23 ++++++-- hive-c0re/src/broker.rs | 33 +++++++++++ hive-c0re/src/dashboard.rs | 30 +++++++++- hive-fr0nt/assets/terminal.js | 103 +++++++++++++++++++++++++--------- 5 files changed, 197 insertions(+), 37 deletions(-) diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 0588c820..e8944836 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")] @@ -216,7 +228,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 +276,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 +285,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 +366,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/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 5eb13fce..941eb298 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, @@ -338,10 +348,15 @@ async fn post_send(State(state): State, Form(form): Form) -> } } -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( diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 93931d38..892acc50 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -3,6 +3,7 @@ use std::path::Path; use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result}; @@ -50,12 +51,14 @@ pub type DueReminder = (String, i64, String, Option); #[serde(rename_all = "snake_case", tag = "kind")] pub enum MessageEvent { Sent { + seq: u64, from: String, to: String, body: String, at: i64, }, Delivered { + seq: u64, from: String, to: String, body: String, @@ -66,6 +69,13 @@ pub enum MessageEvent { pub struct Broker { conn: Mutex, events: broadcast::Sender, + /// Monotonic per-process counter stamped onto every emitted + /// `MessageEvent`. Persisted nowhere — clients always treat a hive-c0re + /// restart as "everything is new" (fresh snapshot, fresh stream of + /// seqs starting at 1). Historical rows replayed via `recent_all` + /// carry `seq = 0` since they predate the live stream the seq is + /// meant to dedupe against. + event_seq: AtomicU64, } impl Broker { @@ -81,6 +91,7 @@ impl Broker { Ok(Self { conn: Mutex::new(conn), events, + event_seq: AtomicU64::new(0), }) } @@ -88,6 +99,20 @@ impl Broker { self.events.subscribe() } + /// 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); any with seq <= snapshot.seq either pre-dates + /// the snapshot or was already captured by it. Clients dedupe their + /// buffered SSE traffic against this value. + 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 + } + pub fn send(&self, message: &Message) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( @@ -96,6 +121,7 @@ impl Broker { )?; drop(conn); let _ = self.events.send(MessageEvent::Sent { + seq: self.next_seq(), from: message.from.clone(), to: message.to.clone(), body: message.body.clone(), @@ -149,6 +175,11 @@ impl Broker { )?; let rows = stmt.query_map(params![limit_i], |row| { Ok(MessageEvent::Sent { + // Historical events: seq=0 (never compared against live + // seqs). Live dedupe windows close against + // history_seq = broker.current_seq() captured at fetch + // time, not against per-row seqs. + seq: 0, from: row.get(0)?, to: row.get(1)?, body: row.get(2)?, @@ -256,6 +287,7 @@ impl Broker { )?; drop(conn); let _ = self.events.send(MessageEvent::Delivered { + seq: self.next_seq(), from: from.clone(), to: to.clone(), body: body.clone(), @@ -332,6 +364,7 @@ impl Broker { tx.commit()?; drop(conn); let _ = self.events.send(MessageEvent::Sent { + seq: self.next_seq(), from: "reminder".to_owned(), to: agent.to_owned(), body: message.to_owned(), diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 6da925cc..22ebcbd3 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -144,6 +144,14 @@ async fn serve_shared_js() -> impl IntoResponse { #[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, @@ -285,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 broker seq *before* any read so the dedupe contract + // is "events with seq > snapshot.seq are post-snapshot, never + // missed." A broker event landing during snapshot construction may + // be doubly applied (snapshot caught the write + client also + // applies the SSE event) — that's a renderer's problem to make + // idempotent, not ours to avoid here. + let seq = state.coord.broker.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(); @@ -319,6 +335,7 @@ async fn api_state(headers: HeaderMap, State(state): State) -> axum::J log_default("questions.recent_answered", state.coord.questions.recent_answered(20)); axum::Json(StateSnapshot { + seq, hostname, manager_port: MANAGER_PORT, any_stale, @@ -711,15 +728,22 @@ fn dir_size_bytes(root: &Path) -> u64 { async fn messages_history(State(state): State) -> Response { // Backfill source for the dashboard message-flow terminal. Returns // up to ~200 historical broker messages as `MessageEvent::Sent` JSON - // — same shape as the live `/messages/stream`, so the renderer - // doesn't branch on history vs. live. + // wrapped in `{ seq, events }`. The seq is the broker'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 message + // that lands between SSE-subscribe and history-fetch isn't shown + // twice and isn't lost. const HISTORY_LIMIT: u64 = 200; + // Capture seq *before* the query so the dedupe contract is + // "drop buffered events you've already seen in history" — never + // "lose an event that fired between the read and the timestamp." + let seq = state.coord.broker.current_seq(); match state.coord.broker.recent_all(HISTORY_LIMIT) { Ok(mut events) => { // recent_all returns newest-first; reverse so the replay // builds chronologically (matches the agent /events/history). events.reverse(); - axum::Json(events).into_response() + axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response() } Err(e) => error_response(&format!("messages/history failed: {e:#}")), } diff --git a/hive-fr0nt/assets/terminal.js b/hive-fr0nt/assets/terminal.js index 625af28a..11dc0512 100644 --- a/hive-fr0nt/assets/terminal.js +++ b/hive-fr0nt/assets/terminal.js @@ -166,36 +166,35 @@ } } - 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); - } - } + // 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 = []; - function subscribe() { 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 +205,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 }; } From fb669c17c8356048d306264c039761a9a06d7775 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 17 May 2026 12:28:04 +0200 Subject: [PATCH 3/9] dashboard: derive operator inbox from message stream (drop snapshot field + refetch workaround) --- hive-c0re/assets/app.js | 47 ++++++++++++++++++++++------------- hive-c0re/src/dashboard.rs | 15 +++-------- hive-fr0nt/assets/terminal.js | 10 +++++++- 3 files changed, 42 insertions(+), 30 deletions(-) diff --git a/hive-c0re/assets/app.js b/hive-c0re/assets/app.js index d9eafbf2..6e5558ca 100644 --- a/hive-c0re/assets/app.js +++ b/hive-c0re/assets/app.js @@ -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 + // `/messages/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)), ' ', @@ -932,7 +938,7 @@ renderContainers(s); renderTombstones(s); renderQuestions(s); - renderInbox(s); + renderInbox(); renderApprovals(s); renderMetaInputs(s); restoreOpenDetails(openDetails); @@ -994,10 +1000,17 @@ 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') { - refreshState(); NOTIF.show( '◆ ' + ev.from + ' → operator', String(ev.body || '').slice(0, 200), diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 22ebcbd3..d324137d 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -161,10 +161,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` @@ -323,13 +319,9 @@ 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)); @@ -344,7 +336,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, diff --git a/hive-fr0nt/assets/terminal.js b/hive-fr0nt/assets/terminal.js index 11dc0512..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,6 +168,10 @@ console.error('terminal renderer threw', ev, err); row('note', '[render err] ' + (err && err.message ? err.message : err)); } + if (opts.onAnyEvent) { + try { opts.onAnyEvent(ev, { fromHistory }); } + catch (err) { console.error('onAnyEvent threw', err); } + } } // Subscribe → buffer → fetch history → dedupe → apply. From d48cee7c2dd284a434bcd020eac2b1318a6ec024 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 17 May 2026 12:30:45 +0200 Subject: [PATCH 4/9] approvals: ship raw diff text instead of pre-rendered html; client classifies per-line --- TODO.md | 1 + hive-c0re/assets/app.js | 23 +++++++++++++++++---- hive-c0re/src/dashboard.rs | 42 ++++++++------------------------------ 3 files changed, 29 insertions(+), 37 deletions(-) diff --git a/TODO.md b/TODO.md index 257282a0..0decc12b 100644 --- a/TODO.md +++ b/TODO.md @@ -34,3 +34,4 @@ - ~~**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**: the enum is `#[serde(tag = "kind", rename_all = "snake_case")]` with `Note(String)` as a newtype variant — `serde_json::to_string` errors at runtime with `cannot serialize tagged newtype variant containing a string`. The SSE handler's `filter_map(... .ok()? ...)` silently drops the event; the sqlite history persists it as the literal string `"null"`. Every `bus.emit(LiveEvent::Note(...))` call site has been a no-op since the variant was added, and the JS terminal's `note` renderer is dead code. Fix: convert to a struct variant `Note { text: String }` (matches what the JS already reads via `ev.text`) and verify the existing call sites still type-check. While there, audit the sqlite-stored `"null"` rows so history-replay doesn't trip on them. diff --git a/hive-c0re/assets/app.js b/hive-c0re/assets/app.js index 6e5558ca..bafe380b 100644 --- a/hive-c0re/assets/app.js +++ b/hive-c0re/assets/app.js @@ -736,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);
       }
diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs
index d324137d..d6f06e04 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;
@@ -256,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,
@@ -639,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,
                 }
             }
@@ -648,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,
             },
         });
@@ -1345,29 +1349,6 @@ fn gc_orphans(coord: &Coordinator, approvals: Vec) -> 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
@@ -1415,8 +1396,3 @@ async fn git_diff_main_to(applied_dir: &Path, target_ref: &str) -> Result String {
-    s.replace('&', "&")
-        .replace('<', "<")
-        .replace('>', ">")
-}

From a47879291445336012e402035a66dccf8977b65f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?m=C3=BCde?= 
Date: Sun, 17 May 2026 12:39:48 +0200
Subject: [PATCH 5/9] dashboard events: unified coord channel +
 /dashboard/{stream,history}; broker forwards

---
 hive-c0re/assets/app.js           |  8 ++--
 hive-c0re/src/broker.rs           | 38 ++-------------
 hive-c0re/src/coordinator.rs      | 51 ++++++++++++++++++++
 hive-c0re/src/dashboard.rs        | 80 ++++++++++++++++++++-----------
 hive-c0re/src/dashboard_events.rs | 47 ++++++++++++++++++
 hive-c0re/src/main.rs             | 47 ++++++++++++++++++
 6 files changed, 205 insertions(+), 66 deletions(-)
 create mode 100644 hive-c0re/src/dashboard_events.rs

diff --git a/hive-c0re/assets/app.js b/hive-c0re/assets/app.js
index bafe380b..02a0f712 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 ────────────────────────────────────────────────────────────
@@ -600,7 +600,7 @@
   // ─── 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
-  // `/messages/history` populates on load, live SSE keeps it current.
+  // `/dashboard/history` populates on load, live SSE keeps it current.
   // Newest-first to match the previous behaviour.
   const INBOX_LIMIT = 50;
   const operatorInbox = [];
@@ -1009,8 +1009,8 @@
     }
     HiveTerminal.create({
       logEl: flow,
-      historyUrl: '/messages/history',
-      streamUrl: '/messages/stream',
+      historyUrl: '/dashboard/history',
+      streamUrl: '/dashboard/stream',
       renderers: {
         sent:      (ev, api) => renderMsg(ev, api, '→'),
         delivered: (ev, api) => renderMsg(ev, api, '✓'),
diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs
index 892acc50..c25d2c1b 100644
--- a/hive-c0re/src/broker.rs
+++ b/hive-c0re/src/broker.rs
@@ -3,7 +3,6 @@
 
 use std::path::Path;
 use std::sync::Mutex;
-use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use anyhow::{Context, Result};
@@ -47,18 +46,21 @@ 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 {
     Sent {
-        seq: u64,
         from: String,
         to: String,
         body: String,
         at: i64,
     },
     Delivered {
-        seq: u64,
         from: String,
         to: String,
         body: String,
@@ -69,13 +71,6 @@ pub enum MessageEvent {
 pub struct Broker {
     conn: Mutex,
     events: broadcast::Sender,
-    /// Monotonic per-process counter stamped onto every emitted
-    /// `MessageEvent`. Persisted nowhere — clients always treat a hive-c0re
-    /// restart as "everything is new" (fresh snapshot, fresh stream of
-    /// seqs starting at 1). Historical rows replayed via `recent_all`
-    /// carry `seq = 0` since they predate the live stream the seq is
-    /// meant to dedupe against.
-    event_seq: AtomicU64,
 }
 
 impl Broker {
@@ -91,7 +86,6 @@ impl Broker {
         Ok(Self {
             conn: Mutex::new(conn),
             events,
-            event_seq: AtomicU64::new(0),
         })
     }
 
@@ -99,20 +93,6 @@ impl Broker {
         self.events.subscribe()
     }
 
-    /// 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); any with seq <= snapshot.seq either pre-dates
-    /// the snapshot or was already captured by it. Clients dedupe their
-    /// buffered SSE traffic against this value.
-    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
-    }
-
     pub fn send(&self, message: &Message) -> Result<()> {
         let conn = self.conn.lock().unwrap();
         conn.execute(
@@ -121,7 +101,6 @@ impl Broker {
         )?;
         drop(conn);
         let _ = self.events.send(MessageEvent::Sent {
-            seq: self.next_seq(),
             from: message.from.clone(),
             to: message.to.clone(),
             body: message.body.clone(),
@@ -175,11 +154,6 @@ impl Broker {
         )?;
         let rows = stmt.query_map(params![limit_i], |row| {
             Ok(MessageEvent::Sent {
-                // Historical events: seq=0 (never compared against live
-                // seqs). Live dedupe windows close against
-                // history_seq = broker.current_seq() captured at fetch
-                // time, not against per-row seqs.
-                seq: 0,
                 from: row.get(0)?,
                 to: row.get(1)?,
                 body: row.get(2)?,
@@ -287,7 +261,6 @@ impl Broker {
         )?;
         drop(conn);
         let _ = self.events.send(MessageEvent::Delivered {
-            seq: self.next_seq(),
             from: from.clone(),
             to: to.clone(),
             body: body.clone(),
@@ -364,7 +337,6 @@ impl Broker {
         tx.commit()?;
         drop(conn);
         let _ = self.events.send(MessageEvent::Sent {
-            seq: self.next_seq(),
             from: "reminder".to_owned(),
             to: agent.to_owned(),
             body: message.to_owned(),
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 d6f06e04..90880b50 100644
--- a/hive-c0re/src/dashboard.rs
+++ b/hive-c0re/src/dashboard.rs
@@ -57,8 +57,8 @@ 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("/messages/history", get(messages_history))
+        .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));
@@ -73,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 —
@@ -293,13 +293,13 @@ 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 broker seq *before* any read so the dedupe contract
-    // is "events with seq > snapshot.seq are post-snapshot, never
-    // missed." A broker event landing during snapshot construction may
-    // be doubly applied (snapshot caught the write + client also
-    // applies the SSE event) — that's a renderer's problem to make
-    // idempotent, not ours to avoid here.
-    let seq = state.coord.broker.current_seq();
+    // 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);
@@ -720,36 +720,58 @@ fn dir_size_bytes(root: &Path) -> u64 {
     total
 }
 
-async fn messages_history(State(state): State) -> Response {
-    // Backfill source for the dashboard message-flow terminal. Returns
-    // up to ~200 historical broker messages as `MessageEvent::Sent` JSON
-    // wrapped in `{ seq, events }`. The seq is the broker'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 message
+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.
+    // twice and isn't lost. Historical rows carry `seq = 0`; the
+    // boundary seq is what closes the dedupe window.
     const HISTORY_LIMIT: u64 = 200;
-    // Capture seq *before* the query so the dedupe contract is
-    // "drop buffered events you've already seen in history" — never
-    // "lose an event that fired between the read and the timestamp."
-    let seq = state.coord.broker.current_seq();
+    let seq = state.coord.current_seq();
     match state.coord.broker.recent_all(HISTORY_LIMIT) {
-        Ok(mut events) => {
-            // recent_all returns newest-first; reverse so the replay
-            // builds chronologically (matches the agent /events/history).
-            events.reverse();
+        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!("messages/history failed: {e:#}")),
+        Err(e) => error_response(&format!("dashboard/history failed: {e:#}")),
     }
 }
 
-async fn messages_stream(
+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)))
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 {

From 616ca38199041798c1de6f438745bd8574bd9c33 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?m=C3=BCde?= 
Date: Sun, 17 May 2026 12:41:04 +0200
Subject: [PATCH 6/9] dashboard: /op-send returns 200; client relies on SSE for
 visual update

---
 hive-c0re/assets/app.js    | 9 +++++----
 hive-c0re/src/dashboard.rs | 8 +++++++-
 2 files changed, 12 insertions(+), 5 deletions(-)

diff --git a/hive-c0re/assets/app.js b/hive-c0re/assets/app.js
index 02a0f712..06f5e108 100644
--- a/hive-c0re/assets/app.js
+++ b/hive-c0re/assets/app.js
@@ -1143,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/src/dashboard.rs b/hive-c0re/src/dashboard.rs
index 90880b50..4e78acde 100644
--- a/hive-c0re/src/dashboard.rs
+++ b/hive-c0re/src/dashboard.rs
@@ -1141,7 +1141,13 @@ async fn post_op_send(State(state): State, Form(form): Form
Date: Sun, 17 May 2026 12:41:37 +0200
Subject: [PATCH 7/9] agent: /send returns 200 (terminal + turn-end refresh
 already cover the visual update)

---
 hive-ag3nt/src/web_ui.rs | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs
index 941eb298..69f22f39 100644
--- a/hive-ag3nt/src/web_ui.rs
+++ b/hive-ag3nt/src/web_ui.rs
@@ -343,7 +343,13 @@ 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}")),
     }
 }

From b60774a66cf5061a1db34ef3e6f5f092d279af70 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?m=C3=BCde?= 
Date: Sun, 17 May 2026 13:14:09 +0200
Subject: [PATCH 8/9] events: LiveEvent::Note becomes struct variant so serde
 can actually serialize it

---
 TODO.md                         |  2 +-
 hive-ag3nt/src/bin/hive-m1nd.rs |  4 +++-
 hive-ag3nt/src/events.rs        | 13 +++++++++++--
 hive-ag3nt/src/turn.rs          | 24 +++++++++++++++---------
 hive-ag3nt/src/web_ui.rs        | 32 ++++++++++++++++----------------
 5 files changed, 46 insertions(+), 29 deletions(-)

diff --git a/TODO.md b/TODO.md
index 0decc12b..cce021ea 100644
--- a/TODO.md
+++ b/TODO.md
@@ -34,4 +34,4 @@
 
 - ~~**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**: the enum is `#[serde(tag = "kind", rename_all = "snake_case")]` with `Note(String)` as a newtype variant — `serde_json::to_string` errors at runtime with `cannot serialize tagged newtype variant containing a string`. The SSE handler's `filter_map(... .ok()? ...)` silently drops the event; the sqlite history persists it as the literal string `"null"`. Every `bus.emit(LiveEvent::Note(...))` call site has been a no-op since the variant was added, and the JS terminal's `note` renderer is dead code. Fix: convert to a struct variant `Note { text: String }` (matches what the JS already reads via `ev.text`) and verify the existing call sites still type-check. While there, audit the sqlite-stored `"null"` rows so history-replay doesn't trip on them.
+- ~~**`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 e8944836..8af3e049 100644
--- a/hive-ag3nt/src/events.rs
+++ b/hive-ag3nt/src/events.rs
@@ -105,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 },
@@ -138,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());
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 69f22f39..44243abe 100644
--- a/hive-ag3nt/src/web_ui.rs
+++ b/hive-ag3nt/src/web_ui.rs
@@ -372,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()?;
@@ -448,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()
@@ -501,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()
 }
 
@@ -524,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()
 }
 

From 87f8f8a123b1560c50c84ed7d26f8b3233630540 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?m=C3=BCde?= 
Date: Sun, 17 May 2026 13:15:32 +0200
Subject: [PATCH 9/9] =?UTF-8?q?todo:=20phase=205b=20=E2=80=94=20mutation?=
 =?UTF-8?q?=20events=20for=20approvals/questions/transients?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 TODO.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/TODO.md b/TODO.md
index cce021ea..3fa31eb4 100644
--- a/TODO.md
+++ b/TODO.md
@@ -29,6 +29,7 @@
 - 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