rchd 2.1.20

//! Worker cache cleanup scheduler.
//!
//! Periodically cleans up old project caches on workers to prevent disk exhaustion.
//! Respects worker busy state and configurable thresholds.

#![allow(dead_code)] // Scaffold code - will be wired into main.rs

use crate::config::CacheCleanupConfig;
use crate::workers::{WorkerPool, WorkerState};
use rch_common::{SshClient, SshOptions, WorkerId, WorkerStatus};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tokio::time::interval;
use tracing::{debug, info, warn};

/// Result of a cleanup operation on a single worker.
#[derive(Debug, Clone)]
pub struct CleanupResult {
    /// Worker that was cleaned.
    pub worker_id: WorkerId,
    /// Whether cleanup was successful.
    pub success: bool,
    /// Number of directories removed (if known).
    pub dirs_removed: Option<u64>,
    /// Bytes freed (if known).
    pub bytes_freed: Option<u64>,
    /// Duration of cleanup operation.
    pub duration: Duration,
    /// Error message if cleanup failed.
    pub error: Option<String>,
}

/// Stats for a cleanup run across all workers.
#[derive(Debug, Default)]
pub struct CleanupStats {
    /// Number of workers checked.
    pub workers_checked: u32,
    /// Number of workers cleaned.
    pub workers_cleaned: u32,
    /// Number of workers skipped (busy or unhealthy).
    pub workers_skipped: u32,
    /// Number of cleanup errors.
    pub errors: u32,
    /// Total bytes freed across all workers.
    pub total_bytes_freed: u64,
    /// Total directories removed.
    pub total_dirs_removed: u64,
}

#[derive(Debug, Clone)]
struct CleanupMetrics {
    removed_dirs: u64,
    freed_bytes: u64,
    free_before_gb: f64,
    free_after_gb: f64,
    low_disk_mode: bool,
    remove_errors: u64,
}

fn cleanup_threshold_kb(min_free_gb: u64) -> u64 {
    min_free_gb.saturating_mul(1024).saturating_mul(1024)
}

/// Where the remote launcher publishes process records (`transfer.rs`).
const REMOTE_RUN_ROOT: &str = "/tmp/rch-run";

fn build_cleanup_command(
    escaped_base: &str,
    escaped_run_root: &str,
    max_cache_age_hours: u64,
    min_free_gb: u64,
) -> String {
    let max_age_minutes = max_cache_age_hours.saturating_mul(60);
    let threshold_kb = cleanup_threshold_kb(min_free_gb);
    let active_grace_minutes = 5_u64;
    // Worker-side orphan watchdog: a job whose process record is older than
    // this and whose recorded leader is still alive escaped the in-session
    // watchdog (ssh drop or daemon restart). 240 min is far above any
    // legitimate test cap (default 30 min). The shared identity helpers bind
    // the group to worker boot and leader start time, so a reused pgid never
    // matches. A record whose group is gone, whose leader no longer matches,
    // or which is malformed can authorize nothing and is deleted; a failed
    // process observation keeps it for the next cycle.
    let orphan_after_minutes = 240_u64;
    // Run directories hold only records; an empty one untouched for a day is
    // garbage. A launch racing its removal fails typed setup and fails over.
    let empty_run_dir_minutes = 1440_u64;
    format!(
        "set -u; \
         base={base}; \
         max_age_minutes={max_age_minutes}; \
         threshold_kb={threshold_kb}; \
         active_grace_minutes={active_grace_minutes}; \
         orphan_after_minutes={orphan_after_minutes}; \
         if [ ! -d \"$base\" ]; then mkdir -p \"$base\"; fi; \
         if physical=$(cd \"$base\" 2>/dev/null && pwd -P); then base=$physical; fi; \
         before_kb=$(df -Pk \"$base\" 2>/dev/null | awk 'NR==2 {{print $4}}'); \
         if [ -z \"$before_kb\" ]; then before_kb=0; fi; \
         removed=0; freed_kb=0; low_disk=0; remove_errors=0; orphans_killed=0; \
         if [ \"$before_kb\" -lt \"$threshold_kb\" ]; then low_disk=1; fi; \
         run_root={run_root}; \
         if [ -d \"$run_root\" ]; then \
           tally=$(mktemp /tmp/rch-cleanup.XXXXXX); \
           (set -- \"$run_root\"/*/*.pgid\n{identity}\n killed=0; \
             for pgf in \"$@\"; do \
               [ -f \"$pgf\" ] && [ ! -L \"$pgf\" ] || continue; \
               [ -n \"$(find \"$pgf\" -mmin +\"$orphan_after_minutes\" -print 2>/dev/null)\" ] || continue; \
               build=${{pgf##*/}}; build=${{build%.pgid}}; verdict=stale; \
               if rch_read_record \"$pgf\" \"$build\"; then \
                 rch_group_state; \
                 case $? in \
                   0) ;; \
                   1) if rch_signal_group KILL; then verdict=killed; fi ;; \
                   *) verdict=unknown ;; \
                 esac; \
               fi; \
               case $verdict in \
                 stale) rm -f -- \"$pgf\" 2>/dev/null ;; \
                 killed) killed=$((killed + 1)) ;; \
               esac; \
             done; \
             echo \"$killed\") > \"$tally\" 2>/dev/null; \
           orphans_killed=$(cat \"$tally\" 2>/dev/null); rm -f \"$tally\"; \
           case \"$orphans_killed\" in ''|*[!0-9]*) orphans_killed=0 ;; esac; \
           find \"$run_root\" -mindepth 1 -maxdepth 1 -type d -empty -mmin +{empty_run_dir_minutes} -exec rmdir -- {{}} + 2>/dev/null || :; \
         fi; \
         cwds=$(mktemp /tmp/rch-cleanup.XXXXXX); \
         find /proc -mindepth 2 -maxdepth 2 -name cwd -printf '%l\\n' 2>/dev/null > \"$cwds\" || :; \
         candidates=$(mktemp /tmp/rch-cleanup.XXXXXX); \
         if [ \"$low_disk\" -eq 1 ]; then \
           find \"$base\" -mindepth 2 -maxdepth 2 -type d -printf '%T@ %p\\n' 2>/dev/null \
             | sort -n | sed 's/^[^ ]* //' > \"$candidates\"; \
         else \
           find \"$base\" -mindepth 2 -maxdepth 2 -type d -mmin +\"$max_age_minutes\" -print 2>/dev/null > \"$candidates\"; \
         fi; \
         while IFS= read -r dir; do \
           [ -z \"$dir\" ] && continue; \
           [ ! -d \"$dir\" ] && continue; \
           case \"$dir\" in \"$base\"/*) ;; *) continue ;; esac; \
           recent_active=$(find \"$dir\" -type f -mmin -\"$active_grace_minutes\" -print -quit 2>/dev/null || true); \
           if [ -n \"$recent_active\" ]; then continue; fi; \
           if D=\"$dir\" awk 'BEGIN {{ d = ENVIRON[\"D\"] }} $0 == d || index($0, d \"/\") == 1 {{ f = 1; exit }} END {{ exit !f }}' \"$cwds\"; then continue; fi; \
           size_kb=$(du -sk \"$dir\" 2>/dev/null | awk '{{print $1}}'); \
           if [ -z \"$size_kb\" ]; then size_kb=0; fi; \
         if rm -rf \"$dir\" 2>/dev/null; then \
           removed=$((removed + 1)); \
           freed_kb=$((freed_kb + size_kb)); \
         elif [ ! -d \"$dir\" ]; then \
           removed=$((removed + 1)); \
           freed_kb=$((freed_kb + size_kb)); \
         elif [ -z \"$(find \"$dir\" -mindepth 1 -print -quit 2>/dev/null)\" ]; then \
           if rmdir \"$dir\" 2>/dev/null; then \
             removed=$((removed + 1)); \
             freed_kb=$((freed_kb + size_kb)); \
             printf 'RCH_CLEANUP_HUSK_REAPED dir=%s\\n' \"$dir\" >&2; \
           else \
             remove_errors=$((remove_errors + 1)); \
             printf 'RCH_CLEANUP_RM_FAILURE dir=%s kb=%s\\n' \"$dir\" \"$size_kb\" >&2; \
           fi; \
         else \
           remove_errors=$((remove_errors + 1)); \
           printf 'RCH_CLEANUP_RM_FAILURE dir=%s kb=%s\\n' \"$dir\" \"$size_kb\" >&2; \
         fi; \
           if [ \"$low_disk\" -eq 1 ]; then \
             current_kb=$(df -Pk \"$base\" 2>/dev/null | awk 'NR==2 {{print $4}}'); \
             if [ -n \"$current_kb\" ] && [ \"$current_kb\" -ge \"$threshold_kb\" ]; then break; fi; \
           fi; \
         done < \"$candidates\"; \
         rm -f \"$candidates\" \"$cwds\"; \
         after_kb=$(df -Pk \"$base\" 2>/dev/null | awk 'NR==2 {{print $4}}'); \
         if [ -z \"$after_kb\" ]; then after_kb=0; fi; \
         printf 'RCH_CLEANUP_METRICS removed=%s freed_kb=%s before_kb=%s after_kb=%s low_disk=%s remove_errors=%s orphans_killed=%s\\n' \"$removed\" \"$freed_kb\" \"$before_kb\" \"$after_kb\" \"$low_disk\" \"$remove_errors\" \"$orphans_killed\"; \
         if [ \"$remove_errors\" -gt 0 ]; then exit 1; fi",
        base = escaped_base,
        run_root = escaped_run_root,
        max_age_minutes = max_age_minutes,
        threshold_kb = threshold_kb,
        active_grace_minutes = active_grace_minutes,
        orphan_after_minutes = orphan_after_minutes,
        empty_run_dir_minutes = empty_run_dir_minutes,
        identity = rch_common::REMOTE_PROCESS_IDENTITY_SCRIPT,
    )
}

fn parse_cleanup_metrics(stdout: &str) -> Option<CleanupMetrics> {
    let line = stdout
        .lines()
        .find(|line| line.contains("RCH_CLEANUP_METRICS"))?;

    let mut removed_dirs = None;
    let mut freed_kb = None;
    let mut before_kb = None;
    let mut after_kb = None;
    let mut low_disk = None;
    let mut remove_errors = None;

    for token in line.split_whitespace().skip(1) {
        let Some((key, value)) = token.split_once('=') else {
            continue;
        };
        match key {
            "removed" => removed_dirs = value.parse::<u64>().ok(),
            "freed_kb" => freed_kb = value.parse::<u64>().ok(),
            "before_kb" => before_kb = value.parse::<u64>().ok(),
            "after_kb" => after_kb = value.parse::<u64>().ok(),
            "low_disk" => low_disk = Some(value == "1"),
            "remove_errors" => remove_errors = value.parse::<u64>().ok(),
            _ => {}
        }
    }

    Some(CleanupMetrics {
        removed_dirs: removed_dirs.unwrap_or(0),
        freed_bytes: freed_kb.unwrap_or(0).saturating_mul(1024),
        free_before_gb: before_kb.unwrap_or(0) as f64 / (1024.0 * 1024.0),
        free_after_gb: after_kb.unwrap_or(0) as f64 / (1024.0 * 1024.0),
        low_disk_mode: low_disk.unwrap_or(false),
        remove_errors: remove_errors.unwrap_or(0),
    })
}

/// Cache cleanup scheduler service.
pub struct CacheCleanupScheduler {
    /// Worker pool reference.
    pool: WorkerPool,
    /// Configuration.
    config: CacheCleanupConfig,
    /// Last cleanup time per worker.
    last_cleanup: Arc<RwLock<HashMap<WorkerId, Instant>>>,
    /// First observed idle instant per worker (used for idle-threshold gating).
    idle_since: Arc<RwLock<HashMap<WorkerId, Instant>>>,
    /// SSH options for cleanup commands (throwaway fallback path).
    ssh_options: SshOptions,
    /// Shared SSH connection pool. When `Some`, cleanup commands run over a warm
    /// reused ControlMaster; when `None`, the legacy throwaway path is used.
    ssh_pool: Option<Arc<rch_common::SshPool>>,
}

impl CacheCleanupScheduler {
    /// Create a new cache cleanup scheduler.
    pub fn new(pool: WorkerPool, config: CacheCleanupConfig) -> Self {
        Self {
            pool,
            config,
            last_cleanup: Arc::new(RwLock::new(HashMap::new())),
            idle_since: Arc::new(RwLock::new(HashMap::new())),
            ssh_options: SshOptions::default(),
            ssh_pool: None,
        }
    }

    /// Attach a shared SSH connection pool for warm ControlMaster reuse.
    #[must_use]
    pub fn with_ssh_pool(mut self, pool: Option<Arc<rch_common::SshPool>>) -> Self {
        self.ssh_pool = pool;
        self
    }

    /// Run one remote command against a worker, using the shared pool (warm
    /// master reuse) when configured, otherwise a throwaway SSH session.
    async fn run_remote(
        &self,
        config: &rch_common::WorkerConfig,
        command: &str,
    ) -> anyhow::Result<rch_common::CommandResult> {
        if let Some(pool) = &self.ssh_pool {
            return pool
                .run_with_timeout(config, command, self.ssh_options.command_timeout)
                .await;
        }
        let mut ssh_client = SshClient::new(config.clone(), self.ssh_options.clone());
        ssh_client.connect().await?;
        let result = ssh_client.execute(command).await;
        let _ = ssh_client.disconnect().await;
        result
    }

    /// Start the cleanup scheduler.
    ///
    /// Returns a handle to the spawned task.
    pub fn start(self: Arc<Self>) -> tokio::task::JoinHandle<()> {
        let scheduler = self;
        tokio::spawn(async move {
            if !scheduler.config.enabled {
                info!("Cache cleanup scheduler disabled");
                return;
            }

            info!(
                "Cache cleanup scheduler started (interval={}s, max_age={}h, min_free={}GB)",
                scheduler.config.interval_secs,
                scheduler.config.max_cache_age_hours,
                scheduler.config.min_free_gb
            );

            let mut ticker = interval(Duration::from_secs(scheduler.config.interval_secs));
            loop {
                ticker.tick().await;
                let stats = scheduler.run_cleanup_cycle().await;
                if stats.workers_cleaned > 0 || stats.errors > 0 {
                    info!(
                        "Cache cleanup cycle: checked={}, cleaned={}, skipped={}, errors={}, freed={}MB",
                        stats.workers_checked,
                        stats.workers_cleaned,
                        stats.workers_skipped,
                        stats.errors,
                        stats.total_bytes_freed / (1024 * 1024)
                    );
                } else {
                    debug!("Cache cleanup cycle: no action needed");
                }
            }
        })
    }

    /// Run a single cleanup cycle across all workers.
    async fn run_cleanup_cycle(&self) -> CleanupStats {
        let mut stats = CleanupStats::default();
        let workers = self.pool.all_workers().await;

        for worker_state in workers {
            stats.workers_checked += 1;
            let worker_id = worker_state.config.read().await.id.clone();

            // Check if worker is eligible for cleanup
            if !self.is_worker_eligible(&worker_state).await {
                debug!("Worker {} not eligible for cleanup", worker_id);
                stats.workers_skipped += 1;
                continue;
            }

            // Perform cleanup
            match self.cleanup_worker(&worker_state).await {
                Ok(result) => {
                    if result.success {
                        stats.workers_cleaned += 1;
                        if let Some(bytes) = result.bytes_freed {
                            stats.total_bytes_freed += bytes;
                        }
                        if let Some(dirs) = result.dirs_removed {
                            stats.total_dirs_removed += dirs;
                        }
                    } else {
                        stats.errors += 1;
                    }
                }
                Err(e) => {
                    warn!("Failed to cleanup worker {}: {}", worker_id, e);
                    stats.errors += 1;
                }
            }
        }

        stats
    }

    /// Check if a worker is eligible for cleanup.
    async fn is_worker_eligible(&self, worker_state: &WorkerState) -> bool {
        let worker_id = worker_state.config.read().await.id.clone();

        // Check worker status
        let status = worker_state.status().await;
        if status != WorkerStatus::Healthy {
            debug!(
                "Worker {:?} not healthy (status={:?}), skipping cleanup",
                worker_id, status
            );
            self.idle_since.write().await.remove(&worker_id);
            return false;
        }

        // Check if worker is busy (has active slots)
        let used_slots = worker_state.used_slots();
        if used_slots != 0 {
            debug!(
                "Worker {} is busy ({} slots reserved), skipping cleanup",
                worker_id, used_slots
            );
            self.idle_since.write().await.remove(&worker_id);
            return false;
        }

        // Enforce idle threshold before cleanup.
        let idle_threshold = Duration::from_secs(self.config.idle_threshold_secs);
        if !idle_threshold.is_zero() {
            let now = Instant::now();
            let idle_for = {
                let mut idle_since = self.idle_since.write().await;
                let first_idle = idle_since.entry(worker_id.clone()).or_insert(now);
                now.saturating_duration_since(*first_idle)
            };
            if idle_for < idle_threshold {
                debug!(
                    "Worker {} idle for {:?} (threshold {:?}), skipping cleanup",
                    worker_id, idle_for, idle_threshold
                );
                return false;
            }
        }

        // Check circuit breaker state
        let circuit_state = worker_state.circuit_state().await;
        if circuit_state != Some(rch_common::CircuitState::Closed) {
            debug!(
                "Worker {} circuit not closed ({:?}), skipping cleanup",
                worker_id, circuit_state
            );
            return false;
        }

        true
    }

    /// Perform cleanup on a single worker.
    async fn cleanup_worker(&self, worker_state: &WorkerState) -> anyhow::Result<CleanupResult> {
        let start = Instant::now();
        let config = worker_state.config.read().await.clone();

        info!(
            "Starting cache cleanup on worker {} (max_age={}h, min_free={}GB)",
            config.id, self.config.max_cache_age_hours, self.config.min_free_gb
        );

        // Build cleanup command:
        // - baseline prune by age
        // - when below disk threshold, escalate and prune oldest caches until threshold is met
        let remote_base = &self.config.remote_base;
        let escaped_base = rch_common::ssh_utils::shell_escape_path_with_home(remote_base)
            .ok_or_else(|| anyhow::anyhow!("Invalid remote_base: contains control characters"))?;
        let cleanup_script = build_cleanup_command(
            escaped_base.as_ref(),
            REMOTE_RUN_ROOT,
            self.config.max_cache_age_hours,
            self.config.min_free_gb,
        );
        // The worker's login shell may be zsh, which neither word-splits nor
        // tolerates unmatched globs the way the POSIX script expects.
        let cleanup_cmd = format!("sh -c {}", shell_escape::escape(cleanup_script.into()));

        // Execute cleanup via SSH (pooled warm master when available).
        let result = self.run_remote(&config, &cleanup_cmd).await?;

        // Update last cleanup time
        {
            let mut last_cleanup = self.last_cleanup.write().await;
            last_cleanup.insert(config.id.clone(), Instant::now());
        }

        let duration = start.elapsed();
        let metrics = parse_cleanup_metrics(&result.stdout);
        let metrics_summary = metrics
            .as_ref()
            .map(|m| {
                format!(
                    "removed={}, freed_mb={}, before_free_gb={:.2}, after_free_gb={:.2}, low_disk={}, remove_errors={}",
                    m.removed_dirs,
                    m.freed_bytes / (1024 * 1024),
                    m.free_before_gb,
                    m.free_after_gb,
                    m.low_disk_mode,
                    m.remove_errors
                )
            })
            .unwrap_or_else(|| "metrics=unavailable".to_string());

        if result.exit_code == 0 {
            info!(
                "Cache cleanup completed on worker {} in {:?} ({})",
                config.id, duration, metrics_summary
            );
            if let Some(ref parsed) = metrics
                && parsed.free_after_gb < self.config.min_free_gb as f64
            {
                warn!(
                    "Worker {} still below free-space threshold after cleanup: {:.2}GB < {}GB",
                    config.id, parsed.free_after_gb, self.config.min_free_gb
                );
            }
            Ok(CleanupResult {
                worker_id: config.id,
                success: true,
                dirs_removed: metrics.as_ref().map(|m| m.removed_dirs),
                bytes_freed: metrics.as_ref().map(|m| m.freed_bytes),
                duration,
                error: None,
            })
        } else {
            let error_msg = format!(
                "Cleanup command failed with exit code {} ({}, stderr: {})",
                result.exit_code,
                metrics_summary,
                result.stderr.trim()
            );
            warn!(
                "Cache cleanup failed on worker {}: {}",
                config.id, error_msg
            );
            Ok(CleanupResult {
                worker_id: config.id,
                success: false,
                dirs_removed: None,
                bytes_freed: None,
                duration,
                error: Some(error_msg),
            })
        }
    }

    /// Trigger immediate cleanup on a specific worker.
    pub async fn cleanup_worker_now(&self, worker_id: &WorkerId) -> anyhow::Result<CleanupResult> {
        let worker_state = self
            .pool
            .get(worker_id)
            .await
            .ok_or_else(|| anyhow::anyhow!("Worker {} not found", worker_id))?;

        self.cleanup_worker(&worker_state).await
    }

    /// Get cleanup status for all workers.
    pub async fn get_status(&self) -> HashMap<WorkerId, Option<Instant>> {
        let last_cleanup = self.last_cleanup.read().await;
        let workers = self.pool.all_workers().await;

        let mut status = HashMap::new();
        for worker in workers {
            let id = worker.config.read().await.id.clone();
            let last = last_cleanup.get(&id).copied();
            status.insert(id, last);
        }
        status
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rch_common::test_guard;
    use rch_common::{CircuitState, WorkerConfig};

    #[test]
    fn test_cleanup_config_defaults() {
        let _guard = test_guard!();
        let config = CacheCleanupConfig::default();
        assert!(config.enabled);
        assert_eq!(config.interval_secs, 3600);
        assert_eq!(config.max_cache_age_hours, 72);
        assert_eq!(config.min_free_gb, 10);
        assert_eq!(config.idle_threshold_secs, 60);
        assert_eq!(config.remote_base, "/data/tmp/rch");
    }

    #[test]
    fn test_cleanup_stats_default() {
        let _guard = test_guard!();
        let stats = CleanupStats::default();
        assert_eq!(stats.workers_checked, 0);
        assert_eq!(stats.workers_cleaned, 0);
        assert_eq!(stats.workers_skipped, 0);
        assert_eq!(stats.errors, 0);
        assert_eq!(stats.total_bytes_freed, 0);
    }

    #[test]
    fn test_cleanup_result_creation() {
        let _guard = test_guard!();
        let result = CleanupResult {
            worker_id: WorkerId::new("test-worker"),
            success: true,
            dirs_removed: Some(5),
            bytes_freed: Some(1024 * 1024 * 100),
            duration: Duration::from_secs(2),
            error: None,
        };

        assert!(result.success);
        assert_eq!(result.dirs_removed, Some(5));
        assert_eq!(result.bytes_freed, Some(104857600));
    }

    #[test]
    fn test_cleanup_result_failure() {
        let _guard = test_guard!();
        let result = CleanupResult {
            worker_id: WorkerId::new("failing-worker"),
            success: false,
            dirs_removed: None,
            bytes_freed: None,
            duration: Duration::from_millis(500),
            error: Some("SSH connection failed".to_string()),
        };

        assert!(!result.success);
        assert!(result.dirs_removed.is_none());
        assert!(result.bytes_freed.is_none());
        assert_eq!(result.error, Some("SSH connection failed".to_string()));
        assert_eq!(result.duration.as_millis(), 500);
    }

    #[test]
    fn test_cleanup_result_partial_info() {
        let _guard = test_guard!();
        // Result where we got some info but not all
        let result = CleanupResult {
            worker_id: WorkerId::new("partial-worker"),
            success: true,
            dirs_removed: Some(10),
            bytes_freed: None, // couldn't determine bytes
            duration: Duration::from_secs(5),
            error: None,
        };

        assert!(result.success);
        assert_eq!(result.dirs_removed, Some(10));
        assert!(result.bytes_freed.is_none());
    }

    #[test]
    fn test_cleanup_result_debug_format() {
        let _guard = test_guard!();
        let result = CleanupResult {
            worker_id: WorkerId::new("debug-worker"),
            success: true,
            dirs_removed: Some(3),
            bytes_freed: Some(1024),
            duration: Duration::from_millis(100),
            error: None,
        };

        let debug_str = format!("{:?}", result);
        assert!(debug_str.contains("debug-worker"));
        assert!(debug_str.contains("success: true"));
    }

    #[test]
    fn test_cleanup_result_clone() {
        let _guard = test_guard!();
        let result = CleanupResult {
            worker_id: WorkerId::new("clone-worker"),
            success: true,
            dirs_removed: Some(7),
            bytes_freed: Some(2048),
            duration: Duration::from_secs(1),
            error: None,
        };

        let cloned = result.clone();
        assert_eq!(cloned.worker_id, result.worker_id);
        assert_eq!(cloned.success, result.success);
        assert_eq!(cloned.dirs_removed, result.dirs_removed);
        assert_eq!(cloned.bytes_freed, result.bytes_freed);
    }

    #[test]
    fn test_cleanup_stats_accumulation() {
        let _guard = test_guard!();
        let mut stats = CleanupStats::default();

        // Simulate checking workers
        stats.workers_checked += 3;
        stats.workers_cleaned += 2;
        stats.workers_skipped += 1;
        stats.total_bytes_freed = 1024 * 1024 * 500; // 500MB
        stats.total_dirs_removed = 15;

        assert_eq!(stats.workers_checked, 3);
        assert_eq!(stats.workers_cleaned, 2);
        assert_eq!(stats.workers_skipped, 1);
        assert_eq!(stats.total_bytes_freed, 524288000);
        assert_eq!(stats.total_dirs_removed, 15);
    }

    #[test]
    fn test_cleanup_stats_with_errors() {
        let _guard = test_guard!();
        let stats = CleanupStats {
            workers_checked: 5,
            workers_cleaned: 2,
            workers_skipped: 1,
            errors: 2,
            ..Default::default()
        };

        assert_eq!(
            stats.workers_checked,
            stats.workers_cleaned + stats.workers_skipped + stats.errors
        );
    }

    #[test]
    fn test_cleanup_stats_debug_format() {
        let _guard = test_guard!();
        let stats = CleanupStats {
            workers_checked: 10,
            workers_cleaned: 8,
            workers_skipped: 1,
            errors: 1,
            total_bytes_freed: 1_000_000_000,
            total_dirs_removed: 50,
        };

        let debug_str = format!("{:?}", stats);
        assert!(debug_str.contains("workers_checked: 10"));
        assert!(debug_str.contains("workers_cleaned: 8"));
    }

    fn create_test_worker_config(id: &str) -> WorkerConfig {
        WorkerConfig {
            id: WorkerId::new(id),
            host: "test.example.com".to_string(),
            user: "testuser".to_string(),
            identity_file: "/home/test/.ssh/id_rsa".to_string(),
            total_slots: 8,
            priority: 50,
            tags: vec![],
            tools: Vec::new(),
        }
    }

    #[tokio::test]
    async fn test_worker_state_status_healthy() {
        let config = create_test_worker_config("healthy-worker");
        let worker_state = WorkerState::new(config);

        let status = worker_state.status().await;
        assert_eq!(status, WorkerStatus::Healthy);
    }

    #[tokio::test]
    async fn test_worker_state_status_changes() {
        let config = create_test_worker_config("status-worker");
        let worker_state = WorkerState::new(config);

        // Initially healthy
        assert_eq!(worker_state.status().await, WorkerStatus::Healthy);

        // Set to draining
        worker_state.set_status(WorkerStatus::Draining).await;
        assert_eq!(worker_state.status().await, WorkerStatus::Draining);

        // Set to unreachable
        worker_state.set_status(WorkerStatus::Unreachable).await;
        assert_eq!(worker_state.status().await, WorkerStatus::Unreachable);
    }

    #[tokio::test]
    async fn test_worker_state_slots_availability() {
        let config = create_test_worker_config("slots-worker");
        let worker_state = WorkerState::new(config);

        // All slots available initially
        assert_eq!(worker_state.available_slots().await, 8);

        // Reserve some slots
        let reserved = worker_state.reserve_slots(3).await;
        assert!(reserved);
        assert_eq!(worker_state.available_slots().await, 5);

        // Release slots
        worker_state.release_slots(2).await;
        assert_eq!(worker_state.available_slots().await, 7);
    }

    #[tokio::test]
    async fn test_worker_state_circuit_state() {
        let config = create_test_worker_config("circuit-worker");
        let worker_state = WorkerState::new(config);

        // Circuit starts closed
        let circuit = worker_state.circuit_state().await;
        assert_eq!(circuit, Some(CircuitState::Closed));

        // Open circuit
        worker_state.open_circuit().await;
        assert_eq!(worker_state.circuit_state().await, Some(CircuitState::Open));

        // Half-open
        worker_state.half_open_circuit().await;
        assert_eq!(
            worker_state.circuit_state().await,
            Some(CircuitState::HalfOpen)
        );

        // Close again
        worker_state.close_circuit().await;
        assert_eq!(
            worker_state.circuit_state().await,
            Some(CircuitState::Closed)
        );
    }

    #[tokio::test]
    async fn test_cleanup_scheduler_creation() {
        let pool = WorkerPool::new();
        let config = CacheCleanupConfig::default();

        let scheduler = CacheCleanupScheduler::new(pool, config);

        // Verify config is stored
        assert!(scheduler.config.enabled);
        assert_eq!(scheduler.config.interval_secs, 3600);
    }

    #[tokio::test]
    async fn test_cleanup_scheduler_disabled_config() {
        let pool = WorkerPool::new();
        let config = CacheCleanupConfig {
            enabled: false,
            ..Default::default()
        };

        let scheduler = CacheCleanupScheduler::new(pool, config);
        assert!(!scheduler.config.enabled);
    }

    #[tokio::test]
    async fn test_cleanup_scheduler_custom_config() {
        let pool = WorkerPool::new();
        let config = CacheCleanupConfig {
            enabled: true,
            interval_secs: 1800, // 30 minutes
            max_cache_age_hours: 24,
            min_free_gb: 20,
            idle_threshold_secs: 120,
            remote_base: "/var/rch/cache".to_string(),
        };

        let scheduler = CacheCleanupScheduler::new(pool, config);

        assert!(scheduler.config.enabled);
        assert_eq!(scheduler.config.interval_secs, 1800);
        assert_eq!(scheduler.config.max_cache_age_hours, 24);
        assert_eq!(scheduler.config.min_free_gb, 20);
        assert_eq!(scheduler.config.remote_base, "/var/rch/cache");
    }

    #[tokio::test]
    async fn test_cleanup_scheduler_get_status_empty() {
        let pool = WorkerPool::new();
        let config = CacheCleanupConfig::default();

        let scheduler = CacheCleanupScheduler::new(pool, config);
        let status = scheduler.get_status().await;

        assert!(status.is_empty());
    }

    #[tokio::test]
    async fn test_cleanup_scheduler_get_status_with_workers() {
        let pool = WorkerPool::new();
        let config1 = create_test_worker_config("worker1");
        let config2 = create_test_worker_config("worker2");

        pool.add_worker(config1).await;
        pool.add_worker(config2).await;

        let cleanup_config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool, cleanup_config);
        let status = scheduler.get_status().await;

        assert_eq!(status.len(), 2);
        // No cleanups yet, so all values should be None
        for (_, last_cleanup) in status {
            assert!(last_cleanup.is_none());
        }
    }

    #[tokio::test]
    async fn test_is_worker_eligible_healthy_idle() {
        let pool = WorkerPool::new();
        let config = create_test_worker_config("eligible-worker");
        pool.add_worker(config).await;

        let cleanup_config = CacheCleanupConfig {
            idle_threshold_secs: 0,
            ..Default::default()
        };
        let scheduler = CacheCleanupScheduler::new(pool.clone(), cleanup_config);

        let worker_state = pool.get(&WorkerId::new("eligible-worker")).await.unwrap();

        // Worker is healthy, all slots available, circuit closed
        let eligible = scheduler.is_worker_eligible(&worker_state).await;
        assert!(eligible);
    }

    #[tokio::test]
    async fn test_is_worker_eligible_respects_idle_threshold() {
        let pool = WorkerPool::new();
        let config = create_test_worker_config("idle-threshold-worker");
        pool.add_worker(config).await;

        let cleanup_config = CacheCleanupConfig {
            idle_threshold_secs: 60,
            ..Default::default()
        };
        let scheduler = CacheCleanupScheduler::new(pool.clone(), cleanup_config);
        let worker_state = pool
            .get(&WorkerId::new("idle-threshold-worker"))
            .await
            .unwrap();

        let eligible_first = scheduler.is_worker_eligible(&worker_state).await;
        assert!(!eligible_first);

        {
            let mut idle_since = scheduler.idle_since.write().await;
            idle_since.insert(
                WorkerId::new("idle-threshold-worker"),
                Instant::now() - Duration::from_secs(120),
            );
        }

        let eligible_after_threshold = scheduler.is_worker_eligible(&worker_state).await;
        assert!(eligible_after_threshold);
    }

    #[tokio::test]
    async fn test_disk_slots_idle_worker_remains_eligible_for_cleanup() {
        let pool = WorkerPool::new();
        pool.add_worker(create_test_worker_config("disk-cleanup"))
            .await;
        let worker = pool.get(&WorkerId::new("disk-cleanup")).await.unwrap();
        let scheduler = CacheCleanupScheduler::new(
            pool,
            CacheCleanupConfig {
                idle_threshold_secs: 0,
                ..Default::default()
            },
        );
        assert!(worker.reserve_slots(2).await);
        for free_gb in [40.0, 10.0] {
            worker
                .set_pressure_assessment(crate::disk_pressure::PressureAssessment {
                    disk_free_gb: Some(free_gb),
                    ..Default::default()
                })
                .await;
            assert!(!scheduler.is_worker_eligible(&worker).await);
        }
        worker.release_slots(2).await;
        assert_eq!(worker.available_slots().await, 0);
        assert!(scheduler.is_worker_eligible(&worker).await);
        worker
            .set_pressure_assessment(crate::disk_pressure::PressureAssessment {
                disk_free_gb: Some(40.0),
                ..Default::default()
            })
            .await;
        assert_eq!(worker.available_slots().await, 3);
        assert!(scheduler.is_worker_eligible(&worker).await);
    }

    #[tokio::test]
    async fn test_is_worker_eligible_unhealthy() {
        let pool = WorkerPool::new();
        let config = create_test_worker_config("unhealthy-worker");
        pool.add_worker(config).await;

        let cleanup_config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool.clone(), cleanup_config);

        let worker_state = pool.get(&WorkerId::new("unhealthy-worker")).await.unwrap();

        // Set worker to unreachable
        worker_state.set_status(WorkerStatus::Unreachable).await;

        let eligible = scheduler.is_worker_eligible(&worker_state).await;
        assert!(!eligible);
    }

    #[tokio::test]
    async fn test_is_worker_eligible_busy() {
        let pool = WorkerPool::new();
        let config = create_test_worker_config("busy-worker");
        pool.add_worker(config).await;

        let cleanup_config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool.clone(), cleanup_config);

        let worker_state = pool.get(&WorkerId::new("busy-worker")).await.unwrap();

        // Reserve some slots (worker is now busy)
        worker_state.reserve_slots(4).await;

        let eligible = scheduler.is_worker_eligible(&worker_state).await;
        assert!(!eligible);
    }

    #[tokio::test]
    async fn test_is_worker_eligible_circuit_open() {
        let pool = WorkerPool::new();
        let config = create_test_worker_config("circuit-open-worker");
        pool.add_worker(config).await;

        let cleanup_config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool.clone(), cleanup_config);

        let worker_state = pool
            .get(&WorkerId::new("circuit-open-worker"))
            .await
            .unwrap();

        // Open the circuit
        worker_state.open_circuit().await;

        let eligible = scheduler.is_worker_eligible(&worker_state).await;
        assert!(!eligible);
    }

    #[tokio::test]
    async fn test_is_worker_eligible_draining() {
        let pool = WorkerPool::new();
        let config = create_test_worker_config("draining-worker");
        pool.add_worker(config).await;

        let cleanup_config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool.clone(), cleanup_config);

        let worker_state = pool.get(&WorkerId::new("draining-worker")).await.unwrap();

        // Set to draining
        worker_state.set_status(WorkerStatus::Draining).await;

        let eligible = scheduler.is_worker_eligible(&worker_state).await;
        assert!(!eligible);
    }

    #[tokio::test]
    async fn test_cleanup_worker_now_not_found() {
        let pool = WorkerPool::new();
        let config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool, config);

        let result = scheduler
            .cleanup_worker_now(&WorkerId::new("nonexistent"))
            .await;
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("Worker nonexistent not found")
        );
    }

    #[tokio::test]
    async fn test_run_cleanup_cycle_empty_pool() {
        let pool = WorkerPool::new();
        let config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool, config);

        let stats = scheduler.run_cleanup_cycle().await;

        assert_eq!(stats.workers_checked, 0);
        assert_eq!(stats.workers_cleaned, 0);
        assert_eq!(stats.workers_skipped, 0);
        assert_eq!(stats.errors, 0);
    }

    #[tokio::test]
    async fn test_run_cleanup_cycle_all_skipped() {
        let pool = WorkerPool::new();
        // Add workers that will be skipped (busy)
        let config1 = create_test_worker_config("busy-worker1");
        let config2 = create_test_worker_config("busy-worker2");
        pool.add_worker(config1).await;
        pool.add_worker(config2).await;

        // Make workers busy
        let worker1 = pool.get(&WorkerId::new("busy-worker1")).await.unwrap();
        let worker2 = pool.get(&WorkerId::new("busy-worker2")).await.unwrap();
        worker1.reserve_slots(4).await;
        worker2.reserve_slots(4).await;

        let cleanup_config = CacheCleanupConfig::default();
        let scheduler = CacheCleanupScheduler::new(pool, cleanup_config);

        let stats = scheduler.run_cleanup_cycle().await;

        assert_eq!(stats.workers_checked, 2);
        assert_eq!(stats.workers_cleaned, 0);
        assert_eq!(stats.workers_skipped, 2);
        assert_eq!(stats.errors, 0);
    }

    #[test]
    fn test_cleanup_config_custom_values() {
        let _guard = test_guard!();
        let config = CacheCleanupConfig {
            enabled: false,
            interval_secs: 7200,
            max_cache_age_hours: 168, // 1 week
            min_free_gb: 50,
            idle_threshold_secs: 300,
            remote_base: "/custom/path".to_string(),
        };

        assert!(!config.enabled);
        assert_eq!(config.interval_secs, 7200);
        assert_eq!(config.max_cache_age_hours, 168);
        assert_eq!(config.min_free_gb, 50);
        assert_eq!(config.idle_threshold_secs, 300);
        assert_eq!(config.remote_base, "/custom/path");
    }

    #[test]
    fn test_cleanup_result_with_large_values() {
        let _guard = test_guard!();
        let result = CleanupResult {
            worker_id: WorkerId::new("large-cleanup"),
            success: true,
            dirs_removed: Some(1000),
            bytes_freed: Some(1024 * 1024 * 1024 * 10), // 10GB
            duration: Duration::from_secs(120),         // 2 minutes
            error: None,
        };

        assert!(result.success);
        assert_eq!(result.dirs_removed, Some(1000));
        assert_eq!(result.bytes_freed, Some(10737418240));
        assert_eq!(result.duration.as_secs(), 120);
    }

    #[test]
    fn test_cleanup_stats_large_accumulation() {
        let _guard = test_guard!();
        // Simulate large-scale cleanup
        let stats = CleanupStats {
            workers_checked: 100,
            workers_cleaned: 95,
            workers_skipped: 3,
            errors: 2,
            total_bytes_freed: 1024 * 1024 * 1024 * 500, // 500GB
            total_dirs_removed: 5000,
        };

        assert_eq!(stats.workers_checked, 100);
        assert_eq!(stats.total_bytes_freed, 536870912000);
        assert_eq!(stats.total_dirs_removed, 5000);
    }

    #[test]
    fn test_parse_cleanup_metrics_parses_emitted_line() {
        let _guard = test_guard!();
        let stdout = "noise\nRCH_CLEANUP_METRICS removed=4 freed_kb=2048 before_kb=4096 after_kb=6144 low_disk=1 remove_errors=2\n";
        let metrics = parse_cleanup_metrics(stdout).expect("expected parse result");
        assert_eq!(metrics.removed_dirs, 4);
        assert_eq!(metrics.freed_bytes, 2 * 1024 * 1024);
        assert!((metrics.free_before_gb - (4096.0 / (1024.0 * 1024.0))).abs() < f64::EPSILON);
        assert!((metrics.free_after_gb - (6144.0 / (1024.0 * 1024.0))).abs() < f64::EPSILON);
        assert!(metrics.low_disk_mode);
        assert_eq!(metrics.remove_errors, 2);
    }

    #[test]
    fn test_build_cleanup_command_contains_threshold_and_marker() {
        let _guard = test_guard!();
        let command = build_cleanup_command("'/tmp/rch'", REMOTE_RUN_ROOT, 72, 10);
        assert!(command.contains("threshold_kb=10485760"));
        assert!(command.contains("active_grace_minutes=5"));
        assert!(command.contains("-mmin -\"$active_grace_minutes\""));
        assert!(command.contains("case \"$dir\" in \"$base\"/*)"));
        assert!(command.contains("if rm -rf \"$dir\""));
        assert!(command.contains("remove_errors=$((remove_errors + 1))"));
        assert!(command.contains("if [ \"$remove_errors\" -gt 0 ]; then exit 1; fi"));
        assert!(command.contains("RCH_CLEANUP_METRICS"));
        assert!(command.contains("low_disk"));
        assert!(command.contains("remove_errors"));
    }

    /// Backdate a file far past the 240-minute orphan threshold.
    fn age_record(path: &std::path::Path) {
        let backdated = std::time::SystemTime::now() - std::time::Duration::from_secs(5 * 3600);
        std::fs::File::options()
            .write(true)
            .open(path)
            .expect("open record")
            .set_modified(backdated)
            .expect("backdate record");
    }

    /// Stale records (malformed, or naming a group with no members) are
    /// deleted; fresh ones are kept; an empty run dir is removed only once
    /// it is a day old.
    #[test]
    fn cleanup_command_deletes_stale_process_records_only() {
        let _guard = test_guard!();
        let base = tempfile::tempdir().expect("base");
        let runs = tempfile::tempdir().expect("run root");
        let run = runs.path().join("project-0123456789abcdef");
        std::fs::create_dir(&run).expect("run dir");
        let malformed = run.join("41.pgid");
        std::fs::write(&malformed, "garbage\n").expect("malformed");
        age_record(&malformed);
        // Legacy bare pgid of a group that cannot exist (above pid_max).
        let vanished = run.join("42.pgid");
        std::fs::write(&vanished, "2147483000\n").expect("vanished");
        age_record(&vanished);
        let fresh = run.join("43.pgid");
        std::fs::write(&fresh, "garbage\n").expect("fresh");
        let empty_old = runs.path().join("empty-old");
        std::fs::create_dir(&empty_old).expect("empty old");
        std::fs::File::open(&empty_old)
            .unwrap()
            .set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(2 * 86400))
            .unwrap();
        let empty_new = runs.path().join("empty-new");
        std::fs::create_dir(&empty_new).expect("empty new");

        let (stdout, stderr, code) = run_sweep_with_runs(base.path(), runs.path(), &[]);
        assert_eq!(code, Some(0), "{stdout}{stderr}");
        assert!(stdout.contains("orphans_killed=0"), "{stdout}");
        assert!(
            !malformed.exists() && !vanished.exists(),
            "stale records remain"
        );
        assert!(
            fresh.exists(),
            "a record younger than the threshold was touched"
        );
        assert!(!empty_old.exists(), "day-old empty run dir must go");
        assert!(
            empty_new.exists() && run.exists(),
            "live run dirs must stay"
        );
    }

    /// A live group whose recorded leader matches is killed; the identity
    /// check comes from the launcher's own record format.
    #[cfg(target_os = "linux")]
    #[test]
    fn cleanup_command_kills_identified_orphan_group() {
        use std::os::unix::process::ExitStatusExt;
        let _guard = test_guard!();
        let base = tempfile::tempdir().expect("base");
        let runs = tempfile::tempdir().expect("run root");
        let run = runs.path().join("project-0123456789abcdef");
        std::fs::create_dir(&run).expect("run dir");
        let record = run.join("77.pgid");
        let launcher = format!(
            "{}\nrch_remote_record \"$1\" 77 || exit 9\nexec sleep 600",
            rch_common::REMOTE_PROCESS_IDENTITY_SCRIPT
        );
        let mut leader = std::process::Command::new("setsid")
            .args(["sh", "-c", &launcher, "rch-test"])
            .arg(&record)
            .spawn()
            .expect("spawn orphan");
        let deadline = Instant::now() + Duration::from_secs(10);
        while !record.exists() {
            assert!(Instant::now() < deadline, "launcher never published");
            std::thread::sleep(Duration::from_millis(20));
        }
        age_record(&record);

        let (stdout, stderr, code) = run_sweep_with_runs(base.path(), runs.path(), &[]);
        let status = leader.wait().expect("reap leader");
        assert_eq!(code, Some(0), "{stdout}{stderr}");
        assert!(stdout.contains("orphans_killed=1"), "{stdout}{stderr}");
        assert_eq!(status.signal(), Some(9));
    }

    /// The sweep (both modes share the loop) must not delete a directory a
    /// live process works in, even when nothing in it was written recently.
    #[cfg(target_os = "linux")]
    #[test]
    fn cleanup_command_keeps_directory_used_as_process_cwd() {
        let _guard = test_guard!();
        let base = tempfile::tempdir().expect("base");
        let pool = base.path().join("repo").join(".rch-target-pool-busy");
        std::fs::create_dir_all(pool.join("deps")).expect("pool tree");
        age_dir(&pool);
        let mut worker = std::process::Command::new("sleep")
            .arg("600")
            .current_dir(pool.join("deps"))
            .spawn()
            .expect("spawn cwd holder");
        let (stdout, _stderr, code) = run_sweep(base.path(), &[]);
        worker.kill().expect("kill cwd holder");
        worker.wait().expect("reap cwd holder");
        assert_eq!(code, Some(0));
        assert!(stdout.contains("removed=0"), "{stdout}");
        assert!(pool.is_dir(), "in-use pool was deleted");
    }

    #[test]
    fn test_build_cleanup_command_reconciles_failed_rm_and_reports_diagnostics() {
        let _guard = test_guard!();
        let command = build_cleanup_command("'/tmp/rch'", REMOTE_RUN_ROOT, 72, 10);
        // bd-kwvy8: a nonzero rm does NOT mean nothing was reaped. A
        // candidate that is already gone was reaped by another writer —
        // count it.
        assert!(command.contains("elif [ ! -d \"$dir\" ]; then"));
        // An empty husk left by a raced/partial removal gets exactly one
        // rmdir retry; success counts as removed and says so on stderr.
        assert!(command.contains("RCH_CLEANUP_HUSK_REAPED dir=%s"));
        // A surviving candidate or an unreapable husk produces a per-dir
        // diagnostic instead of being silently swallowed by 2>/dev/null.
        assert!(command.contains("RCH_CLEANUP_RM_FAILURE dir=%s kb=%s"));
    }

    /// Backdate a fixture directory so the sweep's `-mmin +max_age` filter
    /// (max_age 0 in these tests) deterministically selects it.
    ///
    /// A directory created moments before the sweep has age 0, which is
    /// NOT "older than 0 minutes" under either `find`: BSD rounds
    /// `(age + 59) / 60` and so needs the entry to be at least one second
    /// old, GNU truncates and needs a full minute. Fresh fixtures therefore
    /// only became candidates when the test process happened to be slow,
    /// which made the tests pass under a loaded full gate and fail when
    /// run alone.
    fn age_dir(path: &std::path::Path) {
        let backdated = std::time::SystemTime::now() - std::time::Duration::from_secs(10 * 60);
        std::fs::File::open(path)
            .expect("open fixture dir")
            .set_modified(backdated)
            .expect("backdate fixture dir");
    }

    /// Write an executable `name` into `dir` so `dir` can be prepended to
    /// the sweep's PATH (see `run_sweep`).
    fn write_shim(dir: &std::path::Path, name: &str, body: &str) {
        use std::os::unix::fs::PermissionsExt;
        let shim = dir.join(name);
        std::fs::write(&shim, body).expect("write shim");
        std::fs::set_permissions(&shim, std::fs::Permissions::from_mode(0o755))
            .expect("chmod shim");
    }

    /// Runs the generated sweep against a scratch tree, optionally with
    /// directories prepended to PATH (for command shims), capturing
    /// stdout, stderr, and exit status.
    fn run_sweep(
        base: &std::path::Path,
        prepend_path: &[std::path::PathBuf],
    ) -> (String, String, Option<i32>) {
        // Never let a test sweep the host's real process records.
        let runs = tempfile::tempdir().expect("run root");
        run_sweep_with_runs(base, runs.path(), prepend_path)
    }

    fn run_sweep_with_runs(
        base: &std::path::Path,
        runs: &std::path::Path,
        prepend_path: &[std::path::PathBuf],
    ) -> (String, String, Option<i32>) {
        use std::env::{join_paths, split_paths};
        let escaped = format!("'{}'", base.display());
        let command = build_cleanup_command(&escaped, &format!("'{}'", runs.display()), 0, 0);
        let mut full_path: Vec<std::path::PathBuf> = prepend_path.to_vec();
        full_path.extend(split_paths(&std::env::var_os("PATH").unwrap_or_default()));
        let output = std::process::Command::new("sh")
            .arg("-c")
            .arg(&command)
            .env("PATH", join_paths(full_path.iter()).expect("join PATH"))
            .output()
            .expect("run sweep");
        (
            String::from_utf8_lossy(&output.stdout).into_owned(),
            String::from_utf8_lossy(&output.stderr).into_owned(),
            output.status.code(),
        )
    }

    #[test]
    fn cleanup_command_reaps_aged_pool_normally() {
        let _guard = test_guard!();
        let base = tempfile::tempdir().expect("base");
        let pool = base
            .path()
            .join("gcverify_pool_repo")
            .join(".rch-target-pool-deadbeefcafe");
        std::fs::create_dir_all(pool.join("deps")).expect("pool tree");
        age_dir(&pool);
        let (stdout, _stderr, code) = run_sweep(base.path(), &[]);
        assert_eq!(code, Some(0));
        assert!(stdout.contains("removed=1"), "{stdout}");
        assert!(stdout.contains("remove_errors=0"), "{stdout}");
    }

    #[test]
    fn cleanup_command_reaps_empty_husk_when_rmdir_succeeds() {
        let _guard = test_guard!();
        let base = tempfile::tempdir().expect("base");
        // Empty husk: contents were removed by an interfering writer,
        // the directory itself remains (bd-kwvy8's hz2 observation).
        let husk = base
            .path()
            .join("gcverify_pool_repo")
            .join(".rch-target-pool-deadbeefcafe");
        std::fs::create_dir_all(&husk).expect("husk");
        age_dir(&husk);
        // A shim `rm` that refuses -rf models rm failing mid-removal;
        // the sweep must reconcile via rmdir and count the reap.
        let shim_dir = tempfile::tempdir().expect("shim dir");
        write_shim(
            shim_dir.path(),
            "rm",
            "#!/bin/sh\n[ \"$1\" = \"-rf\" ] && exit 42\nfor real in /usr/bin/rm /bin/rm; do [ -x \"$real\" ] && exec \"$real\" \"$@\"; done\nexit 127\n",
        );
        let (stdout, stderr, code) = run_sweep(base.path(), &[shim_dir.path().to_path_buf()]);
        assert_eq!(code, Some(0));
        assert!(stdout.contains("removed=1"), "{stdout}");
        assert!(stdout.contains("remove_errors=0"), "{stdout}");
        assert!(
            stderr.contains("RCH_CLEANUP_HUSK_REAPED")
                && stderr.contains(".rch-target-pool-deadbeefcafe"),
            "{stderr}"
        );
    }

    #[test]
    fn cleanup_command_reports_survivor_and_unreapable_husk_as_errors() {
        let _guard = test_guard!();
        let base = tempfile::tempdir().expect("base");
        let survivor = base.path().join("repo_a").join(".rch-target-pool-survivor");
        std::fs::create_dir_all(survivor.join("nested")).expect("survivor tree");
        age_dir(&survivor);
        let husk = base.path().join("repo_b").join(".rch-target-pool-husk");
        std::fs::create_dir_all(&husk).expect("husk");
        age_dir(&husk);
        // Inject the failures through command shims rather than read-only
        // parents: root ignores directory permissions, so an EACCES-based
        // fixture reports the wrong verdict on the root-run workers the
        // gate actually executes on. `rm -rf` refusing leaves the survivor
        // non-empty (RM_FAILURE via the final branch); `rmdir` refusing
        // leaves the husk an unreapable empty shell (RM_FAILURE via the
        // rmdir-retry branch). Both directories must still exist after.
        let shim_dir = tempfile::tempdir().expect("shim dir");
        write_shim(
            shim_dir.path(),
            "rm",
            "#!/bin/sh\n[ \"$1\" = \"-rf\" ] && exit 1\nfor real in /usr/bin/rm /bin/rm; do [ -x \"$real\" ] && exec \"$real\" \"$@\"; done\nexit 127\n",
        );
        write_shim(shim_dir.path(), "rmdir", "#!/bin/sh\nexit 1\n");
        let (stdout, stderr, code) = run_sweep(base.path(), &[shim_dir.path().to_path_buf()]);
        assert!(survivor.join("nested").is_dir(), "survivor must be intact");
        assert!(husk.is_dir(), "husk must be intact");
        assert_eq!(code, Some(1));
        assert!(stdout.contains("removed=0"), "{stdout}");
        assert!(stdout.contains("remove_errors=2"), "{stdout}");
        assert!(
            stderr.matches("RCH_CLEANUP_RM_FAILURE").count() == 2,
            "{stderr}"
        );
        assert!(stderr.contains(".rch-target-pool-survivor"), "{stderr}");
        assert!(stderr.contains(".rch-target-pool-husk"), "{stderr}");
    }
}