fix(#942): host-side bash-tasks vacuum, 48h retention

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/<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` / `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<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(|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)
+}