hyperhive/hive-agent/src/vacuum.rs

//! 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;

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;
/// Keep delivered (soft-deleted, `sent_at` set) reminder rows this long
/// before reaping them — same window as `STREAM_KEEP_SECS`, kept around
/// only to serve the trailing-window `ReminderRollup` stats.
const REMINDER_KEEP_SECS: i64 = 14 * 24 * 3600;
/// Keep acked (agent-dismissed) todo rows this long before reaping them —
/// see `todos.rs`'s module doc for why acking doesn't delete outright. Long
/// enough that a genuinely quiet month is the only way to trigger the "one
/// spurious re-announcement" fallback path, short enough the table doesn't
/// grow meaningfully from one-shot todos that will never be upserted again.
const TODO_ACKED_KEEP_SECS: i64 = 30 * 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, chrono::Utc::now().timestamp() - 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"),
        }
    }

    let state_db = crate::paths::state_db();
    if state_db.exists() {
        match vacuum_reminders(&state_db) {
            Ok(0) => {}
            Ok(n) => tracing::info!(removed = n, "reminders vacuum"),
            Err(e) => tracing::warn!(error = ?e, "reminders vacuum failed"),
        }
        match vacuum_todos(&state_db) {
            Ok(0) => {}
            Ok(n) => tracing::info!(removed = n, "todos vacuum"),
            Err(e) => tracing::warn!(error = ?e, "todos vacuum failed"),
        }
    }
}

/// Reap delivered reminder rows older than [`REMINDER_KEEP_SECS`] via the
/// typed store API (own short-lived connection — mirrors `vacuum_events`'s
/// own connection to `events.sqlite` rather than sharing the harness's live
/// `Reminders` handle).
fn vacuum_reminders(path: &Path) -> anyhow::Result<usize> {
    let store = crate::reminders::Reminders::open(path)?;
    store.prune_delivered_older_than(chrono::Utc::now().timestamp() - REMINDER_KEEP_SECS)
}

/// Reap acked todo rows older than [`TODO_ACKED_KEEP_SECS`] via the typed
/// store API — same own-short-lived-connection shape as `vacuum_reminders`.
fn vacuum_todos(path: &Path) -> anyhow::Result<usize> {
    let store = crate::todos::Todos::open(path)?;
    store.reap_acked(chrono::Utc::now().timestamp() - TODO_ACKED_KEEP_SECS)
}

/// 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;
    }
    // `completed_at` serializes as an RFC3339 string (`TaskFile.completed_at`
    // is `Option<DateTime<Utc>>`), not a bare epoch-seconds integer — parse
    // it the same way, falling back to "never expired" on anything
    // unparseable so a corrupt/legacy field never causes a premature delete.
    let completed_at = v
        .get("completed_at")
        .and_then(serde_json::Value::as_str)
        .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
        .map_or(i64::MAX, |dt| dt.timestamp());
    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 = chrono::Utc::now().timestamp() - 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))
}