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 only verified clean Git worktrees and empty directory scaffolding.
593/// Unknown files or unreadable Git state require recovery, never recursive deletion.
594async fn remove_clean_bay(root: &Path) -> Result<()> {
595    let mut pending = vec![root.to_path_buf()];
596    let mut directories = Vec::new();
597    let mut worktrees = Vec::new();
598    while let Some(path) = pending.pop() {
599        if path.join(".git").exists() {
600            if !crate::git::is_clean(&path).await? {
601                anyhow::bail!("uncommitted work in {}; recovery required", path.display());
602            }
603            worktrees.push(path);
604        } else {
605            for entry in std::fs::read_dir(&path)? {
606                let entry = entry?;
607                if !entry.file_type()?.is_dir() {
608                    anyhow::bail!(
609                        "unknown worktree content in {}; recovery required",
610                        path.display()
611                    );
612                }
613                pending.push(entry.path());
614            }
615            directories.push(path);
616        }
617    }
618    for path in worktrees {
619        if !crate::git::worktree_remove_clean(&path, &path).await? {
620            anyhow::bail!("clean worktree removal refused: {}", path.display());
621        }
622    }
623    for path in directories.into_iter().rev() {
624        std::fs::remove_dir(path)?;
625    }
626    Ok(())
627}
628
629/// Remove a run that cannot be read: its state directory under `runs` and its
630/// worktree directory under `worktrees_root`.
631///
632/// Inspect worktrees before deleting any state or artifacts. Only clean registered
633/// worktrees and empty directory scaffolding can be reclaimed; unknown or dirty
634/// content remains available for recovery. Nonstandard worktree roots remain the
635/// operator's responsibility when the state file cannot identify them.
636pub async fn fold_unreadable(runs: &Path, worktrees_root: &Path, id: &str) -> Result<Vec<String>> {
637    let resolved = resolve_id_path(runs, id)?;
638    let mut removed = Vec::new();
639    let run_dir = runs.join(&resolved);
640    let wt = worktrees_root.join(short_of(&resolved));
641    if wt.exists() {
642        remove_clean_bay(&wt).await?;
643        removed.push(wt.to_string_lossy().into_owned());
644    }
645    // Keep the recovery artifacts if any worktree could not be removed safely.
646    if run_dir.exists() {
647        std::fs::remove_dir_all(&run_dir)
648            .with_context(|| format!("remove {}", run_dir.display()))?;
649        removed.push(format!("runs/{resolved}"));
650    }
651    Ok(removed)
652}
653
654/// Reclaim worktrees under `worktrees_root` that no run record in `runs`
655/// claims anymore, and return how many were removed.
656///
657/// [`fold_due`] only ever sees a worktree by walking `runs/` first, so a
658/// worktree whose run record is already gone — `magi run rm`, or a record
659/// deleted before its worktree — never enters that loop at all: nothing there
660/// is looking for it. This walks the worktree bay directly instead, and
661/// removes any `<short>` directory that no run id maps to.
662///
663/// Two things must never happen, and this checks both before ever touching a
664/// directory:
665///
666/// - **A worktree bay is never the only kind of thing under `worktrees_root`,
667///   and this must not assume it is.** A hand-placed scratch directory, or
668///   anything else an operator or another tool left in the same bay, has the
669///   same "no run claims it" shape as a genuine orphan but is not one -
670///   [`looks_like_a_worktree_bay`] is the same tag shape [`crate::run::is_run_id`]
671///   already requires of a real run's short id, and anything else is left
672///   alone regardless of what else is true about it.
673/// - **A worktree that was only just created might not have a `run.json` yet
674///   for a reason that has nothing to do with being orphaned.** `Runner::start`
675///   and `Runner::review` both create the worktree before the first
676///   `RunState::save` lands, and that gap - several `git` subprocesses wide -
677///   is invisible to [`crate::daemon::is_working_on_short`] whenever the run
678///   is not being driven through this daemon's own `poll` loop at all (a
679///   `magi review` invocation, for one). A directory whose own modification
680///   time is within `grace_secs` of `now` is left alone on that basis alone,
681///   the same margin [`fold_due`] gives a run before treating it as truly
682///   finished - long enough that no realistic gap between a `worktree add`
683///   and its `run.json` could ever be mistaken for one.
684///
685/// The one failure this must never cause is deleting the worktree of a run
686/// that is genuinely in flight. [`crate::daemon::is_working_on_short`] is the
687/// same liveness check [`fold_due`] trusts everywhere else in this module,
688/// checked by short id because there is no full id to compare here; when it
689/// cannot tell, this leaves the directory alone. Best-effort like the rest of
690/// housekeeping: one directory git or the filesystem refuses to give up is a
691/// `tracing::warn`, not a reason to abandon the rest of the pass.
692pub async fn fold_orphaned_worktrees(
693    runs: &Path,
694    worktrees_root: &Path,
695    home: &Path,
696    grace_secs: u64,
697    now: Timestamp,
698) -> usize {
699    let known: std::collections::HashSet<String> = std::fs::read_dir(runs)
700        .into_iter()
701        .flatten()
702        .flatten()
703        .map(|e| e.file_name().to_string_lossy().into_owned())
704        .filter(|name| crate::run::is_run_id(name))
705        .map(|id| short_of(&id).to_owned())
706        .collect();
707
708    let mut folded = 0usize;
709    for entry in std::fs::read_dir(worktrees_root)
710        .into_iter()
711        .flatten()
712        .flatten()
713    {
714        if !entry.path().is_dir() {
715            continue;
716        }
717        let short = entry.file_name().to_string_lossy().into_owned();
718        if !looks_like_a_worktree_bay(&short) {
719            continue;
720        }
721        if known.contains(&short) || crate::daemon::is_working_on_short(home, &short, now) {
722            continue;
723        }
724        let wt = entry.path();
725        if !stale_enough(&wt, grace_secs, now) {
726            continue;
727        }
728        match remove_clean_bay(&wt).await {
729            Ok(()) => folded += 1,
730            Err(e) => tracing::warn!(
731                "housekeep: remove orphaned worktree {}: {e:#}",
732                wt.display()
733            ),
734        }
735    }
736    folded
737}
738
739/// Does `name` have the shape a run's own worktree bay is named with: the
740/// same 4-character alphanumeric tag [`crate::run::is_run_id`] requires of a
741/// full id's trailing block (see [`short_of`])?
742///
743/// Anything else under `worktrees_root` is not a bay this function reclaims
744/// at all, claimed or not - answering "does a run claim this?" about a
745/// directory that was never a run's worktree in the first place is exactly
746/// the wrong question to ask before deleting it.
747fn looks_like_a_worktree_bay(name: &str) -> bool {
748    name.len() == 4 && name.bytes().all(|b| b.is_ascii_alphanumeric())
749}
750
751/// Is `dir`'s own modification time old enough, against `grace_secs` and
752/// `now`, that its emptiness of a run record can be trusted rather than
753/// caught mid-creation?
754///
755/// A directory this pass cannot stat at all - a race with its own removal, a
756/// permission error - is treated as not yet stale: unreadable metadata is not
757/// evidence of anything, and the janitor already keeps rather than deletes
758/// whenever it cannot tell (see the module docs).
759///
760/// `grace_secs` is floored at [`MIN_ORPHAN_AGE_SECS`] regardless of what the
761/// caller passes: `0` is a documented, legitimate value for
762/// [`crate::config::Disk::fold_grace_secs`] (`due`'s own "always due" case),
763/// because that grace answers a policy question the operator owns - how long
764/// a *known, finished* run's worktree lingers before cleanup. Whether an
765/// orphan worktree is actually a race with `Runner::review`'s `git worktree
766/// add` landing before its `run.json` is not a policy question, and must not
767/// collapse to zero just because the operator turned the other grace off -
768/// that would defeat the very check meant to catch it.
769fn stale_enough(dir: &Path, grace_secs: u64, now: Timestamp) -> bool {
770    let Ok(modified) = std::fs::metadata(dir).and_then(|m| m.modified()) else {
771        return false;
772    };
773    let Ok(ts) = Timestamp::try_from(modified) else {
774        return false;
775    };
776    due(now, ts, grace_secs.max(MIN_ORPHAN_AGE_SECS))
777}
778
779/// The floor under [`stale_enough`]'s grace, independent of
780/// [`crate::config::Disk::fold_grace_secs`].
781///
782/// Ample next to the race it guards: the gap between `Runner::start` or
783/// `Runner::review` creating a worktree and the first `RunState::save`
784/// landing is a handful of `git` subprocess calls, not minutes - but the
785/// janitor cannot tell "still mid-setup" from "orphaned" by any other signal
786/// for a run that never registers with `daemon::Status` at all (a `magi
787/// review` invocation, for one), so this is generous on purpose rather than
788/// tuned to the observed case.
789const MIN_ORPHAN_AGE_SECS: u64 = 5 * 60;
790
791/// Resolve an id or prefix against an explicit runs directory, exactly the way
792/// [`crate::run::resolve_id`] does against the global home.
793fn resolve_id_path(runs: &Path, prefix: &str) -> Result<String> {
794    // Keyed on the directory, not on a readable state file: the record this
795    // route exists to remove may be a lone `run.json.tmp` from a save that
796    // ran out of disk, and that is precisely the one a human needs a way to
797    // clear (see `crate::run::list_ids`).
798    if runs.join(prefix).is_dir() && crate::run::is_run_id(prefix) {
799        return Ok(prefix.to_owned());
800    }
801    let mut hits: Vec<String> = Vec::new();
802    for e in std::fs::read_dir(runs).into_iter().flatten().flatten() {
803        if !e.path().is_dir() {
804            continue;
805        }
806        let id = e.file_name().to_string_lossy().into_owned();
807        if crate::run::is_run_id(&id) && (id.starts_with(prefix) || id.ends_with(prefix)) {
808            hits.push(id);
809        }
810    }
811    match hits.len() {
812        1 => Ok(hits.into_iter().next().expect("exactly one hit")),
813        0 => bail!("no run matches `{prefix}`"),
814        _ => bail!(
815            "`{prefix}` matches {} runs: {}",
816            hits.len(),
817            hits.join(", ")
818        ),
819    }
820}
821
822/// `magi fold`'s recovery path for a run whose worktrees are already gone —
823/// so [`crate::graph::fold_run`] removed nothing — but whose `run.json` still
824/// lists active seats nobody is left to answer for: no live daemon claims the
825/// run, and every one of those seats has overrun its own timeout (see
826/// [`RunState::active_all_overrun`]). Clearing them and failing the run is
827/// what lets it be deleted afterward — [`RunState::ensure_can_delete`] only
828/// ever checks whether a live daemon is working on the run and whether its
829/// candidates are folded, not `status`, but a run stuck `implementing`
830/// forever with an empty worktree still reads as unresolved everywhere else
831/// (`magi show`, the deck, the phone) until this runs.
832///
833/// Returns `false` without changing anything when a live daemon still claims
834/// the run, or when some active seat has not actually overrun its budget yet
835/// — a run that is merely between waves must never be guessed at.
836pub fn clear_abandoned_active(state: &mut RunState, home: &Path, now: Timestamp) -> Result<bool> {
837    if crate::daemon::is_working_on(home, &state.id, now) || !state.active_all_overrun(now) {
838        return Ok(false);
839    }
840    state.abandon("fold");
841    state.save_under(home)?;
842    // The seat that asked is gone for good now - the same door
843    // `graph::Runner::settle_questions` closes the moment `status` lands
844    // somewhere non-resumable, see that method's own doc. Without this, an
845    // open question the abandoned seat left behind would keep badging the
846    // operator until the next daemon startup's `abandon_settled_questions`
847    // pass happened to notice it, or forever if nothing is running `magi
848    // serve` at all.
849    if let Err(e) = Questions::at(home.join("questions")).settle_run(&state.id, state.status) {
850        tracing::warn!("abandon questions for {}: {e:#}", state.id);
851    }
852    Ok(true)
853}
854
855/// Delete files from the shared build cache until it fits its cap — but only
856/// while nobody live is registered as using it. [`crate::cache::maintenance_prune`]
857/// takes out the same lease a build would, so a prune can never race a
858/// compile in flight (this run's own, another run's, or a human's `magi
859/// review`) into deleting a file that build still needs. `Ok(None)` when the
860/// cache is in use right now; the next pass catches it once the borrower
861/// releases it, the same way a cap of `0` or a missing `CARGO_TARGET_DIR`
862/// already meant "nothing to do this time" here.
863pub fn prune_cache(home: &Path, cache: &Path, limit_bytes: u64) -> Result<Option<Prune>> {
864    crate::cache::maintenance_prune(home, cache, limit_bytes)
865}
866
867/// [`prune_cache`], but resolving the operator's opt-out and missing
868/// `CARGO_TARGET_DIR` first — the same two checks [`housekeep`]'s idle pass
869/// makes before ever measuring the cache, factored out so
870/// [`crate::daemon`]'s between-runs check (see the module's own doc for why
871/// congestion can make "idle" arrive too rarely to matter) makes them
872/// identically rather than growing its own copy that could drift. `Ok(None)`
873/// covers a cap of `0` (see the module docs on `cache_limit_bytes`), a config
874/// that renders no `CARGO_TARGET_DIR` to aggregate at all, and a cache
875/// currently in use (see [`prune_cache`]).
876pub fn prune_cache_if_over_limit(
877    cfg: &crate::config::Config,
878    home: &Path,
879) -> Result<Option<Prune>> {
880    if cfg.disk.cache_limit_bytes == 0 {
881        return Ok(None);
882    }
883    let Some(cache) = cfg.cache_dir() else {
884        return Ok(None);
885    };
886    prune_cache(home, &cache, cfg.disk.cache_limit_bytes)
887}
888
889/// The cache's path, size and cap, for `magi cache show` and the health view.
890/// `None` when the config declares no `CARGO_TARGET_DIR` to aggregate.
891///
892/// A cap of `0` means the operator opted out of pruning; the size is then
893/// reported but never acted on.
894pub fn cache_report(cfg: &crate::config::Config) -> Option<(PathBuf, u64, u64)> {
895    let cache = cfg.cache_dir()?;
896    Some((
897        cache.clone(),
898        cache_size(&cache),
899        cfg.disk.cache_limit_bytes,
900    ))
901}
902
903/// Size in bytes of the shared build cache.
904pub fn cache_size(cache: &Path) -> u64 {
905    dir_size(cache)
906}
907
908#[cfg(test)]
909mod tests {
910
911    async fn sh(cwd: &Path, args: &[&str]) {
912        let out = crate::git::git_raw(cwd, args).await.expect("spawn git");
913        assert!(out.ok(), "git {args:?}: {}", out.stderr);
914    }
915
916    async fn commit(cwd: &Path, file: &str, body: &str) {
917        std::fs::write(cwd.join(file), body).unwrap();
918        sh(cwd, &["add", file]).await;
919        sh(
920            cwd,
921            &[
922                "-c",
923                "user.name=t",
924                "-c",
925                "user.email=t@example.com",
926                "commit",
927                "-q",
928                "-m",
929                body,
930            ],
931        )
932        .await;
933    }
934
935    const LONG: std::time::Duration = std::time::Duration::from_secs(30);
936
937    #[tokio::test]
938    async fn fetch_advances_origin_main_and_leaves_the_checkout_alone() {
939        let tmp = tempfile::tempdir().unwrap();
940        let t = tmp.path();
941        let bare = t.join("remote.git");
942        sh(
943            t,
944            &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
945        )
946        .await;
947        let dir = t.join("root/h/o/r");
948        std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
949        sh(
950            t,
951            &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
952        )
953        .await;
954        sh(&dir, &["checkout", "-q", "-b", "main"]).await;
955        commit(&dir, "a.txt", "one").await;
956        sh(&dir, &["push", "-q", "origin", "main"]).await;
957        sh(&dir, &["checkout", "-q", "--detach"]).await;
958        std::fs::write(dir.join("a.txt"), "dirty").unwrap();
959
960        let other = t.join("other");
961        sh(
962            t,
963            &[
964                "clone",
965                "-q",
966                bare.to_str().unwrap(),
967                other.to_str().unwrap(),
968            ],
969        )
970        .await;
971        commit(&other, "b.txt", "two").await;
972        sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
973
974        let rev = |d: PathBuf, r: &'static str| async move {
975            crate::git::git(&d, &["rev-parse", r]).await.unwrap()
976        };
977        let head = rev(dir.clone(), "HEAD").await;
978        let before = rev(dir.clone(), "origin/main").await;
979        let status = crate::git::git(&dir, &["status", "--porcelain"])
980            .await
981            .unwrap();
982
983        let r = fetch_origins(&[t.join("root")], LONG, || false).await;
984        assert_eq!(
985            r,
986            FetchReport {
987                fetched: 1,
988                no_origin: 0,
989                failed: 0
990            }
991        );
992
993        assert_ne!(rev(dir.clone(), "origin/main").await, before);
994        assert_eq!(rev(dir.clone(), "HEAD").await, head);
995        assert_eq!(
996            crate::git::git(&dir, &["status", "--porcelain"])
997                .await
998                .unwrap(),
999            status
1000        );
1001        assert_eq!(std::fs::read_to_string(dir.join("a.txt")).unwrap(), "dirty");
1002    }
1003
1004    #[tokio::test]
1005    async fn fetch_never_writes_a_local_branch_whatever_the_configured_refspec() {
1006        let tmp = tempfile::tempdir().unwrap();
1007        let t = tmp.path();
1008        let bare = t.join("remote.git");
1009        sh(
1010            t,
1011            &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
1012        )
1013        .await;
1014        let dir = t.join("root/h/o/r");
1015        std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
1016        sh(
1017            t,
1018            &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
1019        )
1020        .await;
1021        sh(&dir, &["checkout", "-q", "-b", "main"]).await;
1022        commit(&dir, "a.txt", "one").await;
1023        sh(&dir, &["push", "-q", "origin", "main"]).await;
1024        sh(&dir, &["checkout", "-q", "--detach"]).await;
1025        // A hostile mapping onto the local branch, and no origin/* at all.
1026        sh(
1027            &dir,
1028            &[
1029                "config",
1030                "remote.origin.fetch",
1031                "+refs/heads/main:refs/heads/main",
1032            ],
1033        )
1034        .await;
1035        let _ = crate::git::git_raw(&dir, &["update-ref", "-d", "refs/remotes/origin/main"]).await;
1036
1037        let other = t.join("other");
1038        sh(
1039            t,
1040            &[
1041                "clone",
1042                "-q",
1043                bare.to_str().unwrap(),
1044                other.to_str().unwrap(),
1045            ],
1046        )
1047        .await;
1048        commit(&other, "b.txt", "two").await;
1049        sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
1050        let remote_tip = crate::git::git(&other, &["rev-parse", "HEAD"])
1051            .await
1052            .unwrap();
1053        let local_main = crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1054            .await
1055            .unwrap();
1056
1057        let r = fetch_origins(&[t.join("root")], LONG, || false).await;
1058        assert_eq!(r.fetched, 1);
1059        assert_eq!(
1060            crate::git::git(&dir, &["rev-parse", "refs/remotes/origin/main"])
1061                .await
1062                .unwrap(),
1063            remote_tip
1064        );
1065        assert_eq!(
1066            crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1067                .await
1068                .unwrap(),
1069            local_main
1070        );
1071    }
1072
1073    #[tokio::test]
1074    async fn fetch_skips_no_origin_and_survives_an_unreachable_origin() {
1075        let tmp = tempfile::tempdir().unwrap();
1076        let t = tmp.path();
1077        let root = t.join("root");
1078        let bare = t.join("remote.git");
1079        sh(t, &["init", "-q", "--bare", bare.to_str().unwrap()]).await;
1080
1081        let none = root.join("h/o/none");
1082        let dead = root.join("h/o/dead");
1083        let good = root.join("h/o/good");
1084        for d in [&none, &dead, &good] {
1085            std::fs::create_dir_all(d).unwrap();
1086            sh(d, &["init", "-q"]).await;
1087        }
1088        sh(
1089            &dead,
1090            &[
1091                "remote",
1092                "add",
1093                "origin",
1094                t.join("missing").to_str().unwrap(),
1095            ],
1096        )
1097        .await;
1098        sh(&good, &["remote", "add", "origin", bare.to_str().unwrap()]).await;
1099
1100        let r = fetch_origins(&[root], LONG, || false).await;
1101        assert_eq!(
1102            r,
1103            FetchReport {
1104                fetched: 1,
1105                no_origin: 1,
1106                failed: 1
1107            }
1108        );
1109    }
1110    use super::*;
1111    use crate::config::Disk;
1112    use std::fs;
1113
1114    fn ts(s: &str) -> Timestamp {
1115        s.parse().expect("rfc3339")
1116    }
1117
1118    fn block_on<F: std::future::Future>(f: F) -> F::Output {
1119        tokio::runtime::Runtime::new().expect("runtime").block_on(f)
1120    }
1121
1122    #[test]
1123    fn a_run_is_due_after_its_grace_and_not_before() {
1124        let now = ts("2026-09-05T00:00:00Z");
1125        let grace = 600;
1126        let old = now - SignedDuration::new(601, 0);
1127        let fresh = now - SignedDuration::new(599, 0);
1128        assert!(due(now, old, grace));
1129        assert!(!due(now, fresh, grace));
1130        // Exactly at the edge: not yet due.
1131        let edge = now - SignedDuration::new(600, 0);
1132        assert!(!due(now, edge, grace));
1133        // A zero grace folds everything, ever.
1134        assert!(due(now, old, 0));
1135    }
1136
1137    #[test]
1138    fn rotate_from_moves_the_starting_point_as_the_cursor_advances() {
1139        let ids: Vec<String> = ["a", "b", "c", "d", "e"]
1140            .iter()
1141            .map(|s| s.to_string())
1142            .collect();
1143
1144        // A cursor of 0 starts at the front, same as an unrotated list.
1145        assert_eq!(rotate_from(&ids, 0), ids);
1146
1147        // A cursor of 2 starts two ids further along, with the skipped
1148        // front wrapping to the tail rather than being dropped.
1149        let at_2 = rotate_from(&ids, 2);
1150        assert_eq!(at_2, vec!["c", "d", "e", "a", "b"]);
1151        for id in &ids {
1152            assert!(at_2.contains(id));
1153        }
1154
1155        // A cursor past the list's own length wraps rather than panicking -
1156        // the caller never has to reduce it modulo anything itself.
1157        assert_eq!(rotate_from(&ids, 7), rotate_from(&ids, 2));
1158
1159        // An empty list has no start point to compute and must not panic.
1160        assert_eq!(rotate_from(&[], 0), Vec::<String>::new());
1161    }
1162
1163    /// A fleet with more eligible runs than one pass can check must not park
1164    /// the same handful at the front of the window forever: advancing the
1165    /// cursor by exactly how many ids one pass actually looked at - the
1166    /// arithmetic [`reconcile_external_merges`] itself does - walks every id
1167    /// to the front within one full rotation, with no reliance on how often
1168    /// or how regularly passes happen to run.
1169    #[test]
1170    fn advancing_the_cursor_by_the_window_size_gives_every_id_a_turn() {
1171        let ids: Vec<String> = (0..12).map(|n| format!("run-{n}")).collect();
1172        let cap = 5usize;
1173        let mut cursor = 0usize;
1174        let mut ever_seen: std::collections::HashSet<String> = std::collections::HashSet::new();
1175        for _ in 0..ids.len() {
1176            let window: Vec<String> = rotate_from(&ids, cursor).into_iter().take(cap).collect();
1177            ever_seen.extend(window.iter().cloned());
1178            cursor = (cursor + window.len()) % ids.len();
1179        }
1180        assert_eq!(
1181            ever_seen.len(),
1182            ids.len(),
1183            "every id must be checked at least once across a full rotation, whatever \
1184             the cadence between passes"
1185        );
1186    }
1187
1188    #[test]
1189    fn the_external_merge_cursor_round_trips_through_a_files_absence() {
1190        let dir = tempfile::tempdir().unwrap();
1191        let home = dir.path();
1192
1193        // Nothing written yet: starts at 0, not an error.
1194        assert_eq!(read_external_merge_cursor(home), 0);
1195
1196        write_external_merge_cursor(home, 7);
1197        assert_eq!(read_external_merge_cursor(home), 7);
1198
1199        // Anything unreadable as a plain number falls back to 0 rather than
1200        // wedging the sweep on a corrupt file.
1201        std::fs::write(home.join(EXTERNAL_MERGE_CURSOR_FILE), "not a number").unwrap();
1202        assert_eq!(read_external_merge_cursor(home), 0);
1203    }
1204
1205    #[test]
1206    fn a_run_qualifies_for_an_external_merge_check_only_when_blocked_unmerged_and_due() {
1207        let now = ts("2026-09-05T00:00:00Z");
1208        let grace = 600;
1209        let old = now - SignedDuration::new(601, 0);
1210        let fresh = now - SignedDuration::new(599, 0);
1211
1212        assert!(
1213            eligible_for_external_merge_check(RunStatus::Blocked, false, old, now, grace),
1214            "blocked, unmerged, and past its grace period is exactly the run this exists for"
1215        );
1216        assert!(
1217            !eligible_for_external_merge_check(RunStatus::Blocked, false, fresh, now, grace),
1218            "an operator mid-fix deserves the same grace window `fold_due` gives before \
1219             the janitor starts asking GitHub about it"
1220        );
1221        assert!(
1222            !eligible_for_external_merge_check(RunStatus::Blocked, true, old, now, grace),
1223            "a run `land::land` already recorded a merge outcome for has its own answer \
1224             already; this check is only for a run with nothing recorded at all"
1225        );
1226        assert!(
1227            !eligible_for_external_merge_check(RunStatus::Stalled, false, old, now, grace),
1228            "stalled is not blocked - it means the tally never reached quorum, which a \
1229             pull request cannot fix"
1230        );
1231        assert!(
1232            !eligible_for_external_merge_check(RunStatus::Ready, false, old, now, grace),
1233            "ready has nothing to correct - it was never landed by design"
1234        );
1235        assert!(
1236            !eligible_for_external_merge_check(RunStatus::Superseded, false, old, now, grace),
1237            "a superseded run's task already has its answer from a later attempt; \
1238             there is nothing left for GitHub to confirm here"
1239        );
1240    }
1241
1242    /// A `Blocked` run with a decided winner, so `land::find_external_merge`
1243    /// has a branch to ask `gh` about instead of returning early for lack of
1244    /// one - which is what makes an unreachable `/nonexistent/repo` fail
1245    /// loudly (an `Err`, raising a notice) rather than silently (an early
1246    /// `Ok(None)`, raising nothing) when this run's sweep is exercised.
1247    fn write_blocked_run(runs: &Path, id: &str, updated_at: Timestamp) {
1248        use crate::run::{Candidate, Tally};
1249        use std::collections::BTreeMap;
1250
1251        let mut state = RunState::new(
1252            PathBuf::from("/nonexistent/repo"),
1253            "main".to_owned(),
1254            "0000000000000000000000000000000000000000".to_owned(),
1255            String::new(),
1256            crate::config::Config::default(),
1257        );
1258        state.id = id.to_owned();
1259        state.status = RunStatus::Blocked;
1260        state.updated_at = updated_at;
1261        state.candidates.push(Candidate {
1262            index: 0,
1263            label: 'A',
1264            agent: "agent".to_owned(),
1265            branch: format!("magi/{id}/A"),
1266            worktree: PathBuf::from("/nonexistent/repo"),
1267            summary: String::new(),
1268            stat: String::new(),
1269            files: 0,
1270            commits: 0,
1271            empty: false,
1272            failed: None,
1273            verified_noop: None,
1274            duration_ms: 0,
1275            folded: false,
1276        });
1277        state.tally = Some(Tally {
1278            first_choice: BTreeMap::from([('A', 1)]),
1279            borda: BTreeMap::new(),
1280            winner: 'A',
1281            rankings: 1,
1282            unanimous_initial: true,
1283            deliberated: false,
1284            changed_votes: 0,
1285            unanimous_final: true,
1286            tie_break: None,
1287            judges: 1,
1288            present: 1,
1289            quorum: 1,
1290            met_quorum: true,
1291            uncontested: None,
1292        });
1293        std::fs::create_dir_all(runs.join(id)).unwrap();
1294        std::fs::write(
1295            runs.join(id).join("run.json"),
1296            serde_json::to_string_pretty(&state).unwrap(),
1297        )
1298        .unwrap();
1299    }
1300
1301    /// No run here can be confirmed merged - one has no repository `gh` can
1302    /// even ask about, one has not sat blocked long enough, and one is
1303    /// claimed by a live daemon - so the sweep must leave every one of them
1304    /// exactly as it found them and never panic on the failures in between.
1305    #[tokio::test]
1306    async fn reconcile_external_merges_leaves_ineligible_and_unconfirmable_runs_alone() {
1307        let dir = tempfile::tempdir().unwrap();
1308        let runs = dir.path().join("runs");
1309        let home = dir.path().to_path_buf();
1310        std::fs::create_dir_all(&runs).unwrap();
1311
1312        let now = ts("2026-09-05T00:00:00Z");
1313        let disk = Disk {
1314            fold_grace_secs: 600,
1315            ..Disk::default()
1316        };
1317
1318        let due_id = "20260905-000000-blkd";
1319        write_blocked_run(&runs, due_id, ts("2026-08-01T00:00:00Z"));
1320
1321        let fresh_id = "20260905-000000-fres";
1322        write_blocked_run(&runs, fresh_id, now);
1323
1324        let live_id = "20260905-000000-live";
1325        write_blocked_run(&runs, live_id, ts("2026-08-01T00:00:00Z"));
1326        let mut status = crate::daemon::Status::new();
1327        status.current = vec![crate::daemon::Current {
1328            task: "20260905-000000-task".to_owned(),
1329            run: live_id.to_owned(),
1330        }];
1331        status.updated_at = now;
1332        crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1333
1334        let reconciled = reconcile_external_merges(&runs, &home, &disk, now).await;
1335        assert_eq!(
1336            reconciled, 0,
1337            "an unreachable repository can never be confirmed merged"
1338        );
1339        for id in [due_id, fresh_id, live_id] {
1340            assert_eq!(
1341                read_meta(&runs, id).unwrap().status,
1342                RunStatus::Blocked,
1343                "{id} must be left exactly as it was found"
1344            );
1345        }
1346        let notices = crate::notices::Notices::at(home.join("notifications")).list();
1347        let due_notice = notices
1348            .iter()
1349            .find(|n| n.key == format!("merged-unrecorded:{due_id}"))
1350            .expect("notice for due_id must be raised");
1351        assert_eq!(
1352            due_notice.link,
1353            Some(crate::notices::Link::Run {
1354                id: due_id.to_owned(),
1355            })
1356        );
1357    }
1358
1359    /// The bug this exists to pin: a first version rotated on
1360    /// `now.as_second() % len`, which is fully determined by wall-clock time
1361    /// and nothing else. Two passes seconds apart - as they would be when a
1362    /// short poll interval and a fast pass alias onto the same handful of
1363    /// `now.as_second()` residues - landed on the *same* starting point and
1364    /// never covered a fleet bigger than the per-pass cap, however many
1365    /// times housekeeping ran. Persisting a cursor makes forward progress a
1366    /// function of how many ids got looked at, not of when the clock reads
1367    /// this pass happened to run.
1368    #[tokio::test]
1369    async fn reconcile_external_merges_covers_a_larger_fleet_across_repeated_passes_at_one_instant()
1370    {
1371        let dir = tempfile::tempdir().unwrap();
1372        let runs = dir.path().join("runs");
1373        let home = dir.path().to_path_buf();
1374        std::fs::create_dir_all(&runs).unwrap();
1375
1376        let now = ts("2026-09-05T00:00:00Z");
1377        let disk = Disk {
1378            fold_grace_secs: 600,
1379            ..Disk::default()
1380        };
1381        let old = ts("2026-08-01T00:00:00Z");
1382
1383        let ids: Vec<String> = (0..8).map(|n| format!("20260905-000000-r{n:03}")).collect();
1384        for id in &ids {
1385            write_blocked_run(&runs, id, old);
1386        }
1387
1388        let notices = crate::notices::Notices::at(home.join("notifications"));
1389
1390        // Two passes at the exact same `now`, exactly what a fast pass on a
1391        // short poll interval looks like. Each of the 8 unreachable-repo
1392        // runs raises its own notice the first time it is looked at, so the
1393        // count of distinct notices is a direct readout of how many distinct
1394        // ids have been checked so far.
1395        reconcile_external_merges(&runs, &home, &disk, now).await;
1396        let after_first = notices.list().len();
1397        assert_eq!(
1398            after_first, MAX_EXTERNAL_MERGE_CHECKS_PER_PASS,
1399            "the first pass checks exactly one cap's worth"
1400        );
1401
1402        reconcile_external_merges(&runs, &home, &disk, now).await;
1403        let after_second = notices.list().len();
1404        assert_eq!(
1405            after_second,
1406            ids.len(),
1407            "a second pass at the same instant must still reach every id the \
1408             first pass had no room for, not repeat the same cap's worth"
1409        );
1410    }
1411
1412    #[test]
1413    fn the_meta_reader_is_tolerant_of_everything_except_the_deciders() {
1414        let dir = tempfile::tempdir().unwrap();
1415        let runs = dir.path().join("runs");
1416        let id = "20260905-000000-abcd";
1417        std::fs::create_dir_all(runs.join(id)).unwrap();
1418        std::fs::write(
1419            runs.join(id).join("run.json"),
1420            r#"{"schema": 99, "id": "20260905-000000-abcd", "updated_at": "2026-09-05T00:00:00Z", "status": "ready", "junk_from_another_build": [1, 2, 3]}"#,
1421        )
1422        .unwrap();
1423        let meta = read_meta(&runs, id).expect("readable");
1424        assert_eq!(meta.status, RunStatus::Ready);
1425        assert_eq!(meta.updated_at, ts("2026-09-05T00:00:00Z"));
1426        assert!(read_meta(&runs, "nope").is_err(), "missing file unreadable");
1427        std::fs::write(runs.join(id).join("run.json"), "not json at all").unwrap();
1428        assert!(read_meta(&runs, id).is_err(), "garbage unreadable");
1429    }
1430
1431    #[test]
1432    fn fold_unreadable_preserves_unknown_content_and_recovery_artifacts() {
1433        let dir = tempfile::tempdir().unwrap();
1434        let runs = dir.path().join("runs");
1435        let wt = dir.path().join("wt");
1436        let id = "20260905-000000-abcd";
1437        std::fs::create_dir_all(runs.join(id).join("artifacts")).unwrap();
1438        std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1439        std::fs::create_dir_all(wt.join("abcd")).unwrap();
1440        std::fs::write(wt.join("abcd/leftover"), "owned edits").unwrap();
1441        assert!(block_on(fold_unreadable(&runs, &wt, id)).is_err());
1442        assert!(runs.join(id).join("artifacts").exists());
1443        assert_eq!(
1444            std::fs::read_to_string(wt.join("abcd/leftover")).unwrap(),
1445            "owned edits"
1446        );
1447    }
1448
1449    #[test]
1450    fn fold_unreadable_releases_run_dir_and_worktrees() {
1451        let dir = tempfile::tempdir().unwrap();
1452        let runs = dir.path().join("runs");
1453        let wt = dir.path().join("wt");
1454        let id = "20260905-000000-abcd";
1455        std::fs::create_dir_all(runs.join(id)).unwrap();
1456        std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1457        std::fs::create_dir_all(wt.join("abcd")).unwrap();
1458
1459        let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold");
1460        assert_eq!(removed.len(), 2);
1461        assert!(!runs.join(id).exists(), "run dir gone");
1462        assert!(!wt.join("abcd").exists(), "worktrees gone");
1463
1464        // A prefix resolves like `run::resolve_id` does.
1465        std::fs::create_dir_all(runs.join(id)).unwrap();
1466        std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1467        std::fs::create_dir_all(wt.join("abcd")).unwrap();
1468        let removed = block_on(fold_unreadable(&runs, &wt, "20260905")).expect("by prefix");
1469        assert_eq!(removed.len(), 2);
1470        // Once gone, `id` cannot be resolved at all - same as `run::resolve_id`
1471        // on an id nothing on disk matches - so a repeat pass errors rather
1472        // than silently reporting nothing removed.
1473        assert!(
1474            block_on(fold_unreadable(&runs, &wt, id)).is_err(),
1475            "a run already gone cannot be resolved again"
1476        );
1477    }
1478
1479    #[test]
1480    fn prune_cache_sheds_the_oldest_generation_until_it_fits() {
1481        let home = tempfile::tempdir().unwrap();
1482        let dir = tempfile::tempdir().unwrap();
1483        // Same size, different age: only the age decides, and the newest
1484        // generation - the one the next build reuses - is what survives.
1485        fs::write(dir.path().join("old"), b"xx").unwrap();
1486        fs::write(dir.path().join("new"), b"yy").unwrap();
1487        touch(&dir.path().join("old"), 1_000_000);
1488        touch(&dir.path().join("new"), 2_000_000);
1489
1490        let out = prune_cache(home.path(), dir.path(), 2)
1491            .expect("prune")
1492            .expect("the cache is free");
1493        assert_eq!(out.files, 1, "one deletion is enough to reach the cap");
1494        assert_eq!(out.remaining, 2);
1495        assert!(!dir.path().join("old").exists(), "the older file went");
1496        assert!(dir.path().join("new").exists(), "the newer one stayed");
1497
1498        // A whole generation shares one timestamp tick, so the tie has to be
1499        // decided too: largest first, which reaches the cap in the fewest
1500        // deletions. Left to `read_dir` and an unstable sort this deleted
1501        // both files on Linux and one on Windows.
1502        let tied = tempfile::tempdir().unwrap();
1503        fs::write(tied.path().join("big"), b"xxxx").unwrap();
1504        fs::write(tied.path().join("small"), b"yy").unwrap();
1505        touch(&tied.path().join("big"), 1_000_000);
1506        touch(&tied.path().join("small"), 1_000_000);
1507        let out = prune_cache(home.path(), tied.path(), 2)
1508            .expect("prune")
1509            .expect("the cache is free");
1510        assert_eq!(out.files, 1, "the big one alone gets under the cap");
1511        assert_eq!(out.remaining, 2);
1512        assert!(tied.path().join("small").exists());
1513    }
1514
1515    #[test]
1516    fn prune_cache_if_over_limit_resolves_the_opt_outs_before_ever_measuring() {
1517        let home = tempfile::tempdir().unwrap();
1518        let dir = tempfile::tempdir().unwrap();
1519        fs::write(dir.path().join("big"), vec![0u8; 10]).unwrap();
1520
1521        let mut cfg = crate::config::Config::default();
1522        cfg.verify.gate = vec![format!(
1523            "CARGO_TARGET_DIR={} cargo make check",
1524            dir.path().display()
1525        )];
1526
1527        // A cap of `0` is the operator's opt-out: never measured, never
1528        // pruned, regardless of what is actually on disk.
1529        cfg.disk.cache_limit_bytes = 0;
1530        assert_eq!(
1531            prune_cache_if_over_limit(&cfg, home.path()).unwrap(),
1532            None,
1533            "a zero cap must not even look at the directory"
1534        );
1535        assert!(dir.path().join("big").exists());
1536
1537        // No `CARGO_TARGET_DIR` in either verify command: nothing to
1538        // aggregate, so there is nothing to prune either.
1539        let mut no_cache = crate::config::Config::default();
1540        no_cache.disk.cache_limit_bytes = 1;
1541        assert_eq!(
1542            prune_cache_if_over_limit(&no_cache, home.path()).unwrap(),
1543            None
1544        );
1545
1546        // Over the cap and configured: pruned exactly like `prune_cache`
1547        // itself would.
1548        cfg.disk.cache_limit_bytes = 1;
1549        let pruned = prune_cache_if_over_limit(&cfg, home.path())
1550            .unwrap()
1551            .expect("a real cache dir over its cap prunes");
1552        assert_eq!(pruned.files, 1);
1553        assert!(!dir.path().join("big").exists());
1554    }
1555
1556    /// Pin a file's mtime, so a test asserts the policy and not the runner's
1557    /// timestamp granularity.
1558    fn touch(path: &Path, secs: u64) {
1559        let f = fs::File::options().write(true).open(path).unwrap();
1560        f.set_times(fs::FileTimes::new().set_modified(
1561            std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs),
1562        ))
1563        .unwrap();
1564    }
1565
1566    /// The disk-full casualty: a run whose first save left `run.json.tmp` and
1567    /// nothing else. It has to be clearable, or the record is permanent.
1568    #[test]
1569    fn fold_unreadable_clears_a_run_whose_state_never_landed() {
1570        let dir = tempfile::tempdir().unwrap();
1571        let runs = dir.path().join("runs");
1572        let wt = dir.path().join("wt");
1573        let id = "20260904-014540-88c0";
1574        std::fs::create_dir_all(runs.join(id)).unwrap();
1575        std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1576
1577        let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold by id");
1578        assert_eq!(removed, vec![format!("runs/{id}")]);
1579        assert!(!runs.join(id).exists(), "record gone");
1580
1581        // And by prefix, the way the deck and the phone address a run.
1582        std::fs::create_dir_all(runs.join(id)).unwrap();
1583        std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1584        assert!(
1585            block_on(fold_unreadable(&runs, &wt, "88c0")).is_ok(),
1586            "by prefix"
1587        );
1588
1589        // A directory under `runs` that is not a run is never a fold target.
1590        std::fs::create_dir_all(runs.join("scratch")).unwrap();
1591        assert!(
1592            block_on(fold_unreadable(&runs, &wt, "scratch")).is_err(),
1593            "a stray directory is not a run"
1594        );
1595    }
1596
1597    #[test]
1598    fn fold_due_folds_terminal_runs_of_any_schema_but_leaves_genuinely_unreadable_ones() {
1599        let dir = tempfile::tempdir().unwrap();
1600        let runs = dir.path().join("runs");
1601        let wt = dir.path().join("wt");
1602        let home = dir.path().to_path_buf();
1603        let disk = Disk::default();
1604        let now = ts("2026-09-05T00:00:00Z");
1605
1606        // 1. Runnable (judging): never folded, however old.
1607        let judging = "20260801-000000-0001";
1608        write_meta(&runs, judging, "judging", "2026-08-01T00:00:00Z");
1609
1610        // 2. Finished but fresh: grace not elapsed. Within the default 6h
1611        //    grace of `now`, so `fold_due` must stop at the freshness check
1612        //    and never even reach `read_state` - `write_meta`'s minimal JSON
1613        //    would fail that full parse anyway, and this case exists to
1614        //    prove freshness is why the run survives, not an accident of the
1615        //    fixture being unparseable as a whole `RunState`.
1616        let ready_fresh = "20260904-220000-0002";
1617        write_meta(&runs, ready_fresh, "ready", "2026-09-04T22:00:00Z");
1618
1619        // 3. Genuinely unreadable: broken JSON, not merely an unfamiliar
1620        //    schema number. Left alone and counted - this is the one case
1621        //    automatic housekeeping must never touch (see `fold_due`'s docs);
1622        //    discarding it is an explicit operator action, not something a
1623        //    background pass does.
1624        let garbage = "20260901-000000-0004";
1625        std::fs::create_dir_all(runs.join(garbage)).unwrap();
1626        std::fs::write(runs.join(garbage).join("run.json"), "not json").unwrap();
1627        std::fs::create_dir_all(wt.join("0004")).unwrap();
1628
1629        // 4. Finished, well past grace, current schema: the ordinary case
1630        //    `fold_due` has always acted on.
1631        let due_ready = due_run(&runs, &wt, "20260801-000000-ffff", SCHEMA);
1632
1633        // 5. Finished, well past grace, but written by a schema number this
1634        //    build no longer matches - the defect this task exists to fix.
1635        //    It still parses cleanly, so only the version number differs, and
1636        //    that alone must not block folding.
1637        let due_old_schema = due_run(&runs, &wt, "20260801-000000-eeee", SCHEMA - 1);
1638
1639        let (folded, unreadable) =
1640            block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
1641        assert_eq!(
1642            folded, 2,
1643            "both due, parseable runs fold regardless of their schema number"
1644        );
1645        assert_eq!(
1646            unreadable, 1,
1647            "only the run with broken JSON counts as unreadable"
1648        );
1649        assert!(runs.join(judging).exists(), "runnable never folded");
1650        assert!(runs.join(ready_fresh).exists(), "fresh never folded");
1651        assert!(runs.join(garbage).exists(), "unreadable record kept");
1652        assert!(wt.join("0004").exists(), "unreadable worktree kept");
1653        assert!(
1654            runs.join(&due_ready).exists(),
1655            "folding drops worktrees, not the record"
1656        );
1657        assert!(
1658            runs.join(&due_old_schema).exists(),
1659            "an old-schema record survives its fold exactly like a current one"
1660        );
1661        // `graph::fold_run` saves the state it just folded back to disk. That
1662        // write must land under this test's own `runs` - the argument it
1663        // passed to `fold_due`, not the process-global `run::home` - or a
1664        // fold that landed somewhere else entirely would still be counted
1665        // above as one of the two `folded` runs.
1666        for id in [&due_ready, &due_old_schema] {
1667            let saved = read_meta(&runs, id).expect("folded run still parses");
1668            assert_ne!(
1669                saved.updated_at,
1670                ts("2026-08-01T00:00:00Z"),
1671                "fold_run must have saved the updated state back through the \
1672                 `runs` directory this test passed to fold_due"
1673            );
1674        }
1675    }
1676
1677    /// `[disk] auto_fold = false` must leave the janitor's fold-and-reclaim
1678    /// passes completely inert - a due run's worktree and record both
1679    /// survive exactly as if `housekeep` had never run at all. Cache pruning
1680    /// is a separate opt-out (`cache_limit_bytes`) and stays disabled here
1681    /// too, so this test is only ever about `auto_fold`.
1682    #[tokio::test]
1683    async fn housekeep_leaves_everything_alone_when_auto_fold_is_disabled() {
1684        let dir = tempfile::tempdir().unwrap();
1685        let runs = dir.path().join("runs");
1686        let wt = dir.path().join("wt");
1687        let home = dir.path().to_path_buf();
1688        crate::run::set_home(dir.path().to_path_buf());
1689
1690        let due_id = due_run(&runs, &wt, "20260801-000000-abcd", SCHEMA);
1691        std::fs::create_dir_all(wt.join("orphan").join("cand-A")).unwrap();
1692
1693        let mut cfg = crate::config::Config::default();
1694        cfg.disk.auto_fold = false;
1695        cfg.disk.cache_limit_bytes = 0;
1696
1697        let out = housekeep(&cfg, &home, &wt, &dir.path().join("repo"), Timestamp::now()).await;
1698
1699        assert_eq!(out.folded, 0);
1700        assert_eq!(out.unreadable, 0);
1701        assert_eq!(out.orphaned_worktrees, 0);
1702        assert!(
1703            runs.join(&due_id).exists(),
1704            "a due run's record survives untouched"
1705        );
1706        assert!(
1707            wt.join("orphan").exists(),
1708            "an orphaned worktree survives untouched: the reclaim pass never ran"
1709        );
1710    }
1711
1712    #[test]
1713    fn fold_orphaned_worktrees_removes_only_worktrees_no_run_claims_and_none_in_flight() {
1714        let dir = tempfile::tempdir().unwrap();
1715        let runs = dir.path().join("runs");
1716        let wt = dir.path().join("wt");
1717        let home = dir.path().to_path_buf();
1718
1719        // A run record exists for this one: its worktree is claimed, not
1720        // orphaned, however old the record.
1721        write_meta(
1722            &runs,
1723            "20260801-000000-aaaa",
1724            "ready",
1725            "2026-08-01T00:00:00Z",
1726        );
1727        std::fs::create_dir_all(wt.join("aaaa").join("cand-A")).unwrap();
1728
1729        // No run record at all, and nobody is working on it: this is the
1730        // leftover `fold_due` can never see, because it only ever walks
1731        // `runs/`.
1732        std::fs::create_dir_all(wt.join("bbbb").join("cand-A")).unwrap();
1733
1734        // No run record either, but a live daemon status names a run with
1735        // this short id - the save-timing gap between the daemon claiming a
1736        // task and `RunState::new` writing its first `run.json`. Must survive
1737        // untouched.
1738        std::fs::create_dir_all(wt.join("cccc")).unwrap();
1739
1740        // Not shaped like a run's short id at all - a scratch directory an
1741        // operator or another tool left in the same bay - so it is never a
1742        // reclaim target regardless of what runs claim it or not.
1743        std::fs::create_dir_all(wt.join("scratch")).unwrap();
1744
1745        // `now` pushed comfortably past `MIN_ORPHAN_AGE_SECS`, so a zero
1746        // grace - the same "always due" escape hatch `due` itself documents
1747        // - still reclaims once a worktree is genuinely old, without faking
1748        // an mtime: real directory creation just above is already in the
1749        // past relative to this `now`, by design rather than by timing.
1750        let now = Timestamp::now() + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1751        let mut status = crate::daemon::Status::new();
1752        status.current = vec![crate::daemon::Current {
1753            task: "20260905-000000-t111".to_owned(),
1754            run: "20260905-000000-cccc".to_owned(),
1755        }];
1756        status.updated_at = now;
1757        crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1758
1759        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1760        assert_eq!(
1761            folded, 1,
1762            "only the truly orphaned, idle, bay-shaped worktree is removed"
1763        );
1764        assert!(wt.join("aaaa").exists(), "claimed by a run record");
1765        assert!(!wt.join("bbbb").exists(), "orphaned and idle: reclaimed");
1766        assert!(wt.join("cccc").exists(), "a run in flight is never touched");
1767        assert!(
1768            wt.join("scratch").exists(),
1769            "not shaped like a worktree bay, so never a reclaim target"
1770        );
1771    }
1772
1773    /// The gap this closes: `Runner::review` (`magi review`) creates the
1774    /// worktree with `git worktree add` before `RunState::save` ever writes a
1775    /// `run.json`, and that path never runs through the daemon's own `poll`
1776    /// loop at all, so `daemon::Status` never names it either. Without a
1777    /// grace window, a janitor pass landing in that gap would read the
1778    /// worktree as an orphan nothing is waiting on and delete a review still
1779    /// being set up.
1780    #[test]
1781    fn fold_orphaned_worktrees_leaves_a_freshly_created_bay_alone() {
1782        let dir = tempfile::tempdir().unwrap();
1783        let runs = dir.path().join("runs");
1784        let wt = dir.path().join("wt");
1785        let home = dir.path().to_path_buf();
1786
1787        std::fs::create_dir_all(wt.join("dddd").join("under-review")).unwrap();
1788
1789        let now = Timestamp::now();
1790        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 6 * 60 * 60, now));
1791        assert_eq!(
1792            folded, 0,
1793            "too fresh to tell apart from a run still being set up"
1794        );
1795        assert!(wt.join("dddd").exists());
1796    }
1797
1798    /// A fresh open question on `run`, stored and handed back for assertions.
1799    fn open_question(store: &Questions, run: &str) -> crate::ask::Question {
1800        let mut q = crate::ask::Question::new(
1801            run.to_owned(),
1802            "implement".to_owned(),
1803            "impl-A".to_owned(),
1804            "Which storage backend should the cache use?".to_owned(),
1805            String::new(),
1806            vec!["SQLite".to_owned(), "Redis".to_owned()],
1807        );
1808        store.put(&mut q).unwrap();
1809        q
1810    }
1811
1812    /// The exact ghost the phone showed: a run that already finished, with a
1813    /// question its dead seat asked still sitting `open` because it reached
1814    /// that status before `graph::Runner::settle_questions` existed (or
1815    /// missed it in the crash window `daemon::reclaim_orphaned_running`
1816    /// covers). This sweep is the second door to the same fact.
1817    #[test]
1818    fn a_finished_runs_open_question_is_swept_up() {
1819        let dir = tempfile::tempdir().unwrap();
1820        let runs = dir.path().join("runs");
1821        let store = Questions::at(dir.path().join("questions"));
1822
1823        let failed = "20260908-205802-c9eb";
1824        write_meta(&runs, failed, "failed", "2026-09-08T20:58:02Z");
1825        let failed_q = open_question(&store, failed);
1826
1827        let merged = "20260908-205501-ca67";
1828        write_meta(&runs, merged, "merged", "2026-09-08T20:55:01Z");
1829        let merged_q = open_question(&store, merged);
1830
1831        let n = abandon_settled_questions(&store, &runs);
1832        assert_eq!(n, 2, "both dead runs' questions are swept in one pass");
1833
1834        for (id, run) in [(&failed_q.id, failed), (&merged_q.id, merged)] {
1835            let back = store.get(id).unwrap();
1836            assert!(!back.status.open(), "{run} is done; nobody reads an answer");
1837            assert!(back.detail.contains(run), "{}", back.detail);
1838        }
1839    }
1840
1841    #[test]
1842    fn a_still_alive_runs_open_question_survives_the_sweep() {
1843        let dir = tempfile::tempdir().unwrap();
1844        let runs = dir.path().join("runs");
1845        let store = Questions::at(dir.path().join("questions"));
1846
1847        // `Blocked` and `Stalled` are `RunStatus::resumable`: the run can
1848        // still be picked back up, so its question may yet get a real
1849        // answer. A run still mid-competition is even more obviously alive.
1850        for (id, status) in [
1851            ("20260908-000000-b10c", "blocked"),
1852            ("20260908-000000-5ta1", "stalled"),
1853            ("20260908-000000-jud6", "judging"),
1854        ] {
1855            write_meta(&runs, id, status, "2026-09-08T00:00:00Z");
1856            let q = open_question(&store, id);
1857
1858            let n = abandon_settled_questions(&store, &runs);
1859            assert_eq!(n, 0, "{status} run is not done; nothing to sweep");
1860            assert!(
1861                store.get(&q.id).unwrap().status.open(),
1862                "{status} run's question must still be waiting"
1863            );
1864        }
1865    }
1866
1867    #[test]
1868    fn the_sweep_leaves_an_answered_question_and_an_unreadable_run_alone() {
1869        let dir = tempfile::tempdir().unwrap();
1870        let runs = dir.path().join("runs");
1871        let store = Questions::at(dir.path().join("questions"));
1872
1873        // Already decided: a sweep must never revisit it, whatever the run
1874        // that asked went on to become.
1875        let done = "20260908-000000-answ";
1876        write_meta(&runs, done, "failed", "2026-09-08T00:00:00Z");
1877        let mut answered = open_question(&store, done);
1878        answered
1879            .answer(crate::ask::Answer::Choice("SQLite".to_owned()))
1880            .unwrap();
1881        store.put(&mut answered).unwrap();
1882
1883        // No `run.json` at all for this one - deleted, or never landed.
1884        let gone = "20260908-000000-gone";
1885        let orphan = open_question(&store, gone);
1886
1887        assert_eq!(abandon_settled_questions(&store, &runs), 0);
1888        assert_eq!(
1889            store.get(&answered.id).unwrap().status,
1890            crate::ask::QuestionStatus::Answered,
1891            "a real answer is never overwritten by a sweep"
1892        );
1893        assert!(
1894            store.get(&orphan.id).unwrap().status.open(),
1895            "a run this sweep cannot read is left exactly as it was, not guessed at"
1896        );
1897    }
1898
1899    /// A grace of `0` is a legitimate, documented value for the operator's
1900    /// own `Disk::fold_grace_secs` - `due`'s "always due" case - but the
1901    /// freshness check this guards is not that policy, and must not collapse
1902    /// to it: a `0` handed straight through would reclaim a worktree the
1903    /// instant it exists, exactly the race `fold_orphaned_worktrees_leaves_a_
1904    /// freshly_created_bay_alone` exists to rule out, just with the operator
1905    /// having turned the other grace off instead of leaving it at its
1906    /// default.
1907    #[test]
1908    fn fold_orphaned_worktrees_floors_a_zero_grace_at_the_race_safe_minimum() {
1909        let dir = tempfile::tempdir().unwrap();
1910        let runs = dir.path().join("runs");
1911        let wt = dir.path().join("wt");
1912        let home = dir.path().to_path_buf();
1913
1914        std::fs::create_dir_all(wt.join("eeee").join("under-review")).unwrap();
1915
1916        // Too fresh, even with the grace argument at zero.
1917        let now = Timestamp::now();
1918        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1919        assert_eq!(
1920            folded, 0,
1921            "a zero grace must not defeat the race-safety floor"
1922        );
1923        assert!(wt.join("eeee").exists());
1924
1925        // Once genuinely past the floor, a zero grace reclaims it - the
1926        // floor is a minimum, not a replacement policy that never fires.
1927        let later = now + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1928        let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, later));
1929        assert_eq!(folded, 1, "old enough now, regardless of the zero grace");
1930        assert!(!wt.join("eeee").exists());
1931    }
1932
1933    #[test]
1934    fn clear_abandoned_active_only_acts_once_dead_and_overrun() {
1935        let dir = tempfile::tempdir().unwrap();
1936        // Harmless if another test in this binary already pinned the global
1937        // home first (see `run::set_home`'s own doc): this test only checks
1938        // the in-memory mutation `clear_abandoned_active` makes, never a
1939        // write that landed under this exact directory.
1940        crate::run::set_home(dir.path().to_path_buf());
1941        let home = dir.path().to_path_buf();
1942        let now = ts("2026-09-14T12:00:00Z");
1943        let overrun_seat = || crate::run::ActiveSeat {
1944            node: "implement".to_owned(),
1945            started_at: now - SignedDuration::new(21_000, 0),
1946            timeout_secs: 3_600,
1947            attempt: 0,
1948            task: None,
1949            command: None,
1950            index: None,
1951            total: None,
1952        };
1953
1954        let mut state = RunState::new(
1955            PathBuf::from("/repo"),
1956            "main".to_owned(),
1957            "abc1234".to_owned(),
1958            "fixture".to_owned(),
1959            crate::config::Config::default(),
1960        );
1961        state.status = RunStatus::Implementing;
1962        state.active.insert("impl-A".to_owned(), overrun_seat());
1963
1964        // A seat still within its own budget: not provably dead yet, so this
1965        // must change nothing.
1966        let mut fresh = state.clone();
1967        fresh.active.insert(
1968            "impl-B".to_owned(),
1969            crate::run::ActiveSeat {
1970                node: "implement".to_owned(),
1971                started_at: now,
1972                timeout_secs: 3_600,
1973                attempt: 0,
1974                task: None,
1975                command: None,
1976                index: None,
1977                total: None,
1978            },
1979        );
1980        assert!(!clear_abandoned_active(&mut fresh, &home, now).unwrap());
1981        assert!(!fresh.active.is_empty());
1982        assert_eq!(fresh.status, RunStatus::Implementing);
1983
1984        let store = Questions::at(home.join("questions"));
1985        let q = open_question(&store, &state.id);
1986
1987        assert!(clear_abandoned_active(&mut state, &home, now).unwrap());
1988        assert!(state.active.is_empty());
1989        assert_eq!(state.status, RunStatus::Failed);
1990        assert!(
1991            !store.get(&q.id).unwrap().status.open(),
1992            "the abandoned seat's own open question must not keep badging the \
1993             operator until some later daemon startup notices it"
1994        );
1995    }
1996
1997    /// Write a whole `run.json` that magi can read, over the given state.
1998    fn write_meta(runs: &Path, id: &str, status: &str, updated_at: &str) {
1999        let day = &updated_at[..10];
2000        std::fs::create_dir_all(runs.join(id)).unwrap();
2001        let body = format!(
2002            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}}"#
2003        );
2004        std::fs::write(runs.join(id).join("run.json"), body).unwrap();
2005    }
2006
2007    /// Write a fully-formed, `Ready`, well-past-grace `run.json` tagged with
2008    /// an arbitrary schema number - so a test can write one this build's own
2009    /// `RunState::new` could never produce on its own. Returns the id.
2010    ///
2011    /// `wt` becomes this run's `graph.worktree_root`: left at the config
2012    /// default, `RunState::worktree_root` falls through to
2013    /// `run::default_worktree_root` - the operator's real `~/wt/magi` - and
2014    /// `graph::fold_run`'s second sweep would then `read_dir` and remove
2015    /// worktrees there instead of anything this test owns.
2016    fn due_run(runs: &Path, wt: &Path, id: &str, schema: u32) -> String {
2017        due_run_with_status(runs, wt, id, schema, RunStatus::Ready)
2018    }
2019
2020    /// [`due_run`], with the terminal status a caller wants instead of the
2021    /// `Ready` every other caller here happens to want.
2022    fn due_run_with_status(
2023        runs: &Path,
2024        wt: &Path,
2025        id: &str,
2026        schema: u32,
2027        status: RunStatus,
2028    ) -> String {
2029        let mut config = crate::config::Config::default();
2030        config.graph.worktree_root = Some(wt.to_path_buf());
2031        let mut state = RunState::new(
2032            PathBuf::from("/nonexistent/repo"),
2033            "main".to_owned(),
2034            "0000000000000000000000000000000000000000".to_owned(),
2035            String::new(),
2036            config,
2037        );
2038        state.id = id.to_owned();
2039        state.status = status;
2040        state.updated_at = ts("2026-08-01T00:00:00Z");
2041        let mut value = serde_json::to_value(&state).unwrap();
2042        value["schema"] = serde_json::json!(schema);
2043        std::fs::create_dir_all(runs.join(id)).unwrap();
2044        std::fs::write(
2045            runs.join(id).join("run.json"),
2046            serde_json::to_string_pretty(&value).unwrap(),
2047        )
2048        .unwrap();
2049        id.to_owned()
2050    }
2051
2052    #[test]
2053    fn fold_due_folds_a_superseded_run_same_as_any_other_terminal_one() {
2054        let dir = tempfile::tempdir().unwrap();
2055        let runs = dir.path().join("runs");
2056        let wt = dir.path().join("wt");
2057        let home = dir.path().to_path_buf();
2058        let disk = Disk::default();
2059        let now = ts("2026-09-05T00:00:00Z");
2060
2061        let superseded = due_run_with_status(
2062            &runs,
2063            &wt,
2064            "20260801-000000-cccc",
2065            SCHEMA,
2066            RunStatus::Superseded,
2067        );
2068
2069        let (folded, unreadable) =
2070            block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
2071        assert_eq!(
2072            folded, 1,
2073            "a superseded run has nothing left for a human to check, so it folds \
2074             exactly like a merged one"
2075        );
2076        assert_eq!(unreadable, 0);
2077        assert!(
2078            read_meta(&runs, &superseded).is_ok(),
2079            "folding drops the worktree, not the record"
2080        );
2081    }
2082
2083    /// A throwaway repo with one commit on `main`, for a test that needs a
2084    /// real winner worktree `fold_run` can actually remove.
2085    fn init_repo(dir: &Path) {
2086        use crate::proc::Quiet as _;
2087        let run = |args: &[&str]| {
2088            let out = std::process::Command::new("git")
2089                .args(args)
2090                .current_dir(dir)
2091                .quiet()
2092                .output()
2093                .expect("spawn git");
2094            assert!(
2095                out.status.success(),
2096                "git {args:?} failed: {}",
2097                String::from_utf8_lossy(&out.stderr)
2098            );
2099        };
2100        run(&["init", "-b", "main"]);
2101        run(&["config", "user.name", "magi test"]);
2102        run(&["config", "user.email", "magi@example.com"]);
2103        std::fs::write(dir.join("README.md"), "# fixture\n").unwrap();
2104        run(&["add", "-A"]);
2105        run(&["commit", "-m", "init"]);
2106    }
2107
2108    /// A due, terminal run with a decided winner that still has a real
2109    /// worktree and branch on disk - so a test can tell whether folding it
2110    /// actually removed the winner or only left it registered as folded.
2111    async fn due_run_with_winner_worktree(
2112        runs: &Path,
2113        wt: &Path,
2114        repo: &Path,
2115        id: &str,
2116        status: RunStatus,
2117    ) -> (String, PathBuf) {
2118        use crate::run::{Candidate, Tally};
2119        use std::collections::BTreeMap;
2120
2121        let mut config = crate::config::Config::default();
2122        config.graph.worktree_root = Some(wt.to_path_buf());
2123        let mut state = RunState::new(
2124            repo.to_path_buf(),
2125            "main".to_owned(),
2126            "0000000000000000000000000000000000000000".to_owned(),
2127            String::new(),
2128            config,
2129        );
2130        state.id = id.to_owned();
2131        state.status = status;
2132        state.updated_at = ts("2026-08-01T00:00:00Z");
2133
2134        let winner_wt = state.worktree_root().join("cand-A");
2135        crate::git::worktree_add_branch(repo, &winner_wt, &format!("magi/{id}/A"), "main")
2136            .await
2137            .expect("winner worktree");
2138
2139        state.candidates.push(Candidate {
2140            index: 0,
2141            label: 'A',
2142            agent: "agent".to_owned(),
2143            branch: format!("magi/{id}/A"),
2144            worktree: winner_wt.clone(),
2145            summary: String::new(),
2146            stat: String::new(),
2147            files: 0,
2148            commits: 0,
2149            empty: false,
2150            failed: None,
2151            verified_noop: None,
2152            duration_ms: 0,
2153            folded: false,
2154        });
2155        state.tally = Some(Tally {
2156            first_choice: BTreeMap::from([('A', 1)]),
2157            borda: BTreeMap::new(),
2158            winner: 'A',
2159            rankings: 1,
2160            unanimous_initial: true,
2161            deliberated: false,
2162            changed_votes: 0,
2163            unanimous_final: true,
2164            tie_break: None,
2165            judges: 1,
2166            present: 1,
2167            quorum: 1,
2168            met_quorum: true,
2169            uncontested: None,
2170        });
2171        std::fs::create_dir_all(runs.join(id)).unwrap();
2172        std::fs::write(
2173            runs.join(id).join("run.json"),
2174            serde_json::to_string_pretty(&state).unwrap(),
2175        )
2176        .unwrap();
2177        (id.to_owned(), winner_wt)
2178    }
2179
2180    #[tokio::test]
2181    async fn recovery_retains_historical_sources_after_success_and_automatic_cleanup() {
2182        let dir = tempfile::tempdir().unwrap();
2183        let home = dir.path();
2184        let runs = home.join("runs");
2185        let wt = home.join("wt");
2186        let repo = home.join("repo");
2187        std::fs::create_dir_all(&repo).unwrap();
2188        init_repo(&repo);
2189        let mut task = crate::queue::Task::new(
2190            "recover".into(),
2191            "keep measured behavior".into(),
2192            repo.clone(),
2193            crate::queue::Source::Human,
2194        );
2195        let mut sources = Vec::new();
2196        for (id, status) in [
2197            ("20260801-000000-old1", RunStatus::Blocked),
2198            ("20260801-000000-old2", RunStatus::Stalled),
2199        ] {
2200            let (id, worktree) = due_run_with_winner_worktree(&runs, &wt, &repo, id, status).await;
2201            let mut state = RunState::load_under(&id, home).unwrap();
2202            state.driver_exited = true;
2203            state.save_under(home).unwrap();
2204            task.link_run(&id);
2205            sources.push((id, worktree));
2206        }
2207        crate::pr_recovery::retain_sources(&task, home).unwrap();
2208        let mut recovered = RunState::new(
2209            repo.clone(),
2210            "main".into(),
2211            "base".into(),
2212            task.instruction.clone(),
2213            crate::config::Config::default(),
2214        );
2215        recovered.config.graph.worktree_root = Some(wt.clone());
2216        recovered.status = RunStatus::Ready;
2217        recovered.driver_exited = true;
2218        recovered.save_under(home).unwrap();
2219        task.start(recovered.id.clone());
2220        task.succeed();
2221        crate::daemon::supersede_prior_runs(&task, home);
2222        for (id, _) in &sources {
2223            assert_eq!(
2224                RunState::load_under(id, home).unwrap().status,
2225                RunStatus::Superseded
2226            );
2227        }
2228        let disk = Disk::default();
2229        let now = Timestamp::now()
2230            .checked_add(SignedDuration::from_secs(
2231                i64::try_from(disk.fold_grace_secs + 1).unwrap(),
2232            ))
2233            .unwrap();
2234        let (_, unreadable) = fold_due(&runs, home, &wt, &disk, now).await.unwrap();
2235        assert_eq!(unreadable, 0);
2236        for (id, worktree) in sources {
2237            assert!(worktree.exists());
2238            assert!(
2239                crate::git::branch_exists(&repo, &format!("magi/{id}/A"))
2240                    .await
2241                    .unwrap()
2242            );
2243            let mut historical = RunState::load_under(&id, home).unwrap();
2244            assert!(!historical.candidates[0].folded);
2245            assert!(
2246                crate::graph::fold_run(&mut historical, true, home)
2247                    .await
2248                    .unwrap()
2249                    .is_empty()
2250            );
2251            assert!(
2252                worktree.exists(),
2253                "every automatic fold entry preserves retained sources"
2254            );
2255        }
2256    }
2257
2258    #[tokio::test]
2259    async fn fold_due_drops_a_superseded_runs_own_winner_worktree_but_keeps_a_readys() {
2260        let dir = tempfile::tempdir().unwrap();
2261        let runs = dir.path().join("runs");
2262        let wt = dir.path().join("wt");
2263        let repo = dir.path().join("repo");
2264        let home = dir.path().to_path_buf();
2265        let disk = Disk::default();
2266        let now = ts("2026-09-05T00:00:00Z");
2267        std::fs::create_dir_all(&repo).unwrap();
2268        init_repo(&repo);
2269
2270        let (superseded_id, superseded_wt) = due_run_with_winner_worktree(
2271            &runs,
2272            &wt,
2273            &repo,
2274            "20260801-000000-supw",
2275            RunStatus::Superseded,
2276        )
2277        .await;
2278        // `Ready` is auto-folded exactly like `Superseded` (both terminal and
2279        // non-resumable), but never landed anywhere - its own winner is
2280        // deliberately kept, which is what makes it the right contrast here.
2281        // `Blocked`/`Stalled` are `resumable()` and so never even reach
2282        // `fold_run` in the first place; they would not exercise the
2283        // `drop_winner` choice this test is about.
2284        let (ready_id, ready_wt) = due_run_with_winner_worktree(
2285            &runs,
2286            &wt,
2287            &repo,
2288            "20260801-000000-rdyw",
2289            RunStatus::Ready,
2290        )
2291        .await;
2292
2293        let (folded, unreadable) = fold_due(&runs, &home, &wt, &disk, now)
2294            .await
2295            .expect("fold_due");
2296        assert_eq!(folded, 2);
2297        assert_eq!(unreadable, 0);
2298
2299        assert!(
2300            !superseded_wt.exists(),
2301            "a superseded run's own winner never lands anywhere else, so its worktree \
2302             must be dropped exactly like a merged run's"
2303        );
2304        assert!(
2305            ready_wt.exists(),
2306            "a still-ready run's winner may yet be merged by hand - folding must not \
2307             touch it"
2308        );
2309
2310        // Both records survive the fold; only the worktrees differ.
2311        assert!(read_meta(&runs, &superseded_id).is_ok());
2312        assert!(read_meta(&runs, &ready_id).is_ok());
2313    }
2314}