rchd 2.1.0

//! Daemon-side worker sweep that reaps *stale* per-job `.rch-target-*` dirs.
//!
//! # Why this exists
//!
//! rch gives every forwarded-`CARGO_TARGET_DIR` build a per-job target dir
//! (`.rch-target-<worker>-job-<id>-…` / `…-pid-…`) placed **directly inside** the
//! synced repo dir, i.e. `<sync-root>/<repo>/.rch-target-*` (e.g.
//! `/data/projects/<repo>/.rch-target-<worker>-job-…`). The orchestrator hook
//! already reaps *abandoned* such dirs — but **only for the single repo currently
//! being built**, and only as a side-effect of an offloaded build. Per-job dirs in
//! any repo that nobody is actively rch-building therefore accumulate forever (we
//! hand-reclaimed ~1.6 TB across the fleet because of exactly this gap).
//!
//! This module closes that gap with a periodic background task inside `rchd`:
//! every `interval_mins` it SSHes to each healthy worker and applies the
//! **identical** idle predicate as the orchestrator reaper — shared via
//! [`rch_common::stale_target_reap`] so the two cannot drift — across **all** repo
//! dirs under the worker's `remote_base` (the repo sync-root, default
//! `/data/projects`).
//!
//! # Safety
//!
//! - Matches only the shared globs `.rch-target-*-job-*` / `.rch-target-*-pid-*`
//!   ([`rch_common::stale_target_reap::REAP_GLOBS`]) at the per-job depth
//!   (`<base>/<repo>/<glob>`). A bare `target`, a source dir, `.git`, `.beads`,
//!   etc. are structurally unreachable by the glob + depth.
//! - A dir is removed only if **neither it nor any descendant** was modified
//!   within the idle window (default 12h), and the durable source registry has
//!   no overlapping active or pending claim. Quiet outputs awaiting recovery
//!   remain owned even when their original process has exited.
//! - The shared builder reserves each deletion durably under the admission
//!   metadata lock. A killed sweep leaves its reservation intact, preventing a
//!   surviving removal child from racing a newly admitted source writer.
//! - The `remote_base` is validated against
//!   [`rch_common::stale_target_reap::is_safe_reap_base`] before being embedded in
//!   the generated shell (the security boundary).
//! - Only `WorkerStatus::Healthy` workers with a closed circuit are swept; SSH and
//!   reap failures are swallowed (best-effort, never load-bearing).

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

/// Metrics emitted by the worker-side reap script.
#[derive(Debug, Clone, Default)]
struct ReapMetrics {
    removed: u64,
    freed_bytes: u64,
}

/// Aggregate stats for one sweep cycle across all workers.
#[derive(Debug, Default, Clone)]
pub struct ReapCycleStats {
    pub workers_checked: u32,
    pub workers_swept: u32,
    pub workers_skipped: u32,
    pub errors: u32,
    pub total_removed: u64,
    pub total_bytes_freed: u64,
}

/// Build the worker-wide sweep script.
///
/// `escaped_base` is the remote repo sync-root rch rsyncs each repo into. rch
/// places per-job target dirs **directly inside** the synced repo dir, so the
/// real on-worker layout is `<base>/<repo>/.rch-target-*-job-*` — i.e. the
/// per-job dir sits one level below `<base>/<repo>` (depth 2 under `<base>`), NOT
/// `<base>/<project_id>/<hash>/.rch-target-*`. This mirrors the orchestrator
/// reaper, which `cd`s into `remote_path()` (= the synced repo dir) and globs
/// `.rch-target-*-job-*` directly. For every repo dir one level under
/// `escaped_base` we iterate the shared per-job globs and apply the shared
/// [`stale_target_reap::reap_loop_body`] predicate, accumulating a removed-count
/// and freed-KB, then print a parseable metrics line.
///
/// # Depth-robust matching
///
/// rch places a per-job target dir directly inside the synced repo dir, but the
/// synced "repo dir" is not always one level under `escaped_base`: a top-level
/// project sits at `<base>/<repo>/.rch-target-*` (depth 2), a canonical multi-repo
/// layout at `<base>/<id>/<hash>/.rch-target-*`, and a **workspace member** even
/// deeper (`<base>/<repo>/crates/<member>/.rch-target-*`). A fixed depth-2 glob
/// (`for proj in "$base"/*; …`) misses the nested per-job dirs. So we walk with a
/// single `find "$__rt" -maxdepth 8 -type d ( -name ".rch-target-*-job-*" -o -name
/// ".rch-target-*-pid-*" ) -prune` that matches the per-job dir at *any* depth and
/// `-prune`s so it never descends into the (large) artifact tree. This mirrors the
/// orchestrator hook's own depth-robust full-tree sweep (now retired in favor of
/// this daemon sweep). The bound is 8 to give deep nested layouts ample headroom
/// while still bounding the source-tree walk.
///
/// # Metrics survive the loop
///
/// A `find … | while read … done` pipe would run the loop body in a **subshell**
/// in POSIX `sh`, discarding the `removed`/`freed_kb` increments. To keep the
/// counters in the parent shell we redirect `find` output to a temp file and feed
/// the `while read` loop *from the file* (`done < "$tmpf"`) — a redirect-from-file
/// (unlike the right side of a pipe) does not spawn a subshell, so the shared
/// [`stale_target_reap::reap_loop_body`] increments persist and the emitted
/// `RCH_WORKER_REAP_METRICS` line reflects the true removed/freed totals.
///
/// `escaped_base` is canonicalized at runtime with `cd … && pwd -P` (defensive
/// against a symlinked base) before the walk, and MUST already be validated by
/// [`stale_target_reap::is_safe_reap_base`] in the Rust caller; it is embedded
/// inside double quotes. The `find -name` globs are double-quoted so they survive
/// the single-quoted `sh -c '…'` dispatch unchanged.
fn build_sweep_command(
    escaped_base: &str,
    idle_minutes: u64,
    pooled_idle_minutes: Option<u64>,
    max_cache_kb: Option<u64>,
) -> String {
    // Delegates to the shared builder (moved to `rch_common::stale_target_reap`
    // for 6dj11's `rch gc`, which must sweep EXACTLY what this periodic sweep
    // sweeps — one builder so the two can never drift). The legacy
    // `/data/tmp/rch_target_*` pass, the optional long-window pooled pass, the
    // optional byte-cap LRU eviction, the depth guards, and the
    // metrics-survive-the-loop file-redirect shape all live there now; the
    // shape tests below still pin the emitted script.
    stale_target_reap::worker_sweep_command(
        escaped_base,
        idle_minutes,
        pooled_idle_minutes,
        max_cache_kb,
    )
}

/// Convert the configured pooled idle window (hours; 0 = disabled) into the
/// builder's `Option<minutes>`.
fn pooled_idle_minutes_from_hours(pooled_idle_hours: u32) -> Option<u64> {
    (pooled_idle_hours != 0).then(|| u64::from(pooled_idle_hours) * 60)
}

/// Convert the configured byte budget (GiB; 0 = disabled) into the builder's
/// `Option<KiB>`.
fn max_cache_kb_from_gb(max_cache_gb: u32) -> Option<u64> {
    (max_cache_gb != 0).then(|| u64::from(max_cache_gb) * 1024 * 1024)
}

fn parse_reap_metrics(stdout: &str) -> Option<ReapMetrics> {
    let (removed, freed_kb) = stale_target_reap::parse_worker_reap_metrics(stdout)?;
    Some(ReapMetrics {
        removed,
        freed_bytes: freed_kb.saturating_mul(1024),
    })
}

fn ensure_sweep_command_success(result: &CommandResult) -> anyhow::Result<()> {
    if result.success() {
        return Ok(());
    }

    let stderr = result.stderr.trim();
    if stderr.is_empty() {
        anyhow::bail!("sweep command exited with code {}", result.exit_code);
    }

    anyhow::bail!(
        "sweep command exited with code {}: {}",
        result.exit_code,
        stderr
    )
}

/// Periodic worker-side stale-target reaper.
pub struct StaleTargetReaper {
    pool: WorkerPool,
    config: StaleTargetReapConfig,
    ssh_options: SshOptions,
    ssh_pool: Option<Arc<rch_common::SshPool>>,
}

impl StaleTargetReaper {
    pub fn new(pool: WorkerPool, config: StaleTargetReapConfig) -> Self {
        Self {
            pool,
            config,
            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
    }

    /// Spawn the background sweep loop. Returns the task handle.
    pub fn start(self: Arc<Self>) -> tokio::task::JoinHandle<()> {
        let reaper = self;
        tokio::spawn(async move {
            if !reaper.config.enabled {
                info!("Worker stale-target reaper disabled");
                return;
            }
            // Validate the base once up-front; a bad base disables the whole sweep
            // rather than risking an unsafe embedded path.
            if let Some(unsafe_base) = reaper
                .scan_bases()
                .into_iter()
                .find(|base| !stale_target_reap::is_safe_reap_base(base))
            {
                warn!("Worker stale-target reaper disabled: unsafe reap base {unsafe_base:?}");
                return;
            }
            info!(
                "Worker stale-target reaper started (interval={}min, idle>{}h, base={})",
                reaper.config.interval_mins,
                reaper
                    .config
                    .idle_hours
                    .max(stale_target_reap::MIN_IDLE_HOURS),
                reaper.config.remote_base
            );
            let mut ticker = interval(Duration::from_secs(reaper.config.interval_mins.max(1) * 60));
            loop {
                ticker.tick().await;
                let stats = reaper.run_cycle().await;
                if stats.total_removed > 0 || stats.errors > 0 {
                    info!(
                        "Worker stale-target reap cycle: checked={}, swept={}, skipped={}, errors={}, removed={} dirs, freed={}MB",
                        stats.workers_checked,
                        stats.workers_swept,
                        stats.workers_skipped,
                        stats.errors,
                        stats.total_removed,
                        stats.total_bytes_freed / (1024 * 1024),
                    );
                } else {
                    debug!("Worker stale-target reap cycle: nothing to reap");
                }
            }
        })
    }

    /// Run one sweep across all workers.
    pub async fn run_cycle(&self) -> ReapCycleStats {
        let mut stats = ReapCycleStats::default();
        let idle_minutes = stale_target_reap::idle_minutes_from_hours(self.config.idle_hours);
        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();

            // Only sweep workers we can reliably SSH to. The mtime predicate — not
            // worker idleness — is what protects active builds, so we deliberately
            // do NOT skip busy workers: a busy worker may have stale dirs in repos
            // it is not currently building, which is the entire reason this sweep
            // exists. Active dirs on a busy worker are still preserved because they
            // are touched within the idle window.
            if !self.is_worker_sweepable(&worker_state, &worker_id).await {
                stats.workers_skipped += 1;
                continue;
            }

            match self.sweep_worker(&worker_state, idle_minutes).await {
                Ok(metrics) => {
                    stats.workers_swept += 1;
                    stats.total_removed += metrics.removed;
                    stats.total_bytes_freed += metrics.freed_bytes;
                    if metrics.removed > 0 {
                        info!(
                            worker = %worker_id,
                            removed = metrics.removed,
                            freed_mb = metrics.freed_bytes / (1024 * 1024),
                            "Reaped stale per-job target dirs on worker"
                        );
                    }
                }
                Err(e) => {
                    warn!(worker = %worker_id, error = %e, "Worker stale-target sweep failed");
                    stats.errors += 1;
                }
            }
        }

        stats
    }

    /// Whether a worker is in a state where SSH-based sweeping is sensible.
    async fn is_worker_sweepable(&self, worker_state: &WorkerState, worker_id: &WorkerId) -> bool {
        let status = worker_state.status().await;
        if status != WorkerStatus::Healthy {
            debug!(worker = %worker_id, ?status, "Skipping non-healthy worker for stale-target sweep");
            return false;
        }
        let circuit_state = worker_state.circuit_state().await;
        if circuit_state != Some(rch_common::CircuitState::Closed) {
            debug!(worker = %worker_id, ?circuit_state, "Skipping worker with non-closed circuit for stale-target sweep");
            return false;
        }
        true
    }

    /// Every root this sweep must walk: the repo mirror tree, plus the pooled
    /// `store_base` when pooled stores are placed off that tree (issue #64).
    /// De-duplicated, so the common `store_base == remote_base` case sweeps
    /// once.
    fn scan_bases(&self) -> Vec<String> {
        let normalize = |base: &str| base.trim_end_matches('/').to_string();
        let mut bases = vec![normalize(&self.config.remote_base)];
        if let Some(store_base) = self.config.store_base.as_deref().map(normalize)
            && !bases.contains(&store_base)
        {
            bases.push(store_base);
        }
        bases
    }

    /// Run the sweep script on a single worker and return its metrics.
    async fn sweep_worker(
        &self,
        worker_state: &WorkerState,
        idle_minutes: u64,
    ) -> anyhow::Result<ReapMetrics> {
        let start = Instant::now();
        let config = worker_state.config.read().await.clone();
        let worker_id = config.id.clone();

        let mut metrics = ReapMetrics::default();
        for base in self.scan_bases() {
            // Bases were validated once in `start()`, but re-check
            // defensively: config can be hot-reloaded between cycles.
            if !stale_target_reap::is_safe_reap_base(&base) {
                anyhow::bail!("unsafe reap base {base:?}");
            }
            let cmd = build_sweep_command(
                &base,
                idle_minutes,
                pooled_idle_minutes_from_hours(self.config.pooled_idle_hours),
                max_cache_kb_from_gb(self.config.max_cache_gb),
            );

            debug!(worker = %worker_id, idle_minutes, base = %base, "Starting stale-target sweep");

            let result = self.run_remote(&config, &cmd).await;
            let result = result?;
            ensure_sweep_command_success(&result)?;

            let base_metrics = parse_reap_metrics(&result.stdout).unwrap_or_default();
            metrics.removed += base_metrics.removed;
            metrics.freed_bytes += base_metrics.freed_bytes;
            log_reap_audit(&worker_id, &result.stdout);
            debug!(
                worker = %worker_id,
                base = %base,
                removed = base_metrics.removed,
                freed_bytes = base_metrics.freed_bytes,
                exit = result.exit_code,
                "Stale-target sweep finished for base"
            );
        }

        debug!(
            worker = %worker_id,
            removed = metrics.removed,
            freed_bytes = metrics.freed_bytes,
            duration_ms = start.elapsed().as_millis() as u64,
            "Stale-target sweep finished"
        );
        Ok(metrics)
    }
}

/// Emit the per-removal audit trail from one sweep's stdout.
///
/// Bead 6dj11: one info line per reaped dir with size and the policy that
/// removed it, so any deletion can be reconstructed from the daemon log alone.
/// bd-kwvy8: failed removals surface at warn — a half-removed dir with an
/// unincremented counter was undiagnosable while `rm` was silenced.
fn log_reap_audit(worker_id: &WorkerId, stdout: &str) {
    for event in stale_target_reap::parse_reap_events(stdout) {
        info!(
            worker = %worker_id,
            kb = event.kb,
            trigger = %event.trigger,
            path = %event.path,
            "Reaped remote target dir"
        );
    }
    for error in stale_target_reap::parse_reap_errors(stdout) {
        warn!(
            worker = %worker_id,
            trigger = %error.trigger,
            path = %error.path,
            "Remote target dir removal FAILED: {}",
            error.message
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Run a generated sweep script under `sh` with `TMPDIR` pointed at a
    /// directory private to this test's fixture.
    ///
    /// The sweep also scans the worker's tmp base (`$TMPDIR`, resolved into
    /// `$__tmpscan`) for reap-class dirs. Every `tempdir()` fixture lives under
    /// the inherited `TMPDIR`, so with it inherited each test's sweep sees —
    /// and reaps — the fixtures of every OTHER test running in parallel (on
    /// rch workers `TMPDIR` is the in-workspace `.rch-tmp`, and the suite
    /// failed there on exactly that cross-talk). A private base is also what
    /// a real worker has: one tmp base per daemon, not one shared with
    /// strangers. It sits beside the fixture's `projects/` so the tmp-base
    /// pass scans only this test's own tree.
    fn run_sweep(cmd: &str, fixture_root: &std::path::Path) -> std::process::Output {
        let private_tmp = fixture_root.join("tmp");
        std::fs::create_dir_all(&private_tmp).expect("create private TMPDIR");
        std::process::Command::new("sh")
            .arg("-c")
            .arg(cmd)
            .env("TMPDIR", &private_tmp)
            .output()
            .expect("run sweep script under sh")
    }

    #[test]
    fn sweep_command_confines_to_base_and_per_job_globs() {
        let cmd = build_sweep_command("/data/projects", 720, None, None);
        // Depth-robust find at ANY depth under the canonicalized base — NOT a
        // fixed depth-2 `for proj in "$base"/*` glob (which would miss nested
        // workspace-member per-job dirs).
        assert!(cmd.contains(
            "find \"$__rt\" -maxdepth 8 -type d \\( -name \".rch-target-*-job-*\" -o -name \".rch-target-*-pid-*\" \\) -prune"
        ));
        assert!(!cmd.contains("for proj in \"$base\"/*"));
        // Base is canonicalized with `cd … && pwd -P` (defensive vs symlinks).
        assert!(cmd.contains("__rt=$(cd \"$base\" 2>/dev/null && pwd -P)"));
        // Defense-in-depth: the RESOLVED base must still be >=2 segments deep, so a
        // base that symlinks to `/` or a shallow root can never be swept (mirrors
        // the orchestrator-side guard; is_safe_reap_base alone permits 1 segment).
        assert!(cmd.contains("case \"$__rt\" in */*/*) ;; *)"));
        // The validated sync-root base is embedded.
        assert!(cmd.contains("base=\"/data/projects\""));
        // Only the per-job globs are matched.
        assert!(cmd.contains(".rch-target-*-job-*"));
        assert!(cmd.contains(".rch-target-*-pid-*"));
        // Shared idle predicate (no -type f) is present.
        assert!(cmd.contains("find \"$d\" -mmin -720 -print -quit"));
        assert!(!cmd.contains("-type f"));
        // Never a bare `target` glob.
        assert!(!cmd.contains("/target "));
        // Metrics counters survive the loop: the `while read` is fed from a temp
        // file (redirect-from-file, NOT the right side of a pipe → no subshell).
        // The tmp base comes from the SAME prelude that creates the durable
        // per-worker Cargo caches and stages target dirs — not a second copy
        // of the `$TMPDIR` -> /data/tmp -> /tmp ladder. Two copies is exactly
        // how the sweep came to look somewhere other than where rch writes.
        assert!(cmd.contains(&rch_common::gc_roots::remote_cargo_home_base_prelude()));
        assert!(cmd.contains("__tmpbase=\"${RCH_CH_BASE}\""));
        assert!(!cmd.contains("__tmpbase=\"${TMPDIR:-}\""));
        assert!(cmd.contains("mktemp -p \"$__tmpbase\""));
        assert!(!cmd.contains("mktemp -p \"${TMPDIR:-/data/tmp}\""));
        // 6dj11: the legacy /data/tmp/rch_target_* trees (the css
        // incident class) are swept too — -maxdepth 1, name-anchored.
        // a4047390: the tmp base is scanned through `$__tmpscan`, its
        // `pwd -P` canonicalization (so the dedup against `$__rt` is sound),
        // depth-guarded BEFORE and AFTER resolution — emptied unless both
        // spellings are >=2 segments deep — so a bare `/tmp` fallback (or
        // one resolving to macOS `/private/tmp`) is never scanned, and every
        // tmp-base find is gated on that resolved root being non-empty.
        assert!(cmd.contains(
            "case \"$__tmpbase\" in /*/*) __tmpscan=$(cd \"$__tmpbase\" 2>/dev/null && pwd -P)"
        ));
        assert!(cmd.contains("case \"$__tmpscan\" in /*/*) ;; *) __tmpscan=\"\";; esac"));
        assert!(cmd.contains(
            "if [ -n \"$__tmpscan\" ]; then find \"$__tmpscan\" -maxdepth 1 -type d -name \"rch_target_*\" -prune"
        ));
        // The staged per-job/per-pid trees below the tmp base are scanned
        // through the same resolved root (a second, deeper find).
        assert!(
            cmd.matches("find \"$__tmpscan\" -maxdepth ").count() >= 2,
            "legacy and staged tmp-base passes must both use the resolved root: {cmd}"
        );
        // Never the unresolved, unguarded spelling.
        assert!(!cmd.contains("find \"$__tmpbase\""));
        assert!(!cmd.contains("-name \"rch_target_*\" -prune 2>/dev/null | "));
        assert!(cmd.contains("done < \"$__tmpf\""));
        assert!(!cmd.contains("-prune 2>/dev/null | \\\n"));
        assert!(!cmd.contains("-prune 2>/dev/null | while"));
        // Emits parseable metrics.
        assert!(cmd.contains("RCH_WORKER_REAP_METRICS"));
        // Exits cleanly if base is missing.
        assert!(cmd.contains("if [ ! -d \"$base\" ]"));
    }

    /// Run the generated sweep script under a real `sh` against a fabricated
    /// sync-root and assert which per-job dirs are reaped vs kept AND that the
    /// emitted metrics line reflects the actual removals. Linux/macOS only (needs
    /// `sh`, `find`, `du`, `mktemp`).
    #[cfg(unix)]
    #[test]
    fn sweep_script_reaps_idle_dirs_at_all_depths_and_keeps_live_and_empty() {
        use std::fs;
        use std::process::Command;
        use tempfile::tempdir;

        let tmp = tempdir().expect("create tmp root");
        let base = tmp.path().join("projects");
        fs::create_dir_all(&base).unwrap();

        // Helper: make a per-job dir at `path` with one file, then backdate the
        // whole subtree (via `touch -t`, portable across GNU/BSD) so it reads as
        // idle — well beyond the 12h window.
        let make_idle = |path: &std::path::Path| {
            fs::create_dir_all(path).unwrap();
            fs::write(path.join("artifact.o"), b"x").unwrap();
            let ok = Command::new("find")
                .arg(path)
                .args(["-exec", "touch", "-t", "202601010000", "{}", ";"])
                .status()
                .map(|s| s.success())
                .unwrap_or(false);
            assert!(ok, "aging {path:?} should succeed");
        };

        // Depth 1 directly under base: <base>/.rch-target-…-job-…. Depth 2:
        // <base>/<repo>/.rch-target-…. Depth 3 (workspace member):
        // <base>/<repo>/crates/<member>/.rch-target-….
        let idle_d1 = base.join(".rch-target-w-job-1-1-0");
        let idle_d2 = base.join("repoA").join(".rch-target-w-job-2-2-0");
        let idle_d3 = base
            .join("repoB")
            .join("crates")
            .join("member")
            .join(".rch-target-w-pid-3-3-0");
        make_idle(&idle_d1);
        make_idle(&idle_d2);
        make_idle(&idle_d3);

        // A FRESH (recently-touched) per-job dir must be kept.
        let live = base.join("repoC").join(".rch-target-w-job-4-4-0");
        fs::create_dir_all(&live).unwrap();
        fs::write(live.join("fresh.o"), b"y").unwrap(); // current mtime → live

        // An EMPTY just-created per-job dir (mkdir but no first write) must be kept
        // (recent dir mtime, zero files — the v1.0.35 race case).
        let empty = base.join("repoD").join(".rch-target-w-pid-5-5-0");
        fs::create_dir_all(&empty).unwrap();

        // A non-rch sibling dir must never be considered.
        let bystander = base.join("repoE").join("target");
        fs::create_dir_all(&bystander).unwrap();
        make_idle(&bystander); // even idle, the glob must not match it

        let cmd = build_sweep_command(base.to_str().unwrap(), 720, None, None);
        let out = run_sweep(&cmd, tmp.path());
        let stdout = String::from_utf8_lossy(&out.stdout);

        // Idle dirs at every depth reaped.
        assert!(!idle_d1.exists(), "idle depth-1 dir should be reaped");
        assert!(!idle_d2.exists(), "idle depth-2 dir should be reaped");
        assert!(
            !idle_d3.exists(),
            "idle depth-3 workspace-member dir should be reaped"
        );
        // Live, empty, and the bystander `target` kept.
        assert!(live.exists(), "freshly-touched dir must be kept");
        assert!(empty.exists(), "empty just-created dir must be kept");
        assert!(bystander.exists(), "non-rch `target` must never be touched");

        // Metrics survive the loop: exactly 3 removals are reported.
        let m = parse_reap_metrics(&stdout).expect("metrics line present");
        assert_eq!(
            m.removed, 3,
            "metrics must count all 3 reaped dirs: {stdout}"
        );
        assert!(
            m.freed_bytes > 0,
            "freed bytes must be > 0 when dirs were removed: {stdout}"
        );
    }

    #[test]
    fn sweep_script_pooled_pass_reaps_long_idle_pools_and_keeps_active_ones() {
        use std::fs;
        use std::process::Command;
        use tempfile::tempdir;

        let tmp = tempdir().expect("create tmp root");
        let base = tmp.path().join("projects");
        fs::create_dir_all(&base).unwrap();

        let make_aged = |path: &std::path::Path| {
            fs::create_dir_all(path).unwrap();
            fs::write(path.join("artifact.o"), b"x").unwrap();
            let ok = Command::new("find")
                .arg(path)
                .args(["-exec", "touch", "-t", "202601010000", "{}", ";"])
                .status()
                .map(|s| s.success())
                .unwrap_or(false);
            assert!(ok, "aging {path:?} should succeed");
        };

        // A pooled dir idle far beyond the 7-day pooled window: corpse, reap.
        let dead_pool = base.join("repoA").join(".rch-target-w-pool-deadbeef");
        make_aged(&dead_pool);
        // A freshly-written pooled dir: warm cache, keep.
        let live_pool = base.join("repoB").join(".rch-target-w-pool-cafef00d");
        fs::create_dir_all(&live_pool).unwrap();
        fs::write(live_pool.join("artifact.o"), b"x").unwrap();
        // An aged pooled dir with the pooled pass DISABLED must survive.
        let cmd_no_pool = build_sweep_command(base.to_str().unwrap(), 720, None, None);
        let out = run_sweep(&cmd_no_pool, tmp.path());
        assert!(out.status.success());
        assert!(
            dead_pool.exists(),
            "pooled dirs untouched without the pooled pass"
        );

        // With the pooled pass on, only the long-idle pool is reaped.
        let cmd = build_sweep_command(base.to_str().unwrap(), 720, Some(7 * 24 * 60), None);
        let out = run_sweep(&cmd, tmp.path());
        assert!(out.status.success());
        let stdout = String::from_utf8_lossy(&out.stdout);
        assert!(
            !dead_pool.exists(),
            "long-idle pooled dir must be reaped: {stdout}"
        );
        assert!(live_pool.exists(), "active pooled dir must be kept");
        let m = parse_reap_metrics(&stdout).expect("metrics line present");
        assert_eq!(m.removed, 1, "exactly the dead pool is counted: {stdout}");
    }

    #[test]
    fn sweep_script_cap_pass_evicts_oldest_until_under_budget() {
        use std::fs;
        use std::process::Command;
        use tempfile::tempdir;

        let tmp = tempdir().expect("create tmp root");
        let base = tmp.path().join("projects");
        fs::create_dir_all(&base).unwrap();

        // Three ~200KB pooled dirs with staggered ages (all idle beyond the
        // short window, all within the pooled TTL since we pass None), plus a
        // fresh oversized one that the safety floor must protect.
        let aged = |path: &std::path::Path, stamp: &str| {
            fs::create_dir_all(path).unwrap();
            fs::write(path.join("blob.bin"), vec![0u8; 200 * 1024]).unwrap();
            let ok = Command::new("find")
                .arg(path)
                .args(["-exec", "touch", "-t", stamp, "{}", ";"])
                .status()
                .map(|s| s.success())
                .unwrap_or(false);
            assert!(ok, "aging {path:?} should succeed");
        };
        let oldest = base.join("repoA").join(".rch-target-w-pool-oldest");
        let middle = base.join("repoB").join(".rch-target-w-pool-middle");
        let newest = base.join("repoC").join(".rch-target-w-pool-newest");
        aged(&oldest, "202601010000");
        aged(&middle, "202602010000");
        aged(&newest, "202603010000");
        let fresh_big = base.join("repoD").join(".rch-target-w-pool-freshbig");
        fs::create_dir_all(&fresh_big).unwrap();
        fs::write(fresh_big.join("blob.bin"), vec![0u8; 400 * 1024]).unwrap();

        // Budget 700KB. Total is ~1020KB (3×~204 + ~404), so eviction fires;
        // the fresh dir is protected by the -mmin floor; evicting the two
        // oldest aged dirs (~408KB) brings the total to ~612KB ≤ 700 and the
        // loop stops before the newest aged dir.
        let cmd = build_sweep_command(base.to_str().unwrap(), 720, None, Some(700));
        let out = run_sweep(&cmd, tmp.path());
        assert!(out.status.success());
        let stdout = String::from_utf8_lossy(&out.stdout);

        assert!(
            !oldest.exists(),
            "oldest aged pool must be evicted: {stdout}"
        );
        assert!(
            !middle.exists(),
            "middle aged pool must be evicted: {stdout}"
        );
        assert!(
            newest.exists(),
            "newest aged pool must survive once under budget: {stdout}"
        );
        assert!(
            fresh_big.exists(),
            "fresh dir is protected by the -mmin floor"
        );
        let m = parse_reap_metrics(&stdout).expect("metrics line present");
        assert_eq!(m.removed, 2, "exactly the two oldest are counted: {stdout}");
        assert!(
            m.freed_bytes >= 300 * 1024,
            "freed reflects evictions: {stdout}"
        );
    }

    #[test]
    fn sweep_script_reports_failed_removals_instead_of_silence() {
        use std::fs;
        use std::os::unix::fs::PermissionsExt;
        use std::process::Command;
        use tempfile::tempdir;

        let tmp = tempdir().expect("create tmp root");
        let base = tmp.path().join("projects");
        let parent = base.join("repoA");
        let victim = parent.join(".rch-target-w-job-err-1-0");
        fs::create_dir_all(&victim).unwrap();
        fs::write(victim.join("artifact.o"), b"x").unwrap();
        let ok = Command::new("find")
            .arg(&victim)
            .args(["-exec", "touch", "-t", "202601010000", "{}", ";"])
            .status()
            .map(|s| s.success())
            .unwrap_or(false);
        assert!(ok, "aging should succeed");

        // A read-only parent makes the unlink fail (as non-root).
        fs::set_permissions(&parent, fs::Permissions::from_mode(0o555)).unwrap();
        // Root (CAP_DAC_OVERRIDE) ignores directory permissions, so the
        // read-only parent cannot defeat the rm — and some rch workers run
        // this suite as root. Probe the precondition instead of asserting it:
        // when a write into the read-only parent succeeds, the failure path
        // under test is unreachable on this host, so skip rather than fail.
        if fs::write(parent.join(".rch-perm-probe"), b"").is_ok() {
            fs::set_permissions(&parent, fs::Permissions::from_mode(0o755)).unwrap();
            eprintln!(
                "skipping sweep_script_reports_failed_removals_instead_of_silence: \
                 a read-only parent does not defeat writes here (running as root?)"
            );
            return;
        }
        let cmd = build_sweep_command(base.to_str().unwrap(), 720, None, None);
        let out = run_sweep(&cmd, tmp.path());
        // Restore before asserting so tempdir cleanup works even on failure.
        fs::set_permissions(&parent, fs::Permissions::from_mode(0o755)).unwrap();

        assert!(out.status.success());
        let stdout = String::from_utf8_lossy(&out.stdout);
        assert!(victim.exists(), "read-only parent must defeat the rm");
        let errors = stale_target_reap::parse_reap_errors(&stdout);
        assert_eq!(errors.len(), 1, "the failed rm must be reported: {stdout}");
        assert_eq!(errors[0].trigger, "ttl");
        assert!(errors[0].path.ends_with(".rch-target-w-job-err-1-0"));
        let m = parse_reap_metrics(&stdout).expect("metrics line present");
        assert_eq!(
            m.removed, 0,
            "failed removals must not be counted: {stdout}"
        );
    }

    #[test]
    fn parse_metrics_reads_values() {
        let out = "noise\nRCH_WORKER_REAP_METRICS removed=3 freed_kb=2048\n";
        let m = parse_reap_metrics(out).expect("parse");
        assert_eq!(m.removed, 3);
        assert_eq!(m.freed_bytes, 2048 * 1024);
    }

    #[test]
    fn parse_metrics_missing_line() {
        assert!(parse_reap_metrics("no metrics\n").is_none());
    }

    #[test]
    fn nonzero_sweep_command_result_is_error() {
        let result = CommandResult {
            exit_code: 2,
            stdout: "RCH_WORKER_REAP_METRICS removed=0 freed_kb=0\n".to_string(),
            stderr: "find failed".to_string(),
            duration_ms: 10,
        };

        let error = ensure_sweep_command_success(&result).expect_err("nonzero exit is an error");

        assert!(error.to_string().contains("exited with code 2"));
        assert!(error.to_string().contains("find failed"));
    }

    #[test]
    fn config_default_is_off_and_conservative() {
        let c = StaleTargetReapConfig::default();
        // Ships default-OFF: an autonomous periodic deleter pointed at
        // /data/projects is opt-in until canary-soaked (carnage-history lesson).
        assert!(!c.enabled);
        assert_eq!(c.interval_mins, 120);
        assert_eq!(c.idle_hours, 12);
        // Default base is the remote repo sync-root (canonical project root),
        // where rch actually places per-job target dirs — NOT `/tmp/rch`.
        assert_eq!(c.remote_base, rch_common::DEFAULT_CANONICAL_PROJECT_ROOT);
        assert_eq!(c.remote_base, "/data/projects");
    }

    #[tokio::test]
    async fn cycle_skips_unhealthy_workers() {
        use rch_common::WorkerConfig;
        let pool = WorkerPool::new();
        let worker_id = WorkerId::new("w1");
        pool.add_worker(WorkerConfig {
            id: worker_id.clone(),
            host: "h".into(),
            user: "u".into(),
            identity_file: "/home/u/.ssh/id".into(),
            total_slots: 4,
            priority: 50,
            tags: vec![],
            tools: Vec::new(),
        })
        .await;
        // Newly-added workers default to `Healthy` (see `WorkerState::new`), so
        // force a non-healthy status to exercise the skip path: an Unreachable
        // worker must be skipped and no SSH attempted.
        pool.set_status(&worker_id, WorkerStatus::Unreachable).await;
        let reaper = StaleTargetReaper::new(pool, StaleTargetReapConfig::default());
        let stats = reaper.run_cycle().await;
        assert_eq!(stats.workers_checked, 1);
        assert_eq!(stats.workers_skipped, 1);
        assert_eq!(stats.workers_swept, 0);
        assert_eq!(stats.errors, 0);
    }

    /// Issue #64: pooled stores placed off the mirror tree must still be
    /// swept, so the reaper walks `store_base` in addition to `remote_base` —
    /// and sweeps once when the two coincide.
    #[test]
    fn scan_bases_cover_the_pooled_store_base() {
        let reaper = |store_base: Option<&str>| {
            StaleTargetReaper::new(
                WorkerPool::new(),
                StaleTargetReapConfig {
                    remote_base: "/data/projects".to_string(),
                    store_base: store_base.map(str::to_string),
                    ..StaleTargetReapConfig::default()
                },
            )
            .scan_bases()
        };

        assert_eq!(reaper(None), vec!["/data/projects".to_string()]);
        assert_eq!(
            reaper(Some("/bigdisk/rch-pools")),
            vec![
                "/data/projects".to_string(),
                "/bigdisk/rch-pools".to_string()
            ]
        );
        assert_eq!(
            reaper(Some("/data/projects")),
            vec!["/data/projects".to_string()],
            "a store_base equal to remote_base must not sweep twice"
        );
    }
}