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:300be8afa9ed999c58b9675879824e2d7cc18ce3
Branches Tags
vinzenz:main
..
compare: vinzenz:a9ed33d94f40925acad9d2596ad79d749560da64
Branches Tags
vinzenz:main
13 changed files with 28 additions and 284 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

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

10
TODO.md
View file

@ -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.).

1
hive-ag3nt/Cargo.toml
View file

@ -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

20
hive-ag3nt/assets/agent.css
View file

@ -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;

33
hive-ag3nt/assets/app.js
View file

@ -161,24 +161,10 @@
let termAPI = null;
const SLASH_COMMANDS = [
{ name: '/help', desc: 'list slash commands' },
{ name: '/clear', desc: 'wipe the terminal panel (local-only)' },
{ name: '/cancel', desc: 'SIGINT the in-flight claude turn' },
{ name: '/help', desc: 'list slash commands' },
{ name: '/clear', desc: 'wipe the terminal panel (local-only)' },
];
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.

1
hive-ag3nt/assets/index.html
View file

@ -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">

19
hive-ag3nt/src/bin/hive-ag3nt.rs
View file

@ -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,
}
}

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

@ -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?;

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

@ -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<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
/// at construction — we keep going so the harness doesn't die on a
/// missing `/state/` mount in dev / test scenarios.
store: Option<Arc<EventStore>>,
history: Arc<Mutex<VecDeque<LiveEvent>>>,
}
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<LiveEvent> {
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()
}
}

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

@ -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.

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

@ -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

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

@ -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 {

3
nix/templates/harness-base.nix
View file

@ -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