diff --git a/hive-c0re/src/bash_tasks_vacuum.rs b/hive-c0re/src/bash_tasks_vacuum.rs new file mode 100644 index 00000000..545faf26 --- /dev/null +++ b/hive-c0re/src/bash_tasks_vacuum.rs @@ -0,0 +1,123 @@ +//! 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` / `stats_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_secs(3600); +/// 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(|t| t.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() { + if 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) +}