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 crate::retention::sweep(home, &cfg.disk, now).await;
89 if cfg.disk.auto_fold {
90 let runs = home.join("runs");
91 match fold_due(&runs, home, worktrees_root, &cfg.disk, now).await {
92 Ok((folded, unreadable)) => {
93 out.folded = folded;
94 out.unreadable = unreadable;
95 }
96 Err(e) => {
97 tracing::warn!("housekeep: fold due runs: {e:#}");
98 crate::notices::raise_in(
101 home,
102 crate::notices::Notice::warn(
103 "housekeep:fold",
104 "Automatic cleanup of finished runs failed; disk usage may keep growing.",
105 ),
106 );
107 }
108 }
109 out.orphaned_worktrees =
110 fold_orphaned_worktrees(&runs, worktrees_root, home, cfg.disk.fold_grace_secs, now)
111 .await;
112 if let Err(e) = crate::git::worktree_prune(repo).await {
116 tracing::warn!("housekeep: prune worktree registrations: {e:#}");
117 }
118 out.external_merges_recorded = reconcile_external_merges(&runs, home, &cfg.disk, now).await;
119 out.stale_pr_states_repaired = crate::land::repair_stale_pr_states(home, 5).await.0;
122 }
123 match prune_cache_if_over_limit(cfg, home) {
124 Ok(Some(pruned)) => {
125 out.cache_files = pruned.files;
126 out.cache_freed = pruned.freed;
127 }
128 Ok(None) => {}
129 Err(e) => tracing::warn!("housekeep: prune cache: {e:#}"),
130 }
131 out.questions_abandoned =
139 abandon_settled_questions(&Questions::at(home.join("questions")), &home.join("runs"));
140 out
141}
142
143pub fn abandon_settled_questions(store: &Questions, runs: &Path) -> usize {
154 let waiting_on: BTreeSet<String> = store
155 .list()
156 .into_iter()
157 .filter(|q| q.status.open())
158 .map(|q| q.run)
159 .collect();
160 let mut abandoned = 0;
161 for run in waiting_on {
162 let Ok(meta) = read_meta(runs, &run) else {
163 continue;
164 };
165 match store.settle_run(&run, meta.status) {
166 Ok(n) => abandoned += n,
167 Err(e) => tracing::warn!("housekeep: abandon questions for {run}: {e:#}"),
168 }
169 }
170 abandoned
171}
172
173pub async fn fold_due(
198 runs: &Path,
199 home: &Path,
200 _worktrees_root: &Path,
201 disk: &Disk,
202 now: Timestamp,
203) -> Result<(usize, usize)> {
204 let mut folded = 0usize;
205 let mut unreadable = 0usize;
206 let mut ids: Vec<String> = std::fs::read_dir(runs)
207 .into_iter()
208 .flatten()
209 .flatten()
210 .filter(|e| e.path().join("run.json").is_file())
211 .map(|e| e.file_name().to_string_lossy().into_owned())
212 .collect();
213 ids.sort_unstable();
214 for id in ids {
215 if crate::daemon::is_working_on(home, &id, now) {
216 continue;
217 }
218 let meta = match read_meta(runs, &id) {
219 Ok(meta) => meta,
220 Err(e) => {
221 unreadable += 1;
222 tracing::warn!("housekeep: run {id} unreadable, left alone: {e:#}");
223 continue;
224 }
225 };
226 if meta.status.resumable() || !due(now, meta.updated_at, disk.fold_grace_secs) {
227 continue;
228 }
229 let mut state = match read_state(runs, &id) {
238 Ok(state) => state,
239 Err(e) => {
240 unreadable += 1;
241 tracing::warn!("housekeep: run {id} unreadable, left alone: {e:#}");
242 continue;
243 }
244 };
245 if state.startup.as_ref().is_some_and(|s| !s.complete) {
246 continue;
247 }
248 if let Some(reason) = crate::pr_recovery::source_retention_reason(home, &state) {
249 tracing::info!(%id, %reason, "housekeep: source retained; fold refused");
250 continue;
251 }
252 if state.schema != SCHEMA {
253 tracing::info!(
254 "housekeep: run {id} was written by schema {} (this build speaks {SCHEMA}); \
255 folding it anyway",
256 state.schema
257 );
258 }
259 let drop_winner = matches!(
269 state.status,
270 RunStatus::Merged | RunStatus::Superseded | RunStatus::AlreadyInBase
271 );
272 match crate::graph::fold_run(&mut state, drop_winner, home).await {
279 Ok(_) => folded += 1,
280 Err(e) => tracing::warn!("housekeep: fold {id}: {e:#}"),
281 }
282 }
283 Ok((folded, unreadable))
284}
285
286#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
288pub struct FetchReport {
289 pub fetched: usize,
291 pub no_origin: usize,
293 pub failed: usize,
295}
296
297pub async fn fetch_origins(
305 roots: &[PathBuf],
306 per_repo: std::time::Duration,
307 stop: impl Fn() -> bool,
308) -> FetchReport {
309 let mut report = FetchReport::default();
310 for repo in crate::repos::scan(roots) {
311 if stop() {
312 break;
313 }
314 match crate::git::git_raw(&repo.path, &["remote", "get-url", "origin"]).await {
315 Ok(out) if out.ok() => {}
316 _ => {
317 report.no_origin += 1;
318 continue;
319 }
320 }
321 match crate::git::fetch_origin(&repo.path, per_repo).await {
322 Ok(()) => report.fetched += 1,
323 Err(e) => {
324 tracing::warn!("fetch origin in {}: {e:#}", repo.name);
325 report.failed += 1;
326 }
327 }
328 }
329 report
330}
331
332pub fn due(now: Timestamp, updated: Timestamp, grace_secs: u64) -> bool {
338 now.duration_since(updated) > SignedDuration::new(grace_secs as i64, 0)
339}
340
341#[derive(Deserialize)]
344struct Meta {
345 status: RunStatus,
346 updated_at: Timestamp,
347}
348
349fn read_meta(runs: &Path, id: &str) -> Result<Meta> {
353 let path = runs.join(id).join("run.json");
354 let body =
355 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
356 let meta: Meta =
357 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
358 Ok(meta)
359}
360
361const MAX_EXTERNAL_MERGE_CHECKS_PER_PASS: usize = 5;
367
368#[derive(Deserialize)]
371struct ExternalMergeMeta {
372 status: RunStatus,
373 updated_at: Timestamp,
374 #[serde(default)]
375 merge: Option<crate::run::MergeOutcome>,
376}
377
378fn read_external_merge_meta(runs: &Path, id: &str) -> Result<ExternalMergeMeta> {
379 let path = runs.join(id).join("run.json");
380 let body =
381 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
382 let meta: ExternalMergeMeta =
383 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
384 Ok(meta)
385}
386
387pub fn eligible_for_external_merge_check(
396 status: RunStatus,
397 merge_recorded: bool,
398 updated_at: Timestamp,
399 now: Timestamp,
400 grace_secs: u64,
401) -> bool {
402 status == RunStatus::Blocked && !merge_recorded && due(now, updated_at, grace_secs)
403}
404
405fn rotate_from(ids: &[String], start: usize) -> Vec<String> {
423 if ids.is_empty() {
424 return Vec::new();
425 }
426 let start = start % ids.len();
427 ids[start..]
428 .iter()
429 .chain(ids[..start].iter())
430 .cloned()
431 .collect()
432}
433
434const EXTERNAL_MERGE_CURSOR_FILE: &str = "external-merge-cursor";
438
439fn read_external_merge_cursor(home: &Path) -> usize {
444 std::fs::read_to_string(home.join(EXTERNAL_MERGE_CURSOR_FILE))
445 .ok()
446 .and_then(|s| s.trim().parse().ok())
447 .unwrap_or(0)
448}
449
450fn write_external_merge_cursor(home: &Path, cursor: usize) {
455 if let Err(e) = std::fs::write(home.join(EXTERNAL_MERGE_CURSOR_FILE), cursor.to_string()) {
456 tracing::warn!("housekeep: persist the external-merge sweep's cursor: {e:#}");
457 }
458}
459
460async fn reconcile_external_merges(runs: &Path, home: &Path, disk: &Disk, now: Timestamp) -> usize {
481 let mut ids: Vec<String> = std::fs::read_dir(runs)
482 .into_iter()
483 .flatten()
484 .flatten()
485 .filter(|e| e.path().join("run.json").is_file())
486 .map(|e| e.file_name().to_string_lossy().into_owned())
487 .collect();
488 ids.sort_unstable();
489
490 let mut eligible: Vec<String> = Vec::new();
496 for id in ids {
497 if crate::daemon::is_working_on(home, &id, now) {
498 continue;
499 }
500 let meta = match read_external_merge_meta(runs, &id) {
501 Ok(meta) => meta,
502 Err(_) => continue,
505 };
506 if eligible_for_external_merge_check(
507 meta.status,
508 meta.merge.is_some(),
509 meta.updated_at,
510 now,
511 disk.fold_grace_secs,
512 ) {
513 eligible.push(id);
514 }
515 }
516
517 if eligible.is_empty() {
518 return 0;
519 }
520 let cursor = read_external_merge_cursor(home);
521 let window: Vec<String> = rotate_from(&eligible, cursor)
522 .into_iter()
523 .take(MAX_EXTERNAL_MERGE_CHECKS_PER_PASS)
524 .collect();
525 write_external_merge_cursor(home, (cursor + window.len()) % eligible.len());
530
531 let mut reconciled = 0usize;
532 for id in window {
533 let mut state = match read_state(runs, &id) {
534 Ok(state) => state,
535 Err(_) => continue,
536 };
537 match crate::land::find_external_merge(&state).await {
538 Ok(Some(found)) => {
539 match crate::land::correct_confirmed_external_merge(&mut state, &found.url).await {
540 Ok(_) => {
541 if let Err(e) = crate::graph::fold_run(&mut state, true, home).await {
542 tracing::warn!(
543 "housekeep: fold {id} after recording its external merge: {e:#}"
544 );
545 }
546 reconciled += 1;
547 }
548 Err(e) => tracing::warn!(
549 "housekeep: record external merge {} for {id}: {e:#}",
550 found.url
551 ),
552 }
553 }
554 Ok(None) => {}
555 Err(e) => {
556 tracing::warn!("housekeep: check external merge for {id}: {e:#}");
557 crate::notices::raise_in(
558 home,
559 crate::notices::Notice::warn(
560 &format!("merged-unrecorded:{id}"),
561 format!(
562 "Run {id} is blocked with no recorded merge, and checking GitHub \
563 for a merge failed; check by hand."
564 ),
565 )
566 .link(crate::notices::Link::Run { id: id.clone() }),
567 );
568 }
569 }
570 }
571 reconciled
572}
573
574fn read_state(runs: &Path, id: &str) -> Result<RunState> {
589 let path = runs.join(id).join("run.json");
590 let body =
591 std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
592 let state: RunState =
593 serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
594 Ok(state)
595}
596
597async fn remove_clean_bay(root: &Path) -> Result<()> {
600 let mut pending = vec![root.to_path_buf()];
601 let mut directories = Vec::new();
602 let mut worktrees = Vec::new();
603 while let Some(path) = pending.pop() {
604 if path.join(".git").exists() {
605 if !crate::git::is_clean(&path).await? {
606 anyhow::bail!("uncommitted work in {}; recovery required", path.display());
607 }
608 worktrees.push(path);
609 } else {
610 for entry in std::fs::read_dir(&path)? {
611 let entry = entry?;
612 if !entry.file_type()?.is_dir() {
613 anyhow::bail!(
614 "unknown worktree content in {}; recovery required",
615 path.display()
616 );
617 }
618 pending.push(entry.path());
619 }
620 directories.push(path);
621 }
622 }
623 for path in worktrees {
624 if !crate::git::worktree_remove_clean(&path, &path).await? {
625 anyhow::bail!("clean worktree removal refused: {}", path.display());
626 }
627 }
628 for path in directories.into_iter().rev() {
629 std::fs::remove_dir(path)?;
630 }
631 Ok(())
632}
633
634pub async fn fold_unreadable(runs: &Path, worktrees_root: &Path, id: &str) -> Result<Vec<String>> {
642 let resolved = resolve_id_path(runs, id)?;
643 let mut removed = Vec::new();
644 let run_dir = runs.join(&resolved);
645 let wt = worktrees_root.join(short_of(&resolved));
646 if wt.exists() {
647 remove_clean_bay(&wt).await?;
648 removed.push(wt.to_string_lossy().into_owned());
649 }
650 if run_dir.exists() {
652 std::fs::remove_dir_all(&run_dir)
653 .with_context(|| format!("remove {}", run_dir.display()))?;
654 removed.push(format!("runs/{resolved}"));
655 }
656 Ok(removed)
657}
658
659pub async fn fold_orphaned_worktrees(
698 runs: &Path,
699 worktrees_root: &Path,
700 home: &Path,
701 grace_secs: u64,
702 now: Timestamp,
703) -> usize {
704 let known: std::collections::HashSet<String> = std::fs::read_dir(runs)
705 .into_iter()
706 .flatten()
707 .flatten()
708 .map(|e| e.file_name().to_string_lossy().into_owned())
709 .filter(|name| crate::run::is_run_id(name))
710 .map(|id| short_of(&id).to_owned())
711 .collect();
712
713 let mut folded = 0usize;
714 for entry in std::fs::read_dir(worktrees_root)
715 .into_iter()
716 .flatten()
717 .flatten()
718 {
719 if !entry.path().is_dir() {
720 continue;
721 }
722 let short = entry.file_name().to_string_lossy().into_owned();
723 if !looks_like_a_worktree_bay(&short) {
724 continue;
725 }
726 if known.contains(&short) || crate::daemon::is_working_on_short(home, &short, now) {
727 continue;
728 }
729 let wt = entry.path();
730 if !stale_enough(&wt, grace_secs, now) {
731 continue;
732 }
733 match remove_clean_bay(&wt).await {
734 Ok(()) => folded += 1,
735 Err(e) => tracing::warn!(
736 "housekeep: remove orphaned worktree {}: {e:#}",
737 wt.display()
738 ),
739 }
740 }
741 folded
742}
743
744fn looks_like_a_worktree_bay(name: &str) -> bool {
753 name.len() == 4 && name.bytes().all(|b| b.is_ascii_alphanumeric())
754}
755
756fn stale_enough(dir: &Path, grace_secs: u64, now: Timestamp) -> bool {
775 let Ok(modified) = std::fs::metadata(dir).and_then(|m| m.modified()) else {
776 return false;
777 };
778 let Ok(ts) = Timestamp::try_from(modified) else {
779 return false;
780 };
781 due(now, ts, grace_secs.max(MIN_ORPHAN_AGE_SECS))
782}
783
784const MIN_ORPHAN_AGE_SECS: u64 = 5 * 60;
795
796fn resolve_id_path(runs: &Path, prefix: &str) -> Result<String> {
799 if runs.join(prefix).is_dir() && crate::run::is_run_id(prefix) {
804 return Ok(prefix.to_owned());
805 }
806 let mut hits: Vec<String> = Vec::new();
807 for e in std::fs::read_dir(runs).into_iter().flatten().flatten() {
808 if !e.path().is_dir() {
809 continue;
810 }
811 let id = e.file_name().to_string_lossy().into_owned();
812 if crate::run::is_run_id(&id) && (id.starts_with(prefix) || id.ends_with(prefix)) {
813 hits.push(id);
814 }
815 }
816 match hits.len() {
817 1 => Ok(hits.into_iter().next().expect("exactly one hit")),
818 0 => bail!("no run matches `{prefix}`"),
819 _ => bail!(
820 "`{prefix}` matches {} runs: {}",
821 hits.len(),
822 hits.join(", ")
823 ),
824 }
825}
826
827pub fn clear_abandoned_active(state: &mut RunState, home: &Path, now: Timestamp) -> Result<bool> {
842 if crate::daemon::is_working_on(home, &state.id, now) || !state.active_all_overrun(now) {
843 return Ok(false);
844 }
845 state.abandon("fold");
846 state.save_under(home)?;
847 if let Err(e) = Questions::at(home.join("questions")).settle_run(&state.id, state.status) {
855 tracing::warn!("abandon questions for {}: {e:#}", state.id);
856 }
857 Ok(true)
858}
859
860pub fn prune_cache(home: &Path, cache: &Path, limit_bytes: u64) -> Result<Option<Prune>> {
869 crate::cache::maintenance_prune(home, cache, limit_bytes)
870}
871
872pub fn prune_cache_if_over_limit(
882 cfg: &crate::config::Config,
883 home: &Path,
884) -> Result<Option<Prune>> {
885 if cfg.disk.cache_limit_bytes == 0 {
886 return Ok(None);
887 }
888 let Some(cache) = cfg.cache_dir() else {
889 return Ok(None);
890 };
891 let limit = crate::disk::free_bytes(&cache)
892 .ok()
893 .map_or(cfg.disk.cache_limit_bytes, |free| {
894 crate::disk::pressure_limit(
895 dir_size(&cache),
896 cfg.disk.cache_limit_bytes,
897 cfg.disk.min_free_bytes,
898 free,
899 )
900 });
901 prune_cache(home, &cache, limit)
902}
903
904pub fn cache_report(cfg: &crate::config::Config) -> Option<(PathBuf, u64, u64)> {
910 let cache = cfg.cache_dir()?;
911 Some((
912 cache.clone(),
913 cache_size(&cache),
914 cfg.disk.cache_limit_bytes,
915 ))
916}
917
918pub fn cache_size(cache: &Path) -> u64 {
920 dir_size(cache)
921}
922
923#[cfg(test)]
924mod tests {
925
926 async fn sh(cwd: &Path, args: &[&str]) {
927 let out = crate::git::git_raw(cwd, args).await.expect("spawn git");
928 assert!(out.ok(), "git {args:?}: {}", out.stderr);
929 }
930
931 async fn commit(cwd: &Path, file: &str, body: &str) {
932 std::fs::write(cwd.join(file), body).unwrap();
933 sh(cwd, &["add", file]).await;
934 sh(
935 cwd,
936 &[
937 "-c",
938 "user.name=t",
939 "-c",
940 "user.email=t@example.com",
941 "commit",
942 "-q",
943 "-m",
944 body,
945 ],
946 )
947 .await;
948 }
949
950 const LONG: std::time::Duration = std::time::Duration::from_secs(30);
951
952 #[tokio::test]
953 async fn fetch_advances_origin_main_and_leaves_the_checkout_alone() {
954 let tmp = tempfile::tempdir().unwrap();
955 let t = tmp.path();
956 let bare = t.join("remote.git");
957 sh(
958 t,
959 &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
960 )
961 .await;
962 let dir = t.join("root/h/o/r");
963 std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
964 sh(
965 t,
966 &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
967 )
968 .await;
969 sh(&dir, &["checkout", "-q", "-b", "main"]).await;
970 commit(&dir, "a.txt", "one").await;
971 sh(&dir, &["push", "-q", "origin", "main"]).await;
972 sh(&dir, &["checkout", "-q", "--detach"]).await;
973 std::fs::write(dir.join("a.txt"), "dirty").unwrap();
974
975 let other = t.join("other");
976 sh(
977 t,
978 &[
979 "clone",
980 "-q",
981 bare.to_str().unwrap(),
982 other.to_str().unwrap(),
983 ],
984 )
985 .await;
986 commit(&other, "b.txt", "two").await;
987 sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
988
989 let rev = |d: PathBuf, r: &'static str| async move {
990 crate::git::git(&d, &["rev-parse", r]).await.unwrap()
991 };
992 let head = rev(dir.clone(), "HEAD").await;
993 let before = rev(dir.clone(), "origin/main").await;
994 let status = crate::git::git(&dir, &["status", "--porcelain"])
995 .await
996 .unwrap();
997
998 let r = fetch_origins(&[t.join("root")], LONG, || false).await;
999 assert_eq!(
1000 r,
1001 FetchReport {
1002 fetched: 1,
1003 no_origin: 0,
1004 failed: 0
1005 }
1006 );
1007
1008 assert_ne!(rev(dir.clone(), "origin/main").await, before);
1009 assert_eq!(rev(dir.clone(), "HEAD").await, head);
1010 assert_eq!(
1011 crate::git::git(&dir, &["status", "--porcelain"])
1012 .await
1013 .unwrap(),
1014 status
1015 );
1016 assert_eq!(std::fs::read_to_string(dir.join("a.txt")).unwrap(), "dirty");
1017 }
1018
1019 #[tokio::test]
1020 async fn fetch_never_writes_a_local_branch_whatever_the_configured_refspec() {
1021 let tmp = tempfile::tempdir().unwrap();
1022 let t = tmp.path();
1023 let bare = t.join("remote.git");
1024 sh(
1025 t,
1026 &["init", "-q", "--bare", "-b", "main", bare.to_str().unwrap()],
1027 )
1028 .await;
1029 let dir = t.join("root/h/o/r");
1030 std::fs::create_dir_all(dir.parent().unwrap()).unwrap();
1031 sh(
1032 t,
1033 &["clone", "-q", bare.to_str().unwrap(), dir.to_str().unwrap()],
1034 )
1035 .await;
1036 sh(&dir, &["checkout", "-q", "-b", "main"]).await;
1037 commit(&dir, "a.txt", "one").await;
1038 sh(&dir, &["push", "-q", "origin", "main"]).await;
1039 sh(&dir, &["checkout", "-q", "--detach"]).await;
1040 sh(
1042 &dir,
1043 &[
1044 "config",
1045 "remote.origin.fetch",
1046 "+refs/heads/main:refs/heads/main",
1047 ],
1048 )
1049 .await;
1050 let _ = crate::git::git_raw(&dir, &["update-ref", "-d", "refs/remotes/origin/main"]).await;
1051
1052 let other = t.join("other");
1053 sh(
1054 t,
1055 &[
1056 "clone",
1057 "-q",
1058 bare.to_str().unwrap(),
1059 other.to_str().unwrap(),
1060 ],
1061 )
1062 .await;
1063 commit(&other, "b.txt", "two").await;
1064 sh(&other, &["push", "-q", "origin", "HEAD:main"]).await;
1065 let remote_tip = crate::git::git(&other, &["rev-parse", "HEAD"])
1066 .await
1067 .unwrap();
1068 let local_main = crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1069 .await
1070 .unwrap();
1071
1072 let r = fetch_origins(&[t.join("root")], LONG, || false).await;
1073 assert_eq!(r.fetched, 1);
1074 assert_eq!(
1075 crate::git::git(&dir, &["rev-parse", "refs/remotes/origin/main"])
1076 .await
1077 .unwrap(),
1078 remote_tip
1079 );
1080 assert_eq!(
1081 crate::git::git(&dir, &["rev-parse", "refs/heads/main"])
1082 .await
1083 .unwrap(),
1084 local_main
1085 );
1086 }
1087
1088 #[tokio::test]
1089 async fn fetch_skips_no_origin_and_survives_an_unreachable_origin() {
1090 let tmp = tempfile::tempdir().unwrap();
1091 let t = tmp.path();
1092 let root = t.join("root");
1093 let bare = t.join("remote.git");
1094 sh(t, &["init", "-q", "--bare", bare.to_str().unwrap()]).await;
1095
1096 let none = root.join("h/o/none");
1097 let dead = root.join("h/o/dead");
1098 let good = root.join("h/o/good");
1099 for d in [&none, &dead, &good] {
1100 std::fs::create_dir_all(d).unwrap();
1101 sh(d, &["init", "-q"]).await;
1102 }
1103 sh(
1104 &dead,
1105 &[
1106 "remote",
1107 "add",
1108 "origin",
1109 t.join("missing").to_str().unwrap(),
1110 ],
1111 )
1112 .await;
1113 sh(&good, &["remote", "add", "origin", bare.to_str().unwrap()]).await;
1114
1115 let r = fetch_origins(&[root], LONG, || false).await;
1116 assert_eq!(
1117 r,
1118 FetchReport {
1119 fetched: 1,
1120 no_origin: 1,
1121 failed: 1
1122 }
1123 );
1124 }
1125 use super::*;
1126 use crate::config::Disk;
1127 use std::fs;
1128
1129 fn ts(s: &str) -> Timestamp {
1130 s.parse().expect("rfc3339")
1131 }
1132
1133 fn block_on<F: std::future::Future>(f: F) -> F::Output {
1134 tokio::runtime::Runtime::new().expect("runtime").block_on(f)
1135 }
1136
1137 #[test]
1138 fn a_run_is_due_after_its_grace_and_not_before() {
1139 let now = ts("2026-09-05T00:00:00Z");
1140 let grace = 600;
1141 let old = now - SignedDuration::new(601, 0);
1142 let fresh = now - SignedDuration::new(599, 0);
1143 assert!(due(now, old, grace));
1144 assert!(!due(now, fresh, grace));
1145 let edge = now - SignedDuration::new(600, 0);
1147 assert!(!due(now, edge, grace));
1148 assert!(due(now, old, 0));
1150 }
1151
1152 #[test]
1153 fn rotate_from_moves_the_starting_point_as_the_cursor_advances() {
1154 let ids: Vec<String> = ["a", "b", "c", "d", "e"]
1155 .iter()
1156 .map(|s| s.to_string())
1157 .collect();
1158
1159 assert_eq!(rotate_from(&ids, 0), ids);
1161
1162 let at_2 = rotate_from(&ids, 2);
1165 assert_eq!(at_2, vec!["c", "d", "e", "a", "b"]);
1166 for id in &ids {
1167 assert!(at_2.contains(id));
1168 }
1169
1170 assert_eq!(rotate_from(&ids, 7), rotate_from(&ids, 2));
1173
1174 assert_eq!(rotate_from(&[], 0), Vec::<String>::new());
1176 }
1177
1178 #[test]
1185 fn advancing_the_cursor_by_the_window_size_gives_every_id_a_turn() {
1186 let ids: Vec<String> = (0..12).map(|n| format!("run-{n}")).collect();
1187 let cap = 5usize;
1188 let mut cursor = 0usize;
1189 let mut ever_seen: std::collections::HashSet<String> = std::collections::HashSet::new();
1190 for _ in 0..ids.len() {
1191 let window: Vec<String> = rotate_from(&ids, cursor).into_iter().take(cap).collect();
1192 ever_seen.extend(window.iter().cloned());
1193 cursor = (cursor + window.len()) % ids.len();
1194 }
1195 assert_eq!(
1196 ever_seen.len(),
1197 ids.len(),
1198 "every id must be checked at least once across a full rotation, whatever \
1199 the cadence between passes"
1200 );
1201 }
1202
1203 #[test]
1204 fn the_external_merge_cursor_round_trips_through_a_files_absence() {
1205 let dir = tempfile::tempdir().unwrap();
1206 let home = dir.path();
1207
1208 assert_eq!(read_external_merge_cursor(home), 0);
1210
1211 write_external_merge_cursor(home, 7);
1212 assert_eq!(read_external_merge_cursor(home), 7);
1213
1214 std::fs::write(home.join(EXTERNAL_MERGE_CURSOR_FILE), "not a number").unwrap();
1217 assert_eq!(read_external_merge_cursor(home), 0);
1218 }
1219
1220 #[test]
1221 fn a_run_qualifies_for_an_external_merge_check_only_when_blocked_unmerged_and_due() {
1222 let now = ts("2026-09-05T00:00:00Z");
1223 let grace = 600;
1224 let old = now - SignedDuration::new(601, 0);
1225 let fresh = now - SignedDuration::new(599, 0);
1226
1227 assert!(
1228 eligible_for_external_merge_check(RunStatus::Blocked, false, old, now, grace),
1229 "blocked, unmerged, and past its grace period is exactly the run this exists for"
1230 );
1231 assert!(
1232 !eligible_for_external_merge_check(RunStatus::Blocked, false, fresh, now, grace),
1233 "an operator mid-fix deserves the same grace window `fold_due` gives before \
1234 the janitor starts asking GitHub about it"
1235 );
1236 assert!(
1237 !eligible_for_external_merge_check(RunStatus::Blocked, true, old, now, grace),
1238 "a run `land::land` already recorded a merge outcome for has its own answer \
1239 already; this check is only for a run with nothing recorded at all"
1240 );
1241 assert!(
1242 !eligible_for_external_merge_check(RunStatus::Stalled, false, old, now, grace),
1243 "stalled is not blocked - it means the tally never reached quorum, which a \
1244 pull request cannot fix"
1245 );
1246 assert!(
1247 !eligible_for_external_merge_check(RunStatus::Ready, false, old, now, grace),
1248 "ready has nothing to correct - it was never landed by design"
1249 );
1250 assert!(
1251 !eligible_for_external_merge_check(RunStatus::Superseded, false, old, now, grace),
1252 "a superseded run's task already has its answer from a later attempt; \
1253 there is nothing left for GitHub to confirm here"
1254 );
1255 }
1256
1257 fn write_blocked_run(runs: &Path, id: &str, updated_at: Timestamp) {
1263 use crate::run::{Candidate, Tally};
1264 use std::collections::BTreeMap;
1265
1266 let mut state = RunState::new(
1267 PathBuf::from("/nonexistent/repo"),
1268 "main".to_owned(),
1269 "0000000000000000000000000000000000000000".to_owned(),
1270 String::new(),
1271 crate::config::Config::default(),
1272 );
1273 state.id = id.to_owned();
1274 state.status = RunStatus::Blocked;
1275 state.updated_at = updated_at;
1276 state.candidates.push(Candidate {
1277 index: 0,
1278 label: 'A',
1279 agent: "agent".to_owned(),
1280 branch: format!("magi/{id}/A"),
1281 worktree: PathBuf::from("/nonexistent/repo"),
1282 summary: String::new(),
1283 stat: String::new(),
1284 files: 0,
1285 commits: 0,
1286 empty: false,
1287 failed: None,
1288 verified_noop: None,
1289 duration_ms: 0,
1290 folded: false,
1291 });
1292 state.tally = Some(Tally {
1293 first_choice: BTreeMap::from([('A', 1)]),
1294 borda: BTreeMap::new(),
1295 winner: 'A',
1296 rankings: 1,
1297 unanimous_initial: true,
1298 deliberated: false,
1299 changed_votes: 0,
1300 unanimous_final: true,
1301 tie_break: None,
1302 judges: 1,
1303 present: 1,
1304 quorum: 1,
1305 met_quorum: true,
1306 uncontested: None,
1307 });
1308 std::fs::create_dir_all(runs.join(id)).unwrap();
1309 std::fs::write(
1310 runs.join(id).join("run.json"),
1311 serde_json::to_string_pretty(&state).unwrap(),
1312 )
1313 .unwrap();
1314 }
1315
1316 #[tokio::test]
1321 async fn reconcile_external_merges_leaves_ineligible_and_unconfirmable_runs_alone() {
1322 let dir = tempfile::tempdir().unwrap();
1323 let runs = dir.path().join("runs");
1324 let home = dir.path().to_path_buf();
1325 std::fs::create_dir_all(&runs).unwrap();
1326
1327 let now = ts("2026-09-05T00:00:00Z");
1328 let disk = Disk {
1329 fold_grace_secs: 600,
1330 ..Disk::default()
1331 };
1332
1333 let due_id = "20260905-000000-blkd";
1334 write_blocked_run(&runs, due_id, ts("2026-08-01T00:00:00Z"));
1335
1336 let fresh_id = "20260905-000000-fres";
1337 write_blocked_run(&runs, fresh_id, now);
1338
1339 let live_id = "20260905-000000-live";
1340 write_blocked_run(&runs, live_id, ts("2026-08-01T00:00:00Z"));
1341 let mut status = crate::daemon::Status::new();
1342 status.current = vec![crate::daemon::Current {
1343 worker: None,
1344 task: "20260905-000000-task".to_owned(),
1345 run: live_id.to_owned(),
1346 }];
1347 status.updated_at = now;
1348 crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1349
1350 let reconciled = reconcile_external_merges(&runs, &home, &disk, now).await;
1351 assert_eq!(
1352 reconciled, 0,
1353 "an unreachable repository can never be confirmed merged"
1354 );
1355 for id in [due_id, fresh_id, live_id] {
1356 assert_eq!(
1357 read_meta(&runs, id).unwrap().status,
1358 RunStatus::Blocked,
1359 "{id} must be left exactly as it was found"
1360 );
1361 }
1362 let notices = crate::notices::Notices::at(home.join("notifications")).list();
1363 let due_notice = notices
1364 .iter()
1365 .find(|n| n.key == format!("merged-unrecorded:{due_id}"))
1366 .expect("notice for due_id must be raised");
1367 assert_eq!(
1368 due_notice.link,
1369 Some(crate::notices::Link::Run {
1370 id: due_id.to_owned(),
1371 })
1372 );
1373 }
1374
1375 #[tokio::test]
1385 async fn reconcile_external_merges_covers_a_larger_fleet_across_repeated_passes_at_one_instant()
1386 {
1387 let dir = tempfile::tempdir().unwrap();
1388 let runs = dir.path().join("runs");
1389 let home = dir.path().to_path_buf();
1390 std::fs::create_dir_all(&runs).unwrap();
1391
1392 let now = ts("2026-09-05T00:00:00Z");
1393 let disk = Disk {
1394 fold_grace_secs: 600,
1395 ..Disk::default()
1396 };
1397 let old = ts("2026-08-01T00:00:00Z");
1398
1399 let ids: Vec<String> = (0..8).map(|n| format!("20260905-000000-r{n:03}")).collect();
1400 for id in &ids {
1401 write_blocked_run(&runs, id, old);
1402 }
1403
1404 let notices = crate::notices::Notices::at(home.join("notifications"));
1405
1406 reconcile_external_merges(&runs, &home, &disk, now).await;
1412 let after_first = notices.list().len();
1413 assert_eq!(
1414 after_first, MAX_EXTERNAL_MERGE_CHECKS_PER_PASS,
1415 "the first pass checks exactly one cap's worth"
1416 );
1417
1418 reconcile_external_merges(&runs, &home, &disk, now).await;
1419 let after_second = notices.list().len();
1420 assert_eq!(
1421 after_second,
1422 ids.len(),
1423 "a second pass at the same instant must still reach every id the \
1424 first pass had no room for, not repeat the same cap's worth"
1425 );
1426 }
1427
1428 #[test]
1429 fn the_meta_reader_is_tolerant_of_everything_except_the_deciders() {
1430 let dir = tempfile::tempdir().unwrap();
1431 let runs = dir.path().join("runs");
1432 let id = "20260905-000000-abcd";
1433 std::fs::create_dir_all(runs.join(id)).unwrap();
1434 std::fs::write(
1435 runs.join(id).join("run.json"),
1436 r#"{"schema": 99, "id": "20260905-000000-abcd", "updated_at": "2026-09-05T00:00:00Z", "status": "ready", "junk_from_another_build": [1, 2, 3]}"#,
1437 )
1438 .unwrap();
1439 let meta = read_meta(&runs, id).expect("readable");
1440 assert_eq!(meta.status, RunStatus::Ready);
1441 assert_eq!(meta.updated_at, ts("2026-09-05T00:00:00Z"));
1442 assert!(read_meta(&runs, "nope").is_err(), "missing file unreadable");
1443 std::fs::write(runs.join(id).join("run.json"), "not json at all").unwrap();
1444 assert!(read_meta(&runs, id).is_err(), "garbage unreadable");
1445 }
1446
1447 #[test]
1448 fn fold_unreadable_preserves_unknown_content_and_recovery_artifacts() {
1449 let dir = tempfile::tempdir().unwrap();
1450 let runs = dir.path().join("runs");
1451 let wt = dir.path().join("wt");
1452 let id = "20260905-000000-abcd";
1453 std::fs::create_dir_all(runs.join(id).join("artifacts")).unwrap();
1454 std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1455 std::fs::create_dir_all(wt.join("abcd")).unwrap();
1456 std::fs::write(wt.join("abcd/leftover"), "owned edits").unwrap();
1457 assert!(block_on(fold_unreadable(&runs, &wt, id)).is_err());
1458 assert!(runs.join(id).join("artifacts").exists());
1459 assert_eq!(
1460 std::fs::read_to_string(wt.join("abcd/leftover")).unwrap(),
1461 "owned edits"
1462 );
1463 }
1464
1465 #[test]
1466 fn fold_unreadable_releases_run_dir_and_worktrees() {
1467 let dir = tempfile::tempdir().unwrap();
1468 let runs = dir.path().join("runs");
1469 let wt = dir.path().join("wt");
1470 let id = "20260905-000000-abcd";
1471 std::fs::create_dir_all(runs.join(id)).unwrap();
1472 std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1473 std::fs::create_dir_all(wt.join("abcd")).unwrap();
1474
1475 let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold");
1476 assert_eq!(removed.len(), 2);
1477 assert!(!runs.join(id).exists(), "run dir gone");
1478 assert!(!wt.join("abcd").exists(), "worktrees gone");
1479
1480 std::fs::create_dir_all(runs.join(id)).unwrap();
1482 std::fs::write(runs.join(id).join("run.json"), "garbage").unwrap();
1483 std::fs::create_dir_all(wt.join("abcd")).unwrap();
1484 let removed = block_on(fold_unreadable(&runs, &wt, "20260905")).expect("by prefix");
1485 assert_eq!(removed.len(), 2);
1486 assert!(
1490 block_on(fold_unreadable(&runs, &wt, id)).is_err(),
1491 "a run already gone cannot be resolved again"
1492 );
1493 }
1494
1495 #[test]
1496 fn prune_cache_sheds_the_oldest_generation_until_it_fits() {
1497 let home = tempfile::tempdir().unwrap();
1498 let dir = tempfile::tempdir().unwrap();
1499 fs::write(dir.path().join("old"), b"xx").unwrap();
1502 fs::write(dir.path().join("new"), b"yy").unwrap();
1503 touch(&dir.path().join("old"), 1_000_000);
1504 touch(&dir.path().join("new"), 2_000_000);
1505
1506 let out = prune_cache(home.path(), dir.path(), 2)
1507 .expect("prune")
1508 .expect("the cache is free");
1509 assert_eq!(out.files, 1, "one deletion is enough to reach the cap");
1510 assert_eq!(out.remaining, 2);
1511 assert!(!dir.path().join("old").exists(), "the older file went");
1512 assert!(dir.path().join("new").exists(), "the newer one stayed");
1513
1514 let tied = tempfile::tempdir().unwrap();
1519 fs::write(tied.path().join("big"), b"xxxx").unwrap();
1520 fs::write(tied.path().join("small"), b"yy").unwrap();
1521 touch(&tied.path().join("big"), 1_000_000);
1522 touch(&tied.path().join("small"), 1_000_000);
1523 let out = prune_cache(home.path(), tied.path(), 2)
1524 .expect("prune")
1525 .expect("the cache is free");
1526 assert_eq!(out.files, 1, "the big one alone gets under the cap");
1527 assert_eq!(out.remaining, 2);
1528 assert!(tied.path().join("small").exists());
1529 }
1530
1531 #[test]
1532 fn prune_cache_if_over_limit_resolves_the_opt_outs_before_ever_measuring() {
1533 let home = tempfile::tempdir().unwrap();
1534 let dir = tempfile::tempdir().unwrap();
1535 fs::write(dir.path().join("big"), vec![0u8; 10]).unwrap();
1536
1537 let mut cfg = crate::config::Config::default();
1538 cfg.verify.gate = vec![format!(
1539 "CARGO_TARGET_DIR={} cargo make check",
1540 dir.path().display()
1541 )];
1542
1543 cfg.disk.cache_limit_bytes = 0;
1546 assert_eq!(
1547 prune_cache_if_over_limit(&cfg, home.path()).unwrap(),
1548 None,
1549 "a zero cap must not even look at the directory"
1550 );
1551 assert!(dir.path().join("big").exists());
1552
1553 let mut no_cache = crate::config::Config::default();
1556 no_cache.disk.cache_limit_bytes = 1;
1557 assert_eq!(
1558 prune_cache_if_over_limit(&no_cache, home.path()).unwrap(),
1559 None
1560 );
1561
1562 cfg.disk.cache_limit_bytes = 1;
1565 let pruned = prune_cache_if_over_limit(&cfg, home.path())
1566 .unwrap()
1567 .expect("a real cache dir over its cap prunes");
1568 assert_eq!(pruned.files, 1);
1569 assert!(!dir.path().join("big").exists());
1570 }
1571
1572 fn touch(path: &Path, secs: u64) {
1575 let f = fs::File::options().write(true).open(path).unwrap();
1576 f.set_times(fs::FileTimes::new().set_modified(
1577 std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(secs),
1578 ))
1579 .unwrap();
1580 }
1581
1582 #[test]
1585 fn fold_unreadable_clears_a_run_whose_state_never_landed() {
1586 let dir = tempfile::tempdir().unwrap();
1587 let runs = dir.path().join("runs");
1588 let wt = dir.path().join("wt");
1589 let id = "20260904-014540-88c0";
1590 std::fs::create_dir_all(runs.join(id)).unwrap();
1591 std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1592
1593 let removed = block_on(fold_unreadable(&runs, &wt, id)).expect("fold by id");
1594 assert_eq!(removed, vec![format!("runs/{id}")]);
1595 assert!(!runs.join(id).exists(), "record gone");
1596
1597 std::fs::create_dir_all(runs.join(id)).unwrap();
1599 std::fs::write(runs.join(id).join("run.json.tmp"), b"").unwrap();
1600 assert!(
1601 block_on(fold_unreadable(&runs, &wt, "88c0")).is_ok(),
1602 "by prefix"
1603 );
1604
1605 std::fs::create_dir_all(runs.join("scratch")).unwrap();
1607 assert!(
1608 block_on(fold_unreadable(&runs, &wt, "scratch")).is_err(),
1609 "a stray directory is not a run"
1610 );
1611 }
1612
1613 #[test]
1614 fn fold_due_folds_terminal_runs_of_any_schema_but_leaves_genuinely_unreadable_ones() {
1615 let dir = tempfile::tempdir().unwrap();
1616 let runs = dir.path().join("runs");
1617 let wt = dir.path().join("wt");
1618 let home = dir.path().to_path_buf();
1619 let disk = Disk::default();
1620 let now = ts("2026-09-05T00:00:00Z");
1621
1622 let judging = "20260801-000000-0001";
1624 write_meta(&runs, judging, "judging", "2026-08-01T00:00:00Z");
1625
1626 let ready_fresh = "20260904-220000-0002";
1633 write_meta(&runs, ready_fresh, "ready", "2026-09-04T22:00:00Z");
1634
1635 let garbage = "20260901-000000-0004";
1641 std::fs::create_dir_all(runs.join(garbage)).unwrap();
1642 std::fs::write(runs.join(garbage).join("run.json"), "not json").unwrap();
1643 std::fs::create_dir_all(wt.join("0004")).unwrap();
1644
1645 let due_ready = due_run(&runs, &wt, "20260801-000000-ffff", SCHEMA);
1648
1649 let due_old_schema = due_run(&runs, &wt, "20260801-000000-eeee", SCHEMA - 1);
1654
1655 let (folded, unreadable) =
1656 block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
1657 assert_eq!(
1658 folded, 2,
1659 "both due, parseable runs fold regardless of their schema number"
1660 );
1661 assert_eq!(
1662 unreadable, 1,
1663 "only the run with broken JSON counts as unreadable"
1664 );
1665 assert!(runs.join(judging).exists(), "runnable never folded");
1666 assert!(runs.join(ready_fresh).exists(), "fresh never folded");
1667 assert!(runs.join(garbage).exists(), "unreadable record kept");
1668 assert!(wt.join("0004").exists(), "unreadable worktree kept");
1669 assert!(
1670 runs.join(&due_ready).exists(),
1671 "folding drops worktrees, not the record"
1672 );
1673 assert!(
1674 runs.join(&due_old_schema).exists(),
1675 "an old-schema record survives its fold exactly like a current one"
1676 );
1677 for id in [&due_ready, &due_old_schema] {
1683 let saved = read_meta(&runs, id).expect("folded run still parses");
1684 assert_ne!(
1685 saved.updated_at,
1686 ts("2026-08-01T00:00:00Z"),
1687 "fold_run must have saved the updated state back through the \
1688 `runs` directory this test passed to fold_due"
1689 );
1690 }
1691 }
1692
1693 #[tokio::test]
1699 async fn housekeep_leaves_everything_alone_when_auto_fold_is_disabled() {
1700 let dir = tempfile::tempdir().unwrap();
1701 let runs = dir.path().join("runs");
1702 let wt = dir.path().join("wt");
1703 let home = dir.path().to_path_buf();
1704 crate::run::set_home(dir.path().to_path_buf());
1705
1706 let due_id = due_run(&runs, &wt, "20260801-000000-abcd", SCHEMA);
1707 std::fs::create_dir_all(wt.join("orphan").join("cand-A")).unwrap();
1708
1709 let mut cfg = crate::config::Config::default();
1710 cfg.disk.auto_fold = false;
1711 cfg.disk.cache_limit_bytes = 0;
1712
1713 let out = housekeep(&cfg, &home, &wt, &dir.path().join("repo"), Timestamp::now()).await;
1714
1715 assert_eq!(out.folded, 0);
1716 assert_eq!(out.unreadable, 0);
1717 assert_eq!(out.orphaned_worktrees, 0);
1718 assert!(
1719 runs.join(&due_id).exists(),
1720 "a due run's record survives untouched"
1721 );
1722 assert!(
1723 wt.join("orphan").exists(),
1724 "an orphaned worktree survives untouched: the reclaim pass never ran"
1725 );
1726 }
1727
1728 #[test]
1729 fn fold_orphaned_worktrees_removes_only_worktrees_no_run_claims_and_none_in_flight() {
1730 let dir = tempfile::tempdir().unwrap();
1731 let runs = dir.path().join("runs");
1732 let wt = dir.path().join("wt");
1733 let home = dir.path().to_path_buf();
1734
1735 write_meta(
1738 &runs,
1739 "20260801-000000-aaaa",
1740 "ready",
1741 "2026-08-01T00:00:00Z",
1742 );
1743 std::fs::create_dir_all(wt.join("aaaa").join("cand-A")).unwrap();
1744
1745 std::fs::create_dir_all(wt.join("bbbb").join("cand-A")).unwrap();
1749
1750 std::fs::create_dir_all(wt.join("cccc")).unwrap();
1755
1756 std::fs::create_dir_all(wt.join("scratch")).unwrap();
1760
1761 let now = Timestamp::now() + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1767 let mut status = crate::daemon::Status::new();
1768 status.current = vec![crate::daemon::Current {
1769 worker: None,
1770 task: "20260905-000000-t111".to_owned(),
1771 run: "20260905-000000-cccc".to_owned(),
1772 }];
1773 status.updated_at = now;
1774 crate::daemon::write_status_to(&home.join("daemon.json"), &status).unwrap();
1775
1776 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1777 assert_eq!(
1778 folded, 1,
1779 "only the truly orphaned, idle, bay-shaped worktree is removed"
1780 );
1781 assert!(wt.join("aaaa").exists(), "claimed by a run record");
1782 assert!(!wt.join("bbbb").exists(), "orphaned and idle: reclaimed");
1783 assert!(wt.join("cccc").exists(), "a run in flight is never touched");
1784 assert!(
1785 wt.join("scratch").exists(),
1786 "not shaped like a worktree bay, so never a reclaim target"
1787 );
1788 }
1789
1790 #[test]
1798 fn fold_orphaned_worktrees_leaves_a_freshly_created_bay_alone() {
1799 let dir = tempfile::tempdir().unwrap();
1800 let runs = dir.path().join("runs");
1801 let wt = dir.path().join("wt");
1802 let home = dir.path().to_path_buf();
1803
1804 std::fs::create_dir_all(wt.join("dddd").join("under-review")).unwrap();
1805
1806 let now = Timestamp::now();
1807 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 6 * 60 * 60, now));
1808 assert_eq!(
1809 folded, 0,
1810 "too fresh to tell apart from a run still being set up"
1811 );
1812 assert!(wt.join("dddd").exists());
1813 }
1814
1815 fn open_question(store: &Questions, run: &str) -> crate::ask::Question {
1817 let mut q = crate::ask::Question::new(
1818 run.to_owned(),
1819 "implement".to_owned(),
1820 "impl-A".to_owned(),
1821 "Which storage backend should the cache use?".to_owned(),
1822 String::new(),
1823 vec!["SQLite".to_owned(), "Redis".to_owned()],
1824 );
1825 store.put(&mut q).unwrap();
1826 q
1827 }
1828
1829 #[test]
1835 fn a_finished_runs_open_question_is_swept_up() {
1836 let dir = tempfile::tempdir().unwrap();
1837 let runs = dir.path().join("runs");
1838 let store = Questions::at(dir.path().join("questions"));
1839
1840 let failed = "20260908-205802-c9eb";
1841 write_meta(&runs, failed, "failed", "2026-09-08T20:58:02Z");
1842 let failed_q = open_question(&store, failed);
1843
1844 let merged = "20260908-205501-ca67";
1845 write_meta(&runs, merged, "merged", "2026-09-08T20:55:01Z");
1846 let merged_q = open_question(&store, merged);
1847
1848 let n = abandon_settled_questions(&store, &runs);
1849 assert_eq!(n, 2, "both dead runs' questions are swept in one pass");
1850
1851 for (id, run) in [(&failed_q.id, failed), (&merged_q.id, merged)] {
1852 let back = store.get(id).unwrap();
1853 assert!(!back.status.open(), "{run} is done; nobody reads an answer");
1854 assert!(back.detail.contains(run), "{}", back.detail);
1855 }
1856 }
1857
1858 #[test]
1859 fn a_still_alive_runs_open_question_survives_the_sweep() {
1860 let dir = tempfile::tempdir().unwrap();
1861 let runs = dir.path().join("runs");
1862 let store = Questions::at(dir.path().join("questions"));
1863
1864 for (id, status) in [
1868 ("20260908-000000-b10c", "blocked"),
1869 ("20260908-000000-5ta1", "stalled"),
1870 ("20260908-000000-jud6", "judging"),
1871 ] {
1872 write_meta(&runs, id, status, "2026-09-08T00:00:00Z");
1873 let q = open_question(&store, id);
1874
1875 let n = abandon_settled_questions(&store, &runs);
1876 assert_eq!(n, 0, "{status} run is not done; nothing to sweep");
1877 assert!(
1878 store.get(&q.id).unwrap().status.open(),
1879 "{status} run's question must still be waiting"
1880 );
1881 }
1882 }
1883
1884 #[test]
1885 fn the_sweep_leaves_an_answered_question_and_an_unreadable_run_alone() {
1886 let dir = tempfile::tempdir().unwrap();
1887 let runs = dir.path().join("runs");
1888 let store = Questions::at(dir.path().join("questions"));
1889
1890 let done = "20260908-000000-answ";
1893 write_meta(&runs, done, "failed", "2026-09-08T00:00:00Z");
1894 let mut answered = open_question(&store, done);
1895 answered
1896 .answer(crate::ask::Answer::Choice("SQLite".to_owned()))
1897 .unwrap();
1898 store.put(&mut answered).unwrap();
1899
1900 let gone = "20260908-000000-gone";
1902 let orphan = open_question(&store, gone);
1903
1904 assert_eq!(abandon_settled_questions(&store, &runs), 0);
1905 assert_eq!(
1906 store.get(&answered.id).unwrap().status,
1907 crate::ask::QuestionStatus::Answered,
1908 "a real answer is never overwritten by a sweep"
1909 );
1910 assert!(
1911 store.get(&orphan.id).unwrap().status.open(),
1912 "a run this sweep cannot read is left exactly as it was, not guessed at"
1913 );
1914 }
1915
1916 #[test]
1925 fn fold_orphaned_worktrees_floors_a_zero_grace_at_the_race_safe_minimum() {
1926 let dir = tempfile::tempdir().unwrap();
1927 let runs = dir.path().join("runs");
1928 let wt = dir.path().join("wt");
1929 let home = dir.path().to_path_buf();
1930
1931 std::fs::create_dir_all(wt.join("eeee").join("under-review")).unwrap();
1932
1933 let now = Timestamp::now();
1935 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, now));
1936 assert_eq!(
1937 folded, 0,
1938 "a zero grace must not defeat the race-safety floor"
1939 );
1940 assert!(wt.join("eeee").exists());
1941
1942 let later = now + SignedDuration::new((MIN_ORPHAN_AGE_SECS + 1) as i64, 0);
1945 let folded = block_on(fold_orphaned_worktrees(&runs, &wt, &home, 0, later));
1946 assert_eq!(folded, 1, "old enough now, regardless of the zero grace");
1947 assert!(!wt.join("eeee").exists());
1948 }
1949
1950 #[test]
1951 fn clear_abandoned_active_only_acts_once_dead_and_overrun() {
1952 let dir = tempfile::tempdir().unwrap();
1953 crate::run::set_home(dir.path().to_path_buf());
1958 let home = dir.path().to_path_buf();
1959 let now = ts("2026-09-14T12:00:00Z");
1960 let overrun_seat = || crate::run::ActiveSeat {
1961 node: "implement".to_owned(),
1962 started_at: now - SignedDuration::new(21_000, 0),
1963 timeout_secs: 3_600,
1964 attempt: 0,
1965 task: None,
1966 command: None,
1967 index: None,
1968 total: None,
1969 };
1970
1971 let mut state = RunState::new(
1972 PathBuf::from("/repo"),
1973 "main".to_owned(),
1974 "abc1234".to_owned(),
1975 "fixture".to_owned(),
1976 crate::config::Config::default(),
1977 );
1978 state.status = RunStatus::Implementing;
1979 state.active.insert("impl-A".to_owned(), overrun_seat());
1980
1981 let mut fresh = state.clone();
1984 fresh.active.insert(
1985 "impl-B".to_owned(),
1986 crate::run::ActiveSeat {
1987 node: "implement".to_owned(),
1988 started_at: now,
1989 timeout_secs: 3_600,
1990 attempt: 0,
1991 task: None,
1992 command: None,
1993 index: None,
1994 total: None,
1995 },
1996 );
1997 assert!(!clear_abandoned_active(&mut fresh, &home, now).unwrap());
1998 assert!(!fresh.active.is_empty());
1999 assert_eq!(fresh.status, RunStatus::Implementing);
2000
2001 let store = Questions::at(home.join("questions"));
2002 let q = open_question(&store, &state.id);
2003
2004 assert!(clear_abandoned_active(&mut state, &home, now).unwrap());
2005 assert!(state.active.is_empty());
2006 assert_eq!(state.status, RunStatus::Failed);
2007 assert!(
2008 !store.get(&q.id).unwrap().status.open(),
2009 "the abandoned seat's own open question must not keep badging the \
2010 operator until some later daemon startup notices it"
2011 );
2012 }
2013
2014 fn write_meta(runs: &Path, id: &str, status: &str, updated_at: &str) {
2016 let day = &updated_at[..10];
2017 std::fs::create_dir_all(runs.join(id)).unwrap();
2018 let body = format!(
2019 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}}"#
2020 );
2021 std::fs::write(runs.join(id).join("run.json"), body).unwrap();
2022 }
2023
2024 fn due_run(runs: &Path, wt: &Path, id: &str, schema: u32) -> String {
2034 due_run_with_status(runs, wt, id, schema, RunStatus::Ready)
2035 }
2036
2037 fn due_run_with_status(
2040 runs: &Path,
2041 wt: &Path,
2042 id: &str,
2043 schema: u32,
2044 status: RunStatus,
2045 ) -> String {
2046 let mut config = crate::config::Config::default();
2047 config.graph.worktree_root = Some(wt.to_path_buf());
2048 let mut state = RunState::new(
2049 PathBuf::from("/nonexistent/repo"),
2050 "main".to_owned(),
2051 "0000000000000000000000000000000000000000".to_owned(),
2052 String::new(),
2053 config,
2054 );
2055 state.id = id.to_owned();
2056 state.status = status;
2057 state.updated_at = ts("2026-08-01T00:00:00Z");
2058 let mut value = serde_json::to_value(&state).unwrap();
2059 value["schema"] = serde_json::json!(schema);
2060 std::fs::create_dir_all(runs.join(id)).unwrap();
2061 std::fs::write(
2062 runs.join(id).join("run.json"),
2063 serde_json::to_string_pretty(&value).unwrap(),
2064 )
2065 .unwrap();
2066 id.to_owned()
2067 }
2068
2069 #[test]
2070 fn fold_due_folds_a_superseded_run_same_as_any_other_terminal_one() {
2071 let dir = tempfile::tempdir().unwrap();
2072 let runs = dir.path().join("runs");
2073 let wt = dir.path().join("wt");
2074 let home = dir.path().to_path_buf();
2075 let disk = Disk::default();
2076 let now = ts("2026-09-05T00:00:00Z");
2077
2078 let superseded = due_run_with_status(
2079 &runs,
2080 &wt,
2081 "20260801-000000-cccc",
2082 SCHEMA,
2083 RunStatus::Superseded,
2084 );
2085
2086 let (folded, unreadable) =
2087 block_on(fold_due(&runs, &home, &wt, &disk, now)).expect("fold_due");
2088 assert_eq!(
2089 folded, 1,
2090 "a superseded run has nothing left for a human to check, so it folds \
2091 exactly like a merged one"
2092 );
2093 assert_eq!(unreadable, 0);
2094 assert!(
2095 read_meta(&runs, &superseded).is_ok(),
2096 "folding drops the worktree, not the record"
2097 );
2098 }
2099
2100 fn init_repo(dir: &Path) {
2103 use crate::proc::Quiet as _;
2104 let run = |args: &[&str]| {
2105 let out = std::process::Command::new("git")
2106 .args(args)
2107 .current_dir(dir)
2108 .quiet()
2109 .output()
2110 .expect("spawn git");
2111 assert!(
2112 out.status.success(),
2113 "git {args:?} failed: {}",
2114 String::from_utf8_lossy(&out.stderr)
2115 );
2116 };
2117 run(&["init", "-b", "main"]);
2118 run(&["config", "user.name", "magi test"]);
2119 run(&["config", "user.email", "magi@example.com"]);
2120 std::fs::write(dir.join("README.md"), "# fixture\n").unwrap();
2121 run(&["add", "-A"]);
2122 run(&["commit", "-m", "init"]);
2123 }
2124
2125 async fn due_run_with_winner_worktree(
2129 runs: &Path,
2130 wt: &Path,
2131 repo: &Path,
2132 id: &str,
2133 status: RunStatus,
2134 ) -> (String, PathBuf) {
2135 use crate::run::{Candidate, Tally};
2136 use std::collections::BTreeMap;
2137
2138 let mut config = crate::config::Config::default();
2139 config.graph.worktree_root = Some(wt.to_path_buf());
2140 let mut state = RunState::new(
2141 repo.to_path_buf(),
2142 "main".to_owned(),
2143 "0000000000000000000000000000000000000000".to_owned(),
2144 String::new(),
2145 config,
2146 );
2147 state.id = id.to_owned();
2148 state.status = status;
2149 state.updated_at = ts("2026-08-01T00:00:00Z");
2150
2151 let winner_wt = state.worktree_root().join("cand-A");
2152 crate::git::worktree_add_branch(repo, &winner_wt, &format!("magi/{id}/A"), "main")
2153 .await
2154 .expect("winner worktree");
2155
2156 state.candidates.push(Candidate {
2157 index: 0,
2158 label: 'A',
2159 agent: "agent".to_owned(),
2160 branch: format!("magi/{id}/A"),
2161 worktree: winner_wt.clone(),
2162 summary: String::new(),
2163 stat: String::new(),
2164 files: 0,
2165 commits: 0,
2166 empty: false,
2167 failed: None,
2168 verified_noop: None,
2169 duration_ms: 0,
2170 folded: false,
2171 });
2172 state.tally = Some(Tally {
2173 first_choice: BTreeMap::from([('A', 1)]),
2174 borda: BTreeMap::new(),
2175 winner: 'A',
2176 rankings: 1,
2177 unanimous_initial: true,
2178 deliberated: false,
2179 changed_votes: 0,
2180 unanimous_final: true,
2181 tie_break: None,
2182 judges: 1,
2183 present: 1,
2184 quorum: 1,
2185 met_quorum: true,
2186 uncontested: None,
2187 });
2188 std::fs::create_dir_all(runs.join(id)).unwrap();
2189 std::fs::write(
2190 runs.join(id).join("run.json"),
2191 serde_json::to_string_pretty(&state).unwrap(),
2192 )
2193 .unwrap();
2194 (id.to_owned(), winner_wt)
2195 }
2196
2197 #[tokio::test]
2198 async fn completed_task_retains_independent_historical_sources_without_a_merge_proof() {
2199 let dir = tempfile::tempdir().unwrap();
2200 let home = dir.path();
2201 let runs = home.join("runs");
2202 let wt = home.join("wt");
2203 let repo = home.join("repo");
2204 std::fs::create_dir_all(&repo).unwrap();
2205 init_repo(&repo);
2206 let queue = crate::queue::Queue::at(home.join("queue"));
2207 let mut task = crate::queue::Task::new(
2208 "completed".into(),
2209 "retain independent historical changes".into(),
2210 repo.clone(),
2211 crate::queue::Source::Human,
2212 );
2213 let mut sources = Vec::new();
2214 for (id, status) in [
2215 ("20260801-000000-old1", RunStatus::Superseded),
2216 ("20260801-000000-old2", RunStatus::Failed),
2217 ] {
2218 let (id, worktree) = due_run_with_winner_worktree(&runs, &wt, &repo, id, status).await;
2219 std::fs::write(worktree.join("independent.txt"), &id).unwrap();
2220 crate::git::git(&worktree, &["add", "independent.txt"])
2221 .await
2222 .unwrap();
2223 crate::git::git(
2224 &worktree,
2225 &["commit", "-m", "Independent historical change"],
2226 )
2227 .await
2228 .unwrap();
2229 let head = crate::git::rev_parse(&worktree, "HEAD").await.unwrap();
2230 let mut state = RunState::load_under(&id, home).unwrap();
2231 state.driver_exited = true;
2232 if status == RunStatus::Failed {
2233 state.tally = None;
2234 }
2235 state.save_under(home).unwrap();
2236 task.link_run(&id);
2237 sources.push((id, worktree, head));
2238 }
2239 let mut latest = RunState::new(
2242 repo.clone(),
2243 "main".into(),
2244 crate::git::rev_parse(&repo, "main").await.unwrap(),
2245 task.instruction.clone(),
2246 crate::config::Config::default(),
2247 );
2248 latest.status = RunStatus::Ready;
2249 latest.driver_exited = true;
2250 latest.save_under(home).unwrap();
2251 task.start(latest.id.clone());
2252 task.succeed();
2253 queue.put(&mut task).unwrap();
2254
2255 let (folded, unreadable) = fold_due(
2256 &runs,
2257 home,
2258 &wt,
2259 &Disk::default(),
2260 ts("2026-09-05T00:00:00Z"),
2261 )
2262 .await
2263 .unwrap();
2264 assert_eq!((folded, unreadable), (0, 0));
2265 for (id, worktree, head) in sources {
2266 assert!(worktree.exists());
2267 assert_eq!(
2268 crate::git::rev_parse(&worktree, "HEAD").await.unwrap(),
2269 head
2270 );
2271 let mut state = RunState::load_under(&id, home).unwrap();
2272 assert!(!state.candidates[0].folded);
2273 let error = crate::graph::fold_run(&mut state, true, home)
2274 .await
2275 .unwrap_err();
2276 assert!(
2277 error
2278 .to_string()
2279 .contains(&format!("source of task {}", task.id))
2280 );
2281 assert_eq!(
2282 crate::git::rev_parse(&repo, &state.candidates[0].branch)
2283 .await
2284 .unwrap(),
2285 head
2286 );
2287 assert!(worktree.exists());
2288 assert!(runs.join(&id).join("run.json").exists());
2289 }
2290 }
2291
2292 #[tokio::test]
2293 async fn recovery_retains_historical_sources_after_success_and_automatic_cleanup() {
2294 let dir = tempfile::tempdir().unwrap();
2295 let home = dir.path();
2296 let runs = home.join("runs");
2297 let wt = home.join("wt");
2298 let repo = home.join("repo");
2299 std::fs::create_dir_all(&repo).unwrap();
2300 init_repo(&repo);
2301 let mut task = crate::queue::Task::new(
2302 "recover".into(),
2303 "keep measured behavior".into(),
2304 repo.clone(),
2305 crate::queue::Source::Human,
2306 );
2307 let mut sources = Vec::new();
2308 for (id, status) in [
2309 ("20260801-000000-old1", RunStatus::Blocked),
2310 ("20260801-000000-old2", RunStatus::Stalled),
2311 ] {
2312 let (id, worktree) = due_run_with_winner_worktree(&runs, &wt, &repo, id, status).await;
2313 let mut state = RunState::load_under(&id, home).unwrap();
2314 state.driver_exited = true;
2315 state.save_under(home).unwrap();
2316 task.link_run(&id);
2317 sources.push((id, worktree));
2318 }
2319 crate::pr_recovery::retain_sources(&task, home).unwrap();
2320 let mut recovered = RunState::new(
2321 repo.clone(),
2322 "main".into(),
2323 "base".into(),
2324 task.instruction.clone(),
2325 crate::config::Config::default(),
2326 );
2327 recovered.config.graph.worktree_root = Some(wt.clone());
2328 recovered.status = RunStatus::Ready;
2329 recovered.driver_exited = true;
2330 recovered.save_under(home).unwrap();
2331 task.start(recovered.id.clone());
2332 task.succeed();
2333 crate::daemon::supersede_prior_runs(&task, home);
2334 for (id, _) in &sources {
2335 assert_eq!(
2336 RunState::load_under(id, home).unwrap().status,
2337 RunStatus::Superseded
2338 );
2339 }
2340 let disk = Disk::default();
2341 let now = Timestamp::now()
2342 .checked_add(SignedDuration::from_secs(
2343 i64::try_from(disk.fold_grace_secs + 1).unwrap(),
2344 ))
2345 .unwrap();
2346 let (_, unreadable) = fold_due(&runs, home, &wt, &disk, now).await.unwrap();
2347 assert_eq!(unreadable, 0);
2348 for (id, worktree) in sources {
2349 assert!(worktree.exists());
2350 assert!(
2351 crate::git::branch_exists(&repo, &format!("magi/{id}/A"))
2352 .await
2353 .unwrap()
2354 );
2355 let mut historical = RunState::load_under(&id, home).unwrap();
2356 assert!(!historical.candidates[0].folded);
2357 let error = crate::graph::fold_run(&mut historical, true, home)
2358 .await
2359 .unwrap_err();
2360 assert!(
2361 error
2362 .to_string()
2363 .contains("explicitly retained recovery source")
2364 );
2365 assert!(!historical.candidates[0].folded);
2366 assert!(
2367 worktree.exists(),
2368 "every automatic fold entry preserves retained sources"
2369 );
2370 }
2371 }
2372
2373 #[tokio::test]
2374 async fn fold_due_drops_a_superseded_runs_own_winner_worktree_but_keeps_a_readys() {
2375 let dir = tempfile::tempdir().unwrap();
2376 let runs = dir.path().join("runs");
2377 let wt = dir.path().join("wt");
2378 let repo = dir.path().join("repo");
2379 let home = dir.path().to_path_buf();
2380 let disk = Disk::default();
2381 let now = ts("2026-09-05T00:00:00Z");
2382 std::fs::create_dir_all(&repo).unwrap();
2383 init_repo(&repo);
2384
2385 let (superseded_id, superseded_wt) = due_run_with_winner_worktree(
2386 &runs,
2387 &wt,
2388 &repo,
2389 "20260801-000000-supw",
2390 RunStatus::Superseded,
2391 )
2392 .await;
2393 let (ready_id, ready_wt) = due_run_with_winner_worktree(
2400 &runs,
2401 &wt,
2402 &repo,
2403 "20260801-000000-rdyw",
2404 RunStatus::Ready,
2405 )
2406 .await;
2407
2408 let (folded, unreadable) = fold_due(&runs, &home, &wt, &disk, now)
2409 .await
2410 .expect("fold_due");
2411 assert_eq!(folded, 2);
2412 assert_eq!(unreadable, 0);
2413
2414 assert!(
2415 !superseded_wt.exists(),
2416 "a superseded run's own winner never lands anywhere else, so its worktree \
2417 must be dropped exactly like a merged run's"
2418 );
2419 assert!(
2420 ready_wt.exists(),
2421 "a still-ready run's winner may yet be merged by hand - folding must not \
2422 touch it"
2423 );
2424
2425 assert!(read_meta(&runs, &superseded_id).is_ok());
2427 assert!(read_meta(&runs, &ready_id).is_ok());
2428 }
2429}