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

fix(#2031): move bash-tasks + events vacuum agent-side (privsep ownership)

Browse source
This commit is contained in:
damocles 2026-06-27 00:26:56 +02:00 • committed by mara
commit 49fc61212b
9 changed files with 159 additions and 224 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -429,6 +429,10 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
S::send_to_parent(socket, failure).await;
}
tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf()));
// Agent-side cleanup of this agent's own harness artifacts (completed
// bash-task files + verbose event rows). Runs here, not host-side in
// hive-c0re, because the files are agent-owned — see `vacuum` module docs.
tokio::spawn(hive_ag3nt::vacuum::run());
// Log web_ui::serve's error instead of dropping it. A bare
// `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so
// any Err (e.g. EACCES from `bind_unix` when HIVE_WEB_SOCKET points

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

@ -17,6 +17,7 @@ pub mod serve_common;
pub mod stats;
pub mod turn;
pub mod turn_stats;
pub mod vacuum;
pub mod web_ui;
/// Default socket path inside the container — bind-mounted by `hive-c0re`.

140
hive-ag3nt/src/vacuum.rs Normal file
View file

@ -0,0 +1,140 @@
//! Agent-side cleanup of this agent's own harness artifacts: completed
//! bash-task files and verbose `stream` event rows.
//!
//! Runs IN the harness (not host-side in hive-c0re) because the files are
//! owned by the agent user. Under privsep hive-c0re runs as the unprivileged
//! `hive-core` user and cannot delete agent-owned files — the old host-side
//! sweeps hit `PermissionDenied` on the bash-task trio and an
//! attempt-to-write-a-readonly-database error on `events.sqlite`. The harness
//! owns these paths, so the deletes succeed here.
//!
//! Trade-off (accepted — issue tracker "perms borked"): a misbehaving harness
//! could skip its own cleanup, which the host-side version was meant to
//! prevent. But a compromised harness is already inside the container trust
//! boundary (docs/security.md), and these are ephemeral local artifacts — so
//! the honest fix is to clean them up where they live.
use std::path::Path;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use rusqlite::{Connection, Result, params};
/// How often the sweep runs.
const VACUUM_INTERVAL: Duration = Duration::from_secs(3600);
/// Keep completed bash-task files this long before deleting their trio.
const BASH_KEEP_SECS: i64 = 48 * 3600;
/// Keep verbose `stream` event rows this long before pruning. Other event
/// kinds are never deleted by this sweep — they carry the semantic per-turn
/// history the operator scrolls back through.
const STREAM_KEEP_SECS: i64 = 14 * 24 * 3600;
/// Terminal bash-task statuses whose files are eligible for deletion.
const TERMINAL_STATUSES: &[&str] = &["done", "timed_out", "interrupted"];
/// Background loop: hourly, prune this agent's stale bash-task files and
/// verbose event rows. Detached task — runs for the harness's lifetime;
/// errors are logged, never fatal.
pub async fn run() {
loop {
sweep_once();
tokio::time::sleep(VACUUM_INTERVAL).await;
}
}
fn sweep_once() {
let harness = crate::paths::harness_dir();
let tasks_dir = harness.join("bash-tasks");
if tasks_dir.is_dir() {
let removed = vacuum_bash_tasks(&tasks_dir, now_unix() - BASH_KEEP_SECS);
if removed > 0 {
tracing::info!(removed, "bash-tasks vacuum");
}
}
let events_db = harness.join("hyperhive-events.sqlite");
if events_db.exists() {
match vacuum_events(&events_db) {
Ok(0) => {}
Ok(n) => tracing::info!(removed = n, "events vacuum"),
Err(e) => tracing::warn!(error = ?e, "events vacuum failed"),
}
}
}
/// Delete eligible bash-task trios in `dir`. Returns the count of `.json`
/// sentinels removed (each represents one task; `.out`/`.err` deletions are
/// not counted separately).
fn vacuum_bash_tasks(dir: &Path, cutoff: i64) -> u64 {
let Ok(rd) = std::fs::read_dir(dir) else {
return 0;
};
let mut removed: u64 = 0;
for entry in rd.flatten() {
let path = entry.path();
// Only process the .json sentinel; derive sibling paths from it.
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
let Some(stem) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
continue;
};
if should_delete(&path, cutoff) {
delete_trio(dir, &stem);
removed += 1;
}
}
removed
}
/// Return `true` when the task file has a terminal status and a
/// `completed_at` older than `cutoff`.
fn should_delete(json_path: &Path, cutoff: i64) -> bool {
let Ok(raw) = std::fs::read_to_string(json_path) else {
return false;
};
let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) else {
return false;
};
let status = v.get("status").and_then(|s| s.as_str()).unwrap_or("");
if !TERMINAL_STATUSES.contains(&status) {
return false;
}
let completed_at = v
.get("completed_at")
.and_then(serde_json::Value::as_i64)
.unwrap_or(i64::MAX);
completed_at < cutoff
}
/// Delete the `.json`, `.out`, and `.err` files for a task. Errors are
/// logged but do not abort the sweep.
fn delete_trio(dir: &Path, stem: &str) {
for ext in ["json", "out", "err"] {
let path = dir.join(format!("{stem}.{ext}"));
if path.exists()
&& let Err(e) = std::fs::remove_file(&path)
{
tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed");
}
}
}
/// Prune verbose `stream` event rows older than [`STREAM_KEEP_SECS`] from the
/// agent's `events.sqlite`. Returns the number of rows deleted.
fn vacuum_events(path: &Path) -> Result<u64> {
let conn = Connection::open(path)?;
let cutoff = now_unix() - STREAM_KEEP_SECS;
let removed = conn.execute(
"DELETE FROM events WHERE kind = 'stream' AND ts < ?1",
params![cutoff],
)?;
Ok(u64::try_from(removed).unwrap_or(0))
}
fn now_unix() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
.and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0)
}

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

@ -1,128 +0,0 @@
//! Host-side vacuum of per-agent bash-task files. The harness writes
//! three files per `bash_run` call:
//!
//! ```text
//! harness_dir()/bash-tasks/<id>.json — task metadata + status
//! harness_dir()/bash-tasks/<id>.out — captured stdout
//! harness_dir()/bash-tasks/<id>.err — captured stderr
//! ```
//!
//! Completed tasks (status `done`, `timed_out`, `interrupted`) are
//! never removed by the harness. On a busy agent they accumulate
//! indefinitely. This module sweeps every agent's `bash-tasks/`
//! directory hourly and deletes the `.json`/`.out`/`.err` trio for
//! any terminal task whose `completed_at` timestamp is older than
//! `KEEP_SECS`.
//!
//! Mirrors `events_vacuum` in structure — host-side
//! so a misbehaving harness cannot disable its own cleanup.
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use crate::coordinator::Coordinator;
const VACUUM_INTERVAL: Duration = Duration::from_hours(1);
/// Keep completed task files for 48 hours before sweeping them.
const KEEP_SECS: i64 = 48 * 3600;
/// Terminal task statuses — files for these are eligible for deletion.
const TERMINAL_STATUSES: &[&str] = &["done", "timed_out", "interrupted"];
/// Spawn the background vacuum loop as a detached tokio task.
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!("bash-tasks vacuum: shutdown signal received");
break;
}
}
}
});
}
fn sweep_once() {
let cutoff = now_unix() - KEEP_SECS;
for name in Coordinator::kept_state_names() {
let tasks_dir = Coordinator::agent_harness_dir(&name).join("bash-tasks");
if !tasks_dir.is_dir() {
continue;
}
let removed = vacuum_dir(&tasks_dir, cutoff);
if removed > 0 {
tracing::info!(agent = %name, removed, "bash-tasks vacuum");
}
}
}
/// Delete eligible task trios in `dir`. Returns the count of `.json`
/// files removed (each represents one task; `.out`/`.err` deletions
/// are not counted separately).
fn vacuum_dir(dir: &Path, cutoff: i64) -> u64 {
let Ok(rd) = std::fs::read_dir(dir) else {
return 0;
};
let mut removed: u64 = 0;
for entry in rd.flatten() {
let path = entry.path();
// Only process the .json sentinel; derive sibling paths from it.
if path.extension().and_then(|e| e.to_str()) != Some("json") {
continue;
}
let Some(stem) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else {
continue;
};
if should_delete(&path, cutoff) {
delete_trio(dir, &stem);
removed += 1;
}
}
removed
}
/// Return `true` when the task file has a terminal status and a
/// `completed_at` older than `cutoff`.
fn should_delete(json_path: &Path, cutoff: i64) -> bool {
let Ok(raw) = std::fs::read_to_string(json_path) else {
return false;
};
let Ok(v) = serde_json::from_str::<serde_json::Value>(&raw) else {
return false;
};
let status = v.get("status").and_then(|s| s.as_str()).unwrap_or("");
if !TERMINAL_STATUSES.contains(&status) {
return false;
}
let completed_at = v
.get("completed_at")
.and_then(serde_json::Value::as_i64)
.unwrap_or(i64::MAX);
completed_at < cutoff
}
/// Delete the `.json`, `.out`, and `.err` files for a task. Errors
/// are logged but do not abort the sweep.
fn delete_trio(dir: &Path, stem: &str) {
for ext in ["json", "out", "err"] {
let path = dir.join(format!("{stem}.{ext}"));
if path.exists()
&& let Err(e) = std::fs::remove_file(&path)
{
tracing::warn!(path = %path.display(), error = ?e, "bash-tasks vacuum: remove failed");
}
}
}
fn now_unix() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
.and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0)
}

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

@ -366,8 +366,7 @@ impl BuildLogs {
}
/// Drop rows past their retention window. Returns the number of
/// rows deleted. Called from the existing hourly vacuum loop —
/// see `events_vacuum` for the call site.
/// rows deleted. Called from this module's hourly `spawn` loop.
///
/// Rule:
/// - `status = 'fail'` rows kept for `KEEP_FAIL_SECS` past their
@ -395,10 +394,10 @@ impl BuildLogs {
}
}
/// 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.
/// Spawn the hourly retention sweep. A host-side sweep (build_logs.sqlite
/// is hive-c0re-owned, so no privsep ownership issue). Runs once at startup
/// before its first sleep so a long-uptime instance doesn't accumulate a
/// backlog the first hour after restart.
pub fn spawn_vacuum(coord: &Arc<crate::coordinator::Coordinator>) {
use std::time::Duration;
let logs = coord.build_logs.clone();

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

@ -1,78 +0,0 @@
//! 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.
//!
//! **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;
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);
/// 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
/// 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!("events vacuum: shutdown signal received");
break;
}
}
}
});
}
fn sweep_once() {
for name in Coordinator::kept_state_names() {
let path = Coordinator::agent_harness_dir(&name).join("hyperhive-events.sqlite");
if !path.exists() {
continue;
}
match vacuum_file(&path) {
Ok(0) => {}
Ok(n) => tracing::info!(agent = %name, removed = n, "events vacuum"),
Err(e) => tracing::warn!(agent = %name, error = ?e, "events 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 - 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))
}

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

@ -12,7 +12,8 @@
//! 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 `events_vacuum`). We open
//! `hive-core` can read them fine, but cannot write/delete (which is why
//! retention sweeps run agent-side in the harness, not here). We open
//! read-only so an in-flight harness writer never blocks us.
use std::collections::HashMap;

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

@ -17,7 +17,6 @@ pub mod agent_sockets;
pub mod approvals;
pub mod audit_log;
pub mod auto_update;
pub mod bash_tasks_vacuum;
pub mod broker;
pub mod build_logs;
pub mod capabilities;
@ -28,7 +27,6 @@ pub mod coordinator;
pub mod crash_watch;
pub mod dashboard;
pub mod dashboard_events;
pub mod events_vacuum;
pub mod flake_check;
pub mod forge;
pub mod gateway_nginx;

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

@ -12,9 +12,9 @@ use hive_sh4re::{HostRequest, HostResponse};
// explicit (any new daemon entry point reads off the next add).
use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig};
use hive_c0re::{
agent_sockets, auto_update, bash_tasks_vacuum, broker, client, crash_watch, dashboard,
dashboard_events, events_vacuum, forge, knowledge, matrix, migrate, rebuild_queue,
reminder_scheduler, scheduled_prompts_worker, server, socket_server,
agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events, forge,
knowledge, matrix, migrate, rebuild_queue, reminder_scheduler, scheduled_prompts_worker,
server, socket_server,
};
#[derive(Parser)]
@ -385,15 +385,13 @@ 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 events.sqlite + bash-tasks file cleanup now runs
// agent-side in the harness (`hive_ag3nt::vacuum`): the files are
// agent-owned, so host-side deletes hit PermissionDenied / readonly-db
// under privsep. See issue tracker "perms borked".
// (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);
// Slow per-container disk sampler: a `du` of each agent's state dir
// + writable rootfs every ~5 min, cached so the 5s container-load
// poll stays cheap cgroup-only reads. Feeds `disk_bytes` on the LOAD