use chrono::{Duration, Utc};
use tracing::{debug, error, info, warn};
use super::driver::ManagedTmuxDriver;
use super::manager::{ManagedError, SessionManager};
use super::record::{ManagedSessionId, ManagedSessionState, SessionRecord};
use super::worktree_safety::{
DirtyWorktree, DirtyWorktreePolicy, dirt_blocks_removal, git_worktree_list_agrees, inspect_dirt,
};
#[path = "prune_types.rs"]
mod types;
pub use types::{PruneAction, PruneFilter, PruneOutcome, PrunedSession};
pub const MAX_EPHEMERAL_AGE_HOURS: i64 = 24;
const CANONICALIZE_FAILURE_STREAK_THRESHOLD: u32 = 10;
#[derive(Debug, Default)]
struct CanonicalizeFailureStreaks {
counts: std::collections::HashMap<std::path::PathBuf, u32>,
}
impl CanonicalizeFailureStreaks {
fn record_failure(&mut self, path: &std::path::Path) -> u32 {
let count = self.counts.entry(path.to_path_buf()).or_insert(0);
*count += 1;
*count
}
fn record_success(&mut self, path: &std::path::Path) {
self.counts.remove(path);
}
fn retain_active(&mut self, active: &std::collections::HashSet<std::path::PathBuf>) {
self.counts.retain(|path, _| active.contains(path));
}
}
static CANONICALIZE_FAILURE_STREAKS: std::sync::OnceLock<
std::sync::Mutex<CanonicalizeFailureStreaks>,
> = std::sync::OnceLock::new();
fn canonicalize_failure_streaks() -> &'static std::sync::Mutex<CanonicalizeFailureStreaks> {
CANONICALIZE_FAILURE_STREAKS
.get_or_init(|| std::sync::Mutex::new(CanonicalizeFailureStreaks::default()))
}
pub(super) fn is_running(
record: &SessionRecord,
tmux: &dyn ManagedTmuxDriver,
) -> Result<bool, ManagedError> {
tmux.session_exists_checked(&record.tmux_name)
}
fn is_tombstone(record: &SessionRecord) -> bool {
matches!(
record.state,
ManagedSessionState::Decommissioned | ManagedSessionState::Deleted
)
}
fn matches_filter(record: &SessionRecord, filter: PruneFilter) -> bool {
match filter {
PruneFilter::Ephemeral => {
record.ephemeral && record.state != ManagedSessionState::Decommissioned
}
PruneFilter::Stopped => record.state == ManagedSessionState::Stopped,
PruneFilter::Decommissioned => record.state == ManagedSessionState::Decommissioned,
PruneFilter::Deleted => record.state == ManagedSessionState::Deleted,
PruneFilter::All => true,
PruneFilter::Unresolvable => record.workspace_unresolvable(),
}
}
pub(crate) fn find_orphaned_worktrees(
repos_root: &std::path::Path,
active_set: &std::collections::HashSet<std::path::PathBuf>,
adopted: &[std::path::PathBuf],
) -> Vec<std::path::PathBuf> {
super::worktree_registry::enumerate_registered_worktrees(repos_root, adopted)
.into_iter()
.filter(|candidate| !active_set.contains(candidate))
.collect()
}
#[derive(Debug, Clone, Default)]
pub struct OrphanSweepOutcome {
pub removed: Vec<std::path::PathBuf>,
pub owner_unknown: Vec<std::path::PathBuf>,
pub skipped_dirty: Vec<DirtyWorktree>,
pub agent_owned: Vec<std::path::PathBuf>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct SweepClassification {
candidates: usize,
owner_unknown: usize,
agent_owned: usize,
skipped_live: usize,
skipped_dirty: usize,
reclaimable: usize,
}
impl SweepClassification {
fn summary(&self) -> Option<String> {
if self.candidates == 0 {
return None;
}
Some(format!(
"prune-worktrees: classified {} candidate(s) — {} reclaimable, \
{} owner-unknown, {} agent-owned, {} owner-live, {} dirty \
(per-path detail at debug; #4323)",
self.candidates,
self.reclaimable,
self.owner_unknown,
self.agent_owned,
self.skipped_live,
self.skipped_dirty,
))
}
}
#[cfg(test)]
mod sweep_summary_tests {
use super::SweepClassification;
#[test]
fn sweep_summary_is_silent_without_candidates() {
assert_eq!(SweepClassification::default().summary(), None);
}
#[test]
fn sweep_summary_names_every_count() {
let line = SweepClassification {
candidates: 200,
owner_unknown: 193,
agent_owned: 3,
skipped_live: 2,
skipped_dirty: 1,
reclaimable: 1,
}
.summary()
.expect("a sweep with candidates reports");
for (label, count) in [
("candidate(s)", "200"),
("owner-unknown", "193"),
("agent-owned", "3"),
("owner-live", "2"),
("dirty", "1"),
] {
assert!(
line.contains(count) && line.contains(label),
"summary must name {label}={count}: {line}"
);
}
}
}
impl SessionManager {
pub async fn decommission_all_ephemeral(&self) -> Result<usize, ManagedError> {
Ok(self.sweep_all_ephemeral(false).await?.count())
}
pub async fn sweep_all_ephemeral(&self, dry_run: bool) -> Result<PruneOutcome, ManagedError> {
self.prune_managed(PruneFilter::Ephemeral, dry_run, true, None, None)
.await
}
pub async fn prune_managed(
&self,
filter: PruneFilter,
dry_run: bool,
include_active: bool,
caller: Option<ManagedSessionId>,
invoker: Option<ManagedSessionId>,
) -> Result<PruneOutcome, ManagedError> {
self.store.write().await.reload_if_changed().await?;
let all = self.store.read().await.cached_all();
let tmux = self.tmux.as_ref();
let ignore_liveness = include_active || filter.selects_regardless_of_liveness();
let mut targets: Vec<SessionRecord> = Vec::new();
for record in all.into_iter().filter(|r| matches_filter(r, filter)) {
if invoker == Some(record.id) {
info!(id = %record.id, "prune: skipping the invoking session");
continue;
}
if ignore_liveness || !is_running(&record, tmux)? {
targets.push(record);
}
}
let mut names: Option<trusty_common::workspace_layout::WorktreeDirNames> = None;
let mut sessions = Vec::with_capacity(targets.len());
for record in targets {
let is_tombstone = is_tombstone(&record);
let mut action = if is_tombstone {
PruneAction::Removed
} else {
PruneAction::Decommissioned
};
let mut retained_workspace_path = None;
if !dry_run {
if is_tombstone {
let names = names.get_or_insert_with(super::decommission::worktree_dir_names);
match self.compact_and_release_slot(&record, names).await {
Ok(true) => {}
Ok(false) => continue,
Err(e) => {
warn!(id = %record.id, "prune: compaction failed: {e}; skipping");
continue;
}
}
} else {
match self.decommission(&record.id, caller).await {
Ok((tombstone, workspace_removed)) => {
if !workspace_removed && tombstone.workspace_path.is_some() {
action = PruneAction::DecommissionedWorktreeRetained;
retained_workspace_path = tombstone.workspace_path.clone();
}
}
Err(e) => {
warn!(id = %record.id, "prune: decommission failed: {e}; skipping");
continue;
}
}
}
}
sessions.push(PrunedSession {
id: record.id.to_string(),
tmux_name: record.tmux_name.clone(),
state: record.state.to_string(),
action,
retained_workspace_path,
});
}
if dry_run {
info!(
filter = %filter,
count = sessions.len(),
"prune (dry-run): reporting candidates, no mutation"
);
} else {
info!(
filter = %filter,
count = sessions.len(),
"prune: applied teardown/compaction"
);
}
Ok(PruneOutcome {
dry_run,
filter: filter.as_str().to_string(),
sessions,
})
}
async fn compact_and_release_slot(
&self,
snapshot: &SessionRecord,
names: &trusty_common::workspace_layout::WorktreeDirNames,
) -> Result<bool, ManagedError> {
let mut store = self.store.write().await;
store.reload_if_changed().await?;
let current = store.cached_get(&snapshot.id).ok();
let Some(record) = current.filter(is_tombstone) else {
drop(store);
info!(
id = %snapshot.id,
"prune: kept the record and its slot — the store holds no tombstone for it"
);
return Ok(false);
};
store.remove(&snapshot.id).await?;
drop(store);
if super::retention::workspace_needs_protection(
record.workspace_path.as_deref(),
names,
|p| p.try_exists(),
) {
info!(
id = %record.id,
"prune: compacted the record but kept its slot — its workspace still looks \
like a live session worktree"
);
return Ok(true);
}
if let Some(slot) = self.slots.write().await.release(&record.id) {
info!(id = %record.id, slot, "prune: freed the compacted record's slot");
}
Ok(true)
}
pub async fn compact_record(&self, id: &ManagedSessionId) -> Result<(), ManagedError> {
self.store.write().await.remove(id).await?;
Ok(())
}
pub async fn prune_orphaned_worktrees(
&self,
repos_root: &std::path::Path,
in_use_workspace_paths: &[std::path::PathBuf],
dry_run: bool,
policy: DirtyWorktreePolicy,
adopted: &[std::path::PathBuf],
) -> Result<OrphanSweepOutcome, anyhow::Error> {
use super::decommission::WorktreeRemoval;
use super::worktree_ownership::SentinelOwner;
use std::collections::HashSet;
let repos_root = repos_root.to_path_buf();
let initial_in_use: HashSet<std::path::PathBuf> = in_use_workspace_paths
.iter()
.map(|p| std::fs::canonicalize(p).unwrap_or_else(|_| p.clone()))
.collect();
let adopted = adopted.to_vec();
let candidates = tokio::task::spawn_blocking({
let initial_in_use = initial_in_use.clone();
move || find_orphaned_worktrees(&repos_root, &initial_in_use, &adopted)
})
.await
.map_err(|e| anyhow::anyhow!("prune-worktrees: orphan scan panicked: {e}"))?;
let mut owner_unknown = Vec::new();
let mut agent_owned = Vec::new();
let mut skipped_dirty: Vec<DirtyWorktree> = Vec::new();
let mut reclaimable = Vec::new();
let mut skipped_live = 0usize;
let total_candidates = candidates.len();
for candidate in candidates {
match super::worktree_ownership::read_sentinel_owner(&candidate) {
SentinelOwner::Unknown => {
debug!(
path = %candidate.display(),
"prune-worktrees: owner-unknown sentinel — never auto-deleting; \
run `tm doctor` or inspect manually (#3649)"
);
owner_unknown.push(candidate);
}
SentinelOwner::Agent(agent, _) => {
debug!(
path = %candidate.display(),
agent_id = %agent.agent_id,
"prune-worktrees: owned by a dispatched agent — reclaimed when that \
agent exits, never by this sweep (#4311)"
);
agent_owned.push(candidate);
}
SentinelOwner::Known(owner, created_at) => {
if !self.resolve_ownerless_with_grace(owner, created_at).await {
debug!(
path = %candidate.display(),
owner = %owner,
"prune-worktrees: owner is still live/resumable, or the sentinel \
is too young to rule out a creation race — skipping (#3649)"
);
skipped_live += 1;
continue;
}
if !git_worktree_list_agrees(&candidate) {
warn!(
path = %candidate.display(),
"prune-worktrees: git worktree list disagrees this path is a \
worktree — skipping conservatively (#3649)"
);
continue;
}
if let Some(dirt) = dirt_blocks_removal(&candidate, policy, "scan") {
skipped_dirty.push(dirt);
continue;
}
reclaimable.push(candidate);
}
}
}
let classification = SweepClassification {
candidates: total_candidates,
owner_unknown: owner_unknown.len(),
agent_owned: agent_owned.len(),
skipped_live,
skipped_dirty: skipped_dirty.len(),
reclaimable: reclaimable.len(),
};
if let Some(summary) = classification.summary() {
info!("{summary}");
}
if dry_run {
for p in &reclaimable {
info!(path = %p.display(), "prune-worktrees (dry-run): would remove orphaned worktree");
}
return Ok(OrphanSweepOutcome {
removed: reclaimable,
owner_unknown,
skipped_dirty,
agent_owned,
});
}
let fresh_in_use: HashSet<std::path::PathBuf> = {
let mut set = HashSet::new();
let mut checked_paths: HashSet<std::path::PathBuf> = HashSet::new();
for r in self.store.read().await.cached_all() {
let session_id = r.id;
let Some(p) = r.workspace_path else {
continue;
};
checked_paths.insert(p.clone());
if let Ok(c) = std::fs::canonicalize(&p) {
set.insert(c);
canonicalize_failure_streaks()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.record_success(&p);
} else {
let streak = canonicalize_failure_streaks()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.record_failure(&p);
if streak >= CANONICALIZE_FAILURE_STREAK_THRESHOLD {
error!(
session = %session_id,
path = %p.display(),
streak,
"prune-worktrees: active session path has failed to \
canonicalize for {streak} consecutive real-sweep \
observations — sustained failure, investigate before \
a stop/reap silently reconstitutes this workspace \
root (#3715)"
);
} else {
warn!(
path = %p.display(),
"prune-worktrees: active session path failed to canonicalize; \
using raw path as protective fallback (#1845 F3)"
);
}
}
set.insert(p);
}
canonicalize_failure_streaks()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.retain_active(&checked_paths);
set
};
let mut removed = Vec::new();
for candidate in reclaimable {
let canonical_candidate = match std::fs::canonicalize(&candidate) {
Ok(c) => c,
Err(_) => {
warn!(
path = %candidate.display(),
"prune-worktrees: skipping candidate — canonicalize failed"
);
continue;
}
};
if fresh_in_use.contains(&canonical_candidate) || fresh_in_use.contains(&candidate) {
info!(
path = %candidate.display(),
"prune-worktrees: skipping — active session appeared after initial snapshot"
);
continue;
}
if let Some(dirt) = dirt_blocks_removal(&candidate, policy, "pre-removal") {
skipped_dirty.push(dirt);
continue;
}
info!(path = %candidate.display(), "prune-worktrees: removing orphaned worktree");
let candidate_clone = candidate.clone();
let outcome = tokio::task::spawn_blocking(move || {
super::decommission::remove_session_worktree(
&candidate_clone,
"prune-worktrees orphan sweep: no live session claims this worktree",
)
})
.await
.unwrap_or_else(|e| {
tracing::error!("prune-worktrees: spawn_blocking panicked during removal: {e}");
WorktreeRemoval::Kept(format!("the removal task panicked: {e}"))
});
if let Some(reason) = outcome.reason() {
warn!(
path = %candidate.display(),
"prune-worktrees: worktree kept — {reason}"
);
}
if outcome.removed() {
removed.push(candidate);
}
}
Ok(OrphanSweepOutcome {
removed,
owner_unknown,
skipped_dirty,
agent_owned,
})
}
pub async fn reap_orphaned_worktrees(
&self,
repos_root: &std::path::Path,
adopted: &[std::path::PathBuf],
) -> Result<OrphanSweepOutcome, anyhow::Error> {
let in_use: Vec<std::path::PathBuf> = self
.list()
.await
.into_iter()
.filter_map(|r| r.workspace_path)
.collect();
let policy = DirtyWorktreePolicy::Skip;
self.prune_orphaned_worktrees(repos_root, &in_use, false, policy, adopted)
.await
}
pub async fn reap_aged_ephemeral(&self, max_age: Duration) -> Result<usize, ManagedError> {
let cutoff = Utc::now() - max_age;
self.store.write().await.reload_if_changed().await?;
let all = self.store.read().await.cached_all();
let stale: Vec<SessionRecord> = all
.into_iter()
.filter(|r| {
r.ephemeral
&& r.state != ManagedSessionState::Decommissioned
&& r.created_at < cutoff
})
.collect();
let mut reaped = 0usize;
for record in stale {
if let Some(dirt) = record.workspace_path.as_deref().and_then(inspect_dirt) {
warn!(
id = %record.id, path = %dirt.path.display(), reason = %dirt.reason,
"auto-reap: workspace holds unsaved work — leaving the session in \
place for an operator (#4118)"
);
continue;
}
match self.decommission(&record.id, None).await {
Ok(_) => {
reaped += 1;
info!(
id = %record.id,
name = %record.tmux_name,
created_at = %record.created_at.to_rfc3339(),
"auto-reap: decommissioned stale ephemeral session"
);
}
Err(e) => {
warn!(id = %record.id, "auto-reap: decommission failed: {e}; skipping");
}
}
}
Ok(reaped)
}
}
#[cfg(test)]
#[path = "prune_orphan_tests.rs"]
mod orphan_tests;
#[cfg(test)]
#[path = "prune_slot_tests.rs"]
mod slot_tests;
#[cfg(test)]
#[path = "prune_unresolvable_tests.rs"]
mod prune_unresolvable_tests;