Cargo.lock

@@ -453,7 +453,6 @@ dependencies = [
  "clap",
  "hive-sh4re",
  "rmcp",
- "rusqlite",
  "schemars",
  "serde",
  "serde_json",

TODO.md

@@ -29,6 +29,14 @@ Pick anything from here when relevant. Cross-cutting design notes live in
 
 - **Per-agent UI substance.** Show last N inbox messages, last turn timing,
   link back to dashboard.
+- **Delivered events history persistence.** The `events::Bus` ring
+  buffer (500 events, in-memory) backfills the terminal on page load
+  but dies on harness restart, and only ever holds the most recent
+  turn or two. Persist to sqlite (`events(agent, id, ts, kind,
+  payload_json)`) so the operator can scroll back through prior
+  turns, and so `/events/history` survives restart. Cap rows per
+  agent or auto-vacuum on age, same trade-off as the bounded broker
+  entry below.
 - **State badge: compacting + napping states.** Idle/thinking already
   ship (driven from SSE turn_start/turn_end). Add `compacting 📦` and
   `napping 😴` once the `/compact` trigger and `nap` tool exist —
@@ -106,6 +114,8 @@ Pick anything from here when relevant. Cross-cutting design notes live in
 
 ## Lifecycle / reliability
 
+- **Bounded broker.** Cap rows per recipient or auto-vacuum delivered
+  messages older than a threshold. sqlite is growing unbounded.
 - **Container crash events.** Watch `container@*.service` via D-Bus, push
   `HelperEvent::ContainerCrash` to the manager's inbox so the manager can
   react (restart, escalate, etc.).

hive-ag3nt/Cargo.toml

@@ -12,7 +12,6 @@ axum.workspace = true
 clap.workspace = true
 hive-sh4re.workspace = true
 rmcp.workspace = true
-rusqlite.workspace = true
 schemars.workspace = true
 serde.workspace = true
 serde_json.workspace = true

hive-ag3nt/assets/agent.css

@@ -126,26 +126,6 @@ pre.diff {
 }
 #state-row {
   margin: 0.4em 0 0.2em;
-  display: flex;
-  align-items: center;
-  gap: 0.6em;
-}
-.btn-cancel-turn {
-  font-family: inherit;
-  font-size: 0.8em;
-  letter-spacing: 0.08em;
-  background: transparent;
-  color: var(--red);
-  border: 1px solid var(--red);
-  border-radius: 999px;
-  padding: 0.2em 0.8em;
-  cursor: pointer;
-  text-shadow: 0 0 4px currentColor;
-  transition: box-shadow 0.15s ease, background 0.15s ease;
-}
-.btn-cancel-turn:hover {
-  background: rgba(243, 139, 168, 0.1);
-  box-shadow: 0 0 10px -2px currentColor;
 }
 .state-badge {
   display: inline-block;

hive-ag3nt/assets/app.js

@@ -161,24 +161,10 @@
   let termAPI = null;
 
   const SLASH_COMMANDS = [
-    { name: '/help',   desc: 'list slash commands' },
+    { name: '/help',  desc: 'list slash commands' },
-    { name: '/clear',  desc: 'wipe the terminal panel (local-only)' },
+    { name: '/clear', desc: 'wipe the terminal panel (local-only)' },
-    { name: '/cancel', desc: 'SIGINT the in-flight claude turn' },
   ];
 
-  async function postCancelTurn() {
-    try {
-      const resp = await fetch('/api/cancel', { method: 'POST', redirect: 'manual' });
-      const ok = resp.ok || resp.type === 'opaqueredirect'
-        || (resp.status >= 200 && resp.status < 400);
-      if (!ok && termAPI) {
-        termAPI.row('turn-end-fail', '✗ /cancel failed: http ' + resp.status);
-      }
-    } catch (err) {
-      if (termAPI) termAPI.row('turn-end-fail', '✗ /cancel failed: ' + err);
-    }
-  }
-
   function handleSlashCommand(line) {
     if (!termAPI) return false;
     const trimmed = line.trim();
@@ -195,9 +181,6 @@
         termAPI.clear();
         termAPI.row('note', '· terminal cleared (local view only — server history kept)');
         return true;
-      case '/cancel':
-        postCancelTurn();
-        return true;
       default:
         termAPI.row('turn-end-fail', '✗ unknown slash command: ' + cmd + ' — try /help');
         return true;
@@ -299,8 +282,6 @@
     const age = fmtAge(Date.now() - stateSince);
     badge.textContent = def.glyph + ' ' + def.text + ' · ' + age;
     badge.className = 'state-badge state-' + stateName;
-    const cancelBtn = $('cancel-btn');
-    if (cancelBtn) cancelBtn.hidden = stateName !== 'thinking';
   }
   function setState(next) {
     if (next === stateName) return;
@@ -321,16 +302,6 @@
   }
   startStateTicker();
 
-  // Wire the cancel-turn button (visible only while state === thinking).
-  (() => {
-    const btn = $('cancel-btn');
-    if (!btn) return;
-    btn.addEventListener('click', () => {
-      btn.disabled = true;
-      postCancelTurn().finally(() => { btn.disabled = false; });
-    });
-  })();
-
   // Track banner activity by reference-counting in-flight turns. A turn
   // can begin while the previous turn_end is still in the pipeline (rare
   // but happens on tight wake cycles), so we count rather than toggle.

hive-ag3nt/assets/index.html

@@ -15,7 +15,6 @@
 
   <div id="state-row">
     <span id="state-badge" class="state-badge state-loading">… booting</span>
-    <button type="button" id="cancel-btn" class="btn-cancel-turn" hidden>■ cancel turn</button>
   </div>
 
   <div class="terminal-wrap">

hive-ag3nt/src/bin/hive-ag3nt.rs

@@ -58,7 +58,6 @@ async fn main() -> Result<()> {
             let login_state = Arc::new(Mutex::new(initial));
             let ui_state = login_state.clone();
             let bus = Bus::new();
-            spawn_events_vacuum(bus.clone());
             let ui_bus = bus.clone();
             let ui_socket = cli.socket.clone();
             tokio::spawn(async move {
@@ -154,23 +153,6 @@ async fn serve(
     }
 }
 
-/// Vacuum events older than 7 days, cap to 2000 most-recent rows.
-/// Runs immediately, then hourly.
-fn spawn_events_vacuum(bus: Bus) {
-    tokio::spawn(async move {
-        let interval_secs = 3600u64;
-        let keep_secs: i64 = 7 * 24 * 3600;
-        let keep_rows = 2000;
-        loop {
-            let n = bus.vacuum(keep_secs, keep_rows);
-            if n > 0 {
-                tracing::info!(removed = n, "events vacuum");
-            }
-            tokio::time::sleep(Duration::from_secs(interval_secs)).await;
-        }
-    });
-}
-
 /// Per-turn user prompt. The role/tools/etc. is in the system prompt
 /// (`prompts/agent.md` → `claude --system-prompt-file`); this is just the
 /// wake signal claude reacts to. `unread` is the count of *other*
@@ -194,3 +176,4 @@ async fn inbox_unread(socket: &Path) -> u64 {
         _ => 0,
     }
 }
+

hive-ag3nt/src/bin/hive-m1nd.rs

@@ -61,7 +61,6 @@ async fn main() -> Result<()> {
             let login_state = Arc::new(Mutex::new(initial));
             let ui_state = login_state.clone();
             let bus = Bus::new();
-            spawn_events_vacuum(bus.clone());
             let ui_bus = bus.clone();
             let ui_socket = cli.socket.clone();
             tokio::spawn(async move {
@@ -90,22 +89,6 @@ async fn main() -> Result<()> {
     }
 }
 
-/// Vacuum events older than 7 days, cap to 2000 most-recent rows.
-fn spawn_events_vacuum(bus: Bus) {
-    tokio::spawn(async move {
-        let interval_secs = 3600u64;
-        let keep_secs: i64 = 7 * 24 * 3600;
-        let keep_rows = 2000;
-        loop {
-            let n = bus.vacuum(keep_secs, keep_rows);
-            if n > 0 {
-                tracing::info!(removed = n, "events vacuum");
-            }
-            tokio::time::sleep(Duration::from_secs(interval_secs)).await;
-        }
-    });
-}
-
 async fn serve(socket: &Path, interval: Duration, bus: Bus) -> Result<()> {
     tracing::info!(socket = %socket.display(), "hive-m1nd serve");
     let mcp_config = turn::write_mcp_config(socket).await?;

hive-ag3nt/src/events.rs

@@ -8,31 +8,17 @@
 //! future events; the dashboard JS deals with the cold-start case by
 //! showing "connecting…" until the first event arrives.
 
-use std::path::{Path, PathBuf};
+use std::collections::VecDeque;
 use std::sync::{Arc, Mutex};
 
-use rusqlite::{Connection, params};
 use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast;
 
 const CHANNEL_CAPACITY: usize = 256;
-/// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
+/// Max `LiveEvent`s the `Bus` keeps in its ring buffer. The web UI fetches
-/// sqlite. Older rows are vacuumed on a periodic sweep.
+/// this on page load to backfill the terminal so the operator sees the
-const HISTORY_CAPACITY: usize = 2000;
+/// last turn(s) without having to wait for the next one.
-/// Default sqlite db path. Lives under `/state/` so it survives
+const HISTORY_CAPACITY: usize = 500;
-/// destroy/recreate but goes away on purge. Overridable via the
-/// `HYPERHIVE_EVENTS_DB` env var (used in tests and one-shot tools).
-const DEFAULT_EVENTS_DB: &str = "/state/hyperhive-events.sqlite";
-
-const SCHEMA: &str = "
-CREATE TABLE IF NOT EXISTS events (
-    id           INTEGER PRIMARY KEY AUTOINCREMENT,
-    ts           INTEGER NOT NULL,
-    kind         TEXT    NOT NULL,
-    payload_json TEXT    NOT NULL
-);
-CREATE INDEX IF NOT EXISTS idx_events_ts ON events (ts);
-";
 
 /// One row of the agent's live stream. Serialised to JSON for SSE delivery.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -59,122 +45,29 @@ pub enum LiveEvent {
     TurnEnd { ok: bool, note: Option<String> },
 }
 
-/// sqlite-backed event log. Wraps a `Connection` behind a `Mutex` so the
-/// `Bus` (which clones cheaply) shares one writer.
-struct EventStore {
-    conn: Mutex<Connection>,
-}
-
-impl EventStore {
-    fn open(path: &Path) -> rusqlite::Result<Self> {
-        if let Some(parent) = path.parent() {
-            let _ = std::fs::create_dir_all(parent);
-        }
-        let conn = Connection::open(path)?;
-        conn.execute_batch(SCHEMA)?;
-        Ok(Self {
-            conn: Mutex::new(conn),
-        })
-    }
-
-    fn append(&self, event: &LiveEvent) -> rusqlite::Result<()> {
-        let ts = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .ok()
-            .and_then(|d| i64::try_from(d.as_secs()).ok())
-            .unwrap_or(0);
-        let kind = match event {
-            LiveEvent::TurnStart { .. } => "turn_start",
-            LiveEvent::Stream(_) => "stream",
-            LiveEvent::Note(_) => "note",
-            LiveEvent::TurnEnd { .. } => "turn_end",
-        };
-        let payload = serde_json::to_string(event).unwrap_or_else(|_| "null".into());
-        let conn = self.conn.lock().unwrap();
-        conn.execute(
-            "INSERT INTO events (ts, kind, payload_json) VALUES (?1, ?2, ?3)",
-            params![ts, kind, payload],
-        )?;
-        Ok(())
-    }
-
-    fn recent(&self, limit: usize) -> rusqlite::Result<Vec<LiveEvent>> {
-        let limit_i = i64::try_from(limit).unwrap_or(i64::MAX);
-        let conn = self.conn.lock().unwrap();
-        let mut stmt = conn.prepare(
-            "SELECT payload_json FROM events
-              ORDER BY id DESC
-              LIMIT ?1",
-        )?;
-        let rows = stmt.query_map(params![limit_i], |row| {
-            let s: String = row.get(0)?;
-            Ok(serde_json::from_str::<LiveEvent>(&s).ok())
-        })?;
-        let mut out: Vec<LiveEvent> = rows.flatten().flatten().collect();
-        out.reverse();
-        Ok(out)
-    }
-
-    /// Drop rows older than `older_than_secs` AND any rows beyond
-    /// `keep_rows` newest. Two-stage so a quiet agent keeps a useful
-    /// tail and a chatty one is bounded.
-    fn vacuum(&self, older_than_secs: i64, keep_rows: usize) -> rusqlite::Result<u64> {
-        let now = std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .ok()
-            .and_then(|d| i64::try_from(d.as_secs()).ok())
-            .unwrap_or(0);
-        let cutoff = now - older_than_secs;
-        let conn = self.conn.lock().unwrap();
-        let by_age = conn.execute("DELETE FROM events WHERE ts < ?1", params![cutoff])?;
-        let keep_i = i64::try_from(keep_rows).unwrap_or(i64::MAX);
-        let by_count = conn.execute(
-            "DELETE FROM events
-              WHERE id NOT IN (
-                SELECT id FROM events ORDER BY id DESC LIMIT ?1
-              )",
-            params![keep_i],
-        )?;
-        Ok(u64::try_from(by_age + by_count).unwrap_or(0))
-    }
-}
-
 #[derive(Clone)]
 pub struct Bus {
     tx: Arc<broadcast::Sender<LiveEvent>>,
-    /// Persistent event log. `None` only if opening the sqlite db failed
+    history: Arc<Mutex<VecDeque<LiveEvent>>>,
-    /// at construction — we keep going so the harness doesn't die on a
-    /// missing `/state/` mount in dev / test scenarios.
-    store: Option<Arc<EventStore>>,
 }
 
 impl Bus {
-    /// Open the default events db (`/state/hyperhive-events.sqlite`, or
-    /// `HYPERHIVE_EVENTS_DB`). On failure, fall back to a no-store bus —
-    /// the harness still works, just without persistent history.
     #[must_use]
     pub fn new() -> Self {
-        let path = std::env::var_os("HYPERHIVE_EVENTS_DB")
-            .map_or_else(|| PathBuf::from(DEFAULT_EVENTS_DB), PathBuf::from);
-        let store = match EventStore::open(&path) {
-            Ok(s) => Some(Arc::new(s)),
-            Err(e) => {
-                tracing::warn!(error = ?e, path = %path.display(), "events db open failed; running without history");
-                None
-            }
-        };
         let (tx, _) = broadcast::channel(CHANNEL_CAPACITY);
         Self {
             tx: Arc::new(tx),
-            store,
+            history: Arc::new(Mutex::new(VecDeque::with_capacity(HISTORY_CAPACITY))),
         }
     }
 
     pub fn emit(&self, event: LiveEvent) {
-        if let Some(store) = &self.store
-            && let Err(e) = store.append(&event)
         {
-            tracing::warn!(error = ?e, "events: append failed");
+            let mut h = self.history.lock().unwrap();
+            if h.len() == HISTORY_CAPACITY {
+                h.pop_front();
+            }
+            h.push_back(event.clone());
         }
         // Lagged subscribers drop events — fine; the UI is a tail, not a log.
         let _ = self.tx.send(event);
@@ -184,22 +77,11 @@ impl Bus {
         self.tx.subscribe()
     }
 
-    /// Most recent events, oldest first, capped at `HISTORY_CAPACITY`.
+    /// Snapshot of the in-memory event ring buffer, oldest first. Drives the
-    /// Drives the terminal pre-fill when the operator opens the agent
+    /// terminal pre-fill when the operator opens the agent page.
-    /// page; without a store (db open failed) this is empty.
     #[must_use]
     pub fn history(&self) -> Vec<LiveEvent> {
-        let Some(store) = &self.store else {
+        self.history.lock().unwrap().iter().cloned().collect()
-            return Vec::new();
-        };
-        store.recent(HISTORY_CAPACITY).unwrap_or_default()
-    }
-
-    /// Drop events older than `older_than_secs` and keep only the
-    /// newest `keep_rows`. Called periodically by the harness.
-    pub fn vacuum(&self, older_than_secs: i64, keep_rows: usize) -> u64 {
-        let Some(store) = &self.store else { return 0 };
-        store.vacuum(older_than_secs, keep_rows).unwrap_or(0)
     }
 }
 

hive-ag3nt/src/web_ui.rs

@@ -79,7 +79,6 @@ pub async fn serve(
         .route("/login/start", post(post_login_start))
         .route("/login/code", post(post_login_code))
         .route("/login/cancel", post(post_login_cancel))
-        .route("/api/cancel", post(post_cancel_turn))
         .with_state(state);
     let addr = SocketAddr::from(([0, 0, 0, 0], port));
     let listener = tokio::net::TcpListener::bind(addr)
@@ -274,33 +273,6 @@ async fn post_login_cancel(State(state): State<AppState>) -> Response {
     Redirect::to("/").into_response()
 }
 
-/// Cancel the in-flight claude turn. Coarse-grained: shells out
-/// `pkill -INT claude` since there's at most one claude per container.
-/// SIGINT (not SIGTERM) so claude flushes anything in-flight and emits a
-/// final result row. Emits a Note so the operator sees the cancel
-/// landed; the actual state transition back to `idle` happens when
-/// `run_claude` wakes up and the harness emits `TurnEnd`.
-async fn post_cancel_turn(State(state): State<AppState>) -> Response {
-    let out = tokio::process::Command::new("pkill")
-        .args(["-INT", "claude"])
-        .output()
-        .await;
-    let note = match out {
-        Ok(o) if o.status.success() => "operator: /cancel — sent SIGINT to claude".to_owned(),
-        Ok(o) if o.status.code() == Some(1) => {
-            "operator: /cancel — no claude process to interrupt".to_owned()
-        }
-        Ok(o) => format!(
-            "operator: /cancel — pkill exited {} stderr={}",
-            o.status,
-            String::from_utf8_lossy(&o.stderr).trim()
-        ),
-        Err(e) => format!("operator: /cancel — pkill failed: {e}"),
-    };
-    state.bus.emit(crate::events::LiveEvent::Note(note));
-    Redirect::to("/").into_response()
-}
-
 fn error_response(message: &str) -> Response {
     // Plain text — JS app surfaces in `alert()`, HTML wrapping would just
     // be noise.

hive-c0re/src/broker.rs

@@ -171,21 +171,6 @@ impl Broker {
         }
     }
 
-    /// Delete delivered messages older than `older_than_secs`. Undelivered
-    /// rows are always kept regardless of age — those are still in flight
-    /// from the broker's POV. Returns the number of rows removed.
-    pub fn vacuum_delivered(&self, older_than_secs: i64) -> Result<u64> {
-        let cutoff = now_unix() - older_than_secs;
-        let conn = self.conn.lock().unwrap();
-        let n = conn.execute(
-            "DELETE FROM messages
-              WHERE delivered_at IS NOT NULL
-                AND delivered_at < ?1",
-            params![cutoff],
-        )?;
-        Ok(u64::try_from(n).unwrap_or(0))
-    }
-
     pub fn recv(&self, recipient: &str) -> Result<Option<Message>> {
         let conn = self.conn.lock().unwrap();
         let row: Option<(i64, String, String, String)> = conn

hive-c0re/src/main.rs

@@ -110,22 +110,6 @@ async fn main() -> Result<()> {
                     tracing::warn!(error = ?e, "auto-update task failed");
                 }
             });
-            // Periodic broker vacuum: drop delivered messages older than
-            // 30 days. Undelivered messages are always kept (still in
-            // flight). Runs hourly; first sweep happens immediately.
-            let vacuum_coord = coord.clone();
-            tokio::spawn(async move {
-                let interval_secs = 3600u64;
-                let keep_secs: i64 = 30 * 24 * 3600;
-                loop {
-                    match vacuum_coord.broker.vacuum_delivered(keep_secs) {
-                        Ok(0) => {}
-                        Ok(n) => tracing::info!(removed = n, "broker vacuum"),
-                        Err(e) => tracing::warn!(error = ?e, "broker vacuum failed"),
-                    }
-                    tokio::time::sleep(std::time::Duration::from_secs(interval_secs)).await;
-                }
-            });
             let dash_coord = coord.clone();
             tokio::spawn(async move {
                 if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {

nix/templates/harness-base.nix

@@ -14,9 +14,6 @@
     claude-code
     bashInteractive
     coreutils-full
-    # procps for pkill — used by the web UI's /api/cancel to SIGINT the
-    # in-flight claude turn.
-    procps
   ];
 
   # Git is needed by claude's Bash tool (for the agent <-> manager config