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

feat: type-scope events vacuum to prune only stream rows (14d) + drop turn-stats vacuum

Browse source
This commit is contained in:
damocles 2026-06-05 23:09:09 +02:00 • committed by mara
commit 6e39515669
9 changed files with 50 additions and 99 deletions
Download patch file Download diff file Expand all files Collapse all files

8
CLAUDE.md
View file

@ -104,10 +104,10 @@ hive-c0re/ host daemon + sibling operator CLI (lib + 2 bins)
fans out one Message per active target,
re-arms recurring rows, deletes fired one-shots
src/events_vacuum.rs host-side hourly sweep of every agent's
/harness/hyperhive-events.sqlite
src/stats_vacuum.rs host-side hourly sweep of every agent's
/harness/hyperhive-turn-stats.sqlite —
90-day age-only retention
/harness/hyperhive-events.sqlite — type-scoped:
prunes only verbose `stream` rows older than
14d, keeps semantic events (turn-stats has no
vacuum — it's tiny)
src/bash_tasks_vacuum.rs host-side hourly sweep of every agent's
harness/bash-tasks/ — deletes terminal task
trios (.json/.out/.err) older than 48h

26
docs/persistence.md
View file

@ -74,13 +74,19 @@ One table:
harness emits during turn loop execution.
The harness writes; the host vacuums. `hive-c0re::events_vacuum`
runs hourly and sweeps every existing agent harness dir, deleting
rows older than 7 days. Age-only — no row cap — so a chatty turn
doesn't lose history sooner than a quiet one; disk pressure on a
sustained burst is the cheaper problem to have. Centralising
retention on the host means a misbehaving harness can't disable
its own vacuum and agents don't need any cleanup wiring of their
own.
runs hourly and sweeps every existing agent harness dir. Retention
is **type-scoped**: it deletes only the verbose `stream` rows (the
raw claude `stream-json` deltas — one per text chunk / tool use, the
bulk of the file's size) older than 14 days, and keeps every other
kind (`turn_start`, `turn_end`, `note`, `status_changed`,
`model_changed`, `token_usage_changed`, `turn_state_changed`)
indefinitely — those are small and carry the semantic per-turn
history the operator scrolls back through when debugging a
regression. Age-only within the `stream` kind — no row cap — so a
chatty turn doesn't lose its stream history sooner than a quiet one.
Centralising retention on the host means a misbehaving harness can't
disable its own vacuum and agents don't need any cleanup wiring of
their own.
Path overridable via `HYPERHIVE_EVENTS_DB` (for dev / no-`/harness`
setups). On open failure the `Bus` falls back to no-store mode
@ -111,8 +117,10 @@ host-side). Best-effort and created on first write
(`CREATE TABLE IF NOT EXISTS`), so it's simply absent until a bash
task runs.
`hive-c0re::stats_vacuum` runs hourly and deletes rows older than 90 days
(`started_at < cutoff`). Age-only sweep, same pattern as `events_vacuum`.
turn-stats.sqlite has **no vacuum** — it's one tiny row per turn
(~hundreds of KB even over months), read directly by the `/stats` page
and the hive-wide stats view, so pruning it would only lose trend
history for no space gain.
### `/state/hyperhive-harness.json` (per agent)

2
hive-c0re/src/bash_tasks_vacuum.rs
View file

@ -14,7 +14,7 @@
//! any terminal task whose `completed_at` timestamp is older than
//! `KEEP_SECS`.
//!
//! Mirrors `events_vacuum` / `stats_vacuum` in structure — host-side
//! Mirrors `events_vacuum` in structure — host-side
//! so a misbehaving harness cannot disable its own cleanup.
use std::path::Path;

4
hive-c0re/src/build_logs.rs
View file

@ -367,7 +367,7 @@ impl BuildLogs {
/// Drop rows past their retention window. Returns the number of
/// rows deleted. Called from the existing hourly vacuum loop —
/// see `stats_vacuum` for the call site.
/// see `events_vacuum` for the call site.
///
/// Rule:
/// - `status = 'fail'` rows kept for `KEEP_FAIL_SECS` past their
@ -395,7 +395,7 @@ impl BuildLogs {
}
}
/// Spawn the hourly retention sweep. Mirrors `stats_vacuum::spawn` /
/// Spawn the hourly retention sweep. Mirrors `events_vacuum::spawn` /
/// `events_vacuum::spawn` in cadence + shutdown handling. Runs once
/// at startup before its first sleep so a long-uptime instance
/// doesn't accumulate a backlog the first hour after restart.

31
hive-c0re/src/events_vacuum.rs
View file

@ -1,13 +1,18 @@
//! Host-side vacuum of every per-agent events.sqlite. The harness
//! writes to `/agents/<name>/harness/hyperhive-events.sqlite`
//! (bind-mounted from `/var/lib/hyperhive/agents/<name>/harness/`);
//! we open the same file from the host every hour and delete rows
//! older than `KEEP_SECS`.
//! Age-only — no row cap — so a chatty turn doesn't lose history
//! sooner than a quiet one; disk pressure on a sustained burst is
//! a cheaper problem than a missing event when the operator is
//! debugging a regression. Keeping retention on the host means
//! agents don't need any cleanup wiring of their own, and a
//! we open the same file from the host every hour.
//!
//! **Type-scoped retention**: only the verbose `stream` rows (the raw
//! claude `stream-json` deltas — one per text chunk / `tool_use`, the
//! bulk of the file's size) are pruned, and only once they're older
//! than [`STREAM_KEEP_SECS`]. Every other kind (`turn_start`,
//! `turn_end`, `note`, `status_changed`, `model_changed`,
//! `token_usage_changed`, `turn_state_changed`) is kept — they're
//! small and carry the meaningful per-turn history the operator wants
//! to scroll back through. This keeps events.sqlite small-and-bounded
//! without throwing away semantic history. Keeping retention on the
//! host means agents don't need any cleanup wiring of their own, and a
//! misbehaving harness can't disable its own vacuum.
use std::path::Path;
@ -19,7 +24,9 @@ use rusqlite::{Connection, Result, params};
use crate::coordinator::Coordinator;
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
const KEEP_SECS: i64 = 7 * 24 * 3600;
/// How long verbose `stream` rows are kept before pruning. Other event
/// kinds are never deleted by this sweep.
const STREAM_KEEP_SECS: i64 = 14 * 24 * 3600;
/// Background loop: sweep every existing agent state dir hourly, run
/// the vacuum SQL against its events.sqlite if present. Errors are
@ -61,7 +68,11 @@ fn vacuum_file(path: &Path) -> Result<u64> {
.ok()
.and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0);
let cutoff = now - KEEP_SECS;
let removed = conn.execute("DELETE FROM events WHERE ts < ?1", params![cutoff])?;
let cutoff = now - STREAM_KEEP_SECS;
// Prune only the verbose `stream` deltas; semantic events are kept.
let removed = conn.execute(
"DELETE FROM events WHERE kind = 'stream' AND ts < ?1",
params![cutoff],
)?;
Ok(u64::try_from(removed).unwrap_or(0))
}

2
hive-c0re/src/hive_stats.rs
View file

@ -12,7 +12,7 @@
//! be lifted into a shared crate — overkill for now.
//!
//! Privsep: the sqlite files are mode 0644 owned by the agent user;
//! `hive-core` reads them fine (same as `stats_vacuum`). We open
//! `hive-core` reads them fine (same as `events_vacuum`). We open
//! read-only so an in-flight harness writer never blocks us.
use std::collections::HashMap;

1
hive-c0re/src/lib.rs
View file

@ -51,6 +51,5 @@ pub mod reminder_scheduler;
pub mod scheduled_prompts;
pub mod scheduled_prompts_worker;
pub mod server;
pub mod stats_vacuum;
pub mod tool_groups;
pub mod topology;

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

@ -14,7 +14,7 @@ use hive_c0re::coordinator::Coordinator;
use hive_c0re::{
agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard,
dashboard_events, events_vacuum, forge, knowledge, manager_server, matrix, migrate,
rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server, stats_vacuum,
rebuild_queue, reminder_scheduler, scheduled_prompts_worker, server,
};
#[derive(Parser)]
@ -364,9 +364,9 @@ async fn cmd_serve(
// Per-agent events.sqlite vacuum: host-side so the harness
// doesn't need any retention wiring of its own.
events_vacuum::spawn(&coord);
// Per-agent turn-stats.sqlite vacuum: same pattern, 90-day
// retention so trend analysis has enough history.
stats_vacuum::spawn(&coord);
// (turn-stats.sqlite has no vacuum — it's one tiny row per turn,
// ~hundreds of KB, and the /stats + hive-stats views read it
// directly; pruning it would just lose trend history for no gain.)
// Per-agent bash-tasks file vacuum: host-side so the harness
// cannot disable it. Deletes terminal task trios older than 48h.
bash_tasks_vacuum::spawn(&coord);

67
hive-c0re/src/stats_vacuum.rs
View file

@ -1,67 +0,0 @@
//! Host-side vacuum of every per-agent turn-stats.sqlite. The harness
//! writes to `/agents/<name>/harness/hyperhive-turn-stats.sqlite`
//! (bind-mounted from `/var/lib/hyperhive/agents/<name>/harness/`); we open the same file
//! from the host every hour and delete rows older than `KEEP_SECS`.
//! Mirrors `events_vacuum` in structure — host-side so the harness
//! can't disable it, age-only so a chatty burst doesn't evict old
//! rows sooner than expected. 90-day retention keeps enough history
//! for trend analysis without unbounded growth.
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use rusqlite::{Connection, Result, params};
use crate::coordinator::Coordinator;
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
const KEEP_SECS: i64 = 90 * 24 * 3600;
/// Background loop: sweep every existing agent state dir hourly, run
/// the vacuum SQL against its turn-stats.sqlite if present. Errors
/// are logged but don't tear the loop down.
pub fn spawn(coord: &Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move {
loop {
sweep_once();
tokio::select! {
() = tokio::time::sleep(VACUUM_INTERVAL) => {}
_ = shutdown.changed() => {
tracing::info!("stats vacuum: shutdown signal received");
break;
}
}
}
});
}
fn sweep_once() {
for name in Coordinator::kept_state_names() {
let path = Coordinator::agent_harness_dir(&name).join("hyperhive-turn-stats.sqlite");
if !path.exists() {
continue;
}
match vacuum_file(&path) {
Ok(0) => {}
Ok(n) => tracing::info!(agent = %name, removed = n, "turn-stats vacuum"),
Err(e) => tracing::warn!(agent = %name, error = ?e, "turn-stats vacuum failed"),
}
}
}
fn vacuum_file(path: &Path) -> Result<u64> {
let conn = Connection::open(path)?;
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
.and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0);
let cutoff = now - KEEP_SECS;
let removed = conn.execute(
"DELETE FROM turn_stats WHERE started_at < ?1",
params![cutoff],
)?;
Ok(u64::try_from(removed).unwrap_or(0))
}