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
592pub async fn fold_unreadable(runs: &Path, worktrees_root: &Path, id: &str) -> Result<Vec<String>> {
608 let resolved = resolve_id_path(runs, id)?;
609 let mut removed = Vec::new();
610 let run_dir = runs.join(&resolved);
611 if run_dir.exists() {
612 std::fs::remove_dir_all(&run_dir)
613 .with_context(|| format!("remove {}", run_dir.display()))?;
614 removed.push(format!("runs/{resolved}"));
615 }
616 let wt = worktrees_root.join(short_of(&resolved));
617 if wt.exists() {
618 crate::git::remove_worktree_from_linked(&wt).await;
619 for e in std::fs::read_dir(&wt).into_iter().flatten().flatten() {
620 crate::git::remove_worktree_from_linked(&e.path()).await;
621 }
622 std::fs::remove_dir_all(&wt).with_context(|| format!("remove {}", wt.display()))?;
623 removed.push(wt.to_string_lossy().into_owned());
624 }
625 Ok(removed)
626}
627
628pub async fn fold_orphaned_worktrees(
667 runs: &Path,
668 worktrees_root: &Path,
669 home: &Path,
670 grace_secs: u64,
671 now: Timestamp,
672) -> usize {
673 let known: std::collections::HashSet<String> = std::fs::read_dir(runs)
674 .into_iter()
675 .flatten()
676 .flatten()
677 .map(|e| e.file_name().to_string_lossy().into_owned())
678 .filter(|name| crate::run::is_run_id(name))
679 .map(|id| short_of(&id).to_owned())
680 .collect();
681
682 let mut folded = 0usize;
683 for entry in std::fs::read_dir(worktrees_root)
684 .into_iter()
685 .flatten()
686 .flatten()
687 {
688 if !entry.path().is_dir() {
689 continue;
690 }
691 let short = entry.file_name().to_string_lossy().into_owned();
692 if !looks_like_a_worktree_bay(&short) {
693 continue;
694 }
695 if known.contains(&short) || crate::daemon::is_working_on_short(home, &short, now) {
696 continue;
697 }
698 let wt = entry.path();
699 if !stale_enough(&wt, grace_secs, now) {
700 continue;
701 }
702 crate::git::remove_worktree_from_linked(&wt).await;
703 for e in std::fs::read_dir(&wt).into_iter().flatten().flatten() {
704 crate::git::remove_worktree_from_linked(&e.path()).await;
705 }
706 match std::fs::remove_dir_all(&wt) {
707 Ok(()) => folded += 1,
708 Err(e) => tracing::warn!(
709 "housekeep: remove orphaned worktree {}: {e:#}",
710 wt.display()
711 ),
712 }
713 }
714 folded
715}
716
717fn looks_like_a_worktree_bay(name: &str) -> bool {
726 name.len() == 4 && name.bytes().all(|b| b.is_ascii_alphanumeric())
727}
728
729fn stale_enough(dir: &Path, grace_secs: u64, now: Timestamp) -> bool {
748 let Ok(modified) = std::fs::metadata(dir).and_then(|m| m.modified()) else {
749 return false;
750 };
751 let Ok(ts) = Timestamp::try_from(modified) else {
752 return false;
753 };
754 due(now, ts, grace_secs.max(MIN_ORPHAN_AGE_SECS))
755}
756
757const MIN_ORPHAN_AGE_SECS: u64 = 5 * 60;
768
769fn resolve_id_path(runs: &Path, prefix: &str) -> Result<String> {
772 if runs.join(prefix).is_dir() && crate::run::is_run_id(prefix) {
777 return Ok(prefix.to_owned());
778 }
779 let mut hits: Vec<String> = Vec::new();
780 for e in std::fs::read_dir(runs).into_iter().flatten().flatten() {
781 if !e.path().is_dir() {
782 continue;
783 }
784 let id = e.file_name().to_string_lossy().into_owned();
785 if crate::run::is_run_id(&id) && (id.starts_with(prefix) || id.ends_with(prefix)) {
786 hits.push(id);
787 }
788 }
789 match hits.len() {
790 1 => Ok(hits.into_iter().next().expect("exactly one hit")),
791 0 => bail!("no run matches `{prefix}`"),
792 _ => bail!(
793 "`{prefix}` matches {} runs: {}",
794 hits.len(),
795 hits.join(", ")
796 ),
797 }
798}
799
800pub fn clear_abandoned_active(state: &mut RunState, home: &Path, now: Timestamp) -> Result<bool> {
815 if crate::daemon::is_working_on(home, &state.id, now) || !state.active_all_overrun(now) {
816 return Ok(false);
817 }
818 state.abandon("fold");
819 state.save_under(home)?;
820 if let Err(e) = Questions::at(home.join("questions")).settle_run(&state.id, state.status) {
828 tracing::warn!("abandon questions for {}: {e:#}", state.id);
829 }
830 Ok(true)
831}
832
833pub fn prune_cache(home: &Path, cache: &Path, limit_bytes: u64) -> Result<Option<Prune>> {
842 crate::cache::maintenance_prune(home, cache, limit_bytes)
843}
844
845pub fn prune_cache_if_over_limit(
855 cfg: &crate::config::Config,
856 home: &Path,
857) -> Result<Option<Prune>> {
858 if cfg.disk.cache_limit_bytes == 0 {
859 return Ok(None);
860 }
861 let Some(cache) = cfg.cache_dir() else {
862 return Ok(None);
863 };
864 prune_cache(home, &cache, cfg.disk.cache_limit_bytes)
865}
866
867pub fn cache_report(cfg: &crate::config::Config) -> Option<(PathBuf, u64, u64)> {
873 let cache = cfg.cache_dir()?;
874 Some((
875 cache.clone(),
876 cache_size(&cache),
877 cfg.disk.cache_limit_bytes,
878 ))
879}
880
881pub fn cache_size(cache: &Path) -> u64 {
883 dir_size(cache)
884}
885
886#[cfg(test)]
887mod tests {
888
889 async fn sh(cwd: &Path, args: &[&str]) {
890 let out = crate::git::git_raw(cwd, args).await.expect("spawn git");
891 assert!(out.ok(), "git {args:?}: {}", out.stderr);
892 }
893
894 async fn commit(cwd: &Path, file: &str, body: &str) {
895 std::fs::write(cwd.join(file), body).unwrap();
896 sh(cwd, &["add", file]).await;
897 sh(
898 cwd,
899 &[
900 "-c",
901 "user.name=t",
902 "-c",
903 "user.email=t@example.com",
904 "commit",
905 "-q",
906 "-m",
907 body,
908 ],
909 )
910 .await;
911 }
912
913 const LONG: std::time::Duration = std::time::Duration::from_secs(30);
914
915 #[tokio::test]
916 async fn fetch_advances_origin_main_and_leaves_the_checkout_alone() {
917 let tmp = tempfile::tempdir().unwrap();
918 let t = tmp.path();
919 let bare = t.join("remote.git");
920 sh(
921 t,
922 &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
923 )
924 .await;
925 let dir = t.join("root/h/o/r");
926 std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
927 sh(
928 t,
929 &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
930 )
931 .await;
932 sh(&dir, &["checkout", "-q", "-b", "main"]).await;
933 commit(&dir, "a.txt", "one").await;
934 sh(&dir, &["push", "-q", "origin", "main"]).await;
935 sh(&dir, &["checkout", "-q", "--detach"]).await;
936 std::fs::write(dir.join("a.txt"), "dirty").unwrap();
937
938 let other = t.join("other");
939 sh(
940 t,
941 &[
942 "clone",
943 "-q",
944 bare.to_str().unwrap(),
945 other.to_str().unwrap(),
946 ],
947 )
948 .await;
949 commit(&other, "b.txt", "two").await;
950 sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
951
952 let rev = |d: PathBuf, r: &'static str| async move {
953 crate::git::git(&d, &["rev-parse", r]).await.unwrap()
954 };
955 let head = rev(dir.clone(), "HEAD").await;
956 let before = rev(dir.clone(), "origin/main").await;
957 let status = crate::git::git(&dir, &["status", "--porcelain"])
958 .await
959 .unwrap();
960
961 let r = fetch_origins(&[t.join("root")], LONG, || false).await;
962 assert_eq!(
963 r,
964 FetchReport {
965 fetched: 1,
966 no_origin: 0,
967 failed: 0
968 }
969 );
970
971 assert_ne!(rev(dir.clone(), "origin/main").await, before);
972 assert_eq!(rev(dir.clone(), "HEAD").await, head);
973 assert_eq!(
974 crate::git::git(&dir, &["status", "--porcelain"])
975 .await
976 .unwrap(),
977 status
978 );
979 assert_eq!(std::fs::read_to_string(dir.join("a.txt")).unwrap(), "dirty");
980 }
981
982 #[tokio::test]
983 async fn fetch_never_writes_a_local_branch_whatever_the_configured_refspec() {
984 let tmp = tempfile::tempdir().unwrap();
985 let t = tmp.path();
986 let bare = t.join("remote.git");
987 sh(
988 t,
989 &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
990 )
991 .await;
992 let dir = t.join("root/h/o/r");
993 std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
994 sh(
995 t,
996 &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
997 )
998 .await;
999 sh(&dir, &["checkout", "-q", "-b", "main"]).await;
1000 commit(&dir, "a.txt", "one").await;
1001 sh(&dir, &["push", "-q", "origin", "main"]).await;
1002 sh(&dir, &["checkout", "-q", "--detach"]).await;
1003 sh(
1005 &dir,
1006 &[
1007 "config",
1008 "remote.origin.fetch",
1009 "+refs/heads/main:refs/heads/main",
1010 ],
1011 )
1012 .await;
1013 let _ = crate::git::git_raw(&dir, &["update-ref", "-d", "refs/remotes/origin/main"]).await;
1014
1015 let other = t.join("other");
1016 sh(
1017 t,
1018 &[
1019 "clone",
1020 "-q",
1021 bare.to_str().unwrap(),
1022 other.to_str().unwrap(),
1023 ],
1024 )
1025 .await;
1026 commit(&other, "b.txt", "two").await;
1027 sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
1028 let remote_tip = crate::git::git(&other, &["rev-parse", "HEAD"])
1029 .await
1030 .unwrap();
1031 let local_main = crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1032 .await
1033 .unwrap();
1034
1035 let r = fetch_origins(&[t.join("root")], LONG, || false).await;
1036 assert_eq!(r.fetched, 1);
1037 assert_eq!(
1038 crate::git::git(&dir, &["rev-parse", "refs/remotes/origin/main"])
1039 .await
1040 .unwrap(),
1041 remote_tip
1042 );
1043 assert_eq!(
1044 crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1045 .await
1046 .unwrap(),
1047 local_main
1048 );
1049 }
1050
1051 #[tokio::test]
1052 async fn fetch_skips_no_origin_and_survives_an_unreachable_origin() {
1053 let tmp = tempfile::tempdir().unwrap();
1054 let t = tmp.path();
1055 let root = t.join("root");
1056 let bare = t.join("remote.git");
1057 sh(t, &["init", "-q", "--bare", bare.to_str().unwrap()]).await;
1058
1059 let none = root.join("h/o/none");
1060 let dead = root.join("h/o/dead");
1061 let good = root.join("h/o/good");
1062 for d in [&none, &dead, &good] {
1063 std::fs::create_dir_all(d).unwrap();
1064 sh(d, &["init", "-q"]).await;
1065 }
1066 sh(
1067 &dead,
1068 &[
1069 "remote",
1070 "add",
1071 "origin",
1072 t.join("missing").to_str().unwrap(),
1073 ],
1074 )
1075 .await;
1076 sh(&good, &["remote", "add", "origin", bare.to_str().unwrap()]).await;
1077
1078 let r = fetch_origins(&[root], LONG, || false).await;
1079 assert_eq!(
1080 r,
1081 FetchReport {
1082 fetched: 1,
1083 no_origin: 1,
1084 failed: 1
1085 }
1086 );
1087 }
1088 use super::*;
1089 use crate::config::Disk;
1090 use std::fs;
1091
1092 fn ts(s: &str) -> Timestamp {
1093 s.parse().expect("rfc3339")
1094 }
1095
1096 fn block_on<F: std::future::Future>(f: F) -> F::Output {
1097 tokio::runtime::Runtime::new().expect("runtime").block_on(f)
1098 }
1099
1100 #[test]
1101 fn a_run_is_due_after_its_grace_and_not_before() {
1102 let now = ts("2026-09-05T00:00:00Z");
1103 let grace = 600;
1104 let old = now - SignedDuration::new(601, 0);
1105 let fresh = now - SignedDuration::new(599, 0);
1106 assert!(due(now, old, grace));
1107 assert!(!due(now, fresh, grace));
1108 let edge = now - SignedDuration::new(600, 0);
1110 assert!(!due(now, edge, grace));
1111 assert!(due(now, old, 0));
1113 }
1114
1115 #[test]
1116 fn rotate_from_moves_the_starting_point_as_the_cursor_advances() {
1117 let ids: Vec<String> = ["a", "b", "c", "d", "e"]
1118 .iter()
1119 .map(|s| s.to_string())
1120 .collect();
1121
1122 assert_eq!(rotate_from(&ids, 0), ids);
1124
1125 let at_2 = rotate_from(&ids, 2);
1128 assert_eq!(at_2, vec!["c", "d", "e", "a", "b"]);
1129 for id in &ids {
1130 assert!(at_2.contains(id));
1131 }
1132
1133 assert_eq!(rotate_from(&ids, 7), rotate_from(&ids, 2));
1136
1137 assert_eq!(rotate_from(&[], 0), Vec::<String>::new());
1139 }
1140
1141 #[test]
1148 fn advancing_the_cursor_by_the_window_size_gives_every_id_a_turn() {
1149 let ids: Vec<String> = (0..12).map(|n| format!("run-{n}")).collect();
1150 let cap = 5usize;
1151 let mut cursor = 0usize;
1152 let mut ever_seen: std::collections::HashSet<String> = std::collections::HashSet::new();
1153 for _ in 0..ids.len() {
1154 let window: Vec<String> = rotate_from(&ids, cursor).into_iter().take(cap).collect();
1155 ever_seen.extend(window.iter().cloned());
1156 cursor = (cursor + window.len()) % ids.len();
1157 }
1158 assert_eq!(
1159 ever_seen.len(),
1160 ids.len(),
1161 "every id must be checked at least once across a full rotation, whatever \
1162 the cadence between passes"
1163 );
1164 }
1165
1166 #[test]
1167 fn the_external_merge_cursor_round_trips_through_a_files_absence() {
1168 let dir = tempfile::tempdir().unwrap();
1169 let home = dir.path();
1170
1171 assert_eq!(read_external_merge_cursor(home), 0);
1173
1174 write_external_merge_cursor(home, 7);
1175 assert_eq!(read_external_merge_cursor(home), 7);
1176
1177 std::fs::write(home.join(EXTERNAL_MERGE_CURSOR_FILE), "not a number").unwrap();
1180 assert_eq!(read_external_merge_cursor(home), 0);
1181 }
1182
1183 #[test]
1184 fn a_run_qualifies_for_an_external_merge_check_only_when_blocked_unmerged_and_due() {
1185 let now = ts("2026-09-05T00:00:00Z");
1186 let grace = 600;
1187 let old = now - SignedDuration::new(601, 0);
1188 let fresh = now - SignedDuration::new(599, 0);
1189
1190 assert!(
1191 eligible_for_external_merge_check(RunStatus::Blocked, false, old, now, grace),
1192 "blocked, unmerged, and past its grace period is exactly the run this exists for"
1193 );
1194 assert!(
1195 !eligible_for_external_merge_check(RunStatus::Blocked, false, fresh, now, grace),
1196 "an operator mid-fix deserves the same grace window `fold_due` gives before \
1197 the janitor starts asking GitHub about it"
1198 );
1199 assert!(
1200 !eligible_for_external_merge_check(RunStatus::Blocked, true, old, now, grace),
1201 "a run `land::land` already recorded a merge outcome for has its own answer \
1202 already; this check is only for a run with nothing recorded at all"
1203 );
1204 assert!(
1205 !eligible_for_external_merge_check(RunStatus::Stalled, false, old, now, grace),
1206 "stalled is not blocked - it means the tally never reached quorum, which a \
1207 pull request cannot fix"
1208 );
1209 assert!(
1210 !eligible_for_external_merge_check(RunStatus::Ready, false, old, now, grace),
1211 "ready has nothing to correct - it was never landed by design"
1212 );
1213 assert!(
1214 !eligible_for_external_merge_check(RunStatus::Superseded, false, old, now, grace),
1215 "a superseded run's task already has its answer from a later attempt; \
1216 there is nothing left for GitHub to confirm here"
1217 );
1218 }
1219
1220 fn write_blocked_run(runs: &Path, id: &str, updated_at: Timestamp) {
1226 use crate::run::{Candidate, Tally};
1227 use std::collections::BTreeMap;
1228
1229 let mut state = RunState::new(
1230 PathBuf::from("/nonexistent/repo"),
1231 "main".to_owned(),
1232 "0000000000000000000000000000000000000000".to_owned(),
1233 String::new(),
1234 crate::config::Config::default(),
1235 );
1236 state.id = id.to_owned();
1237 state.status = RunStatus::Blocked;
1238 state.updated_at = updated_at;
1239 state.candidates.push(Candidate {
1240 index: 0,
1241 label: 'A',
1242 agent: "agent".to_owned(),
1243 branch: format!("magi/{id}/A"),
1244 worktree: PathBuf::from("/nonexistent/repo"),
1245 summary: String::new(),
1246 stat: String::new(),
1247 files: 0,
1248 commits: 0,
1249 empty: false,
1250 failed: None,
1251 verified_noop: None,
1252 duration_ms: 0,
1253 folded: false,
1254 });
1255 state.tally = Some(Tally {
1256 first_choice: BTreeMap::from([('A', 1)]),
1257 borda: BTreeMap::new(),
1258 winner: 'A',
1259 rankings: 1,
1260 unanimous_initial: true,
1261 deliberated: false,
1262 changed_votes: 0,
1263 unanimous_final: true,
1264 tie_break: None,
1265 judges: 1,
1266 present: 1,
1267 quorum: 1,
1268 met_quorum: true,
1269 uncontested: None,
1270 });
1271 std::fs::create_dir_all(runs.join(id)).unwrap();
1272 std::fs::write(
1273 runs.join(id).join("run.json"),
1274 serde_json::to_string_pretty(&state).unwrap(),
1275 )
1276 .unwrap();
1277 }
1278
1279 #[tokio::test]
1284 async fn reconcile_external_merges_leaves_ineligible_and_unconfirmable_runs_alone() {
1285 let dir = tempfile::tempdir().unwrap();
1286 let runs = dir.path().join("runs");
1287 let home = dir.path().to_path_buf();
1288 std::fs::create_dir_all(&runs).unwrap();
1289
1290 let now = ts("2026-09-05T00:00:00Z");
1291 let disk = Disk {
1292 fold_grace_secs: 600,
1293 ..Disk::default()
1294 };
1295
1296 let due_id = "20260905-000000-blkd";
1297 write_blocked_run(&runs, due_id, ts("2026-08-01T00:00:00Z"));
1298
1299 let fresh_id = "20260905-000000-fres";
1300 write_blocked_run(&runs, fresh_id, now);
1301
1302 let live_id = "20260905-000000-live";
1303 write_blocked_run(&runs, live_id, ts("2026-08-01T00:00:00Z"));
1304 let mut status = crate::daemon::Status::new();
1305 status.current = vec![crate::daemon::Current {
1306 task: "20260905-000000-task".to_owned(),
1307 run: live_id.to_owned(),
1308 }];
1309 status.updated_at = now;
1310 crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1311
1312 let reconciled = reconcile_external_merges(&runs, &home, &disk, now).await;
1313 assert_eq!(
1314 reconciled, 0,
1315 "an unreachable repository can never be confirmed merged"
1316 );
1317 for id in [due_id, fresh_id, live_id] {
1318 assert_eq!(
1319 read_meta(&runs, id).unwrap().status,
1320 RunStatus::Blocked,
1321 "{id} must be left exactly as it was found"
1322 );
1323 }
1324 let notices = crate::notices::Notices::at(home.join("notifications")).list();
1325 let due_notice = notices
1326 .iter()
1327 .find(|n| n.key == format!("merged-unrecorded:{due_id}"))
1328 .expect("notice for due_id must be raised");
1329 assert_eq!(
1330 due_notice.link,
1331 Some(crate::notices::Link::Run {
1332 id: due_id.to_owned(),
1333 })
1334 );
1335 }
1336
1337 #[tokio::test]
1347 async fn reconcile_external_merges_covers_a_larger_fleet_across_repeated_passes_at_one_instant()
1348 {
1349 let dir = tempfile::tempdir().unwrap();
1350 let runs = dir.path().join("runs");
1351 let home = dir.path().to_path_buf();
1352 std::fs::create_dir_all(&runs).unwrap();
1353
1354 let now = ts("2026-09-05T00:00:00Z");
1355 let disk = Disk {
1356 fold_grace_secs: 600,
1357 ..Disk::default()
1358 };
1359 let old = ts("2026-08-01T00:00:00Z");
1360
1361 let ids: Vec<String> = (0..8).map(|n| format!("20260905-000000-r{n:03}")).collect();
1362 for id in &ids {
1363 write_blocked_run(&runs, id, old);
1364 }
1365
1366 let notices = crate::notices::Notices::at(home.join("notifications"));
1367
1368 reconcile_external_merges(&runs, &home, &disk, now).await;
1374 let after_first = notices.list().len();
1375 assert_eq!(
1376 after_first, MAX_EXTERNAL_MERGE_CHECKS_PER_PASS,
1377 "the first pass checks exactly one cap's worth"
1378 );
1379
1380 reconcile_external_merges(&runs, &home, &disk, now).await;
1381 let after_second = notices.list().len();
1382 assert_eq!(
1383 after_second,
1384 ids.len(),
1385 "a second pass at the same instant must still reach every id the \
1386 first pass had no room for, not repeat the same cap's worth"
1387 );
1388 }
1389
1390 #[test]
1391 fn the_meta_reader_is_tolerant_of_everything_except_the_deciders() {
1392 let dir = tempfile::tempdir().unwrap();
1393 let runs = dir.path().join("runs");
1394 let id = "20260905-000000-abcd";
1395 std::fs::create_dir_all(runs.join(id)).unwrap();
1396 std::fs::write(
1397 runs.join(id).join("run.json"),
1398 r#"{"schema": 99, "id": "20260905-000000-abcd", "updated_at": "2026-09-05T00:00:00Z", "status": "ready", "junk_from_another_build": [1, 2, 3]}"#,
1399 )
1400 .unwrap();
1401 let meta = read_meta(&runs, id).expect("readable");
1402 assert_eq!(meta.status, RunStatus::Ready);
1403 assert_eq!(meta.updated_at, ts("2026-09-05T00:00:00Z"));
1404 assert!(read_meta(&runs, "nope").is_err(), "missing file unreadable");
1405 std::fs::write(runs.join(id).join("run.json"), "not json at all").unwrap();
1406 assert!(read_meta(&runs, id).is_err(), "garbage unreadable");
1407 }
1408
1409 #[test]
1410 fn fold_unreadable_releases_run_dir_and_worktrees() {
1411 let dir = tempfile::tempdir().unwrap();
1412 let runs = dir.path().join("runs");
1413 let wt = dir.path().join("wt");
1414 let id = "20260905-000000-abcd";
1415 std::fs::create_dir_all(runs.join(id)).unwrap();
1416 std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1417 std::fs::create_dir_all(wt.join("abcd")).unwrap();
1418 std::fs::write(wt.join("abcd").join("leftover"), b"x").unwrap();
1419
1420 let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold");
1421 assert_eq!(removed.len(), 2);
1422 assert!(!runs.join(id).exists(), "run dir gone");
1423 assert!(!wt.join("abcd").exists(), "worktrees gone");
1424
1425 std::fs::create_dir_all(runs.join(id)).unwrap();
1427 std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1428 std::fs::create_dir_all(wt.join("abcd")).unwrap();
1429 std::fs::write(wt.join("abcd").join("leftover"), b"x").unwrap();
1430 let removed = block_on(fold_unreadable(&runs, &wt, "20260905")).expect("by prefix");
1431 assert_eq!(removed.len(), 2);
1432 assert!(
1436 block_on(fold_unreadable(&runs, &wt, id)).is_err(),
1437 "a run already gone cannot be resolved again"
1438 );
1439 }
1440
1441 #[test]
1442 fn prune_cache_sheds_the_oldest_generation_until_it_fits() {
1443 let home = tempfile::tempdir().unwrap();
1444 let dir = tempfile::tempdir().unwrap();
1445 fs::write(dir.path().join("old"), b"xx").unwrap();
1448 fs::write(dir.path().join("new"), b"yy").unwrap();
1449 touch(&dir.path().join("old"), 1_000_000);
1450 touch(&dir.path().join("new"), 2_000_000);
1451
1452 let out = prune_cache(home.path(), dir.path(), 2)
1453 .expect("prune")
1454 .expect("the cache is free");
1455 assert_eq!(out.files, 1, "one deletion is enough to reach the cap");
1456 assert_eq!(out.remaining, 2);
1457 assert!(!dir.path().join("old").exists(), "the older file went");
1458 assert!(dir.path().join("new").exists(), "the newer one stayed");
1459
1460 let tied = tempfile::tempdir().unwrap();
1465 fs::write(tied.path().join("big"), b"xxxx").unwrap();
1466 fs::write(tied.path().join("small"), b"yy").unwrap();
1467 touch(&tied.path().join("big"), 1_000_000);
1468 touch(&tied.path().join("small"), 1_000_000);
1469 let out = prune_cache(home.path(), tied.path(), 2)
1470 .expect("prune")
1471 .expect("the cache is free");
1472 assert_eq!(out.files, 1, "the big one alone gets under the cap");
1473 assert_eq!(out.remaining, 2);
1474 assert!(tied.path().join("small").exists());
1475 }
1476
1477 #[test]
1478 fn prune_cache_if_over_limit_resolves_the_opt_outs_before_ever_measuring() {
1479 let home = tempfile::tempdir().unwrap();
1480 let dir = tempfile::tempdir().unwrap();
1481 fs::write(dir.path().join("big"), vec![0u8; 10]).unwrap();
1482
1483 let mut cfg = crate::config::Config::default();
1484 cfg.verify.gate = vec![format!(
1485 "CARGO_TARGET_DIR={} cargo make check",
1486 dir.path().display()
1487 )];
1488
1489 cfg.disk.cache_limit_bytes = 0;
1492 assert_eq!(
1493 prune_cache_if_over_limit(&cfg, home.path()).unwrap(),
1494 None,
1495 "a zero cap must not even look at the directory"
1496 );
1497 assert!(dir.path().join("big").exists());
1498
1499 let mut no_cache = crate::config::Config::default();
1502 no_cache.disk.cache_limit_bytes = 1;
1503 assert_eq!(
1504 prune_cache_if_over_limit(&no_cache, home.path()).unwrap(),
1505 None
1506 );
1507
1508 cfg.disk.cache_limit_bytes = 1;
1511 let pruned = prune_cache_if_over_limit(&cfg, home.path())
1512 .unwrap()
1513 .expect("a real cache dir over its cap prunes");
1514 assert_eq!(pruned.files, 1);
1515 assert!(!dir.path().join("big").exists());
1516 }
1517
1518 fn touch(path: &Path, secs: u64) {
1521 let f = fs::File::options().write(true).open(path).unwrap();
1522 f.set_times(fs::FileTimes::new().set_modified(
1523 std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs),
1524 ))
1525 .unwrap();
1526 }
1527
1528 #[test]
1531 fn fold_unreadable_clears_a_run_whose_state_never_landed() {
1532 let dir = tempfile::tempdir().unwrap();
1533 let runs = dir.path().join("runs");
1534 let wt = dir.path().join("wt");
1535 let id = "20260904-014540-88c0";
1536 std::fs::create_dir_all(runs.join(id)).unwrap();
1537 std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1538
1539 let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold by id");
1540 assert_eq!(removed, vec![format!("runs/{id}")]);
1541 assert!(!runs.join(id).exists(), "record gone");
1542
1543 std::fs::create_dir_all(runs.join(id)).unwrap();
1545 std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1546 assert!(
1547 block_on(fold_unreadable(&runs, &wt, "88c0")).is_ok(),
1548 "by prefix"
1549 );
1550
1551 std::fs::create_dir_all(runs.join("scratch")).unwrap();
1553 assert!(
1554 block_on(fold_unreadable(&runs, &wt, "scratch")).is_err(),
1555 "a stray directory is not a run"
1556 );
1557 }
1558
1559 #[test]
1560 fn fold_due_folds_terminal_runs_of_any_schema_but_leaves_genuinely_unreadable_ones() {
1561 let dir = tempfile::tempdir().unwrap();
1562 let runs = dir.path().join("runs");
1563 let wt = dir.path().join("wt");
1564 let home = dir.path().to_path_buf();
1565 let disk = Disk::default();
1566 let now = ts("2026-09-05T00:00:00Z");
1567
1568 let judging = "20260801-000000-0001";
1570 write_meta(&runs, judging, "judging", "2026-08-01T00:00:00Z");
1571
1572 let ready_fresh = "20260904-220000-0002";
1579 write_meta(&runs, ready_fresh, "ready", "2026-09-04T22:00:00Z");
1580
1581 let garbage = "20260901-000000-0004";
1587 std::fs::create_dir_all(runs.join(garbage)).unwrap();
1588 std::fs::write(runs.join(garbage).join("run.json"), "not json").unwrap();
1589 std::fs::create_dir_all(wt.join("0004")).unwrap();
1590
1591 let due_ready = due_run(&runs, &wt, "20260801-000000-ffff", SCHEMA);
1594
1595 let due_old_schema = due_run(&runs, &wt, "20260801-000000-eeee", SCHEMA - 1);
1600
1601 let (folded, unreadable) =
1602 block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
1603 assert_eq!(
1604 folded, 2,
1605 "both due, parseable runs fold regardless of their schema number"
1606 );
1607 assert_eq!(
1608 unreadable, 1,
1609 "only the run with broken JSON counts as unreadable"
1610 );
1611 assert!(runs.join(judging).exists(), "runnable never folded");
1612 assert!(runs.join(ready_fresh).exists(), "fresh never folded");
1613 assert!(runs.join(garbage).exists(), "unreadable record kept");
1614 assert!(wt.join("0004").exists(), "unreadable worktree kept");
1615 assert!(
1616 runs.join(&due_ready).exists(),
1617 "folding drops worktrees, not the record"
1618 );
1619 assert!(
1620 runs.join(&due_old_schema).exists(),
1621 "an old-schema record survives its fold exactly like a current one"
1622 );
1623 for id in [&due_ready, &due_old_schema] {
1629 let saved = read_meta(&runs, id).expect("folded run still parses");
1630 assert_ne!(
1631 saved.updated_at,
1632 ts("2026-08-01T00:00:00Z"),
1633 "fold_run must have saved the updated state back through the \
1634 `runs` directory this test passed to fold_due"
1635 );
1636 }
1637 }
1638
1639 #[tokio::test]
1645 async fn housekeep_leaves_everything_alone_when_auto_fold_is_disabled() {
1646 let dir = tempfile::tempdir().unwrap();
1647 let runs = dir.path().join("runs");
1648 let wt = dir.path().join("wt");
1649 let home = dir.path().to_path_buf();
1650 crate::run::set_home(dir.path().to_path_buf());
1651
1652 let due_id = due_run(&runs, &wt, "20260801-000000-abcd", SCHEMA);
1653 std::fs::create_dir_all(wt.join("orphan").join("cand-A")).unwrap();
1654
1655 let mut cfg = crate::config::Config::default();
1656 cfg.disk.auto_fold = false;
1657 cfg.disk.cache_limit_bytes = 0;
1658
1659 let out = housekeep(&cfg, &home, &wt, &dir.path().join("repo"), Timestamp::now()).await;
1660
1661 assert_eq!(out.folded, 0);
1662 assert_eq!(out.unreadable, 0);
1663 assert_eq!(out.orphaned_worktrees, 0);
1664 assert!(
1665 runs.join(&due_id).exists(),
1666 "a due run's record survives untouched"
1667 );
1668 assert!(
1669 wt.join("orphan").exists(),
1670 "an orphaned worktree survives untouched: the reclaim pass never ran"
1671 );
1672 }
1673
1674 #[test]
1675 fn fold_orphaned_worktrees_removes_only_worktrees_no_run_claims_and_none_in_flight() {
1676 let dir = tempfile::tempdir().unwrap();
1677 let runs = dir.path().join("runs");
1678 let wt = dir.path().join("wt");
1679 let home = dir.path().to_path_buf();
1680
1681 write_meta(
1684 &runs,
1685 "20260801-000000-aaaa",
1686 "ready",
1687 "2026-08-01T00:00:00Z",
1688 );
1689 std::fs::create_dir_all(wt.join("aaaa").join("cand-A")).unwrap();
1690
1691 std::fs::create_dir_all(wt.join("bbbb").join("cand-A")).unwrap();
1695
1696 std::fs::create_dir_all(wt.join("cccc")).unwrap();
1701
1702 std::fs::create_dir_all(wt.join("scratch")).unwrap();
1706
1707 let now = Timestamp::now() + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1713 let mut status = crate::daemon::Status::new();
1714 status.current = vec![crate::daemon::Current {
1715 task: "20260905-000000-t111".to_owned(),
1716 run: "20260905-000000-cccc".to_owned(),
1717 }];
1718 status.updated_at = now;
1719 crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1720
1721 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1722 assert_eq!(
1723 folded, 1,
1724 "only the truly orphaned, idle, bay-shaped worktree is removed"
1725 );
1726 assert!(wt.join("aaaa").exists(), "claimed by a run record");
1727 assert!(!wt.join("bbbb").exists(), "orphaned and idle: reclaimed");
1728 assert!(wt.join("cccc").exists(), "a run in flight is never touched");
1729 assert!(
1730 wt.join("scratch").exists(),
1731 "not shaped like a worktree bay, so never a reclaim target"
1732 );
1733 }
1734
1735 #[test]
1743 fn fold_orphaned_worktrees_leaves_a_freshly_created_bay_alone() {
1744 let dir = tempfile::tempdir().unwrap();
1745 let runs = dir.path().join("runs");
1746 let wt = dir.path().join("wt");
1747 let home = dir.path().to_path_buf();
1748
1749 std::fs::create_dir_all(wt.join("dddd").join("under-review")).unwrap();
1750
1751 let now = Timestamp::now();
1752 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 6 * 60 * 60, now));
1753 assert_eq!(
1754 folded, 0,
1755 "too fresh to tell apart from a run still being set up"
1756 );
1757 assert!(wt.join("dddd").exists());
1758 }
1759
1760 fn open_question(store: &Questions, run: &str) -> crate::ask::Question {
1762 let mut q = crate::ask::Question::new(
1763 run.to_owned(),
1764 "implement".to_owned(),
1765 "impl-A".to_owned(),
1766 "Which storage backend should the cache use?".to_owned(),
1767 String::new(),
1768 vec!["SQLite".to_owned(), "Redis".to_owned()],
1769 );
1770 store.put(&mut q).unwrap();
1771 q
1772 }
1773
1774 #[test]
1780 fn a_finished_runs_open_question_is_swept_up() {
1781 let dir = tempfile::tempdir().unwrap();
1782 let runs = dir.path().join("runs");
1783 let store = Questions::at(dir.path().join("questions"));
1784
1785 let failed = "20260908-205802-c9eb";
1786 write_meta(&runs, failed, "failed", "2026-09-08T20:58:02Z");
1787 let failed_q = open_question(&store, failed);
1788
1789 let merged = "20260908-205501-ca67";
1790 write_meta(&runs, merged, "merged", "2026-09-08T20:55:01Z");
1791 let merged_q = open_question(&store, merged);
1792
1793 let n = abandon_settled_questions(&store, &runs);
1794 assert_eq!(n, 2, "both dead runs' questions are swept in one pass");
1795
1796 for (id, run) in [(&failed_q.id, failed), (&merged_q.id, merged)] {
1797 let back = store.get(id).unwrap();
1798 assert!(!back.status.open(), "{run} is done; nobody reads an answer");
1799 assert!(back.detail.contains(run), "{}", back.detail);
1800 }
1801 }
1802
1803 #[test]
1804 fn a_still_alive_runs_open_question_survives_the_sweep() {
1805 let dir = tempfile::tempdir().unwrap();
1806 let runs = dir.path().join("runs");
1807 let store = Questions::at(dir.path().join("questions"));
1808
1809 for (id, status) in [
1813 ("20260908-000000-b10c", "blocked"),
1814 ("20260908-000000-5ta1", "stalled"),
1815 ("20260908-000000-jud6", "judging"),
1816 ] {
1817 write_meta(&runs, id, status, "2026-09-08T00:00:00Z");
1818 let q = open_question(&store, id);
1819
1820 let n = abandon_settled_questions(&store, &runs);
1821 assert_eq!(n, 0, "{status} run is not done; nothing to sweep");
1822 assert!(
1823 store.get(&q.id).unwrap().status.open(),
1824 "{status} run's question must still be waiting"
1825 );
1826 }
1827 }
1828
1829 #[test]
1830 fn the_sweep_leaves_an_answered_question_and_an_unreadable_run_alone() {
1831 let dir = tempfile::tempdir().unwrap();
1832 let runs = dir.path().join("runs");
1833 let store = Questions::at(dir.path().join("questions"));
1834
1835 let done = "20260908-000000-answ";
1838 write_meta(&runs, done, "failed", "2026-09-08T00:00:00Z");
1839 let mut answered = open_question(&store, done);
1840 answered
1841 .answer(crate::ask::Answer::Choice("SQLite".to_owned()))
1842 .unwrap();
1843 store.put(&mut answered).unwrap();
1844
1845 let gone = "20260908-000000-gone";
1847 let orphan = open_question(&store, gone);
1848
1849 assert_eq!(abandon_settled_questions(&store, &runs), 0);
1850 assert_eq!(
1851 store.get(&answered.id).unwrap().status,
1852 crate::ask::QuestionStatus::Answered,
1853 "a real answer is never overwritten by a sweep"
1854 );
1855 assert!(
1856 store.get(&orphan.id).unwrap().status.open(),
1857 "a run this sweep cannot read is left exactly as it was, not guessed at"
1858 );
1859 }
1860
1861 #[test]
1870 fn fold_orphaned_worktrees_floors_a_zero_grace_at_the_race_safe_minimum() {
1871 let dir = tempfile::tempdir().unwrap();
1872 let runs = dir.path().join("runs");
1873 let wt = dir.path().join("wt");
1874 let home = dir.path().to_path_buf();
1875
1876 std::fs::create_dir_all(wt.join("eeee").join("under-review")).unwrap();
1877
1878 let now = Timestamp::now();
1880 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1881 assert_eq!(
1882 folded, 0,
1883 "a zero grace must not defeat the race-safety floor"
1884 );
1885 assert!(wt.join("eeee").exists());
1886
1887 let later = now + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1890 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, later));
1891 assert_eq!(folded, 1, "old enough now, regardless of the zero grace");
1892 assert!(!wt.join("eeee").exists());
1893 }
1894
1895 #[test]
1896 fn clear_abandoned_active_only_acts_once_dead_and_overrun() {
1897 let dir = tempfile::tempdir().unwrap();
1898 crate::run::set_home(dir.path().to_path_buf());
1903 let home = dir.path().to_path_buf();
1904 let now = ts("2026-09-14T12:00:00Z");
1905 let overrun_seat = || crate::run::ActiveSeat {
1906 node: "implement".to_owned(),
1907 started_at: now - SignedDuration::new(21_000, 0),
1908 timeout_secs: 3_600,
1909 attempt: 0,
1910 task: None,
1911 command: None,
1912 index: None,
1913 total: None,
1914 };
1915
1916 let mut state = RunState::new(
1917 PathBuf::from("/repo"),
1918 "main".to_owned(),
1919 "abc1234".to_owned(),
1920 "fixture".to_owned(),
1921 crate::config::Config::default(),
1922 );
1923 state.status = RunStatus::Implementing;
1924 state.active.insert("impl-A".to_owned(), overrun_seat());
1925
1926 let mut fresh = state.clone();
1929 fresh.active.insert(
1930 "impl-B".to_owned(),
1931 crate::run::ActiveSeat {
1932 node: "implement".to_owned(),
1933 started_at: now,
1934 timeout_secs: 3_600,
1935 attempt: 0,
1936 task: None,
1937 command: None,
1938 index: None,
1939 total: None,
1940 },
1941 );
1942 assert!(!clear_abandoned_active(&mut fresh, &home, now).unwrap());
1943 assert!(!fresh.active.is_empty());
1944 assert_eq!(fresh.status, RunStatus::Implementing);
1945
1946 let store = Questions::at(home.join("questions"));
1947 let q = open_question(&store, &state.id);
1948
1949 assert!(clear_abandoned_active(&mut state, &home, now).unwrap());
1950 assert!(state.active.is_empty());
1951 assert_eq!(state.status, RunStatus::Failed);
1952 assert!(
1953 !store.get(&q.id).unwrap().status.open(),
1954 "the abandoned seat's own open question must not keep badging the \
1955 operator until some later daemon startup notices it"
1956 );
1957 }
1958
1959 fn write_meta(runs: &Path, id: &str, status: &str, updated_at: &str) {
1961 let day = &updated_at[..10];
1962 std::fs::create_dir_all(runs.join(id)).unwrap();
1963 let body = format!(
1964 r#"{{"schema": {SCHEMA}, "id": "{id}", "repo": "/nonexistent/repo", "base_branch": "main", "base_commit": "0000000000000000000000000000000000000000", "instruction": "", "created_at": "{day}T00:00:00Z", "updated_at": "{updated_at}", "status": "{status}", "seed": 1}}"#
1965 );
1966 std::fs::write(runs.join(id).join("run.json"), body).unwrap();
1967 }
1968
1969 fn due_run(runs: &Path, wt: &Path, id: &str, schema: u32) -> String {
1979 due_run_with_status(runs, wt, id, schema, RunStatus::Ready)
1980 }
1981
1982 fn due_run_with_status(
1985 runs: &Path,
1986 wt: &Path,
1987 id: &str,
1988 schema: u32,
1989 status: RunStatus,
1990 ) -> String {
1991 let mut config = crate::config::Config::default();
1992 config.graph.worktree_root = Some(wt.to_path_buf());
1993 let mut state = RunState::new(
1994 PathBuf::from("/nonexistent/repo"),
1995 "main".to_owned(),
1996 "0000000000000000000000000000000000000000".to_owned(),
1997 String::new(),
1998 config,
1999 );
2000 state.id = id.to_owned();
2001 state.status = status;
2002 state.updated_at = ts("2026-08-01T00:00:00Z");
2003 let mut value = serde_json::to_value(&state).unwrap();
2004 value["schema"] = serde_json::json!(schema);
2005 std::fs::create_dir_all(runs.join(id)).unwrap();
2006 std::fs::write(
2007 runs.join(id).join("run.json"),
2008 serde_json::to_string_pretty(&value).unwrap(),
2009 )
2010 .unwrap();
2011 id.to_owned()
2012 }
2013
2014 #[test]
2015 fn fold_due_folds_a_superseded_run_same_as_any_other_terminal_one() {
2016 let dir = tempfile::tempdir().unwrap();
2017 let runs = dir.path().join("runs");
2018 let wt = dir.path().join("wt");
2019 let home = dir.path().to_path_buf();
2020 let disk = Disk::default();
2021 let now = ts("2026-09-05T00:00:00Z");
2022
2023 let superseded = due_run_with_status(
2024 &runs,
2025 &wt,
2026 "20260801-000000-cccc",
2027 SCHEMA,
2028 RunStatus::Superseded,
2029 );
2030
2031 let (folded, unreadable) =
2032 block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
2033 assert_eq!(
2034 folded, 1,
2035 "a superseded run has nothing left for a human to check, so it folds \
2036 exactly like a merged one"
2037 );
2038 assert_eq!(unreadable, 0);
2039 assert!(
2040 read_meta(&runs, &superseded).is_ok(),
2041 "folding drops the worktree, not the record"
2042 );
2043 }
2044
2045 fn init_repo(dir: &Path) {
2048 use crate::proc::Quiet as _;
2049 let run = |args: &[&str]| {
2050 let out = std::process::Command::new("git")
2051 .args(args)
2052 .current_dir(dir)
2053 .quiet()
2054 .output()
2055 .expect("spawn git");
2056 assert!(
2057 out.status.success(),
2058 "git {args:?} failed: {}",
2059 String::from_utf8_lossy(&out.stderr)
2060 );
2061 };
2062 run(&["init", "-b", "main"]);
2063 run(&["config", "user.name", "magi test"]);
2064 run(&["config", "user.email", "magi@example.com"]);
2065 std::fs::write(dir.join("README.md"), "# fixture\n").unwrap();
2066 run(&["add", "-A"]);
2067 run(&["commit", "-m", "init"]);
2068 }
2069
2070 async fn due_run_with_winner_worktree(
2074 runs: &Path,
2075 wt: &Path,
2076 repo: &Path,
2077 id: &str,
2078 status: RunStatus,
2079 ) -> (String, PathBuf) {
2080 use crate::run::{Candidate, Tally};
2081 use std::collections::BTreeMap;
2082
2083 let mut config = crate::config::Config::default();
2084 config.graph.worktree_root = Some(wt.to_path_buf());
2085 let mut state = RunState::new(
2086 repo.to_path_buf(),
2087 "main".to_owned(),
2088 "0000000000000000000000000000000000000000".to_owned(),
2089 String::new(),
2090 config,
2091 );
2092 state.id = id.to_owned();
2093 state.status = status;
2094 state.updated_at = ts("2026-08-01T00:00:00Z");
2095
2096 let winner_wt = state.worktree_root().join("cand-A");
2097 crate::git::worktree_add_branch(repo, &winner_wt, &format!("magi/{id}/A"), "main")
2098 .await
2099 .expect("winner worktree");
2100
2101 state.candidates.push(Candidate {
2102 index: 0,
2103 label: 'A',
2104 agent: "agent".to_owned(),
2105 branch: format!("magi/{id}/A"),
2106 worktree: winner_wt.clone(),
2107 summary: String::new(),
2108 stat: String::new(),
2109 files: 0,
2110 commits: 0,
2111 empty: false,
2112 failed: None,
2113 verified_noop: None,
2114 duration_ms: 0,
2115 folded: false,
2116 });
2117 state.tally = Some(Tally {
2118 first_choice: BTreeMap::from([('A', 1)]),
2119 borda: BTreeMap::new(),
2120 winner: 'A',
2121 rankings: 1,
2122 unanimous_initial: true,
2123 deliberated: false,
2124 changed_votes: 0,
2125 unanimous_final: true,
2126 tie_break: None,
2127 judges: 1,
2128 present: 1,
2129 quorum: 1,
2130 met_quorum: true,
2131 uncontested: None,
2132 });
2133 std::fs::create_dir_all(runs.join(id)).unwrap();
2134 std::fs::write(
2135 runs.join(id).join("run.json"),
2136 serde_json::to_string_pretty(&state).unwrap(),
2137 )
2138 .unwrap();
2139 (id.to_owned(), winner_wt)
2140 }
2141
2142 #[tokio::test]
2143 async fn recovery_retains_historical_sources_after_success_and_automatic_cleanup() {
2144 let dir = tempfile::tempdir().unwrap();
2145 let home = dir.path();
2146 let runs = home.join("runs");
2147 let wt = home.join("wt");
2148 let repo = home.join("repo");
2149 std::fs::create_dir_all(&repo).unwrap();
2150 init_repo(&repo);
2151 let mut task = crate::queue::Task::new(
2152 "recover".into(),
2153 "keep measured behavior".into(),
2154 repo.clone(),
2155 crate::queue::Source::Human,
2156 );
2157 let mut sources = Vec::new();
2158 for (id, status) in [
2159 ("20260801-000000-old1", RunStatus::Blocked),
2160 ("20260801-000000-old2", RunStatus::Stalled),
2161 ] {
2162 let (id, worktree) = due_run_with_winner_worktree(&runs, &wt, &repo, id, status).await;
2163 let mut state = RunState::load_under(&id, home).unwrap();
2164 state.driver_exited = true;
2165 state.save_under(home).unwrap();
2166 task.link_run(&id);
2167 sources.push((id, worktree));
2168 }
2169 crate::pr_recovery::retain_sources(&task, home).unwrap();
2170 let mut recovered = RunState::new(
2171 repo.clone(),
2172 "main".into(),
2173 "base".into(),
2174 task.instruction.clone(),
2175 crate::config::Config::default(),
2176 );
2177 recovered.config.graph.worktree_root = Some(wt.clone());
2178 recovered.status = RunStatus::Ready;
2179 recovered.driver_exited = true;
2180 recovered.save_under(home).unwrap();
2181 task.start(recovered.id.clone());
2182 task.succeed();
2183 crate::daemon::supersede_prior_runs(&task, home);
2184 for (id, _) in &sources {
2185 assert_eq!(
2186 RunState::load_under(id, home).unwrap().status,
2187 RunStatus::Superseded
2188 );
2189 }
2190 let disk = Disk::default();
2191 let now = Timestamp::now()
2192 .checked_add(SignedDuration::from_secs(
2193 i64::try_from(disk.fold_grace_secs + 1).unwrap(),
2194 ))
2195 .unwrap();
2196 let (_, unreadable) = fold_due(&runs, home, &wt, &disk, now).await.unwrap();
2197 assert_eq!(unreadable, 0);
2198 for (id, worktree) in sources {
2199 assert!(worktree.exists());
2200 assert!(
2201 crate::git::branch_exists(&repo, &format!("magi/{id}/A"))
2202 .await
2203 .unwrap()
2204 );
2205 let mut historical = RunState::load_under(&id, home).unwrap();
2206 assert!(!historical.candidates[0].folded);
2207 assert!(
2208 crate::graph::fold_run(&mut historical, true, home)
2209 .await
2210 .unwrap()
2211 .is_empty()
2212 );
2213 assert!(
2214 worktree.exists(),
2215 "every automatic fold entry preserves retained sources"
2216 );
2217 }
2218 }
2219
2220 #[tokio::test]
2221 async fn fold_due_drops_a_superseded_runs_own_winner_worktree_but_keeps_a_readys() {
2222 let dir = tempfile::tempdir().unwrap();
2223 let runs = dir.path().join("runs");
2224 let wt = dir.path().join("wt");
2225 let repo = dir.path().join("repo");
2226 let home = dir.path().to_path_buf();
2227 let disk = Disk::default();
2228 let now = ts("2026-09-05T00:00:00Z");
2229 std::fs::create_dir_all(&repo).unwrap();
2230 init_repo(&repo);
2231
2232 let (superseded_id, superseded_wt) = due_run_with_winner_worktree(
2233 &runs,
2234 &wt,
2235 &repo,
2236 "20260801-000000-supw",
2237 RunStatus::Superseded,
2238 )
2239 .await;
2240 let (ready_id, ready_wt) = due_run_with_winner_worktree(
2247 &runs,
2248 &wt,
2249 &repo,
2250 "20260801-000000-rdyw",
2251 RunStatus::Ready,
2252 )
2253 .await;
2254
2255 let (folded, unreadable) = fold_due(&runs, &home, &wt, &disk, now)
2256 .await
2257 .expect("fold_due");
2258 assert_eq!(folded, 2);
2259 assert_eq!(unreadable, 0);
2260
2261 assert!(
2262 !superseded_wt.exists(),
2263 "a superseded run's own winner never lands anywhere else, so its worktree \
2264 must be dropped exactly like a merged run's"
2265 );
2266 assert!(
2267 ready_wt.exists(),
2268 "a still-ready run's winner may yet be merged by hand - folding must not \
2269 touch it"
2270 );
2271
2272 assert!(read_meta(&runs, &superseded_id).is_ok());
2274 assert!(read_meta(&runs, &ready_id).is_ok());
2275 }
2276}