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>,
pub removed_sessions: Vec<crate::types::SessionId>,
pub would_remove_sessions: Vec<crate::types::SessionId>,
}
const SECONDS_PER_MINUTE: u64 = 60;
const SECONDS_PER_HOUR: u64 = 60 * SECONDS_PER_MINUTE;
const SECONDS_PER_DAY: u64 = 24 * SECONDS_PER_HOUR;
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))
.filter(|seconds| *seconds <= i64::MAX as u64)
.ok_or_else(error)?;
Ok(Duration::from_secs(seconds))
}
pub fn prune(
state: &Path,
repo: Option<&Path>,
older_than: Option<Duration>,
dry_run: bool,
) -> Result<PruneReport, String> {
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)?;
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| {
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)?;
}
}
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> {
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(())
}