diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index e31b26e7..0697ddca 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -429,10 +429,6 @@ async fn serve_main(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 diff --git a/hive-ag3nt/src/lib.rs b/hive-ag3nt/src/lib.rs index 56a62a29..f02ec897 100644 --- a/hive-ag3nt/src/lib.rs +++ b/hive-ag3nt/src/lib.rs @@ -17,7 +17,6 @@ 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`. diff --git a/hive-ag3nt/src/vacuum.rs b/hive-ag3nt/src/vacuum.rs deleted file mode 100644 index 8455a371..00000000 --- a/hive-ag3nt/src/vacuum.rs +++ /dev/null @@ -1,140 +0,0 @@ -//! 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_hours(1); -/// 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::(&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 { - 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) -} diff --git a/hive-c0re/src/bash_tasks_vacuum.rs b/hive-c0re/src/bash_tasks_vacuum.rs new file mode 100644 index 00000000..6cce4dbc --- /dev/null +++ b/hive-c0re/src/bash_tasks_vacuum.rs @@ -0,0 +1,128 @@ +//! Host-side vacuum of per-agent bash-task files. The harness writes +//! three files per `bash_run` call: +//! +//! ```text +//! harness_dir()/bash-tasks/.json — task metadata + status +//! harness_dir()/bash-tasks/.out — captured stdout +//! harness_dir()/bash-tasks/.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) { + 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::(&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) +} diff --git a/hive-c0re/src/build_logs.rs b/hive-c0re/src/build_logs.rs index 76679abc..608cb9a1 100644 --- a/hive-c0re/src/build_logs.rs +++ b/hive-c0re/src/build_logs.rs @@ -366,7 +366,8 @@ impl BuildLogs { } /// Drop rows past their retention window. Returns the number of - /// rows deleted. Called from this module's hourly `spawn` loop. + /// rows deleted. Called from the existing hourly vacuum loop — + /// see `events_vacuum` for the call site. /// /// Rule: /// - `status = 'fail'` rows kept for `KEEP_FAIL_SECS` past their @@ -394,10 +395,10 @@ impl BuildLogs { } } -/// 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. +/// 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. pub fn spawn_vacuum(coord: &Arc) { use std::time::Duration; let logs = coord.build_logs.clone(); diff --git a/hive-c0re/src/events_vacuum.rs b/hive-c0re/src/events_vacuum.rs new file mode 100644 index 00000000..56aaa933 --- /dev/null +++ b/hive-c0re/src/events_vacuum.rs @@ -0,0 +1,78 @@ +//! Host-side vacuum of every per-agent events.sqlite. The harness +//! writes to `/agents//harness/hyperhive-events.sqlite` +//! (bind-mounted from `/var/lib/hyperhive/agents//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) { + 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 { + 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)) +} diff --git a/hive-c0re/src/hive_stats.rs b/hive-c0re/src/hive_stats.rs index f3bd6e2e..6f19c841 100644 --- a/hive-c0re/src/hive_stats.rs +++ b/hive-c0re/src/hive_stats.rs @@ -12,8 +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` can read them fine, but cannot write/delete (which is why -//! retention sweeps run agent-side in the harness, not here). 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; diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index e58f2d0f..53f66241 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -17,6 +17,7 @@ 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; @@ -27,6 +28,7 @@ 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; diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 6cee9a28..0dbad747 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -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, broker, client, crash_watch, dashboard, dashboard_events, forge, - knowledge, matrix, migrate, rebuild_queue, reminder_scheduler, scheduled_prompts_worker, - server, socket_server, + 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, }; #[derive(Parser)] @@ -385,13 +385,15 @@ async fn cmd_serve( } } }); - // 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". + // Per-agent events.sqlite vacuum: host-side so the harness + // doesn't need any retention wiring of its own. + events_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); // 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