diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs
index bddee64c..bc5af540 100644
--- a/hive-ag3nt/src/bin/hive-ag3nt.rs
+++ b/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,
}
}
+
diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs
index 256c4abe..d3cbd8d7 100644
--- a/hive-ag3nt/src/bin/hive-m1nd.rs
+++ b/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?;
diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs
index 54297d32..fb95223e 100644
--- a/hive-ag3nt/src/events.rs
+++ b/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
-/// 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);
-";
+/// 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;
/// 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
},
}
-/// sqlite-backed event log. Wraps a `Connection` behind a `Mutex` so the
-/// `Bus` (which clones cheaply) shares one writer.
-struct EventStore {
- conn: Mutex,
-}
-
-impl EventStore {
- fn open(path: &Path) -> rusqlite::Result {
- 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> {
- 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::(&s).ok())
- })?;
- let mut out: Vec = 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 {
- 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>,
- /// 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>,
+ history: Arc>>,
}
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`.
- /// Drives the terminal pre-fill when the operator opens the agent
- /// page; without a store (db open failed) this is empty.
+ /// Snapshot of the in-memory event ring buffer, oldest first. Drives the
+ /// terminal pre-fill when the operator opens the agent page.
#[must_use]
pub fn history(&self) -> Vec {
- 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)
+ self.history.lock().unwrap().iter().cloned().collect()
}
}
diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs
index 3f7bb781..4e5c1207 100644
--- a/hive-ag3nt/src/web_ui.rs
+++ b/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) -> 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) -> 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.
diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs
index bd6b45f9..232b15c1 100644
--- a/hive-c0re/src/broker.rs
+++ b/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 {
- 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> {
let conn = self.conn.lock().unwrap();
let row: Option<(i64, String, String, String)> = conn
diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs
index e3fb11d7..6561703a 100644
--- a/hive-c0re/src/main.rs
+++ b/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 {
diff --git a/nix/templates/harness-base.nix b/nix/templates/harness-base.nix
index a612a327..814e6051 100644
--- a/nix/templates/harness-base.nix
+++ b/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