operator control: /cancel slash command + cancel button

new POST /api/cancel on the per-agent web UI: shells out
pkill -INT claude (procps added to harness-base.nix). emits a Note
on the bus so the operator sees the cancel landed; state goes back
to idle when run_claude wakes and emits TurnEnd as usual.

frontend:
- /cancel slash command in the terminal input
- ■ cancel turn button in the state row, visible only while
  state === 'thinking' (driven from the same SSE-based state
  machine). disabled briefly during the POST.

claude gets SIGINT (not TERM) so it flushes anything in-flight and
emits a final result row before exiting.

Cargo.lock

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

TODO.md

@@ -29,14 +29,6 @@ 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 —
@@ -114,8 +106,6 @@ 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,6 +12,7 @@ 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,6 +126,26 @@ 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,10 +161,24 @@
   let termAPI = null;
 
   const SLASH_COMMANDS = [
-    { name: '/help',  desc: 'list slash commands' },
-    { name: '/clear', desc: 'wipe the terminal panel (local-only)' },
+    { name: '/help',   desc: 'list slash commands' },
+    { 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();
@@ -181,6 +195,9 @@
         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;
@@ -282,6 +299,8 @@
     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;
@@ -302,6 +321,16 @@
   }
   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,6 +15,7 @@
 
   <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,6 +58,7 @@ 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 {
@@ -153,6 +154,23 @@ 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*
@@ -176,4 +194,3 @@ async fn inbox_unread(socket: &Path) -> u64 {
         _ => 0,
     }
 }
-

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

@@ -61,6 +61,7 @@ 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 {
@@ -89,6 +90,22 @@ 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,17 +8,31 @@
 //! future events; the dashboard JS deals with the cold-start case by
 //! showing "connecting…" until the first event arrives.
 
-use std::collections::VecDeque;
+use std::path::{Path, PathBuf};
 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` keeps in its ring buffer. The web UI fetches
-/// this on page load to backfill the terminal so the operator sees the
-/// last turn(s) without having to wait for the next one.
-const HISTORY_CAPACITY: usize = 500;
+/// Max `LiveEvent`s the `Bus` returns from `history()` and keeps in
+/// sqlite. Older rows are vacuumed on a periodic sweep.
+const HISTORY_CAPACITY: usize = 2000;
+/// Default sqlite db path. Lives under `/state/` so it survives
+/// 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)]
@@ -45,29 +59,122 @@ 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>>,
-    history: Arc<Mutex<VecDeque<LiveEvent>>>,
+    /// 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/` 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),
-            history: Arc::new(Mutex::new(VecDeque::with_capacity(HISTORY_CAPACITY))),
+            store,
         }
     }
 
     pub fn emit(&self, event: LiveEvent) {
+        if let Some(store) = &self.store
+            && let Err(e) = store.append(&event)
         {
-            let mut h = self.history.lock().unwrap();
-            if h.len() == HISTORY_CAPACITY {
-                h.pop_front();
-            }
-            h.push_back(event.clone());
+            tracing::warn!(error = ?e, "events: append failed");
         }
         // Lagged subscribers drop events — fine; the UI is a tail, not a log.
         let _ = self.tx.send(event);
@@ -77,11 +184,22 @@ impl Bus {
         self.tx.subscribe()
     }
 
-    /// Snapshot of the in-memory event ring buffer, oldest first. Drives the
-    /// terminal pre-fill when the operator opens the agent page.
+    /// Most recent events, oldest first, capped at `HISTORY_CAPACITY`.
+    /// Drives the terminal pre-fill when the operator opens the agent
+    /// page; without a store (db open failed) this is empty.
     #[must_use]
     pub fn history(&self) -> Vec<LiveEvent> {
-        self.history.lock().unwrap().iter().cloned().collect()
+        let Some(store) = &self.store else {
+            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,6 +79,7 @@ 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)
@@ -273,6 +274,33 @@ 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,6 +171,21 @@ 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,6 +110,22 @@ 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,6 +14,9 @@
     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