artifactize_core/store/

prune.rs

use std::{
    fs,
    path::{Path, PathBuf},
    time::Duration,
};

use rusqlite::OpenFlags;
use serde::Serialize;
use time::OffsetDateTime;

use super::{DATABASE, STATE_SCHEMA_VERSION, receipts::check_files};
use crate::{platform, process, workspace};

#[derive(Default, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct PruneReport {
    #[serde(serialize_with = "crate::platform::serialize_paths")]
    pub removed: Vec<PathBuf>,
    #[serde(serialize_with = "crate::platform::serialize_paths")]
    pub would_remove: Vec<PathBuf>,
    pub skipped_runs: Vec<crate::types::RunId>,
    /// Agent sessions the size-bound collection deleted, oldest first.
    pub removed_sessions: Vec<crate::types::SessionId>,
    /// With `--dry-run`, the Agent sessions it would delete.
    pub would_remove_sessions: Vec<crate::types::SessionId>,
}

// `--older-than` units in seconds, each built from the one below it. A day is a fixed
// 24 hours because pruning compares elapsed time, not calendar dates.
/// 60 seconds in a minute.
const SECONDS_PER_MINUTE: u64 = 60;
/// 60 minutes in an hour.
const SECONDS_PER_HOUR: u64 = 60 * SECONDS_PER_MINUTE;
/// 24 hours in a day.
const SECONDS_PER_DAY: u64 = 24 * SECONDS_PER_HOUR;
/// 7 days in a week.
const SECONDS_PER_WEEK: u64 = 7 * SECONDS_PER_DAY;

pub fn parse_duration(value: &str) -> Result<Duration, String> {
    let error = || "Use a whole-number duration with s, m, h, d or w (for example 7d).".to_owned();
    let split = value
        .find(|c: char| !c.is_ascii_digit())
        .ok_or_else(error)?;
    let multiplier = match &value[split..] {
        "s" => 1,
        "m" => SECONDS_PER_MINUTE,
        "h" => SECONDS_PER_HOUR,
        "d" => SECONDS_PER_DAY,
        "w" => SECONDS_PER_WEEK,
        _ => return Err(error()),
    };
    let seconds = value[..split]
        .parse::<u64>()
        .ok()
        .and_then(|amount| amount.checked_mul(multiplier))
        // `time::Duration` holds whole seconds as an `i64`; the cutoff converts into it.
        .filter(|seconds| *seconds <= i64::MAX as u64)
        .ok_or_else(error)?;
    Ok(Duration::from_secs(seconds))
}

/// Prune only known scratch directories. Terminal Runs never reacquire an owner.
pub fn prune(
    state: &Path,
    repo: Option<&Path>,
    older_than: Option<Duration>,
    dry_run: bool,
) -> Result<PruneReport, String> {
    // Canonical like the repositories it is compared with: on Windows, a state path typed in
    // another case, or with 8.3 short names, would otherwise slip past `outside_workspace`.
    // System aliases such as macOS /tmp are OS-owned boundaries, not operator-made links.
    let state = platform::resolve_system_aliases(state).map_err(|e| e.to_string())?;
    let state = workspace::canonical_target(&real_path(&state)?).map_err(|e| e.to_string())?;
    let mut report = PruneReport::default();
    let mut repositories = Vec::new();
    if let Some(repo) = repo {
        repositories.push(workspace::canonical_target(repo).map_err(|e| e.to_string())?);
    }
    for repo in &repositories {
        workspace::outside_workspace(repo, &state).map_err(|e| e.to_string())?;
    }
    check_files(&state)?;
    // Saved Agent conversations past their size bound (limits.json agentSessions).
    let bounds = crate::limits::Limits::read(&state)?.agent_sessions();
    let sessions = crate::agent::session::collect(&state, bounds, dry_run)?;
    if dry_run {
        report.would_remove_sessions = sessions.removed.into_iter().collect();
    } else {
        report.removed_sessions = sessions.removed.into_iter().collect();
    }
    if !state
        .join(DATABASE)
        .try_exists()
        .map_err(|e| e.to_string())?
    {
        return Ok(report);
    }
    let mut db = rusqlite::Connection::open_with_flags(
        state.join(DATABASE),
        OpenFlags::SQLITE_OPEN_READ_ONLY,
    )
    .map_err(|e| e.to_string())?;
    db.busy_timeout(super::SQLITE_BUSY_TIMEOUT)
        .map_err(|e| e.to_string())?;
    let transaction = db.transaction().map_err(|e| e.to_string())?;
    let version: u32 = transaction
        .pragma_query_value(None, "user_version", |row| row.get(0))
        .map_err(|e| e.to_string())?;
    if version != STATE_SCHEMA_VERSION {
        return Err(super::schema_error(version));
    }
    let cutoff = older_than
        .map(|age| {
            time::Duration::try_from(age)
                .ok()
                .and_then(|age| OffsetDateTime::now_utc().checked_sub(age))
                .ok_or_else(|| "Prune duration is too large.".to_owned())
        })
        .transpose()?;
    let runs = {
        let mut statement = transaction
            .prepare("SELECT id,repo,data,status FROM runs ORDER BY id")
            .map_err(|e| e.to_string())?;
        statement
            .query_map([], |row| {
                Ok((
                    row.get::<_, crate::types::RunId>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, crate::types::RunStatus>(3)?,
                ))
            })
            .map_err(|e| e.to_string())?
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| e.to_string())?
    };
    let mut eligible = Vec::new();
    for (id, repo, data, status) in runs {
        let repo = workspace::canonical_target(Path::new(&repo)).map_err(|e| e.to_string())?;
        workspace::outside_workspace(&repo, &state).map_err(|e| e.to_string())?;
        repositories.push(repo);
        let Some(run) = super::unreadable::evidence::<super::Run>("run", &data) else {
            report.skipped_runs.push(id);
            continue;
        };
        let finished = run.completed_at().map(crate::types::Timestamp::time);
        let requests_terminal: bool = transaction
            .query_row(
                "SELECT NOT EXISTS(SELECT 1 FROM requests WHERE run_id=? AND status NOT IN ('GREEN','RED','ERROR','INCOMPLETE','BLOCKED','BUDGET_EXHAUSTED'))",
                [&id],
                |row| row.get(0),
            )
            .map_err(|e| e.to_string())?;
        let owners = {
            let mut statement = transaction
                .prepare(
                    "SELECT owner_pid,owner_start_time,status FROM executions WHERE json_extract(data,'$.provenance.runId')=?1 OR id IN (SELECT execution_id FROM requests WHERE run_id=?1)",
                )
                .map_err(|e| e.to_string())?;
            statement
                .query_map([&id], |row| {
                    // A mirrored remote result has no local owner: a saved pid of 0.
                    let pid: u32 = row.get(0)?;
                    let start_time = row.get::<_, i64>(1)? as u64;
                    let owner = std::num::NonZeroU32::new(pid).map(|pid| process::ChildIdentity {
                        pid: process::ProcessId::from(pid),
                        start_time,
                    });
                    Ok((owner, row.get::<_, crate::types::ExecutionStatus>(2)?))
                })
                .map_err(|e| e.to_string())?
                .collect::<Result<Vec<_>, _>>()
                .map_err(|e| e.to_string())?
        };
        let mut owned = false;
        for (owner, status) in owners {
            owned |= !matches!(
                status,
                crate::types::ExecutionStatus::Green
                    | crate::types::ExecutionStatus::Red
                    | crate::types::ExecutionStatus::Error
            ) || process::is_alive(owner).map_err(|e| e.to_string())?;
        }
        if !matches!(
            status,
            crate::types::RunStatus::Green
                | crate::types::RunStatus::Red
                | crate::types::RunStatus::Error
                | crate::types::RunStatus::Incomplete
        ) || finished.is_none()
            || cutoff.is_some_and(|cutoff| finished.is_some_and(|date| date > cutoff))
            || !requests_terminal
            || owned
        {
            report.skipped_runs.push(id);
        } else {
            eligible.push(id);
        }
    }
    transaction.commit().map_err(|e| e.to_string())?;
    drop(db);

    let runs = state.join("runs");
    if !directory(&runs)? {
        return Ok(report);
    }
    let mut targets = Vec::new();
    for id in eligible {
        let path = runs.join(id.as_str());
        if directory(&path)? {
            collect(&path, &mut targets)?;
        }
    }
    // Validate the complete deletion set before removing anything; never follow links.
    for path in &targets {
        validate_tree(path, &runs, &repositories)?;
    }
    for path in targets {
        if dry_run {
            report.would_remove.push(path);
        } else {
            real_path(&path)?;
            validate_tree(&path, &runs, &repositories)?;
            fs::remove_dir_all(&path)
                .map_err(|e| format!("Cannot prune {}: {e}", crate::platform::path_text(&path)))?;
            report.removed.push(path);
        }
    }
    Ok(report)
}

fn segment(value: &str) -> bool {
    !value.is_empty()
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
}

fn prefixed(name: &str, prefix: &str) -> bool {
    name.strip_prefix(prefix).is_some_and(segment)
}

fn collect(parent: &Path, targets: &mut Vec<PathBuf>) -> Result<(), String> {
    let parent_handle = platform::open_directory(parent).map_err(|e| e.to_string())?;
    let mut entries = platform::read_dir(&parent_handle)
        .map_err(|e| e.to_string())?
        .collect::<Result<Vec<_>, _>>()
        .map_err(|e| e.to_string())?;
    entries.sort_by_key(|entry| entry.file_name());
    for entry in entries {
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        let path = parent.join(entry.file_name());
        if matches!(name, "output" | "tmp" | "home" | "cache" | "human-tools")
            || ["tool-output-", "tool-"]
                .iter()
                .any(|prefix| prefixed(name, prefix))
        {
            if directory(&path)? {
                targets.push(path);
            }
        } else if ["runtime-", "fingerprint-"]
            .iter()
            .any(|prefix| prefixed(name, prefix))
            && directory(&path)?
        {
            collect(&path, targets)?;
        }
    }
    Ok(())
}

fn directory(path: &Path) -> Result<bool, String> {
    match platform::path_kind(path) {
        Ok(platform::FileKind::Directory) => {
            reject_repository(path)?;
            Ok(true)
        }
        Ok(_) => Err(format!(
            "Prune requires a real directory, not a symlink or file: {}",
            crate::platform::path_text(path)
        )),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
        Err(error) => Err(error.to_string()),
    }
}

fn real_path(path: &Path) -> Result<PathBuf, String> {
    platform::paths::absolute_without_links(path).map_err(|refusal| match refusal {
        platform::paths::LinkedPath::ParentTraversal => {
            "Prune paths must not contain parent traversal.".to_owned()
        }
        platform::paths::LinkedPath::Link(path) => {
            format!("Prune refuses symlinks: {}", platform::path_text(&path))
        }
        platform::paths::LinkedPath::Io(error) => error.to_string(),
    })
}

fn reject_repository(path: &Path) -> Result<(), String> {
    if workspace::has_artifact_marker(path).map_err(|error| error.to_string())? {
        return Err(format!(
            "Prune refuses repository content: {}",
            crate::platform::path_text(path)
        ));
    }
    match platform::path_kind(&path.join(".git")) {
        Ok(_) => Err(format!(
            "Prune refuses repository content: {}",
            crate::platform::path_text(path)
        )),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error.to_string()),
    }
}

fn validate_tree(path: &Path, runs: &Path, repositories: &[PathBuf]) -> Result<(), String> {
    // A link is refused first: the identity checks below would see where it leads.
    let kind = platform::path_kind(path).map_err(|e| e.to_string())?;
    if matches!(
        kind,
        platform::FileKind::Symlink | platform::FileKind::Other
    ) {
        return Err(format!(
            "Prune refuses symlinks: {}",
            crate::platform::path_text(path)
        ));
    }
    if !platform::is_within(path, runs)
        || platform::paths_equal(path, runs)
        || repositories
            .iter()
            .any(|repo| platform::is_within(path, repo) || platform::is_within(repo, path))
    {
        return Err(format!(
            "Prune target is outside Run output or overlaps a repository: {}",
            crate::platform::path_text(path)
        ));
    }
    if kind == platform::FileKind::Directory {
        reject_repository(path)?;
        let directory = platform::open_directory(path).map_err(|e| e.to_string())?;
        for entry in platform::read_dir(&directory).map_err(|e| e.to_string())? {
            validate_tree(
                &path.join(entry.map_err(|e| e.to_string())?.file_name()),
                runs,
                repositories,
            )?;
        }
    }
    Ok(())
}