1use 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#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
39pub struct Housekeeping {
40 pub folded: usize,
42 pub unreadable: usize,
52 pub orphaned_worktrees: usize,
55 pub cache_files: usize,
57 pub cache_freed: u64,
59 pub questions_abandoned: usize,
62 pub external_merges_recorded: usize,
67 pub stale_pr_states_repaired: usize,
71}
72
73pub 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 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 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 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 out.questions_abandoned =
138 abandon_settled_questions(&Questions::at(home.join("questions")), &home.join("runs"));
139 out
140}
141
142pub 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
172pub 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 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 let drop_winner = matches!(
264 state.status,
265 RunStatus::Merged | RunStatus::Superseded | RunStatus::AlreadyInBase
266 );
267 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#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
283pub struct FetchReport {
284 pub fetched: usize,
286 pub no_origin: usize,
288 pub failed: usize,
290}
291
292pub 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
327pub 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#[derive(Deserialize)]
339struct Meta {
340 status: RunStatus,
341 updated_at: Timestamp,
342}
343
344fn 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
356const MAX_EXTERNAL_MERGE_CHECKS_PER_PASS: usize = 5;
362
363#[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
382pub 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
400fn 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
429const EXTERNAL_MERGE_CURSOR_FILE: &str = "external-merge-cursor";
433
434fn 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
445fn 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
455async 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 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 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 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
569fn 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
592async 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
629pub 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 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
654pub 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
739fn looks_like_a_worktree_bay(name: &str) -> bool {
748 name.len() == 4 && name.bytes().all(|b| b.is_ascii_alphanumeric())
749}
750
751fn 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
779const MIN_ORPHAN_AGE_SECS: u64 = 5 * 60;
790
791fn resolve_id_path(runs: &Path, prefix: &str) -> Result<String> {
794 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
822pub 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 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
855pub fn prune_cache(home: &Path, cache: &Path, limit_bytes: u64) -> Result<Option<Prune>> {
864 crate::cache::maintenance_prune(home, cache, limit_bytes)
865}
866
867pub 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
889pub 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
903pub 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 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 let edge = now - SignedDuration::new(600, 0);
1132 assert!(!due(now, edge, grace));
1133 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 assert_eq!(rotate_from(&ids, 0), ids);
1146
1147 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 assert_eq!(rotate_from(&ids, 7), rotate_from(&ids, 2));
1158
1159 assert_eq!(rotate_from(&[], 0), Vec::<String>::new());
1161 }
1162
1163 #[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 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 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 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 #[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 #[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 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 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 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 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 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 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 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 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 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 #[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 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 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 let judging = "20260801-000000-0001";
1608 write_meta(&runs, judging, "judging", "2026-08-01T00:00:00Z");
1609
1610 let ready_fresh = "20260904-220000-0002";
1617 write_meta(&runs, ready_fresh, "ready", "2026-09-04T22:00:00Z");
1618
1619 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 let due_ready = due_run(&runs, &wt, "20260801-000000-ffff", SCHEMA);
1632
1633 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 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 #[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 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 std::fs::create_dir_all(wt.join("bbbb").join("cand-A")).unwrap();
1733
1734 std::fs::create_dir_all(wt.join("cccc")).unwrap();
1739
1740 std::fs::create_dir_all(wt.join("scratch")).unwrap();
1744
1745 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 #[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 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 #[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 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 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 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 #[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 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 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 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 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 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 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 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 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 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 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 assert!(read_meta(&runs, &superseded_id).is_ok());
2312 assert!(read_meta(&runs, &ready_id).is_ok());
2313 }
2314}