vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Compare commits

base: vinzenz:87f8f8a123b1560c50c84ed7d26f8b3233630540
Branches Tags
vinzenz:main
..
compare: vinzenz:f27108aecfb8206b51c7b23accc536b0c6a862c8
Branches Tags
vinzenz:main
14 changed files with 212 additions and 590 deletions
Download patch file Download diff file Expand all files Collapse all files

2
TODO.md
View file

@ -29,10 +29,8 @@
- Per-agent reminder status (pending, delivered) - Per-agent reminder status (pending, delivered)
- Reminder query interface for debugging - Reminder query interface for debugging
- Display reminder delivery errors (failed sends, mark failures) - 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 ## 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. - ~~**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. - **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.

4
hive-ag3nt/src/bin/hive-m1nd.rs
View file

@ -136,9 +136,7 @@ async fn serve(
} else { } else {
tracing::info!(%from, %body, "system message"); tracing::info!(%from, %body, "system message");
} }
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note(format!("[system] {body}")));
text: format!("[system] {body}"),
});
// Fall through: drive a turn with the event in the wake // Fall through: drive a turn with the event in the wake
// prompt body so claude sees it. Sender stays "system" // prompt body so claude sees it. Sender stays "system"
// so the wake prompt can label it as such. // so the wake prompt can label it as such.

58
hive-ag3nt/src/events.rs
View file

@ -9,7 +9,7 @@
//! showing "connecting…" until the first event arrives. //! showing "connecting…" until the first event arrives.
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use rusqlite::{Connection, params}; use rusqlite::{Connection, params};
@ -74,18 +74,6 @@ CREATE TABLE IF NOT EXISTS events (
CREATE INDEX IF NOT EXISTS idx_events_ts ON events (ts); 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. /// One row of the agent's live stream. Serialised to JSON for SSE delivery.
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")] #[serde(tag = "kind", rename_all = "snake_case")]
@ -105,16 +93,7 @@ pub enum LiveEvent {
/// Free-form note from the harness (e.g. "claude exited 0", /// Free-form note from the harness (e.g. "claude exited 0",
/// "stream-json parse error: ..."). Useful when stream-json itself /// "stream-json parse error: ..."). Useful when stream-json itself
/// fails so the UI doesn't just go silent. /// 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 /// Turn finished. `ok=false` means claude exited non-zero or the
/// harness hit a transport error. /// harness hit a transport error.
TurnEnd { ok: bool, note: Option<String> }, TurnEnd { ok: bool, note: Option<String> },
@ -147,7 +126,7 @@ impl EventStore {
let kind = match event { let kind = match event {
LiveEvent::TurnStart { .. } => "turn_start", LiveEvent::TurnStart { .. } => "turn_start",
LiveEvent::Stream(_) => "stream", LiveEvent::Stream(_) => "stream",
LiveEvent::Note { .. } => "note", LiveEvent::Note(_) => "note",
LiveEvent::TurnEnd { .. } => "turn_end", LiveEvent::TurnEnd { .. } => "turn_end",
}; };
let payload = serde_json::to_string(event).unwrap_or_else(|_| "null".into()); let payload = serde_json::to_string(event).unwrap_or_else(|_| "null".into());
@ -237,13 +216,7 @@ pub const DEFAULT_MODEL: &str = "haiku";
#[derive(Clone)] #[derive(Clone)]
pub struct Bus { pub struct Bus {
tx: Arc<broadcast::Sender<BusEvent>>, tx: Arc<broadcast::Sender<LiveEvent>>,
/// 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 /// Persistent event log. `None` only if opening the sqlite db failed
/// at construction — we keep going so the harness doesn't die on a /// at construction — we keep going so the harness doesn't die on a
/// missing state dir mount in dev / test scenarios. /// missing state dir mount in dev / test scenarios.
@ -285,7 +258,6 @@ impl Bus {
let initial_model = load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()); let initial_model = load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned());
Self { Self {
tx: Arc::new(tx), tx: Arc::new(tx),
event_seq: Arc::new(AtomicU64::new(0)),
store, store,
state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))), state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))),
model: Arc::new(Mutex::new(initial_model)), model: Arc::new(Mutex::new(initial_model)),
@ -294,20 +266,6 @@ 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 /// Arm the one-shot: the next claude invocation will run without
/// `--continue`, dropping any prior session context. Idempotent /// `--continue`, dropping any prior session context. Idempotent
/// — calling twice in a row before the next turn still consumes /// — calling twice in a row before the next turn still consumes
@ -375,15 +333,11 @@ impl Bus {
{ {
tracing::warn!(error = ?e, "events: append failed"); 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. // Lagged subscribers drop events — fine; the UI is a tail, not a log.
let _ = self.tx.send(envelope); let _ = self.tx.send(event);
} }
pub fn subscribe(&self) -> broadcast::Receiver<BusEvent> { pub fn subscribe(&self) -> broadcast::Receiver<LiveEvent> {
self.tx.subscribe() self.tx.subscribe()
} }

24
hive-ag3nt/src/turn.rs
View file

@ -206,13 +206,11 @@ pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome
/// compact state matches a normal turn's. Only the prompt over stdin /// compact state matches a normal turn's. Only the prompt over stdin
/// differs (`/compact` vs the wake-up payload). /// differs (`/compact` vs the wake-up payload).
pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> Result<()> { pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> Result<()> {
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note(
text: "context overflow — running /compact on the persistent session".into(), "context overflow — running /compact on the persistent session".into(),
}); ));
let _ = run_claude("/compact", files, bus).await?; let _ = run_claude("/compact", files, bus).await?;
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note("/compact done".into()));
text: "/compact done".into(),
});
Ok(()) Ok(())
} }
@ -220,9 +218,9 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<bool>
let model = bus.model(); let model = bus.model();
let resume = !bus.take_skip_continue(); let resume = !bus.take_skip_continue();
if !resume { if !resume {
bus.emit(LiveEvent::Note { bus.emit(LiveEvent::Note(
text: "fresh session (--continue suppressed for this turn)".into(), "fresh session (--continue suppressed for this turn)".into(),
}); ));
} }
let mut cmd = Command::new("claude"); let mut cmd = Command::new("claude");
// Spawn inside the agent's state dir so relative paths in tool calls // Spawn inside the agent's state dir so relative paths in tool calls
@ -284,9 +282,7 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<bool>
} }
bus_out.emit(LiveEvent::Stream(v)); bus_out.emit(LiveEvent::Stream(v));
} }
Err(_) => bus_out.emit(LiveEvent::Note { Err(_) => bus_out.emit(LiveEvent::Note(format!("(non-json) {line}"))),
text: format!("(non-json) {line}"),
}),
} }
} }
}); });
@ -308,9 +304,7 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<bool>
// renders; the tracing line is what `journalctl -M <c> -b` // renders; the tracing line is what `journalctl -M <c> -b`
// surfaces when claude exits non-zero. // surfaces when claude exits non-zero.
tracing::warn!(line = %line, "claude stderr"); tracing::warn!(line = %line, "claude stderr");
bus_err.emit(LiveEvent::Note { bus_err.emit(LiveEvent::Note(format!("stderr: {line}")));
text: format!("stderr: {line}"),
});
let mut t = tail_clone.lock().unwrap(); let mut t = tail_clone.lock().unwrap();
if t.len() >= STDERR_TAIL_LINES { if t.len() >= STDERR_TAIL_LINES {
t.pop_front(); t.pop_front();

63
hive-ag3nt/src/web_ui.rs
View file

@ -191,12 +191,6 @@ async fn serve_shared_js() -> impl IntoResponse {
#[derive(Serialize)] #[derive(Serialize)]
struct StateSnapshot { 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, label: String,
dashboard_port: u16, dashboard_port: u16,
/// `"online"` | `"needs_login_idle"` | `"needs_login_in_progress"`. /// `"online"` | `"needs_login_idle"` | `"needs_login_in_progress"`.
@ -232,9 +226,6 @@ struct SessionView {
} }
async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> { 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); drop_if_finished(&state.session);
let login = *state.login.lock().unwrap(); let login = *state.login.lock().unwrap();
let session_snapshot = state.session.lock().unwrap().clone(); let session_snapshot = state.session.lock().unwrap().clone();
@ -260,7 +251,6 @@ async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
let model = state.bus.model(); let model = state.bus.model();
let token_usage = state.bus.last_usage(); let token_usage = state.bus.last_usage();
axum::Json(StateSnapshot { axum::Json(StateSnapshot {
seq,
label: state.label.clone(), label: state.label.clone(),
dashboard_port, dashboard_port,
status, status,
@ -343,26 +333,15 @@ async fn post_send(State(state): State<AppState>, Form(form): Form<SendForm>) ->
}, },
}; };
match result { match result {
// 200 instead of 303 → the client doesn't refetch /api/state. Ok(()) => Redirect::to("/").into_response(),
// 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}")), Err(e) => error_response(&format!("send failed: {e}")),
} }
} }
async fn events_history(State(state): State<AppState>) -> axum::Json<serde_json::Value> { async fn events_history(
// Capture seq *before* the read so dedupe is "drop buffered events State(state): State<AppState>,
// you've already seen in history", never "lose an event that fired ) -> axum::Json<Vec<crate::events::LiveEvent>> {
// between the read and the timestamp." Historical rows have no axum::Json(state.bus.history())
// 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( async fn events_stream(
@ -372,9 +351,9 @@ async fn events_stream(
let rx = state.bus.subscribe(); let rx = state.bus.subscribe();
// Drop a "hello" note into the bus so every new subscriber sees at // Drop a "hello" note into the bus so every new subscriber sees at
// least one event immediately and can clear the connecting placeholder. // least one event immediately and can clear the connecting placeholder.
state.bus.emit(crate::events::LiveEvent::Note { state.bus.emit(crate::events::LiveEvent::Note(
text: "live stream attached".into(), "live stream attached".into(),
}); ));
let stream = BroadcastStream::new(rx).filter_map(|res| { let stream = BroadcastStream::new(rx).filter_map(|res| {
let ev = res.ok()?; let ev = res.ok()?;
let json = serde_json::to_string(&ev).ok()?; let json = serde_json::to_string(&ev).ok()?;
@ -448,9 +427,9 @@ async fn post_set_model(State(state): State<AppState>, Form(form): Form<ModelFor
return error_response("model: name required"); return error_response("model: name required");
} }
state.bus.set_model(name); state.bus.set_model(name);
state.bus.emit(crate::events::LiveEvent::Note { state.bus.emit(crate::events::LiveEvent::Note(format!(
text: format!("operator: /model — claude model set to '{name}' for future turns"), "operator: /model — claude model set to '{name}' for future turns"
}); )));
tracing::info!(%name, "operator set model"); tracing::info!(%name, "operator set model");
Redirect::to("/").into_response() Redirect::to("/").into_response()
} }
@ -471,16 +450,16 @@ async fn post_compact(State(state): State<AppState>) -> Response {
let files = state.files.clone(); let files = state.files.clone();
tokio::spawn(async move { tokio::spawn(async move {
let _guard = guard; // keep lock alive for the duration of compaction let _guard = guard; // keep lock alive for the duration of compaction
bus.emit(crate::events::LiveEvent::Note { bus.emit(crate::events::LiveEvent::Note(
text: "operator: /compact — running on persistent session".into(), "operator: /compact — running on persistent session".into(),
}); ));
bus.set_state(crate::events::TurnState::Compacting); bus.set_state(crate::events::TurnState::Compacting);
let r = crate::turn::compact_session(&files, &bus).await; let r = crate::turn::compact_session(&files, &bus).await;
bus.set_state(crate::events::TurnState::Idle); bus.set_state(crate::events::TurnState::Idle);
if let Err(e) = r { if let Err(e) = r {
bus.emit(crate::events::LiveEvent::Note { bus.emit(crate::events::LiveEvent::Note(format!(
text: format!("/compact failed: {e:#}"), "/compact failed: {e:#}"
}); )));
} }
}); });
Redirect::to("/").into_response() Redirect::to("/").into_response()
@ -501,9 +480,9 @@ async fn post_compact(State(state): State<AppState>) -> Response {
/// than asking claude to forget mid-stream. /// than asking claude to forget mid-stream.
async fn post_new_session(State(state): State<AppState>) -> Response { async fn post_new_session(State(state): State<AppState>) -> Response {
state.bus.request_new_session(); state.bus.request_new_session();
state.bus.emit(crate::events::LiveEvent::Note { state.bus.emit(crate::events::LiveEvent::Note(
text: "operator: new session armed — next turn runs without --continue".into(), "operator: new session armed — next turn runs without --continue".into(),
}); ));
Redirect::to("/").into_response() Redirect::to("/").into_response()
} }
@ -524,7 +503,7 @@ async fn post_cancel_turn(State(state): State<AppState>) -> Response {
), ),
Err(e) => format!("operator: /cancel — pkill failed: {e}"), Err(e) => format!("operator: /cancel — pkill failed: {e}"),
}; };
state.bus.emit(crate::events::LiveEvent::Note { text: note }); state.bus.emit(crate::events::LiveEvent::Note(note));
Redirect::to("/").into_response() Redirect::to("/").into_response()
} }

159
hive-c0re/assets/app.js
View file

@ -1,6 +1,6 @@
// Dashboard SPA. Renders containers + approvals from `/api/state`, wires // Dashboard SPA. Renders containers + approvals from `/api/state`, wires
// up async-form submission (URL-encoded POST + spinner + state refresh), // up async-form submission (URL-encoded POST + spinner + state refresh),
// and tails the unified dashboard event channel over `/dashboard/stream`. // and tails the broker over `/messages/stream` SSE.
(() => { (() => {
// ─── helpers ──────────────────────────────────────────────────────────── // ─── helpers ────────────────────────────────────────────────────────────
@ -118,20 +118,20 @@
// when the page reloads. // when the page reloads.
const seenApprovals = new Set(); const seenApprovals = new Set();
const seenQuestions = new Set(); const seenQuestions = new Set();
const seenInboxIds = new Set();
let seededNotify = false; let seededNotify = false;
function notifyDeltas(s) { function notifyDeltas(s) {
const approvals = s.approvals || []; const approvals = s.approvals || [];
const questions = s.questions || []; const questions = s.questions || [];
const inbox = s.operator_inbox || [];
if (!seededNotify) { if (!seededNotify) {
// First render after page load — fill the "seen" sets without // First render after page load — fill the "seen" sets without
// firing notifications. We only want to notify on NEW items // firing notifications. We only want to notify on NEW items
// that arrived while the page is open. The inbox no longer // that arrived while the page is open.
// 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 a of approvals) seenApprovals.add(a.id);
for (const q of questions) seenQuestions.add(q.id); for (const q of questions) seenQuestions.add(q.id);
for (const m of inbox) seenInboxIds.add(m.id);
seededNotify = true; seededNotify = true;
return; return;
} }
@ -148,6 +148,14 @@
NOTIF.show('◆ manager asks', q.question.slice(0, 120), NOTIF.show('◆ manager asks', q.question.slice(0, 120),
'hyperhive:question:' + q.id); '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 ──────────────────────────────────────────────────────── // ─── async forms ────────────────────────────────────────────────────────
@ -597,30 +605,16 @@
} }
} }
// ─── operator inbox (derived from the broker message stream) ─────────── function renderInbox(s) {
// 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'); const root = $('inbox-section');
if (!root) return;
root.innerHTML = ''; root.innerHTML = '';
if (!operatorInbox.length) { if (!s.operator_inbox || !s.operator_inbox.length) {
root.append(el('p', { class: 'empty' }, 'no messages')); root.append(el('p', { class: 'empty' }, 'no messages'));
return; return;
} }
const fmt = (n) => new Date(n * 1000).toISOString().replace('T', ' ').slice(0, 19); const fmt = (n) => new Date(n * 1000).toISOString().replace('T', ' ').slice(0, 19);
const ul = el('ul', { class: 'inbox' }); const ul = el('ul', { class: 'inbox' });
for (const m of operatorInbox) { for (const m of s.operator_inbox) {
const li = el('li'); const li = el('li');
li.append( li.append(
el('span', { class: 'msg-ts' }, fmt(m.at)), ' ', el('span', { class: 'msg-ts' }, fmt(m.at)), ' ',
@ -736,29 +730,14 @@
denyForm, denyForm,
); );
li.append(row); li.append(row);
if (a.diff) { if (a.diff_html) {
const details = el('details', { const details = el('details', {
'data-restore-key': 'approval-diff:' + a.id, 'data-restore-key': 'approval-diff:' + a.id,
}); });
details.append(el('summary', {}, 'diff vs applied')); details.append(el('summary', {}, 'diff vs applied'));
// Server ships the raw unified diff; classify each line by its // diff_html is pre-rendered server-side (per-line class spans inside
// leading char so `.diff-add` / `.diff-del` / `.diff-hunk` / // a <pre>); inject as innerHTML.
// `.diff-file` / `.diff-ctx` colour the output. Building spans const pre = el('pre', { class: 'diff', html: a.diff_html });
// 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); details.append(pre);
li.append(details); li.append(details);
} }
@ -953,7 +932,7 @@
renderContainers(s); renderContainers(s);
renderTombstones(s); renderTombstones(s);
renderQuestions(s); renderQuestions(s);
renderInbox(); renderInbox(s);
renderApprovals(s); renderApprovals(s);
renderMetaInputs(s); renderMetaInputs(s);
restoreOpenDetails(openDetails); restoreOpenDetails(openDetails);
@ -976,19 +955,17 @@
refreshState(); refreshState();
NOTIF.bind(); NOTIF.bind();
// ─── message flow: shared terminal pane ──────────────────────────────── // ─── message flow SSE ───────────────────────────────────────────────────
// 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'); const flow = $('msgflow');
if (!flow || !window.HiveTerminal) return; if (!flow) return;
flow.innerHTML = ''; flow.innerHTML = '';
const es = new EventSource('/messages/stream');
const MAX_ROWS = 200;
const tsFmt = (n) => new Date(n * 1000).toISOString().slice(11, 19); const tsFmt = (n) => new Date(n * 1000).toISOString().slice(11, 19);
// Pulse the page banner whenever a broker event lands. Each event // Animate the banner whenever a broker event lands. Each event nudges
// nudges the shimmer window; if traffic stops, the class falls off // the shimmer window; if traffic stops, the class falls off after the
// after the grace timer. // grace timer.
const banner = document.querySelector('.banner'); const banner = document.querySelector('.banner');
let bannerOffTimer = null; let bannerOffTimer = null;
function pulseBanner() { function pulseBanner() {
@ -997,45 +974,40 @@
if (bannerOffTimer) clearTimeout(bannerOffTimer); if (bannerOffTimer) clearTimeout(bannerOffTimer);
bannerOffTimer = setTimeout(() => banner.classList.remove('active'), 4000); bannerOffTimer = setTimeout(() => banner.classList.remove('active'), 4000);
} }
function renderMsg(ev, api, glyph) { es.onmessage = (e) => {
const el = api.row('msgrow ' + ev.kind, ''); let m;
el.innerHTML = try { m = JSON.parse(e.data); } catch { return; }
'<span class="msg-ts">' + tsFmt(ev.at) + '</span>' + pulseBanner();
'<span class="msg-arrow">' + glyph + '</span>' + // Live-update the inbox when claude sends to operator + ping
'<span class="msg-from">' + esc(ev.from) + '</span>' + // 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>' +
'<span class="msg-sep">→</span>' + '<span class="msg-sep">→</span>' +
'<span class="msg-to">' + esc(ev.to) + '</span>' + '<span class="msg-to">' + esc(m.to) + '</span>' +
'<span class="msg-body">' + esc(ev.body) + '</span>'; '<span class="msg-body">' + esc(m.body) + '</span>';
} flow.insertBefore(row, flow.firstChild);
HiveTerminal.create({ while (flow.childNodes.length > MAX_ROWS) flow.removeChild(flow.lastChild);
logEl: flow, };
historyUrl: '/dashboard/history', es.onerror = () => {
streamUrl: '/dashboard/stream', flow.insertBefore(Object.assign(document.createElement('div'), {
renderers: { className: 'msgrow meta', textContent: '[connection lost — retrying]',
sent: (ev, api) => renderMsg(ev, api, '→'), }), flow.firstChild);
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 ─────────────────────────── // ─── compose: @-mention with sticky recipient ───────────────────────────
@ -1143,15 +1115,14 @@
fd.append('body', body); fd.append('body', body);
input.disabled = true; input.disabled = true;
try { 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', { const resp = await fetch('/op-send', {
method: 'POST', method: 'POST',
body: new URLSearchParams(fd), body: new URLSearchParams(fd),
redirect: 'manual',
}); });
if (!resp.ok) { const ok = resp.ok || resp.type === 'opaqueredirect'
|| (resp.status >= 200 && resp.status < 400);
if (!ok) {
flashError(`send failed: http ${resp.status}`); flashError(`send failed: http ${resp.status}`);
return; return;
} }

35
hive-c0re/assets/dashboard.css
View file

@ -537,28 +537,43 @@ summary:hover { color: var(--purple); }
.inbox .msg-from { color: var(--amber); } .inbox .msg-from { color: var(--amber); }
.inbox .msg-sep { color: var(--muted); } .inbox .msg-sep { color: var(--muted); }
.inbox .msg-body { color: var(--fg); white-space: pre-wrap; word-break: break-word; } .inbox .msg-body { color: var(--fg); white-space: pre-wrap; word-break: break-word; }
/* `#msgflow` is a shared `.live` pane inside `.terminal-wrap` (see .msgflow {
hive-fr0nt::TERMINAL_CSS). The msgrow / msg-* rules below are background: rgba(24, 24, 37, 0.78);
dashboard-specific: each broker event becomes a grid of timestamp + -webkit-backdrop-filter: blur(8px) saturate(120%);
arrow + from/sep/to + body inside the `.row` shell. */ backdrop-filter: blur(8px) saturate(120%);
.live .msgrow { display: grid; grid-template-columns: auto auto auto auto auto 1fr; gap: 0.6em; align-items: baseline; padding: 0.1em 0; } border: 1px solid var(--border);
.live .msgrow.sent .msg-arrow { color: var(--cyan); } padding: 0.8em;
.live .msgrow.delivered .msg-arrow { color: var(--green); } 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); }
.msg-ts { color: var(--muted); font-size: 0.85em; } .msg-ts { color: var(--muted); font-size: 0.85em; }
.msg-arrow { font-weight: bold; } .msg-arrow { font-weight: bold; }
.msg-from { color: var(--amber); } .msg-from { color: var(--amber); }
.msg-sep { color: var(--muted); } .msg-sep { color: var(--muted); }
.msg-to { color: var(--pink); } .msg-to { color: var(--pink); }
.msg-body { color: var(--fg); white-space: pre-wrap; word-break: break-word; } .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 { .op-compose {
position: relative; position: relative;
display: flex; display: flex;
align-items: flex-start; align-items: flex-start;
gap: 0.6em; gap: 0.6em;
margin-top: 0.4em;
padding: 0.55em 0.8em; padding: 0.55em 0.8em;
border-top: 1px dashed var(--purple-dim); background: rgba(24, 24, 37, 0.85);
border: 1px solid var(--border);
border-top: none;
} }
.op-compose-prompt { .op-compose-prompt {
color: var(--purple); color: var(--purple);

17
hive-c0re/assets/index.html
View file

@ -61,15 +61,13 @@
<h2>◆ MESS4GE FL0W ◆</h2> <h2>◆ MESS4GE FL0W ◆</h2>
<div class="divider">══════════════════════════════════════════════════════════════</div> <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> <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 class="terminal-wrap"> <div id="msgflow" class="msgflow"><span class="meta">connecting…</span></div>
<div id="msgflow" class="live"><div class="meta">connecting…</div></div> <div id="op-compose" class="op-compose">
<div id="op-compose" class="op-compose"> <span id="op-compose-prompt" class="op-compose-prompt">@—&gt;</span>
<span id="op-compose-prompt" class="op-compose-prompt">@—&gt;</span> <textarea id="op-compose-input" class="op-compose-input"
<textarea id="op-compose-input" class="op-compose-input" placeholder="@agent message… (enter sends, shift+enter newline, tab completes @-mention)"
placeholder="@agent message… (enter sends, shift+enter newline, tab completes @-mention)" rows="1" autocomplete="off"></textarea>
rows="1" autocomplete="off"></textarea> <div id="op-compose-suggest" class="op-compose-suggest" hidden></div>
<div id="op-compose-suggest" class="op-compose-suggest" hidden></div>
</div>
</div> </div>
<footer> <footer>
@ -77,7 +75,6 @@
<p>▲△▲ <a href="https://git.berlin.ccc.de/vinzenz/hyperhive">hyperhive</a> ▲△▲ hive-c0re on this host ▲△▲</p> <p>▲△▲ <a href="https://git.berlin.ccc.de/vinzenz/hyperhive">hyperhive</a> ▲△▲ hive-c0re on this host ▲△▲</p>
</footer> </footer>
<script src="/static/hive-fr0nt.js" defer></script>
<script src="/static/app.js" defer></script> <script src="/static/app.js" defer></script>
</body> </body>
</html> </html>

35
hive-c0re/src/broker.rs
View file

@ -46,11 +46,6 @@ const EVENT_CHANNEL: usize = 256;
/// self-documenting. /// self-documenting.
pub type DueReminder = (String, i64, String, Option<String>); 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)] #[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind")] #[serde(rename_all = "snake_case", tag = "kind")]
pub enum MessageEvent { pub enum MessageEvent {
@ -134,36 +129,6 @@ impl Broker {
.map_err(Into::into) .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 /// Number of undelivered messages addressed to `recipient`. Non-mutating
/// — used by the harness to surface "N unread" in tool-result status /// — used by the harness to surface "N unread" in tool-result status
/// lines without popping the queue. /// lines without popping the queue.

51
hive-c0re/src/coordinator.rs
View file

@ -4,23 +4,15 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use tokio::sync::broadcast;
use crate::agent_server::{self, AgentSocket}; use crate::agent_server::{self, AgentSocket};
use crate::approvals::Approvals; use crate::approvals::Approvals;
use crate::broker::Broker; use crate::broker::Broker;
use crate::dashboard_events::DashboardEvent;
use crate::operator_questions::OperatorQuestions; 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 AGENT_RUNTIME_ROOT: &str = "/run/hyperhive/agents";
const MANAGER_RUNTIME_ROOT: &str = "/run/hyperhive/manager"; const MANAGER_RUNTIME_ROOT: &str = "/run/hyperhive/manager";
/// Manager-editable per-agent config repos. Bind-mounted RW into the manager /// Manager-editable per-agent config repos. Bind-mounted RW into the manager
@ -55,15 +47,6 @@ pub struct Coordinator {
/// Read by the dashboard to render a spinner; cleared when the action /// Read by the dashboard to render a spinner; cleared when the action
/// resolves (success or failure). /// resolves (success or failure).
transient: Mutex<HashMap<String, TransientState>>, 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 /// Per-agent in-progress state that the dashboard surfaces between approve
@ -115,7 +98,6 @@ impl Coordinator {
let broker = Broker::open(db_path).context("open broker")?; let broker = Broker::open(db_path).context("open broker")?;
let approvals = Approvals::open(db_path).context("open approvals")?; let approvals = Approvals::open(db_path).context("open approvals")?;
let questions = OperatorQuestions::open(db_path).context("open operator_questions")?; let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
Ok(Self { Ok(Self {
broker: Arc::new(broker), broker: Arc::new(broker),
approvals: Arc::new(approvals), approvals: Arc::new(approvals),
@ -125,42 +107,9 @@ impl Coordinator {
operator_pronouns, operator_pronouns,
agents: Mutex::new(HashMap::new()), agents: Mutex::new(HashMap::new()),
transient: 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> { pub fn register_agent(self: &Arc<Self>, name: &str) -> Result<PathBuf> {
// Idempotent: drop any existing listener so re-registration (e.g. on rebuild, // 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. // or after a hive-c0re restart cleared /run/hyperhive) gets a fresh socket.

147
hive-c0re/src/dashboard.rs
View file

@ -3,6 +3,7 @@
//! repo, plus approve/deny buttons), and the manager. //! repo, plus approve/deny buttons), and the manager.
use std::convert::Infallible; use std::convert::Infallible;
use std::fmt::Write as _;
use std::net::SocketAddr; use std::net::SocketAddr;
use std::path::Path; use std::path::Path;
use std::sync::Arc; use std::sync::Arc;
@ -57,9 +58,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
.route("/request-spawn", post(post_request_spawn)) .route("/request-spawn", post(post_request_spawn))
.route("/op-send", post(post_op_send)) .route("/op-send", post(post_op_send))
.route("/meta-update", post(post_meta_update)) .route("/meta-update", post(post_meta_update))
.route("/dashboard/stream", get(dashboard_stream)) .route("/messages/stream", get(messages_stream))
.route("/dashboard/history", get(dashboard_history))
.route("/static/hive-fr0nt.js", get(serve_shared_js))
.with_state(AppState { coord }); .with_state(AppState { coord });
let addr = SocketAddr::from(([0, 0, 0, 0], port)); let addr = SocketAddr::from(([0, 0, 0, 0], port));
let listener = bind_with_retry(addr).await?; let listener = bind_with_retry(addr).await?;
@ -73,7 +72,7 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
// (static) shell; `GET /static/*` serves the CSS + JS app; `GET /api/state` // (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, // returns the current snapshot as JSON. The JS app fetches state on load,
// re-fetches after every async-form submit, and listens on // re-fetches after every async-form submit, and listens on
// `/dashboard/stream` for the unified live event channel. // `/messages/stream` for broker traffic.
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
/// `SO_REUSEADDR` bind with retry. Mirrors the per-agent variant — /// `SO_REUSEADDR` bind with retry. Mirrors the per-agent variant —
@ -134,23 +133,8 @@ async fn serve_app_js() -> impl IntoResponse {
) )
} }
async fn serve_shared_js() -> impl IntoResponse {
(
[("content-type", "application/javascript")],
hive_fr0nt::TERMINAL_JS,
)
}
#[derive(Serialize)] #[derive(Serialize)]
struct StateSnapshot { 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, hostname: String,
manager_port: u16, manager_port: u16,
any_stale: bool, any_stale: bool,
@ -160,6 +144,10 @@ struct StateSnapshot {
/// Last 30 resolved approvals (approved / denied / failed), newest- /// Last 30 resolved approvals (approved / denied / failed), newest-
/// first. Drives the "history" tab on the approvals section. /// first. Drives the "history" tab on the approvals section.
approval_history: Vec<ApprovalHistoryView>, 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). /// Pending operator questions (currently only from the manager).
/// `ask_operator` returns immediately with the id; on `/answer-question` /// `ask_operator` returns immediately with the id; on `/answer-question`
/// we mark the row answered and fire `HelperEvent::OperatorAnswered` /// we mark the row answered and fire `HelperEvent::OperatorAnswered`
@ -255,13 +243,8 @@ struct ApprovalView {
kind: &'static str, kind: &'static str,
/// First 12 chars of the `commit_ref`, for `ApplyCommit` only. /// First 12 chars of the `commit_ref`, for `ApplyCommit` only.
sha_short: Option<String>, sha_short: Option<String>,
/// Raw unified diff text, for `ApplyCommit` only. The client splits /// Pre-rendered syntax-coloured diff HTML, for `ApplyCommit` only.
/// on `\n` and per-line classifies (`+` / `-` / `@@` / `--- ` / `+++ ` diff_html: Option<String>,
/// → 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. /// Manager-supplied description shown on the approval card.
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
description: Option<String>, description: Option<String>,
@ -293,14 +276,6 @@ async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::J
.unwrap_or("localhost"); .unwrap_or("localhost");
let hostname = host.split(':').next().unwrap_or(host).to_owned(); 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 raw_containers = log_default("nixos-container list", lifecycle::list().await);
let current_rev = crate::auto_update::current_flake_rev(&state.coord.hyperhive_flake); let current_rev = crate::auto_update::current_flake_rev(&state.coord.hyperhive_flake);
let transient_snapshot = state.coord.transient_snapshot(); let transient_snapshot = state.coord.transient_snapshot();
@ -323,15 +298,18 @@ async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::J
let tombstones = build_tombstone_views(&state.coord, &containers, &transient_snapshot); let tombstones = build_tombstone_views(&state.coord, &containers, &transient_snapshot);
let port_conflicts = build_port_conflicts(&containers); let port_conflicts = build_port_conflicts(&containers);
// operator_inbox used to be served here as a 50-row array; the let operator_inbox = log_default(
// dashboard now derives it client-side from the message stream "broker.recent_for(operator)",
// (terminal backfill + live SSE), so the snapshot stops shipping it. state
.coord
.broker
.recent_for(hive_sh4re::OPERATOR_RECIPIENT, 50),
);
let questions = log_default("questions.pending", state.coord.questions.pending()); let questions = log_default("questions.pending", state.coord.questions.pending());
let question_history = let question_history =
log_default("questions.recent_answered", state.coord.questions.recent_answered(20)); log_default("questions.recent_answered", state.coord.questions.recent_answered(20));
axum::Json(StateSnapshot { axum::Json(StateSnapshot {
seq,
hostname, hostname,
manager_port: MANAGER_PORT, manager_port: MANAGER_PORT,
any_stale, any_stale,
@ -340,6 +318,7 @@ async fn api_state(headers: HeaderMap, State(state): State<AppState>) -> axum::J
approvals, approvals,
approval_history, approval_history,
meta_inputs: read_meta_inputs(), meta_inputs: read_meta_inputs(),
operator_inbox,
questions, questions,
question_history, question_history,
tombstones, tombstones,
@ -643,7 +622,7 @@ async fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
agent: a.agent.clone(), agent: a.agent.clone(),
kind: "apply_commit", kind: "apply_commit",
sha_short: Some(sha), sha_short: Some(sha),
diff: Some(diff), diff_html: Some(render_diff_lines(&diff)),
description: a.description, description: a.description,
} }
} }
@ -652,7 +631,7 @@ async fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
agent: a.agent, agent: a.agent,
kind: "spawn", kind: "spawn",
sha_short: None, sha_short: None,
diff: None, diff_html: None,
description: a.description, description: a.description,
}, },
}); });
@ -720,58 +699,12 @@ fn dir_size_bytes(root: &Path) -> u64 {
total total
} }
async fn dashboard_history(State(state): State<AppState>) -> Response { async fn messages_stream(
// 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>, State(state): State<AppState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> { ) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
let rx = state.coord.dashboard_subscribe(); let rx = state.coord.broker.subscribe();
let stream = BroadcastStream::new(rx).filter_map(|res| { let stream = BroadcastStream::new(rx).filter_map(|res| {
// Drop lagged frames. Browsers reconnect; the seq dedupe on // Drop lagged events. Browsers reconnect; nothing to do here.
// reconnect skips any frame already reflected in the snapshot.
let event = res.ok()?; let event = res.ok()?;
let json = serde_json::to_string(&event).ok()?; let json = serde_json::to_string(&event).ok()?;
Some(Ok(Event::default().data(json))) Some(Ok(Event::default().data(json)))
@ -1141,13 +1074,7 @@ async fn post_op_send(State(state): State<AppState>, Form(form): Form<OpSendForm
}) { }) {
return error_response(&format!("op-send to {to} failed: {e:#}")); return error_response(&format!("op-send to {to} failed: {e:#}"));
} }
// 200 instead of 303 → the client doesn't refetch /api/state. The Redirect::to("/").into_response()
// 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( async fn post_request_spawn(
@ -1377,6 +1304,29 @@ fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<Approval> {
.collect() .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 /// 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 /// 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 /// dashboard reads this each render so logins driven from the agent web UI
@ -1424,3 +1374,8 @@ async fn git_diff_main_to(applied_dir: &Path, target_ref: &str) -> Result<String
Ok(String::from_utf8_lossy(&out.stdout).into_owned()) Ok(String::from_utf8_lossy(&out.stdout).into_owned())
} }
fn html_escape(s: &str) -> String {
s.replace('&', "&amp;")
.replace('<', "&lt;")
.replace('>', "&gt;")
}

47
hive-c0re/src/dashboard_events.rs
View file

@ -1,47 +0,0 @@
//! 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,
},
}

47
hive-c0re/src/main.rs
View file

@ -14,7 +14,6 @@ mod client;
mod coordinator; mod coordinator;
mod crash_watch; mod crash_watch;
mod dashboard; mod dashboard;
mod dashboard_events;
mod events_vacuum; mod events_vacuum;
mod forge; mod forge;
mod lifecycle; mod lifecycle;
@ -171,12 +170,6 @@ async fn main() -> Result<()> {
// Reminder scheduler: drains due reminders + handles // Reminder scheduler: drains due reminders + handles
// file_path payload persistence. See reminder_scheduler.rs. // file_path payload persistence. See reminder_scheduler.rs.
reminder_scheduler::spawn(coord.clone()); 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(); let dash_coord = coord.clone();
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await { if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {
@ -209,46 +202,6 @@ 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<()> { fn render(resp: HostResponse) -> Result<()> {
println!("{}", serde_json::to_string_pretty(&resp)?); println!("{}", serde_json::to_string_pretty(&resp)?);
if !resp.ok { if !resp.ok {

113
hive-fr0nt/assets/terminal.js
View file

@ -14,11 +14,7 @@
// delivered: (ev, api) => api.row('msgrow delivered', ...), // delivered: (ev, api) => api.row('msgrow delivered', ...),
// _default: (ev, api) => api.row('note', JSON.stringify(ev)), // _default: (ev, api) => api.row('note', JSON.stringify(ev)),
// }, // },
// onLiveEvent: (ev) => { /* live-only side effects (notif, state pokes) */ }, // onLiveEvent: (ev) => { /* side effects: notifications, 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 */ }, // onBackfillDone: (count) => { /* one-shot after history replay */ },
// pillAnchor: document.getElementById('msgflow').parentElement, // pillAnchor: document.getElementById('msgflow').parentElement,
// }); // });
@ -168,41 +164,38 @@
console.error('terminal renderer threw', ev, err); console.error('terminal renderer threw', ev, err);
row('note', '[render err] ' + (err && err.message ? err.message : 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); } 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. function subscribe() {
//
// 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); const es = new EventSource(opts.streamUrl);
es.onmessage = (e) => { es.onmessage = (e) => {
let ev; let ev;
try { ev = JSON.parse(e.data); } try { ev = JSON.parse(e.data); }
catch (err) { row('note', '[parse err] ' + e.data); return; } catch (err) { row('note', '[parse err] ' + e.data); return; }
if (!live) { buffered.push(ev); return; }
dispatch(ev, false); dispatch(ev, false);
if (opts.onLiveEvent) { if (opts.onLiveEvent) {
try { opts.onLiveEvent(ev); } try { opts.onLiveEvent(ev); }
@ -213,62 +206,10 @@
if (es.readyState === EventSource.CONNECTING) row('note', '[reconnecting…]'); if (es.readyState === EventSource.CONNECTING) row('note', '[reconnecting…]');
else row('note', '[disconnected]'); 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 = start(); const ready = backfill().then(subscribe);
return { row, details, detailsDiff, placeholder, ready }; return { row, details, detailsDiff, placeholder, ready };
} }