1use std::path::{Path, PathBuf};
50use std::sync::Arc;
51use std::sync::atomic::{AtomicBool, Ordering};
52use std::sync::{Mutex, MutexGuard};
53use std::time::Duration;
54
55use anyhow::{Context, Result, bail};
56use jiff::Timestamp;
57use serde::{Deserialize, Serialize};
58use tokio::sync::Notify;
59
60use crate::ask::{self, Questions};
61use crate::clean;
62use crate::conduct::Conductor;
63use crate::config::{Config, MergeMode};
64use crate::graph::Runner;
65use crate::land;
66use crate::notices::{self, Link, Notice};
67use crate::queue::{Queue, Task, TaskStatus};
68use crate::run::{Liveness, QuotaLoss, RunState, RunStatus};
69use crate::triage;
70
71pub const SCHEMA: u32 = 1;
73
74pub const HEARTBEAT: Duration = Duration::from_secs(5);
78
79pub const STALE_SECS: i64 = 30;
88
89pub const POLL: Duration = Duration::from_secs(5);
91
92pub const STALE_CLAIM: Duration = Duration::from_secs(6 * 60 * 60);
96
97pub const STALLED_RUNNING: Duration = Duration::from_secs(30 * 60);
116
117#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
119#[serde(default)]
120pub struct Current {
121 pub task: String,
123 pub run: String,
125}
126
127#[derive(Debug, Clone, Serialize, Deserialize)]
134pub struct Status {
135 pub schema: u32,
137 pub pid: u32,
139 pub started_at: Timestamp,
141 pub updated_at: Timestamp,
143 pub idle: bool,
145 pub current: Vec<Current>,
151 pub completed: usize,
153 pub polls: u64,
155}
156
157impl Status {
158 #[must_use]
160 pub fn new() -> Self {
161 let now = Timestamp::now();
162 Self {
163 schema: SCHEMA,
164 pid: std::process::id(),
165 started_at: now,
166 updated_at: now,
167 idle: true,
168 current: Vec::new(),
169 completed: 0,
170 polls: 0,
171 }
172 }
173}
174
175impl Default for Status {
176 fn default() -> Self {
177 Self::new()
178 }
179}
180
181#[derive(Debug, Clone)]
183pub struct Opts {
184 pub repo: PathBuf,
186 pub config: Option<PathBuf>,
188 pub poll: Duration,
190 pub max_attempts: usize,
192 pub once: bool,
194 pub merge: Option<String>,
196 pub worktrees_root: Option<PathBuf>,
205}
206
207impl Default for Opts {
208 fn default() -> Self {
209 Self {
210 repo: PathBuf::from("."),
211 config: None,
212 poll: POLL,
213 max_attempts: 2,
214 once: false,
215 merge: None,
216 worktrees_root: None,
217 }
218 }
219}
220
221fn max_concurrent(n: usize) -> usize {
226 n.max(1)
227}
228
229#[must_use]
231pub fn status_path() -> PathBuf {
232 crate::run::home().join("daemon.json")
233}
234
235pub fn write_status(status: &Status) -> Result<()> {
237 write_status_to(&status_path(), status)
238}
239
240pub fn write_status_to(path: &Path, status: &Status) -> Result<()> {
245 if let Some(parent) = path.parent() {
246 std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
247 }
248 let body = serde_json::to_string_pretty(status).context("serialize daemon status")?;
249 let tmp = path.with_extension("json.tmp");
250 std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
251 std::fs::rename(&tmp, path).with_context(|| format!("replace {}", path.display()))?;
252 Ok(())
253}
254
255pub fn clear_status() {
258 clear_status_at(&status_path());
259}
260
261fn clear_status_at(path: &Path) {
265 let _ = std::fs::remove_file(path);
266}
267
268#[derive(Debug, Clone, Default)]
281pub struct Stop {
282 stopped: Arc<AtomicBool>,
286 busy: Arc<std::sync::atomic::AtomicUsize>,
292 wake: Arc<Notify>,
296 pause: crate::graph::Pause,
299}
300
301impl Stop {
302 #[must_use]
304 pub fn new() -> Self {
305 Self::default()
306 }
307
308 pub fn stop(&self) {
311 self.stopped.store(true, Ordering::SeqCst);
312 self.wake.notify_one();
316 }
317
318 #[must_use]
320 pub fn stopped(&self) -> bool {
321 self.stopped.load(Ordering::SeqCst)
322 }
323
324 #[must_use]
332 pub fn finishing(&self) -> bool {
333 self.stopped() && self.busy_now()
334 }
335
336 pub fn park(&self, cause: crate::run::ParkCause) {
347 self.pause.park_for(cause);
348 self.stop();
349 }
350
351 #[must_use]
353 pub fn parking(&self) -> bool {
354 self.pause.parked()
355 }
356
357 #[must_use]
359 pub fn pause(&self) -> crate::graph::Pause {
360 self.pause.clone()
361 }
362
363 #[must_use]
369 pub fn busy_now(&self) -> bool {
370 self.busy.load(Ordering::SeqCst) > 0
371 }
372
373 fn enter(&self) {
375 self.busy.fetch_add(1, Ordering::SeqCst);
376 }
377
378 fn exit(&self) {
381 self.busy.fetch_sub(1, Ordering::SeqCst);
382 }
383
384 async fn idle(&self, poll: Duration) {
386 tokio::select! {
387 () = tokio::time::sleep(poll) => {}
388 () = self.wake.notified() => {}
389 }
390 }
391}
392
393#[derive(Debug, Clone, Default, Deserialize)]
400#[serde(default)]
401pub struct Reading {
402 pub schema: u32,
404 pub pid: Option<u32>,
406 pub started_at: Option<Timestamp>,
408 pub updated_at: Option<Timestamp>,
410 pub idle: bool,
412 #[serde(deserialize_with = "de_current")]
424 pub current: Vec<Current>,
425 pub completed: u64,
427 pub polls: u64,
429}
430
431fn de_current<'de, D>(deserializer: D) -> std::result::Result<Vec<Current>, D::Error>
434where
435 D: serde::Deserializer<'de>,
436{
437 #[derive(Deserialize)]
438 #[serde(untagged)]
439 enum Shape {
440 Many(Vec<Current>),
441 One(Current),
442 }
443 Ok(
444 Option::<Shape>::deserialize(deserializer)?.map_or_else(Vec::new, |shape| match shape {
445 Shape::Many(v) => v,
446 Shape::One(c) => vec![c],
447 }),
448 )
449}
450
451impl Reading {
452 #[must_use]
455 pub fn age_secs(&self, now: Timestamp) -> Option<i64> {
456 self.updated_at
457 .map(|at| (now.as_second() - at.as_second()).max(0))
458 }
459
460 #[must_use]
464 pub fn running(&self, now: Timestamp) -> bool {
465 self.age_secs(now).is_some_and(|secs| secs <= STALE_SECS)
466 }
467}
468
469#[must_use]
476pub fn read_status(home: &Path) -> Option<Reading> {
477 let body = std::fs::read_to_string(home.join("daemon.json")).ok()?;
478 serde_json::from_str(&body).ok()
479}
480
481#[must_use]
493pub fn current_work(home: &Path, now: Timestamp) -> Vec<Current> {
494 read_status(home)
495 .filter(|reading| reading.running(now))
496 .map(|reading| reading.current)
497 .unwrap_or_default()
498}
499
500#[must_use]
502pub fn is_working_on(home: &Path, run: &str, now: Timestamp) -> bool {
503 current_work(home, now).iter().any(|c| c.run == run)
504}
505
506#[must_use]
516pub fn is_working_on_short(home: &Path, short: &str, now: Timestamp) -> bool {
517 current_work(home, now)
518 .iter()
519 .any(|c| crate::run::short_of(&c.run) == short)
520}
521
522#[must_use]
524pub fn is_working_on_task(home: &Path, task: &str, now: Timestamp) -> bool {
525 current_work(home, now).iter().any(|c| c.task == task)
526}
527
528pub fn sweep_stale_claims(queue: &Queue, older_than: Duration) -> Vec<String> {
572 sweep_stale_claims_with(queue, older_than, crate::proc::pid_alive)
573}
574
575fn sweep_stale_claims_with<F>(queue: &Queue, older_than: Duration, pid_alive: F) -> Vec<String>
579where
580 F: Fn(u32) -> bool,
581{
582 let this_process = std::process::id();
583 let mut swept: Vec<String> = std::fs::read_dir(queue.root())
584 .into_iter()
585 .flatten()
586 .flatten()
587 .map(|e| e.path())
588 .filter(|p| p.extension().is_some_and(|x| x == "lock"))
589 .filter(|p| {
590 match std::fs::read_to_string(p)
591 .ok()
592 .and_then(|body| body.trim().parse::<u32>().ok())
593 {
594 Some(pid) if pid == this_process => false,
598 Some(pid) => !pid_alive(pid),
599 None => p
600 .metadata()
601 .and_then(|m| m.modified())
602 .and_then(|t| t.elapsed().map_err(std::io::Error::other))
603 .is_ok_and(|age| age >= older_than),
604 }
605 })
606 .filter(|p| std::fs::remove_file(p).is_ok())
607 .filter_map(|p| {
608 p.file_stem()
609 .and_then(|s| s.to_str())
610 .map(std::borrow::ToOwned::to_owned)
611 })
612 .collect();
613 swept.sort_unstable();
614 swept
615}
616
617fn is_stalled(task: &Task, home: &Path, now: Timestamp) -> bool {
622 task.status == TaskStatus::Running
623 && (now.as_second() - task.updated_at.as_second()) >= STALLED_RUNNING.as_secs() as i64
624 && !is_working_on_task(home, &task.id, now)
625}
626
627fn stalled_tasks(queue: &Queue, home: &Path, now: Timestamp) -> Vec<Task> {
630 queue
631 .list()
632 .into_iter()
633 .filter(|t| is_stalled(t, home, now))
634 .collect()
635}
636
637fn queued_tasks(queue: &Queue) -> Vec<Task> {
643 queue
644 .list()
645 .into_iter()
646 .filter(|t| t.status == TaskStatus::Queued)
647 .collect()
648}
649
650fn finished_tasks(queue: &Queue) -> Vec<Task> {
653 queue
654 .list()
655 .into_iter()
656 .filter(|t| matches!(t.status, TaskStatus::Failed | TaskStatus::Held))
657 .collect()
658}
659
660fn resolve_blockers(queue: &Queue, questions: &Questions) {
677 for listed in queue.list() {
678 if listed.status != TaskStatus::Blocked || listed.blocked_by.is_empty() {
679 continue;
680 }
681 let Ok(_claim) = queue.claim(&listed.id) else {
682 continue;
683 };
684 let Ok(mut task) = queue.get(&listed.id) else {
685 continue;
686 };
687 if task.status != TaskStatus::Blocked {
688 continue;
689 }
690 let deleted = queue.apply_deleted_blockers(&mut task);
694 if !deleted.is_empty() {
695 record(queue, &mut task);
696 for id in &deleted {
697 queue.note_dependency_deleted(&task, id);
698 }
699 if task.status != TaskStatus::Blocked {
700 continue;
701 }
702 }
703 let missing = crate::queue::missing_blockers(queue, questions, &task.blocked_by);
704 if !missing.is_empty() {
705 let language = language_of(&task, Path::new("."));
706 task.hold_machine(Some(crate::queue::missing_blocker_hold_reason_in(
707 &task.blocked_by,
708 &missing,
709 &language,
710 )));
711 record(queue, &mut task);
712 continue;
713 }
714 if let Some(q) = task.blocked_by.iter().find_map(|id| {
717 questions.get(id).ok().filter(|q| {
718 q.node == crate::conduct::NODE && q.status == ask::QuestionStatus::Abandoned
719 })
720 }) {
721 let language = language_of(&task, Path::new("."));
722 task.hold_machine(Some(unanswered_question_hold_reason(&q, &language)));
723 record(queue, &mut task);
724 continue;
725 }
726 let mut changed = false;
727 for id in task.blocked_by.clone() {
728 if let Ok(dep) = queue.get(&id) {
729 if dep.status == TaskStatus::Done {
730 task.unblock(&id);
731 changed = true;
732 }
733 continue;
734 }
735 if let Ok(q) = questions.get(&id)
736 && q.status == ask::QuestionStatus::Answered
737 {
738 let answer = match &q.answer {
739 Some(ask::Answer::Choice(c) | ask::Answer::Text(c)) => c.clone(),
740 None => String::new(),
741 };
742 task.record_answer(q.summary.clone(), answer);
743 task.unblock(&id);
744 changed = true;
745 }
746 }
747 if changed {
748 record(queue, &mut task);
749 }
750 }
751}
752
753fn unanswered_question_hold_reason(q: &ask::Question, language: &str) -> String {
755 if crate::lang::is_japanese(language) {
756 format!(
757 "質問 {} 「{}」 に期限内の回答がなく、取り下げられました - `magi task triage` を参照",
758 q.short(),
759 q.summary
760 )
761 } else {
762 format!(
763 "question {} \"{}\" went unanswered and was abandoned - see `magi task triage`",
764 q.short(),
765 q.summary
766 )
767 }
768}
769
770#[derive(Debug, Clone, Copy, PartialEq, Eq)]
773pub(crate) enum ActionStanding {
774 NoAction,
776 Pending,
778 Applied,
781 Stale,
783 Busy,
785}
786
787impl ActionStanding {
788 pub(crate) fn daemon_owns(self) -> bool {
790 matches!(self, Self::Pending | Self::Applied | Self::Stale)
791 }
792}
793
794pub(crate) fn action_standing(task: &Task, q: &ask::Question) -> ActionStanding {
798 if q.chosen_action().is_none() {
799 ActionStanding::NoAction
800 } else if task.action_applied(&q.id) {
801 ActionStanding::Applied
802 } else if q.node != crate::conduct::NODE && task.runs.last() != Some(&q.run) {
803 ActionStanding::Stale
804 } else if matches!(
805 task.status,
806 TaskStatus::Running | TaskStatus::Blocked | TaskStatus::Done
807 ) {
808 ActionStanding::Busy
809 } else {
810 ActionStanding::Pending
811 }
812}
813
814#[derive(Debug, Clone, PartialEq, Eq)]
816enum ActionDecision {
817 Skip,
819 Resume(String),
821 Requeue,
823 Done,
825 Stale,
828 Refuse(String),
831}
832
833fn decide_action<F>(task: &Task, q: &ask::Question, p: &Phrases, load: F) -> ActionDecision
841where
842 F: FnOnce(&str) -> Result<RunState>,
843{
844 let Some(action) = q.chosen_action() else {
845 return ActionDecision::Skip;
846 };
847 if task.action_applied(&q.id)
848 || matches!(
849 task.status,
850 TaskStatus::Running | TaskStatus::Blocked | TaskStatus::Done
851 )
852 {
853 return ActionDecision::Skip;
854 }
855 if q.node != crate::conduct::NODE && task.runs.last() != Some(&q.run) {
858 return ActionDecision::Stale;
859 }
860 let repair = q.node == land::APPROVAL_NODE && q.resolution().as_deref() == Some(land::REPAIR);
861 let decision_at = if repair {
865 q.answered_at.unwrap_or(q.asked_at)
866 } else {
867 q.asked_at
868 };
869 if task.operator_decided_at.is_some_and(|at| at > decision_at) {
870 return if repair {
871 ActionDecision::Refuse(format!(
872 "repair answer {} was superseded by a later operator decision; the task stays held. Inspect `magi show {}`; if repair is still wanted, explicitly release the existing owner with `magi task release {}`",
873 q.id, q.run, task.id
874 ))
875 } else {
876 ActionDecision::Stale
877 };
878 }
879 match action {
880 ask::ChoiceAction::Requeue => ActionDecision::Requeue,
881 ask::ChoiceAction::Done => ActionDecision::Done,
882 ask::ChoiceAction::Resume { run } => {
883 if task.runs.last() != Some(run) {
884 return ActionDecision::Refuse((p.resume_not_latest)(
885 q.short(),
886 ask::short_id(run),
887 ));
888 }
889 match load(run) {
890 Ok(s)
891 if q.node == land::APPROVAL_NODE
892 && q.resolution().as_deref() == Some(land::REPAIR) =>
893 {
894 if s.merge_repairs.iter().any(|r| r.question == q.id)
895 || s.land_approval.as_ref().is_some_and(|a| a.question != q.id)
896 {
897 ActionDecision::Stale
900 } else if s.liveness(false) == Liveness::Live {
901 ActionDecision::Skip
902 } else if s.status != RunStatus::Landing
903 || s.released()
904 || !land::repair_available(&s)
905 || !implementers_match(task, &s)
906 || !s
907 .land_approval
908 .as_ref()
909 .is_some_and(|a| a.question == q.id && !a.head.is_empty())
910 {
911 ActionDecision::Refuse(format!(
912 "repair question {} no longer names a supported current approval on run {}; inspect `magi show {}` and restore the existing owner's approval before retrying; do not file another task",
913 q.id, run, run
914 ))
915 } else {
916 ActionDecision::Resume(run.clone())
917 }
918 }
919 Ok(s) if s.status.resumable() && !s.released() && !exhausted_review_budget(&s) => {
920 ActionDecision::Resume(run.clone())
921 }
922 Ok(_) => ActionDecision::Refuse((p.resume_cannot_progress)(
923 q.short(),
924 ask::short_id(run),
925 )),
926 Err(e) => ActionDecision::Refuse((p.resume_unreadable)(
927 q.short(),
928 ask::short_id(run),
929 &format!("{e:#}"),
930 )),
931 }
932 }
933 }
934}
935
936struct Phrases {
950 graph_stopped: fn(&str, &str) -> String,
952 quorum_lost: &'static str,
953 quota_took_out: &'static str,
955 run_ended: &'static str,
957 waiting_for_answer: &'static str,
959 recovered_running: &'static str,
960 no_run_to_recover: &'static str,
961 could_not_start: &'static str,
962 resume_not_latest: fn(&str, &str) -> String,
963 resume_cannot_progress: fn(&str, &str) -> String,
964 resume_unreadable: fn(&str, &str, &str) -> String,
965 handover_refused: fn(&crate::handover::Refused) -> String,
968 handover_hint: &'static str,
970 reviewers_never_answered: fn(usize, usize) -> String,
973}
974
975const PHRASES_EN: Phrases = Phrases {
976 graph_stopped: |status, detail| {
977 format!("the graph stopped at `{status}` without reaching a terminal status: {detail}")
978 },
979 quorum_lost: "the judging panel lost its quorum",
980 quota_took_out: "; quota took out ",
981 run_ended: "run ended ",
982 waiting_for_answer: " - waiting for operator answer to question ",
983 recovered_running: "recovered a `running` task whose daemon never recorded the outcome: ",
984 no_run_to_recover: "task was `running` with no live daemon and no readable \
985 run to recover; held for a human to check what happened",
986 could_not_start: "could not start the run: ",
987 resume_not_latest: |q, run| {
988 format!("question {q} asked to resume run {run}, which is not this task's latest run")
989 },
990 resume_cannot_progress: |q, run| {
991 format!("question {q} asked to resume run {run}, which cannot make progress")
992 },
993 resume_unreadable: |q, run, e| {
994 format!("question {q} asked to resume run {run}, which could not be read: {e}")
995 },
996 handover_refused: |r| r.to_string(),
997 handover_hint: " (clean up the other worktree, then release the task from the queue)",
998 reviewers_never_answered: |missing, rounds| {
999 format!(
1000 "{missing} reviewer seat(s) never answered after {rounds} rounds; \
1001 refusing to call it clean"
1002 )
1003 },
1004};
1005
1006const PHRASES_JA: Phrases = Phrases {
1007 graph_stopped: |status, detail| {
1008 format!("グラフが終端状態に達しないまま `{status}` で停止しました: {detail}")
1009 },
1010 quorum_lost: "審査パネルが定足数を失いました",
1011 quota_took_out: "。クォータで脱落: ",
1012 run_ended: "run 終了: ",
1013 waiting_for_answer: " - オペレーターの回答待ち: 質問 ",
1014 recovered_running: "daemon が結果を記録しないまま `running` だったタスクを回収しました: ",
1015 no_run_to_recover: "タスクは `running` でしたが、生きた daemon も回収できる run も見つかりません。\
1016 何が起きたか人が確認するため保留にしました",
1017 could_not_start: "run を開始できませんでした: ",
1018 resume_not_latest: |q, run| {
1019 format!(
1020 "質問 {q} は run {run} の再開を求めましたが、これはタスクの最新の run ではありません"
1021 )
1022 },
1023 resume_cannot_progress: |q, run| {
1024 format!("質問 {q} は run {run} の再開を求めましたが、これは進行できません")
1025 },
1026 resume_unreadable: |q, run, e| {
1027 format!("質問 {q} は run {run} の再開を求めましたが、読み込めませんでした: {e}")
1028 },
1029 handover_refused: |r| {
1030 use crate::handover::Refused;
1031 match r {
1032 Refused::Foreign { branch, path, why } => format!(
1033 "ブランチ `{branch}` は {path} にチェックアウトされており、magi は自動では\
1034 削除しません。不要なら `git worktree remove` でその worktree を削除して\
1035 から、やり直してください(詳細: {why})"
1036 ),
1037 Refused::Unsafe { branch, path, why } => format!(
1038 "ブランチ `{branch}` は {path} にチェックアウトされています。そこの作業を\
1039 コミットか破棄したうえで `git worktree remove` で worktree を削除する\
1040 か、run を破棄してよいと伝えてから、やり直してください(詳細: {why})"
1041 ),
1042 Refused::ReleaseFailed { branch, path, run } => format!(
1043 "ブランチ `{branch}` は {path} で run {run} がチェックアウトしており、その\
1044 worktree の解放に失敗したか、変更が見つかりました(未コミットの変更が\
1045 ある worktree は git が削除を拒否します)。worktree はそのまま残しました"
1046 ),
1047 }
1048 },
1049 handover_hint: "(他の worktree を片付けてから、タスクをキューから解放してください)",
1050 reviewers_never_answered: |missing, rounds| {
1051 format!(
1052 "{missing} 席のレビュアーが {rounds} ラウンドの間に一度も回答しなかったため、\
1053 クリーンとは認めません"
1054 )
1055 },
1056};
1057
1058fn phrases(language: &str) -> &'static Phrases {
1059 if crate::lang::is_japanese(language) {
1060 &PHRASES_JA
1061 } else {
1062 &PHRASES_EN
1063 }
1064}
1065
1066fn language_of(task: &Task, fallback: &Path) -> String {
1070 crate::lang::of_repo(&repo_for(task, fallback))
1071}
1072
1073pub(crate) fn task_of_question<'a>(tasks: &'a [Task], q: &ask::Question) -> Option<&'a Task> {
1077 if q.node == crate::conduct::NODE {
1078 return tasks.iter().find(|t| t.id == q.run);
1079 }
1080 tasks.iter().find(|t| t.runs.contains(&q.run))
1081}
1082
1083pub(crate) fn apply_choice_actions(queue: &Queue, questions: &Questions, home: &Path) {
1092 let tasks = queue.list();
1093 for q in questions.list() {
1094 if q.chosen_action().is_none() {
1095 continue;
1096 }
1097 let Some(listed) = task_of_question(&tasks, &q) else {
1098 if q.node == land::APPROVAL_NODE && q.resolution().as_deref() == Some(land::REPAIR) {
1099 notices::raise_in(
1100 home,
1101 Notice::warn(
1102 &format!("repair-owner:{}", q.id),
1103 format!(
1104 "Repair is waiting: no task owns run {}. Inspect `magi show {}` and restore its original queue owner; do not file a competing repair task.",
1105 q.run, q.run
1106 ),
1107 ),
1108 );
1109 }
1110 continue;
1111 };
1112 if listed.action_applied(&q.id) {
1113 continue;
1114 }
1115 let Ok(_claim) = queue.claim(&listed.id) else {
1116 continue;
1117 };
1118 let Ok(mut task) = queue.get(&listed.id) else {
1119 continue;
1120 };
1121 let language = language_of(&task, Path::new("."));
1122 let decision = decide_action(&task, &q, phrases(&language), |id| {
1123 RunState::load_under(id, home)
1124 });
1125 let ran = matches!(
1126 decision,
1127 ActionDecision::Resume(_) | ActionDecision::Requeue | ActionDecision::Done
1128 );
1129 if ran && asker_may_still_read(questions, &q, home) {
1130 continue;
1139 }
1140 let done = matches!(decision, ActionDecision::Done);
1141 match decision {
1142 ActionDecision::Skip => continue,
1143 ActionDecision::Resume(run) => {
1144 task.release_operator("choice", Some(&q.id));
1145 task.resume_override = Some(crate::queue::OperatorResume {
1146 question_id: q.id.clone(),
1147 at: Timestamp::now(),
1148 conductor_rehold: None,
1149 forced: true,
1150 pinned_run: Some(run),
1151 });
1152 }
1153 ActionDecision::Stale => {}
1154 ActionDecision::Requeue => {
1155 task.release_operator("choice-requeue", Some(&q.id));
1156 task.requeue();
1157 }
1158 ActionDecision::Done => task.succeed(),
1159 ActionDecision::Refuse(why) => {
1160 if task.operator_held() {
1161 task.hold_reason = Some(match task.hold_reason.take() {
1164 Some(reason) => format!("{reason}\n\n{why}"),
1165 None => why,
1166 });
1167 } else {
1168 task.hold_machine(Some(why));
1169 }
1170 notices::raise(
1171 Notice::warn(
1172 &format!("action:{}", q.id),
1173 "An answer asked the daemon to resume a run that cannot be resumed; the task stays held.",
1174 )
1175 .link(Link::Task {
1176 id: task.id.clone(),
1177 }),
1178 );
1179 }
1180 }
1181 task.mark_action_applied(&q.id);
1182 record(queue, &mut task);
1185 if done && queue.get(&task.id).is_ok_and(|t| t.action_applied(&q.id)) {
1186 supersede_prior_runs(&task, home);
1187 }
1188 let _ = questions.update(&q.id, |r| {
1191 r.answer_delivered = true;
1192 Ok(())
1193 });
1194 }
1195}
1196
1197fn asker_may_still_read(questions: &Questions, q: &ask::Question, home: &Path) -> bool {
1201 let now = Timestamp::now();
1202 if questions.read_lease(&q.id).is_some_and(|l| l.fresh(now)) {
1203 return true;
1204 }
1205 q.node != crate::conduct::NODE
1206 && RunState::load_under(&q.run, home).is_ok_and(|s| {
1207 s.seats_active()
1208 .any(|(k, a)| *k == q.seat && a.remaining_secs(now) > 0)
1209 })
1210}
1211
1212fn reconcile_task_questions(queue: &Queue, questions: &Questions) {
1223 let tasks = queue.list();
1224 let by_id: std::collections::BTreeMap<&str, &Task> =
1225 tasks.iter().map(|t| (t.id.as_str(), t)).collect();
1226 let referenced: std::collections::BTreeSet<&str> = tasks
1227 .iter()
1228 .flat_map(|task| task.blocked_by.iter().map(String::as_str))
1229 .collect();
1230
1231 for mut question in questions.list() {
1232 if !question.status.open() || question.node != crate::conduct::NODE {
1233 continue;
1234 }
1235 if referenced.contains(question.id.as_str()) {
1238 continue;
1239 }
1240 let Some(task) = by_id.get(question.run.as_str()) else {
1241 continue;
1242 };
1243 question.abandon(format!(
1244 "task {} no longer waits for this answer",
1245 task.short()
1246 ));
1247 if let Err(e) = questions.put(&mut question) {
1248 tracing::warn!(
1249 "could not retire question {} for task {}: {e:#}",
1250 question.short(),
1251 task.short()
1252 );
1253 }
1254 }
1255}
1256
1257#[derive(Debug, Clone, Copy)]
1264pub struct Verdict {
1265 pub status: RunStatus,
1267 pub left_pr: bool,
1269 pub quota_hit: bool,
1271 pub parked: bool,
1273 pub no_viable_candidates: bool,
1280}
1281
1282pub fn settle(task: &mut Task, verdict: Verdict, detail: &str, max_attempts: usize) {
1339 settle_in(task, verdict, detail, max_attempts, &PHRASES_EN)
1340}
1341
1342fn settle_in(task: &mut Task, verdict: Verdict, detail: &str, max_attempts: usize, p: &Phrases) {
1344 if verdict.parked {
1350 task.park(detail);
1351 return;
1352 }
1353 match verdict.status {
1354 RunStatus::Merged | RunStatus::Ready => task.succeed(),
1355 RunStatus::AlreadyInBase => task.already_landed(detail),
1356 RunStatus::Stalled if verdict.quota_hit => task.stall(detail),
1357 RunStatus::Failed if verdict.quota_hit && verdict.no_viable_candidates => {
1358 task.stall(detail)
1359 }
1360 RunStatus::Stalled | RunStatus::Failed => task.fail(detail, max_attempts),
1361 RunStatus::Blocked if verdict.left_pr => task.handed_off(detail),
1362 RunStatus::Blocked => task.fail(detail, max_attempts),
1363 RunStatus::VerifiedNoop => task.handed_off(detail),
1364 other => task.fail((p.graph_stopped)(label(other), detail), max_attempts),
1365 }
1366}
1367
1368pub fn supersede_prior_runs(task: &Task, home: &Path) {
1417 let now = Timestamp::now();
1418 let last_run_succeeded = task
1425 .runs
1426 .last()
1427 .and_then(|id| RunState::load_under(id, home).ok())
1428 .is_some_and(|s| matches!(s.status, RunStatus::Merged | RunStatus::Ready));
1429 for id in task.superseded_attempts(last_run_succeeded) {
1430 let mut state = match RunState::load_under(id, home) {
1431 Ok(s) => s,
1432 Err(e) => {
1433 tracing::warn!("could not load run {id} to mark it superseded: {e:#}");
1434 continue;
1435 }
1436 };
1437 if !matches!(state.status, RunStatus::Blocked | RunStatus::Stalled) {
1438 continue;
1439 }
1440 let daemon_claims = is_working_on(home, id, now);
1441 if state.liveness(daemon_claims) == Liveness::Live {
1442 continue;
1443 }
1444 state.status = RunStatus::Superseded;
1445 if let Err(e) = state.save_under(home) {
1446 tracing::warn!("could not mark run {id} superseded: {e:#}");
1447 }
1448 }
1449}
1450
1451fn resweep_superseded_attempts(queue: &Queue, home: &Path) {
1473 for task in queue.list() {
1474 if task.status != TaskStatus::Done || task.runs.len() < 2 {
1475 continue;
1476 }
1477 supersede_prior_runs(&task, home);
1478 }
1479}
1480
1481fn settle_and_diagnose(
1488 task: &mut Task,
1489 verdict: Verdict,
1490 detail: &str,
1491 max_attempts: usize,
1492 state: &RunState,
1493) {
1494 let p = phrases(&state.config.graph.language);
1495 let attempts_before = task.attempts;
1496 if let Some(block) = state.resource_block.as_ref().filter(|_| !verdict.parked) {
1502 task.last_error = Some(detail.to_owned());
1503 task.diagnostic = None;
1504 task.attempts = task.attempts.saturating_sub(1);
1505 task.hold_machine(Some(format!(
1506 "{}{} stayed busy past the resource-wait limit before `{}` could start; \
1507 no agent was asked and no attempt was spent",
1508 crate::cache::RESOURCE_HOLD,
1509 block.kind,
1510 block.node
1511 )));
1512 if let Some(pass) = state.passes.last() {
1513 task.record_pass(&state.id, pass.index, attempts_before, None);
1514 }
1515 return;
1516 }
1517 settle_in(task, verdict, detail, max_attempts, p);
1518 if let Some(pass) = state.passes.last() {
1521 let parked = match &pass.end {
1522 Some(crate::run::PassEnd::Parked { cause }) => Some(cause.clone()),
1523 _ => None,
1524 };
1525 task.record_pass(&state.id, pass.index, attempts_before, parked);
1526 }
1527 if task.status == TaskStatus::Held {
1528 task.diagnostic = diagnostic(state);
1529 note_open_question(task, &state.id, p);
1530 }
1531}
1532
1533fn note_open_question(task: &mut Task, run: &str, p: &Phrases) {
1548 let Some(home) = crate::run::try_home() else {
1549 return;
1550 };
1551 let open = Questions::at(home.join("questions")).open_for(run);
1552 let Some(q) = open.first() else {
1553 return;
1554 };
1555 let base = task.hold_reason.clone().unwrap_or_default();
1556 task.hold_reason = Some(format!("{base}{}{}", p.waiting_for_answer, q.short()));
1557}
1558
1559fn reclaim(task: &mut Task, last_run: Option<RunState>, max_attempts: usize, language: &str) {
1570 match last_run {
1571 Some(state) => {
1572 let verdict = Verdict {
1573 status: state.status,
1574 left_pr: state.pr.is_some(),
1575 quota_hit: !state.quota.is_empty(),
1576 parked: state.parked,
1577 no_viable_candidates: state.viable().is_empty(),
1578 };
1579 let detail = format!(
1580 "{}{}",
1581 phrases(&state.config.graph.language).recovered_running,
1582 describe(&state)
1583 );
1584 settle_and_diagnose(task, verdict, &detail, max_attempts, &state);
1585 }
1586 None => {
1587 let why = phrases(language).no_run_to_recover;
1590 task.last_error = Some(why.to_owned());
1591 task.hold_machine(Some(why.to_owned()));
1594 }
1595 }
1596}
1597
1598fn reclaim_orphaned_running(queue: &Queue, max_attempts: usize) -> Vec<String> {
1620 let mut reclaimed = Vec::new();
1621 for listed in queue.list() {
1622 if listed.status != TaskStatus::Running {
1623 continue;
1624 }
1625 let Ok(_claim) = queue.claim(&listed.id) else {
1626 continue;
1627 };
1628 let Ok(mut task) = queue.get(&listed.id) else {
1632 continue;
1633 };
1634 if task.status != TaskStatus::Running {
1635 continue;
1636 }
1637 let last_run = task.runs.last().and_then(|id| RunState::load(id).ok());
1638 if let Some(state) = &last_run
1648 && let Err(e) = ask::Questions::open().settle_run(&state.id, state.status)
1649 {
1650 tracing::warn!("abandon questions for {}: {e:#}", state.id);
1651 }
1652 let language = if last_run.is_none() {
1653 language_of(&task, Path::new("."))
1654 } else {
1655 String::new()
1656 };
1657 reclaim(&mut task, last_run, max_attempts, &language);
1658 if task.status == TaskStatus::Done {
1659 supersede_prior_runs(&task, &crate::run::home());
1660 }
1661 record(queue, &mut task);
1662 reclaimed.push(task.id.clone());
1663 }
1664 reclaimed
1665}
1666
1667fn reclaim_abandoned_runs(home: &Path, now: Timestamp) -> Vec<String> {
1699 reclaim_abandoned_runs_with(
1700 home,
1701 now,
1702 crate::proc::pid_status,
1703 crate::proc::process_started_at,
1704 )
1705}
1706
1707fn reclaim_abandoned_runs_with<F, G>(
1713 home: &Path,
1714 now: Timestamp,
1715 query: F,
1716 identity: G,
1717) -> Vec<String>
1718where
1719 F: Fn(u32) -> Option<bool>,
1720 G: Fn(u32) -> Option<String>,
1721{
1722 let mut abandoned = Vec::new();
1723 for entry in std::fs::read_dir(home.join("runs"))
1724 .into_iter()
1725 .flatten()
1726 .flatten()
1727 {
1728 let id = entry.file_name().to_string_lossy().into_owned();
1729 if !crate::run::is_run_id(&id) {
1730 continue;
1731 }
1732 let Ok(body) = std::fs::read_to_string(entry.path().join("run.json")) else {
1740 continue;
1741 };
1742 let Ok(mut state) = serde_json::from_str::<RunState>(&body) else {
1743 continue;
1744 };
1745 if state.status.done() || !state.active_all_overrun(now) {
1746 continue;
1747 }
1748 let daemon_claims = is_working_on(home, &id, now);
1760 if state.liveness_with(daemon_claims, &query, &identity) != crate::run::Liveness::Dead {
1761 continue;
1762 }
1763 state.abandon("daemon");
1764 if let Err(e) = state.save_under(home) {
1765 tracing::warn!("could not persist abandoned run {id}: {e:#}");
1766 continue;
1767 }
1768 if let Some(notice) = notices::run_ended(&state) {
1771 notices::raise_in(home, notice);
1772 }
1773 if let Err(e) = Questions::at(home.join("questions")).settle_run(&id, state.status) {
1781 tracing::warn!("abandon questions for {id}: {e:#}");
1782 }
1783 abandoned.push(id);
1784 }
1785 abandoned
1786}
1787
1788pub async fn serve(opts: Opts) -> Result<()> {
1794 serve_until(opts, Stop::new()).await
1795}
1796
1797pub async fn serve_until(opts: Opts, stop: Stop) -> Result<()> {
1814 let signal = {
1815 let stop = stop.clone();
1816 tokio::spawn(async move {
1817 if tokio::signal::ctrl_c().await.is_ok() {
1818 stop.stop();
1819 tracing::info!("shutdown requested; a run in flight will be finished first");
1820 }
1821 })
1822 };
1823
1824 let worktrees_root = opts
1825 .worktrees_root
1826 .clone()
1827 .unwrap_or_else(crate::run::default_worktree_root);
1828 let outcome = drive(
1829 &opts,
1830 &Queue::open(),
1831 &status_path(),
1832 &crate::run::home(),
1833 &worktrees_root,
1834 &stop,
1835 )
1836 .await;
1837
1838 signal.abort();
1839 outcome
1840}
1841
1842async fn drive(
1855 opts: &Opts,
1856 queue: &Queue,
1857 status_file: &Path,
1858 home: &Path,
1859 worktrees_root: &Path,
1860 stop: &Stop,
1861) -> Result<()> {
1862 let status = Arc::new(Mutex::new(Status::new()));
1870 write_status_to(status_file, &lock(&status)).context("publish the daemon status file")?;
1871 let beat = tokio::spawn(heartbeat(Arc::clone(&status), status_file.to_path_buf()));
1872
1873 let daemon_cfg = prepare(&opts.repo, opts)
1879 .map(|c| c.daemon)
1880 .unwrap_or_default();
1881 let concurrency = max_concurrent(daemon_cfg.max_concurrent_runs);
1882
1883 let waiter = tokio::spawn(crate::waiter::run(
1888 crate::waiter::Waiter::new(
1889 crate::ask::Questions::at(home.join("questions")),
1890 home.to_path_buf(),
1891 prepare(&opts.repo, opts).ok(),
1892 ),
1893 stop.clone(),
1894 ));
1895
1896 let deputies = tokio::spawn(crate::deputy::run(
1900 crate::deputy::Deputies::new(
1901 crate::ask::Questions::at(home.join("questions")),
1902 home.to_path_buf(),
1903 prepare(&opts.repo, opts).ok(),
1904 opts.repo.clone(),
1905 daemon_cfg.max_deputies,
1906 {
1907 let stop = stop.clone();
1908 Arc::new(move || stop.parking())
1909 },
1910 ),
1911 stop.clone(),
1912 ));
1913
1914 let fetcher = tokio::spawn(fetch_loop(opts.repo.clone(), opts.clone(), stop.clone()));
1917
1918 let release_watch = tokio::spawn(crate::release_watch::run(
1921 crate::release_watch::Watcher::new(
1922 Box::new(crate::release_watch::GhForge),
1923 home.to_path_buf(),
1924 ),
1925 {
1926 let (repo, opts) = (opts.repo.clone(), opts.clone());
1927 move || release_watch_settings(&repo, &opts)
1928 },
1929 stop.clone(),
1930 ));
1931
1932 tracing::info!(
1933 "magi serve: queue {} (poll {}s, {} attempts per task, {} run(s) at once{})",
1934 queue.root().display(),
1935 opts.poll.as_secs(),
1936 opts.max_attempts,
1937 concurrency,
1938 if daemon_cfg.pause_for_interrupts {
1939 ", interrupts enabled"
1940 } else {
1941 ""
1942 }
1943 );
1944
1945 janitor(&opts.repo, opts, home, worktrees_root).await;
1948 resweep_superseded_attempts(queue, home);
1949
1950 let outcome = poll(
1951 opts,
1952 queue,
1953 &status,
1954 home,
1955 worktrees_root,
1956 stop,
1957 DispatchLimits {
1958 max_concurrent: concurrency,
1959 pause_for_interrupts: daemon_cfg.pause_for_interrupts,
1960 },
1961 )
1962 .await;
1963
1964 beat.abort();
1965 waiter.abort();
1966 deputies.abort();
1967 fetcher.abort();
1968 release_watch.abort();
1969 clear_status_at(status_file);
1970 outcome
1971}
1972
1973fn release_watch_settings(repo: &Path, opts: &Opts) -> (Vec<PathBuf>, u64) {
1976 let Ok(cfg) = prepare(repo, opts) else {
1977 return (Vec::new(), 0);
1978 };
1979 let mut paths: Vec<PathBuf> = crate::repos::scan(&cfg.repos.roots)
1980 .into_iter()
1981 .map(|r| r.path)
1982 .collect();
1983 let mut known: Vec<PathBuf> = Queue::open().list().into_iter().map(|t| t.repo).collect();
1986 let runs = crate::run::home().join("runs");
1987 for id in crate::run::list_ids_in(&runs) {
1988 let repo = std::fs::read_to_string(runs.join(&id).join("run.json"))
1989 .ok()
1990 .and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
1991 .and_then(|v| v.get("repo")?.as_str().map(PathBuf::from));
1992 known.extend(repo);
1993 }
1994 for k in known {
1995 let k = k.canonicalize().unwrap_or(k);
1996 if k.join(".git").exists() && !paths.contains(&k) {
1997 paths.push(k);
1998 }
1999 }
2000 let own = repo.canonicalize().unwrap_or_else(|_| repo.to_path_buf());
2001 if !paths.contains(&own) {
2002 paths.push(own);
2003 }
2004 (paths, cfg.daemon.release_stall_minutes)
2005}
2006
2007const FETCH_TIMEOUT: Duration = Duration::from_secs(30);
2009
2010async fn fetch_loop(repo: PathBuf, opts: Opts, stop: Stop) {
2014 while !stop.stopped() {
2015 let (interval, roots) = match prepare(&repo, &opts) {
2016 Ok(c) => (c.repos.fetch_interval, c.repos.roots),
2017 Err(_) => (0, Vec::new()),
2018 };
2019 if interval > 0 && !roots.is_empty() {
2020 let r = crate::clean::fetch_origins(&roots, FETCH_TIMEOUT, || stop.stopped()).await;
2021 tracing::debug!("fetch origins: {r:?}");
2022 }
2023 let wait = if interval > 0 { interval } else { 60 };
2025 let mut slept = 0;
2026 while slept < wait && !stop.stopped() {
2027 tokio::time::sleep(Duration::from_secs(1)).await;
2028 slept += 1;
2029 }
2030 }
2031}
2032
2033async fn heartbeat(status: Arc<Mutex<Status>>, path: PathBuf) {
2039 loop {
2040 tokio::time::sleep(HEARTBEAT).await;
2041 let snapshot = {
2042 let mut guard = lock(&status);
2043 guard.updated_at = Timestamp::now();
2044 guard.clone()
2045 };
2046 if let Err(e) = write_status_to(&path, &snapshot) {
2047 tracing::warn!("could not refresh the daemon status file: {e:#}");
2050 }
2051 }
2052}
2053
2054#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2057enum LandResume {
2058 NotLanding,
2061 StillWaiting,
2066 Ready,
2070}
2071
2072fn land_resume_state(task: &Task) -> LandResume {
2076 let Some(run_id) = task.runs.last() else {
2077 return LandResume::NotLanding;
2078 };
2079 let Ok(state) = RunState::load(run_id) else {
2080 return LandResume::NotLanding;
2081 };
2082 if !implementers_match(task, &state) || !state.parked {
2083 return LandResume::NotLanding;
2084 }
2085 let node = if state.status == RunStatus::Landing {
2089 land::APPROVAL_NODE
2090 } else if state.github_text.as_ref().is_some_and(|g| !g.resolved) {
2091 crate::github_text::ASK_NODE
2092 } else {
2093 return LandResume::NotLanding;
2094 };
2095 let store = ask::Questions::open();
2096 let waiting = store
2097 .list()
2098 .into_iter()
2099 .filter(|q| &q.run == run_id && q.node == node)
2100 .max_by(|a, b| a.id.cmp(&b.id));
2101 let Some(mut q) = waiting else {
2102 return LandResume::Ready;
2103 };
2104 if !q.status.open() {
2105 if q.resolution().as_deref() == Some(land::REPAIR) {
2108 return if state.liveness(false) == Liveness::Live {
2109 LandResume::StillWaiting
2110 } else {
2111 LandResume::NotLanding
2112 };
2113 }
2114 return LandResume::Ready;
2115 }
2116 let timeout = Duration::from_secs(state.config.graph.answer_timeout);
2123 let elapsed = Timestamp::now().as_second() - q.asked_at.as_second();
2124 if elapsed >= 0 && elapsed as u64 >= timeout.as_secs() {
2125 q.abandon(format!(
2126 "no answer within {}s of asking",
2127 timeout.as_secs().max(1)
2128 ));
2129 if store.put(&mut q).is_ok() {
2132 return LandResume::Ready;
2133 }
2134 }
2135 LandResume::StillWaiting
2136}
2137
2138const RECHECK_WHILE_BUSY: Duration = Duration::from_millis(200);
2146
2147const CACHE_CHECK_INTERVAL_SECS: u64 = 5 * 60;
2159
2160struct InFlightGuard<'a> {
2173 status: &'a Arc<Mutex<Status>>,
2174 stop: &'a Stop,
2175 task_id: &'a str,
2176}
2177
2178impl Drop for InFlightGuard<'_> {
2179 fn drop(&mut self) {
2180 lock(self.status).current.retain(|c| c.task != self.task_id);
2181 self.stop.exit();
2182 }
2183}
2184
2185#[derive(Debug, Clone, PartialEq, Eq)]
2207enum Interrupt {
2208 Idle,
2210 Parking {
2222 parked: Vec<String>,
2223 interrupt_task: String,
2224 },
2225 Running {
2233 parked: Vec<String>,
2234 interrupt_task: String,
2235 },
2236 Resuming { parked: Vec<String> },
2243}
2244
2245fn advance_interrupt(state: Interrupt, in_flight: &[String], runnable: &[Task]) -> Interrupt {
2266 match state {
2267 Interrupt::Idle => {
2268 if in_flight.len() != 1 {
2280 return Interrupt::Idle;
2281 }
2282 match runnable.iter().find(|t| t.interrupt) {
2283 Some(t) => Interrupt::Parking {
2284 parked: in_flight.to_vec(),
2285 interrupt_task: t.id.clone(),
2286 },
2287 None => Interrupt::Idle,
2288 }
2289 }
2290 Interrupt::Parking {
2291 parked,
2292 interrupt_task,
2293 } => {
2294 if in_flight.iter().any(|id| parked.contains(id)) {
2295 Interrupt::Parking {
2297 parked,
2298 interrupt_task,
2299 }
2300 } else if in_flight.contains(&interrupt_task) {
2301 Interrupt::Running {
2302 parked,
2303 interrupt_task,
2304 }
2305 } else if runnable.iter().any(|t| t.id == interrupt_task) {
2306 Interrupt::Parking {
2310 parked,
2311 interrupt_task,
2312 }
2313 } else {
2314 Interrupt::Resuming { parked }
2319 }
2320 }
2321 Interrupt::Running {
2322 parked,
2323 interrupt_task,
2324 } => {
2325 if in_flight.contains(&interrupt_task) {
2326 Interrupt::Running {
2327 parked,
2328 interrupt_task,
2329 }
2330 } else {
2331 Interrupt::Resuming { parked }
2337 }
2338 }
2339 Interrupt::Resuming { parked } => {
2340 if in_flight.iter().any(|id| parked.contains(id)) {
2341 Interrupt::Idle
2347 } else if runnable.iter().any(|t| parked.contains(&t.id)) {
2348 Interrupt::Resuming { parked }
2349 } else {
2350 Interrupt::Idle
2353 }
2354 }
2355 }
2356}
2357
2358fn advance_interrupt_tick(
2364 enabled: bool,
2365 state: Interrupt,
2366 in_flight: &[String],
2367 runnable: &[Task],
2368) -> Interrupt {
2369 if !enabled {
2370 return Interrupt::Idle;
2371 }
2372 advance_interrupt(state, in_flight, runnable)
2373}
2374
2375fn interrupt_gate(state: &Interrupt, in_flight: &[String], candidates: Vec<Task>) -> Vec<Task> {
2380 match state {
2381 Interrupt::Idle => candidates,
2382 Interrupt::Parking {
2383 parked,
2384 interrupt_task,
2385 } => {
2386 if in_flight.iter().any(|id| parked.contains(id)) {
2387 Vec::new()
2388 } else {
2389 candidates
2390 .into_iter()
2391 .filter(|t| &t.id == interrupt_task)
2392 .collect()
2393 }
2394 }
2395 Interrupt::Running { .. } => Vec::new(),
2396 Interrupt::Resuming { parked } => candidates
2404 .into_iter()
2405 .find(|t| parked.contains(&t.id))
2406 .into_iter()
2407 .collect(),
2408 }
2409}
2410
2411struct DispatchLimits {
2415 max_concurrent: usize,
2418 pause_for_interrupts: bool,
2420}
2421
2422#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2431enum PermitKind {
2432 None,
2437 Urgent,
2444 Ordinary,
2447}
2448
2449fn permit_kind(priority: bool, urgent: bool) -> PermitKind {
2452 if priority {
2453 PermitKind::None
2454 } else if urgent {
2455 PermitKind::Urgent
2456 } else {
2457 PermitKind::Ordinary
2458 }
2459}
2460
2461struct DispatchClaim {
2462 task: Task,
2463 claim: crate::queue::Claim,
2464 permit: Option<tokio::sync::OwnedSemaphorePermit>,
2465}
2466
2467fn claim_dispatch(
2470 queue: &Queue,
2471 candidate: &Task,
2472 cooling_down: bool,
2473 ordinary: &Arc<tokio::sync::Semaphore>,
2474 urgent: &Arc<tokio::sync::Semaphore>,
2475) -> Option<DispatchClaim> {
2476 let Ok(claim) = queue.claim(&candidate.id) else {
2477 tracing::info!("task {} is claimed elsewhere; skipping", candidate.short());
2478 return None;
2479 };
2480 let task = match queue.get(&candidate.id) {
2481 Ok(t) if t.status.runnable() => t,
2482 Ok(_) => return None,
2483 Err(e) => {
2484 tracing::warn!("could not re-read task {}: {e:#}", candidate.short());
2485 return None;
2486 }
2487 };
2488 let resume = land_resume_state(&task);
2489 if resume == LandResume::StillWaiting {
2490 return None;
2491 }
2492 let priority = resume == LandResume::Ready;
2493 if !priority && cooling_down {
2494 return None;
2495 }
2496 let permit = match permit_kind(priority, task.urgent) {
2499 PermitKind::None => None,
2500 PermitKind::Urgent => Some(Arc::clone(urgent).try_acquire_owned().ok()?),
2501 PermitKind::Ordinary => Some(Arc::clone(ordinary).try_acquire_owned().ok()?),
2502 };
2503 Some(DispatchClaim {
2504 task,
2505 claim,
2506 permit,
2507 })
2508}
2509
2510async fn poll(
2529 opts: &Opts,
2530 queue: &Queue,
2531 status: &Arc<Mutex<Status>>,
2532 home: &Path,
2533 worktrees_root: &Path,
2534 stop: &Stop,
2535 limits: DispatchLimits,
2536) -> Result<()> {
2537 let DispatchLimits {
2538 max_concurrent,
2539 pause_for_interrupts,
2540 } = limits;
2541 let mut attempted: Vec<String> = Vec::new();
2546 let sem = Arc::new(tokio::sync::Semaphore::new(max_concurrent));
2547 let urgent_sem = Arc::new(tokio::sync::Semaphore::new(1));
2555 let quota_cooldown_until: Arc<Mutex<Option<Timestamp>>> = Arc::new(Mutex::new(None));
2561 let mut inflight: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
2562 let mut conductor = Conductor::new();
2563 let mut cache_last_checked: Option<Timestamp> = None;
2566 let mut interrupt = Interrupt::Idle;
2568 let mut interrupt_pauses: std::collections::HashMap<String, crate::graph::Pause> =
2574 std::collections::HashMap::new();
2575
2576 while !stop.stopped() {
2577 lock(status).polls += 1;
2578
2579 while let Some(result) = inflight.try_join_next() {
2584 if let Err(e) = result {
2585 tracing::error!("a spawned attempt did not finish cleanly: {e}");
2586 notices::raise(Notice::error(
2587 "loop:attempt",
2588 "A queued attempt ended abnormally; check the task it was running.",
2589 ));
2590 }
2591 }
2592
2593 let swept = sweep_stale_claims(queue, STALE_CLAIM);
2594 if !swept.is_empty() {
2595 tracing::warn!(
2596 "swept {} stale claim(s) left behind by an earlier daemon: {}",
2597 swept.len(),
2598 swept.join(", ")
2599 );
2600 }
2601 let now = Timestamp::now();
2606
2607 if !stop.busy_now() {
2612 maybe_prune_cache_between_runs(
2613 &opts.repo,
2614 opts,
2615 home,
2616 stop,
2617 &mut cache_last_checked,
2618 now,
2619 )
2620 .await;
2621 }
2622
2623 let stalled = stalled_tasks(queue, home, now);
2624 let stalled_ids: std::collections::BTreeSet<_> =
2625 stalled.iter().map(|task| task.id.clone()).collect();
2626 let reclaimed = reclaim_orphaned_running(queue, opts.max_attempts);
2627 if !reclaimed.is_empty() {
2628 tracing::warn!(
2629 "reclaimed {} task(s) left `running` by a daemon that never \
2630 recorded the outcome: {}",
2631 reclaimed.len(),
2632 reclaimed.join(", ")
2633 );
2634 }
2635 let abandoned_runs = reclaim_abandoned_runs(home, now);
2636 if !abandoned_runs.is_empty() {
2637 tracing::warn!(
2638 "failed {} run(s) left behind by a killed process, past every \
2639 active seat's own timeout: {}",
2640 abandoned_runs.len(),
2641 abandoned_runs.join(", ")
2642 );
2643 }
2644
2645 let questions = Questions::at(home.join("questions"));
2650
2651 resolve_blockers(queue, &questions);
2654 apply_choice_actions(queue, &questions, home);
2655 triage::apply_answers(queue, &questions);
2656 reconcile_task_questions(queue, &questions);
2657 {
2661 let talks = crate::talk::Talks::at(home.join("talks"));
2662 crate::chat_notify::sweep(queue, &talks, home);
2663 crate::chat_watch::sweep(home, &talks);
2664 crate::chat_notify::start_turns(&talks, stop.parking(), &mut inflight);
2665 }
2666
2667 let finished: Vec<Task> = finished_tasks(queue)
2673 .into_iter()
2674 .filter(|task| !stalled_ids.contains(&task.id))
2675 .filter(|task| !awaiting_resume_with(task, |id| RunState::load_under(id, home)))
2679 .collect();
2680 let queued = queued_tasks(queue);
2681 if !(queued.is_empty() && stalled.is_empty() && finished.is_empty())
2685 && conductor.worth_a_look(queue, &stalled, &finished)
2686 {
2687 match prepare(&opts.repo, opts) {
2688 Ok(cfg) => {
2689 conductor
2690 .maybe_run(
2691 &cfg,
2692 &opts.repo,
2693 queue,
2694 &questions,
2695 home,
2696 &queued,
2697 &stalled,
2698 &finished,
2699 opts.max_attempts,
2700 )
2701 .await;
2702 }
2703 Err(e) => {
2704 tracing::warn!("conductor: no config: {e:#}");
2705 notices::raise(Notice::warn(
2706 "loop:no-config",
2707 "The loop could not read this repository's config, so held tasks are not being triaged.",
2708 ));
2709 }
2710 }
2711 }
2712
2713 let candidates: Vec<Task> = runnable(queue)
2714 .into_iter()
2715 .filter(|t| !opts.once || !attempted.contains(&t.id))
2716 .collect();
2717
2718 let in_flight: Vec<String> = lock(status)
2723 .current
2724 .iter()
2725 .map(|c| c.task.clone())
2726 .collect();
2727 interrupt_pauses.retain(|id, _| in_flight.contains(id));
2728
2729 interrupt =
2730 advance_interrupt_tick(pause_for_interrupts, interrupt, &in_flight, &candidates);
2731 if let Interrupt::Parking {
2732 parked,
2733 interrupt_task,
2734 } = &interrupt
2735 {
2736 let reason = format!(
2737 "task {} asked to run first",
2738 crate::run::short_of(interrupt_task)
2739 );
2740 for id in parked {
2741 if let Some(pause) = interrupt_pauses.get(id) {
2742 pause.park_because(reason.clone());
2743 }
2744 }
2745 }
2746 let candidates = interrupt_gate(&interrupt, &in_flight, candidates);
2747
2748 let cooling_down =
2749 lock("a_cooldown_until).is_some_and(|until| Timestamp::now() < until);
2750
2751 let mut started_any = false;
2752 for candidate in candidates {
2753 if stop.stopped() {
2754 break;
2755 }
2756
2757 let Some(DispatchClaim {
2758 mut task,
2759 claim,
2760 permit,
2761 }) = claim_dispatch(queue, &candidate, cooling_down, &sem, &urgent_sem)
2762 else {
2763 continue;
2764 };
2765
2766 let task_id = task.id.clone();
2767 attempted.push(task_id.clone());
2768 lock(status).idle = false;
2769 stop.enter();
2772 started_any = true;
2773
2774 let run_pause = crate::graph::Pause::new();
2778 interrupt_pauses.insert(task_id.clone(), run_pause.clone());
2779
2780 let opts = opts.clone();
2781 let queue = queue.clone();
2782 let status = Arc::clone(status);
2783 let stop = stop.clone();
2784 let quota_cooldown_until = Arc::clone("a_cooldown_until);
2785 inflight.spawn(async move {
2786 let _claim = claim;
2790 let _permit = permit;
2791 let _inflight = InFlightGuard {
2793 status: &status,
2794 stop: &stop,
2795 task_id: &task_id,
2796 };
2797 let quota = attempt(&opts, &queue, &status, &stop, run_pause, &mut task).await;
2798 lock(&status).completed += 1;
2799 let now = Timestamp::now();
2805 if let Some(until) = cooldown_until("a, now) {
2806 let wait = until.as_second() - now.as_second();
2807 *lock("a_cooldown_until) = Some(until);
2808 let hint = quota
2809 .iter()
2810 .find(|q| q.reset.is_some())
2811 .and_then(|q| q.reset.as_deref());
2812 match hint {
2813 Some(h) => tracing::warn!(
2814 "quota hit; waiting {wait}s before taking another ordinary task \
2815 (CLI reported reset: {h})"
2816 ),
2817 None => tracing::warn!(
2818 "quota hit; waiting {wait}s before taking another ordinary task \
2819 (no reset hint reported)"
2820 ),
2821 }
2822 }
2823 });
2824 }
2825
2826 if started_any {
2827 continue;
2828 }
2829
2830 if stop.busy_now() {
2831 stop.idle(RECHECK_WHILE_BUSY.min(opts.poll)).await;
2836 continue;
2837 }
2838
2839 lock(status).idle = true;
2841 if opts.once {
2842 janitor(&opts.repo, opts, home, worktrees_root).await;
2846 resweep_superseded_attempts(queue, home);
2847 triage_held(queue, home, opts).await;
2848 break;
2849 }
2850 stop.idle(opts.poll).await;
2851 if stop.stopped() {
2852 continue;
2853 }
2854 janitor(&opts.repo, opts, home, worktrees_root).await;
2860 resweep_superseded_attempts(queue, home);
2861 triage_held(queue, home, opts).await;
2862 }
2863
2864 while let Some(result) = inflight.join_next().await {
2869 if let Err(e) = result {
2870 tracing::error!("a spawned attempt did not finish cleanly: {e}");
2871 notices::raise(Notice::error(
2872 "loop:attempt",
2873 "A queued attempt ended abnormally; check the task it was running.",
2874 ));
2875 }
2876 }
2877 Ok(())
2878}
2879
2880async fn attempt(
2886 opts: &Opts,
2887 queue: &Queue,
2888 status: &Arc<Mutex<Status>>,
2889 stop: &Stop,
2890 interrupt_pause: crate::graph::Pause,
2891 task: &mut Task,
2892) -> Vec<QuotaLoss> {
2893 let repo = repo_for(task, &opts.repo);
2894 tracing::info!(
2895 "task {} — {} (repo {})",
2896 task.short(),
2897 task.title,
2898 repo.display()
2899 );
2900
2901 let explicit = match &task.overrides {
2905 Some(o) if !o.inherit_daemon_options => o.config.as_deref(),
2906 Some(_) => opts.config.as_deref(),
2907 None => opts.config.as_deref(),
2908 };
2909 if let Err(e) = Config::discover_fetched(&repo, explicit).await {
2910 let reason = format!("[config] {e:#}");
2911 task.last_error = Some(reason.clone());
2912 task.hold_machine(Some(reason.clone()));
2913 record(queue, task);
2914 tracing::warn!("holding {}: {reason}", task.short());
2915 return Vec::new();
2916 }
2917 let mut config = match prepare_for(&repo, opts, task) {
2918 Ok(c) => c,
2919 Err(e) => {
2920 task.attempts += 1;
2924 task.fail(format!("config: {e:#}"), opts.max_attempts);
2925 record(queue, task);
2926 return Vec::new();
2927 }
2928 };
2929 apply_task_implementers(&mut config, task);
2930 let p = phrases(&config.graph.language);
2931 let start_failed = p.could_not_start;
2932
2933 if let Some(reason) = disk_gate(&repo, &config) {
2941 task.last_error = Some(reason.clone());
2942 task.hold_machine(Some(reason.clone()));
2943 record(queue, task);
2944 tracing::warn!("holding {} for want of disk space: {reason}", task.short());
2945 notices::raise(
2948 Notice::warn(
2949 &format!("disk:{}", repo.display()),
2950 "A task was held for want of free disk space; free some, then release it from the queue.",
2951 )
2952 .link(Link::Task {
2953 id: task.id.clone(),
2954 }),
2955 );
2956 return Vec::new();
2957 }
2958
2959 let unfinished = (!task.fresh_start)
2979 .then(|| unfinished_run(&task.runs, task.short()))
2980 .flatten()
2981 .filter(|id| RunState::load(id).is_ok_and(|run| may_resume_run(&run, task)));
2982 let previous = task.runs.last().and_then(|id| RunState::load(id).ok());
2988 let review_branch = take_review_branch(task, previous.as_ref());
2989 let branch_exists = match &review_branch {
2990 Some(branch) => crate::git::branch_exists(&repo, branch)
2991 .await
2992 .unwrap_or(false),
2993 None => false,
2994 };
2995 let proposed = match (&review_branch, &unfinished, &task.review_of) {
2999 (None, None, Some(branch)) => Starter::Review(branch.clone()),
3000 _ => choose_starter(
3001 review_branch.as_deref(),
3002 branch_exists,
3003 unfinished.as_deref(),
3004 ),
3005 };
3006 let starter = if let Some(selected) = &task.pr_recovery {
3007 if selected.check_task(&task.id, &task.instruction).is_ok() {
3008 selected_starter(selected, previous.as_ref(), unfinished.as_deref())
3009 } else {
3010 Starter::Review(selected.branch.clone())
3011 }
3012 } else {
3013 starter_with_implementers(task, previous.as_ref(), proposed)
3014 };
3015 if let Some(pin) = &task.resume_override
3018 && let Some(run) = &pin.pinned_run
3019 && Questions::open().get(&pin.question_id).map_or(true, |q| {
3020 q.node == land::APPROVAL_NODE && q.resolution().as_deref() == Some(land::REPAIR)
3021 })
3022 && starter != Starter::Resume(run.clone())
3023 {
3024 task.hold_machine(Some(format!(
3025 "repair owner cannot resume; inspect `magi show {run}` and restore the existing run before `magi task release {}`; no replacement task was started",
3026 task.id
3027 )));
3028 record(queue, task);
3029 return Vec::new();
3030 }
3031 let attachments = match task_attachments(queue, task) {
3036 Ok(a) => a,
3037 Err(e) => {
3038 task.attempts += 1;
3039 task.fail(format!("{start_failed}{e:#}"), opts.max_attempts);
3040 record(queue, task);
3041 return Vec::new();
3042 }
3043 };
3044 let started = match &starter {
3045 Starter::Review(_) if task.pr_recovery.is_some() => {
3046 let selected = task.pr_recovery.clone().expect("selected recovery");
3047 match selected.check_task(&task.id, &task.instruction) {
3048 Ok(()) => Runner::recover_pr(&repo, &task.id, selected, config)
3049 .await
3050 .map(|mut r| {
3051 r.state.attachments = attachments.clone();
3052 r
3053 }),
3054 Err(e) => Err(e),
3055 }
3056 }
3057 Starter::Review(branch) => {
3058 tracing::info!(
3059 "task {} reopens `{branch}` as a review-only pass",
3060 task.short()
3061 );
3062 let takeover = crate::handover::Takeover {
3065 earlier: task.earlier_attempts().to_vec(),
3066 home: crate::run::home(),
3067 choice: take_divergence_answer(branch, &config.merge.remote, task),
3068 task_instruction: Some(instruction_for(task)),
3069 };
3070 Runner::review_taking_over(
3071 &repo,
3072 branch,
3073 config,
3074 Some(takeover),
3075 crate::run::Origin::queue(&task.id),
3076 )
3077 .await
3078 }
3079 Starter::Resume(id) => {
3080 tracing::info!("resuming run {id} rather than competing again");
3081 Runner::resume(id).and_then(|mut r| {
3082 if r.state.prep_failure.is_some() {
3083 let approval = task
3084 .resumption
3085 .as_ref()
3086 .context("preparation retry requires an explicit task release")?;
3087 r.retry_preparation(approval.at)?;
3088 }
3089 if let Some(instruction) =
3090 prepare_instruction(&starter, Some(&r.state.instruction), task)
3091 {
3092 r.state.instruction = instruction;
3093 }
3094 r.state.attachments = attachments.clone();
3095 if let Some(mode) = task
3098 .overrides
3099 .as_ref()
3100 .and_then(|o| o.merge.as_deref())
3101 .and_then(|m| merge_mode(m).ok())
3102 {
3103 r.state.config.merge.mode = mode;
3104 }
3105 Ok(r)
3106 })
3107 }
3108 Starter::Start => {
3109 if let Some(branch) = &review_branch {
3110 tracing::warn!(
3111 "conductor chose review for task {} but branch `{branch}` no longer \
3112 exists; requeuing as a fresh competition instead",
3113 task.short()
3114 );
3115 }
3116 let instruction = prepare_instruction(&starter, None, task)
3117 .unwrap_or_else(|| task.instruction.clone());
3118 Runner::start_naming(
3119 &repo,
3120 instruction,
3121 &task.title,
3122 config,
3123 crate::run::Origin::queue(&task.id),
3124 )
3125 .await
3126 .map(|mut r| {
3127 r.state.attachments = attachments.clone();
3128 r
3129 })
3130 }
3131 };
3132 let mut runner = match started {
3133 Ok(r) => r,
3134 Err(e) if task.pr_recovery.is_some() => {
3138 task.attempts += 1;
3139 if let Some(partial) = e.downcast_ref::<crate::pr_recovery::StartupFailed>() {
3140 task.link_run(&partial.run);
3141 task.record_pass(&partial.run, 0, task.attempts, None);
3142 }
3143 task.last_error = Some(format!("{start_failed}{e:#}"));
3144 task.hold_unrequested_retry(Some(format!(
3145 "Selected public-head recovery could not start: {e:#}. Correct the cause and use `magi task recover-pr` with the approved public head; automatic recovery is stopped."
3146 )));
3147 record(queue, task);
3148 return Vec::new();
3149 }
3150 Err(e) if e.downcast_ref::<crate::handover::Refused>().is_some() => {
3151 let detail = match e.downcast_ref::<crate::handover::Refused>() {
3152 Some(r) => (p.handover_refused)(r),
3153 None => format!("{e:#}"),
3154 };
3155 let reason = format!("{start_failed}{detail}");
3156 task.last_error = Some(reason.clone());
3157 let branch = match &starter {
3160 Starter::Review(branch) => Some(branch.clone()),
3161 _ => None,
3162 };
3163 task.hold_for_handover(branch, format!("{reason}{}", p.handover_hint));
3167 record(queue, task);
3168 tracing::warn!(
3169 "holding {} for a branch it cannot take over: {e:#}",
3170 task.short()
3171 );
3172 return Vec::new();
3173 }
3174 Err(e) if e.downcast_ref::<crate::reconcile::Diverged>().is_some() => {
3179 let d = e
3180 .downcast_ref::<crate::reconcile::Diverged>()
3181 .expect("checked by the guard");
3182 let mut q = ask::Question::new(
3183 task.id.clone(),
3184 crate::reconcile::NODE.to_owned(),
3185 crate::reconcile::SEAT.to_owned(),
3186 d.summary(),
3187 d.detail(),
3188 d.choices(),
3189 );
3190 match Questions::open().put(&mut q) {
3191 Ok(()) => {
3192 task.last_error = Some(format!("{e:#}"));
3193 task.review_branch = Some(d.branch.clone());
3196 task.review_instruction = Some(task.instruction.clone());
3197 task.block(vec![q.id.clone()], Some(d.summary()));
3198 }
3199 Err(put) => {
3200 tracing::warn!("could not file the divergence question: {put:#}");
3201 task.attempts += 1;
3202 task.fail(format!("{start_failed}{e:#}"), opts.max_attempts);
3203 }
3204 }
3205 record(queue, task);
3206 return Vec::new();
3207 }
3208 Err(e) => {
3209 if e.downcast_ref::<crate::reconcile::Stale>().is_some()
3215 && let Starter::Review(branch) = &starter
3216 {
3217 task.review_branch = Some(branch.clone());
3218 task.review_instruction = Some(task.instruction.clone());
3219 task.last_error = Some(format!("{start_failed}{e:#}"));
3220 task.status = crate::queue::TaskStatus::Failed;
3221 record(queue, task);
3222 return Vec::new();
3223 }
3224 task.attempts += 1;
3225 task.last_error = Some(format!("{start_failed}{e:#}"));
3226 task.hold_unrequested_retry(Some(format!(
3227 "Startup failed: {e:#}. Correct the startup source or configuration, then explicitly release the task; automatic recovery is stopped."
3228 )));
3229 record(queue, task);
3230 return Vec::new();
3231 }
3232 };
3233 runner.state.task_instruction = Some(task.instruction.clone());
3234 runner.state.followup_generation = Some(task.followup.as_ref().map_or(0, |f| f.generation));
3237 if runner.state.origin_chat.is_none() {
3238 runner.state.origin_chat = task.chat_talk().map(str::to_owned);
3239 }
3240 runner.on_pause(stop.pause());
3242 runner.watch_interrupt(interrupt_pause);
3246
3247 let run = runner.state.id.clone();
3250 task.start(run.clone());
3251 record(queue, task);
3252 lock(status).current.push(Current {
3253 task: task.id.clone(),
3254 run,
3255 });
3256
3257 let quota_before = runner.state.quota.clone();
3260 runner.state.next_pass_task_entry = true;
3261 let result = runner.execute().await;
3262 finish_attempt(
3263 opts.max_attempts,
3264 queue,
3265 task,
3266 &runner.state,
3267 "a_before,
3268 result,
3269 )
3270}
3271
3272pub fn finish_attempt(
3277 max_attempts: usize,
3278 queue: &Queue,
3279 task: &mut Task,
3280 state: &RunState,
3281 quota_before: &[QuotaLoss],
3282 result: Result<()>,
3283) -> Vec<QuotaLoss> {
3284 let detail = match result {
3285 Ok(()) => describe(state),
3286 Err(e) => format!("{e:#}"),
3287 };
3288 let fresh = losses_this_attempt(quota_before, &state.quota);
3289 let verdict = Verdict {
3290 status: state.status,
3291 left_pr: state.pr.is_some(),
3294 quota_hit: !fresh.is_empty(),
3300 parked: state.parked,
3304 no_viable_candidates: state.viable().is_empty(),
3307 };
3308 settle_and_diagnose(task, verdict, &detail, max_attempts, state);
3309 if state.status == RunStatus::Blocked && state.prep_failure.is_some() {
3310 task.hold_unrequested_retry(Some(format!(
3311 "Candidate setup failed: {detail}. Select a complete implementation with `magi task recover-pr`, or correct the referenced source before explicitly releasing the task."
3312 )));
3313 }
3314 if task.status == TaskStatus::Done {
3315 supersede_prior_runs(task, &crate::run::home());
3316 }
3317 record(queue, task);
3318 tracing::info!(
3319 "task {} is {} after run {} ({})",
3320 task.short(),
3321 task.status.as_str(),
3322 state.short(),
3323 label(state.status)
3324 );
3325 fresh
3326}
3327
3328pub fn hold_if_runnable(queue: &Queue, task: &mut Task) {
3332 let waits_on_owner = withheld_text_wait(task).is_some();
3337 if task.status.runnable() && !waits_on_owner {
3338 let why = task.last_error.clone().map_or_else(
3339 || "the run did not finish".to_owned(),
3340 |e| format!("the run did not finish: {e}"),
3341 );
3342 task.hold_unrequested_retry(Some(format!(
3343 "{why}. It was started by hand, so it is not retried \
3344 automatically; `magi task release` retries it."
3345 )));
3346 record(queue, task);
3347 }
3348}
3349
3350#[must_use]
3354pub fn loop_would_resume(run: &str) -> bool {
3355 unfinished_run(&[run.to_owned()], crate::run::short_of(run)).is_some()
3356}
3357
3358#[must_use]
3368pub fn foreign_loop(
3369 reading: Option<&Reading>,
3370 now: Timestamp,
3371 own_pid: u32,
3372) -> Option<Option<u32>> {
3373 let reading = reading.filter(|r| r.running(now))?;
3374 match reading.pid {
3375 Some(pid) if pid == own_pid => None,
3376 pid => Some(pid),
3377 }
3378}
3379
3380pub async fn run_claimed(opts: &Opts, queue: &Queue, task: &mut Task) {
3395 let status = Arc::new(Mutex::new(Status::new()));
3396 let stop = Stop::new();
3397 attempt(
3398 opts,
3399 queue,
3400 &status,
3401 &stop,
3402 crate::graph::Pause::new(),
3403 task,
3404 )
3405 .await;
3406 let mut resumes = 0;
3415 while resumes < MAX_WITHHELD_RESUMES {
3416 let Some(timeout) = withheld_text_wait(task) else {
3417 break;
3418 };
3419 eprintln!(
3420 "waiting for the owner's answer on a withheld pull request text \
3421 (up to {timeout}s; Ctrl-C leaves the run parked for a later `magi serve`)"
3422 );
3423 let limit = std::time::Instant::now() + Duration::from_secs(timeout + 60);
3426 let ready = loop {
3427 match land_resume_state(task) {
3428 LandResume::StillWaiting if std::time::Instant::now() < limit => {
3429 tokio::time::sleep(WITHHELD_POLL).await;
3430 }
3431 LandResume::Ready => break true,
3432 _ => break false,
3433 }
3434 };
3435 if !ready {
3436 break;
3437 }
3438 resumes += 1;
3439 attempt(
3440 opts,
3441 queue,
3442 &status,
3443 &stop,
3444 crate::graph::Pause::new(),
3445 task,
3446 )
3447 .await;
3448 }
3449 hold_if_runnable(queue, task);
3450}
3451
3452const MAX_WITHHELD_RESUMES: u32 = 3;
3455
3456const WITHHELD_POLL: Duration = Duration::from_secs(15);
3458
3459fn withheld_text_wait(task: &Task) -> Option<u64> {
3462 let id = task.runs.last()?;
3463 let s = RunState::load(id).ok()?;
3464 (s.parked && s.github_text.as_ref().is_some_and(|g| !g.resolved))
3465 .then_some(s.config.graph.answer_timeout)
3466}
3467
3468fn losses_this_attempt(before: &[QuotaLoss], after: &[QuotaLoss]) -> Vec<QuotaLoss> {
3478 after
3479 .iter()
3480 .filter(|q| !before.contains(q))
3481 .cloned()
3482 .collect()
3483}
3484
3485fn cooldown_until(quota: &[QuotaLoss], now: Timestamp) -> Option<Timestamp> {
3488 if quota.is_empty() {
3489 return None;
3490 }
3491 let with_hint = quota.iter().find(|q| q.reset.is_some());
3492 let reset_at = with_hint.and_then(|q| parse_reset_hint(q.reset.as_deref()?, now, q.at));
3493 let wait = quota_wait(reset_at, now, QUOTA_WAIT_FALLBACK, QUOTA_WAIT_CAP);
3494 let secs = i64::try_from(wait.as_secs()).unwrap_or(i64::MAX);
3495 Some(
3496 now.checked_add(jiff::SignedDuration::from_secs(secs))
3497 .unwrap_or(Timestamp::MAX),
3498 )
3499}
3500
3501fn implementers_match(task: &Task, state: &RunState) -> bool {
3503 task.implementers()
3504 .is_none_or(|n| n == state.config.graph.implementers)
3505}
3506
3507fn starter_with_implementers(
3509 task: &Task,
3510 previous: Option<&RunState>,
3511 proposed: Starter,
3512) -> Starter {
3513 if previous.is_some_and(|s| !implementers_match(task, s)) {
3514 task.review_of
3515 .as_ref()
3516 .map_or(Starter::Start, |branch| Starter::Review(branch.clone()))
3517 } else {
3518 proposed
3519 }
3520}
3521
3522fn apply_task_implementers(config: &mut Config, task: &Task) {
3524 if let Some(n) = task.implementers() {
3525 config.graph.implementers = n;
3526 }
3527}
3528
3529fn prepare_for(repo: &Path, opts: &Opts, task: &Task) -> Result<Config> {
3534 let Some(o) = &task.overrides else {
3535 return prepare(repo, opts);
3536 };
3537 let own = Opts {
3541 config: o.config.clone(),
3542 merge: None,
3543 ..opts.clone()
3544 };
3545 let mut config = prepare(repo, if o.inherit_daemon_options { opts } else { &own })?;
3546 o.apply(&mut config);
3547 Ok(config)
3548}
3549
3550fn prepare(repo: &Path, opts: &Opts) -> Result<Config> {
3552 let (mut config, _layers) = Config::discover(repo, opts.config.as_deref())?;
3553 if let Some(mode) = &opts.merge {
3554 config.merge.mode = merge_mode(mode)?;
3555 }
3556 Ok(config)
3557}
3558
3559async fn maybe_prune_cache_between_runs(
3590 repo: &Path,
3591 opts: &Opts,
3592 home: &Path,
3593 stop: &Stop,
3594 last_checked: &mut Option<Timestamp>,
3595 now: Timestamp,
3596) {
3597 if stop.stopped() || !cache_check_due(*last_checked, now, CACHE_CHECK_INTERVAL_SECS) {
3598 return;
3599 }
3600 *last_checked = Some(now);
3601 let cfg = match prepare(repo, opts) {
3602 Ok(cfg) => cfg,
3603 Err(e) => {
3604 tracing::warn!("cache check: no config: {e:#}");
3605 return;
3606 }
3607 };
3608 match clean::prune_cache_if_over_limit(&cfg, home) {
3609 Ok(Some(pruned)) if pruned.files > 0 => tracing::info!(
3610 "housekeep: pruned {} file(s) ({} bytes) from the shared cache between runs",
3611 pruned.files,
3612 pruned.freed
3613 ),
3614 Ok(_) => {}
3615 Err(e) => {
3616 tracing::warn!("housekeep: prune cache: {e:#}");
3617 notices::raise_in(
3618 home,
3619 Notice::warn(
3620 "housekeep:cache",
3621 "Pruning the shared build cache failed; disk usage may keep growing.",
3622 ),
3623 );
3624 }
3625 }
3626}
3627
3628fn cache_check_due(last_checked: Option<Timestamp>, now: Timestamp, interval_secs: u64) -> bool {
3632 last_checked.is_none_or(|last| clean::due(now, last, interval_secs))
3633}
3634
3635async fn janitor(repo: &Path, opts: &Opts, home: &Path, worktrees_root: &Path) {
3654 let cfg = match prepare(repo, opts) {
3655 Ok(cfg) => cfg,
3656 Err(e) => {
3657 tracing::warn!("housekeep: no config: {e:#}");
3658 return;
3659 }
3660 };
3661 let worktrees_root = cfg.graph.worktree_root.as_deref().unwrap_or(worktrees_root);
3670 let out = clean::housekeep(&cfg, home, worktrees_root, repo, Timestamp::now()).await;
3671 if out.folded > 0 || out.unreadable > 0 || out.orphaned_worktrees > 0 {
3676 let mut extra = Vec::new();
3677 if out.unreadable > 0 {
3678 extra.push(format!("{} unreadable", out.unreadable));
3679 }
3680 if out.orphaned_worktrees > 0 {
3681 extra.push(format!("{} orphaned worktree(s)", out.orphaned_worktrees));
3682 }
3683 let detail = if extra.is_empty() {
3684 String::new()
3685 } else {
3686 format!(" ({})", extra.join(", "))
3687 };
3688 tracing::info!("housekeep: folded {} run(s){detail}", out.folded);
3689 }
3690 if out.external_merges_recorded > 0 {
3691 tracing::info!(
3692 "housekeep: recorded {} run(s) as merged externally",
3693 out.external_merges_recorded
3694 );
3695 }
3696 if out.stale_pr_states_repaired > 0 {
3697 tracing::info!(
3698 "housekeep: rewrote {} run record(s) whose pull request had already settled",
3699 out.stale_pr_states_repaired
3700 );
3701 }
3702 if out.cache_files > 0 {
3703 tracing::info!(
3704 "housekeep: pruned {} file(s) ({} bytes) from the shared cache",
3705 out.cache_files,
3706 out.cache_freed
3707 );
3708 }
3709 if out.questions_abandoned > 0 {
3710 tracing::info!(
3711 "housekeep: abandoned {} question(s) left open by a finished run",
3712 out.questions_abandoned
3713 );
3714 }
3715}
3716
3717async fn triage_held(queue: &Queue, home: &Path, opts: &Opts) {
3726 let questions = Questions::at(home.join("questions"));
3727 let report = triage::run_once(queue, &questions, opts.config.as_deref(), Timestamp::now());
3728 if report.is_empty() {
3729 return;
3730 }
3731 if !report.quarantined.is_empty() {
3732 tracing::info!(
3733 "triage: held {} blocked task(s) whose blocked-on task or \
3734 question no longer exists: {}",
3735 report.quarantined.len(),
3736 report.quarantined.join(", ")
3737 );
3738 }
3739 if !report.resumed.is_empty() {
3740 tracing::info!(
3741 "triage: resumed {} held task(s) whose machine hold had resolved: {}",
3742 report.resumed.len(),
3743 report.resumed.join(", ")
3744 );
3745 }
3746 if !report.asked.is_empty() {
3747 tracing::info!(
3748 "triage: asked about {} held task(s): {}",
3749 report.asked.len(),
3750 report.asked.join(", ")
3751 );
3752 }
3753 if !report.answered.is_empty() {
3754 tracing::info!(
3755 "triage: applied {} operator answer(s): {}",
3756 report.answered.len(),
3757 report.answered.join(", ")
3758 );
3759 }
3760}
3761
3762fn disk_gate(repo: &Path, config: &Config) -> Option<String> {
3769 disk_gate_with(repo, config, crate::disk::free_bytes)
3770}
3771
3772fn disk_gate_with<F: Fn(&Path) -> Result<u64>>(
3776 repo: &Path,
3777 config: &Config,
3778 free_bytes: F,
3779) -> Option<String> {
3780 let min = config.disk.min_free_bytes;
3781 if min == 0 {
3782 return None;
3783 }
3784 match free_bytes(repo) {
3785 Ok(free) => crate::disk::gate_in(free, min, &config.graph.language),
3786 Err(e) => Some(crate::disk::unmeasured_in(repo, &e, &config.graph.language)),
3787 }
3788}
3789
3790const QUOTA_WAIT_FALLBACK: Duration = Duration::from_secs(5 * 60);
3797
3798const QUOTA_WAIT_CAP: Duration = Duration::from_secs(30 * 60);
3802
3803fn quota_wait(
3812 reset_at: Option<Timestamp>,
3813 now: Timestamp,
3814 fallback: Duration,
3815 cap: Duration,
3816) -> Duration {
3817 match reset_at {
3818 Some(at) if at > now => {
3819 let secs = u64::try_from(at.as_second() - now.as_second()).unwrap_or(0);
3820 Duration::from_secs(secs).min(cap)
3821 }
3822 _ => fallback,
3823 }
3824}
3825
3826fn parse_reset_hint(text: &str, now: Timestamp, recorded: Timestamp) -> Option<Timestamp> {
3840 parse_reset_hint_zoned(text, now)
3841 .or_else(|| parse_reset_hint_dated(text))
3842 .or_else(|| parse_reset_hint_relative(text, recorded))
3843}
3844
3845fn parse_reset_hint_relative(text: &str, recorded: Timestamp) -> Option<Timestamp> {
3849 let rest = text.trim().trim_end_matches('.').strip_prefix("in ")?;
3850 let mut rest = rest.trim();
3851 if rest.is_empty() {
3852 return None;
3853 }
3854 let mut total: i64 = 0;
3855 let mut matched = false;
3856 for (unit, secs) in [('h', 3600), ('m', 60), ('s', 1)] {
3857 if let Some((digits, tail)) = rest.split_once(unit)
3858 && !digits.is_empty()
3859 && digits.bytes().all(|b| b.is_ascii_digit())
3860 {
3861 total += digits.parse::<i64>().ok()?.checked_mul(secs)?;
3862 rest = tail;
3863 matched = true;
3864 }
3865 }
3866 if !rest.is_empty() || !matched {
3867 return None;
3868 }
3869 recorded
3870 .checked_add(jiff::SignedDuration::from_secs(total))
3871 .ok()
3872}
3873
3874fn parse_12h_clock(clock: &str) -> Option<(i8, i8)> {
3878 let clock = clock.trim().to_lowercase();
3879 let (digits, pm) = clock
3880 .strip_suffix("am")
3881 .map(|d| (d, false))
3882 .or_else(|| clock.strip_suffix("pm").map(|d| (d, true)))?;
3883 let (h, m) = digits.trim().split_once(':')?;
3884 let mut hour: i8 = h.trim().parse().ok()?;
3885 let minute: i8 = m.trim().parse().ok()?;
3886 if !(1..=12).contains(&hour) || !(0..=59).contains(&minute) {
3887 return None;
3888 }
3889 if pm && hour != 12 {
3890 hour += 12;
3891 } else if !pm && hour == 12 {
3892 hour = 0;
3893 }
3894 Some((hour, minute))
3895}
3896
3897fn parse_reset_hint_zoned(text: &str, now: Timestamp) -> Option<Timestamp> {
3902 let open = text.find('(')?;
3903 let close = text.rfind(')')?;
3904 if close <= open {
3905 return None;
3906 }
3907 let zone = text[open + 1..close].trim();
3908 let (hour, minute) = parse_12h_clock(&text[..open])?;
3909 let tz = jiff::tz::TimeZone::get(zone).ok()?;
3910 let candidate = now
3911 .to_zoned(tz)
3912 .with()
3913 .hour(hour)
3914 .minute(minute)
3915 .second(0)
3916 .millisecond(0)
3917 .microsecond(0)
3918 .nanosecond(0)
3919 .build()
3920 .ok()?;
3921 let mut at = candidate.timestamp();
3922 if at <= now {
3923 at += jiff::SignedDuration::from_hours(24);
3924 }
3925 Some(at)
3926}
3927
3928fn parse_reset_hint_dated(text: &str) -> Option<Timestamp> {
3937 let words: Vec<&str> = text.split_whitespace().collect();
3938 if words.len() < 5 {
3939 return None;
3940 }
3941 (0..=words.len() - 5)
3942 .find_map(|start| parse_dated_window(&words[start..start + 5], words.get(start + 5)))
3943}
3944
3945fn parse_dated_window(window: &[&str], trailing: Option<&&str>) -> Option<Timestamp> {
3951 if trailing.is_some_and(|next| next.starts_with('(')) {
3952 return None;
3953 }
3954 let month = month_number(window[0])?;
3955 let day_token = window[1].strip_suffix(',')?.to_lowercase();
3956 let day_digits = ["st", "nd", "rd", "th"]
3957 .iter()
3958 .find_map(|suffix| day_token.strip_suffix(*suffix))?;
3959 let day: i8 = day_digits.parse().ok()?;
3960 let year_token = window[2];
3961 if year_token.len() != 4 || !year_token.bytes().all(|b| b.is_ascii_digit()) {
3962 return None;
3963 }
3964 let year: i16 = year_token.parse().ok()?;
3965 let ampm = window[4].trim_matches(|c: char| !c.is_ascii_alphabetic());
3969 let (hour, minute) = parse_12h_clock(&format!("{}{}", window[3], ampm))?;
3970 let date = jiff::civil::Date::new(year, month, day).ok()?;
3971 let candidate = date
3972 .at(hour, minute, 0, 0)
3973 .to_zoned(jiff::tz::TimeZone::UTC)
3974 .ok()?;
3975 Some(candidate.timestamp())
3976}
3977
3978fn month_number(name: &str) -> Option<i8> {
3981 const NAMES: [&str; 12] = [
3982 "jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec",
3983 ];
3984 let lower = name.to_lowercase();
3985 NAMES
3986 .iter()
3987 .position(|n| *n == lower.as_str())
3988 .map(|i| i as i8 + 1)
3989}
3990
3991fn exhausted_review_budget(state: &RunState) -> bool {
4003 state.status == RunStatus::Blocked
4006 && !state.pr_publication_pending
4007 && state.reviews.len() >= state.config.graph.review_rounds
4008}
4009
4010fn unfinished_run(runs: &[String], short: &str) -> Option<String> {
4047 unfinished_run_with(runs, short, RunState::load)
4048}
4049
4050fn unfinished_run_with<F>(runs: &[String], short: &str, load: F) -> Option<String>
4053where
4054 F: FnOnce(&str) -> Result<RunState>,
4055{
4056 let id = runs.last()?;
4057 match load(id) {
4058 Ok(s)
4063 if s.status.resumable()
4064 && !s.released()
4065 && !exhausted_review_budget(&s)
4066 && s.liveness(false) != crate::run::Liveness::Live =>
4067 {
4068 Some(id.clone())
4069 }
4070 Ok(_) => None,
4071 Err(e) => {
4072 tracing::warn!("could not read run {id} for task {short}: {e:#}");
4073 None
4074 }
4075 }
4076}
4077
4078fn awaiting_resume_with<F>(task: &Task, load: F) -> bool
4086where
4087 F: FnOnce(&str) -> Result<RunState>,
4088{
4089 if !matches!(task.status, TaskStatus::Parked | TaskStatus::Failed)
4091 || task.fresh_start
4092 || task.review_branch.is_some()
4093 {
4094 return false;
4095 }
4096 let Some(id) = task.runs.last() else {
4097 return false;
4098 };
4099 load(id).is_ok_and(|s| {
4100 s.parked && s.status.resumable() && !s.released() && !exhausted_review_budget(&s)
4101 })
4102}
4103
4104#[derive(Debug, Clone, PartialEq, Eq)]
4107enum Starter {
4108 Review(String),
4111 Resume(String),
4113 Start,
4115}
4116
4117fn selected_starter(
4119 selected: &crate::pr_recovery::Selection,
4120 previous: Option<&RunState>,
4121 unfinished: Option<&str>,
4122) -> Starter {
4123 if let Some(id) = unfinished
4124 && previous.is_some_and(|r| {
4125 r.id == id && r.pr_recovery.as_ref() == Some(selected) && r.reviewed_commits.is_some()
4126 })
4127 {
4128 Starter::Resume(id.to_owned())
4129 } else {
4130 Starter::Review(selected.branch.clone())
4131 }
4132}
4133
4134fn take_divergence_answer(
4138 branch: &str,
4139 remote: &str,
4140 task: &mut Task,
4141) -> Option<crate::reconcile::Choice> {
4142 let summary = crate::reconcile::summary_for(branch, remote);
4143 let (idx, choice) = task.answers.iter().enumerate().rev().find_map(|(i, a)| {
4144 (a.question == summary)
4145 .then(|| crate::reconcile::Choice::from_answer(&a.answer))
4146 .flatten()
4147 .map(|c| (i, c))
4148 })?;
4149 task.answers.remove(idx);
4150 Some(choice)
4151}
4152
4153fn choose_starter(
4165 review_branch: Option<&str>,
4166 branch_exists: bool,
4167 unfinished: Option<&str>,
4168) -> Starter {
4169 match review_branch {
4170 Some(branch) if branch_exists => Starter::Review(branch.to_owned()),
4171 Some(_) => Starter::Start,
4172 None => match unfinished {
4173 Some(id) => Starter::Resume(id.to_owned()),
4174 None => Starter::Start,
4175 },
4176 }
4177}
4178
4179fn take_review_branch(task: &mut Task, previous: Option<&RunState>) -> Option<String> {
4181 let branch = task.review_branch.take();
4182 let instruction = task.review_instruction.take();
4183 branch.filter(|branch| match &instruction {
4184 Some(authored) => *authored == task.instruction,
4185 None => previous.map_or_else(
4188 || task.review_of.as_ref() == Some(branch),
4189 |run| compatible_run_instruction(run, task),
4190 ),
4191 })
4192}
4193
4194fn may_resume_run(run: &RunState, task: &Task) -> bool {
4197 task.resume_override
4198 .as_ref()
4199 .is_some_and(|r| r.pinned_run.as_ref() == Some(&run.id))
4200 || compatible_run_instruction(run, task)
4201}
4202
4203pub(crate) fn compatible_run_instruction(run: &RunState, task: &Task) -> bool {
4205 if let Some(authored) = &run.task_instruction {
4206 return compatible_instruction(authored, task);
4207 }
4208 if compatible_instruction(&run.instruction, task) {
4209 return true;
4210 }
4211 if run.reviewed_commits.is_some() {
4214 if let Some((authored, _)) = run.instruction.split_once("\n\n# Work already on branch `") {
4215 return strip_answers_block(authored, task).trim() == task.instruction.trim();
4216 }
4217 if let Some(branch) = &task.review_of {
4218 let canonical = format!(
4219 "Review the branch `{branch}` (review-only: no implementation, no judging)."
4220 );
4221 let generated = format!(
4222 "Review the work already on branch `{branch}`. There is no task statement: what the change claims to do is whatever its commits say.\n\n"
4223 );
4224 return task.instruction == canonical && run.instruction.starts_with(&generated);
4225 }
4226 }
4227 false
4228}
4229
4230pub(crate) fn compatible_instruction(old: &str, task: &Task) -> bool {
4232 strip_answers_block(old, task) == task.instruction
4233}
4234
4235fn repo_for(task: &Task, fallback: &Path) -> PathBuf {
4238 if task.repo.as_os_str().is_empty() || task.repo == Path::new(".") {
4239 return fallback.to_path_buf();
4240 }
4241 task.repo.clone()
4242}
4243
4244const ANSWERS_HEADER: &str = "\n\n# Operator answers\n\n";
4248
4249fn answers_block(task: &Task, count: usize) -> String {
4251 let mut s = ANSWERS_HEADER.to_owned();
4252 for a in &task.answers[..count] {
4253 s.push_str(&format!("- {}: {}\n", a.question, a.answer));
4254 }
4255 s
4256}
4257
4258fn resumption_block(r: &crate::queue::Resumption) -> String {
4261 format!(
4262 "\n\n# Explicit operator resumption\n\nGeneration {} at {} approved this instruction:\n{}\n\
4263 Earlier answers (first {}) are audit history; their waiting premises are superseded by this approval.\n",
4264 r.generation, r.at, r.instruction, r.answers_through,
4265 )
4266}
4267
4268fn append_answers(base: &str, task: &Task) -> String {
4270 let mut s = base.to_owned();
4271 if !task.answers.is_empty() {
4272 s.push_str(&answers_block(task, task.answers.len()));
4273 }
4274 if let Some(r) = &task.resumption {
4275 s.push_str(&resumption_block(r));
4276 }
4277 s
4278}
4279
4280fn strip_answers_block<'a>(instruction: &'a str, task: &Task) -> &'a str {
4284 let mut instruction = instruction;
4285 for approval in task
4286 .resumption
4287 .iter()
4288 .chain(task.resumption_history.iter().rev())
4289 {
4290 if let Some(base) = instruction.strip_suffix(&resumption_block(approval)) {
4291 instruction = base;
4292 break;
4293 }
4294 }
4295 for count in (1..=task.answers.len()).rev() {
4296 let block = answers_block(task, count);
4297 if let Some(base) = instruction.strip_suffix(&block) {
4298 return base;
4299 }
4300 }
4301 instruction
4302}
4303
4304fn instruction_for(task: &Task) -> String {
4312 append_answers(&task.instruction, task)
4313}
4314
4315fn resumed_instruction(old_instruction: &str, task: &Task) -> String {
4327 append_answers(strip_answers_block(old_instruction, task), task)
4328}
4329
4330fn task_attachments(queue: &Queue, task: &Task) -> Result<Vec<PathBuf>> {
4332 let paths = queue.attachment_paths(task);
4333 for (name, path) in task.attachments.iter().zip(&paths) {
4334 if !path.is_file() {
4335 bail!(
4336 "attachment `{name}` is recorded on the task but {} is missing",
4337 path.display()
4338 );
4339 }
4340 }
4341 Ok(paths)
4342}
4343
4344fn prepare_instruction(
4355 starter: &Starter,
4356 old_instruction: Option<&str>,
4357 task: &Task,
4358) -> Option<String> {
4359 match starter {
4360 Starter::Start => Some(instruction_for(task)),
4361 Starter::Resume(_) => Some(resumed_instruction(
4362 old_instruction.expect("a resumed run always has a prior instruction"),
4363 task,
4364 )),
4365 Starter::Review(_) => None,
4366 }
4367}
4368
4369fn record(queue: &Queue, task: &mut Task) {
4373 if let Err(e) = queue.put(task) {
4374 tracing::error!("could not record task {}: {e:#}", task.short());
4375 let notice = Notice::error(
4376 "loop:record",
4377 "The loop could not save a task's state; check the disk.",
4378 )
4379 .link(Link::Task {
4380 id: task.id.clone(),
4381 });
4382 match queue.root().parent().filter(|p| !p.as_os_str().is_empty()) {
4384 Some(home) => notices::raise_in(home, notice),
4385 None => notices::raise(notice),
4386 }
4387 }
4388}
4389
4390fn runnable(queue: &Queue) -> Vec<Task> {
4396 let mut tasks: Vec<Task> = queue
4397 .list()
4398 .into_iter()
4399 .filter(|t| t.status.runnable())
4400 .collect();
4401 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
4402 tasks
4403}
4404
4405fn describe(state: &RunState) -> String {
4419 let p = phrases(&state.config.graph.language);
4420 let mut detail = if state.status == RunStatus::Stalled {
4421 let mut seats: Vec<&str> = state.quota.iter().map(|q| q.seat.as_str()).collect();
4422 seats.sort_unstable();
4423 seats.dedup();
4424 if seats.is_empty() {
4425 p.quorum_lost.to_owned()
4426 } else {
4427 format!("{}{}{}", p.quorum_lost, p.quota_took_out, seats.join(", "))
4428 }
4429 } else {
4430 format!("{}{}", p.run_ended, state.status.display_label())
4431 };
4432 if let Some(last) = state.events.last() {
4433 let unanswered = state
4437 .reviews
4438 .last()
4439 .filter(|r| {
4440 state.status == RunStatus::Blocked
4441 && last.node == "review"
4442 && r.incomplete()
4443 && r.blocking == 0
4444 && r.round == state.config.graph.review_rounds
4445 && r.e2e.iter().all(crate::run::CommandOutcome::ok)
4446 })
4447 .map(|r| (r.expected - r.answered, r.round));
4448 match unanswered {
4449 Some((missing, rounds)) if crate::lang::is_japanese(&state.config.graph.language) => {
4450 detail.push_str(&format!(
4451 " ({}: {})",
4452 last.node,
4453 (p.reviewers_never_answered)(missing, rounds)
4454 ));
4455 }
4456 _ => detail.push_str(&format!(" ({}: {})", last.node, last.message)),
4457 }
4458 }
4459 detail.push_str(&format!(" [run {}]", state.id));
4460 detail
4461}
4462
4463const DIAGNOSTIC_MAX: usize = 4_000;
4469
4470const DIAGNOSTIC_OUTPUT_TAIL: usize = 800;
4475
4476fn diagnostic(state: &RunState) -> Option<String> {
4490 let mut parts: Vec<String> = Vec::new();
4491
4492 for o in state.gate.iter().filter(|o| !o.ok()) {
4494 parts.push(format!(
4495 "gate `{}` failed ({:?}):\n{}",
4496 o.command,
4497 o.code,
4498 crate::run::tail(&o.output_tail, DIAGNOSTIC_OUTPUT_TAIL)
4499 ));
4500 }
4501
4502 if let Some(last) = state
4505 .events
4506 .iter()
4507 .rev()
4508 .find(|e| e.node == "land" && e.message.contains("fixer produced no commit"))
4509 {
4510 parts.push(last.message.clone());
4511 }
4512
4513 if state.viable().is_empty() {
4520 for c in &state.candidates {
4521 if let Some(evidence) = &c.verified_noop {
4522 parts.push(format!(
4523 "candidate {} (agent-verified no-op, unconfirmed by magi): {evidence}",
4524 c.label
4525 ));
4526 } else if !c.summary.trim().is_empty() {
4527 parts.push(format!("candidate {}: {}", c.label, c.summary.trim()));
4528 } else if let Some(why) = &c.failed {
4529 parts.push(format!("candidate {}: {why}", c.label));
4530 }
4531 }
4532 }
4533
4534 if parts.is_empty() {
4535 return None;
4536 }
4537 Some(crate::run::tail(
4542 &parts.join("\n\n"),
4543 DIAGNOSTIC_MAX.saturating_sub(100),
4544 ))
4545}
4546
4547fn label(status: RunStatus) -> &'static str {
4555 status.as_str()
4556}
4557
4558pub(crate) fn merge_mode(mode: &str) -> Result<MergeMode> {
4560 match mode {
4561 "none" => Ok(MergeMode::None),
4562 "local" => Ok(MergeMode::Local),
4563 "pr" => Ok(MergeMode::Pr),
4564 other => bail!("unknown merge mode `{other}`; expected none, local or pr"),
4565 }
4566}
4567
4568fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
4573 mutex
4574 .lock()
4575 .unwrap_or_else(std::sync::PoisonError::into_inner)
4576}
4577
4578#[cfg(test)]
4579mod tests {
4580 use super::*;
4581 use crate::queue::{Source, TaskStatus};
4582 use crate::run::{Candidate, CommandOutcome};
4583 use pretty_assertions::assert_eq;
4584
4585 fn task() -> Task {
4586 Task::new(
4587 "add retries".to_owned(),
4588 "add retries".to_owned(),
4589 PathBuf::from("/repo"),
4590 Source::Human,
4591 )
4592 }
4593
4594 #[test]
4595 fn selected_public_head_overrides_historical_resume_and_pin() {
4596 let mut t = task();
4597 let mut old = run_state(RunStatus::Blocked);
4598 let selection = crate::pr_recovery::Selection {
4599 url: "url".into(),
4600 branch: format!("magi/task-{}", t.id),
4601 head: "a".repeat(40),
4602 instruction: t.instruction.clone(),
4603 approved_at: Timestamp::now(),
4604 };
4605 t.resume_override = Some(crate::queue::OperatorResume {
4606 pinned_run: Some(old.id.clone()),
4607 question_id: "choice".into(),
4608 at: Timestamp::now(),
4609 conductor_rehold: None,
4610 forced: false,
4611 });
4612 assert_eq!(
4613 selected_starter(&selection, Some(&old), Some(&old.id)),
4614 Starter::Review(selection.branch.clone())
4615 );
4616 old.pr_recovery = Some(selection.clone());
4617 assert_eq!(
4618 selected_starter(&selection, Some(&old), Some(&old.id)),
4619 Starter::Review(selection.branch.clone())
4620 );
4621 old.reviewed_commits = Some(vec!["implementation".into()]);
4622 assert_eq!(
4623 selected_starter(&selection, Some(&old), Some(&old.id)),
4624 Starter::Resume(old.id.clone())
4625 );
4626 let mut changed = selection.clone();
4627 changed.head = "b".repeat(40);
4628 assert_eq!(
4629 selected_starter(&changed, Some(&old), Some(&old.id)),
4630 Starter::Review(changed.branch)
4631 );
4632 }
4633
4634 #[test]
4635 fn candidate_transplant_failure_is_one_actionable_hold() {
4636 let dir = tempfile::tempdir().unwrap();
4637 let q = Queue::at(dir.path().join("queue"));
4638 let mut t = task();
4639 let mut state = run_state(RunStatus::Blocked);
4640 state.event("prep", "cannot apply referenced commit: conflict");
4641 state.prep_failure = Some("conflict".into());
4642 t.start(state.id.clone());
4643 finish_attempt(3, &q, &mut t, &state, &[], Err(anyhow::anyhow!("conflict")));
4644 assert_eq!(t.attempts, 1);
4645 assert_eq!(t.status, TaskStatus::Held);
4646 assert!(t.operator_held());
4647 assert!(!t.status.runnable());
4648 assert!(t.hold_reason.as_deref().unwrap().contains("recover-pr"));
4649 }
4650
4651 #[test]
4652 fn revised_resumption_invalidates_old_pins_and_preserves_unchanged_pins() {
4653 for source in ["cli", "web"] {
4654 let mut t = task();
4655 let mut run = run_state(RunStatus::Blocked);
4656 run.instruction = "Run A original criteria".into();
4657 run.task_instruction = Some(run.instruction.clone());
4658 t.start(run.id.clone());
4659 t.release_operator("choice", Some("resume-question"));
4660 t.resume_override = Some(crate::queue::OperatorResume {
4661 question_id: "resume-question".into(),
4662 at: Timestamp::now(),
4663 conductor_rehold: None,
4664 forced: true,
4665 pinned_run: Some(run.id.clone()),
4666 });
4667 t.hold_machine(Some("not enough free space to start a run".into()));
4668 let mut unchanged = t.clone();
4669 unchanged
4670 .edit("title changed".into(), unchanged.instruction.clone())
4671 .unwrap();
4672 unchanged.release_operator(source, None);
4673 assert_eq!(
4674 unchanged
4675 .resume_override
4676 .as_ref()
4677 .unwrap()
4678 .pinned_run
4679 .as_deref(),
4680 Some(run.id.as_str())
4681 );
4682 assert!(!compatible_run_instruction(&run, &unchanged));
4683 assert!(may_resume_run(&run, &unchanged));
4684
4685 t.edit(
4686 "approved recovery".into(),
4687 "Reuse measured PR head B; fix Clippy; review".into(),
4688 )
4689 .unwrap();
4690 assert!(t.resume_override.is_none());
4691 t.release_operator(source, None);
4692 let reloaded: Task = serde_json::from_str(&serde_json::to_string(&t).unwrap()).unwrap();
4693 assert!(reloaded.resume_override.is_none());
4694 assert!(!may_resume_run(&run, &reloaded));
4695 let unfinished =
4696 unfinished_run_with(&reloaded.runs, reloaded.short(), |_| Ok(run.clone()))
4697 .filter(|_| may_resume_run(&run, &reloaded));
4698 assert_eq!(
4699 choose_starter(None, false, unfinished.as_deref()),
4700 Starter::Start
4701 );
4702 assert!(instruction_for(&reloaded).starts_with("Reuse measured PR head B"));
4703 assert_eq!(
4704 reloaded.resumption_history[0].question_id.as_deref(),
4705 Some("resume-question")
4706 );
4707
4708 unchanged.instruction = "Reuse measured PR head B".into();
4711 unchanged.hold_machine(Some("disk hold".into()));
4712 unchanged.release_operator(source, None);
4713 assert!(unchanged.resume_override.is_none());
4714 assert!(!may_resume_run(&run, &unchanged));
4715 }
4716 }
4717
4718 #[test]
4719 fn resumption_preserves_generated_review_runs_and_rejects_revised_targets() {
4720 let mut t = task();
4721 let mut run = run_state(RunStatus::Blocked);
4722 t.review_of = Some("approved-head".into());
4723 t.instruction =
4724 "Review the branch `approved-head` (review-only: no implementation, no judging)."
4725 .into();
4726 run.instruction = "Review the work already on branch `approved-head`. There is no task statement: what the change claims to do is whatever its commits say.\n\nCommit log".into();
4727 run.reviewed_commits = Some(vec!["Commit log".into()]);
4728 t.start(run.id.clone());
4729 t.hold_machine(Some("interrupted review".into()));
4730 t.release_operator("cli", None);
4731 assert!(
4732 compatible_run_instruction(&run, &t),
4733 "legacy explicit review remains resumable"
4734 );
4735 let unfinished = unfinished_run_with(&t.runs, t.short(), |_| Ok(run.clone()))
4736 .filter(|_| compatible_run_instruction(&run, &t));
4737 assert_eq!(
4738 choose_starter(None, false, unfinished.as_deref()),
4739 Starter::Resume(run.id.clone())
4740 );
4741 run.task_instruction = Some(t.instruction.clone());
4742 assert!(compatible_run_instruction(&run, &t));
4743 let reloaded: RunState =
4744 serde_json::from_str(&serde_json::to_string(&run).unwrap()).unwrap();
4745 assert!(compatible_run_instruction(&reloaded, &t));
4746 t.edit(
4747 "new approved head".into(),
4748 "Reuse measured PR head B".into(),
4749 )
4750 .unwrap();
4751 assert!(!compatible_run_instruction(&run, &t));
4752
4753 t.review_of = None;
4755 t.instruction = "Fix Clippy on approved head".into();
4756 run.task_instruction = None;
4757 run.instruction = format!(
4758 "{}\n\n# Work already on branch `approved-head` (commit subjects)\n\nCommit log",
4759 t.instruction
4760 );
4761 assert!(compatible_run_instruction(&run, &t));
4762 t.instruction.push('\n');
4763 assert!(
4764 compatible_run_instruction(&run, &t),
4765 "legacy generated review trimmed the original criteria"
4766 );
4767 t.hold_for_handover(Some("approved-head".into()), "refused".into());
4768 let mut unchanged = t.clone();
4769 assert_eq!(
4770 take_review_branch(&mut unchanged, Some(&run)).as_deref(),
4771 Some("approved-head")
4772 );
4773 t.edit("new plan".into(), "Reuse measured PR head B".into())
4774 .unwrap();
4775 assert!(take_review_branch(&mut t, Some(&run)).is_none());
4776 assert!(!compatible_run_instruction(&run, &t));
4777 t.review_branch = Some("approved-head".into());
4778 t.review_instruction = None;
4779 assert!(
4780 take_review_branch(&mut t, Some(&run)).is_none(),
4781 "legacy retained candidate cannot replace a revised approved head"
4782 );
4783 }
4784
4785 #[test]
4786 fn resumption_actions_survive_automatic_holds_of_hand_started_runs() {
4787 crate::run::pin_test_home();
4788 let dir = tempfile::tempdir().unwrap();
4789 let queue = Queue::at(dir.path().join("queue"));
4790 for resume in [true, false] {
4791 let state = run_state(RunStatus::Blocked);
4792 let mut t = task();
4793 t.start(state.id.clone());
4794 let action = if resume {
4795 ask::ChoiceAction::Resume {
4796 run: state.id.clone(),
4797 }
4798 } else {
4799 ask::ChoiceAction::Requeue
4800 };
4801 let mut q = action_question(&state.id, action);
4802 q.status = ask::QuestionStatus::Open;
4803 q.answer = None;
4804 q.answered_at = None;
4805 t.fail("run stopped with a question", 3);
4806 hold_if_runnable(&queue, &mut t);
4807 assert!(t.operator_decided_at.is_none());
4808 assert!(t.operator_held());
4809 q.answer(ask::Answer::Choice(q.choices[0].clone())).unwrap();
4810 let expected = if resume {
4811 ActionDecision::Resume(state.id.clone())
4812 } else {
4813 ActionDecision::Requeue
4814 };
4815 assert_eq!(
4816 decide_action(&t, &q, phrases("en"), |_| Ok(state.clone())),
4817 expected
4818 );
4819 t.hold_manual(Some("later explicit hold".into()));
4820 assert_eq!(
4821 decide_action(&t, &q, phrases("en"), |_| Ok(state.clone())),
4822 ActionDecision::Stale
4823 );
4824 }
4825 }
4826
4827 #[test]
4828 fn explicit_resumption_keeps_revised_instruction_and_audit_answers() {
4829 let mut t = task();
4830 t.answers.push(crate::queue::AnsweredQuestion {
4831 question: "Inspect the remote?".into(),
4832 answer: "Inspect then wait".into(),
4833 });
4834 let old = instruction_for(&t);
4835 assert!(compatible_instruction(&old, &t));
4836 t.instruction = "Reuse approved PR head and review".into();
4837 t.release_operator("cli", None);
4838 assert!(!compatible_instruction(&old, &t));
4839 let approved = instruction_for(&t);
4840 assert!(approved.contains("waiting premises are superseded"));
4841 assert!(compatible_instruction(&approved, &t));
4842 assert_eq!(resumed_instruction(&approved, &t), approved);
4843 t.release_operator("web", None);
4844 assert!(compatible_instruction(&approved, &t));
4845 assert_eq!(t.resumption_history.len(), 1);
4846 t.start("run".into());
4847 assert!(instruction_for(&t).contains("Explicit operator resumption"));
4848 }
4849
4850 #[test]
4854 fn a_resource_blocked_run_holds_its_task_and_spends_no_attempt() {
4855 let mut state = RunState::new(
4856 PathBuf::from("/repo"),
4857 "main".to_owned(),
4858 "abc1234".to_owned(),
4859 "task".to_owned(),
4860 Config::default(),
4861 );
4862 state.status = RunStatus::Blocked;
4863 state.resource_block = Some(crate::run::ResourceBlock {
4864 node: "graft".to_owned(),
4865 kind: "shared build cache".to_owned(),
4866 detail: "timed out".to_owned(),
4867 at: jiff::Timestamp::now(),
4868 });
4869 let mut t = task();
4870 t.start("20260902-000000-aaaa".to_owned());
4871 assert_eq!(t.attempts, 1);
4872 let verdict = Verdict {
4873 status: RunStatus::Blocked,
4874 left_pr: false,
4875 quota_hit: false,
4876 parked: false,
4877 no_viable_candidates: false,
4878 };
4879 settle_and_diagnose(&mut t, verdict, "blocked", 2, &state);
4880 assert_eq!(t.status, TaskStatus::Held);
4881 assert_eq!(t.attempts, 0, "the attempt is refunded");
4882 assert!(t.hold_reason.as_deref().unwrap().starts_with("[resource] "));
4883 assert!(
4884 !t.operator_held(),
4885 "a machine hold, released without a human"
4886 );
4887
4888 let mut p = task();
4890 p.start("20260902-000000-bbbb".to_owned());
4891 settle_and_diagnose(
4892 &mut p,
4893 Verdict {
4894 parked: true,
4895 ..verdict
4896 },
4897 "parked",
4898 2,
4899 &state,
4900 );
4901 assert_eq!(p.status, TaskStatus::Parked);
4902 }
4903
4904 #[test]
4905 fn record_failure_notice_links_the_task() {
4906 let dir = tempfile::tempdir().unwrap();
4907 std::fs::write(dir.path().join("queue"), b"not a directory").unwrap();
4910 let queue = Queue::at(dir.path().join("queue"));
4911 let mut t = interrupt_task("rec-fail-1");
4912 record(&queue, &mut t);
4913
4914 let found = crate::notices::Notices::at(dir.path().join("notifications")).list();
4915 assert_eq!(found.len(), 1);
4916 assert_eq!(found[0].key, "loop:record");
4917 assert_eq!(
4918 found[0].link,
4919 Some(Link::Task {
4920 id: "rec-fail-1".to_owned()
4921 })
4922 );
4923 }
4924
4925 fn interrupt_task(id: &str) -> Task {
4928 let mut t = task();
4929 t.id = id.to_owned();
4930 t.interrupt = true;
4931 t
4932 }
4933
4934 fn task_with_id(id: &str) -> Task {
4936 let mut t = task();
4937 t.id = id.to_owned();
4938 t
4939 }
4940
4941 fn urgent_task(id: &str) -> Task {
4943 let mut t = task();
4944 t.id = id.to_owned();
4945 t.urgent = true;
4946 t
4947 }
4948
4949 #[test]
4950 fn urgent_dispatch_has_one_extra_slot_and_never_uses_a_stale_flag() {
4951 let dir = tempfile::tempdir().unwrap();
4952 let queue = Queue::at(dir.path().join("queue"));
4953 let ordinary = Arc::new(tokio::sync::Semaphore::new(1));
4954 let urgent = Arc::new(tokio::sync::Semaphore::new(1));
4955 let occupied = Arc::clone(&ordinary).try_acquire_owned().unwrap();
4956 let mut first = task();
4957 queue.put(&mut first).unwrap();
4958 let stale_plain = queue.get(&first.id).unwrap();
4959 first.set_urgent(true).unwrap();
4960 queue.put(&mut first).unwrap();
4961 let dispatched = claim_dispatch(&queue, &stale_plain, false, &ordinary, &urgent).unwrap();
4962 let mut second = urgent_task("second");
4963 queue.put(&mut second).unwrap();
4964 assert!(claim_dispatch(&queue, &second, false, &ordinary, &urgent).is_none());
4965 drop(queue.claim(&second.id).unwrap());
4966 assert_eq!(queue.get(&second.id).unwrap().attempts, 0);
4967 drop(dispatched);
4968 let stale_urgent = queue.get(&first.id).unwrap();
4969 first.set_urgent(false).unwrap();
4970 queue.put(&mut first).unwrap();
4971 assert!(claim_dispatch(&queue, &stale_urgent, false, &ordinary, &urgent).is_none());
4972 assert_eq!(urgent.available_permits(), 1);
4973 drop(queue.claim(&first.id).unwrap());
4974 drop(occupied);
4975 }
4976
4977 #[test]
4978 fn urgent_dispatch_uses_saved_lane_and_releases_unsuccessful_claims() {
4979 let dir = tempfile::tempdir().unwrap();
4980 let queue = Queue::at(dir.path().join("queue"));
4981 let ordinary = Arc::new(tokio::sync::Semaphore::new(1));
4982 let urgent = Arc::new(tokio::sync::Semaphore::new(1));
4983 let mut t = task();
4984 queue.put(&mut t).unwrap();
4985 for enabled in [true, false] {
4986 let stale = queue.get(&t.id).unwrap();
4987 t.set_urgent(enabled).unwrap();
4988 queue.put(&mut t).unwrap();
4989 assert_ne!(stale.urgent, enabled);
4990 let DispatchClaim {
4991 task: fresh,
4992 claim,
4993 permit,
4994 } = claim_dispatch(&queue, &stale, false, &ordinary, &urgent).unwrap();
4995 assert_eq!(fresh.urgent, enabled);
4996 assert_eq!(ordinary.available_permits(), usize::from(enabled));
4997 assert_eq!(urgent.available_permits(), usize::from(!enabled));
4998 assert!(queue.claim(&t.id).is_err());
4999 assert_eq!(fresh.attempts, stale.attempts);
5000 drop((claim, permit));
5001 assert_eq!(ordinary.available_permits(), 1);
5002 assert_eq!(urgent.available_permits(), 1);
5003 }
5004 t.set_urgent(true).unwrap();
5005 queue.put(&mut t).unwrap();
5006 let occupied = Arc::clone(&urgent).try_acquire_owned().unwrap();
5007 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5008 assert_eq!(ordinary.available_permits(), 1, "no ordinary fallback");
5009 drop(queue.claim(&t.id).unwrap());
5010 assert_eq!(queue.get(&t.id).unwrap().attempts, t.attempts);
5011 drop(occupied);
5012 assert!(claim_dispatch(&queue, &t, true, &ordinary, &urgent).is_none());
5013 drop(queue.claim(&t.id).unwrap());
5014 let held = queue.claim(&t.id).unwrap();
5015 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5016 drop(held);
5017 t.status = TaskStatus::Held;
5018 queue.put(&mut t).unwrap();
5019 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5020 drop(queue.claim(&t.id).unwrap());
5021 std::fs::remove_file(queue.path_of(&t.id)).unwrap();
5022 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5023 drop(queue.claim(&t.id).unwrap());
5024 assert_eq!(urgent.available_permits(), 1);
5025 }
5026
5027 #[test]
5032 fn permit_kind_prefers_a_land_resume_over_the_urgent_slot() {
5033 assert_eq!(permit_kind(true, false), PermitKind::None);
5034 assert_eq!(permit_kind(true, true), PermitKind::None);
5035 }
5036
5037 #[test]
5042 fn permit_kind_separates_urgent_from_ordinary() {
5043 assert_eq!(permit_kind(false, true), PermitKind::Urgent);
5044 assert_eq!(permit_kind(false, false), PermitKind::Ordinary);
5045 }
5046
5047 #[test]
5054 fn disk_gate_with_holds_a_task_below_the_threshold_and_names_both_numbers() {
5055 let cfg = Config::default();
5056 let repo = Path::new("/any/repo/path");
5057
5058 let reason =
5059 disk_gate_with(repo, &cfg, |_| Ok(1024)).expect("must hold below the threshold");
5060 assert!(reason.contains("1024"), "{reason}");
5061 assert!(
5062 reason.contains(&cfg.disk.min_free_bytes.to_string()),
5063 "{reason}"
5064 );
5065
5066 assert_eq!(
5067 disk_gate_with(repo, &cfg, |_| Ok(cfg.disk.min_free_bytes)),
5068 None,
5069 "exactly at the floor is open"
5070 );
5071 assert_eq!(
5072 disk_gate_with(repo, &cfg, |_| Ok(cfg.disk.min_free_bytes + 1)),
5073 None,
5074 "comfortably above the floor is open"
5075 );
5076 }
5077
5078 #[test]
5079 fn disk_gate_with_opens_unconditionally_when_the_operator_opted_out() {
5080 let mut cfg = Config::default();
5081 cfg.disk.min_free_bytes = 0;
5082 let repo = Path::new("/any/repo/path");
5083 assert_eq!(
5084 disk_gate_with(repo, &cfg, |_| Ok(0)),
5085 None,
5086 "a zero floor never measures at all"
5087 );
5088 }
5089
5090 #[test]
5091 fn disk_gate_with_closes_rather_than_starts_blind_when_it_cannot_measure() {
5092 let cfg = Config::default();
5093 let repo = Path::new("/any/repo/path");
5094 let reason = disk_gate_with(repo, &cfg, |_| Err(anyhow::anyhow!("no df on this box")))
5095 .expect("a measurement failure must close the gate, not open it");
5096 assert!(reason.contains("could not measure"), "{reason}");
5097 }
5098
5099 #[test]
5100 fn no_interrupt_task_leaves_the_sequence_idle_even_with_something_in_flight() {
5101 let ordinary = task();
5102 let next = advance_interrupt(
5103 Interrupt::Idle,
5104 std::slice::from_ref(&ordinary.id),
5105 std::slice::from_ref(&ordinary),
5106 );
5107 assert_eq!(next, Interrupt::Idle);
5108 }
5109
5110 #[test]
5111 fn an_interrupt_task_with_nothing_in_flight_never_starts_a_sequence() {
5112 let marked = interrupt_task("marked");
5115 let next = advance_interrupt(Interrupt::Idle, &[], std::slice::from_ref(&marked));
5116 assert_eq!(next, Interrupt::Idle);
5117 }
5118
5119 #[test]
5120 fn an_interrupt_task_with_something_in_flight_starts_parking_it() {
5121 let marked = interrupt_task("marked");
5122 let next = advance_interrupt(
5123 Interrupt::Idle,
5124 &["running".to_owned()],
5125 std::slice::from_ref(&marked),
5126 );
5127 assert_eq!(
5128 next,
5129 Interrupt::Parking {
5130 parked: vec!["running".to_owned()],
5131 interrupt_task: "marked".to_owned(),
5132 }
5133 );
5134 }
5135
5136 #[test]
5145 fn more_than_one_run_in_flight_never_starts_an_interrupt_sequence() {
5146 let marked = interrupt_task("marked");
5147
5148 let two = advance_interrupt(
5149 Interrupt::Idle,
5150 &["a".to_owned(), "b".to_owned()],
5151 std::slice::from_ref(&marked),
5152 );
5153 assert_eq!(two, Interrupt::Idle);
5154
5155 let none = advance_interrupt(Interrupt::Idle, &[], std::slice::from_ref(&marked));
5156 assert_eq!(none, Interrupt::Idle, "nothing to interrupt either");
5157 }
5158
5159 #[test]
5160 fn parking_holds_until_every_parked_id_has_actually_left_flight() {
5161 let state = Interrupt::Parking {
5162 parked: vec!["running".to_owned()],
5163 interrupt_task: "marked".to_owned(),
5164 };
5165 let still_going = advance_interrupt(state.clone(), &["running".to_owned()], &[]);
5167 assert_eq!(still_going, state);
5168
5169 let stopped_but_not_yet_dispatched =
5173 advance_interrupt(state.clone(), &[], &[interrupt_task("marked")]);
5174 assert_eq!(stopped_but_not_yet_dispatched, state);
5175
5176 let dispatched = advance_interrupt(state, &["marked".to_owned()], &[]);
5178 assert_eq!(
5179 dispatched,
5180 Interrupt::Running {
5181 parked: vec!["running".to_owned()],
5182 interrupt_task: "marked".to_owned(),
5183 }
5184 );
5185 }
5186
5187 #[test]
5188 fn the_sequence_moves_to_resuming_the_instant_the_interrupt_tasks_own_run_leaves_flight() {
5189 let state = Interrupt::Running {
5190 parked: vec!["running".to_owned()],
5191 interrupt_task: "marked".to_owned(),
5192 };
5193 let still_running = advance_interrupt(state.clone(), &["marked".to_owned()], &[]);
5194 assert_eq!(still_running, state);
5195
5196 let ended = advance_interrupt(state, &[], &[task_with_id("running")]);
5203 assert_eq!(
5204 ended,
5205 Interrupt::Resuming {
5206 parked: vec!["running".to_owned()]
5207 }
5208 );
5209 }
5210
5211 #[test]
5212 fn resuming_ends_the_instant_a_parked_task_is_seen_in_flight() {
5213 let state = Interrupt::Resuming {
5214 parked: vec!["running".to_owned()],
5215 };
5216 let still_waiting = advance_interrupt(state.clone(), &[], &[task_with_id("running")]);
5217 assert_eq!(still_waiting, state);
5218
5219 let dispatched = advance_interrupt(state, &["running".to_owned()], &[]);
5220 assert_eq!(dispatched, Interrupt::Idle);
5221 }
5222
5223 #[test]
5229 fn an_interrupt_task_that_stops_being_runnable_abandons_the_wait_without_losing_the_parked_run()
5230 {
5231 let state = Interrupt::Parking {
5232 parked: vec!["running".to_owned()],
5233 interrupt_task: "marked".to_owned(),
5234 };
5235 let next = advance_interrupt(state, &[], &[]);
5238 assert_eq!(
5239 next,
5240 Interrupt::Resuming {
5241 parked: vec!["running".to_owned()]
5242 },
5243 "abandoning the interrupt must not abandon the resume it owes"
5244 );
5245 }
5246
5247 #[test]
5250 fn resuming_abandons_a_parked_task_that_stops_being_runnable() {
5251 let state = Interrupt::Resuming {
5252 parked: vec!["running".to_owned()],
5253 };
5254 let next = advance_interrupt(state, &[], &[]);
5255 assert_eq!(
5256 next,
5257 Interrupt::Idle,
5258 "nothing is left to wait for; the loop must not stay wedged"
5259 );
5260 }
5261
5262 #[test]
5263 fn disabled_by_config_the_sequence_can_never_leave_idle() {
5264 let marked = interrupt_task("marked");
5265 let next = advance_interrupt_tick(
5266 false,
5267 Interrupt::Idle,
5268 &["running".to_owned()],
5269 std::slice::from_ref(&marked),
5270 );
5271 assert_eq!(
5272 next,
5273 Interrupt::Idle,
5274 "an unmarked, unconfigured daemon must behave exactly as before"
5275 );
5276 }
5277
5278 #[test]
5279 fn the_gate_blocks_everyone_while_something_parked_is_still_in_flight() {
5280 let state = Interrupt::Parking {
5281 parked: vec!["running".to_owned()],
5282 interrupt_task: "marked".to_owned(),
5283 };
5284 let candidates = vec![interrupt_task("marked"), task()];
5285 let allowed = interrupt_gate(&state, &["running".to_owned()], candidates);
5286 assert!(
5287 allowed.is_empty(),
5288 "nothing may dispatch - not even the interrupt task itself - \
5289 until the parked run has actually stopped"
5290 );
5291 }
5292
5293 #[test]
5307 fn urgent_gains_no_exemption_from_an_active_interrupt_sequence() {
5308 for state in [
5309 Interrupt::Parking {
5310 parked: vec!["running".to_owned()],
5311 interrupt_task: "marked".to_owned(),
5312 },
5313 Interrupt::Running {
5314 parked: vec!["running".to_owned()],
5315 interrupt_task: "marked".to_owned(),
5316 },
5317 Interrupt::Resuming {
5318 parked: vec!["running".to_owned()],
5319 },
5320 ] {
5321 let candidates = vec![
5322 interrupt_task("marked"),
5323 urgent_task("hot"),
5324 task_with_id("ordinary"),
5325 ];
5326 let allowed = interrupt_gate(&state, &["running".to_owned()], candidates);
5327 assert!(
5328 !allowed.iter().any(|t| t.id == "hot"),
5329 "an urgent candidate must wait out the same gate as anything \
5330 else while the run it would run alongside has not actually \
5331 left flight, for state {state:?}: {allowed:?}"
5332 );
5333 }
5334 }
5335
5336 #[test]
5342 fn a_task_marked_both_urgent_and_interrupt_is_admitted_once_the_gate_itself_says_so() {
5343 let state = Interrupt::Resuming {
5344 parked: vec!["hot".to_owned()],
5345 };
5346 let candidates = vec![urgent_task("hot"), task()];
5347 let allowed = interrupt_gate(&state, &[], candidates);
5348 assert_eq!(
5349 allowed.iter().filter(|t| t.id == "hot").count(),
5350 1,
5351 "the gate's own decision is unaffected by the urgent flag: {allowed:?}"
5352 );
5353 }
5354
5355 #[test]
5356 fn the_gate_lets_only_the_interrupt_task_through_once_parked_work_has_stopped() {
5357 let state = Interrupt::Parking {
5358 parked: vec!["running".to_owned()],
5359 interrupt_task: "marked".to_owned(),
5360 };
5361 let other = task();
5362 let candidates = vec![interrupt_task("marked"), other.clone()];
5363 let allowed = interrupt_gate(&state, &[], candidates);
5364 assert_eq!(allowed.len(), 1);
5365 assert_eq!(allowed[0].id, "marked");
5366 }
5367
5368 #[test]
5369 fn the_gate_blocks_everyone_while_the_interrupt_task_itself_is_in_flight() {
5370 let state = Interrupt::Running {
5371 parked: vec!["running".to_owned()],
5372 interrupt_task: "marked".to_owned(),
5373 };
5374 let candidates = vec![task(), task()];
5375 let allowed = interrupt_gate(&state, &["marked".to_owned()], candidates);
5376 assert!(allowed.is_empty());
5377 }
5378
5379 #[test]
5386 fn the_gate_offers_at_most_one_candidate_while_resuming_even_with_two_parked() {
5387 let state = Interrupt::Resuming {
5388 parked: vec!["a".to_owned(), "c".to_owned()],
5389 };
5390 let candidates = vec![task_with_id("a"), task_with_id("c"), task_with_id("other")];
5391 let allowed = interrupt_gate(&state, &[], candidates);
5392 assert_eq!(
5393 allowed.len(),
5394 1,
5395 "at most one candidate may be offered while resuming: {allowed:?}"
5396 );
5397 assert_eq!(allowed[0].id, "a");
5398 }
5399
5400 #[test]
5401 fn the_gate_offers_nothing_while_resuming_if_no_parked_task_is_runnable() {
5402 let state = Interrupt::Resuming {
5403 parked: vec!["a".to_owned()],
5404 };
5405 let allowed = interrupt_gate(&state, &[], vec![task_with_id("other")]);
5406 assert!(allowed.is_empty());
5407 }
5408
5409 #[test]
5415 fn a_full_sequence_never_gates_two_runs_through_at_once_and_resumes_exactly_one() {
5416 let running = task(); let marked = interrupt_task("marked");
5418
5419 let mut state = Interrupt::Idle;
5420 let in_flight = vec![running.id.clone()];
5422 state = advance_interrupt_tick(true, state, &in_flight, std::slice::from_ref(&marked));
5423 let gated = interrupt_gate(&state, &in_flight, vec![marked.clone(), running.clone()]);
5424 assert!(gated.is_empty(), "still waiting on `running` to park");
5425
5426 state = advance_interrupt_tick(true, state, &[], &[marked.clone(), running.clone()]);
5428 let gated = interrupt_gate(&state, &[], vec![marked.clone(), running.clone()]);
5429 assert_eq!(
5430 gated.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5431 vec!["marked"],
5432 "only the interrupt task may be offered to the dispatcher now"
5433 );
5434
5435 state = advance_interrupt_tick(
5437 true,
5438 state,
5439 &["marked".to_owned()],
5440 std::slice::from_ref(&running),
5441 );
5442 let gated = interrupt_gate(
5443 &state,
5444 &["marked".to_owned()],
5445 vec![marked.clone(), running.clone()],
5446 );
5447 assert!(
5448 gated.is_empty(),
5449 "the parked run must not be offered back while the interrupt \
5450 task is still running"
5451 );
5452
5453 let other = task_with_id("other");
5457 state = advance_interrupt_tick(true, state, &[], &[running.clone(), other.clone()]);
5458 assert_eq!(
5459 state,
5460 Interrupt::Resuming {
5461 parked: vec![running.id.clone()]
5462 }
5463 );
5464 let gated = interrupt_gate(&state, &[], vec![other.clone(), running.clone()]);
5465 assert_eq!(
5466 gated.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5467 vec![running.id.as_str()],
5468 "exactly the parked run resumes - not the unrelated task, even \
5469 though it was offered first"
5470 );
5471
5472 state = advance_interrupt_tick(
5476 true,
5477 state,
5478 std::slice::from_ref(&running.id),
5479 std::slice::from_ref(&other),
5480 );
5481 assert_eq!(state, Interrupt::Idle);
5482 let gated = interrupt_gate(
5483 &state,
5484 std::slice::from_ref(&running.id),
5485 vec![other.clone()],
5486 );
5487 assert_eq!(
5488 gated.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5489 vec![other.id.as_str()],
5490 "ordinary dispatch is unrestricted again"
5491 );
5492 }
5493
5494 #[test]
5495 fn every_run_status_settles_the_task_it_came_from() {
5496 let table = [
5498 (RunStatus::Merged, TaskStatus::Done, 1),
5499 (RunStatus::Ready, TaskStatus::Done, 1),
5500 (RunStatus::Stalled, TaskStatus::Failed, 0),
5501 (RunStatus::Blocked, TaskStatus::Failed, 1),
5502 (RunStatus::Failed, TaskStatus::Failed, 1),
5503 (RunStatus::VerifiedNoop, TaskStatus::Held, 1),
5504 (RunStatus::Prep, TaskStatus::Failed, 1),
5505 (RunStatus::Implementing, TaskStatus::Failed, 1),
5506 (RunStatus::Judging, TaskStatus::Failed, 1),
5507 (RunStatus::Deliberating, TaskStatus::Failed, 1),
5508 (RunStatus::Voting, TaskStatus::Failed, 1),
5509 (RunStatus::Reviewing, TaskStatus::Failed, 1),
5510 (RunStatus::Gating, TaskStatus::Failed, 1),
5511 ];
5512 for (run, want, attempts) in table {
5513 let mut t = task();
5514 t.start("20260902-000000-aaaa".to_owned());
5515 settle(
5516 &mut t,
5517 Verdict {
5518 status: run,
5519 left_pr: false,
5520 parked: false,
5521 quota_hit: matches!(run, RunStatus::Stalled),
5522 no_viable_candidates: false,
5523 },
5524 "why",
5525 2,
5526 );
5527 assert_eq!(t.status, want, "task status after {}", label(run));
5528 assert_eq!(t.attempts, attempts, "attempts after {}", label(run));
5529 }
5530 }
5531
5532 #[test]
5533 fn a_quota_stall_costs_the_task_no_attempt_but_a_block_does() {
5534 let mut stalled = task();
5535 stalled.start("20260902-000000-aaaa".to_owned());
5536 settle(
5537 &mut stalled,
5538 Verdict {
5539 status: RunStatus::Stalled,
5540 left_pr: false,
5541 parked: false,
5542 quota_hit: true,
5543 no_viable_candidates: false,
5544 },
5545 "quota",
5546 1,
5547 );
5548 assert_eq!(stalled.attempts, 0);
5549 assert!(
5550 stalled.status.runnable(),
5551 "a machine problem must leave the task in line"
5552 );
5553
5554 let mut blocked = task();
5555 blocked.start("20260902-000000-aaaa".to_owned());
5556 settle(
5557 &mut blocked,
5558 Verdict {
5559 status: RunStatus::Blocked,
5560 left_pr: false,
5561 parked: false,
5562 quota_hit: false,
5563 no_viable_candidates: false,
5564 },
5565 "findings open",
5566 1,
5567 );
5568 assert_eq!(blocked.attempts, 1);
5569 assert_eq!(
5570 blocked.status,
5571 TaskStatus::Held,
5572 "the last attempt hands the task to a human"
5573 );
5574 }
5575
5576 #[test]
5577 fn a_run_that_opened_a_pull_request_is_never_re_competed() {
5578 let mut delivered = task();
5581 delivered.start("20260903-080619-01c2".to_owned());
5582 settle(
5583 &mut delivered,
5584 Verdict {
5585 status: RunStatus::Blocked,
5586 left_pr: true,
5587 parked: false,
5588 quota_hit: false,
5589 no_viable_candidates: false,
5590 },
5591 "no check status",
5592 4,
5593 );
5594 assert_eq!(
5595 delivered.status,
5596 TaskStatus::Held,
5597 "a pull request waiting on CI or a person is not a retryable failure"
5598 );
5599 assert!(
5600 !delivered.status.runnable(),
5601 "the loop must not pick this task up again"
5602 );
5603 assert_eq!(
5604 delivered.last_error.as_deref(),
5605 Some("no check status"),
5606 "the operator needs to be told what the gate was waiting for"
5607 );
5608
5609 let mut empty_handed = task();
5612 empty_handed.start("20260903-080619-01c2".to_owned());
5613 settle(
5614 &mut empty_handed,
5615 Verdict {
5616 status: RunStatus::Blocked,
5617 left_pr: false,
5618 parked: false,
5619 quota_hit: false,
5620 no_viable_candidates: false,
5621 },
5622 "findings open",
5623 4,
5624 );
5625 assert_eq!(empty_handed.status, TaskStatus::Failed);
5626 assert!(empty_handed.status.runnable());
5627 }
5628
5629 #[test]
5630 fn a_verified_noop_run_hands_off_rather_than_closing_or_auto_retrying() {
5631 let mut noop = task();
5637 noop.start("20260912-131304-391f".to_owned());
5638 settle(
5639 &mut noop,
5640 Verdict {
5641 status: RunStatus::VerifiedNoop,
5642 left_pr: false,
5643 parked: false,
5644 quota_hit: false,
5645 no_viable_candidates: true,
5646 },
5647 "candidate A: already fixed by b32cfc4, on main",
5648 4,
5649 );
5650 assert_eq!(
5651 noop.status,
5652 TaskStatus::Held,
5653 "an unverified claim is a request for a human, not a failure"
5654 );
5655 assert!(
5656 !noop.status.runnable(),
5657 "the loop must not requeue this on the same unverified claim"
5658 );
5659 assert_eq!(noop.attempts, 1);
5664 }
5665
5666 #[test]
5667 fn parking_costs_the_task_no_attempt_and_leaves_it_in_line() {
5668 let mut parked = task();
5673 parked.start("20260903-183634-2d98".to_owned());
5674 settle(
5675 &mut parked,
5676 Verdict {
5677 status: RunStatus::Implementing,
5678 left_pr: false,
5679 quota_hit: false,
5680 parked: true,
5681 no_viable_candidates: false,
5682 },
5683 "parked after `implementing`",
5684 2,
5685 );
5686 assert_eq!(parked.attempts, 0, "a park is refunded");
5687 assert!(
5688 parked.status.runnable(),
5689 "and the task stays in line so the next loop resumes its run"
5690 );
5691 assert_eq!(parked.status, TaskStatus::Parked);
5692 assert_eq!(
5693 parked.park_reason.as_deref(),
5694 Some("parked after `implementing`"),
5695 "the card says where it stopped"
5696 );
5697 assert_eq!(parked.last_error, None, "nothing failed");
5698
5699 let mut broken = task();
5703 broken.start("20260903-183634-2d98".to_owned());
5704 settle(
5705 &mut broken,
5706 Verdict {
5707 status: RunStatus::Implementing,
5708 left_pr: false,
5709 quota_hit: false,
5710 parked: false,
5711 no_viable_candidates: false,
5712 },
5713 "returned mid-flight",
5714 2,
5715 );
5716 assert_eq!(broken.attempts, 1);
5717 }
5718
5719 #[test]
5720 fn only_a_rate_limit_buys_the_task_its_attempt_back() {
5721 let mut flaky = task();
5726 flaky.start("20260903-123023-e633".to_owned());
5727 settle(
5728 &mut flaky,
5729 Verdict {
5730 status: RunStatus::Stalled,
5731 left_pr: false,
5732 parked: false,
5733 quota_hit: false,
5734 no_viable_candidates: false,
5735 },
5736 "verdict rests on 1 of 3 judges",
5737 2,
5738 );
5739 assert_eq!(
5740 flaky.attempts, 1,
5741 "flakiness spends an attempt, so `max_attempts` still bounds it"
5742 );
5743 assert!(flaky.status.runnable(), "and it is still worth retrying");
5744
5745 let mut limited = task();
5747 limited.start("20260903-123023-e633".to_owned());
5748 settle(
5749 &mut limited,
5750 Verdict {
5751 status: RunStatus::Stalled,
5752 left_pr: false,
5753 parked: false,
5754 quota_hit: true,
5755 no_viable_candidates: false,
5756 },
5757 "judge-2, judge-3 out of quota",
5758 2,
5759 );
5760 assert_eq!(limited.attempts, 0, "a quota window is refunded");
5761 assert!(limited.status.runnable());
5762
5763 let mut worn = task();
5766 for _ in 0..2 {
5767 worn.release();
5768 }
5769 worn.start("20260903-123023-e633".to_owned());
5770 worn.attempts = 2;
5771 settle(
5772 &mut worn,
5773 Verdict {
5774 status: RunStatus::Stalled,
5775 left_pr: false,
5776 parked: false,
5777 quota_hit: false,
5778 no_viable_candidates: false,
5779 },
5780 "no quorum again",
5781 2,
5782 );
5783 assert_eq!(worn.status, TaskStatus::Held);
5784 assert!(!worn.status.runnable());
5785 }
5786
5787 #[test]
5788 fn a_quota_wipeout_that_leaves_nothing_to_judge_also_costs_no_attempt() {
5789 let mut wiped_out = task();
5796 wiped_out.start("20260907-025000-a1b2".to_owned());
5797 settle(
5798 &mut wiped_out,
5799 Verdict {
5800 status: RunStatus::Failed,
5801 left_pr: false,
5802 parked: false,
5803 quota_hit: true,
5804 no_viable_candidates: true,
5805 },
5806 "no candidate produced a change; nothing to judge",
5807 2,
5808 );
5809 assert_eq!(wiped_out.attempts, 0, "a total quota wipeout is refunded");
5810 assert!(
5811 wiped_out.status.runnable(),
5812 "a machine problem must leave the task in line"
5813 );
5814
5815 let mut partial_progress = task();
5821 partial_progress.start("20260907-025500-c3d4".to_owned());
5822 settle(
5823 &mut partial_progress,
5824 Verdict {
5825 status: RunStatus::Failed,
5826 left_pr: false,
5827 parked: false,
5828 quota_hit: true,
5829 no_viable_candidates: false,
5830 },
5831 "gate failed on the winning candidate",
5832 2,
5833 );
5834 assert_eq!(
5835 partial_progress.attempts, 1,
5836 "a candidate that actually produced a change spends the attempt \
5837 even though some other seat hit its quota"
5838 );
5839 assert!(partial_progress.status.runnable());
5840 }
5841
5842 #[test]
5843 fn reclaim_refunds_a_recovered_quota_wipeout_the_same_way_a_live_settle_does() {
5844 let mut t = task();
5850 t.start("20260907-025000-a1b2".to_owned());
5851 let mut state = run_state(RunStatus::Failed);
5852 state.quota.push(QuotaLoss {
5853 seat: "cand-a".to_owned(),
5854 node: "implement".to_owned(),
5855 at: Timestamp::now(),
5856 reset: None,
5857 });
5858 assert!(
5859 state.viable().is_empty(),
5860 "no candidate was added, so nothing is viable"
5861 );
5862 reclaim(&mut t, Some(state), 2, "en");
5863 assert_eq!(t.attempts, 0, "a recovered quota wipeout is refunded");
5864 assert!(t.status.runnable());
5865 }
5866
5867 #[test]
5868 fn a_held_task_is_never_offered_to_the_loop() {
5869 let dir = tempfile::tempdir().unwrap();
5870 let queue = Queue::at(dir.path().to_path_buf());
5871 for (n, priority) in [(1, 0), (2, 5), (3, 5)] {
5872 let mut t = task();
5873 t.id = format!("2026090{n}-000000-000{n}");
5874 t.priority = priority;
5875 queue.put(&mut t).unwrap();
5876 }
5877 let mut held = task();
5878 held.id = "20260909-000000-9999".to_owned();
5879 held.priority = 99;
5880 held.hold_machine(None);
5881 queue.put(&mut held).unwrap();
5882
5883 let order: Vec<String> = runnable(&queue).into_iter().map(|t| t.id).collect();
5884 assert_eq!(order.len(), 3);
5885 assert!(!order.contains(&held.id));
5886 assert_eq!(
5887 order.first().cloned(),
5888 queue.next_runnable().map(|t| t.id),
5889 "the loop's first candidate is exactly what the queue offers"
5890 );
5891 assert_eq!(
5892 order,
5893 vec![
5894 "20260902-000000-0002".to_owned(),
5895 "20260903-000000-0003".to_owned(),
5896 "20260901-000000-0001".to_owned(),
5897 ],
5898 "priority first, then oldest, so nothing starves"
5899 );
5900 }
5901
5902 #[test]
5903 fn sweep_removes_an_old_unparseable_lock_and_keeps_a_live_one() {
5904 let dir = tempfile::tempdir().unwrap();
5905 let queue = Queue::at(dir.path().to_path_buf());
5906 let mut old = task();
5907 old.id = "20260101-000000-old0".to_owned();
5908 queue.put(&mut old).unwrap();
5909 let mut fresh = task();
5910 fresh.id = "20260101-000000-new0".to_owned();
5911 queue.put(&mut fresh).unwrap();
5912
5913 std::fs::write(dir.path().join(format!("{}.lock", old.id)), "not a pid").unwrap();
5917 std::thread::sleep(Duration::from_millis(60));
5918 let live = queue.claim(&fresh.id).unwrap();
5919
5920 let swept = sweep_stale_claims(&queue, Duration::from_millis(50));
5921 assert_eq!(swept, vec![old.id.clone()]);
5922 assert!(
5923 queue.claim(&old.id).is_ok(),
5924 "an unparseable lock older than the threshold is swept"
5925 );
5926 assert!(
5927 queue.claim(&fresh.id).is_err(),
5928 "a live pid protects its lock regardless of age"
5929 );
5930 drop(live);
5931 }
5932
5933 #[test]
5934 fn an_old_lock_whose_pid_is_still_alive_is_never_swept_by_age_alone() {
5935 let dir = tempfile::tempdir().unwrap();
5945 let queue = Queue::at(dir.path().to_path_buf());
5946 let mut t = task();
5947 t.id = "20260101-000000-live".to_owned();
5948 queue.put(&mut t).unwrap();
5949
5950 let claim = queue.claim(&t.id).unwrap();
5951 std::thread::sleep(Duration::from_millis(60));
5952
5953 let swept = sweep_stale_claims(&queue, Duration::from_millis(50));
5954 assert!(
5955 swept.is_empty(),
5956 "a lock naming a live pid must never be swept by age, no matter how old: {swept:?}"
5957 );
5958 assert!(
5959 queue.claim(&t.id).is_err(),
5960 "the lock still protects its task"
5961 );
5962 drop(claim);
5963 }
5964
5965 fn injected_dead_pid() -> u32 {
5968 std::process::id().checked_add(1).unwrap_or(1)
5969 }
5970
5971 #[test]
5972 fn a_lock_naming_a_dead_pid_is_swept_at_once_regardless_of_age() {
5973 let dir = tempfile::tempdir().unwrap();
5974 let queue = Queue::at(dir.path().to_path_buf());
5975 let mut t = task();
5976 t.id = "20260101-000000-dead".to_owned();
5977 queue.put(&mut t).unwrap();
5978 let dead_pid = injected_dead_pid();
5979
5980 std::fs::write(
5985 dir.path().join(format!("{}.lock", t.id)),
5986 dead_pid.to_string(),
5987 )
5988 .unwrap();
5989
5990 let swept = sweep_stale_claims_with(&queue, Duration::from_secs(6 * 60 * 60), |pid| {
5991 pid != dead_pid
5992 });
5993 assert_eq!(
5994 swept,
5995 vec![t.id.clone()],
5996 "a dead owner is reclaimed immediately, not after STALE_CLAIM"
5997 );
5998 assert!(queue.claim(&t.id).is_ok(), "the task is claimable again");
5999 }
6000
6001 #[test]
6002 fn sweeping_on_every_poll_catches_a_lock_that_appears_after_the_first_sweep() {
6003 let dir = tempfile::tempdir().unwrap();
6004 let queue = Queue::at(dir.path().to_path_buf());
6005 let mut t = task();
6006 t.id = "20260101-000000-late".to_owned();
6007 queue.put(&mut t).unwrap();
6008 let dead_pid = injected_dead_pid();
6009
6010 assert!(
6013 sweep_stale_claims(&queue, Duration::from_secs(6 * 60 * 60)).is_empty(),
6014 "nothing has claimed the task yet"
6015 );
6016
6017 std::fs::write(
6020 dir.path().join(format!("{}.lock", t.id)),
6021 dead_pid.to_string(),
6022 )
6023 .unwrap();
6024
6025 let swept = sweep_stale_claims_with(&queue, Duration::from_secs(6 * 60 * 60), |pid| {
6029 pid != dead_pid
6030 });
6031 assert_eq!(swept, vec![t.id.clone()]);
6032 }
6033
6034 #[test]
6035 fn a_running_task_behind_a_dead_daemons_lock_recovers_once_swept_and_keeps_its_history() {
6036 crate::run::pin_test_home();
6041 let dir = tempfile::tempdir().unwrap();
6042 let queue = Queue::at(dir.path().to_path_buf());
6043 let mut t = task();
6044 t.id = "20260101-000000-crsh".to_owned();
6045 t.status = TaskStatus::Running;
6046 t.attempts = 1;
6047 t.runs.push("20260904-000000-4043".to_owned());
6051 queue.put(&mut t).unwrap();
6052 let dead_pid = injected_dead_pid();
6053
6054 std::fs::write(
6057 dir.path().join(format!("{}.lock", t.id)),
6058 dead_pid.to_string(),
6059 )
6060 .unwrap();
6061
6062 assert!(reclaim_orphaned_running(&queue, 2).is_empty());
6068 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Running);
6069
6070 let swept = sweep_stale_claims_with(&queue, Duration::from_secs(6 * 60 * 60), |pid| {
6071 pid != dead_pid
6072 });
6073 assert_eq!(swept, vec![t.id.clone()]);
6074
6075 let reclaimed = reclaim_orphaned_running(&queue, 2);
6076 assert_eq!(reclaimed, vec![t.id.clone()]);
6077 let after = queue.get(&t.id).unwrap();
6078 assert_eq!(
6079 after.status,
6080 TaskStatus::Held,
6081 "no run.json to recover from, so a human is asked"
6082 );
6083 assert_eq!(
6084 after.runs,
6085 vec!["20260904-000000-4043".to_owned()],
6086 "the crashed run's id is kept as evidence, not discarded"
6087 );
6088 }
6089
6090 #[test]
6091 fn a_lock_is_kept_when_the_process_query_is_unavailable() {
6092 let dir = tempfile::tempdir().unwrap();
6093 let queue = Queue::at(dir.path().to_path_buf());
6094 let mut t = task();
6095 t.id = "20260101-000000-unknown".to_owned();
6096 queue.put(&mut t).unwrap();
6097 let dead_pid = injected_dead_pid();
6098 std::fs::write(
6099 dir.path().join(format!("{}.lock", t.id)),
6100 dead_pid.to_string(),
6101 )
6102 .unwrap();
6103
6104 let swept = sweep_stale_claims_with(&queue, Duration::ZERO, |_| true);
6105 assert!(swept.is_empty(), "an unknown pid must keep its lock");
6106 assert!(queue.claim(&t.id).is_err(), "the lock remains protective");
6107 }
6108
6109 fn run_state_in(status: RunStatus, language: &str) -> RunState {
6110 let mut s = run_state(status);
6111 s.config.graph.language = language.to_owned();
6112 s
6113 }
6114
6115 fn unstarted_verdict(status: RunStatus) -> Verdict {
6116 Verdict {
6117 status,
6118 left_pr: false,
6119 quota_hit: false,
6120 parked: false,
6121 no_viable_candidates: false,
6122 }
6123 }
6124
6125 #[test]
6126 fn an_already_in_base_run_finishes_the_task_without_spending_an_attempt() {
6127 let mut t = task();
6128 t.attempts = 1;
6129 settle_in(
6130 &mut t,
6131 unstarted_verdict(RunStatus::AlreadyInBase),
6132 "already in main",
6133 1,
6134 phrases("en"),
6135 );
6136 assert_eq!(t.status, TaskStatus::Done);
6137 assert_eq!(t.attempts, 0);
6138 }
6139
6140 #[test]
6141 fn settle_renders_the_non_terminal_reason_in_the_configured_language() {
6142 let reason = |language: &str| {
6143 let mut t = task();
6144 settle_in(
6145 &mut t,
6146 unstarted_verdict(RunStatus::Judging),
6147 "boom",
6148 1,
6149 phrases(language),
6150 );
6151 t.last_error.or(t.hold_reason).unwrap_or_default()
6152 };
6153 assert!(
6154 reason("en").starts_with("the graph stopped at `"),
6155 "{}",
6156 reason("en")
6157 );
6158 assert!(reason("ja").starts_with("グラフが終端状態に達しないまま"));
6159 assert!(reason("日本語").contains("boom"));
6160 assert_eq!(reason("fr"), reason("en"));
6161 }
6162
6163 #[test]
6164 fn describe_follows_the_run_language_and_keeps_the_run_id() {
6165 let en = describe(&run_state_in(RunStatus::Stalled, "en"));
6166 assert!(en.starts_with("the judging panel lost its quorum"), "{en}");
6167 let ja = describe(&run_state_in(RunStatus::Stalled, "ja"));
6168 assert!(ja.starts_with("審査パネルが定足数を失いました"), "{ja}");
6169 assert!(ja.contains("[run "), "{ja}");
6170 let ended = describe(&run_state_in(RunStatus::Failed, "jp"));
6171 assert!(ended.starts_with("run 終了: "), "{ended}");
6172 let mut de = run_state_in(RunStatus::Failed, "de");
6173 let mut en = run_state_in(RunStatus::Failed, "en");
6174 de.id = "same".to_owned();
6175 en.id = "same".to_owned();
6176 assert_eq!(describe(&de), describe(&en));
6177 }
6178
6179 #[test]
6180 fn handover_refusals_follow_the_language_and_keep_the_detail_apart() {
6181 use crate::handover::Refused;
6182 let cases = [
6183 Refused::Foreign {
6184 branch: "b".into(),
6185 path: "/w/x".into(),
6186 why: "made by hand".into(),
6187 },
6188 Refused::Unsafe {
6189 branch: "b".into(),
6190 path: "/w/x".into(),
6191 why: "its worktree has uncommitted changes (a.rs)".into(),
6192 },
6193 Refused::ReleaseFailed {
6194 branch: "b".into(),
6195 path: "/w/x".into(),
6196 run: "ab12".into(),
6197 },
6198 ];
6199 for r in &cases {
6200 let en = (phrases("en").handover_refused)(r);
6201 assert_eq!(en, r.to_string());
6202 let ja = (phrases("ja").handover_refused)(r);
6203 assert!(
6204 !ja.contains("is checked out") && !ja.contains("try again"),
6205 "{ja}"
6206 );
6207 assert!(ja.contains("`b`") && ja.contains("/w/x"), "{ja}");
6208 if let Refused::Foreign { why, .. } | Refused::Unsafe { why, .. } = r {
6209 assert!(ja.contains(&format!("(詳細: {why})")), "{ja}");
6210 }
6211 }
6212 assert!(phrases("en").handover_hint.contains("release the task"));
6213 assert!(phrases("ja").handover_hint.contains("解放"));
6214 }
6215
6216 #[test]
6217 fn describe_translates_the_unanswered_reviewer_stop_only_in_ja() {
6218 let build = |lang: &str| {
6219 let mut s = run_state_in(RunStatus::Blocked, lang);
6220 s.id = "same".to_owned();
6221 s.config.graph.review_rounds = 3;
6222 let mut r = review_round(3);
6223 r.expected = 3;
6224 r.answered = 1;
6225 s.reviews.push(r);
6226 s.event(
6227 "review",
6228 "2 reviewer seat(s) never answered after 3 rounds; refusing to call it clean",
6229 );
6230 s
6231 };
6232 let en = describe(&build("en"));
6233 assert!(en.contains("(review: 2 reviewer seat(s) never answered after 3 rounds; refusing to call it clean)"), "{en}");
6234 let ja = describe(&build("ja"));
6235 assert!(ja.contains("2 席のレビュアーが 3 ラウンド"), "{ja}");
6236 assert!(!ja.contains("never answered"), "{ja}");
6237 assert!(ja.contains("[run "), "{ja}");
6238
6239 let mut failed = build("ja");
6242 failed.reviews[0].e2e.push(crate::run::CommandOutcome {
6243 command: "cargo test".to_owned(),
6244 code: Some(1),
6245 output_tail: String::new(),
6246 duration_ms: 0,
6247 resource_blocked: false,
6248 });
6249 failed.event("review", "stopped; e2e failed: cargo test");
6250 let ja = describe(&failed);
6251 assert!(ja.contains("stopped; e2e failed: cargo test"), "{ja}");
6252 assert!(!ja.contains("席のレビュアー"), "{ja}");
6253 }
6254
6255 #[test]
6256 fn refusals_and_recovery_prose_follow_the_language() {
6257 let t = held_task_with("r1");
6258 let q = action_question(
6259 "r1",
6260 ask::ChoiceAction::Resume {
6261 run: "r1".to_owned(),
6262 },
6263 );
6264 let refuse = |p: &Phrases| match decide_action(&t, &q, p, |_: &str| bail!("gone")) {
6265 ActionDecision::Refuse(s) => s,
6266 other => panic!("{other:?}"),
6267 };
6268 assert!(refuse(phrases("en")).contains("could not be read: gone"));
6269 assert!(refuse(phrases("ja")).contains("読み込めませんでした: gone"));
6270 assert_eq!(refuse(phrases("fr")), refuse(phrases("en")));
6271
6272 let mut held = task();
6273 reclaim(&mut held, None, 2, "ja");
6274 assert!(held.hold_reason.unwrap().contains("保留にしました"));
6275 let mut held = task();
6276 reclaim(&mut held, None, 2, "xx");
6277 assert!(held.hold_reason.unwrap().contains("held for a human"));
6278 }
6279
6280 fn run_state(status: RunStatus) -> RunState {
6281 let mut state = RunState::new(
6282 PathBuf::from("/repo"),
6283 "main".to_owned(),
6284 "abc1234def".to_owned(),
6285 "add retries".to_owned(),
6286 Config::default(),
6287 );
6288 state.status = status;
6289 state
6290 }
6291
6292 fn candidate(label: char, summary: &str, empty: bool, failed: Option<&str>) -> Candidate {
6293 Candidate {
6294 index: 0,
6295 label,
6296 agent: "claude".to_owned(),
6297 branch: format!("magi/x/{label}"),
6298 worktree: PathBuf::from("/repo"),
6299 summary: summary.to_owned(),
6300 stat: String::new(),
6301 files: 0,
6302 commits: usize::from(!empty),
6303 empty,
6304 failed: failed.map(str::to_owned),
6305 verified_noop: None,
6306 duration_ms: 0,
6307 folded: false,
6308 }
6309 }
6310
6311 #[test]
6312 fn diagnostic_names_the_failing_gate_checks_and_their_output() {
6313 let mut state = run_state(RunStatus::Blocked);
6314 state.gate = vec![
6315 CommandOutcome {
6316 command: "cargo make check".to_owned(),
6317 code: Some(0),
6318 output_tail: "ok".to_owned(),
6319 duration_ms: 0,
6320 resource_blocked: false,
6321 },
6322 CommandOutcome {
6323 command: "cargo test".to_owned(),
6324 code: Some(101),
6325 output_tail: "thread 'x' panicked: assertion failed".to_owned(),
6326 duration_ms: 0,
6327 resource_blocked: false,
6328 },
6329 ];
6330 let d = diagnostic(&state).expect("a failing gate must produce a diagnostic");
6331 assert!(d.contains("cargo test"), "{d}");
6332 assert!(
6333 !d.contains("cargo make check"),
6334 "a passing check is not a diagnostic: {d}"
6335 );
6336 assert!(d.contains("assertion failed"), "{d}");
6337 }
6338
6339 #[test]
6340 fn diagnostic_names_the_checks_the_fixer_gave_up_in_front_of() {
6341 let mut state = run_state(RunStatus::Blocked);
6342 state.event(
6343 "land",
6344 "stopped: the fixer produced no commit while 2 check(s) were failing \
6345 (build, lint); stopping instead of looping on an unchanged tree",
6346 );
6347 let d = diagnostic(&state).expect("a stalled land loop must produce a diagnostic");
6348 assert!(d.contains("build"), "{d}");
6349 assert!(d.contains("lint"), "{d}");
6350 assert!(d.contains("fixer produced no commit"), "{d}");
6351 }
6352
6353 #[test]
6354 fn describe_never_leaves_a_verified_noop_reading_as_a_bare_status_code() {
6355 let state = run_state(RunStatus::VerifiedNoop);
6360 let d = describe(&state);
6361 assert!(
6362 d.contains("agent-verified no-op"),
6363 "expected the display label, not the wire spelling: {d}"
6364 );
6365 assert!(!d.contains("verified_noop"), "{d}");
6366 }
6367
6368 #[test]
6369 fn diagnostic_carries_a_candidates_own_final_word_when_none_was_viable() {
6370 let mut state = run_state(RunStatus::Failed);
6376 state.candidates = vec![candidate(
6377 'A',
6378 "opened pull request #42, merged it, tagged v1.2.3 and published the release",
6379 true,
6380 None,
6381 )];
6382 let d = diagnostic(&state).expect("an empty candidate with a summary must be surfaced");
6383 assert!(d.contains("candidate A"), "{d}");
6384 assert!(d.contains("tagged v1.2.3"), "{d}");
6385 }
6386
6387 #[test]
6388 fn diagnostic_falls_back_to_a_candidates_failure_reason_when_it_has_no_summary() {
6389 let mut state = run_state(RunStatus::Failed);
6390 state.candidates = vec![candidate('A', "", true, Some("agent timed out"))];
6391 let d = diagnostic(&state).expect("a candidate's own failure reason must be surfaced");
6392 assert!(d.contains("candidate A"), "{d}");
6393 assert!(d.contains("agent timed out"), "{d}");
6394 }
6395
6396 #[test]
6397 fn diagnostic_is_none_when_nothing_recognisable_explains_the_hold() {
6398 let mut state = run_state(RunStatus::Failed);
6401 state.candidates = vec![candidate('A', "did the work", false, None)];
6402 assert!(diagnostic(&state).is_none());
6403 }
6404
6405 #[test]
6406 fn diagnostic_is_bounded_however_much_a_run_printed() {
6407 let mut state = run_state(RunStatus::Blocked);
6408 state.gate = vec![
6409 CommandOutcome {
6410 command: "cargo test".to_owned(),
6411 code: Some(101),
6412 output_tail: "x".repeat(50_000),
6413 duration_ms: 0,
6414 resource_blocked: false,
6415 },
6416 CommandOutcome {
6417 command: "cargo clippy".to_owned(),
6418 code: Some(1),
6419 output_tail: "y".repeat(50_000),
6420 duration_ms: 0,
6421 resource_blocked: false,
6422 },
6423 ];
6424 state.candidates = vec![
6425 candidate('A', &"z".repeat(50_000), true, None),
6426 candidate('B', &"w".repeat(50_000), true, None),
6427 ];
6428 let d = diagnostic(&state).expect("plenty here to diagnose");
6429 assert!(
6430 d.len() <= DIAGNOSTIC_MAX,
6431 "diagnostic grew to {} bytes, unbounded",
6432 d.len()
6433 );
6434 }
6435
6436 #[test]
6437 fn settle_and_diagnose_attaches_a_diagnostic_only_once_the_task_is_held() {
6438 let mut state = run_state(RunStatus::Blocked);
6439 state.gate = vec![CommandOutcome {
6440 command: "cargo test".to_owned(),
6441 code: Some(101),
6442 output_tail: "assertion failed".to_owned(),
6443 duration_ms: 0,
6444 resource_blocked: false,
6445 }];
6446 let verdict = Verdict {
6447 status: RunStatus::Blocked,
6448 left_pr: false,
6449 quota_hit: false,
6450 parked: false,
6451 no_viable_candidates: false,
6452 };
6453
6454 let mut t = task();
6457 t.start("run-1".to_owned());
6458 settle_and_diagnose(&mut t, verdict, "gate failed", 2, &state);
6459 assert_eq!(t.status, TaskStatus::Failed);
6460 assert!(t.diagnostic.is_none());
6461
6462 t.start("run-2".to_owned());
6465 settle_and_diagnose(&mut t, verdict, "gate failed", 2, &state);
6466 assert_eq!(t.status, TaskStatus::Held);
6467 let d = t.diagnostic.expect("a held task must carry its diagnostic");
6468 assert!(d.contains("cargo test"), "{d}");
6469 }
6470
6471 #[test]
6472 fn a_held_task_names_the_open_question_it_is_waiting_on() {
6473 crate::run::pin_test_home();
6478 let home = crate::run::home();
6479 let state = run_state(RunStatus::VerifiedNoop);
6480 let mut q = ask::Question::new(
6481 state.id.clone(),
6482 "implement".to_owned(),
6483 "impl-A".to_owned(),
6484 "is this really a no-op?".to_owned(),
6485 String::new(),
6486 Vec::new(),
6487 );
6488 Questions::at(home.join("questions")).put(&mut q).unwrap();
6489
6490 let verdict = Verdict {
6491 status: RunStatus::VerifiedNoop,
6492 left_pr: false,
6493 quota_hit: false,
6494 parked: false,
6495 no_viable_candidates: false,
6496 };
6497 let mut t = task();
6498 t.start(state.id.clone());
6499 settle_and_diagnose(&mut t, verdict, "run ended agent-verified no-op", 5, &state);
6500
6501 assert_eq!(t.status, TaskStatus::Held);
6502 let reason = t.hold_reason.expect("a held task must record why");
6503 assert!(
6504 reason.starts_with("run ended agent-verified no-op"),
6505 "the original settle reason must survive unchanged: {reason}"
6506 );
6507 assert!(
6508 reason.contains(q.short()),
6509 "the open question's id must be named so the notice is actionable: {reason}"
6510 );
6511 }
6512
6513 #[test]
6514 fn a_held_task_with_no_open_question_keeps_its_plain_reason() {
6515 crate::run::pin_test_home();
6516 let state = run_state(RunStatus::VerifiedNoop);
6517
6518 let verdict = Verdict {
6519 status: RunStatus::VerifiedNoop,
6520 left_pr: false,
6521 quota_hit: false,
6522 parked: false,
6523 no_viable_candidates: false,
6524 };
6525 let mut t = task();
6526 t.start(state.id.clone());
6527 settle_and_diagnose(&mut t, verdict, "run ended agent-verified no-op", 5, &state);
6528
6529 assert_eq!(t.status, TaskStatus::Held);
6530 assert_eq!(
6531 t.hold_reason.as_deref(),
6532 Some("run ended agent-verified no-op"),
6533 "nothing to append when the question was already answered or never asked"
6534 );
6535 }
6536
6537 #[test]
6538 fn supersede_prior_runs_rewrites_an_earlier_blocked_attempt_once_a_later_one_lands() {
6539 crate::run::pin_test_home();
6540 let mut first = run_state(RunStatus::Blocked);
6541 first.id = "20260101-000000-sup1".to_owned();
6542 first.save().unwrap();
6543 let mut second = run_state(RunStatus::Merged);
6544 second.id = "20260101-000000-sup2".to_owned();
6545 second.save().unwrap();
6546
6547 let mut t = task();
6548 t.runs = vec![first.id.clone(), second.id.clone()];
6549 t.status = TaskStatus::Done;
6550
6551 supersede_prior_runs(&t, &crate::run::home());
6552
6553 assert_eq!(
6554 RunState::load(&first.id).unwrap().status,
6555 RunStatus::Superseded,
6556 "the first attempt's Blocked no longer needs anyone's attention"
6557 );
6558 assert_eq!(
6559 RunState::load(&second.id).unwrap().status,
6560 RunStatus::Merged,
6561 "the run that actually succeeded is left exactly as it was"
6562 );
6563 }
6564
6565 #[test]
6566 fn supersede_prior_runs_leaves_a_manually_resumed_attempt_alone() {
6567 crate::run::pin_test_home();
6573 let mut first = run_state(RunStatus::Blocked);
6574 first.id = "20260101-000000-sup9".to_owned();
6575 first.driver_pid = Some(std::process::id());
6578 first.driver_started_at = Some(
6579 crate::proc::process_started_at(std::process::id())
6580 .expect("this test process's own start time must be queryable"),
6581 );
6582 first.save().unwrap();
6583 let mut second = run_state(RunStatus::Merged);
6584 second.id = "20260101-000000-supa".to_owned();
6585 second.save().unwrap();
6586
6587 let mut t = task();
6588 t.runs = vec![first.id.clone(), second.id.clone()];
6589 t.status = TaskStatus::Done;
6590
6591 supersede_prior_runs(&t, &crate::run::home());
6592
6593 assert_eq!(
6594 RunState::load(&first.id).unwrap().status,
6595 RunStatus::Blocked,
6596 "a live driver_pid means something is still actually working this run, \
6597 even though no daemon claims it - rewriting under it would just be \
6598 undone the next time that process saves"
6599 );
6600 }
6601
6602 #[test]
6603 fn resweep_catches_up_a_run_left_live_once_its_manual_process_is_no_longer_driving_it() {
6604 let dir = tempfile::tempdir().unwrap();
6610 let home = dir.path().to_path_buf();
6611 let queue = Queue::at(dir.path().join("queue"));
6612
6613 let mut first = run_state(RunStatus::Blocked);
6614 first.id = "20260101-000000-supd".to_owned();
6615 first.driver_pid = Some(std::process::id());
6616 first.driver_started_at = Some(
6617 crate::proc::process_started_at(std::process::id())
6618 .expect("this test process's own start time must be queryable"),
6619 );
6620 first.save_under(&home).unwrap();
6621 let mut second = run_state(RunStatus::Merged);
6622 second.id = "20260101-000000-supe".to_owned();
6623 second.save_under(&home).unwrap();
6624
6625 let mut t = task();
6626 t.runs = vec![first.id.clone(), second.id.clone()];
6627 t.status = TaskStatus::Done;
6628 queue.put(&mut t).unwrap();
6629
6630 resweep_superseded_attempts(&queue, &home);
6631 assert_eq!(
6632 RunState::load_under(&first.id, &home).unwrap().status,
6633 RunStatus::Blocked,
6634 "still live on the first pass, so still untouched"
6635 );
6636
6637 let mut stale = RunState::load_under(&first.id, &home).unwrap();
6643 stale.driver_started_at = Some("1".to_owned());
6644 stale.save_under(&home).unwrap();
6645
6646 resweep_superseded_attempts(&queue, &home);
6647 assert_eq!(
6648 RunState::load_under(&first.id, &home).unwrap().status,
6649 RunStatus::Superseded,
6650 "the second pass catches up what the first one correctly skipped"
6651 );
6652 }
6653
6654 #[test]
6655 fn supersede_prior_runs_leaves_concurrent_blocked_attempts_alone_while_the_task_is_not_done() {
6656 crate::run::pin_test_home();
6657 let mut first = run_state(RunStatus::Blocked);
6658 first.id = "20260101-000000-sup3".to_owned();
6659 first.save().unwrap();
6660 let mut second = run_state(RunStatus::Blocked);
6661 second.id = "20260101-000000-sup4".to_owned();
6662 second.save().unwrap();
6663
6664 let mut t = task();
6665 t.runs = vec![first.id.clone(), second.id.clone()];
6666 t.status = TaskStatus::Failed;
6670
6671 supersede_prior_runs(&t, &crate::run::home());
6672
6673 assert_eq!(
6674 RunState::load(&first.id).unwrap().status,
6675 RunStatus::Blocked
6676 );
6677 assert_eq!(
6678 RunState::load(&second.id).unwrap().status,
6679 RunStatus::Blocked
6680 );
6681 }
6682
6683 #[test]
6684 fn supersede_prior_runs_does_nothing_when_the_task_was_closed_by_hand() {
6685 crate::run::pin_test_home();
6689 let mut first = run_state(RunStatus::Blocked);
6690 first.id = "20260101-000000-sup5".to_owned();
6691 first.save().unwrap();
6692
6693 let mut t = task();
6694 t.runs = vec![first.id.clone()];
6695 t.status = TaskStatus::Done;
6696
6697 supersede_prior_runs(&t, &crate::run::home());
6698
6699 assert_eq!(
6700 RunState::load(&first.id).unwrap().status,
6701 RunStatus::Blocked,
6702 "a single-attempt task has no earlier run to supersede"
6703 );
6704 }
6705
6706 #[test]
6707 fn supersede_prior_runs_does_nothing_when_the_last_recorded_attempt_never_landed() {
6708 crate::run::pin_test_home();
6715 let mut first = run_state(RunStatus::Blocked);
6716 first.id = "20260101-000000-supb".to_owned();
6717 first.save().unwrap();
6718 let mut second = run_state(RunStatus::Failed);
6719 second.id = "20260101-000000-supc".to_owned();
6720 second.save().unwrap();
6721
6722 let mut t = task();
6723 t.runs = vec![first.id.clone(), second.id.clone()];
6724 t.status = TaskStatus::Done;
6725
6726 supersede_prior_runs(&t, &crate::run::home());
6727
6728 assert_eq!(
6729 RunState::load(&first.id).unwrap().status,
6730 RunStatus::Blocked,
6731 "the task's last attempt never landed, so there is nothing here \
6732 actually superseding it"
6733 );
6734 }
6735
6736 #[test]
6737 fn supersede_prior_runs_leaves_a_failed_or_verified_noop_attempt_as_is() {
6738 crate::run::pin_test_home();
6742 let mut failed = run_state(RunStatus::Failed);
6743 failed.id = "20260101-000000-sup6".to_owned();
6744 failed.save().unwrap();
6745 let mut noop = run_state(RunStatus::VerifiedNoop);
6746 noop.id = "20260101-000000-sup7".to_owned();
6747 noop.save().unwrap();
6748 let mut winner = run_state(RunStatus::Ready);
6749 winner.id = "20260101-000000-sup8".to_owned();
6750 winner.save().unwrap();
6751
6752 let mut t = task();
6753 t.runs = vec![failed.id.clone(), noop.id.clone(), winner.id.clone()];
6754 t.status = TaskStatus::Done;
6755
6756 supersede_prior_runs(&t, &crate::run::home());
6757
6758 assert_eq!(
6759 RunState::load(&failed.id).unwrap().status,
6760 RunStatus::Failed
6761 );
6762 assert_eq!(
6763 RunState::load(&noop.id).unwrap().status,
6764 RunStatus::VerifiedNoop
6765 );
6766 }
6767
6768 fn approval_question(run: &str) -> ask::Question {
6769 ask::Question::new(
6770 run.to_owned(),
6771 land::APPROVAL_NODE.to_owned(),
6772 "land".to_owned(),
6773 "merge?".to_owned(),
6774 String::new(),
6775 vec!["merge".to_owned(), "hold".to_owned()],
6776 )
6777 }
6778
6779 #[test]
6780 fn land_resume_state_leaves_a_fresh_open_question_waiting() {
6781 crate::run::pin_test_home();
6782 let mut state = run_state(RunStatus::Landing);
6783 state.id = "20260101-000000-fre1".to_owned();
6784 state.parked = true;
6785 state.save().unwrap();
6786 ask::Questions::open()
6787 .put(&mut approval_question(&state.id))
6788 .unwrap();
6789
6790 let mut t = task();
6791 t.runs.push(state.id.clone());
6792 assert_eq!(
6793 land_resume_state(&t),
6794 LandResume::StillWaiting,
6795 "nobody has answered and the timeout has not passed"
6796 );
6797 }
6798
6799 #[test]
6800 fn land_resume_state_waits_on_a_withheld_text_question_then_resumes() {
6801 crate::run::pin_test_home();
6802 let mut state = run_state(RunStatus::Gating);
6803 state.id = "20260101-000000-gt01".to_owned();
6804 state.parked = true;
6805 state.config.graph.answer_timeout = 60;
6806 state.github_text = Some(crate::run::GithubTextAsk {
6807 rewrite: None,
6808 fingerprint: "f".into(),
6809 title: true,
6810 body: false,
6811 categories: vec!["title-language".into()],
6812 question: None,
6813 asks: 1,
6814 resolved: false,
6815 chosen_title: None,
6816 choice: None,
6817 });
6818 state.save().unwrap();
6819 let store = ask::Questions::open();
6820 let mut q = ask::Question::new(
6821 state.id.clone(),
6822 crate::github_text::ASK_NODE.to_owned(),
6823 crate::github_text::ASK_SEAT.to_owned(),
6824 "s".into(),
6825 String::new(),
6826 vec!["use fallback".into()],
6827 );
6828 store.put(&mut q).unwrap();
6829 let mut t = task();
6830 t.runs.push(state.id.clone());
6831 assert_eq!(land_resume_state(&t), LandResume::StillWaiting);
6832 q.asked_at = Timestamp::now() - jiff::SignedDuration::from_secs(120);
6833 store.put(&mut q).unwrap();
6834 assert_eq!(land_resume_state(&t), LandResume::Ready);
6835 assert!(!store.get(&q.id).unwrap().status.open());
6836 state.github_text.as_mut().unwrap().resolved = true;
6838 state.save().unwrap();
6839 assert_eq!(land_resume_state(&t), LandResume::NotLanding);
6840 }
6841
6842 #[test]
6843 fn land_resume_state_abandons_a_question_that_outlived_answer_timeout() {
6844 crate::run::pin_test_home();
6849 let mut state = run_state(RunStatus::Landing);
6850 state.id = "20260101-000000-exp1".to_owned();
6851 state.parked = true;
6852 state.config.graph.answer_timeout = 60;
6853 state.save().unwrap();
6854
6855 let store = ask::Questions::open();
6856 let mut q = approval_question(&state.id);
6857 q.asked_at = Timestamp::now() - jiff::SignedDuration::from_secs(120);
6858 store.put(&mut q).unwrap();
6859
6860 let mut t = task();
6861 t.runs.push(state.id.clone());
6862 assert_eq!(
6863 land_resume_state(&t),
6864 LandResume::Ready,
6865 "an expired question must not be waited on forever"
6866 );
6867
6868 let after = store.get(&q.id).unwrap();
6869 assert!(
6870 !after.status.open(),
6871 "the question is abandoned, not silently ignored"
6872 );
6873 assert!(
6874 after.resolution().is_none(),
6875 "an abandoned question is not read as a decision"
6876 );
6877 }
6878
6879 #[test]
6880 fn reclaim_settles_a_running_task_against_its_last_run() {
6881 let mut t = task();
6882 t.start("20260904-000000-4043".to_owned());
6883 reclaim(&mut t, Some(run_state(RunStatus::Ready)), 2, "en");
6884 assert_eq!(
6885 t.status,
6886 TaskStatus::Done,
6887 "a run that actually finished must not stay `running` forever"
6888 );
6889 }
6890
6891 #[test]
6892 fn reclaim_maps_a_parked_run_to_parked_without_touching_last_error() {
6893 let mut t = task();
6894 t.start("20260904-000000-4043".to_owned());
6895 t.last_error = Some("earlier trouble".to_owned());
6896 let mut state = run_state(RunStatus::Judging);
6897 state.parked = true;
6898 reclaim(&mut t, Some(state), 1, "en");
6899 assert_eq!(t.status, TaskStatus::Parked);
6900 assert_eq!(t.attempts, 0, "the park is refunded, even at max_attempts");
6901 assert_eq!(t.last_error.as_deref(), Some("earlier trouble"));
6902 assert!(t.park_reason.is_some());
6903 assert_eq!(t.runs, ["20260904-000000-4043"], "the same run is kept");
6904 assert!(!t.fresh_start);
6905 assert!(t.status.runnable());
6906 }
6907
6908 #[test]
6909 fn settle_parks_a_task_without_a_failure_and_a_new_start_clears_the_reason() {
6910 let mut t = task();
6911 t.start("20260904-000000-4043".to_owned());
6912 t.last_error = Some("earlier trouble".to_owned());
6913 settle(
6914 &mut t,
6915 Verdict {
6916 status: RunStatus::Judging,
6917 left_pr: false,
6918 quota_hit: false,
6919 parked: true,
6920 no_viable_candidates: false,
6921 },
6922 "parked after `judging`",
6923 1,
6924 );
6925 assert_eq!(t.status, TaskStatus::Parked);
6926 assert_eq!(t.attempts, 0);
6927 assert_eq!(t.last_error.as_deref(), Some("earlier trouble"));
6928 assert_eq!(t.park_reason.as_deref(), Some("parked after `judging`"));
6929 assert_eq!(t.schema, crate::queue::SCHEMA);
6930 assert!(
6931 crate::notices::task_held(&t).is_none(),
6932 "a park is not news"
6933 );
6934 t.start("20260904-000000-4043".to_owned());
6935 assert_eq!(t.park_reason, None);
6936 }
6937
6938 #[test]
6939 fn reclaim_reuses_the_same_retry_policy_as_a_live_settle() {
6940 let mut t = task();
6944 t.start("20260904-000000-4043".to_owned());
6945 reclaim(&mut t, Some(run_state(RunStatus::Blocked)), 2, "en");
6946 assert_eq!(t.status, TaskStatus::Failed);
6947 assert!(t.status.runnable());
6948 }
6949
6950 #[test]
6951 fn a_settled_parked_pass_records_its_cause_and_the_refund() {
6952 let mut t = task();
6953 t.start("20260904-000000-4043".to_owned());
6954 assert_eq!(t.attempts, 1);
6955 let mut run = run_state(RunStatus::Implementing);
6956 run.id = "20260904-000000-4043".to_owned();
6957 run.begin_pass();
6958 let cause = crate::run::ParkCause::Upgrade {
6959 from: "0.121.0".to_owned(),
6960 to: Some("0.121.1".to_owned()),
6961 };
6962 run.end_pass_parked(cause.clone());
6963 run.parked = true;
6964 let verdict = Verdict {
6965 status: run.status,
6966 left_pr: false,
6967 quota_hit: false,
6968 parked: true,
6969 no_viable_candidates: false,
6970 };
6971 settle_and_diagnose(&mut t, verdict, "parked", 2, &run);
6972 assert_eq!(t.attempts, 0);
6973 assert_eq!(t.passes.len(), 1);
6974 assert_eq!(t.passes[0].cost, crate::queue::PassCost::Refunded);
6975 assert_eq!(t.passes[0].parked, Some(cause));
6976 assert_eq!(t.passes[0].pass_index, 0);
6977
6978 t.start("20260904-000000-4043".to_owned());
6981 run.begin_pass();
6982 run.status = RunStatus::Failed;
6983 let verdict = Verdict {
6984 status: RunStatus::Failed,
6985 left_pr: false,
6986 quota_hit: false,
6987 parked: false,
6988 no_viable_candidates: false,
6989 };
6990 settle_and_diagnose(&mut t, verdict, "failed", 2, &run);
6991 assert_eq!(t.passes.len(), 2);
6992 assert_eq!(t.passes[0].cost, crate::queue::PassCost::Refunded);
6993 assert_eq!(t.passes[1].cost, crate::queue::PassCost::Spent);
6994 assert_eq!(t.passes[1].pass_index, 1);
6995 assert_eq!(t.attempts, 1);
6996 }
6997
6998 #[test]
6999 fn reclaim_holds_a_running_task_whose_run_cannot_be_found() {
7000 let mut t = task();
7001 t.start("20260904-000000-4043".to_owned());
7002 reclaim(&mut t, None, 2, "en");
7003 assert_eq!(t.status, TaskStatus::Held);
7004 assert!(
7005 t.last_error
7006 .as_deref()
7007 .is_some_and(|e| e.contains("running")),
7008 "the operator needs to know why this task was held"
7009 );
7010 }
7011
7012 #[test]
7013 fn orphaned_running_tasks_are_reclaimed_but_live_ones_are_left_alone() {
7014 let dir = tempfile::tempdir().unwrap();
7015 let queue = Queue::at(dir.path().to_path_buf());
7016
7017 let mut orphaned = task();
7019 orphaned.id = "20260904-000000-orph".to_owned();
7020 orphaned.status = TaskStatus::Running;
7021 orphaned.attempts = 1;
7022 queue.put(&mut orphaned).unwrap();
7023
7024 let mut alive = task();
7025 alive.id = "20260904-000000-live".to_owned();
7026 alive.status = TaskStatus::Running;
7027 alive.attempts = 1;
7028 queue.put(&mut alive).unwrap();
7029 let _held_by_a_live_daemon = queue.claim(&alive.id).unwrap();
7030
7031 let mut queued = task();
7032 queued.id = "20260904-000000-wait".to_owned();
7033 queue.put(&mut queued).unwrap();
7034
7035 let reclaimed = reclaim_orphaned_running(&queue, 2);
7036 assert_eq!(reclaimed, vec![orphaned.id.clone()]);
7037
7038 assert_eq!(
7039 queue.get(&orphaned.id).unwrap().status,
7040 TaskStatus::Held,
7041 "nothing was driving it and there was no run to recover"
7042 );
7043 assert_eq!(
7044 queue.get(&alive.id).unwrap().status,
7045 TaskStatus::Running,
7046 "a live claim must protect the task it belongs to"
7047 );
7048 assert_eq!(queue.get(&queued.id).unwrap().status, TaskStatus::Queued);
7049 }
7050
7051 fn read_run_under(home: &Path, id: &str) -> RunState {
7057 let body = std::fs::read_to_string(home.join("runs").join(id).join("run.json")).unwrap();
7058 serde_json::from_str(&body).unwrap()
7059 }
7060
7061 #[test]
7062 fn reclaim_abandoned_runs_fails_a_run_whose_active_seats_are_all_provably_dead() {
7063 let dir = tempfile::tempdir().unwrap();
7064 let home = dir.path().to_path_buf();
7065 let now = Timestamp::now();
7066 let overrun_seat = || crate::run::ActiveSeat {
7067 node: "implement".to_owned(),
7068 started_at: now - jiff::SignedDuration::new(21_000, 0),
7069 timeout_secs: 3_600,
7070 attempt: 0,
7071 task: None,
7072 command: None,
7073 index: None,
7074 total: None,
7075 };
7076
7077 let mut dead = run_state(RunStatus::Implementing);
7078 dead.id = "20260101-000000-dead".to_owned();
7079 dead.active.insert("impl-A".to_owned(), overrun_seat());
7080 dead.driver_pid = Some(4242);
7083 dead.save_under(&home).unwrap();
7084
7085 let mut alive = run_state(RunStatus::Implementing);
7088 alive.id = "20260101-000000-aliv".to_owned();
7089 alive.active.insert("impl-A".to_owned(), overrun_seat());
7090 alive.save_under(&home).unwrap();
7091 let mut status = Status::new();
7092 status.current = vec![Current {
7093 task: "20260101-000000-task".to_owned(),
7094 run: alive.id.clone(),
7095 }];
7096 write_status_to(&home.join("daemon.json"), &status).unwrap();
7097
7098 let questions = Questions::at(home.join("questions"));
7102 let mut q = ask::Question::new(
7103 dead.id.clone(),
7104 "implement".to_owned(),
7105 "impl-A".to_owned(),
7106 "Which storage backend?".to_owned(),
7107 String::new(),
7108 vec!["SQLite".to_owned(), "Redis".to_owned()],
7109 );
7110 questions.put(&mut q).unwrap();
7111
7112 let abandoned = reclaim_abandoned_runs_with(
7113 &home,
7114 now,
7115 |pid| if pid == 4242 { Some(false) } else { None },
7116 |_| panic!("a query answering Dead outright needs no identity corroboration"),
7117 );
7118 assert_eq!(abandoned, vec![dead.id.clone()]);
7119
7120 let reloaded = read_run_under(&home, &dead.id);
7121 assert_eq!(reloaded.status, RunStatus::Failed);
7122 assert!(reloaded.active.is_empty());
7123 assert!(
7124 !questions.get(&q.id).unwrap().status.open(),
7125 "the failed run's own open question must be settled in the same pass"
7126 );
7127
7128 let still_alive = read_run_under(&home, &alive.id);
7129 assert_eq!(
7130 still_alive.status,
7131 RunStatus::Implementing,
7132 "a live daemon's claim protects it"
7133 );
7134 assert!(!still_alive.active.is_empty());
7135 }
7136
7137 #[test]
7147 fn reclaim_abandoned_runs_leaves_a_live_manual_run_alone_even_though_no_daemon_claims_it() {
7148 let dir = tempfile::tempdir().unwrap();
7149 let home = dir.path().to_path_buf();
7150 let now = Timestamp::now();
7151
7152 let mut manual = run_state(RunStatus::Reviewing);
7153 manual.id = "20260101-000000-manl".to_owned();
7154 manual.active.insert(
7155 "review-1".to_owned(),
7156 crate::run::ActiveSeat {
7157 node: "review".to_owned(),
7158 started_at: now - jiff::SignedDuration::new(21_000, 0),
7159 timeout_secs: 3_600,
7160 attempt: 0,
7161 task: None,
7162 command: None,
7163 index: None,
7164 total: None,
7165 },
7166 );
7167 manual.driver_pid = Some(4242);
7171 manual.driver_started_at = Some("1790000000".to_owned());
7172 manual.save_under(&home).unwrap();
7173
7174 let abandoned = reclaim_abandoned_runs_with(
7175 &home,
7176 now,
7177 |pid| if pid == 4242 { Some(true) } else { None },
7178 |pid| {
7179 if pid == 4242 {
7180 Some("1790000000".to_owned())
7181 } else {
7182 None
7183 }
7184 },
7185 );
7186 assert!(
7187 abandoned.is_empty(),
7188 "a manual run a real process is still driving must never be reclaimed: {abandoned:?}"
7189 );
7190
7191 let reloaded = read_run_under(&home, &manual.id);
7192 assert_eq!(reloaded.status, RunStatus::Reviewing);
7193 assert!(!reloaded.active.is_empty());
7194 }
7195
7196 #[test]
7197 fn an_already_claimed_task_is_skipped_rather_than_failed() {
7198 let dir = tempfile::tempdir().unwrap();
7199 let queue = Queue::at(dir.path().to_path_buf());
7200 let mut only = task();
7201 queue.put(&mut only).unwrap();
7202
7203 let _elsewhere = queue.claim(&only.id).unwrap();
7204 let candidates = runnable(&queue);
7205 assert_eq!(candidates.len(), 1, "the task is still runnable");
7206 assert!(
7207 queue.claim(&candidates[0].id).is_err(),
7208 "the loop cannot take a claim somebody else holds"
7209 );
7210
7211 let after = queue.get(&only.id).unwrap();
7212 assert_eq!(after.status, TaskStatus::Queued);
7213 assert_eq!(
7214 after.attempts, 0,
7215 "losing the race is not an attempt at the task"
7216 );
7217 assert_eq!(after.last_error, None);
7218 }
7219
7220 #[test]
7221 fn the_status_file_round_trips_and_its_heartbeat_advances() {
7222 let dir = tempfile::tempdir().unwrap();
7223 let path = dir.path().join("daemon.json");
7224
7225 let mut status = Status::new();
7226 status.idle = false;
7227 status.completed = 7;
7228 status.current = vec![Current {
7229 task: "20260902-000000-t111".to_owned(),
7230 run: "20260902-000001-r111".to_owned(),
7231 }];
7232 write_status_to(&path, &status).unwrap();
7233 let first: Status = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
7234 assert_eq!(first.schema, SCHEMA);
7235 assert_eq!(first.pid, std::process::id());
7236 assert!(!first.idle);
7237 assert_eq!(first.completed, 7);
7238 assert_eq!(first.current, status.current);
7239 assert!(
7240 !path.with_extension("json.tmp").exists(),
7241 "the temp file is renamed, not left behind"
7242 );
7243
7244 std::thread::sleep(Duration::from_millis(5));
7245 status.updated_at = Timestamp::now();
7246 status.polls = 3;
7247 write_status_to(&path, &status).unwrap();
7248 let second: Status =
7249 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
7250 assert!(
7251 second.updated_at > first.updated_at,
7252 "a reader can only detect staleness if the heartbeat moves"
7253 );
7254 assert_eq!(
7255 second.started_at, first.started_at,
7256 "the start time is not a heartbeat"
7257 );
7258 assert_eq!(second.polls, 3);
7259 }
7260
7261 #[test]
7262 fn reading_counts_as_running_only_while_its_heartbeat_is_fresh() {
7263 let dir = tempfile::tempdir().unwrap();
7264
7265 assert!(read_status(dir.path()).is_none(), "no file, no daemon");
7266
7267 let mut status = Status::new();
7268 status.updated_at = Timestamp::now() - jiff::SignedDuration::from_secs(60);
7269 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7270 let stale = read_status(dir.path()).unwrap();
7271 assert!(
7272 !stale.running(Timestamp::now()),
7273 "a minute without a heartbeat is a dead daemon, not a busy one"
7274 );
7275 assert!(stale.age_secs(Timestamp::now()).is_some_and(|s| s >= 55));
7276
7277 status.updated_at = Timestamp::now();
7278 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7279 let fresh = read_status(dir.path()).unwrap();
7280 assert!(fresh.running(Timestamp::now()));
7281 }
7282
7283 #[test]
7284 fn only_a_live_daemon_on_this_very_run_counts_as_working_on_it() {
7285 let dir = tempfile::tempdir().unwrap();
7286 let now = Timestamp::now();
7287 let mine = "20260903-080619-01c2";
7288
7289 assert!(
7290 !is_working_on(dir.path(), mine, now),
7291 "no status file means nobody is working on anything"
7292 );
7293
7294 let mut status = Status::new();
7295 status.current = vec![Current {
7296 task: "20260903-080340-0167".to_owned(),
7297 run: mine.to_owned(),
7298 }];
7299 status.updated_at = now;
7300 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7301 assert!(is_working_on(dir.path(), mine, now));
7302 assert!(
7303 !is_working_on(dir.path(), "20260903-105039-3cbf", now),
7304 "a daemon busy with one run is not working on another"
7305 );
7306
7307 status.updated_at = now - jiff::SignedDuration::from_secs(600);
7310 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7311 assert!(
7312 !is_working_on(dir.path(), mine, now),
7313 "a stale heartbeat is a dead daemon, so its run is a leftover"
7314 );
7315 }
7316
7317 #[test]
7318 fn is_working_on_short_matches_by_the_worktree_bays_own_name() {
7319 let dir = tempfile::tempdir().unwrap();
7320 let now = Timestamp::now();
7321
7322 assert!(
7323 !is_working_on_short(dir.path(), "01c2", now),
7324 "no status file means nobody is working on anything"
7325 );
7326
7327 let mut status = Status::new();
7328 status.current = vec![Current {
7329 task: "20260903-080340-0167".to_owned(),
7330 run: "20260903-080619-01c2".to_owned(),
7331 }];
7332 status.updated_at = now;
7333 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7334 assert!(
7335 is_working_on_short(dir.path(), "01c2", now),
7336 "the run's short id is the last block of its full id"
7337 );
7338 assert!(
7339 !is_working_on_short(dir.path(), "3cbf", now),
7340 "a daemon busy with one worktree bay is not working on another"
7341 );
7342 }
7343
7344 #[test]
7345 fn a_newer_status_file_still_yields_a_reading() {
7346 let dir = tempfile::tempdir().unwrap();
7347 std::fs::write(
7350 dir.path().join("daemon.json"),
7351 serde_json::json!({
7352 "schema": 2,
7353 "updated_at": Timestamp::now().to_string(),
7354 "idle": true,
7355 "surprise": { "nested": [1, 2, 3] },
7356 })
7357 .to_string(),
7358 )
7359 .unwrap();
7360
7361 let reading = read_status(dir.path()).expect("a forward-compatible read");
7362 assert!(reading.running(Timestamp::now()));
7363 assert!(reading.idle);
7364 assert!(reading.current.is_empty());
7365 }
7366
7367 #[test]
7368 fn an_older_daemons_single_object_current_still_reads_as_a_one_item_list() {
7369 let dir = tempfile::tempdir().unwrap();
7375 std::fs::write(
7376 dir.path().join("daemon.json"),
7377 serde_json::json!({
7378 "schema": 1,
7379 "pid": 4242,
7380 "updated_at": Timestamp::now().to_string(),
7381 "idle": false,
7382 "current": {"task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb"},
7383 "completed": 3,
7384 "polls": 9,
7385 })
7386 .to_string(),
7387 )
7388 .unwrap();
7389
7390 let reading = read_status(dir.path()).expect("an older shape must still parse");
7391 assert!(reading.running(Timestamp::now()));
7392 assert_eq!(
7393 reading.current,
7394 vec![Current {
7395 task: "20260902-140501-aaaa".to_owned(),
7396 run: "20260902-140502-bbbb".to_owned(),
7397 }]
7398 );
7399 }
7400
7401 #[test]
7402 fn an_absent_or_null_current_reads_as_idle_not_a_parse_failure() {
7403 let dir = tempfile::tempdir().unwrap();
7404 std::fs::write(
7405 dir.path().join("daemon.json"),
7406 serde_json::json!({
7407 "schema": 1,
7408 "updated_at": Timestamp::now().to_string(),
7409 "idle": true,
7410 "current": null,
7411 })
7412 .to_string(),
7413 )
7414 .unwrap();
7415 let with_null = read_status(dir.path()).expect("null must still parse");
7416 assert!(with_null.current.is_empty());
7417
7418 std::fs::write(
7419 dir.path().join("daemon.json"),
7420 serde_json::json!({
7421 "schema": 1,
7422 "updated_at": Timestamp::now().to_string(),
7423 "idle": true,
7424 })
7425 .to_string(),
7426 )
7427 .unwrap();
7428 let absent = read_status(dir.path()).expect("a missing field must still parse");
7429 assert!(absent.current.is_empty());
7430 }
7431
7432 #[test]
7433 fn a_task_without_a_repository_runs_in_the_daemons_default() {
7434 let fallback = Path::new("/default");
7435 let mut blank = task();
7436 blank.repo = PathBuf::new();
7437 assert_eq!(repo_for(&blank, fallback), PathBuf::from("/default"));
7438 let mut dot = task();
7439 dot.repo = PathBuf::from(".");
7440 assert_eq!(repo_for(&dot, fallback), PathBuf::from("/default"));
7441 assert_eq!(
7442 repo_for(&task(), fallback),
7443 PathBuf::from("/repo"),
7444 "a task that names a repository keeps it"
7445 );
7446 }
7447
7448 #[test]
7449 fn task_implementers_edit_inherits_daemon_config_and_merge() {
7450 let dir = tempfile::tempdir().unwrap();
7451 let path = dir.path().join("explicit.toml");
7452 std::fs::write(&path, "[graph]\nimplementers = 3\njudges = 1\n").unwrap();
7453 let opts = Opts {
7454 config: Some(path),
7455 merge: Some("none".into()),
7456 ..Default::default()
7457 };
7458 let mut t = task();
7459 t.set_implementers(
7460 2,
7461 &toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7462 .unwrap(),
7463 )
7464 .unwrap();
7465 let cfg = prepare_for(dir.path(), &opts, &t).unwrap();
7466 assert_eq!(cfg.graph.implementers, 2);
7467 assert_eq!(cfg.graph.judges, 1);
7468 assert_eq!(cfg.merge.mode, MergeMode::None);
7469 t.set_implementers(1, &cfg).unwrap();
7470 let mut cfg = prepare_for(dir.path(), &opts, &t).unwrap();
7471 apply_task_implementers(&mut cfg, &t);
7472 assert_eq!(cfg.graph.implementers, 1);
7473 assert_eq!(cfg.graph.judges, 1);
7474 assert_eq!(cfg.merge.mode, MergeMode::None);
7475 }
7476
7477 #[test]
7478 fn task_implementers_direct_overrides_stay_isolated_after_edits() {
7479 let dir = tempfile::tempdir().unwrap();
7480 let repo = dir.path().join("repo");
7481 std::fs::create_dir(&repo).unwrap();
7482 std::fs::write(
7483 repo.join("magi.toml"),
7484 "[graph]\njudges = 2\n[merge]\nmode = 'none'\n",
7485 )
7486 .unwrap();
7487 let daemon_config = dir.path().join("daemon.toml");
7488 std::fs::write(
7489 &daemon_config,
7490 "[graph]\njudges = 1\n[merge]\nmode = 'pr'\n",
7491 )
7492 .unwrap();
7493 let opts = Opts {
7494 config: Some(daemon_config),
7495 merge: Some("pr".into()),
7496 ..Default::default()
7497 };
7498 let q = Queue::at(dir.path().join("queue"));
7499 let validation: Config =
7500 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7501 .unwrap();
7502 for record in ["{}", r#"{"candidates":2}"#] {
7504 let overrides = serde_json::from_str(record).unwrap();
7505 let filing = crate::direct::Filing {
7506 instruction: "work".into(),
7507 title: "direct".into(),
7508 repo: repo.clone(),
7509 source: crate::queue::Source::Human,
7510 solo: false,
7511 overrides,
7512 review_of: None,
7513 };
7514 let filed =
7515 crate::direct::file(&q, dir.path(), Timestamp::now(), 1, filing, true).unwrap();
7516 let crate::direct::Filed::Standalone {
7517 mut task,
7518 claim: _claim,
7519 } = filed
7520 else {
7521 panic!("local filing")
7522 };
7523 let before = prepare_for(&repo, &opts, &task).unwrap();
7524 assert_eq!(before.graph.judges, 2, "daemon --config must not leak");
7525 assert_eq!(
7526 before.merge.mode,
7527 MergeMode::None,
7528 "daemon --merge must not leak"
7529 );
7530 for n in [2, 1, 2] {
7531 task.set_implementers(n, &validation).unwrap();
7532 q.put(&mut task).unwrap();
7533 task = q.get(&task.id).unwrap();
7534 assert!(!task.overrides.as_ref().unwrap().inherit_daemon_options);
7535 let mut cfg = prepare_for(&repo, &opts, &task).unwrap();
7536 apply_task_implementers(&mut cfg, &task);
7537 assert_eq!(cfg.graph.implementers, usize::from(n));
7538 assert_eq!(cfg.graph.judges, 2);
7539 assert_eq!(cfg.merge.mode, MergeMode::None);
7540 }
7541 }
7542 }
7543
7544 #[test]
7545 fn task_implementers_mismatch_keeps_explicit_reviews_review_only() {
7546 let cfg: Config =
7547 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7548 .unwrap();
7549 let mut t = task();
7550 t.review_of = Some("review-target".into());
7551 t.hold_manual(Some("retry".into()));
7552 let mut previous = RunState::new(
7553 PathBuf::from("/repo"),
7554 "main".into(),
7555 "abc1234".into(),
7556 "review".into(),
7557 cfg.clone(),
7558 );
7559 previous.config.graph.implementers = 2;
7560 for count in [1, 3] {
7561 t.set_implementers(count, &cfg).unwrap();
7562 t.release();
7563 assert_eq!(t.review_of.as_deref(), Some("review-target"));
7564 for proposed in [
7565 Starter::Resume(previous.id.clone()),
7566 Starter::Review("recovery-target".into()),
7567 ] {
7568 assert_eq!(
7569 starter_with_implementers(&t, Some(&previous), proposed),
7570 Starter::Review("review-target".into())
7571 );
7572 }
7573 let mut configured = cfg.clone();
7574 apply_task_implementers(&mut configured, &t);
7575 assert_eq!(configured.graph.implementers, usize::from(count));
7576 assert_eq!(previous.config.graph.implementers, 2);
7577 }
7578 }
7579
7580 #[test]
7581 fn task_implementers_dispatch_and_resume_agree() {
7582 let mut t = task();
7583 let cfg: crate::config::Config =
7584 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7585 .unwrap();
7586 let mut run = RunState::new(
7587 PathBuf::from("/repo"),
7588 "main".into(),
7589 "abc1234".into(),
7590 "work".into(),
7591 cfg.clone(),
7592 );
7593 run.config.graph.implementers = 2;
7594 for n in [2, 1, 3, 2] {
7595 t.set_implementers(n, &cfg).unwrap();
7596 let mut dispatched = cfg.clone();
7597 apply_task_implementers(&mut dispatched, &t);
7598 assert_eq!(dispatched.graph.implementers, usize::from(n));
7599 assert_eq!(implementers_match(&t, &run), n == 2);
7600 for proposed in [
7601 Starter::Resume(run.id.clone()),
7602 Starter::Review("old-branch".into()),
7603 ] {
7604 let expected = if n == 2 {
7605 proposed.clone()
7606 } else {
7607 Starter::Start
7608 };
7609 assert_eq!(
7610 starter_with_implementers(&t, Some(&run), proposed),
7611 expected
7612 );
7613 }
7614 assert_eq!(
7615 run.config.graph.implementers, 2,
7616 "history must not be rewritten"
7617 );
7618 }
7619 t.solo = false;
7620 t.overrides = None;
7621 assert!(implementers_match(&t, &run));
7622 }
7623
7624 #[test]
7625 fn a_solo_task_runs_with_one_candidate_and_a_plain_task_keeps_the_configs() {
7626 let mut solo_cfg = Config::default();
7632 solo_cfg.graph.implementers = 3;
7633 let mut solo_task = task();
7634 solo_task.solo = true;
7635 apply_task_implementers(&mut solo_cfg, &solo_task);
7636 assert_eq!(solo_cfg.graph.implementers, 1);
7637
7638 let mut plain_cfg = Config::default();
7639 plain_cfg.graph.implementers = 3;
7640 let plain_task = task();
7641 assert!(!plain_task.solo);
7642 apply_task_implementers(&mut plain_cfg, &plain_task);
7643 assert_eq!(
7644 plain_cfg.graph.implementers, 3,
7645 "a task that did not ask to run alone keeps the config's candidates"
7646 );
7647 }
7648
7649 fn loss(seat: &str, at: &str, reset: Option<&str>) -> QuotaLoss {
7650 QuotaLoss {
7651 seat: seat.into(),
7652 node: "judge".into(),
7653 at: at.parse().unwrap(),
7654 reset: reset.map(str::to_string),
7655 }
7656 }
7657
7658 #[test]
7659 fn a_resumed_run_with_only_old_quota_losses_arms_no_cooldown() {
7660 let old: Vec<QuotaLoss> = (1..=4)
7661 .map(|i| {
7662 loss(
7663 &format!("judge-{i}"),
7664 "2026-09-23T05:23:00Z",
7665 Some("2:40pm (Asia/Tokyo)"),
7666 )
7667 })
7668 .collect();
7669 let fresh = losses_this_attempt(&old, &old);
7670 assert!(fresh.is_empty());
7671 assert_eq!(cooldown_until(&fresh, Timestamp::now()), None);
7672 }
7674
7675 #[test]
7676 fn a_new_quota_loss_during_the_attempt_still_arms_the_cooldown() {
7677 let old = vec![loss("judge-1", "2026-09-23T05:23:00Z", None)];
7678 let now = Timestamp::now();
7679 let mut after = old.clone();
7680 after.push(loss("judge-2", &now.to_string(), None));
7681 let fresh = losses_this_attempt(&old, &after);
7682 assert_eq!(fresh, vec![after[1].clone()]);
7683 let until = cooldown_until(&fresh, now).expect("a fresh loss arms the cooldown");
7684 assert_eq!(
7685 until,
7686 now + jiff::SignedDuration::from_secs(QUOTA_WAIT_FALLBACK.as_secs() as i64)
7687 );
7688 }
7689
7690 #[test]
7691 fn a_recovered_seat_dropping_out_of_the_history_does_not_hide_a_new_loss() {
7692 let before = vec![
7695 loss("judge-1", "2026-09-23T05:23:00Z", None),
7696 loss("judge-2", "2026-09-23T05:24:00Z", None),
7697 ];
7698 let after = vec![
7699 loss("judge-2", "2026-09-23T05:24:00Z", None),
7700 loss("judge-1", "2026-09-24T01:00:00Z", None),
7701 ];
7702 assert_eq!(losses_this_attempt(&before, &after), vec![after[1].clone()]);
7703 }
7704
7705 #[test]
7706 fn merge_overrides_are_parsed_or_refused() {
7707 assert_eq!(merge_mode("none").unwrap(), MergeMode::None);
7708 assert_eq!(merge_mode("local").unwrap(), MergeMode::Local);
7709 assert_eq!(merge_mode("pr").unwrap(), MergeMode::Pr);
7710 assert!(merge_mode("squash").is_err());
7711 }
7712
7713 #[test]
7714 fn quota_wait_uses_a_future_reset_time_capped_and_falls_back_otherwise() {
7715 let now = Timestamp::now();
7716 let fallback = Duration::from_secs(300);
7717 let cap = Duration::from_secs(1800);
7718
7719 assert_eq!(quota_wait(None, now, fallback, cap), fallback);
7721
7722 let soon = now + jiff::SignedDuration::from_secs(600);
7724 assert_eq!(
7725 quota_wait(Some(soon), now, fallback, cap),
7726 Duration::from_secs(600)
7727 );
7728
7729 let past = now - jiff::SignedDuration::from_secs(60);
7732 assert_eq!(quota_wait(Some(past), now, fallback, cap), fallback);
7733
7734 let far = now + jiff::SignedDuration::from_secs(3 * 3600);
7737 assert_eq!(quota_wait(Some(far), now, fallback, cap), cap);
7738 }
7739
7740 #[test]
7741 fn parse_reset_hint_reads_the_claude_cli_shape_and_rolls_a_past_clock_to_tomorrow() {
7742 let now = "2026-09-07T02:50:00Z".parse::<Timestamp>().unwrap();
7743
7744 let at = parse_reset_hint("4:50am (UTC)", now, now).expect("a recognised shape parses");
7745 assert_eq!(at.to_string(), "2026-09-07T04:50:00Z");
7746
7747 let already_past =
7751 parse_reset_hint("1:00am (UTC)", now, now).expect("a recognised shape parses");
7752 assert_eq!(already_past.to_string(), "2026-09-08T01:00:00Z");
7753
7754 assert!(
7755 parse_reset_hint("session limit reached", now, now).is_none(),
7756 "free text with no recognised shape is not guessed at"
7757 );
7758 assert!(
7759 parse_reset_hint("4:50am (Nowhere/Fake)", now, now).is_none(),
7760 "an unresolvable zone name is not guessed at either"
7761 );
7762 }
7763
7764 #[test]
7765 fn parse_reset_hint_reads_the_codex_cli_shape_with_no_year_rollover_needed() {
7766 let now = "2026-09-07T02:50:00Z".parse::<Timestamp>().unwrap();
7767
7768 let at = parse_reset_hint(
7769 "You've hit your usage limit. Visit \
7770 https://chatgpt.com/codex/settings/usage to purchase more \
7771 credits or try again at Sep 19th, 2026 5:10 PM.",
7772 now,
7773 now,
7774 )
7775 .expect("the codex reset wording is a recognised shape");
7776 assert_eq!(at.to_string(), "2026-09-19T17:10:00Z");
7777
7778 let earlier = parse_reset_hint("try again at Jan 2nd, 2026 1:00 AM.", now, now)
7783 .expect("an explicit year needs no rollover");
7784 assert_eq!(earlier.to_string(), "2026-01-02T01:00:00Z");
7785
7786 assert!(
7787 parse_reset_hint("try again at Sep 19th, 26 5:10 PM.", now, now).is_none(),
7788 "a two-digit year is not the documented shape and is not guessed at"
7789 );
7790 assert!(
7791 parse_reset_hint("try again at Sept 19th, 2026 5:10 PM.", now, now).is_none(),
7792 "a four-letter month name is not the documented three-letter abbreviation"
7793 );
7794 assert!(
7795 parse_reset_hint("try again at Sep 19th, 2026 5:10 PM (UTC).", now, now).is_none(),
7796 "an explicit zone on the dated shape is a format nobody has \
7797 documented, and is refused rather than guessed at as UTC"
7798 );
7799 }
7800
7801 #[test]
7802 fn parse_reset_hint_reads_agys_relative_shape_from_when_the_loss_was_recorded() {
7803 let now = "2026-09-24T12:00:00Z".parse::<Timestamp>().unwrap();
7804 let recorded = "2026-09-24T08:00:00Z".parse::<Timestamp>().unwrap();
7805
7806 let at = parse_reset_hint("in 1h2m49s", now, recorded).expect("agy's shape parses");
7807 assert_eq!(at.as_second() - recorded.as_second(), 3769);
7808
7809 let partial = parse_reset_hint("in 45m", now, recorded).expect("units are optional");
7810 assert_eq!(partial.as_second() - recorded.as_second(), 45 * 60);
7811
7812 for bad in ["in ", "in 45", "in 3x", "in m", "in 1h junk", "1h2m"] {
7813 assert!(
7814 parse_reset_hint(bad, now, recorded).is_none(),
7815 "{bad:?} must not be guessed at"
7816 );
7817 }
7818 }
7819
7820 fn idle_loop(dir: &Path) -> (Opts, Queue, PathBuf, PathBuf, PathBuf) {
7824 let config = dir.join("magi.toml");
7825 std::fs::write(
7826 &config,
7827 "[disk]\nmin_free_bytes = 0\nauto_fold = false\ncache_limit_bytes = 0\n",
7828 )
7829 .unwrap();
7830 let opts = Opts {
7831 poll: Duration::from_secs(30),
7832 config: Some(config),
7833 repo: dir.join("repo"),
7837 ..Opts::default()
7838 };
7839 let home = dir.join("home");
7848 let worktrees = dir.join("wt");
7849 (
7850 opts,
7851 Queue::at(dir.join("queue")),
7852 home.join("daemon.json"),
7853 home,
7854 worktrees,
7855 )
7856 }
7857
7858 #[test]
7859 fn a_stop_is_idempotent_and_once_set_stays_set() {
7860 let stop = Stop::new();
7861 assert!(!stop.stopped());
7862
7863 stop.stop();
7864 assert!(stop.stopped());
7865 stop.stop();
7866 assert!(stop.stopped(), "a second stop is not a toggle");
7867
7868 let shared = stop.clone();
7869 assert!(
7870 shared.stopped(),
7871 "a clone is the same stop; that is how the loop and its caller share one"
7872 );
7873 }
7874
7875 #[test]
7876 fn only_a_stop_with_a_run_in_flight_reads_as_finishing() {
7877 let stop = Stop::new();
7878 stop.enter();
7879 assert!(
7880 !stop.finishing(),
7881 "a busy loop nobody has asked to stop is just running"
7882 );
7883
7884 stop.stop();
7885 assert!(
7886 stop.finishing(),
7887 "a stop asked for mid-run has not landed until the run is settled"
7888 );
7889
7890 stop.exit();
7891 assert!(
7892 !stop.finishing(),
7893 "once the run is settled the stop has landed and there is nothing to finish"
7894 );
7895 }
7896
7897 #[test]
7898 fn finishing_stays_true_until_the_last_of_several_runs_exits() {
7899 let stop = Stop::new();
7900 stop.enter();
7901 stop.enter();
7902 stop.stop();
7903 assert!(stop.finishing(), "two runs still in flight");
7904
7905 stop.exit();
7906 assert!(
7907 stop.finishing(),
7908 "one run finished, but a sibling is still working"
7909 );
7910
7911 stop.exit();
7912 assert!(
7913 !stop.finishing(),
7914 "the last run out is what actually lands the stop"
7915 );
7916 }
7917
7918 #[tokio::test]
7919 async fn a_loop_already_asked_to_stop_returns_without_waiting_out_a_poll() {
7920 let dir = tempfile::tempdir().unwrap();
7921 let (opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
7922 let stop = Stop::new();
7923 stop.stop();
7924
7925 let began = std::time::Instant::now();
7926 tokio::time::timeout(
7927 Duration::from_secs(2),
7928 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
7929 )
7930 .await
7931 .expect("a stopped loop must return, not sit out its poll interval")
7932 .expect("the loop's own setup and teardown must not fail");
7933 assert!(
7934 began.elapsed() < opts.poll,
7935 "returned only after {:?}, which is a poll interval, not a stop",
7936 began.elapsed()
7937 );
7938 }
7939
7940 #[tokio::test]
7941 async fn a_stop_while_idle_wakes_the_wait_instead_of_sleeping_it_out() {
7942 let dir = tempfile::tempdir().unwrap();
7943 let (opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
7944 let stop = Stop::new();
7945
7946 let asker = {
7949 let stop = stop.clone();
7950 tokio::spawn(async move {
7951 tokio::time::sleep(Duration::from_millis(20)).await;
7952 stop.stop();
7953 })
7954 };
7955
7956 let began = std::time::Instant::now();
7957 tokio::time::timeout(
7958 Duration::from_secs(2),
7959 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
7960 )
7961 .await
7962 .expect("a stop asked for while idle must wake the wait")
7963 .expect("the loop's own setup and teardown must not fail");
7964 asker.await.unwrap();
7965 assert!(
7966 began.elapsed() < opts.poll,
7967 "returned only after {:?}, so the stop waited on the sleep",
7968 began.elapsed()
7969 );
7970 }
7971
7972 #[tokio::test]
7973 async fn a_stopped_loop_leaves_no_status_file_claiming_it_is_running() {
7974 let dir = tempfile::tempdir().unwrap();
7975 let (opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
7976 let stop = Stop::new();
7977 stop.stop();
7978
7979 tokio::time::timeout(
7980 Duration::from_secs(2),
7981 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
7982 )
7983 .await
7984 .expect("a stopped loop must return")
7985 .expect("the loop's own setup and teardown must not fail");
7986
7987 assert!(
7988 home.is_dir(),
7989 "the loop did publish a status file, so its removal is the teardown and not an absence"
7990 );
7991 assert!(
7992 !status_file.exists(),
7993 "a stopped loop clears its status file"
7994 );
7995 assert!(
7996 read_status(&home).is_none(),
7997 "a reader must see no daemon at all, not a heartbeat that merely stopped"
7998 );
7999 }
8000
8001 #[tokio::test]
8002 async fn once_runs_startup_housekeeping_before_an_empty_queue_exits() {
8003 let dir = tempfile::tempdir().unwrap();
8004 let (mut opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
8005 opts.once = true;
8006
8007 let mut settled = RunState::new(
8008 dir.path().join("repo"),
8009 "main".to_owned(),
8010 "abc1234".to_owned(),
8011 "fixture".to_owned(),
8012 Config::default(),
8013 );
8014 settled.status = RunStatus::Ready;
8015 let run_dir = home.join("runs").join(&settled.id);
8016 std::fs::create_dir_all(&run_dir).unwrap();
8017 std::fs::write(
8018 run_dir.join("run.json"),
8019 serde_json::to_string_pretty(&settled).unwrap(),
8020 )
8021 .unwrap();
8022 let questions = Questions::at(home.join("questions"));
8023 let mut question = ask::Question::new(
8024 settled.id.clone(),
8025 "review".to_owned(),
8026 "reviewer-1".to_owned(),
8027 "Continue?".to_owned(),
8028 String::new(),
8029 Vec::new(),
8030 );
8031 questions.put(&mut question).unwrap();
8032
8033 drive(&opts, &queue, &status_file, &home, &worktrees, &Stop::new())
8034 .await
8035 .unwrap();
8036
8037 assert_eq!(
8038 questions.get(&question.id).unwrap().status,
8039 ask::QuestionStatus::Abandoned,
8040 "an empty --once drain still performs startup question cleanup"
8041 );
8042 }
8043
8044 #[test]
8045 fn cache_check_due_fires_immediately_then_waits_out_its_own_interval() {
8046 let t0 = "2026-09-15T00:00:00Z".parse::<Timestamp>().unwrap();
8047
8048 assert!(
8049 cache_check_due(None, t0, CACHE_CHECK_INTERVAL_SECS),
8050 "never checked before: due at once"
8051 );
8052
8053 let one_sec_later = t0 + jiff::SignedDuration::from_secs(1);
8054 assert!(
8055 !cache_check_due(Some(t0), one_sec_later, CACHE_CHECK_INTERVAL_SECS),
8056 "well inside the interval: not due yet"
8057 );
8058
8059 let at_the_edge = t0 + jiff::SignedDuration::from_secs(CACHE_CHECK_INTERVAL_SECS as i64);
8060 assert!(
8061 !cache_check_due(Some(t0), at_the_edge, CACHE_CHECK_INTERVAL_SECS),
8062 "exactly at the edge: not yet due, same convention as `clean::due`"
8063 );
8064
8065 let past_it = t0 + jiff::SignedDuration::from_secs(CACHE_CHECK_INTERVAL_SECS as i64 + 1);
8066 assert!(
8067 cache_check_due(Some(t0), past_it, CACHE_CHECK_INTERVAL_SECS),
8068 "past the interval: due again"
8069 );
8070 }
8071
8072 fn cache_check_opts(dir: &Path, cache_dir: &Path, limit_bytes: u64) -> Opts {
8077 let config = dir.join("magi.toml");
8078 std::fs::write(
8084 &config,
8085 format!(
8086 "[disk]\nmin_free_bytes = 0\nauto_fold = false\ncache_limit_bytes = {limit_bytes}\n\n\
8087 [verify]\ngate = ['CARGO_TARGET_DIR={} cargo make check']\n",
8088 cache_dir.display()
8089 ),
8090 )
8091 .unwrap();
8092 Opts {
8093 config: Some(config),
8094 repo: dir.join("repo"),
8095 ..Opts::default()
8096 }
8097 }
8098
8099 #[tokio::test]
8100 async fn maybe_prune_cache_between_runs_reprunes_only_once_its_own_interval_elapses() {
8101 let dir = tempfile::tempdir().unwrap();
8102 let home = dir.path().join("home");
8103 let cache_dir = dir.path().join("cache");
8104 std::fs::create_dir_all(&cache_dir).unwrap();
8105 std::fs::write(cache_dir.join("a"), vec![0u8; 10]).unwrap();
8106 let opts = cache_check_opts(dir.path(), &cache_dir, 1);
8107
8108 let running = Stop::new();
8111 let mut last_checked = None;
8112 let t0 = "2026-09-15T00:00:00Z".parse::<Timestamp>().unwrap();
8113 maybe_prune_cache_between_runs(&opts.repo, &opts, &home, &running, &mut last_checked, t0)
8114 .await;
8115 assert_eq!(
8116 crate::disk::dir_size(&cache_dir),
8117 0,
8118 "over the cap on the first check ever: pruned at once, no idle queue required"
8119 );
8120 assert_eq!(last_checked, Some(t0));
8121
8122 std::fs::write(cache_dir.join("b"), vec![0u8; 10]).unwrap();
8124 let too_soon = t0 + jiff::SignedDuration::from_secs(1);
8125 maybe_prune_cache_between_runs(
8126 &opts.repo,
8127 &opts,
8128 &home,
8129 &running,
8130 &mut last_checked,
8131 too_soon,
8132 )
8133 .await;
8134 assert_eq!(
8135 crate::disk::dir_size(&cache_dir),
8136 10,
8137 "too soon since the last check: left alone rather than rescanned every call"
8138 );
8139 assert_eq!(
8140 last_checked,
8141 Some(t0),
8142 "an idle check does not reset the clock"
8143 );
8144
8145 let due_again = t0 + jiff::SignedDuration::from_secs(CACHE_CHECK_INTERVAL_SECS as i64 + 1);
8147 maybe_prune_cache_between_runs(
8148 &opts.repo,
8149 &opts,
8150 &home,
8151 &running,
8152 &mut last_checked,
8153 due_again,
8154 )
8155 .await;
8156 assert_eq!(
8157 crate::disk::dir_size(&cache_dir),
8158 0,
8159 "due again: pruned back under the cap"
8160 );
8161 }
8162
8163 #[tokio::test]
8171 async fn a_stop_already_asked_for_skips_the_between_runs_cache_walk() {
8172 let dir = tempfile::tempdir().unwrap();
8173 let home = dir.path().join("home");
8174 let cache_dir = dir.path().join("cache");
8175 std::fs::create_dir_all(&cache_dir).unwrap();
8176 std::fs::write(cache_dir.join("a"), vec![0u8; 10]).unwrap();
8177 let opts = cache_check_opts(dir.path(), &cache_dir, 1);
8178
8179 let stop = Stop::new();
8180 stop.stop();
8181 assert!(
8182 !stop.finishing(),
8183 "no run is in flight at a between-runs boundary, so nothing else \
8184 would tell the operator this stop had not taken effect yet"
8185 );
8186
8187 let mut last_checked = None;
8188 let t0 = "2026-09-15T00:00:00Z".parse::<Timestamp>().unwrap();
8189 maybe_prune_cache_between_runs(&opts.repo, &opts, &home, &stop, &mut last_checked, t0)
8190 .await;
8191 assert_eq!(
8192 crate::disk::dir_size(&cache_dir),
8193 10,
8194 "over its cap, and due for the first check ever, but a stop outranks \
8195 it: the cap is a standing policy the next start measures again"
8196 );
8197 assert_eq!(
8198 last_checked, None,
8199 "a check that never happened must not claim the interval"
8200 );
8201 }
8202
8203 #[tokio::test]
8214 async fn cache_prune_reaches_a_queue_that_never_goes_idle() {
8215 let dir = tempfile::tempdir().unwrap();
8216 let cache_dir = dir.path().join("cache");
8217 std::fs::create_dir_all(&cache_dir).unwrap();
8218 std::fs::write(cache_dir.join("stale"), vec![0u8; 4096]).unwrap();
8219
8220 let mut opts = cache_check_opts(dir.path(), &cache_dir, 1);
8221 opts.poll = Duration::from_millis(20);
8222 opts.max_attempts = 1_000;
8223
8224 let queue = Queue::at(dir.path().join("queue"));
8225 for i in 0..128 {
8226 let mut task = Task::new(
8227 format!("queued work {i}"),
8228 format!("queued work {i}"),
8229 opts.repo.clone(),
8230 Source::Human,
8231 );
8232 queue.put(&mut task).unwrap();
8233 }
8234
8235 let home = dir.path().join("home");
8236 let worktrees = dir.path().join("wt");
8237 let status_file = home.join("daemon.json");
8238 let stop = Stop::new();
8239 let stopper = {
8240 let stop = stop.clone();
8241 let queue = Queue::at(dir.path().join("queue"));
8242 tokio::spawn(async move {
8243 for _ in 0..2_000 {
8247 if queue.list().iter().filter(|t| t.attempts == 1).count() >= 2 {
8248 break;
8249 }
8250 tokio::time::sleep(Duration::from_millis(5)).await;
8251 }
8252 stop.stop();
8253 })
8254 };
8255
8256 tokio::time::timeout(
8257 Duration::from_secs(20),
8258 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
8259 )
8260 .await
8261 .expect("the loop must not hang on a queue that keeps producing failing work")
8262 .expect("the loop's own setup and teardown must not fail");
8263 stopper.await.unwrap();
8264
8265 let after = queue.list();
8266 let completed = after.iter().filter(|t| t.attempts == 1).count();
8267 assert!(
8268 completed >= 2,
8269 "the harness must have processed multiple tasks to exercise a busy queue \
8270 (got {completed} completed task(s))"
8271 );
8272 assert!(
8273 after.iter().any(|t| t.status.runnable()),
8274 "unprocessed tasks must remain: the external stop, rather than an \
8275 exhausted backlog, must end the test"
8276 );
8277 assert!(
8278 after
8279 .iter()
8280 .filter(|t| t.attempts > 0)
8281 .all(|t| { t.attempts == 1 && t.status == TaskStatus::Held }),
8282 "a busy backlog must not retry a task's unchanged startup failure"
8283 );
8284
8285 assert_eq!(
8286 crate::disk::dir_size(&cache_dir),
8287 0,
8288 "an oversized cache must not be left to grow unboundedly just because the \
8289 queue kept the loop busy the whole time"
8290 );
8291 }
8292
8293 #[test]
8294 fn task_question_reconciliation_keeps_references_and_retires_manual_releases() {
8295 let dir = tempfile::tempdir().unwrap();
8296 let queue = Queue::at(dir.path().join("queue"));
8297 let questions = Questions::at(dir.path().join("questions"));
8298 let mut task = task();
8299 queue.put(&mut task).unwrap();
8300
8301 let mut task_question = ask::Question::new(
8302 task.id.clone(),
8303 crate::conduct::NODE.to_owned(),
8304 "conduct".to_owned(),
8305 "Which backend?".to_owned(),
8306 String::new(),
8307 Vec::new(),
8308 );
8309 questions.put(&mut task_question).unwrap();
8310 task.block(vec![task_question.id.clone()], None);
8311 queue.put(&mut task).unwrap();
8312
8313 let mut run_question = ask::Question::new(
8314 "20260101-000000-run1".to_owned(),
8315 "review".to_owned(),
8316 "reviewer-1".to_owned(),
8317 "Run question".to_owned(),
8318 String::new(),
8319 Vec::new(),
8320 );
8321 questions.put(&mut run_question).unwrap();
8322
8323 let mut coincidental = ask::Question::new(
8328 task.id.clone(),
8329 "review".to_owned(),
8330 "reviewer-1".to_owned(),
8331 "Unrelated review question".to_owned(),
8332 String::new(),
8333 Vec::new(),
8334 );
8335 questions.put(&mut coincidental).unwrap();
8336
8337 reconcile_task_questions(&queue, &questions);
8338 assert!(questions.get(&task_question.id).unwrap().status.open());
8339 assert!(questions.get(&run_question.id).unwrap().status.open());
8340 assert!(questions.get(&coincidental.id).unwrap().status.open());
8341
8342 task.release();
8343 queue.put(&mut task).unwrap();
8344 reconcile_task_questions(&queue, &questions);
8345 assert_eq!(
8346 questions.get(&task_question.id).unwrap().status,
8347 ask::QuestionStatus::Abandoned
8348 );
8349 assert!(
8350 questions.get(&run_question.id).unwrap().status.open(),
8351 "run questions remain the run janitor's responsibility"
8352 );
8353 assert!(
8354 questions.get(&coincidental.id).unwrap().status.open(),
8355 "a non-conductor question must not be abandoned just because its \
8356 run id coincides with a task id"
8357 );
8358 }
8359
8360 #[test]
8361 fn a_freshly_started_running_task_is_never_stalled() {
8362 let dir = tempfile::tempdir().unwrap();
8363 let mut t = task();
8364 t.start("run-1".to_owned());
8365 assert!(!is_stalled(&t, dir.path(), Timestamp::now()));
8368 }
8369
8370 #[test]
8371 fn a_long_running_task_with_no_live_daemon_is_stalled() {
8372 let dir = tempfile::tempdir().unwrap();
8373 let mut t = task();
8374 t.start("run-1".to_owned());
8375 t.updated_at = Timestamp::now()
8376 - jiff::SignedDuration::from_secs(STALLED_RUNNING.as_secs() as i64 + 60);
8377 assert!(is_stalled(&t, dir.path(), Timestamp::now()));
8378 assert_eq!(
8379 stalled_tasks(
8380 &Queue::at(dir.path().join("q")),
8381 dir.path(),
8382 Timestamp::now()
8383 )
8384 .len(),
8385 0,
8386 "the task was never written to this queue"
8387 );
8388 }
8389
8390 #[test]
8391 fn a_long_running_task_a_live_daemon_still_names_is_not_stalled() {
8392 let dir = tempfile::tempdir().unwrap();
8393 let mut t = task();
8394 t.id = "20260903-080340-0167".to_owned();
8395 t.start("20260903-080619-01c2".to_owned());
8396 t.updated_at = Timestamp::now()
8397 - jiff::SignedDuration::from_secs(STALLED_RUNNING.as_secs() as i64 + 60);
8398
8399 let mut status = Status::new();
8400 status.current = vec![Current {
8401 task: t.id.clone(),
8402 run: "20260903-080619-01c2".to_owned(),
8403 }];
8404 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
8405
8406 assert!(
8407 !is_stalled(&t, dir.path(), Timestamp::now()),
8408 "a live daemon's own heartbeat rules out stalled, however long the task has run"
8409 );
8410 }
8411
8412 fn backdate_task(queue: &Queue, id: &str, seconds_ago: i64) {
8416 let path = queue.path_of(id);
8417 let body = std::fs::read_to_string(&path).unwrap();
8418 let mut v: serde_json::Value = serde_json::from_str(&body).unwrap();
8419 let old = Timestamp::now() - jiff::SignedDuration::from_secs(seconds_ago);
8420 v["updated_at"] = serde_json::Value::String(old.to_string());
8421 std::fs::write(&path, serde_json::to_string_pretty(&v).unwrap()).unwrap();
8422 }
8423
8424 #[test]
8425 fn stalled_tasks_still_reaches_a_task_reclaim_could_not_claim_yet() {
8426 let dir = tempfile::tempdir().unwrap();
8439 let queue = Queue::at(dir.path().join("queue"));
8440 let home = dir.path().join("home");
8441
8442 let mut t = task();
8443 t.id = "20260101-000001-lock".to_owned();
8444 t.start("run-1".to_owned());
8445 queue.put(&mut t).unwrap();
8446 backdate_task(&queue, &t.id, STALLED_RUNNING.as_secs() as i64 + 60);
8447 std::fs::write(
8448 dir.path().join("queue").join(format!("{}.lock", t.id)),
8449 "not a pid",
8450 )
8451 .unwrap();
8452
8453 let now = Timestamp::now();
8454 assert!(
8455 reclaim_orphaned_running(&queue, 2).is_empty(),
8456 "the unparseable lock is still well within STALE_CLAIM, so the claim fails \
8457 and reclaim must leave the task alone"
8458 );
8459 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Running);
8460
8461 let stalled = stalled_tasks(&queue, &home, now);
8462 assert_eq!(
8463 stalled.len(),
8464 1,
8465 "reclaim's inability to claim it yet must not hide it from the conductor"
8466 );
8467 assert_eq!(stalled[0].id, t.id);
8468 }
8469
8470 #[test]
8471 fn ordinary_dead_daemon_task_is_shown_stalled_before_reclaim_and_can_be_requeued() {
8472 let dir = tempfile::tempdir().unwrap();
8473 crate::run::set_home(dir.path().join("run-home"));
8474 let queue = Queue::at(dir.path().join("queue"));
8475 let home = dir.path().join("home");
8476 let questions = Questions::at(dir.path().join("questions"));
8477
8478 let mut t = task();
8479 t.id = "20260101-000003-dead".to_owned();
8480 t.start("missing-run".to_owned());
8481 queue.put(&mut t).unwrap();
8482 backdate_task(&queue, &t.id, STALLED_RUNNING.as_secs() as i64 + 60);
8483
8484 let stalled = stalled_tasks(&queue, &home, Timestamp::now());
8487 assert_eq!(
8488 stalled.iter().map(|task| &task.id).collect::<Vec<_>>(),
8489 [&t.id]
8490 );
8491 assert_eq!(reclaim_orphaned_running(&queue, 2), [t.id.clone()]);
8492 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
8493
8494 crate::conduct::apply(
8497 &queue,
8498 &questions,
8499 &crate::conduct::Verdict {
8500 decisions: vec![crate::conduct::Decision {
8501 id: t.id.clone(),
8502 recovery: Some(crate::conduct::Recovery::Requeue),
8503 ..crate::conduct::Decision::default()
8504 }],
8505 },
8506 )
8507 .unwrap();
8508 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
8509 }
8510
8511 #[test]
8512 fn stalled_tasks_reports_exactly_the_tasks_is_stalled_agrees_on() {
8513 let dir = tempfile::tempdir().unwrap();
8514 let queue = Queue::at(dir.path().join("queue"));
8515 let home = dir.path().join("home");
8516
8517 let mut fresh = task();
8518 fresh.id = "20260101-000001-aaaa".to_owned();
8519 fresh.start("run-1".to_owned());
8520 queue.put(&mut fresh).unwrap();
8521
8522 let mut old = task();
8523 old.id = "20260101-000002-bbbb".to_owned();
8524 old.start("run-2".to_owned());
8525 queue.put(&mut old).unwrap();
8526 backdate_task(&queue, &old.id, STALLED_RUNNING.as_secs() as i64 + 60);
8527
8528 let stalled = stalled_tasks(&queue, &home, Timestamp::now());
8529 assert_eq!(stalled.len(), 1);
8530 assert_eq!(stalled[0].id, old.id);
8531 }
8532
8533 #[test]
8534 fn queued_and_finished_task_views_partition_by_status() {
8535 let dir = tempfile::tempdir().unwrap();
8536 let queue = Queue::at(dir.path().join("queue"));
8537
8538 let mut queued = task();
8539 queued.id = "20260101-000001-aaaa".to_owned();
8540 queue.put(&mut queued).unwrap();
8541
8542 let mut failed = task();
8543 failed.id = "20260101-000002-bbbb".to_owned();
8544 failed.start("run-1".to_owned());
8545 failed.fail("gate red", 5);
8546 queue.put(&mut failed).unwrap();
8547
8548 let mut held = task();
8549 held.id = "20260101-000003-cccc".to_owned();
8550 held.hold_machine(None);
8551 queue.put(&mut held).unwrap();
8552
8553 let mut running = task();
8554 running.id = "20260101-000004-dddd".to_owned();
8555 running.start("run-2".to_owned());
8556 queue.put(&mut running).unwrap();
8557
8558 let queued_ids: Vec<String> = queued_tasks(&queue).into_iter().map(|t| t.id).collect();
8559 assert_eq!(queued_ids, [queued.id.clone()]);
8560
8561 let mut finished_ids: Vec<String> =
8562 finished_tasks(&queue).into_iter().map(|t| t.id).collect();
8563 finished_ids.sort_unstable();
8564 let mut want = vec![failed.id.clone(), held.id.clone()];
8565 want.sort_unstable();
8566 assert_eq!(finished_ids, want);
8567 }
8568
8569 #[test]
8570 fn resolve_blockers_clears_a_done_dependency_and_keeps_an_unresolved_one() {
8571 let dir = tempfile::tempdir().unwrap();
8572 let queue = Queue::at(dir.path().join("queue"));
8573 let questions = ask::Questions::at(dir.path().join("questions"));
8574
8575 let mut dep = task();
8576 dep.id = "20260101-000001-dep0".to_owned();
8577 dep.succeed();
8578 queue.put(&mut dep).unwrap();
8579
8580 let mut still_going = task();
8581 still_going.id = "20260101-000002-dep1".to_owned();
8582 queue.put(&mut still_going).unwrap();
8583
8584 let mut blocked = task();
8585 blocked.id = "20260101-000003-main".to_owned();
8586 blocked.block(
8587 vec![dep.id.clone(), still_going.id.clone()],
8588 Some("waits on both".to_owned()),
8589 );
8590 queue.put(&mut blocked).unwrap();
8591
8592 resolve_blockers(&queue, &questions);
8593
8594 let after = queue.get(&blocked.id).unwrap();
8595 assert_eq!(
8596 after.status,
8597 TaskStatus::Blocked,
8598 "one dependency is still outstanding"
8599 );
8600 assert_eq!(after.blocked_by, [still_going.id.clone()]);
8601 }
8602
8603 #[test]
8604 fn resolve_blockers_carries_an_answers_content_onto_the_task_and_unblocks_it() {
8605 let dir = tempfile::tempdir().unwrap();
8606 let queue = Queue::at(dir.path().join("queue"));
8607 let questions = ask::Questions::at(dir.path().join("questions"));
8608
8609 let mut q = crate::ask::Question::new(
8610 "20260101-000001-main".to_owned(),
8611 crate::conduct::NODE.to_owned(),
8612 "conduct".to_owned(),
8613 "Which backend?".to_owned(),
8614 String::new(),
8615 Vec::new(),
8616 );
8617 questions.put(&mut q).unwrap();
8618 q.answer(crate::ask::Answer::Text("SQLite".to_owned()))
8619 .unwrap();
8620 questions.put(&mut q).unwrap();
8621
8622 let mut blocked = task();
8623 blocked.id = "20260101-000001-main".to_owned();
8624 blocked.block(vec![q.id.clone()], Some("which backend?".to_owned()));
8625 queue.put(&mut blocked).unwrap();
8626
8627 resolve_blockers(&queue, &questions);
8628
8629 let after = queue.get(&blocked.id).unwrap();
8630 assert_eq!(
8631 after.status,
8632 TaskStatus::Queued,
8633 "the only blocker resolved"
8634 );
8635 assert_eq!(after.answers.len(), 1);
8636 assert_eq!(after.answers[0].question, "Which backend?");
8637 assert_eq!(after.answers[0].answer, "SQLite");
8638
8639 let instruction = instruction_for(&after);
8641 assert!(instruction.contains("Which backend?"));
8642 assert!(instruction.contains("SQLite"));
8643 }
8644
8645 #[test]
8646 fn resolve_blockers_holds_a_task_whose_conductor_question_was_abandoned() {
8647 let dir = tempfile::tempdir().unwrap();
8648 let queue = Queue::at(dir.path().join("queue"));
8649 let questions = ask::Questions::at(dir.path().join("questions"));
8650
8651 let mut q = crate::ask::Question::new(
8652 "20260101-000001-main".to_owned(),
8653 crate::conduct::NODE.to_owned(),
8654 "conduct".to_owned(),
8655 "Is the setup done?".to_owned(),
8656 String::new(),
8657 Vec::new(),
8658 );
8659 q.abandon("no answer within 60s of asking");
8660 questions.put(&mut q).unwrap();
8661
8662 let mut blocked = task();
8663 blocked.id = "20260101-000001-main".to_owned();
8664 blocked.block(vec![q.id.clone()], Some("setup?".to_owned()));
8665 queue.put(&mut blocked).unwrap();
8666
8667 resolve_blockers(&queue, &questions);
8668
8669 let after = queue.get(&blocked.id).unwrap();
8670 assert_eq!(
8671 after.status,
8672 TaskStatus::Held,
8673 "never left blocked on nothing"
8674 );
8675 assert!(!after.operator_held(), "a machine hold, for triage");
8676 assert!(
8677 after
8678 .hold_reason
8679 .as_deref()
8680 .unwrap_or_default()
8681 .contains("went unanswered")
8682 );
8683 }
8684
8685 #[test]
8686 fn resolve_blockers_restores_a_held_task_to_held_instead_of_queuing_it() {
8687 let dir = tempfile::tempdir().unwrap();
8693 let queue = Queue::at(dir.path().join("queue"));
8694 let questions = ask::Questions::at(dir.path().join("questions"));
8695
8696 let mut q = crate::ask::Question::new(
8697 "20260101-000001-main".to_owned(),
8698 crate::conduct::NODE.to_owned(),
8699 "conduct".to_owned(),
8700 "How should this be handled?".to_owned(),
8701 String::new(),
8702 Vec::new(),
8703 );
8704 questions.put(&mut q).unwrap();
8705 q.answer(crate::ask::Answer::Text(
8706 "leave it held, a human will look at it later".to_owned(),
8707 ))
8708 .unwrap();
8709 questions.put(&mut q).unwrap();
8710
8711 let mut held = task();
8712 held.id = "20260101-000001-main".to_owned();
8713 held.hold_machine(Some("out of attempts".to_owned()));
8714 held.block(vec![q.id.clone()], Some("what now?".to_owned()));
8715 queue.put(&mut held).unwrap();
8716
8717 resolve_blockers(&queue, &questions);
8718
8719 let after = queue.get(&held.id).unwrap();
8720 assert_eq!(after.status, TaskStatus::Held);
8721 assert_eq!(after.hold_reason.as_deref(), Some("out of attempts"));
8722 assert_eq!(
8723 after.answers[0].answer,
8724 "leave it held, a human will look at it later"
8725 );
8726 }
8727
8728 #[test]
8729 fn resolve_blockers_releases_a_task_whose_dependency_was_deleted_on_purpose() {
8730 let dir = tempfile::tempdir().unwrap();
8731 let queue = Queue::at(dir.path().join("queue"));
8732 let questions = ask::Questions::at(dir.path().join("questions"));
8733 let mut dep = task();
8734 dep.id = "20260101-000001-gone".to_owned();
8735 queue.put(&mut dep).unwrap();
8736 let mut blocked = task();
8737 blocked.id = "20260101-000003-main".to_owned();
8738 blocked.block(vec![dep.id.clone()], None);
8739 queue.put(&mut blocked).unwrap();
8740
8741 let claim = queue.claim(&blocked.id).unwrap();
8743 queue.remove(&dep.id, false, &questions).unwrap();
8744 drop(claim);
8745 resolve_blockers(&queue, &questions);
8746
8747 let after = queue.get(&blocked.id).unwrap();
8748 assert_eq!(after.status, TaskStatus::Queued);
8749 assert!(after.blocked_by.is_empty());
8750 }
8751
8752 #[test]
8753 fn resolve_blockers_holds_a_task_whose_dependency_was_deleted() {
8754 let dir = tempfile::tempdir().unwrap();
8760 let queue = Queue::at(dir.path().join("queue"));
8761 let questions = ask::Questions::at(dir.path().join("questions"));
8762
8763 let mut still_going = task();
8764 still_going.id = "20260101-000002-dep1".to_owned();
8765 queue.put(&mut still_going).unwrap();
8766
8767 let mut blocked = task();
8768 blocked.id = "20260101-000003-main".to_owned();
8769 blocked.block(
8770 vec!["20260101-000001-gone".to_owned(), still_going.id.clone()],
8771 Some("waits on both".to_owned()),
8772 );
8773 queue.put(&mut blocked).unwrap();
8774
8775 resolve_blockers(&queue, &questions);
8776
8777 let after = queue.get(&blocked.id).unwrap();
8778 assert_eq!(
8779 after.status,
8780 TaskStatus::Held,
8781 "a missing dependency must not leave the task blocked forever"
8782 );
8783 assert_eq!(after.hold_source, Some(crate::queue::HoldSource::Machine));
8784 assert!(after.blocked_by.is_empty());
8785 let reason = after.hold_reason.as_deref().unwrap_or_default();
8786 assert!(
8787 reason.contains("20260101-000001-gone"),
8788 "the missing id must be named so an operator can tell what happened: {reason}"
8789 );
8790 assert!(
8791 reason.contains(&still_going.id),
8792 "the still-valid dependency must not silently vanish from the record: {reason}"
8793 );
8794 }
8795
8796 #[test]
8797 fn instruction_for_is_unchanged_without_any_answers() {
8798 let t = task();
8799 assert_eq!(instruction_for(&t), t.instruction);
8800 }
8801
8802 #[test]
8803 fn task_attachments_are_absolute_and_a_missing_file_is_an_error() {
8804 let dir = tempfile::tempdir().unwrap();
8805 let q = Queue::at(dir.path().join("queue"));
8806 let src = dir.path().join("shot.png");
8807 std::fs::write(&src, "x").unwrap();
8808 let mut t = task();
8809 q.attach(&mut t, &[src]).unwrap();
8810 let paths = task_attachments(&q, &t).unwrap();
8811 assert_eq!(paths.len(), 1);
8812 assert!(paths[0].is_absolute() && paths[0].is_file());
8813 std::fs::remove_file(&paths[0]).unwrap();
8814 let err = task_attachments(&q, &t).unwrap_err().to_string();
8815 assert!(err.contains("shot.png"), "{err}");
8816 }
8817
8818 #[test]
8819 fn resumed_instruction_is_unchanged_without_any_answers() {
8820 let t = task();
8821 assert_eq!(resumed_instruction(&t.instruction, &t), t.instruction);
8822 }
8823
8824 #[test]
8825 fn resumed_instruction_carries_a_new_answer_onto_the_old_run() {
8826 let mut t = task();
8827 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
8828 let old = t.instruction.clone();
8832
8833 let refreshed = resumed_instruction(&old, &t);
8834 assert!(refreshed.starts_with(&old), "the original text is kept");
8835 assert!(refreshed.contains("Which backend?"));
8836 assert!(refreshed.contains("SQLite"));
8837 }
8838
8839 #[test]
8840 fn resumed_instruction_keeps_an_original_answers_heading() {
8841 let mut t = task();
8842 t.instruction = "Context\n\n# Operator answers\n\nThis is part of the task.".to_owned();
8843 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
8844
8845 let refreshed = resumed_instruction(&t.instruction, &t);
8846
8847 assert!(
8848 refreshed.starts_with(&t.instruction),
8849 "an answers heading in the original instruction is not the appended block"
8850 );
8851 assert_eq!(refreshed.matches(ANSWERS_HEADER).count(), 2);
8852 assert!(refreshed.contains("Which backend?"));
8853 assert!(refreshed.contains("SQLite"));
8854
8855 let repeated = resumed_instruction(&refreshed, &t);
8856 assert_eq!(
8857 repeated, refreshed,
8858 "only the final appended block is refreshed"
8859 );
8860 }
8861
8862 #[test]
8863 fn resumed_instruction_does_not_duplicate_across_repeated_resumes() {
8864 let mut t = task();
8865 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
8866
8867 let once = resumed_instruction(&t.instruction, &t);
8871 let twice = resumed_instruction(&once, &t);
8872 assert_eq!(once, twice);
8873 assert_eq!(once.matches("Which backend?").count(), 1);
8874
8875 t.record_answer("Which cache?".to_owned(), "Redis".to_owned());
8877 let refreshed = resumed_instruction(&once, &t);
8878 assert_eq!(refreshed.matches(ANSWERS_HEADER).count(), 1);
8879 assert!(refreshed.contains("Which backend?"));
8880 assert!(refreshed.contains("Which cache?"));
8881 }
8882
8883 #[test]
8884 fn prepare_instruction_covers_all_three_starters() {
8885 let mut t = task();
8886 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
8887
8888 assert_eq!(
8891 prepare_instruction(&Starter::Start, None, &t),
8892 Some(instruction_for(&t))
8893 );
8894
8895 let old = t.instruction.clone();
8898 assert_eq!(
8899 prepare_instruction(&Starter::Resume("some-run".to_owned()), Some(&old), &t),
8900 Some(resumed_instruction(&old, &t))
8901 );
8902
8903 assert_eq!(
8907 prepare_instruction(&Starter::Review("magi/eba2/A".to_owned()), Some(&old), &t),
8908 None
8909 );
8910 }
8911
8912 #[test]
8913 fn choose_starter_prefers_review_over_resume_when_the_branch_survived() {
8914 assert_eq!(
8915 choose_starter(Some("magi/eba2/A"), true, Some("some-run")),
8916 Starter::Review("magi/eba2/A".to_owned())
8917 );
8918 }
8919
8920 #[test]
8921 fn choose_starter_falls_back_to_start_when_the_review_branch_is_gone() {
8922 assert_eq!(
8923 choose_starter(Some("magi/eba2/A"), false, Some("some-run")),
8924 Starter::Start,
8925 "a vanished review branch must not fall back to resuming the old run either"
8926 );
8927 }
8928
8929 #[test]
8930 fn a_refused_handover_retries_as_a_review_of_the_same_branch() {
8931 let mut t = task();
8932 t.start("old-run".to_owned());
8933 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
8934 t.release();
8935 let branch = t.review_branch.take();
8936 assert_eq!(
8937 choose_starter(branch.as_deref(), true, Some("old-run")),
8938 Starter::Review("magi/eba2/A".to_owned()),
8939 "a review wins over resuming the old run"
8940 );
8941 }
8942
8943 #[test]
8944 fn choose_starter_resumes_or_starts_when_there_is_no_review_choice_at_all() {
8945 assert_eq!(
8946 choose_starter(None, false, Some("some-run")),
8947 Starter::Resume("some-run".to_owned())
8948 );
8949 assert_eq!(choose_starter(None, false, None), Starter::Start);
8950 }
8951
8952 #[test]
8953 fn an_explicit_release_forces_a_fresh_competition_even_with_a_resumable_run() {
8954 let mut released = task();
8955 released.start("stalled-run".to_owned());
8956 released.requeue();
8957 let unfinished = (!released.fresh_start)
8958 .then(|| Some("stalled-run".to_owned()))
8959 .flatten();
8960 assert_eq!(
8961 choose_starter(None, false, unfinished.as_deref()),
8962 Starter::Start,
8963 "release keeps run history but must not resume it"
8964 );
8965 assert_eq!(released.runs, ["stalled-run"]);
8966 }
8967
8968 #[test]
8969 fn an_ordinary_release_keeps_a_resumable_run_available() {
8970 let mut released = task();
8971 released.start("stalled-run".to_owned());
8972 released.release();
8973 let unfinished = (!released.fresh_start)
8974 .then(|| Some("stalled-run".to_owned()))
8975 .flatten();
8976 assert_eq!(
8977 choose_starter(None, false, unfinished.as_deref()),
8978 Starter::Resume("stalled-run".to_owned()),
8979 "manual release must preserve the normal resume path"
8980 );
8981 }
8982
8983 #[test]
8984 fn a_blocked_run_that_spent_every_review_round_has_exhausted_its_budget() {
8985 let mut state = run_state(RunStatus::Blocked);
8986 state.config.graph.review_rounds = 3;
8987 state.reviews = vec![review_round(1), review_round(2), review_round(3)];
8988 assert!(exhausted_review_budget(&state));
8989
8990 state.reviews.pop();
8992 assert!(!exhausted_review_budget(&state));
8993
8994 let mut stalled = run_state(RunStatus::Stalled);
8997 stalled.config.graph.review_rounds = 1;
8998 stalled.reviews = vec![review_round(1)];
8999 assert!(!exhausted_review_budget(&stalled));
9000 }
9001
9002 fn review_round(round: usize) -> crate::run::ReviewRound {
9003 crate::run::ReviewRound {
9004 round,
9005 head: "deadbeef".to_owned(),
9006 verified_head: None,
9007 verified_at: None,
9008 reviews: Vec::new(),
9009 e2e: Vec::new(),
9010 verify_retried: false,
9011 e2e_deferred: false,
9012 e2e_defer_reason: None,
9013 fix: None,
9014 blocking: 0,
9015 answered: 1,
9016 expected: 1,
9017 clean: false,
9018 progressed: true,
9019 vote_split: false,
9020 reconsideration: Vec::new(),
9021 verdict: None,
9022 }
9023 }
9024
9025 #[test]
9026 fn awaiting_resume_is_a_failed_task_whose_last_run_parked_and_can_be_resumed() {
9027 let mut run = RunState::new(
9028 PathBuf::from("/repo"),
9029 "main".to_owned(),
9030 "abc1234def".to_owned(),
9031 "add retries".to_owned(),
9032 Config::default(),
9033 );
9034 run.status = RunStatus::Judging;
9035 run.parked = true;
9036 let mut task = Task::new(
9037 "add retries".to_owned(),
9038 "add retries".to_owned(),
9039 PathBuf::from("/repo"),
9040 crate::queue::Source::Human,
9041 );
9042 task.status = TaskStatus::Failed;
9043 task.runs = vec![run.id.clone()];
9044 let with = |t: &Task, r: &RunState| awaiting_resume_with(t, |_| Ok(r.clone()));
9045 assert!(with(&task, &run), "parked after judging is the case");
9046 let mut parked_task = task.clone();
9047 parked_task.status = TaskStatus::Parked;
9048 assert!(with(&parked_task, &run), "the Parked status is the case");
9049
9050 let mut not_parked = run.clone();
9051 not_parked.parked = false;
9052 not_parked.status = RunStatus::Stalled;
9053 assert!(!with(&task, ¬_parked), "a stall is the conductor's");
9054
9055 let mut fresh = task.clone();
9056 fresh.fresh_start = true;
9057 assert!(!with(&fresh, &run), "a requeue asked for a new competition");
9058
9059 let mut review = task.clone();
9060 review.review_branch = Some("magi/x/A".to_owned());
9061 assert!(!with(&review, &run), "review is ranked before resume");
9062
9063 let mut held = task.clone();
9064 held.status = TaskStatus::Held;
9065 assert!(!with(&held, &run), "a hold stays visible to the conductor");
9066
9067 let mut released = run.clone();
9068 released.released_to = Some("20260901-000000-new1".to_owned());
9069 assert!(!with(&task, &released), "nothing left to resume into");
9070
9071 assert!(!awaiting_resume_with(&task, |_| anyhow::bail!(
9072 "unreadable"
9073 )));
9074 }
9075
9076 #[test]
9077 fn unfinished_run_never_offers_a_run_whose_worktree_was_released() {
9078 let mut released = RunState::new(
9079 PathBuf::from("/repo"),
9080 "main".to_owned(),
9081 "abc1234def".to_owned(),
9082 "add retries".to_owned(),
9083 Config::default(),
9084 );
9085 released.status = RunStatus::Blocked;
9086 assert_eq!(
9087 unfinished_run_with(&[released.id.clone()], "t", |_| Ok(released.clone())),
9088 Some(released.id.clone())
9089 );
9090 released.released_to = Some("20260901-000000-new1".to_owned());
9091 assert_eq!(
9092 unfinished_run_with(&[released.id.clone()], "t", |_| Ok(released.clone())),
9093 None,
9094 "there is nothing left to resume it into"
9095 );
9096 }
9097
9098 #[test]
9099 fn unfinished_run_skips_a_round_exhausted_blocked_run_so_requeue_means_a_fresh_competition() {
9100 let mut exhausted = RunState::new(
9109 PathBuf::from("/repo"),
9110 "main".to_owned(),
9111 "abc1234def".to_owned(),
9112 "add retries".to_owned(),
9113 Config::default(),
9114 );
9115 exhausted.status = RunStatus::Blocked;
9116 exhausted.config.graph.review_rounds = 1;
9117 exhausted.reviews = vec![review_round(1)];
9118
9119 assert_eq!(
9120 unfinished_run_with(&[exhausted.id.clone()], "t", |_| Ok(exhausted.clone())),
9121 None,
9122 "an exhausted `Blocked` run must not be offered as resumable"
9123 );
9124
9125 let mut has_budget_left = RunState::new(
9128 PathBuf::from("/repo"),
9129 "main".to_owned(),
9130 "abc1234def".to_owned(),
9131 "add retries".to_owned(),
9132 Config::default(),
9133 );
9134 has_budget_left.status = RunStatus::Blocked;
9135 has_budget_left.config.graph.review_rounds = 3;
9136 has_budget_left.reviews = vec![review_round(1)];
9137
9138 assert_eq!(
9139 unfinished_run_with(&[has_budget_left.id.clone()], "t", |_| {
9140 Ok(has_budget_left.clone())
9141 }),
9142 Some(has_budget_left.id.clone())
9143 );
9144 }
9145
9146 #[test]
9147 fn unfinished_run_resumes_a_run_whose_task_head_publication_is_pending() {
9148 let mut run = RunState::new(
9149 PathBuf::from("/repo"),
9150 "main".to_owned(),
9151 "abc1234def".to_owned(),
9152 "add retries".to_owned(),
9153 Config::default(),
9154 );
9155 run.status = RunStatus::Blocked;
9156 run.config.graph.review_rounds = 1;
9157 run.reviews = vec![review_round(1)];
9158 run.pr_publication_pending = true;
9159 assert_eq!(
9160 unfinished_run_with(&[run.id.clone()], "t", |_| Ok(run.clone())),
9161 Some(run.id.clone()),
9162 "a spent review budget must not turn a failed publication into a re-competition"
9163 );
9164 }
9165
9166 #[test]
9167 fn unfinished_run_never_falls_back_to_an_older_resumable_run() {
9168 let mut older_stalled = RunState::new(
9176 PathBuf::from("/repo"),
9177 "main".to_owned(),
9178 "abc1234def".to_owned(),
9179 "add retries".to_owned(),
9180 Config::default(),
9181 );
9182 older_stalled.status = RunStatus::Stalled;
9183
9184 let mut newest_exhausted = RunState::new(
9185 PathBuf::from("/repo"),
9186 "main".to_owned(),
9187 "abc1234def".to_owned(),
9188 "add retries".to_owned(),
9189 Config::default(),
9190 );
9191 newest_exhausted.status = RunStatus::Blocked;
9192 newest_exhausted.config.graph.review_rounds = 1;
9193 newest_exhausted.reviews = vec![review_round(1)];
9194
9195 assert_eq!(
9196 unfinished_run_with(
9197 &[older_stalled.id.clone(), newest_exhausted.id.clone()],
9198 "t",
9199 |_| Ok(newest_exhausted.clone())
9200 ),
9201 None,
9202 "the newest run is exhausted, so nothing here is worth resuming - \
9203 least of all the older, already-superseded run"
9204 );
9205 }
9206
9207 #[test]
9208 fn unfinished_run_warns_and_skips_a_run_it_cannot_read() {
9209 assert_eq!(
9210 unfinished_run_with(&["20260101-000000-gone".to_owned()], "t", |_| {
9211 Err(anyhow::anyhow!("fixture is absent"))
9212 }),
9213 None
9214 );
9215 }
9216
9217 fn action_question(run: &str, action: ask::ChoiceAction) -> ask::Question {
9218 let mut q = ask::Question::new(
9219 run.to_owned(),
9220 "implement".to_owned(),
9221 "impl-A".to_owned(),
9222 "continue?".to_owned(),
9223 String::new(),
9224 vec!["resume で続行する".to_owned(), "other".to_owned()],
9225 );
9226 q.actions.insert("resume で続行する".to_owned(), action);
9227 q.answer(ask::Answer::Choice("resume で続行する".to_owned()))
9228 .unwrap();
9229 q
9230 }
9231
9232 fn held_task_with(run: &str) -> Task {
9233 let mut t = task();
9234 t.runs = vec![run.to_owned()];
9235 t.hold_machine(Some("waiting for magi resume to be executed".to_owned()));
9236 t
9237 }
9238
9239 fn resume_action(run: &str) -> ask::ChoiceAction {
9240 ask::ChoiceAction::Resume { run: run.into() }
9241 }
9242
9243 #[test]
9244 fn explicitly_resumed_older_run_remains_the_release_and_question_target() {
9245 for parked in [false, true] {
9246 let mut t = task();
9247 t.start("A".into());
9248 t.start("B".into());
9249 assert_eq!(t.runs, vec!["A", "B"]);
9250 t.start("A".into());
9251 let mut resumed = run_state(RunStatus::Blocked);
9252 resumed.id = "A".into();
9253 resumed.parked = parked;
9254 resumed.driver_exited = true;
9255 t.hold_machine(Some("resumed run stopped".into()));
9256 let q = action_question("A", resume_action("A"));
9257 assert_eq!(action_standing(&t, &q), ActionStanding::Pending);
9258 assert_eq!(
9259 decide_action(&t, &q, &PHRASES_EN, |id| {
9260 assert_eq!(id, "A");
9261 Ok(resumed.clone())
9262 }),
9263 ActionDecision::Resume("A".into())
9264 );
9265 assert_eq!(
9266 action_standing(&t, &action_question("B", resume_action("B"))),
9267 ActionStanding::Stale
9268 );
9269 t.release_operator("cli", None);
9270 let unfinished = unfinished_run_with(&t.runs, t.short(), |id| {
9271 assert_eq!(id, "A");
9272 Ok(resumed.clone())
9273 });
9274 assert_eq!(
9275 choose_starter(None, false, unfinished.as_deref()),
9276 Starter::Resume("A".into())
9277 );
9278 assert_eq!(t.runs, vec!["B", "A"]);
9279 }
9280 }
9281
9282 #[test]
9283 fn merge_repair_dispatch_refuses_obsolete_bindings_and_unsupported_owners() {
9284 let mut state = crate::land::tests::repair_state(true, true);
9285 state.status = RunStatus::Landing;
9286 state.parked = true;
9287 state.driver_exited = true;
9288 let mut q = ask::Question::new(
9289 state.id.clone(),
9290 land::APPROVAL_NODE.into(),
9291 "land".into(),
9292 "Merge?".into(),
9293 String::new(),
9294 vec![land::APPROVE.into(), land::HOLD.into()],
9295 );
9296 q.id = state.id.clone();
9297 state.land_approval = Some(crate::run::LandApproval {
9298 question: q.id.clone(),
9299 head: "current-head".into(),
9300 });
9301 assert!(land::upgrade_repair_choice(&mut q, &state));
9302 q.answer(ask::Answer::Choice(land::REPAIR.into())).unwrap();
9303 let mut task = held_task_with(&state.id);
9304 task.status = TaskStatus::Parked;
9305 assert_eq!(
9306 decide_action(&task, &q, &PHRASES_EN, |_| Ok(state.clone())),
9307 ActionDecision::Resume(state.id.clone())
9308 );
9309 q.asked_at = "2026-01-01T00:00:00Z".parse().unwrap();
9310 q.answered_at = Some("2026-01-01T00:00:20Z".parse().unwrap());
9311 task.operator_decided_at = Some("2026-01-01T00:00:10Z".parse().unwrap());
9312 assert_eq!(
9313 decide_action(&task, &q, &PHRASES_EN, |_| Ok(state.clone())),
9314 ActionDecision::Resume(state.id.clone()),
9315 "a repair answer newer than the manual hold resumes the existing owner"
9316 );
9317 task.operator_decided_at = Some("2026-01-01T00:00:30Z".parse().unwrap());
9318 for answered_at in [q.answered_at, None] {
9319 let mut older = q.clone();
9320 older.answered_at = answered_at;
9321 let refused = decide_action(&task, &older, &PHRASES_EN, |_| {
9322 panic!("a later manual hold must prevent dispatch")
9323 });
9324 let ActionDecision::Refuse(reason) = refused else {
9325 panic!("expected an actionable refusal, got {refused:?}");
9326 };
9327 assert!(reason.contains("later operator decision"));
9328 assert!(reason.contains(&format!("magi task release {}", task.id)));
9329 }
9330 task.operator_decided_at = None;
9331 let mut superseded = state.clone();
9332 superseded.land_approval.as_mut().unwrap().question = "fresh-decision".into();
9333 assert_eq!(
9334 decide_action(&task, &q, &PHRASES_EN, |_| Ok(superseded)),
9335 ActionDecision::Stale
9336 );
9337 for case in 1..4 {
9338 let mut changed = state.clone();
9339 match case {
9340 1 => changed.status = RunStatus::Blocked,
9341 2 => changed.released_to = Some("new-owner".into()),
9342 _ => changed.land_approval.as_mut().unwrap().head.clear(),
9343 }
9344 assert!(matches!(
9345 decide_action(&task, &q, &PHRASES_EN, |_| Ok(changed)),
9346 ActionDecision::Refuse(_)
9347 ));
9348 }
9349 assert!(matches!(
9350 decide_action(&task, &q, &PHRASES_EN, |_| anyhow::bail!("missing owner")),
9351 ActionDecision::Refuse(_)
9352 ));
9353 task.runs.push("replacement".into());
9354 task.operator_decided_at = Some("2026-01-01T00:00:30Z".parse().unwrap());
9355 assert_eq!(
9356 decide_action(&task, &q, &PHRASES_EN, |_| panic!(
9357 "old owner must not load"
9358 )),
9359 ActionDecision::Stale
9360 );
9361 }
9362
9363 #[test]
9364 fn merge_repair_dispatch_obeys_capacity_and_quota_while_hold_stays_decision_only() {
9365 let mut state = crate::land::tests::repair_state(true, true);
9366 state.status = RunStatus::Landing;
9367 state.parked = true;
9368 state.driver_exited = true;
9369 let mut q = ask::Question::new(
9370 state.id.clone(),
9371 land::APPROVAL_NODE.into(),
9372 "land".into(),
9373 "Merge?".into(),
9374 String::new(),
9375 vec![land::APPROVE.into(), land::HOLD.into()],
9376 );
9377 q.id = state.id.clone();
9378 state.land_approval = Some(crate::run::LandApproval {
9379 question: q.id.clone(),
9380 head: "current-head".into(),
9381 });
9382 state.save().unwrap();
9383 land::upgrade_repair_choice(&mut q, &state);
9384 q.answer(ask::Answer::Choice(land::REPAIR.into())).unwrap();
9385 Questions::open().put(&mut q).unwrap();
9386 let tmp = tempfile::tempdir().unwrap();
9387 let queue = Queue::at(tmp.path().join("queue"));
9388 let mut task = held_task_with(&state.id);
9389 task.status = TaskStatus::Parked;
9390 queue.put(&mut task).unwrap();
9391 let empty = Arc::new(tokio::sync::Semaphore::new(0));
9392 let free = Arc::new(tokio::sync::Semaphore::new(1));
9393 assert!(claim_dispatch(&queue, &task, false, &empty, &free).is_none());
9394 assert!(claim_dispatch(&queue, &task, true, &free, &free).is_none());
9395 let claim = claim_dispatch(&queue, &task, false, &free, &free).unwrap();
9396 assert!(claim.permit.is_some());
9397 assert!(claim_dispatch(&queue, &task, false, &free, &free).is_none());
9398 drop(claim);
9399 Questions::open()
9400 .update(&q.id, |q| {
9401 q.answer = Some(ask::Answer::Choice(land::HOLD.into()));
9402 Ok(())
9403 })
9404 .unwrap();
9405 let claim = claim_dispatch(&queue, &task, true, &empty, &empty).unwrap();
9406 assert!(claim.permit.is_none());
9407 assert!(RunState::load(&state.id).unwrap().merge_repairs.is_empty());
9408 }
9409
9410 #[test]
9411 fn decide_action_resumes_only_the_latest_resumable_run() {
9412 let t = held_task_with("r1");
9413 let q = action_question("r1", resume_action("r1"));
9414 let load = |s: RunState| move |_: &str| Ok(s);
9415 assert_eq!(
9416 decide_action(&t, &q, &PHRASES_EN, load(run_state(RunStatus::Blocked))),
9417 ActionDecision::Resume("r1".into())
9418 );
9419 let q_other = action_question("r1", resume_action("r0"));
9421 assert!(matches!(
9422 decide_action(
9423 &t,
9424 &q_other,
9425 &PHRASES_EN,
9426 load(run_state(RunStatus::Blocked))
9427 ),
9428 ActionDecision::Refuse(_)
9429 ));
9430 assert!(matches!(
9432 decide_action(&t, &q, &PHRASES_EN, load(run_state(RunStatus::Ready))),
9433 ActionDecision::Refuse(_)
9434 ));
9435 let mut released = run_state(RunStatus::Blocked);
9437 released.released_to = Some("elsewhere".into());
9438 assert!(matches!(
9439 decide_action(&t, &q, &PHRASES_EN, load(released)),
9440 ActionDecision::Refuse(_)
9441 ));
9442 assert!(matches!(
9444 decide_action(&t, &q, &PHRASES_EN, |_: &str| bail!("gone")),
9445 ActionDecision::Refuse(_)
9446 ));
9447 }
9448
9449 #[test]
9450 fn decide_action_ignores_a_question_about_an_earlier_run() {
9451 let mut t = held_task_with("r1");
9452 t.runs.push("r2".to_owned());
9453 let never = |_: &str| -> Result<RunState> { bail!("not read") };
9454 assert_eq!(
9455 decide_action(
9456 &t,
9457 &action_question("r1", ask::ChoiceAction::Done),
9458 &PHRASES_EN,
9459 never
9460 ),
9461 ActionDecision::Stale
9462 );
9463 }
9464
9465 #[test]
9466 fn decide_action_maps_requeue_and_done_and_never_acts_twice() {
9467 let mut t = held_task_with("r1");
9468 let never = |_: &str| -> Result<RunState> { bail!("not read") };
9469 assert_eq!(
9470 decide_action(
9471 &t,
9472 &action_question("r1", ask::ChoiceAction::Requeue),
9473 &PHRASES_EN,
9474 never
9475 ),
9476 ActionDecision::Requeue
9477 );
9478 let done_q = action_question("r1", ask::ChoiceAction::Done);
9479 assert_eq!(
9480 decide_action(&t, &done_q, &PHRASES_EN, never),
9481 ActionDecision::Done
9482 );
9483 t.mark_action_applied(&done_q.id);
9484 assert_eq!(
9485 decide_action(&t, &done_q, &PHRASES_EN, never),
9486 ActionDecision::Skip
9487 );
9488
9489 let mut plain = action_question("r1", ask::ChoiceAction::Done);
9491 plain.actions.clear();
9492 assert_eq!(
9493 decide_action(&held_task_with("r1"), &plain, &PHRASES_EN, never),
9494 ActionDecision::Skip
9495 );
9496 let mut running = held_task_with("r1");
9498 running.status = TaskStatus::Running;
9499 assert_eq!(
9500 decide_action(
9501 &running,
9502 &action_question("r1", ask::ChoiceAction::Done),
9503 &PHRASES_EN,
9504 never
9505 ),
9506 ActionDecision::Skip
9507 );
9508 }
9509
9510 #[test]
9511 fn merge_repair_rejected_after_manual_hold_keeps_the_conductor_guard() {
9512 crate::run::pin_test_home();
9513 for legacy in [false, true] {
9514 let dir = tempfile::tempdir().unwrap();
9515 let queue = Queue::at(dir.path().join("queue"));
9516 let questions = Questions::at(dir.path().join("questions"));
9517 let run = "20260101-000000-hold";
9518 let mut task = held_task_with(run);
9519 task.status = TaskStatus::Parked;
9520 let mut q = ask::Question::new(
9521 run.into(),
9522 land::APPROVAL_NODE.into(),
9523 "land".into(),
9524 "Repair?".into(),
9525 String::new(),
9526 vec![land::REPAIR.into()],
9527 );
9528 q.actions.insert(land::REPAIR.into(), resume_action(run));
9529 q.answer(ask::Answer::Choice(land::REPAIR.into())).unwrap();
9530 q.asked_at = "2026-01-01T00:00:00Z".parse().unwrap();
9532 q.answered_at = Some("2026-01-01T00:00:01Z".parse().unwrap());
9533 questions.put(&mut q).unwrap();
9534 task.hold_manual(Some("Keep this PR paused".into()));
9535 if legacy {
9536 task.hold_source = None;
9537 }
9538 queue.put(&mut task).unwrap();
9539
9540 apply_choice_actions(&queue, &questions, dir.path());
9541 let after = queue.get(&task.id).unwrap();
9542 assert!(after.operator_held());
9543 assert_eq!(after.hold_source, task.hold_source);
9544 assert_eq!(after.operator_decided_at, task.operator_decided_at);
9545 assert_eq!(after.runs, task.runs);
9546 assert_eq!(after.instruction, task.instruction);
9547 assert!(after.resume_override.is_none());
9548 assert!(after.action_applied(&q.id));
9549 let reason = after.hold_reason.as_deref().unwrap();
9550 assert!(reason.starts_with("Keep this PR paused"));
9551 assert!(reason.contains("later operator decision"));
9552 assert!(reason.contains(&format!("magi task release {}", task.id)));
9553
9554 for recovery in [
9555 crate::conduct::Recovery::Review,
9556 crate::conduct::Recovery::Requeue,
9557 ] {
9558 crate::conduct::apply(
9559 &queue,
9560 &questions,
9561 &crate::conduct::Verdict {
9562 decisions: vec![crate::conduct::Decision {
9563 id: task.id.clone(),
9564 recovery: Some(recovery),
9565 ..Default::default()
9566 }],
9567 },
9568 )
9569 .unwrap();
9570 }
9571 apply_choice_actions(&queue, &questions, dir.path());
9574 let still_held = queue.get(&task.id).unwrap();
9575 assert!(still_held.operator_held());
9576 assert_eq!(still_held.hold_reason, after.hold_reason);
9577 assert_eq!(still_held.runs, task.runs);
9578 assert!(still_held.review_branch.is_none());
9579 assert!(!still_held.fresh_start);
9580 assert!(queue.next_runnable().is_none());
9581 }
9582 }
9583
9584 #[test]
9585 fn apply_choice_actions_releases_the_task_pinned_to_its_run_and_only_once() {
9586 let dir = tempfile::tempdir().unwrap();
9587 let queue = Queue::at(dir.path().join("queue"));
9588 let questions = Questions::at(dir.path().join("questions"));
9589 let home = dir.path().join("home");
9590 let mut state = run_state(RunStatus::Blocked);
9591 state.id = "20260101-000000-act1".to_owned();
9592 state.save_under(&home).unwrap();
9593
9594 let mut t = held_task_with(&state.id);
9595 queue.put(&mut t).unwrap();
9596 let mut q = action_question(&state.id, resume_action(&state.id));
9597 questions.put(&mut q).unwrap();
9598
9599 apply_choice_actions(&queue, &questions, &home);
9600 let after = queue.get(&t.id).unwrap();
9601 assert_eq!(after.status, TaskStatus::Queued);
9602 assert!(!after.fresh_start);
9603 assert!(after.action_applied(&q.id));
9604 let pin = after.resume_override.clone().unwrap();
9605 assert_eq!(pin.pinned_run.as_deref(), Some(state.id.as_str()));
9606 assert!(pin.forced);
9607
9608 let mut again = queue.get(&t.id).unwrap();
9610 again.hold_machine(Some("later".into()));
9611 queue.put(&mut again).unwrap();
9612 apply_choice_actions(&queue, &questions, &home);
9613 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9614 }
9615
9616 #[test]
9617 fn a_delivered_answer_is_still_acted_on_exactly_once() {
9618 let dir = tempfile::tempdir().unwrap();
9619 let queue = Queue::at(dir.path().join("queue"));
9620 let questions = Questions::at(dir.path().join("questions"));
9621 let home = dir.path().join("home");
9622 let mut state = run_state(RunStatus::Blocked);
9623 state.id = "20260101-000000-act2".to_owned();
9624 state.save_under(&home).unwrap();
9625
9626 let mut t = held_task_with(&state.id);
9627 queue.put(&mut t).unwrap();
9628 let mut q = action_question(&state.id, resume_action(&state.id));
9629 q.answer_delivered = true;
9630 questions.put(&mut q).unwrap();
9631
9632 apply_choice_actions(&queue, &questions, &home);
9635 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
9636 assert!(queue.get(&t.id).unwrap().action_applied(&q.id));
9637
9638 let mut again = queue.get(&t.id).unwrap();
9640 again.hold_machine(Some("later".into()));
9641 queue.put(&mut again).unwrap();
9642 apply_choice_actions(&queue, &questions, &home);
9643 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9644 }
9645
9646 #[test]
9647 fn a_fresh_asker_defers_the_action_until_it_goes_quiet() {
9648 let dir = tempfile::tempdir().unwrap();
9649 let queue = Queue::at(dir.path().join("queue"));
9650 let questions = Questions::at(dir.path().join("questions"));
9651 let home = dir.path().join("home");
9652 let mut state = run_state(RunStatus::Blocked);
9653 state.id = "20260101-000000-act3".to_owned();
9654 state.save_under(&home).unwrap();
9655 let mut t = held_task_with(&state.id);
9656 queue.put(&mut t).unwrap();
9657 let mut q = action_question(&state.id, ask::ChoiceAction::Requeue);
9658 questions.put(&mut q).unwrap();
9659
9660 questions.beat(&q.id, crate::ask::WaiterKind::Asker);
9661 apply_choice_actions(&queue, &questions, &home);
9662 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9663
9664 std::fs::remove_file(questions.root().join(format!("{}.lease", q.id))).unwrap();
9665 apply_choice_actions(&queue, &questions, &home);
9666 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
9667 }
9668
9669 #[test]
9670 fn action_standing_tells_the_reasons_apart() {
9671 let mut t = held_task_with("r1");
9672 let q = action_question("r1", ask::ChoiceAction::Requeue);
9673 assert_eq!(action_standing(&t, &q), ActionStanding::Pending);
9674 let mut plain = q.clone();
9675 plain.actions.clear();
9676 assert_eq!(action_standing(&t, &plain), ActionStanding::NoAction);
9677
9678 t.mark_action_applied(&q.id);
9680 assert_eq!(action_standing(&t, &q), ActionStanding::Applied);
9681
9682 let mut t = held_task_with("r1");
9684 t.start("r2".to_owned());
9685 assert_eq!(action_standing(&t, &q), ActionStanding::Stale);
9686 let q2 = action_question("r2", ask::ChoiceAction::Requeue);
9688 assert_eq!(action_standing(&t, &q2), ActionStanding::Busy);
9689 t.status = TaskStatus::Blocked;
9690 assert_eq!(action_standing(&t, &q2), ActionStanding::Busy);
9691
9692 let mut c = action_question("r0", ask::ChoiceAction::Requeue);
9694 c.node = crate::conduct::NODE.to_owned();
9695 assert_eq!(action_standing(&t, &c), ActionStanding::Busy);
9696 assert!(!ActionStanding::Busy.daemon_owns());
9697 }
9698
9699 #[test]
9700 fn a_running_task_waits_and_a_dead_asker_with_a_cwd_is_still_actioned() {
9701 let dir = tempfile::tempdir().unwrap();
9702 let queue = Queue::at(dir.path().join("queue"));
9703 let questions = Questions::at(dir.path().join("questions"));
9704 let home = dir.path().join("home");
9705 let mut state = run_state(RunStatus::Blocked);
9706 state.id = "20260101-000000-act4".to_owned();
9707 state.save_under(&home).unwrap();
9708 let mut t = held_task_with(&state.id);
9709 t.status = TaskStatus::Running;
9710 queue.put(&mut t).unwrap();
9711 let mut q = action_question(&state.id, ask::ChoiceAction::Done);
9712 q.cwd = Some(dir.path().display().to_string());
9713 questions.put(&mut q).unwrap();
9714
9715 let running = queue.get(&t.id).unwrap();
9718 assert_eq!(action_standing(&running, &q), ActionStanding::Busy);
9719 assert!(matches!(
9720 crate::waiter::decide_owned(&q, None, false, 86_400, Timestamp::now(), false),
9721 crate::waiter::Action::Deliver(_)
9722 ));
9723 apply_choice_actions(&queue, &questions, &home);
9725 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Running);
9726 let mut back = queue.get(&t.id).unwrap();
9727 back.hold_machine(Some("later".into()));
9728 queue.put(&mut back).unwrap();
9729 apply_choice_actions(&queue, &questions, &home);
9730 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Done);
9731 let mut again = queue.get(&t.id).unwrap();
9732 again.hold_machine(Some("again".into()));
9733 queue.put(&mut again).unwrap();
9734 apply_choice_actions(&queue, &questions, &home);
9735 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9736 }
9737}