Skip to main content

magi/
clean.rs

1//! The disk janitor: finished runs get their worktrees folded, worktrees whose
2//! run record is already gone get reclaimed too, and the shared build cache is
3//! pruned to its cap.
4//!
5//! A run's state being written by an older schema is not the same thing as it
6//! being unreadable, and this module used to conflate the two: [`fold_due`]
7//! treated any `run.json` its version check rejected exactly like one that
8//! failed to parse at all, so a single schema bump silently stopped every
9//! automatic fold in the fleet the moment it shipped, and did so with no
10//! counter and no log line to say so. A record magi genuinely cannot parse —
11//! missing fields, broken JSON, a schema newer than this build has ever heard
12//! of — is still left alone here, still counted in
13//! [`Housekeeping::unreadable`], and still only ever removed by an explicit
14//! operator action (`magi fold`, or the equivalent phone route). One written
15//! by a schema this build merely disagrees with the *meaning* of is not that:
16//! as long as it still parses, folding proceeds regardless of the number in
17//! its `schema` field.
18//!
19//! Everything policy-shaped — which statuses are foldable, how long a finished
20//! run is left alone, whether the cache is over its limit — is a pure function
21//! injected with numbers, so nothing here has to ask the operating system to
22//! be testable. The only I/O is the removal itself.
23
24use std::collections::BTreeSet;
25use std::path::{Path, PathBuf};
26
27use anyhow::{Context as _, Result, bail};
28use jiff::{SignedDuration, Timestamp};
29use serde::Deserialize;
30
31use crate::ask::Questions;
32use crate::config::Disk;
33use crate::run::{RunState, RunStatus, SCHEMA, short_of};
34
35use crate::disk::{Prune, dir_size};
36
37/// What one janitor pass did, for the caller's log line.
38#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
39pub struct Housekeeping {
40    /// Runs folded (worktrees dropped).
41    pub folded: usize,
42    /// Runs [`fold_due`] left alone because their `run.json` genuinely could
43    /// not be read - missing, broken JSON, a schema this build has never
44    /// heard of - as opposed to one merely written by a different schema
45    /// number, which is folded like any other (see the module docs). This was
46    /// defined but never incremented for a long stretch of this module's
47    /// history, which is exactly how 90 of 93 runs sat unfolded on one
48    /// operator's machine with nothing anywhere saying why: every one of them
49    /// was misclassified as unreadable by a schema check that has since been
50    /// narrowed to only the runs that actually are.
51    pub unreadable: usize,
52    /// Worktrees under the worktree bay reclaimed because no run record in
53    /// `runs/` claims them anymore (see [`fold_orphaned_worktrees`]).
54    pub orphaned_worktrees: usize,
55    /// Files dropped from the shared cache.
56    pub cache_files: usize,
57    /// Bytes freed from the shared cache.
58    pub cache_freed: u64,
59    /// Open questions abandoned because the run that asked them has already
60    /// settled where nothing is coming back to read an answer.
61    pub questions_abandoned: usize,
62    /// Runs found blocked on a merge magi never recorded — a pull request the
63    /// operator merged by hand while `land::land` never got as far as
64    /// opening one itself — and corrected automatically. See
65    /// [`reconcile_external_merges`].
66    pub external_merges_recorded: usize,
67    /// Run records that still called their pull request open after it was
68    /// merged or closed, and were rewritten (see
69    /// [`crate::land::repair_stale_pr_states`]).
70    pub stale_pr_states_repaired: usize,
71}
72
73/// Run the janitor: fold due runs, reclaim orphaned worktrees, prune stale
74/// worktree registrations, then prune the cache if it is over its cap.
75///
76/// Every part is best-effort; a jammed cache lock or a run whose worktree
77/// another borrower holds must not stop the rest. Errors are reported through
78/// `tracing::warn` - this is housekeeping, and the daemon keeps serving
79/// either way.
80pub async fn housekeep(
81    cfg: &crate::config::Config,
82    home: &Path,
83    worktrees_root: &Path,
84    repo: &Path,
85    now: Timestamp,
86) -> Housekeeping {
87    let mut out = Housekeeping::default();
88    if cfg.disk.auto_fold {
89        let runs = home.join("runs");
90        match fold_due(&runs, home, worktrees_root, &cfg.disk, now).await {
91            Ok((folded, unreadable)) => {
92                out.folded = folded;
93                out.unreadable = unreadable;
94            }
95            Err(e) => {
96                tracing::warn!("housekeep: fold due runs: {e:#}");
97                // Stable wording: the reason varies per pass, and a changed
98                // message would relight the bell on every retry.
99                crate::notices::raise_in(
100                    home,
101                    crate::notices::Notice::warn(
102                        "housekeep:fold",
103                        "Automatic cleanup of finished runs failed; disk usage may keep growing.",
104                    ),
105                );
106            }
107        }
108        out.orphaned_worktrees =
109            fold_orphaned_worktrees(&runs, worktrees_root, home, cfg.disk.fold_grace_secs, now)
110                .await;
111        // Best-effort in the same sense as everything else here: a repository
112        // this janitor pass has nothing to do with (or none at all, in a unit
113        // test) must not turn a `warn` into a reason to skip the rest.
114        if let Err(e) = crate::git::worktree_prune(repo).await {
115            tracing::warn!("housekeep: prune worktree registrations: {e:#}");
116        }
117        out.external_merges_recorded = reconcile_external_merges(&runs, home, &cfg.disk, now).await;
118        // Bounded: a handful of forge lookups a pass keeps this cheap, and the
119        // pass is idempotent, so a backlog drains over several passes.
120        out.stale_pr_states_repaired = crate::land::repair_stale_pr_states(home, 5).await.0;
121    }
122    match prune_cache_if_over_limit(cfg, home) {
123        Ok(Some(pruned)) => {
124            out.cache_files = pruned.files;
125            out.cache_freed = pruned.freed;
126        }
127        Ok(None) => {}
128        Err(e) => tracing::warn!("housekeep: prune cache: {e:#}"),
129    }
130    // Unconditional, unlike the two passes above: this is not a disk policy
131    // with a cap or an opt-out, it is closing a gap `graph::Runner` itself
132    // cannot - a run that reached `Merged`/`Ready`/`Failed` before this
133    // cleanup existed, or whose process died between saving that status and
134    // abandoning the question it leaves behind (see `Runner::settle_questions`).
135    // Left alone, that question sits `open` forever: the owner's badge,
136    // banner and title all keep counting a decision nobody is left to read.
137    out.questions_abandoned =
138        abandon_settled_questions(&Questions::at(home.join("questions")), &home.join("runs"));
139    out
140}
141
142/// Abandon every open question whose run has already settled into a status
143/// nothing comes back from, worded with what the run became - the same
144/// cleanup `graph::Runner::settle_questions` runs the moment `status` lands
145/// there, for questions that missed it.
146///
147/// Scans questions rather than runs: the open list is normally short, and a
148/// run that never asked anything costs nothing here. A run this cannot read,
149/// deleted or written by a schema this build does not speak, is left alone
150/// the same as everywhere else in this module; the question stays open
151/// rather than guessed at.
152pub fn abandon_settled_questions(store: &Questions, runs: &Path) -> usize {
153    let waiting_on: BTreeSet<String> = store
154        .list()
155        .into_iter()
156        .filter(|q| q.status.open())
157        .map(|q| q.run)
158        .collect();
159    let mut abandoned = 0;
160    for run in waiting_on {
161        let Ok(meta) = read_meta(runs, &run) else {
162            continue;
163        };
164        match store.settle_run(&run, meta.status) {
165            Ok(n) => abandoned += n,
166            Err(e) => tracing::warn!("housekeep: abandon questions for {run}: {e:#}"),
167        }
168    }
169    abandoned
170}
171
172/// Fold every run that is finished, older than the grace period, and not being
173/// worked on; return `(folded, unreadable)`.
174///
175/// A run whose `run.json` genuinely cannot be parsed — missing fields, broken
176/// JSON, a schema newer than this build has ever heard of — is left exactly
177/// as it is. Automatic housekeeping cannot tell a mid-write file from one that
178/// will never parse again, and `<home>/runs/<id>/` is the evidence `magi
179/// stats` and the deck read; when unsure whether it is safe to touch, the
180/// janitor keeps rather than deletes (see the module docs). Discarding a
181/// record this unreadable is an explicit operator action (`magi fold`, or the
182/// equivalent phone route), never something that happens unattended. Every
183/// such skip is counted in the returned `unreadable` and logged through
184/// `tracing::warn` with the parse failure that caused it - silence here is
185/// exactly the failure mode that let 90 of 93 runs sit unfolded with nothing
186/// to show for it.
187///
188/// A run merely written by a *different* schema number is not unreadable: as
189/// long as `run.json` still parses, it folds like any other terminal run (see
190/// the module docs for why the two are different questions).
191///
192/// Runnable statuses and runs newer than the grace period are also left
193/// alone; folding them would throw away work that is still the answer to
194/// somebody's question. `Merged` runs forget their winner's worktree (the
195/// merge already landed it); `Ready` and `Failed` runs keep it.
196pub async fn fold_due(
197    runs: &Path,
198    home: &Path,
199    _worktrees_root: &Path,
200    disk: &Disk,
201    now: Timestamp,
202) -> Result<(usize, usize)> {
203    let mut folded = 0usize;
204    let mut unreadable = 0usize;
205    let mut ids: Vec<String> = std::fs::read_dir(runs)
206        .into_iter()
207        .flatten()
208        .flatten()
209        .filter(|e| e.path().join("run.json").is_file())
210        .map(|e| e.file_name().to_string_lossy().into_owned())
211        .collect();
212    ids.sort_unstable();
213    for id in ids {
214        if crate::pr_recovery::sources_retained(home, &id) {
215            continue;
216        }
217        if crate::daemon::is_working_on(home, &id, now) {
218            continue;
219        }
220        let meta = match read_meta(runs, &id) {
221            Ok(meta) => meta,
222            Err(e) => {
223                unreadable += 1;
224                tracing::warn!("housekeep: run {id} unreadable, left alone: {e:#}");
225                continue;
226            }
227        };
228        if meta.status.resumable() || !due(now, meta.updated_at, disk.fold_grace_secs) {
229            continue;
230        }
231        // `read_meta` already proved the file parses; `read_state` asks for
232        // the rest of the fields `graph::fold_run` needs (worktree paths,
233        // candidates, tally). A schema mismatch alone does not fail this -
234        // see the module docs - so reaching `Err` here means the JSON itself
235        // is broken in a way `read_meta` did not exercise, which is rare but
236        // not impossible (a body truncated between the two fields it reads
237        // and the rest). That must not cost every other run its turn through
238        // this loop, so it is a skip, not a `?`.
239        let mut state = match read_state(runs, &id) {
240            Ok(state) => state,
241            Err(e) => {
242                unreadable += 1;
243                tracing::warn!("housekeep: run {id} unreadable, left alone: {e:#}");
244                continue;
245            }
246        };
247        if state.schema != SCHEMA {
248            tracing::info!(
249                "housekeep: run {id} was written by schema {} (this build speaks {SCHEMA}); \
250                 folding it anyway",
251                state.schema
252            );
253        }
254        // `Merged`'s winner branch is safe to drop because it already landed
255        // in the real target branch; `Superseded`'s winner branch is safe to
256        // drop for a different reason - a later attempt at the same task is
257        // what actually landed (or didn't), so *this* run's own winner is
258        // never going to be merged by anyone. Every other terminal status
259        // keeps the winner: `Ready`/`Blocked`/`Stalled`/`Failed`/`VerifiedNoop`
260        // may still have a human's decision pending on that exact branch.
261        // `AlreadyInBase` is the same: the change is on the base under other
262        // commit ids, so the branch holds nothing anyone still needs.
263        let drop_winner = matches!(
264            state.status,
265            RunStatus::Merged | RunStatus::Superseded | RunStatus::AlreadyInBase
266        );
267        // One run's fold must not cost every later run its turn. A worktree
268        // another borrower holds, a branch git refuses to delete, a repository
269        // that has since moved: each is a reason this run cannot be folded
270        // now, and none is a reason to stop the pass. Left unfolded, it is
271        // simply due again next time; a `?` here stopped automatic folding
272        // permanently at the first such run (finding R3-1-1 of run 51a3).
273        match crate::graph::fold_run(&mut state, drop_winner, home).await {
274            Ok(_) => folded += 1,
275            Err(e) => tracing::warn!("housekeep: fold {id}: {e:#}"),
276        }
277    }
278    Ok((folded, unreadable))
279}
280
281/// What one [`fetch_origins`] pass did.
282#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
283pub struct FetchReport {
284    /// Repos whose fetch succeeded.
285    pub fetched: usize,
286    /// Repos with no `origin` remote, skipped.
287    pub no_origin: usize,
288    /// Repos whose fetch failed or timed out.
289    pub failed: usize,
290}
291
292/// Run `git fetch origin` in every checkout under `roots` (the set `magi
293/// repos` lists), so `origin/main` stays current.
294///
295/// Fetch only: never a checkout, merge or fast-forward, so no HEAD, branch or
296/// working tree moves. Best-effort like [`fold_due`]: a repo with no `origin`
297/// is skipped silently, a failure or timeout is a warning, and the pass never
298/// errors. `stop` is checked between repos.
299pub async fn fetch_origins(
300    roots: &[PathBuf],
301    per_repo: std::time::Duration,
302    stop: impl Fn() -> bool,
303) -> FetchReport {
304    let mut report = FetchReport::default();
305    for repo in crate::repos::scan(roots) {
306        if stop() {
307            break;
308        }
309        match crate::git::git_raw(&repo.path, &["remote", "get-url", "origin"]).await {
310            Ok(out) if out.ok() => {}
311            _ => {
312                report.no_origin += 1;
313                continue;
314            }
315        }
316        match crate::git::fetch_origin(&repo.path, per_repo).await {
317            Ok(()) => report.fetched += 1,
318            Err(e) => {
319                tracing::warn!("fetch origin in {}: {e:#}", repo.name);
320                report.failed += 1;
321            }
322        }
323    }
324    report
325}
326
327/// Is `updated` old enough, measured against `now`, that the run may fold?
328///
329/// Pure; the janitor compares against wallclock, tests inject both sides. The
330/// comparison is strict, so a run exactly at the edge of its grace period is
331/// left alone one more pass — the same convention as [`crate::disk::over_limit`].
332pub fn due(now: Timestamp, updated: Timestamp, grace_secs: u64) -> bool {
333    now.duration_since(updated) > SignedDuration::new(grace_secs as i64, 0)
334}
335
336/// The two fields the janitor decides on, read with a serde that tolerates
337/// everything else about the run being unreadable.
338#[derive(Deserialize)]
339struct Meta {
340    status: RunStatus,
341    updated_at: Timestamp,
342}
343
344/// Read `status` and `updated_at` straight off the state file, asking for
345/// nothing else. `Err` when the file is missing, not parseable, or a status in
346/// a version this build does not speak - all of which mean "unreadable".
347fn read_meta(runs: &Path, id: &str) -> Result<Meta> {
348    let path = runs.join(id).join("run.json");
349    let body =
350        std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
351    let meta: Meta =
352        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
353    Ok(meta)
354}
355
356/// Runs checked against GitHub in one janitor pass, at most.
357///
358/// A fleet with many stuck runs must not turn every idle tick into a burst of
359/// `gh pr list` calls; a run left unchecked this pass is checked again next
360/// pass, same as an unfolded due run is.
361const MAX_EXTERNAL_MERGE_CHECKS_PER_PASS: usize = 5;
362
363/// The fields [`reconcile_external_merges`] filters on before paying for a
364/// full [`RunState`] parse or a `gh` round trip.
365#[derive(Deserialize)]
366struct ExternalMergeMeta {
367    status: RunStatus,
368    updated_at: Timestamp,
369    #[serde(default)]
370    merge: Option<crate::run::MergeOutcome>,
371}
372
373fn read_external_merge_meta(runs: &Path, id: &str) -> Result<ExternalMergeMeta> {
374    let path = runs.join(id).join("run.json");
375    let body =
376        std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
377    let meta: ExternalMergeMeta =
378        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
379    Ok(meta)
380}
381
382/// Is a run worth asking GitHub about?
383///
384/// Pure, so the filter itself is assertable without a fixture that writes
385/// `run.json` files or spawns `gh`. Narrower than "blocked": `merge_recorded`
386/// excludes a run `land::land` already finished landing (successfully or
387/// not) - that path already carries its own record of what happened - and
388/// `due` gives an operator who is about to fix a run by hand a grace window
389/// before the janitor starts asking the same question about it.
390pub fn eligible_for_external_merge_check(
391    status: RunStatus,
392    merge_recorded: bool,
393    updated_at: Timestamp,
394    now: Timestamp,
395    grace_secs: u64,
396) -> bool {
397    status == RunStatus::Blocked && !merge_recorded && due(now, updated_at, grace_secs)
398}
399
400/// Rotate `ids` so a fixed per-pass cap does not always land on the same
401/// prefix of the list, starting at `start` (already reduced modulo nothing -
402/// callers pass whatever a persisted cursor last recorded, and this takes it
403/// modulo `ids.len()` itself).
404///
405/// Pure: the offset is the caller's problem, not wall-clock time. An earlier
406/// version derived `start` from `now.as_second() % ids.len()`, which looked
407/// independent of any state to persist but is not actually independent of
408/// how often the janitor runs - a poll interval and a pass's own duration
409/// that alias onto the same handful of `now.as_second()` values (say, a
410/// ten-second poll and a pass that finishes in under a second) revisit only
411/// those same few residues forever and never advance into the rest of the
412/// list at all. [`reconcile_external_merges`] instead persists a cursor
413/// across passes ([`read_external_merge_cursor`] /
414/// [`write_external_merge_cursor`]) and advances it by exactly how many ids
415/// this pass actually looked at, which is the only quantity that is
416/// guaranteed to move forward pass over pass regardless of timing.
417fn rotate_from(ids: &[String], start: usize) -> Vec<String> {
418    if ids.is_empty() {
419        return Vec::new();
420    }
421    let start = start % ids.len();
422    ids[start..]
423        .iter()
424        .chain(ids[..start].iter())
425        .cloned()
426        .collect()
427}
428
429/// Where [`reconcile_external_merges`] remembers how far it got, so a fixed
430/// per-pass cap advances through a fleet's eligible runs pass over pass
431/// instead of camping on whichever ones happen to sort first.
432const EXTERNAL_MERGE_CURSOR_FILE: &str = "external-merge-cursor";
433
434/// `0` for anything this cannot read as a plain number - a first run, a
435/// corrupt or missing file, a build that has never written one - which is
436/// exactly as good a starting point as any: the cursor's whole job is to
437/// keep moving, not to encode any particular position as meaningful.
438fn read_external_merge_cursor(home: &Path) -> usize {
439    std::fs::read_to_string(home.join(EXTERNAL_MERGE_CURSOR_FILE))
440        .ok()
441        .and_then(|s| s.trim().parse().ok())
442        .unwrap_or(0)
443}
444
445/// Best-effort, like every other write in this module's sweep: a failure to
446/// persist the cursor costs the fleet one pass's worth of forward progress
447/// on this sweep, never the run whose housekeeping it is racing to keep up
448/// with.
449fn write_external_merge_cursor(home: &Path, cursor: usize) {
450    if let Err(e) = std::fs::write(home.join(EXTERNAL_MERGE_CURSOR_FILE), cursor.to_string()) {
451        tracing::warn!("housekeep: persist the external-merge sweep's cursor: {e:#}");
452    }
453}
454
455/// Close the gap `magi fold --merged` exists for, without waiting for an
456/// operator to notice and go find the pull request URL: a run that stopped
457/// `Blocked` with no `merge` recorded may have been merged anyway, by hand,
458/// on a pull request `land::land` never opened or never got to observe as
459/// merged. `land::find_external_merge` asks GitHub about the run's own
460/// winning branch, and only a pull request it can uniquely tie back to this
461/// run is ever acted on - see that function's own doc for what "uniquely"
462/// excludes.
463///
464/// A run this finds and fixes goes through
465/// [`crate::land::correct_confirmed_external_merge`] - the same correction
466/// `magi fold --merged` performs, minus the same-repo guard that command
467/// needs and this loop doesn't (see that function's own doc: the URL here
468/// was never operator-supplied, it came from asking `state.repo`'s own
469/// remote) - then through [`crate::graph::fold_run`] so it stops holding
470/// worktrees the moment it stops needing them. A run this cannot decide
471/// about - `gh` unreachable, more than one candidate pull
472/// request, nothing found at all - is left exactly as it is; only an error
473/// asking GitHub raises a notice, since "nothing found" is the ordinary,
474/// expected shape of a run that really is just blocked.
475async fn reconcile_external_merges(runs: &Path, home: &Path, disk: &Disk, now: Timestamp) -> usize {
476    let mut ids: Vec<String> = std::fs::read_dir(runs)
477        .into_iter()
478        .flatten()
479        .flatten()
480        .filter(|e| e.path().join("run.json").is_file())
481        .map(|e| e.file_name().to_string_lossy().into_owned())
482        .collect();
483    ids.sort_unstable();
484
485    // Every eligible id first, cheaply (a `Meta` read, no `gh`), then rotated
486    // before the cap is applied. Capping the sorted order directly would
487    // always land on the same lexicographic prefix - a fleet's oldest run ids
488    // - so a handful of long-blocked runs that never turn out to be merged
489    // would starve every id after them of a `gh` check, forever.
490    let mut eligible: Vec<String> = Vec::new();
491    for id in ids {
492        if crate::daemon::is_working_on(home, &id, now) {
493            continue;
494        }
495        let meta = match read_external_merge_meta(runs, &id) {
496            Ok(meta) => meta,
497            // Already counted as unreadable by `fold_due`'s own pass; not
498            // worth a second warning for the same file.
499            Err(_) => continue,
500        };
501        if eligible_for_external_merge_check(
502            meta.status,
503            meta.merge.is_some(),
504            meta.updated_at,
505            now,
506            disk.fold_grace_secs,
507        ) {
508            eligible.push(id);
509        }
510    }
511
512    if eligible.is_empty() {
513        return 0;
514    }
515    let cursor = read_external_merge_cursor(home);
516    let window: Vec<String> = rotate_from(&eligible, cursor)
517        .into_iter()
518        .take(MAX_EXTERNAL_MERGE_CHECKS_PER_PASS)
519        .collect();
520    // Advances by exactly how many ids this pass looked at, regardless of
521    // what came of looking - a run that turned out not to be merged after
522    // all must not be revisited before every other eligible run has had its
523    // own turn.
524    write_external_merge_cursor(home, (cursor + window.len()) % eligible.len());
525
526    let mut reconciled = 0usize;
527    for id in window {
528        let mut state = match read_state(runs, &id) {
529            Ok(state) => state,
530            Err(_) => continue,
531        };
532        match crate::land::find_external_merge(&state).await {
533            Ok(Some(found)) => {
534                match crate::land::correct_confirmed_external_merge(&mut state, &found.url).await {
535                    Ok(_) => {
536                        if let Err(e) = crate::graph::fold_run(&mut state, true, home).await {
537                            tracing::warn!(
538                                "housekeep: fold {id} after recording its external merge: {e:#}"
539                            );
540                        }
541                        reconciled += 1;
542                    }
543                    Err(e) => tracing::warn!(
544                        "housekeep: record external merge {} for {id}: {e:#}",
545                        found.url
546                    ),
547                }
548            }
549            Ok(None) => {}
550            Err(e) => {
551                tracing::warn!("housekeep: check external merge for {id}: {e:#}");
552                crate::notices::raise_in(
553                    home,
554                    crate::notices::Notice::warn(
555                        &format!("merged-unrecorded:{id}"),
556                        format!(
557                            "Run {id} is blocked with no recorded merge, and checking GitHub \
558                             for a merge failed; check by hand."
559                        ),
560                    )
561                    .link(crate::notices::Link::Run { id: id.clone() }),
562                );
563            }
564        }
565    }
566    reconciled
567}
568
569/// Read a whole run state from a runs directory, for folding only.
570///
571/// Deliberately more permissive than [`RunState::load`], which this does not
572/// call: `load` backs `--resume` and every hand-driven command, where a
573/// schema this build disagrees with the *meaning* of must refuse outright
574/// rather than resume a review round or a tally against stale semantics
575/// (`RunState::SCHEMA`'s own docs list what has changed meaning at each
576/// bump). Folding recomputes nothing - it only reads worktree paths, branch
577/// names and a tally winner off the struct to remove them - so an old
578/// schema's values are exactly as good here as a current one's; every schema
579/// bump so far has only ever added a field or a variant, never repurposed an
580/// existing one, and serde already fills an added field's default when an
581/// older record has nothing to say about it. What this cannot tolerate, and
582/// what still surfaces as an `Err`, is `run.json` failing to parse at all.
583fn read_state(runs: &Path, id: &str) -> Result<RunState> {
584    let path = runs.join(id).join("run.json");
585    let body =
586        std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
587    let state: RunState =
588        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
589    Ok(state)
590}
591
592/// Remove a run that cannot be read: its state directory under `runs` and its
593/// worktree directory under `worktrees_root`.
594///
595/// The state file is the only record of a run's repository and branches, so a
596/// run this unreadable is discarded at the filesystem level - there is no
597/// candidate list to fold first. The worktrees live under
598/// [`crate::run::default_worktree_root`] unless the run's config relocated
599/// them, which an unreadable run cannot tell us; the default location is
600/// removed, and anything the run placed elsewhere is a leftover for whoever
601/// knows where it went.
602///
603/// Deleting a worktree directory by hand leaves its registration in git, and a
604/// registered path cannot be re-`worktree add`-ed until it is pruned - so every
605/// worktree is unregistered from its repository first, best-effort, via the
606/// `gitdir:` link git keeps inside the directory.
607pub async fn fold_unreadable(runs: &Path, worktrees_root: &Path, id: &str) -> Result<Vec<String>> {
608    let resolved = resolve_id_path(runs, id)?;
609    let mut removed = Vec::new();
610    let run_dir = runs.join(&resolved);
611    if run_dir.exists() {
612        std::fs::remove_dir_all(&run_dir)
613            .with_context(|| format!("remove {}", run_dir.display()))?;
614        removed.push(format!("runs/{resolved}"));
615    }
616    let wt = worktrees_root.join(short_of(&resolved));
617    if wt.exists() {
618        crate::git::remove_worktree_from_linked(&wt).await;
619        for e in std::fs::read_dir(&wt).into_iter().flatten().flatten() {
620            crate::git::remove_worktree_from_linked(&e.path()).await;
621        }
622        std::fs::remove_dir_all(&wt).with_context(|| format!("remove {}", wt.display()))?;
623        removed.push(wt.to_string_lossy().into_owned());
624    }
625    Ok(removed)
626}
627
628/// Reclaim worktrees under `worktrees_root` that no run record in `runs`
629/// claims anymore, and return how many were removed.
630///
631/// [`fold_due`] only ever sees a worktree by walking `runs/` first, so a
632/// worktree whose run record is already gone — `magi run rm`, or a record
633/// deleted before its worktree — never enters that loop at all: nothing there
634/// is looking for it. This walks the worktree bay directly instead, and
635/// removes any `<short>` directory that no run id maps to.
636///
637/// Two things must never happen, and this checks both before ever touching a
638/// directory:
639///
640/// - **A worktree bay is never the only kind of thing under `worktrees_root`,
641///   and this must not assume it is.** A hand-placed scratch directory, or
642///   anything else an operator or another tool left in the same bay, has the
643///   same "no run claims it" shape as a genuine orphan but is not one -
644///   [`looks_like_a_worktree_bay`] is the same tag shape [`crate::run::is_run_id`]
645///   already requires of a real run's short id, and anything else is left
646///   alone regardless of what else is true about it.
647/// - **A worktree that was only just created might not have a `run.json` yet
648///   for a reason that has nothing to do with being orphaned.** `Runner::start`
649///   and `Runner::review` both create the worktree before the first
650///   `RunState::save` lands, and that gap - several `git` subprocesses wide -
651///   is invisible to [`crate::daemon::is_working_on_short`] whenever the run
652///   is not being driven through this daemon's own `poll` loop at all (a
653///   `magi review` invocation, for one). A directory whose own modification
654///   time is within `grace_secs` of `now` is left alone on that basis alone,
655///   the same margin [`fold_due`] gives a run before treating it as truly
656///   finished - long enough that no realistic gap between a `worktree add`
657///   and its `run.json` could ever be mistaken for one.
658///
659/// The one failure this must never cause is deleting the worktree of a run
660/// that is genuinely in flight. [`crate::daemon::is_working_on_short`] is the
661/// same liveness check [`fold_due`] trusts everywhere else in this module,
662/// checked by short id because there is no full id to compare here; when it
663/// cannot tell, this leaves the directory alone. Best-effort like the rest of
664/// housekeeping: one directory git or the filesystem refuses to give up is a
665/// `tracing::warn`, not a reason to abandon the rest of the pass.
666pub async fn fold_orphaned_worktrees(
667    runs: &Path,
668    worktrees_root: &Path,
669    home: &Path,
670    grace_secs: u64,
671    now: Timestamp,
672) -> usize {
673    let known: std::collections::HashSet<String> = std::fs::read_dir(runs)
674        .into_iter()
675        .flatten()
676        .flatten()
677        .map(|e| e.file_name().to_string_lossy().into_owned())
678        .filter(|name| crate::run::is_run_id(name))
679        .map(|id| short_of(&id).to_owned())
680        .collect();
681
682    let mut folded = 0usize;
683    for entry in std::fs::read_dir(worktrees_root)
684        .into_iter()
685        .flatten()
686        .flatten()
687    {
688        if !entry.path().is_dir() {
689            continue;
690        }
691        let short = entry.file_name().to_string_lossy().into_owned();
692        if !looks_like_a_worktree_bay(&short) {
693            continue;
694        }
695        if known.contains(&short) || crate::daemon::is_working_on_short(home, &short, now) {
696            continue;
697        }
698        let wt = entry.path();
699        if !stale_enough(&wt, grace_secs, now) {
700            continue;
701        }
702        crate::git::remove_worktree_from_linked(&wt).await;
703        for e in std::fs::read_dir(&wt).into_iter().flatten().flatten() {
704            crate::git::remove_worktree_from_linked(&e.path()).await;
705        }
706        match std::fs::remove_dir_all(&wt) {
707            Ok(()) => folded += 1,
708            Err(e) => tracing::warn!(
709                "housekeep: remove orphaned worktree {}: {e:#}",
710                wt.display()
711            ),
712        }
713    }
714    folded
715}
716
717/// Does `name` have the shape a run's own worktree bay is named with: the
718/// same 4-character alphanumeric tag [`crate::run::is_run_id`] requires of a
719/// full id's trailing block (see [`short_of`])?
720///
721/// Anything else under `worktrees_root` is not a bay this function reclaims
722/// at all, claimed or not - answering "does a run claim this?" about a
723/// directory that was never a run's worktree in the first place is exactly
724/// the wrong question to ask before deleting it.
725fn looks_like_a_worktree_bay(name: &str) -> bool {
726    name.len() == 4 && name.bytes().all(|b| b.is_ascii_alphanumeric())
727}
728
729/// Is `dir`'s own modification time old enough, against `grace_secs` and
730/// `now`, that its emptiness of a run record can be trusted rather than
731/// caught mid-creation?
732///
733/// A directory this pass cannot stat at all - a race with its own removal, a
734/// permission error - is treated as not yet stale: unreadable metadata is not
735/// evidence of anything, and the janitor already keeps rather than deletes
736/// whenever it cannot tell (see the module docs).
737///
738/// `grace_secs` is floored at [`MIN_ORPHAN_AGE_SECS`] regardless of what the
739/// caller passes: `0` is a documented, legitimate value for
740/// [`crate::config::Disk::fold_grace_secs`] (`due`'s own "always due" case),
741/// because that grace answers a policy question the operator owns - how long
742/// a *known, finished* run's worktree lingers before cleanup. Whether an
743/// orphan worktree is actually a race with `Runner::review`'s `git worktree
744/// add` landing before its `run.json` is not a policy question, and must not
745/// collapse to zero just because the operator turned the other grace off -
746/// that would defeat the very check meant to catch it.
747fn stale_enough(dir: &Path, grace_secs: u64, now: Timestamp) -> bool {
748    let Ok(modified) = std::fs::metadata(dir).and_then(|m| m.modified()) else {
749        return false;
750    };
751    let Ok(ts) = Timestamp::try_from(modified) else {
752        return false;
753    };
754    due(now, ts, grace_secs.max(MIN_ORPHAN_AGE_SECS))
755}
756
757/// The floor under [`stale_enough`]'s grace, independent of
758/// [`crate::config::Disk::fold_grace_secs`].
759///
760/// Ample next to the race it guards: the gap between `Runner::start` or
761/// `Runner::review` creating a worktree and the first `RunState::save`
762/// landing is a handful of `git` subprocess calls, not minutes - but the
763/// janitor cannot tell "still mid-setup" from "orphaned" by any other signal
764/// for a run that never registers with `daemon::Status` at all (a `magi
765/// review` invocation, for one), so this is generous on purpose rather than
766/// tuned to the observed case.
767const MIN_ORPHAN_AGE_SECS: u64 = 5 * 60;
768
769/// Resolve an id or prefix against an explicit runs directory, exactly the way
770/// [`crate::run::resolve_id`] does against the global home.
771fn resolve_id_path(runs: &Path, prefix: &str) -> Result<String> {
772    // Keyed on the directory, not on a readable state file: the record this
773    // route exists to remove may be a lone `run.json.tmp` from a save that
774    // ran out of disk, and that is precisely the one a human needs a way to
775    // clear (see `crate::run::list_ids`).
776    if runs.join(prefix).is_dir() && crate::run::is_run_id(prefix) {
777        return Ok(prefix.to_owned());
778    }
779    let mut hits: Vec<String> = Vec::new();
780    for e in std::fs::read_dir(runs).into_iter().flatten().flatten() {
781        if !e.path().is_dir() {
782            continue;
783        }
784        let id = e.file_name().to_string_lossy().into_owned();
785        if crate::run::is_run_id(&id) && (id.starts_with(prefix) || id.ends_with(prefix)) {
786            hits.push(id);
787        }
788    }
789    match hits.len() {
790        1 => Ok(hits.into_iter().next().expect("exactly one hit")),
791        0 => bail!("no run matches `{prefix}`"),
792        _ => bail!(
793            "`{prefix}` matches {} runs: {}",
794            hits.len(),
795            hits.join(", ")
796        ),
797    }
798}
799
800/// `magi fold`'s recovery path for a run whose worktrees are already gone —
801/// so [`crate::graph::fold_run`] removed nothing — but whose `run.json` still
802/// lists active seats nobody is left to answer for: no live daemon claims the
803/// run, and every one of those seats has overrun its own timeout (see
804/// [`RunState::active_all_overrun`]). Clearing them and failing the run is
805/// what lets it be deleted afterward — [`RunState::ensure_can_delete`] only
806/// ever checks whether a live daemon is working on the run and whether its
807/// candidates are folded, not `status`, but a run stuck `implementing`
808/// forever with an empty worktree still reads as unresolved everywhere else
809/// (`magi show`, the deck, the phone) until this runs.
810///
811/// Returns `false` without changing anything when a live daemon still claims
812/// the run, or when some active seat has not actually overrun its budget yet
813/// — a run that is merely between waves must never be guessed at.
814pub fn clear_abandoned_active(state: &mut RunState, home: &Path, now: Timestamp) -> Result<bool> {
815    if crate::daemon::is_working_on(home, &state.id, now) || !state.active_all_overrun(now) {
816        return Ok(false);
817    }
818    state.abandon("fold");
819    state.save_under(home)?;
820    // The seat that asked is gone for good now - the same door
821    // `graph::Runner::settle_questions` closes the moment `status` lands
822    // somewhere non-resumable, see that method's own doc. Without this, an
823    // open question the abandoned seat left behind would keep badging the
824    // operator until the next daemon startup's `abandon_settled_questions`
825    // pass happened to notice it, or forever if nothing is running `magi
826    // serve` at all.
827    if let Err(e) = Questions::at(home.join("questions")).settle_run(&state.id, state.status) {
828        tracing::warn!("abandon questions for {}: {e:#}", state.id);
829    }
830    Ok(true)
831}
832
833/// Delete files from the shared build cache until it fits its cap — but only
834/// while nobody live is registered as using it. [`crate::cache::maintenance_prune`]
835/// takes out the same lease a build would, so a prune can never race a
836/// compile in flight (this run's own, another run's, or a human's `magi
837/// review`) into deleting a file that build still needs. `Ok(None)` when the
838/// cache is in use right now; the next pass catches it once the borrower
839/// releases it, the same way a cap of `0` or a missing `CARGO_TARGET_DIR`
840/// already meant "nothing to do this time" here.
841pub fn prune_cache(home: &Path, cache: &Path, limit_bytes: u64) -> Result<Option<Prune>> {
842    crate::cache::maintenance_prune(home, cache, limit_bytes)
843}
844
845/// [`prune_cache`], but resolving the operator's opt-out and missing
846/// `CARGO_TARGET_DIR` first — the same two checks [`housekeep`]'s idle pass
847/// makes before ever measuring the cache, factored out so
848/// [`crate::daemon`]'s between-runs check (see the module's own doc for why
849/// congestion can make "idle" arrive too rarely to matter) makes them
850/// identically rather than growing its own copy that could drift. `Ok(None)`
851/// covers a cap of `0` (see the module docs on `cache_limit_bytes`), a config
852/// that renders no `CARGO_TARGET_DIR` to aggregate at all, and a cache
853/// currently in use (see [`prune_cache`]).
854pub fn prune_cache_if_over_limit(
855    cfg: &crate::config::Config,
856    home: &Path,
857) -> Result<Option<Prune>> {
858    if cfg.disk.cache_limit_bytes == 0 {
859        return Ok(None);
860    }
861    let Some(cache) = cfg.cache_dir() else {
862        return Ok(None);
863    };
864    prune_cache(home, &cache, cfg.disk.cache_limit_bytes)
865}
866
867/// The cache's path, size and cap, for `magi cache show` and the health view.
868/// `None` when the config declares no `CARGO_TARGET_DIR` to aggregate.
869///
870/// A cap of `0` means the operator opted out of pruning; the size is then
871/// reported but never acted on.
872pub fn cache_report(cfg: &crate::config::Config) -> Option<(PathBuf, u64, u64)> {
873    let cache = cfg.cache_dir()?;
874    Some((
875        cache.clone(),
876        cache_size(&cache),
877        cfg.disk.cache_limit_bytes,
878    ))
879}
880
881/// Size in bytes of the shared build cache.
882pub fn cache_size(cache: &Path) -> u64 {
883    dir_size(cache)
884}
885
886#[cfg(test)]
887mod tests {
888
889    async fn sh(cwd: &Path, args: &[&str]) {
890        let out = crate::git::git_raw(cwd, args).await.expect("spawn git");
891        assert!(out.ok(), "git {args:?}: {}", out.stderr);
892    }
893
894    async fn commit(cwd: &Path, file: &str, body: &str) {
895        std::fs::write(cwd.join(file), body).unwrap();
896        sh(cwd, &["add", file]).await;
897        sh(
898            cwd,
899            &[
900                "-c",
901                "user.name=t",
902                "-c",
903                "user.email=t@example.com",
904                "commit",
905                "-q",
906                "-m",
907                body,
908            ],
909        )
910        .await;
911    }
912
913    const LONG: std::time::Duration = std::time::Duration::from_secs(30);
914
915    #[tokio::test]
916    async fn fetch_advances_origin_main_and_leaves_the_checkout_alone() {
917        let tmp = tempfile::tempdir().unwrap();
918        let t = tmp.path();
919        let bare = t.join("remote.git");
920        sh(
921            t,
922            &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
923        )
924        .await;
925        let dir = t.join("root/h/o/r");
926        std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
927        sh(
928            t,
929            &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
930        )
931        .await;
932        sh(&dir, &["checkout", "-q", "-b", "main"]).await;
933        commit(&dir, "a.txt", "one").await;
934        sh(&dir, &["push", "-q", "origin", "main"]).await;
935        sh(&dir, &["checkout", "-q", "--detach"]).await;
936        std::fs::write(dir.join("a.txt"), "dirty").unwrap();
937
938        let other = t.join("other");
939        sh(
940            t,
941            &[
942                "clone",
943                "-q",
944                bare.to_str().unwrap(),
945                other.to_str().unwrap(),
946            ],
947        )
948        .await;
949        commit(&other, "b.txt", "two").await;
950        sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
951
952        let rev = |d: PathBuf, r: &'static str| async move {
953            crate::git::git(&d, &["rev-parse", r]).await.unwrap()
954        };
955        let head = rev(dir.clone(), "HEAD").await;
956        let before = rev(dir.clone(), "origin/main").await;
957        let status = crate::git::git(&dir, &["status", "--porcelain"])
958            .await
959            .unwrap();
960
961        let r = fetch_origins(&[t.join("root")], LONG, || false).await;
962        assert_eq!(
963            r,
964            FetchReport {
965                fetched: 1,
966                no_origin: 0,
967                failed: 0
968            }
969        );
970
971        assert_ne!(rev(dir.clone(), "origin/main").await, before);
972        assert_eq!(rev(dir.clone(), "HEAD").await, head);
973        assert_eq!(
974            crate::git::git(&dir, &["status", "--porcelain"])
975                .await
976                .unwrap(),
977            status
978        );
979        assert_eq!(std::fs::read_to_string(dir.join("a.txt")).unwrap(), "dirty");
980    }
981
982    #[tokio::test]
983    async fn fetch_never_writes_a_local_branch_whatever_the_configured_refspec() {
984        let tmp = tempfile::tempdir().unwrap();
985        let t = tmp.path();
986        let bare = t.join("remote.git");
987        sh(
988            t,
989            &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
990        )
991        .await;
992        let dir = t.join("root/h/o/r");
993        std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
994        sh(
995            t,
996            &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
997        )
998        .await;
999        sh(&dir, &["checkout", "-q", "-b", "main"]).await;
1000        commit(&dir, "a.txt", "one").await;
1001        sh(&dir, &["push", "-q", "origin", "main"]).await;
1002        sh(&dir, &["checkout", "-q", "--detach"]).await;
1003        // A hostile mapping onto the local branch, and no origin/* at all.
1004        sh(
1005            &dir,
1006            &[
1007                "config",
1008                "remote.origin.fetch",
1009                "+refs/heads/main:refs/heads/main",
1010            ],
1011        )
1012        .await;
1013        let _ = crate::git::git_raw(&dir, &["update-ref", "-d", "refs/remotes/origin/main"]).await;
1014
1015        let other = t.join("other");
1016        sh(
1017            t,
1018            &[
1019                "clone",
1020                "-q",
1021                bare.to_str().unwrap(),
1022                other.to_str().unwrap(),
1023            ],
1024        )
1025        .await;
1026        commit(&other, "b.txt", "two").await;
1027        sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
1028        let remote_tip = crate::git::git(&other, &["rev-parse", "HEAD"])
1029            .await
1030            .unwrap();
1031        let local_main = crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1032            .await
1033            .unwrap();
1034
1035        let r = fetch_origins(&[t.join("root")], LONG, || false).await;
1036        assert_eq!(r.fetched, 1);
1037        assert_eq!(
1038            crate::git::git(&dir, &["rev-parse", "refs/remotes/origin/main"])
1039                .await
1040                .unwrap(),
1041            remote_tip
1042        );
1043        assert_eq!(
1044            crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1045                .await
1046                .unwrap(),
1047            local_main
1048        );
1049    }
1050
1051    #[tokio::test]
1052    async fn fetch_skips_no_origin_and_survives_an_unreachable_origin() {
1053        let tmp = tempfile::tempdir().unwrap();
1054        let t = tmp.path();
1055        let root = t.join("root");
1056        let bare = t.join("remote.git");
1057        sh(t, &["init", "-q", "--bare", bare.to_str().unwrap()]).await;
1058
1059        let none = root.join("h/o/none");
1060        let dead = root.join("h/o/dead");
1061        let good = root.join("h/o/good");
1062        for d in [&none, &dead, &good] {
1063            std::fs::create_dir_all(d).unwrap();
1064            sh(d, &["init", "-q"]).await;
1065        }
1066        sh(
1067            &dead,
1068            &[
1069                "remote",
1070                "add",
1071                "origin",
1072                t.join("missing").to_str().unwrap(),
1073            ],
1074        )
1075        .await;
1076        sh(&good, &["remote", "add", "origin", bare.to_str().unwrap()]).await;
1077
1078        let r = fetch_origins(&[root], LONG, || false).await;
1079        assert_eq!(
1080            r,
1081            FetchReport {
1082                fetched: 1,
1083                no_origin: 1,
1084                failed: 1
1085            }
1086        );
1087    }
1088    use super::*;
1089    use crate::config::Disk;
1090    use std::fs;
1091
1092    fn ts(s: &str) -> Timestamp {
1093        s.parse().expect("rfc3339")
1094    }
1095
1096    fn block_on<F: std::future::Future>(f: F) -> F::Output {
1097        tokio::runtime::Runtime::new().expect("runtime").block_on(f)
1098    }
1099
1100    #[test]
1101    fn a_run_is_due_after_its_grace_and_not_before() {
1102        let now = ts("2026-09-05T00:00:00Z");
1103        let grace = 600;
1104        let old = now - SignedDuration::new(601, 0);
1105        let fresh = now - SignedDuration::new(599, 0);
1106        assert!(due(now, old, grace));
1107        assert!(!due(now, fresh, grace));
1108        // Exactly at the edge: not yet due.
1109        let edge = now - SignedDuration::new(600, 0);
1110        assert!(!due(now, edge, grace));
1111        // A zero grace folds everything, ever.
1112        assert!(due(now, old, 0));
1113    }
1114
1115    #[test]
1116    fn rotate_from_moves_the_starting_point_as_the_cursor_advances() {
1117        let ids: Vec<String> = ["a", "b", "c", "d", "e"]
1118            .iter()
1119            .map(|s| s.to_string())
1120            .collect();
1121
1122        // A cursor of 0 starts at the front, same as an unrotated list.
1123        assert_eq!(rotate_from(&ids, 0), ids);
1124
1125        // A cursor of 2 starts two ids further along, with the skipped
1126        // front wrapping to the tail rather than being dropped.
1127        let at_2 = rotate_from(&ids, 2);
1128        assert_eq!(at_2, vec!["c", "d", "e", "a", "b"]);
1129        for id in &ids {
1130            assert!(at_2.contains(id));
1131        }
1132
1133        // A cursor past the list's own length wraps rather than panicking -
1134        // the caller never has to reduce it modulo anything itself.
1135        assert_eq!(rotate_from(&ids, 7), rotate_from(&ids, 2));
1136
1137        // An empty list has no start point to compute and must not panic.
1138        assert_eq!(rotate_from(&[], 0), Vec::<String>::new());
1139    }
1140
1141    /// A fleet with more eligible runs than one pass can check must not park
1142    /// the same handful at the front of the window forever: advancing the
1143    /// cursor by exactly how many ids one pass actually looked at - the
1144    /// arithmetic [`reconcile_external_merges`] itself does - walks every id
1145    /// to the front within one full rotation, with no reliance on how often
1146    /// or how regularly passes happen to run.
1147    #[test]
1148    fn advancing_the_cursor_by_the_window_size_gives_every_id_a_turn() {
1149        let ids: Vec<String> = (0..12).map(|n| format!("run-{n}")).collect();
1150        let cap = 5usize;
1151        let mut cursor = 0usize;
1152        let mut ever_seen: std::collections::HashSet<String> = std::collections::HashSet::new();
1153        for _ in 0..ids.len() {
1154            let window: Vec<String> = rotate_from(&ids, cursor).into_iter().take(cap).collect();
1155            ever_seen.extend(window.iter().cloned());
1156            cursor = (cursor + window.len()) % ids.len();
1157        }
1158        assert_eq!(
1159            ever_seen.len(),
1160            ids.len(),
1161            "every id must be checked at least once across a full rotation, whatever \
1162             the cadence between passes"
1163        );
1164    }
1165
1166    #[test]
1167    fn the_external_merge_cursor_round_trips_through_a_files_absence() {
1168        let dir = tempfile::tempdir().unwrap();
1169        let home = dir.path();
1170
1171        // Nothing written yet: starts at 0, not an error.
1172        assert_eq!(read_external_merge_cursor(home), 0);
1173
1174        write_external_merge_cursor(home, 7);
1175        assert_eq!(read_external_merge_cursor(home), 7);
1176
1177        // Anything unreadable as a plain number falls back to 0 rather than
1178        // wedging the sweep on a corrupt file.
1179        std::fs::write(home.join(EXTERNAL_MERGE_CURSOR_FILE), "not a number").unwrap();
1180        assert_eq!(read_external_merge_cursor(home), 0);
1181    }
1182
1183    #[test]
1184    fn a_run_qualifies_for_an_external_merge_check_only_when_blocked_unmerged_and_due() {
1185        let now = ts("2026-09-05T00:00:00Z");
1186        let grace = 600;
1187        let old = now - SignedDuration::new(601, 0);
1188        let fresh = now - SignedDuration::new(599, 0);
1189
1190        assert!(
1191            eligible_for_external_merge_check(RunStatus::Blocked, false, old, now, grace),
1192            "blocked, unmerged, and past its grace period is exactly the run this exists for"
1193        );
1194        assert!(
1195            !eligible_for_external_merge_check(RunStatus::Blocked, false, fresh, now, grace),
1196            "an operator mid-fix deserves the same grace window `fold_due` gives before \
1197             the janitor starts asking GitHub about it"
1198        );
1199        assert!(
1200            !eligible_for_external_merge_check(RunStatus::Blocked, true, old, now, grace),
1201            "a run `land::land` already recorded a merge outcome for has its own answer \
1202             already; this check is only for a run with nothing recorded at all"
1203        );
1204        assert!(
1205            !eligible_for_external_merge_check(RunStatus::Stalled, false, old, now, grace),
1206            "stalled is not blocked - it means the tally never reached quorum, which a \
1207             pull request cannot fix"
1208        );
1209        assert!(
1210            !eligible_for_external_merge_check(RunStatus::Ready, false, old, now, grace),
1211            "ready has nothing to correct - it was never landed by design"
1212        );
1213        assert!(
1214            !eligible_for_external_merge_check(RunStatus::Superseded, false, old, now, grace),
1215            "a superseded run's task already has its answer from a later attempt; \
1216             there is nothing left for GitHub to confirm here"
1217        );
1218    }
1219
1220    /// A `Blocked` run with a decided winner, so `land::find_external_merge`
1221    /// has a branch to ask `gh` about instead of returning early for lack of
1222    /// one - which is what makes an unreachable `/nonexistent/repo` fail
1223    /// loudly (an `Err`, raising a notice) rather than silently (an early
1224    /// `Ok(None)`, raising nothing) when this run's sweep is exercised.
1225    fn write_blocked_run(runs: &Path, id: &str, updated_at: Timestamp) {
1226        use crate::run::{Candidate, Tally};
1227        use std::collections::BTreeMap;
1228
1229        let mut state = RunState::new(
1230            PathBuf::from("/nonexistent/repo"),
1231            "main".to_owned(),
1232            "0000000000000000000000000000000000000000".to_owned(),
1233            String::new(),
1234            crate::config::Config::default(),
1235        );
1236        state.id = id.to_owned();
1237        state.status = RunStatus::Blocked;
1238        state.updated_at = updated_at;
1239        state.candidates.push(Candidate {
1240            index: 0,
1241            label: 'A',
1242            agent: "agent".to_owned(),
1243            branch: format!("magi/{id}/A"),
1244            worktree: PathBuf::from("/nonexistent/repo"),
1245            summary: String::new(),
1246            stat: String::new(),
1247            files: 0,
1248            commits: 0,
1249            empty: false,
1250            failed: None,
1251            verified_noop: None,
1252            duration_ms: 0,
1253            folded: false,
1254        });
1255        state.tally = Some(Tally {
1256            first_choice: BTreeMap::from([('A', 1)]),
1257            borda: BTreeMap::new(),
1258            winner: 'A',
1259            rankings: 1,
1260            unanimous_initial: true,
1261            deliberated: false,
1262            changed_votes: 0,
1263            unanimous_final: true,
1264            tie_break: None,
1265            judges: 1,
1266            present: 1,
1267            quorum: 1,
1268            met_quorum: true,
1269            uncontested: None,
1270        });
1271        std::fs::create_dir_all(runs.join(id)).unwrap();
1272        std::fs::write(
1273            runs.join(id).join("run.json"),
1274            serde_json::to_string_pretty(&state).unwrap(),
1275        )
1276        .unwrap();
1277    }
1278
1279    /// No run here can be confirmed merged - one has no repository `gh` can
1280    /// even ask about, one has not sat blocked long enough, and one is
1281    /// claimed by a live daemon - so the sweep must leave every one of them
1282    /// exactly as it found them and never panic on the failures in between.
1283    #[tokio::test]
1284    async fn reconcile_external_merges_leaves_ineligible_and_unconfirmable_runs_alone() {
1285        let dir = tempfile::tempdir().unwrap();
1286        let runs = dir.path().join("runs");
1287        let home = dir.path().to_path_buf();
1288        std::fs::create_dir_all(&runs).unwrap();
1289
1290        let now = ts("2026-09-05T00:00:00Z");
1291        let disk = Disk {
1292            fold_grace_secs: 600,
1293            ..Disk::default()
1294        };
1295
1296        let due_id = "20260905-000000-blkd";
1297        write_blocked_run(&runs, due_id, ts("2026-08-01T00:00:00Z"));
1298
1299        let fresh_id = "20260905-000000-fres";
1300        write_blocked_run(&runs, fresh_id, now);
1301
1302        let live_id = "20260905-000000-live";
1303        write_blocked_run(&runs, live_id, ts("2026-08-01T00:00:00Z"));
1304        let mut status = crate::daemon::Status::new();
1305        status.current = vec![crate::daemon::Current {
1306            task: "20260905-000000-task".to_owned(),
1307            run: live_id.to_owned(),
1308        }];
1309        status.updated_at = now;
1310        crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1311
1312        let reconciled = reconcile_external_merges(&runs, &home, &disk, now).await;
1313        assert_eq!(
1314            reconciled, 0,
1315            "an unreachable repository can never be confirmed merged"
1316        );
1317        for id in [due_id, fresh_id, live_id] {
1318            assert_eq!(
1319                read_meta(&runs, id).unwrap().status,
1320                RunStatus::Blocked,
1321                "{id} must be left exactly as it was found"
1322            );
1323        }
1324        let notices = crate::notices::Notices::at(home.join("notifications")).list();
1325        let due_notice = notices
1326            .iter()
1327            .find(|n| n.key == format!("merged-unrecorded:{due_id}"))
1328            .expect("notice for due_id must be raised");
1329        assert_eq!(
1330            due_notice.link,
1331            Some(crate::notices::Link::Run {
1332                id: due_id.to_owned(),
1333            })
1334        );
1335    }
1336
1337    /// The bug this exists to pin: a first version rotated on
1338    /// `now.as_second() % len`, which is fully determined by wall-clock time
1339    /// and nothing else. Two passes seconds apart - as they would be when a
1340    /// short poll interval and a fast pass alias onto the same handful of
1341    /// `now.as_second()` residues - landed on the *same* starting point and
1342    /// never covered a fleet bigger than the per-pass cap, however many
1343    /// times housekeeping ran. Persisting a cursor makes forward progress a
1344    /// function of how many ids got looked at, not of when the clock reads
1345    /// this pass happened to run.
1346    #[tokio::test]
1347    async fn reconcile_external_merges_covers_a_larger_fleet_across_repeated_passes_at_one_instant()
1348    {
1349        let dir = tempfile::tempdir().unwrap();
1350        let runs = dir.path().join("runs");
1351        let home = dir.path().to_path_buf();
1352        std::fs::create_dir_all(&runs).unwrap();
1353
1354        let now = ts("2026-09-05T00:00:00Z");
1355        let disk = Disk {
1356            fold_grace_secs: 600,
1357            ..Disk::default()
1358        };
1359        let old = ts("2026-08-01T00:00:00Z");
1360
1361        let ids: Vec<String> = (0..8).map(|n| format!("20260905-000000-r{n:03}")).collect();
1362        for id in &ids {
1363            write_blocked_run(&runs, id, old);
1364        }
1365
1366        let notices = crate::notices::Notices::at(home.join("notifications"));
1367
1368        // Two passes at the exact same `now`, exactly what a fast pass on a
1369        // short poll interval looks like. Each of the 8 unreachable-repo
1370        // runs raises its own notice the first time it is looked at, so the
1371        // count of distinct notices is a direct readout of how many distinct
1372        // ids have been checked so far.
1373        reconcile_external_merges(&runs, &home, &disk, now).await;
1374        let after_first = notices.list().len();
1375        assert_eq!(
1376            after_first, MAX_EXTERNAL_MERGE_CHECKS_PER_PASS,
1377            "the first pass checks exactly one cap's worth"
1378        );
1379
1380        reconcile_external_merges(&runs, &home, &disk, now).await;
1381        let after_second = notices.list().len();
1382        assert_eq!(
1383            after_second,
1384            ids.len(),
1385            "a second pass at the same instant must still reach every id the \
1386             first pass had no room for, not repeat the same cap's worth"
1387        );
1388    }
1389
1390    #[test]
1391    fn the_meta_reader_is_tolerant_of_everything_except_the_deciders() {
1392        let dir = tempfile::tempdir().unwrap();
1393        let runs = dir.path().join("runs");
1394        let id = "20260905-000000-abcd";
1395        std::fs::create_dir_all(runs.join(id)).unwrap();
1396        std::fs::write(
1397            runs.join(id).join("run.json"),
1398            r#"{"schema": 99, "id": "20260905-000000-abcd", "updated_at": "2026-09-05T00:00:00Z", "status": "ready", "junk_from_another_build": [1, 2, 3]}"#,
1399        )
1400        .unwrap();
1401        let meta = read_meta(&runs, id).expect("readable");
1402        assert_eq!(meta.status, RunStatus::Ready);
1403        assert_eq!(meta.updated_at, ts("2026-09-05T00:00:00Z"));
1404        assert!(read_meta(&runs, "nope").is_err(), "missing file unreadable");
1405        std::fs::write(runs.join(id).join("run.json"), "not json at all").unwrap();
1406        assert!(read_meta(&runs, id).is_err(), "garbage unreadable");
1407    }
1408
1409    #[test]
1410    fn fold_unreadable_releases_run_dir_and_worktrees() {
1411        let dir = tempfile::tempdir().unwrap();
1412        let runs = dir.path().join("runs");
1413        let wt = dir.path().join("wt");
1414        let id = "20260905-000000-abcd";
1415        std::fs::create_dir_all(runs.join(id)).unwrap();
1416        std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1417        std::fs::create_dir_all(wt.join("abcd")).unwrap();
1418        std::fs::write(wt.join("abcd").join("leftover"), b"x").unwrap();
1419
1420        let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold");
1421        assert_eq!(removed.len(), 2);
1422        assert!(!runs.join(id).exists(), "run dir gone");
1423        assert!(!wt.join("abcd").exists(), "worktrees gone");
1424
1425        // A prefix resolves like `run::resolve_id` does.
1426        std::fs::create_dir_all(runs.join(id)).unwrap();
1427        std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1428        std::fs::create_dir_all(wt.join("abcd")).unwrap();
1429        std::fs::write(wt.join("abcd").join("leftover"), b"x").unwrap();
1430        let removed = block_on(fold_unreadable(&runs, &wt, "20260905")).expect("by prefix");
1431        assert_eq!(removed.len(), 2);
1432        // Once gone, `id` cannot be resolved at all - same as `run::resolve_id`
1433        // on an id nothing on disk matches - so a repeat pass errors rather
1434        // than silently reporting nothing removed.
1435        assert!(
1436            block_on(fold_unreadable(&runs, &wt, id)).is_err(),
1437            "a run already gone cannot be resolved again"
1438        );
1439    }
1440
1441    #[test]
1442    fn prune_cache_sheds_the_oldest_generation_until_it_fits() {
1443        let home = tempfile::tempdir().unwrap();
1444        let dir = tempfile::tempdir().unwrap();
1445        // Same size, different age: only the age decides, and the newest
1446        // generation - the one the next build reuses - is what survives.
1447        fs::write(dir.path().join("old"), b"xx").unwrap();
1448        fs::write(dir.path().join("new"), b"yy").unwrap();
1449        touch(&dir.path().join("old"), 1_000_000);
1450        touch(&dir.path().join("new"), 2_000_000);
1451
1452        let out = prune_cache(home.path(), dir.path(), 2)
1453            .expect("prune")
1454            .expect("the cache is free");
1455        assert_eq!(out.files, 1, "one deletion is enough to reach the cap");
1456        assert_eq!(out.remaining, 2);
1457        assert!(!dir.path().join("old").exists(), "the older file went");
1458        assert!(dir.path().join("new").exists(), "the newer one stayed");
1459
1460        // A whole generation shares one timestamp tick, so the tie has to be
1461        // decided too: largest first, which reaches the cap in the fewest
1462        // deletions. Left to `read_dir` and an unstable sort this deleted
1463        // both files on Linux and one on Windows.
1464        let tied = tempfile::tempdir().unwrap();
1465        fs::write(tied.path().join("big"), b"xxxx").unwrap();
1466        fs::write(tied.path().join("small"), b"yy").unwrap();
1467        touch(&tied.path().join("big"), 1_000_000);
1468        touch(&tied.path().join("small"), 1_000_000);
1469        let out = prune_cache(home.path(), tied.path(), 2)
1470            .expect("prune")
1471            .expect("the cache is free");
1472        assert_eq!(out.files, 1, "the big one alone gets under the cap");
1473        assert_eq!(out.remaining, 2);
1474        assert!(tied.path().join("small").exists());
1475    }
1476
1477    #[test]
1478    fn prune_cache_if_over_limit_resolves_the_opt_outs_before_ever_measuring() {
1479        let home = tempfile::tempdir().unwrap();
1480        let dir = tempfile::tempdir().unwrap();
1481        fs::write(dir.path().join("big"), vec![0u8; 10]).unwrap();
1482
1483        let mut cfg = crate::config::Config::default();
1484        cfg.verify.gate = vec![format!(
1485            "CARGO_TARGET_DIR={} cargo make check",
1486            dir.path().display()
1487        )];
1488
1489        // A cap of `0` is the operator's opt-out: never measured, never
1490        // pruned, regardless of what is actually on disk.
1491        cfg.disk.cache_limit_bytes = 0;
1492        assert_eq!(
1493            prune_cache_if_over_limit(&cfg, home.path()).unwrap(),
1494            None,
1495            "a zero cap must not even look at the directory"
1496        );
1497        assert!(dir.path().join("big").exists());
1498
1499        // No `CARGO_TARGET_DIR` in either verify command: nothing to
1500        // aggregate, so there is nothing to prune either.
1501        let mut no_cache = crate::config::Config::default();
1502        no_cache.disk.cache_limit_bytes = 1;
1503        assert_eq!(
1504            prune_cache_if_over_limit(&no_cache, home.path()).unwrap(),
1505            None
1506        );
1507
1508        // Over the cap and configured: pruned exactly like `prune_cache`
1509        // itself would.
1510        cfg.disk.cache_limit_bytes = 1;
1511        let pruned = prune_cache_if_over_limit(&cfg, home.path())
1512            .unwrap()
1513            .expect("a real cache dir over its cap prunes");
1514        assert_eq!(pruned.files, 1);
1515        assert!(!dir.path().join("big").exists());
1516    }
1517
1518    /// Pin a file's mtime, so a test asserts the policy and not the runner's
1519    /// timestamp granularity.
1520    fn touch(path: &Path, secs: u64) {
1521        let f = fs::File::options().write(true).open(path).unwrap();
1522        f.set_times(fs::FileTimes::new().set_modified(
1523            std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs),
1524        ))
1525        .unwrap();
1526    }
1527
1528    /// The disk-full casualty: a run whose first save left `run.json.tmp` and
1529    /// nothing else. It has to be clearable, or the record is permanent.
1530    #[test]
1531    fn fold_unreadable_clears_a_run_whose_state_never_landed() {
1532        let dir = tempfile::tempdir().unwrap();
1533        let runs = dir.path().join("runs");
1534        let wt = dir.path().join("wt");
1535        let id = "20260904-014540-88c0";
1536        std::fs::create_dir_all(runs.join(id)).unwrap();
1537        std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1538
1539        let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold by id");
1540        assert_eq!(removed, vec![format!("runs/{id}")]);
1541        assert!(!runs.join(id).exists(), "record gone");
1542
1543        // And by prefix, the way the deck and the phone address a run.
1544        std::fs::create_dir_all(runs.join(id)).unwrap();
1545        std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1546        assert!(
1547            block_on(fold_unreadable(&runs, &wt, "88c0")).is_ok(),
1548            "by prefix"
1549        );
1550
1551        // A directory under `runs` that is not a run is never a fold target.
1552        std::fs::create_dir_all(runs.join("scratch")).unwrap();
1553        assert!(
1554            block_on(fold_unreadable(&runs, &wt, "scratch")).is_err(),
1555            "a stray directory is not a run"
1556        );
1557    }
1558
1559    #[test]
1560    fn fold_due_folds_terminal_runs_of_any_schema_but_leaves_genuinely_unreadable_ones() {
1561        let dir = tempfile::tempdir().unwrap();
1562        let runs = dir.path().join("runs");
1563        let wt = dir.path().join("wt");
1564        let home = dir.path().to_path_buf();
1565        let disk = Disk::default();
1566        let now = ts("2026-09-05T00:00:00Z");
1567
1568        // 1. Runnable (judging): never folded, however old.
1569        let judging = "20260801-000000-0001";
1570        write_meta(&runs, judging, "judging", "2026-08-01T00:00:00Z");
1571
1572        // 2. Finished but fresh: grace not elapsed. Within the default 6h
1573        //    grace of `now`, so `fold_due` must stop at the freshness check
1574        //    and never even reach `read_state` - `write_meta`'s minimal JSON
1575        //    would fail that full parse anyway, and this case exists to
1576        //    prove freshness is why the run survives, not an accident of the
1577        //    fixture being unparseable as a whole `RunState`.
1578        let ready_fresh = "20260904-220000-0002";
1579        write_meta(&runs, ready_fresh, "ready", "2026-09-04T22:00:00Z");
1580
1581        // 3. Genuinely unreadable: broken JSON, not merely an unfamiliar
1582        //    schema number. Left alone and counted - this is the one case
1583        //    automatic housekeeping must never touch (see `fold_due`'s docs);
1584        //    discarding it is an explicit operator action, not something a
1585        //    background pass does.
1586        let garbage = "20260901-000000-0004";
1587        std::fs::create_dir_all(runs.join(garbage)).unwrap();
1588        std::fs::write(runs.join(garbage).join("run.json"), "not json").unwrap();
1589        std::fs::create_dir_all(wt.join("0004")).unwrap();
1590
1591        // 4. Finished, well past grace, current schema: the ordinary case
1592        //    `fold_due` has always acted on.
1593        let due_ready = due_run(&runs, &wt, "20260801-000000-ffff", SCHEMA);
1594
1595        // 5. Finished, well past grace, but written by a schema number this
1596        //    build no longer matches - the defect this task exists to fix.
1597        //    It still parses cleanly, so only the version number differs, and
1598        //    that alone must not block folding.
1599        let due_old_schema = due_run(&runs, &wt, "20260801-000000-eeee", SCHEMA - 1);
1600
1601        let (folded, unreadable) =
1602            block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
1603        assert_eq!(
1604            folded, 2,
1605            "both due, parseable runs fold regardless of their schema number"
1606        );
1607        assert_eq!(
1608            unreadable, 1,
1609            "only the run with broken JSON counts as unreadable"
1610        );
1611        assert!(runs.join(judging).exists(), "runnable never folded");
1612        assert!(runs.join(ready_fresh).exists(), "fresh never folded");
1613        assert!(runs.join(garbage).exists(), "unreadable record kept");
1614        assert!(wt.join("0004").exists(), "unreadable worktree kept");
1615        assert!(
1616            runs.join(&due_ready).exists(),
1617            "folding drops worktrees, not the record"
1618        );
1619        assert!(
1620            runs.join(&due_old_schema).exists(),
1621            "an old-schema record survives its fold exactly like a current one"
1622        );
1623        // `graph::fold_run` saves the state it just folded back to disk. That
1624        // write must land under this test's own `runs` - the argument it
1625        // passed to `fold_due`, not the process-global `run::home` - or a
1626        // fold that landed somewhere else entirely would still be counted
1627        // above as one of the two `folded` runs.
1628        for id in [&due_ready, &due_old_schema] {
1629            let saved = read_meta(&runs, id).expect("folded run still parses");
1630            assert_ne!(
1631                saved.updated_at,
1632                ts("2026-08-01T00:00:00Z"),
1633                "fold_run must have saved the updated state back through the \
1634                 `runs` directory this test passed to fold_due"
1635            );
1636        }
1637    }
1638
1639    /// `[disk] auto_fold = false` must leave the janitor's fold-and-reclaim
1640    /// passes completely inert - a due run's worktree and record both
1641    /// survive exactly as if `housekeep` had never run at all. Cache pruning
1642    /// is a separate opt-out (`cache_limit_bytes`) and stays disabled here
1643    /// too, so this test is only ever about `auto_fold`.
1644    #[tokio::test]
1645    async fn housekeep_leaves_everything_alone_when_auto_fold_is_disabled() {
1646        let dir = tempfile::tempdir().unwrap();
1647        let runs = dir.path().join("runs");
1648        let wt = dir.path().join("wt");
1649        let home = dir.path().to_path_buf();
1650        crate::run::set_home(dir.path().to_path_buf());
1651
1652        let due_id = due_run(&runs, &wt, "20260801-000000-abcd", SCHEMA);
1653        std::fs::create_dir_all(wt.join("orphan").join("cand-A")).unwrap();
1654
1655        let mut cfg = crate::config::Config::default();
1656        cfg.disk.auto_fold = false;
1657        cfg.disk.cache_limit_bytes = 0;
1658
1659        let out = housekeep(&cfg, &home, &wt, &dir.path().join("repo"), Timestamp::now()).await;
1660
1661        assert_eq!(out.folded, 0);
1662        assert_eq!(out.unreadable, 0);
1663        assert_eq!(out.orphaned_worktrees, 0);
1664        assert!(
1665            runs.join(&due_id).exists(),
1666            "a due run's record survives untouched"
1667        );
1668        assert!(
1669            wt.join("orphan").exists(),
1670            "an orphaned worktree survives untouched: the reclaim pass never ran"
1671        );
1672    }
1673
1674    #[test]
1675    fn fold_orphaned_worktrees_removes_only_worktrees_no_run_claims_and_none_in_flight() {
1676        let dir = tempfile::tempdir().unwrap();
1677        let runs = dir.path().join("runs");
1678        let wt = dir.path().join("wt");
1679        let home = dir.path().to_path_buf();
1680
1681        // A run record exists for this one: its worktree is claimed, not
1682        // orphaned, however old the record.
1683        write_meta(
1684            &runs,
1685            "20260801-000000-aaaa",
1686            "ready",
1687            "2026-08-01T00:00:00Z",
1688        );
1689        std::fs::create_dir_all(wt.join("aaaa").join("cand-A")).unwrap();
1690
1691        // No run record at all, and nobody is working on it: this is the
1692        // leftover `fold_due` can never see, because it only ever walks
1693        // `runs/`.
1694        std::fs::create_dir_all(wt.join("bbbb").join("cand-A")).unwrap();
1695
1696        // No run record either, but a live daemon status names a run with
1697        // this short id - the save-timing gap between the daemon claiming a
1698        // task and `RunState::new` writing its first `run.json`. Must survive
1699        // untouched.
1700        std::fs::create_dir_all(wt.join("cccc")).unwrap();
1701
1702        // Not shaped like a run's short id at all - a scratch directory an
1703        // operator or another tool left in the same bay - so it is never a
1704        // reclaim target regardless of what runs claim it or not.
1705        std::fs::create_dir_all(wt.join("scratch")).unwrap();
1706
1707        // `now` pushed comfortably past `MIN_ORPHAN_AGE_SECS`, so a zero
1708        // grace - the same "always due" escape hatch `due` itself documents
1709        // - still reclaims once a worktree is genuinely old, without faking
1710        // an mtime: real directory creation just above is already in the
1711        // past relative to this `now`, by design rather than by timing.
1712        let now = Timestamp::now() + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1713        let mut status = crate::daemon::Status::new();
1714        status.current = vec![crate::daemon::Current {
1715            task: "20260905-000000-t111".to_owned(),
1716            run: "20260905-000000-cccc".to_owned(),
1717        }];
1718        status.updated_at = now;
1719        crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1720
1721        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1722        assert_eq!(
1723            folded, 1,
1724            "only the truly orphaned, idle, bay-shaped worktree is removed"
1725        );
1726        assert!(wt.join("aaaa").exists(), "claimed by a run record");
1727        assert!(!wt.join("bbbb").exists(), "orphaned and idle: reclaimed");
1728        assert!(wt.join("cccc").exists(), "a run in flight is never touched");
1729        assert!(
1730            wt.join("scratch").exists(),
1731            "not shaped like a worktree bay, so never a reclaim target"
1732        );
1733    }
1734
1735    /// The gap this closes: `Runner::review` (`magi review`) creates the
1736    /// worktree with `git worktree add` before `RunState::save` ever writes a
1737    /// `run.json`, and that path never runs through the daemon's own `poll`
1738    /// loop at all, so `daemon::Status` never names it either. Without a
1739    /// grace window, a janitor pass landing in that gap would read the
1740    /// worktree as an orphan nothing is waiting on and delete a review still
1741    /// being set up.
1742    #[test]
1743    fn fold_orphaned_worktrees_leaves_a_freshly_created_bay_alone() {
1744        let dir = tempfile::tempdir().unwrap();
1745        let runs = dir.path().join("runs");
1746        let wt = dir.path().join("wt");
1747        let home = dir.path().to_path_buf();
1748
1749        std::fs::create_dir_all(wt.join("dddd").join("under-review")).unwrap();
1750
1751        let now = Timestamp::now();
1752        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 6 * 60 * 60, now));
1753        assert_eq!(
1754            folded, 0,
1755            "too fresh to tell apart from a run still being set up"
1756        );
1757        assert!(wt.join("dddd").exists());
1758    }
1759
1760    /// A fresh open question on `run`, stored and handed back for assertions.
1761    fn open_question(store: &Questions, run: &str) -> crate::ask::Question {
1762        let mut q = crate::ask::Question::new(
1763            run.to_owned(),
1764            "implement".to_owned(),
1765            "impl-A".to_owned(),
1766            "Which storage backend should the cache use?".to_owned(),
1767            String::new(),
1768            vec!["SQLite".to_owned(), "Redis".to_owned()],
1769        );
1770        store.put(&mut q).unwrap();
1771        q
1772    }
1773
1774    /// The exact ghost the phone showed: a run that already finished, with a
1775    /// question its dead seat asked still sitting `open` because it reached
1776    /// that status before `graph::Runner::settle_questions` existed (or
1777    /// missed it in the crash window `daemon::reclaim_orphaned_running`
1778    /// covers). This sweep is the second door to the same fact.
1779    #[test]
1780    fn a_finished_runs_open_question_is_swept_up() {
1781        let dir = tempfile::tempdir().unwrap();
1782        let runs = dir.path().join("runs");
1783        let store = Questions::at(dir.path().join("questions"));
1784
1785        let failed = "20260908-205802-c9eb";
1786        write_meta(&runs, failed, "failed", "2026-09-08T20:58:02Z");
1787        let failed_q = open_question(&store, failed);
1788
1789        let merged = "20260908-205501-ca67";
1790        write_meta(&runs, merged, "merged", "2026-09-08T20:55:01Z");
1791        let merged_q = open_question(&store, merged);
1792
1793        let n = abandon_settled_questions(&store, &runs);
1794        assert_eq!(n, 2, "both dead runs' questions are swept in one pass");
1795
1796        for (id, run) in [(&failed_q.id, failed), (&merged_q.id, merged)] {
1797            let back = store.get(id).unwrap();
1798            assert!(!back.status.open(), "{run} is done; nobody reads an answer");
1799            assert!(back.detail.contains(run), "{}", back.detail);
1800        }
1801    }
1802
1803    #[test]
1804    fn a_still_alive_runs_open_question_survives_the_sweep() {
1805        let dir = tempfile::tempdir().unwrap();
1806        let runs = dir.path().join("runs");
1807        let store = Questions::at(dir.path().join("questions"));
1808
1809        // `Blocked` and `Stalled` are `RunStatus::resumable`: the run can
1810        // still be picked back up, so its question may yet get a real
1811        // answer. A run still mid-competition is even more obviously alive.
1812        for (id, status) in [
1813            ("20260908-000000-b10c", "blocked"),
1814            ("20260908-000000-5ta1", "stalled"),
1815            ("20260908-000000-jud6", "judging"),
1816        ] {
1817            write_meta(&runs, id, status, "2026-09-08T00:00:00Z");
1818            let q = open_question(&store, id);
1819
1820            let n = abandon_settled_questions(&store, &runs);
1821            assert_eq!(n, 0, "{status} run is not done; nothing to sweep");
1822            assert!(
1823                store.get(&q.id).unwrap().status.open(),
1824                "{status} run's question must still be waiting"
1825            );
1826        }
1827    }
1828
1829    #[test]
1830    fn the_sweep_leaves_an_answered_question_and_an_unreadable_run_alone() {
1831        let dir = tempfile::tempdir().unwrap();
1832        let runs = dir.path().join("runs");
1833        let store = Questions::at(dir.path().join("questions"));
1834
1835        // Already decided: a sweep must never revisit it, whatever the run
1836        // that asked went on to become.
1837        let done = "20260908-000000-answ";
1838        write_meta(&runs, done, "failed", "2026-09-08T00:00:00Z");
1839        let mut answered = open_question(&store, done);
1840        answered
1841            .answer(crate::ask::Answer::Choice("SQLite".to_owned()))
1842            .unwrap();
1843        store.put(&mut answered).unwrap();
1844
1845        // No `run.json` at all for this one - deleted, or never landed.
1846        let gone = "20260908-000000-gone";
1847        let orphan = open_question(&store, gone);
1848
1849        assert_eq!(abandon_settled_questions(&store, &runs), 0);
1850        assert_eq!(
1851            store.get(&answered.id).unwrap().status,
1852            crate::ask::QuestionStatus::Answered,
1853            "a real answer is never overwritten by a sweep"
1854        );
1855        assert!(
1856            store.get(&orphan.id).unwrap().status.open(),
1857            "a run this sweep cannot read is left exactly as it was, not guessed at"
1858        );
1859    }
1860
1861    /// A grace of `0` is a legitimate, documented value for the operator's
1862    /// own `Disk::fold_grace_secs` - `due`'s "always due" case - but the
1863    /// freshness check this guards is not that policy, and must not collapse
1864    /// to it: a `0` handed straight through would reclaim a worktree the
1865    /// instant it exists, exactly the race `fold_orphaned_worktrees_leaves_a_
1866    /// freshly_created_bay_alone` exists to rule out, just with the operator
1867    /// having turned the other grace off instead of leaving it at its
1868    /// default.
1869    #[test]
1870    fn fold_orphaned_worktrees_floors_a_zero_grace_at_the_race_safe_minimum() {
1871        let dir = tempfile::tempdir().unwrap();
1872        let runs = dir.path().join("runs");
1873        let wt = dir.path().join("wt");
1874        let home = dir.path().to_path_buf();
1875
1876        std::fs::create_dir_all(wt.join("eeee").join("under-review")).unwrap();
1877
1878        // Too fresh, even with the grace argument at zero.
1879        let now = Timestamp::now();
1880        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1881        assert_eq!(
1882            folded, 0,
1883            "a zero grace must not defeat the race-safety floor"
1884        );
1885        assert!(wt.join("eeee").exists());
1886
1887        // Once genuinely past the floor, a zero grace reclaims it - the
1888        // floor is a minimum, not a replacement policy that never fires.
1889        let later = now + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1890        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, later));
1891        assert_eq!(folded, 1, "old enough now, regardless of the zero grace");
1892        assert!(!wt.join("eeee").exists());
1893    }
1894
1895    #[test]
1896    fn clear_abandoned_active_only_acts_once_dead_and_overrun() {
1897        let dir = tempfile::tempdir().unwrap();
1898        // Harmless if another test in this binary already pinned the global
1899        // home first (see `run::set_home`'s own doc): this test only checks
1900        // the in-memory mutation `clear_abandoned_active` makes, never a
1901        // write that landed under this exact directory.
1902        crate::run::set_home(dir.path().to_path_buf());
1903        let home = dir.path().to_path_buf();
1904        let now = ts("2026-09-14T12:00:00Z");
1905        let overrun_seat = || crate::run::ActiveSeat {
1906            node: "implement".to_owned(),
1907            started_at: now - SignedDuration::new(21_000, 0),
1908            timeout_secs: 3_600,
1909            attempt: 0,
1910            task: None,
1911            command: None,
1912            index: None,
1913            total: None,
1914        };
1915
1916        let mut state = RunState::new(
1917            PathBuf::from("/repo"),
1918            "main".to_owned(),
1919            "abc1234".to_owned(),
1920            "fixture".to_owned(),
1921            crate::config::Config::default(),
1922        );
1923        state.status = RunStatus::Implementing;
1924        state.active.insert("impl-A".to_owned(), overrun_seat());
1925
1926        // A seat still within its own budget: not provably dead yet, so this
1927        // must change nothing.
1928        let mut fresh = state.clone();
1929        fresh.active.insert(
1930            "impl-B".to_owned(),
1931            crate::run::ActiveSeat {
1932                node: "implement".to_owned(),
1933                started_at: now,
1934                timeout_secs: 3_600,
1935                attempt: 0,
1936                task: None,
1937                command: None,
1938                index: None,
1939                total: None,
1940            },
1941        );
1942        assert!(!clear_abandoned_active(&mut fresh, &home, now).unwrap());
1943        assert!(!fresh.active.is_empty());
1944        assert_eq!(fresh.status, RunStatus::Implementing);
1945
1946        let store = Questions::at(home.join("questions"));
1947        let q = open_question(&store, &state.id);
1948
1949        assert!(clear_abandoned_active(&mut state, &home, now).unwrap());
1950        assert!(state.active.is_empty());
1951        assert_eq!(state.status, RunStatus::Failed);
1952        assert!(
1953            !store.get(&q.id).unwrap().status.open(),
1954            "the abandoned seat's own open question must not keep badging the \
1955             operator until some later daemon startup notices it"
1956        );
1957    }
1958
1959    /// Write a whole `run.json` that magi can read, over the given state.
1960    fn write_meta(runs: &Path, id: &str, status: &str, updated_at: &str) {
1961        let day = &updated_at[..10];
1962        std::fs::create_dir_all(runs.join(id)).unwrap();
1963        let body = format!(
1964            r#"{{"schema": {SCHEMA}, "id": "{id}", "repo": "/nonexistent/repo", "base_branch": "main", "base_commit": "0000000000000000000000000000000000000000", "instruction": "", "created_at": "{day}T00:00:00Z", "updated_at": "{updated_at}", "status": "{status}", "seed": 1}}"#
1965        );
1966        std::fs::write(runs.join(id).join("run.json"), body).unwrap();
1967    }
1968
1969    /// Write a fully-formed, `Ready`, well-past-grace `run.json` tagged with
1970    /// an arbitrary schema number - so a test can write one this build's own
1971    /// `RunState::new` could never produce on its own. Returns the id.
1972    ///
1973    /// `wt` becomes this run's `graph.worktree_root`: left at the config
1974    /// default, `RunState::worktree_root` falls through to
1975    /// `run::default_worktree_root` - the operator's real `~/wt/magi` - and
1976    /// `graph::fold_run`'s second sweep would then `read_dir` and remove
1977    /// worktrees there instead of anything this test owns.
1978    fn due_run(runs: &Path, wt: &Path, id: &str, schema: u32) -> String {
1979        due_run_with_status(runs, wt, id, schema, RunStatus::Ready)
1980    }
1981
1982    /// [`due_run`], with the terminal status a caller wants instead of the
1983    /// `Ready` every other caller here happens to want.
1984    fn due_run_with_status(
1985        runs: &Path,
1986        wt: &Path,
1987        id: &str,
1988        schema: u32,
1989        status: RunStatus,
1990    ) -> String {
1991        let mut config = crate::config::Config::default();
1992        config.graph.worktree_root = Some(wt.to_path_buf());
1993        let mut state = RunState::new(
1994            PathBuf::from("/nonexistent/repo"),
1995            "main".to_owned(),
1996            "0000000000000000000000000000000000000000".to_owned(),
1997            String::new(),
1998            config,
1999        );
2000        state.id = id.to_owned();
2001        state.status = status;
2002        state.updated_at = ts("2026-08-01T00:00:00Z");
2003        let mut value = serde_json::to_value(&state).unwrap();
2004        value["schema"] = serde_json::json!(schema);
2005        std::fs::create_dir_all(runs.join(id)).unwrap();
2006        std::fs::write(
2007            runs.join(id).join("run.json"),
2008            serde_json::to_string_pretty(&value).unwrap(),
2009        )
2010        .unwrap();
2011        id.to_owned()
2012    }
2013
2014    #[test]
2015    fn fold_due_folds_a_superseded_run_same_as_any_other_terminal_one() {
2016        let dir = tempfile::tempdir().unwrap();
2017        let runs = dir.path().join("runs");
2018        let wt = dir.path().join("wt");
2019        let home = dir.path().to_path_buf();
2020        let disk = Disk::default();
2021        let now = ts("2026-09-05T00:00:00Z");
2022
2023        let superseded = due_run_with_status(
2024            &runs,
2025            &wt,
2026            "20260801-000000-cccc",
2027            SCHEMA,
2028            RunStatus::Superseded,
2029        );
2030
2031        let (folded, unreadable) =
2032            block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
2033        assert_eq!(
2034            folded, 1,
2035            "a superseded run has nothing left for a human to check, so it folds \
2036             exactly like a merged one"
2037        );
2038        assert_eq!(unreadable, 0);
2039        assert!(
2040            read_meta(&runs, &superseded).is_ok(),
2041            "folding drops the worktree, not the record"
2042        );
2043    }
2044
2045    /// A throwaway repo with one commit on `main`, for a test that needs a
2046    /// real winner worktree `fold_run` can actually remove.
2047    fn init_repo(dir: &Path) {
2048        use crate::proc::Quiet as _;
2049        let run = |args: &[&str]| {
2050            let out = std::process::Command::new("git")
2051                .args(args)
2052                .current_dir(dir)
2053                .quiet()
2054                .output()
2055                .expect("spawn git");
2056            assert!(
2057                out.status.success(),
2058                "git {args:?} failed: {}",
2059                String::from_utf8_lossy(&out.stderr)
2060            );
2061        };
2062        run(&["init", "-b", "main"]);
2063        run(&["config", "user.name", "magi test"]);
2064        run(&["config", "user.email", "magi@example.com"]);
2065        std::fs::write(dir.join("README.md"), "# fixture\n").unwrap();
2066        run(&["add", "-A"]);
2067        run(&["commit", "-m", "init"]);
2068    }
2069
2070    /// A due, terminal run with a decided winner that still has a real
2071    /// worktree and branch on disk - so a test can tell whether folding it
2072    /// actually removed the winner or only left it registered as folded.
2073    async fn due_run_with_winner_worktree(
2074        runs: &Path,
2075        wt: &Path,
2076        repo: &Path,
2077        id: &str,
2078        status: RunStatus,
2079    ) -> (String, PathBuf) {
2080        use crate::run::{Candidate, Tally};
2081        use std::collections::BTreeMap;
2082
2083        let mut config = crate::config::Config::default();
2084        config.graph.worktree_root = Some(wt.to_path_buf());
2085        let mut state = RunState::new(
2086            repo.to_path_buf(),
2087            "main".to_owned(),
2088            "0000000000000000000000000000000000000000".to_owned(),
2089            String::new(),
2090            config,
2091        );
2092        state.id = id.to_owned();
2093        state.status = status;
2094        state.updated_at = ts("2026-08-01T00:00:00Z");
2095
2096        let winner_wt = state.worktree_root().join("cand-A");
2097        crate::git::worktree_add_branch(repo, &winner_wt, &format!("magi/{id}/A"), "main")
2098            .await
2099            .expect("winner worktree");
2100
2101        state.candidates.push(Candidate {
2102            index: 0,
2103            label: 'A',
2104            agent: "agent".to_owned(),
2105            branch: format!("magi/{id}/A"),
2106            worktree: winner_wt.clone(),
2107            summary: String::new(),
2108            stat: String::new(),
2109            files: 0,
2110            commits: 0,
2111            empty: false,
2112            failed: None,
2113            verified_noop: None,
2114            duration_ms: 0,
2115            folded: false,
2116        });
2117        state.tally = Some(Tally {
2118            first_choice: BTreeMap::from([('A', 1)]),
2119            borda: BTreeMap::new(),
2120            winner: 'A',
2121            rankings: 1,
2122            unanimous_initial: true,
2123            deliberated: false,
2124            changed_votes: 0,
2125            unanimous_final: true,
2126            tie_break: None,
2127            judges: 1,
2128            present: 1,
2129            quorum: 1,
2130            met_quorum: true,
2131            uncontested: None,
2132        });
2133        std::fs::create_dir_all(runs.join(id)).unwrap();
2134        std::fs::write(
2135            runs.join(id).join("run.json"),
2136            serde_json::to_string_pretty(&state).unwrap(),
2137        )
2138        .unwrap();
2139        (id.to_owned(), winner_wt)
2140    }
2141
2142    #[tokio::test]
2143    async fn recovery_retains_historical_sources_after_success_and_automatic_cleanup() {
2144        let dir = tempfile::tempdir().unwrap();
2145        let home = dir.path();
2146        let runs = home.join("runs");
2147        let wt = home.join("wt");
2148        let repo = home.join("repo");
2149        std::fs::create_dir_all(&repo).unwrap();
2150        init_repo(&repo);
2151        let mut task = crate::queue::Task::new(
2152            "recover".into(),
2153            "keep measured behavior".into(),
2154            repo.clone(),
2155            crate::queue::Source::Human,
2156        );
2157        let mut sources = Vec::new();
2158        for (id, status) in [
2159            ("20260801-000000-old1", RunStatus::Blocked),
2160            ("20260801-000000-old2", RunStatus::Stalled),
2161        ] {
2162            let (id, worktree) = due_run_with_winner_worktree(&runs, &wt, &repo, id, status).await;
2163            let mut state = RunState::load_under(&id, home).unwrap();
2164            state.driver_exited = true;
2165            state.save_under(home).unwrap();
2166            task.link_run(&id);
2167            sources.push((id, worktree));
2168        }
2169        crate::pr_recovery::retain_sources(&task, home).unwrap();
2170        let mut recovered = RunState::new(
2171            repo.clone(),
2172            "main".into(),
2173            "base".into(),
2174            task.instruction.clone(),
2175            crate::config::Config::default(),
2176        );
2177        recovered.config.graph.worktree_root = Some(wt.clone());
2178        recovered.status = RunStatus::Ready;
2179        recovered.driver_exited = true;
2180        recovered.save_under(home).unwrap();
2181        task.start(recovered.id.clone());
2182        task.succeed();
2183        crate::daemon::supersede_prior_runs(&task, home);
2184        for (id, _) in &sources {
2185            assert_eq!(
2186                RunState::load_under(id, home).unwrap().status,
2187                RunStatus::Superseded
2188            );
2189        }
2190        let disk = Disk::default();
2191        let now = Timestamp::now()
2192            .checked_add(SignedDuration::from_secs(
2193                i64::try_from(disk.fold_grace_secs + 1).unwrap(),
2194            ))
2195            .unwrap();
2196        let (_, unreadable) = fold_due(&runs, home, &wt, &disk, now).await.unwrap();
2197        assert_eq!(unreadable, 0);
2198        for (id, worktree) in sources {
2199            assert!(worktree.exists());
2200            assert!(
2201                crate::git::branch_exists(&repo, &format!("magi/{id}/A"))
2202                    .await
2203                    .unwrap()
2204            );
2205            let mut historical = RunState::load_under(&id, home).unwrap();
2206            assert!(!historical.candidates[0].folded);
2207            assert!(
2208                crate::graph::fold_run(&mut historical, true, home)
2209                    .await
2210                    .unwrap()
2211                    .is_empty()
2212            );
2213            assert!(
2214                worktree.exists(),
2215                "every automatic fold entry preserves retained sources"
2216            );
2217        }
2218    }
2219
2220    #[tokio::test]
2221    async fn fold_due_drops_a_superseded_runs_own_winner_worktree_but_keeps_a_readys() {
2222        let dir = tempfile::tempdir().unwrap();
2223        let runs = dir.path().join("runs");
2224        let wt = dir.path().join("wt");
2225        let repo = dir.path().join("repo");
2226        let home = dir.path().to_path_buf();
2227        let disk = Disk::default();
2228        let now = ts("2026-09-05T00:00:00Z");
2229        std::fs::create_dir_all(&repo).unwrap();
2230        init_repo(&repo);
2231
2232        let (superseded_id, superseded_wt) = due_run_with_winner_worktree(
2233            &runs,
2234            &wt,
2235            &repo,
2236            "20260801-000000-supw",
2237            RunStatus::Superseded,
2238        )
2239        .await;
2240        // `Ready` is auto-folded exactly like `Superseded` (both terminal and
2241        // non-resumable), but never landed anywhere - its own winner is
2242        // deliberately kept, which is what makes it the right contrast here.
2243        // `Blocked`/`Stalled` are `resumable()` and so never even reach
2244        // `fold_run` in the first place; they would not exercise the
2245        // `drop_winner` choice this test is about.
2246        let (ready_id, ready_wt) = due_run_with_winner_worktree(
2247            &runs,
2248            &wt,
2249            &repo,
2250            "20260801-000000-rdyw",
2251            RunStatus::Ready,
2252        )
2253        .await;
2254
2255        let (folded, unreadable) = fold_due(&runs, &home, &wt, &disk, now)
2256            .await
2257            .expect("fold_due");
2258        assert_eq!(folded, 2);
2259        assert_eq!(unreadable, 0);
2260
2261        assert!(
2262            !superseded_wt.exists(),
2263            "a superseded run's own winner never lands anywhere else, so its worktree \
2264             must be dropped exactly like a merged run's"
2265        );
2266        assert!(
2267            ready_wt.exists(),
2268            "a still-ready run's winner may yet be merged by hand - folding must not \
2269             touch it"
2270        );
2271
2272        // Both records survive the fold; only the worktrees differ.
2273        assert!(read_meta(&runs, &superseded_id).is_ok());
2274        assert!(read_meta(&runs, &ready_id).is_ok());
2275    }
2276}