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, questions);
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 let crate::queue::Dependency::Question(full) =
718 crate::queue::dependency(queue, questions, id)
719 else {
720 return None;
721 };
722 questions.get(&full).ok().filter(|q| {
723 q.node == crate::conduct::NODE && q.status == ask::QuestionStatus::Abandoned
724 })
725 }) {
726 let language = language_of(&task, Path::new("."));
727 task.hold_machine(Some(unanswered_question_hold_reason(&q, &language)));
728 record(queue, &mut task);
729 continue;
730 }
731 let mut changed = false;
732 for id in task.blocked_by.clone() {
733 let resolved = crate::queue::dependency(queue, questions, &id);
734 if let crate::queue::Dependency::Task(full) = &resolved {
735 if queue
736 .get(full)
737 .is_ok_and(|dep| dep.status == TaskStatus::Done)
738 {
739 task.unblock(&id);
740 changed = true;
741 }
742 continue;
743 }
744 if let crate::queue::Dependency::Question(full) = resolved
745 && let Ok(q) = questions.get(&full)
746 && q.status == ask::QuestionStatus::Answered
747 {
748 let answer = match &q.answer {
749 Some(ask::Answer::Choice(c) | ask::Answer::Text(c)) => c.clone(),
750 None => String::new(),
751 };
752 task.record_answer(q.summary.clone(), answer);
753 task.unblock(&id);
754 changed = true;
755 }
756 }
757 if changed {
758 record(queue, &mut task);
759 }
760 }
761}
762
763fn unanswered_question_hold_reason(q: &ask::Question, language: &str) -> String {
765 if crate::lang::is_japanese(language) {
766 format!(
767 "質問 {} 「{}」 に期限内の回答がなく、取り下げられました - `magi task triage` を参照",
768 q.short(),
769 q.summary
770 )
771 } else {
772 format!(
773 "question {} \"{}\" went unanswered and was abandoned - see `magi task triage`",
774 q.short(),
775 q.summary
776 )
777 }
778}
779
780#[derive(Debug, Clone, Copy, PartialEq, Eq)]
783pub(crate) enum ActionStanding {
784 NoAction,
786 Pending,
788 Applied,
791 Stale,
793 Busy,
795}
796
797impl ActionStanding {
798 pub(crate) fn daemon_owns(self) -> bool {
800 matches!(self, Self::Pending | Self::Applied | Self::Stale)
801 }
802}
803
804pub(crate) fn action_standing(task: &Task, q: &ask::Question) -> ActionStanding {
808 if q.chosen_action().is_none() {
809 ActionStanding::NoAction
810 } else if task.action_applied(&q.id) {
811 ActionStanding::Applied
812 } else if q.node != crate::conduct::NODE && task.runs.last() != Some(&q.run) {
813 ActionStanding::Stale
814 } else if matches!(
815 task.status,
816 TaskStatus::Running | TaskStatus::Blocked | TaskStatus::Done
817 ) {
818 ActionStanding::Busy
819 } else {
820 ActionStanding::Pending
821 }
822}
823
824#[derive(Debug, Clone, PartialEq, Eq)]
826enum ActionDecision {
827 Skip,
829 Resume(String),
831 Requeue,
833 Done,
835 Stale,
838 Refuse(String),
841}
842
843fn decide_action<F>(task: &Task, q: &ask::Question, p: &Phrases, load: F) -> ActionDecision
851where
852 F: FnOnce(&str) -> Result<RunState>,
853{
854 let Some(action) = q.chosen_action() else {
855 return ActionDecision::Skip;
856 };
857 if task.action_applied(&q.id)
858 || matches!(
859 task.status,
860 TaskStatus::Running | TaskStatus::Blocked | TaskStatus::Done
861 )
862 {
863 return ActionDecision::Skip;
864 }
865 if q.node != crate::conduct::NODE && task.runs.last() != Some(&q.run) {
868 return ActionDecision::Stale;
869 }
870 let repair = q.node == land::APPROVAL_NODE && q.resolution().as_deref() == Some(land::REPAIR);
871 let decision_at = if repair {
875 q.answered_at.unwrap_or(q.asked_at)
876 } else {
877 q.asked_at
878 };
879 if task.operator_decided_at.is_some_and(|at| at > decision_at) {
880 return if repair {
881 ActionDecision::Refuse(format!(
882 "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 {}`",
883 q.id, q.run, task.id
884 ))
885 } else {
886 ActionDecision::Stale
887 };
888 }
889 match action {
890 ask::ChoiceAction::Requeue => ActionDecision::Requeue,
891 ask::ChoiceAction::Done => ActionDecision::Done,
892 ask::ChoiceAction::Resume { run } => {
893 if task.runs.last() != Some(run) {
894 return ActionDecision::Refuse((p.resume_not_latest)(
895 q.short(),
896 ask::short_id(run),
897 ));
898 }
899 match load(run) {
900 Ok(s)
901 if q.node == land::APPROVAL_NODE
902 && q.resolution().as_deref() == Some(land::REPAIR) =>
903 {
904 if s.merge_repairs.iter().any(|r| r.question == q.id)
905 || s.land_approval.as_ref().is_some_and(|a| a.question != q.id)
906 {
907 ActionDecision::Stale
910 } else if s.liveness(false) == Liveness::Live {
911 ActionDecision::Skip
912 } else if s.status != RunStatus::Landing
913 || s.released()
914 || !land::repair_available(&s)
915 || !implementers_match(task, &s)
916 || !s
917 .land_approval
918 .as_ref()
919 .is_some_and(|a| a.question == q.id && !a.head.is_empty())
920 {
921 ActionDecision::Refuse(format!(
922 "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",
923 q.id, run, run
924 ))
925 } else {
926 ActionDecision::Resume(run.clone())
927 }
928 }
929 Ok(s) if s.status.resumable() && !s.released() && !exhausted_review_budget(&s) => {
930 ActionDecision::Resume(run.clone())
931 }
932 Ok(_) => ActionDecision::Refuse((p.resume_cannot_progress)(
933 q.short(),
934 ask::short_id(run),
935 )),
936 Err(e) => ActionDecision::Refuse((p.resume_unreadable)(
937 q.short(),
938 ask::short_id(run),
939 &format!("{e:#}"),
940 )),
941 }
942 }
943 }
944}
945
946struct Phrases {
960 graph_stopped: fn(&str, &str) -> String,
962 quorum_lost: &'static str,
963 quota_took_out: &'static str,
965 run_ended: &'static str,
967 waiting_for_answer: &'static str,
969 recovered_running: &'static str,
970 no_run_to_recover: &'static str,
971 could_not_start: &'static str,
972 resume_not_latest: fn(&str, &str) -> String,
973 resume_cannot_progress: fn(&str, &str) -> String,
974 resume_unreadable: fn(&str, &str, &str) -> String,
975 handover_refused: fn(&crate::handover::Refused) -> String,
978 handover_hint: &'static str,
980 reviewers_never_answered: fn(usize, usize) -> String,
983}
984
985const PHRASES_EN: Phrases = Phrases {
986 graph_stopped: |status, detail| {
987 format!("the graph stopped at `{status}` without reaching a terminal status: {detail}")
988 },
989 quorum_lost: "the judging panel lost its quorum",
990 quota_took_out: "; quota took out ",
991 run_ended: "run ended ",
992 waiting_for_answer: " - waiting for operator answer to question ",
993 recovered_running: "recovered a `running` task whose daemon never recorded the outcome: ",
994 no_run_to_recover: "task was `running` with no live daemon and no readable \
995 run to recover; held for a human to check what happened",
996 could_not_start: "could not start the run: ",
997 resume_not_latest: |q, run| {
998 format!("question {q} asked to resume run {run}, which is not this task's latest run")
999 },
1000 resume_cannot_progress: |q, run| {
1001 format!("question {q} asked to resume run {run}, which cannot make progress")
1002 },
1003 resume_unreadable: |q, run, e| {
1004 format!("question {q} asked to resume run {run}, which could not be read: {e}")
1005 },
1006 handover_refused: |r| r.to_string(),
1007 handover_hint: " (clean up the other worktree, then release the task from the queue)",
1008 reviewers_never_answered: |missing, rounds| {
1009 format!(
1010 "{missing} reviewer seat(s) never answered after {rounds} rounds; \
1011 refusing to call it clean"
1012 )
1013 },
1014};
1015
1016const PHRASES_JA: Phrases = Phrases {
1017 graph_stopped: |status, detail| {
1018 format!("グラフが終端状態に達しないまま `{status}` で停止しました: {detail}")
1019 },
1020 quorum_lost: "審査パネルが定足数を失いました",
1021 quota_took_out: "。クォータで脱落: ",
1022 run_ended: "run 終了: ",
1023 waiting_for_answer: " - オペレーターの回答待ち: 質問 ",
1024 recovered_running: "daemon が結果を記録しないまま `running` だったタスクを回収しました: ",
1025 no_run_to_recover: "タスクは `running` でしたが、生きた daemon も回収できる run も見つかりません。\
1026 何が起きたか人が確認するため保留にしました",
1027 could_not_start: "run を開始できませんでした: ",
1028 resume_not_latest: |q, run| {
1029 format!(
1030 "質問 {q} は run {run} の再開を求めましたが、これはタスクの最新の run ではありません"
1031 )
1032 },
1033 resume_cannot_progress: |q, run| {
1034 format!("質問 {q} は run {run} の再開を求めましたが、これは進行できません")
1035 },
1036 resume_unreadable: |q, run, e| {
1037 format!("質問 {q} は run {run} の再開を求めましたが、読み込めませんでした: {e}")
1038 },
1039 handover_refused: |r| {
1040 use crate::handover::Refused;
1041 match r {
1042 Refused::Foreign { branch, path, why } => format!(
1043 "ブランチ `{branch}` は {path} にチェックアウトされており、magi は自動では\
1044 削除しません。不要なら `git worktree remove` でその worktree を削除して\
1045 から、やり直してください(詳細: {why})"
1046 ),
1047 Refused::Unsafe { branch, path, why } => format!(
1048 "ブランチ `{branch}` は {path} にチェックアウトされています。そこの作業を\
1049 コミットか破棄したうえで `git worktree remove` で worktree を削除する\
1050 か、run を破棄してよいと伝えてから、やり直してください(詳細: {why})"
1051 ),
1052 Refused::ReleaseFailed { branch, path, run } => format!(
1053 "ブランチ `{branch}` は {path} で run {run} がチェックアウトしており、その\
1054 worktree の解放に失敗したか、変更が見つかりました(未コミットの変更が\
1055 ある worktree は git が削除を拒否します)。worktree はそのまま残しました"
1056 ),
1057 }
1058 },
1059 handover_hint: "(他の worktree を片付けてから、タスクをキューから解放してください)",
1060 reviewers_never_answered: |missing, rounds| {
1061 format!(
1062 "{missing} 席のレビュアーが {rounds} ラウンドの間に一度も回答しなかったため、\
1063 クリーンとは認めません"
1064 )
1065 },
1066};
1067
1068fn phrases(language: &str) -> &'static Phrases {
1069 if crate::lang::is_japanese(language) {
1070 &PHRASES_JA
1071 } else {
1072 &PHRASES_EN
1073 }
1074}
1075
1076fn language_of(task: &Task, fallback: &Path) -> String {
1080 crate::lang::of_repo(&repo_for(task, fallback))
1081}
1082
1083pub(crate) fn task_of_question<'a>(tasks: &'a [Task], q: &ask::Question) -> Option<&'a Task> {
1087 if q.node == crate::conduct::NODE {
1088 return tasks.iter().find(|t| t.id == q.run);
1089 }
1090 tasks.iter().find(|t| t.runs.contains(&q.run))
1091}
1092
1093pub(crate) fn apply_choice_actions(queue: &Queue, questions: &Questions, home: &Path) {
1102 let tasks = queue.list();
1103 for q in questions.list() {
1104 if q.chosen_action().is_none() {
1105 continue;
1106 }
1107 let Some(listed) = task_of_question(&tasks, &q) else {
1108 if q.node == land::APPROVAL_NODE && q.resolution().as_deref() == Some(land::REPAIR) {
1109 notices::raise_in(
1110 home,
1111 Notice::warn(
1112 &format!("repair-owner:{}", q.id),
1113 format!(
1114 "Repair is waiting: no task owns run {}. Inspect `magi show {}` and restore its original queue owner; do not file a competing repair task.",
1115 q.run, q.run
1116 ),
1117 ),
1118 );
1119 }
1120 continue;
1121 };
1122 if listed.action_applied(&q.id) {
1123 continue;
1124 }
1125 let Ok(_claim) = queue.claim(&listed.id) else {
1126 continue;
1127 };
1128 let Ok(mut task) = queue.get(&listed.id) else {
1129 continue;
1130 };
1131 let language = language_of(&task, Path::new("."));
1132 let decision = decide_action(&task, &q, phrases(&language), |id| {
1133 RunState::load_under(id, home)
1134 });
1135 let ran = matches!(
1136 decision,
1137 ActionDecision::Resume(_) | ActionDecision::Requeue | ActionDecision::Done
1138 );
1139 if ran && asker_may_still_read(questions, &q, home) {
1140 continue;
1149 }
1150 let done = matches!(decision, ActionDecision::Done);
1151 match decision {
1152 ActionDecision::Skip => continue,
1153 ActionDecision::Resume(run) => {
1154 task.release_operator("choice", Some(&q.id));
1155 task.resume_override = Some(crate::queue::OperatorResume {
1156 question_id: q.id.clone(),
1157 at: Timestamp::now(),
1158 conductor_rehold: None,
1159 forced: true,
1160 pinned_run: Some(run),
1161 });
1162 }
1163 ActionDecision::Stale => {}
1164 ActionDecision::Requeue => {
1165 task.release_operator("choice-requeue", Some(&q.id));
1166 task.requeue();
1167 }
1168 ActionDecision::Done => task.succeed(),
1169 ActionDecision::Refuse(why) => {
1170 if task.operator_held() {
1171 task.hold_reason = Some(match task.hold_reason.take() {
1174 Some(reason) => format!("{reason}\n\n{why}"),
1175 None => why,
1176 });
1177 } else {
1178 task.hold_machine(Some(why));
1179 }
1180 notices::raise(
1181 Notice::warn(
1182 &format!("action:{}", q.id),
1183 "An answer asked the daemon to resume a run that cannot be resumed; the task stays held.",
1184 )
1185 .link(Link::Task {
1186 id: task.id.clone(),
1187 }),
1188 );
1189 }
1190 }
1191 task.mark_action_applied(&q.id);
1192 record(queue, &mut task);
1195 if done && queue.get(&task.id).is_ok_and(|t| t.action_applied(&q.id)) {
1196 supersede_prior_runs(&task, home);
1197 }
1198 let _ = questions.update(&q.id, |r| {
1201 r.answer_delivered = true;
1202 Ok(())
1203 });
1204 }
1205}
1206
1207fn asker_may_still_read(questions: &Questions, q: &ask::Question, home: &Path) -> bool {
1211 let now = Timestamp::now();
1212 if questions.read_lease(&q.id).is_some_and(|l| l.fresh(now)) {
1213 return true;
1214 }
1215 q.node != crate::conduct::NODE
1216 && RunState::load_under(&q.run, home).is_ok_and(|s| {
1217 s.seats_active()
1218 .any(|(k, a)| *k == q.seat && a.remaining_secs(now) > 0)
1219 })
1220}
1221
1222fn reconcile_task_questions(queue: &Queue, questions: &Questions) {
1233 let tasks = queue.list();
1234 let by_id: std::collections::BTreeMap<&str, &Task> =
1235 tasks.iter().map(|t| (t.id.as_str(), t)).collect();
1236 let referenced: Vec<&str> = tasks
1237 .iter()
1238 .flat_map(|task| task.blocked_by.iter().map(String::as_str))
1239 .collect();
1240
1241 for mut question in questions.list() {
1242 if !question.status.open() || question.node != crate::conduct::NODE {
1243 continue;
1244 }
1245 if referenced
1250 .iter()
1251 .any(|id| question.id.starts_with(id) || question.id.ends_with(id))
1252 {
1253 continue;
1254 }
1255 let Some(task) = by_id.get(question.run.as_str()) else {
1256 continue;
1257 };
1258 question.abandon(format!(
1259 "task {} no longer waits for this answer",
1260 task.short()
1261 ));
1262 if let Err(e) = questions.put(&mut question) {
1263 tracing::warn!(
1264 "could not retire question {} for task {}: {e:#}",
1265 question.short(),
1266 task.short()
1267 );
1268 }
1269 }
1270}
1271
1272#[derive(Debug, Clone, Copy)]
1279pub struct Verdict {
1280 pub status: RunStatus,
1282 pub left_pr: bool,
1284 pub quota_hit: bool,
1286 pub parked: bool,
1288 pub no_viable_candidates: bool,
1295}
1296
1297pub fn settle(task: &mut Task, verdict: Verdict, detail: &str, max_attempts: usize) {
1354 settle_in(task, verdict, detail, max_attempts, &PHRASES_EN)
1355}
1356
1357fn settle_in(task: &mut Task, verdict: Verdict, detail: &str, max_attempts: usize, p: &Phrases) {
1359 if verdict.parked {
1365 task.park(detail);
1366 return;
1367 }
1368 match verdict.status {
1369 RunStatus::Merged | RunStatus::Ready => task.succeed(),
1370 RunStatus::AlreadyInBase => task.already_landed(detail),
1371 RunStatus::Stalled if verdict.quota_hit => task.stall(detail),
1372 RunStatus::Failed if verdict.quota_hit && verdict.no_viable_candidates => {
1373 task.stall(detail)
1374 }
1375 RunStatus::Stalled | RunStatus::Failed => task.fail(detail, max_attempts),
1376 RunStatus::Blocked if verdict.left_pr => task.handed_off(detail),
1377 RunStatus::Blocked => task.fail(detail, max_attempts),
1378 RunStatus::VerifiedNoop => task.handed_off(detail),
1379 other => task.fail((p.graph_stopped)(label(other), detail), max_attempts),
1380 }
1381}
1382
1383pub fn supersede_prior_runs(task: &Task, home: &Path) {
1432 let now = Timestamp::now();
1433 let last_run_succeeded = task
1440 .runs
1441 .last()
1442 .and_then(|id| RunState::load_under(id, home).ok())
1443 .is_some_and(|s| matches!(s.status, RunStatus::Merged | RunStatus::Ready));
1444 for id in task.superseded_attempts(last_run_succeeded) {
1445 let mut state = match RunState::load_under(id, home) {
1446 Ok(s) => s,
1447 Err(e) => {
1448 tracing::warn!("could not load run {id} to mark it superseded: {e:#}");
1449 continue;
1450 }
1451 };
1452 if !matches!(state.status, RunStatus::Blocked | RunStatus::Stalled) {
1453 continue;
1454 }
1455 let daemon_claims = is_working_on(home, id, now);
1456 if state.liveness(daemon_claims) == Liveness::Live {
1457 continue;
1458 }
1459 state.status = RunStatus::Superseded;
1460 if let Err(e) = state.save_under(home) {
1461 tracing::warn!("could not mark run {id} superseded: {e:#}");
1462 }
1463 }
1464}
1465
1466fn resweep_superseded_attempts(queue: &Queue, home: &Path) {
1488 for task in queue.list() {
1489 if task.status != TaskStatus::Done || task.runs.len() < 2 {
1490 continue;
1491 }
1492 supersede_prior_runs(&task, home);
1493 }
1494}
1495
1496fn settle_and_diagnose(
1503 task: &mut Task,
1504 verdict: Verdict,
1505 detail: &str,
1506 max_attempts: usize,
1507 state: &RunState,
1508) {
1509 let p = phrases(&state.config.graph.language);
1510 let attempts_before = task.attempts;
1511 if let Some(block) = state.resource_block.as_ref().filter(|_| !verdict.parked) {
1517 task.last_error = Some(detail.to_owned());
1518 task.diagnostic = None;
1519 task.attempts = task.attempts.saturating_sub(1);
1520 task.hold_machine(Some(format!(
1521 "{}{} stayed busy past the resource-wait limit before `{}` could start; \
1522 no agent was asked and no attempt was spent",
1523 crate::cache::RESOURCE_HOLD,
1524 block.kind,
1525 block.node
1526 )));
1527 if let Some(pass) = state.passes.last() {
1528 task.record_pass(&state.id, pass.index, attempts_before, None);
1529 }
1530 return;
1531 }
1532 settle_in(task, verdict, detail, max_attempts, p);
1533 if let Some(pass) = state.passes.last() {
1536 let parked = match &pass.end {
1537 Some(crate::run::PassEnd::Parked { cause }) => Some(cause.clone()),
1538 _ => None,
1539 };
1540 task.record_pass(&state.id, pass.index, attempts_before, parked);
1541 }
1542 if task.status == TaskStatus::Held {
1543 task.diagnostic = diagnostic(state);
1544 note_open_question(task, &state.id, p);
1545 }
1546}
1547
1548fn note_open_question(task: &mut Task, run: &str, p: &Phrases) {
1563 let Some(home) = crate::run::try_home() else {
1564 return;
1565 };
1566 let open = Questions::at(home.join("questions")).open_for(run);
1567 let Some(q) = open.first() else {
1568 return;
1569 };
1570 let base = task.hold_reason.clone().unwrap_or_default();
1571 task.hold_reason = Some(format!("{base}{}{}", p.waiting_for_answer, q.short()));
1572}
1573
1574fn reclaim(task: &mut Task, last_run: Option<RunState>, max_attempts: usize, language: &str) {
1585 match last_run {
1586 Some(state) => {
1587 let verdict = Verdict {
1588 status: state.status,
1589 left_pr: state.pr.is_some(),
1590 quota_hit: !state.quota.is_empty(),
1591 parked: state.parked,
1592 no_viable_candidates: state.viable().is_empty(),
1593 };
1594 let detail = format!(
1595 "{}{}",
1596 phrases(&state.config.graph.language).recovered_running,
1597 describe(&state)
1598 );
1599 settle_and_diagnose(task, verdict, &detail, max_attempts, &state);
1600 }
1601 None => {
1602 let why = phrases(language).no_run_to_recover;
1605 task.last_error = Some(why.to_owned());
1606 task.hold_machine(Some(why.to_owned()));
1609 }
1610 }
1611}
1612
1613fn reclaim_orphaned_running(queue: &Queue, max_attempts: usize) -> Vec<String> {
1635 let mut reclaimed = Vec::new();
1636 for listed in queue.list() {
1637 if listed.status != TaskStatus::Running {
1638 continue;
1639 }
1640 let Ok(_claim) = queue.claim(&listed.id) else {
1641 continue;
1642 };
1643 let Ok(mut task) = queue.get(&listed.id) else {
1647 continue;
1648 };
1649 if task.status != TaskStatus::Running {
1650 continue;
1651 }
1652 let last_run = task.runs.last().and_then(|id| RunState::load(id).ok());
1653 if let Some(state) = &last_run
1663 && let Err(e) = ask::Questions::open().settle_run(&state.id, state.status)
1664 {
1665 tracing::warn!("abandon questions for {}: {e:#}", state.id);
1666 }
1667 let language = if last_run.is_none() {
1668 language_of(&task, Path::new("."))
1669 } else {
1670 String::new()
1671 };
1672 reclaim(&mut task, last_run, max_attempts, &language);
1673 if task.status == TaskStatus::Done {
1674 supersede_prior_runs(&task, &crate::run::home());
1675 }
1676 record(queue, &mut task);
1677 reclaimed.push(task.id.clone());
1678 }
1679 reclaimed
1680}
1681
1682fn reclaim_abandoned_runs(home: &Path, now: Timestamp) -> Vec<String> {
1714 reclaim_abandoned_runs_with(
1715 home,
1716 now,
1717 crate::proc::pid_status,
1718 crate::proc::process_started_at,
1719 )
1720}
1721
1722fn reclaim_abandoned_runs_with<F, G>(
1728 home: &Path,
1729 now: Timestamp,
1730 query: F,
1731 identity: G,
1732) -> Vec<String>
1733where
1734 F: Fn(u32) -> Option<bool>,
1735 G: Fn(u32) -> Option<String>,
1736{
1737 let mut abandoned = Vec::new();
1738 for entry in std::fs::read_dir(home.join("runs"))
1739 .into_iter()
1740 .flatten()
1741 .flatten()
1742 {
1743 let id = entry.file_name().to_string_lossy().into_owned();
1744 if !crate::run::is_run_id(&id) {
1745 continue;
1746 }
1747 let Ok(body) = std::fs::read_to_string(entry.path().join("run.json")) else {
1755 continue;
1756 };
1757 let Ok(mut state) = serde_json::from_str::<RunState>(&body) else {
1758 continue;
1759 };
1760 if state.status.done() || !state.active_all_overrun(now) {
1761 continue;
1762 }
1763 let daemon_claims = is_working_on(home, &id, now);
1775 if state.liveness_with(daemon_claims, &query, &identity) != crate::run::Liveness::Dead {
1776 continue;
1777 }
1778 state.abandon("daemon");
1779 if let Err(e) = state.save_under(home) {
1780 tracing::warn!("could not persist abandoned run {id}: {e:#}");
1781 continue;
1782 }
1783 if let Some(notice) = notices::run_ended(&state) {
1786 notices::raise_in(home, notice);
1787 }
1788 if let Err(e) = Questions::at(home.join("questions")).settle_run(&id, state.status) {
1796 tracing::warn!("abandon questions for {id}: {e:#}");
1797 }
1798 abandoned.push(id);
1799 }
1800 abandoned
1801}
1802
1803pub async fn serve(opts: Opts) -> Result<()> {
1809 serve_until(opts, Stop::new()).await
1810}
1811
1812pub async fn serve_until(opts: Opts, stop: Stop) -> Result<()> {
1829 let signal = {
1830 let stop = stop.clone();
1831 tokio::spawn(async move {
1832 if tokio::signal::ctrl_c().await.is_ok() {
1833 stop.stop();
1834 tracing::info!("shutdown requested; a run in flight will be finished first");
1835 }
1836 })
1837 };
1838
1839 let worktrees_root = opts
1840 .worktrees_root
1841 .clone()
1842 .unwrap_or_else(crate::run::default_worktree_root);
1843 let outcome = drive(
1844 &opts,
1845 &Queue::open(),
1846 &status_path(),
1847 &crate::run::home(),
1848 &worktrees_root,
1849 &stop,
1850 )
1851 .await;
1852
1853 signal.abort();
1854 outcome
1855}
1856
1857async fn drive(
1870 opts: &Opts,
1871 queue: &Queue,
1872 status_file: &Path,
1873 home: &Path,
1874 worktrees_root: &Path,
1875 stop: &Stop,
1876) -> Result<()> {
1877 let status = Arc::new(Mutex::new(Status::new()));
1885 write_status_to(status_file, &lock(&status)).context("publish the daemon status file")?;
1886 let beat = tokio::spawn(heartbeat(Arc::clone(&status), status_file.to_path_buf()));
1887
1888 let daemon_cfg = prepare(&opts.repo, opts)
1894 .map(|c| c.daemon)
1895 .unwrap_or_default();
1896 let concurrency = max_concurrent(daemon_cfg.max_concurrent_runs);
1897
1898 let waiter = tokio::spawn(crate::waiter::run(
1903 crate::waiter::Waiter::new(
1904 crate::ask::Questions::at(home.join("questions")),
1905 home.to_path_buf(),
1906 prepare(&opts.repo, opts).ok(),
1907 ),
1908 stop.clone(),
1909 ));
1910
1911 let deputies = tokio::spawn(crate::deputy::run(
1915 crate::deputy::Deputies::new(
1916 crate::ask::Questions::at(home.join("questions")),
1917 home.to_path_buf(),
1918 prepare(&opts.repo, opts).ok(),
1919 opts.repo.clone(),
1920 daemon_cfg.max_deputies,
1921 {
1922 let stop = stop.clone();
1923 Arc::new(move || stop.parking())
1924 },
1925 ),
1926 stop.clone(),
1927 ));
1928
1929 let fetcher = tokio::spawn(fetch_loop(opts.repo.clone(), opts.clone(), stop.clone()));
1932
1933 let release_watch = tokio::spawn(crate::release_watch::run(
1936 crate::release_watch::Watcher::new(
1937 Box::new(crate::release_watch::GhForge),
1938 home.to_path_buf(),
1939 ),
1940 {
1941 let (repo, opts) = (opts.repo.clone(), opts.clone());
1942 move || release_watch_settings(&repo, &opts)
1943 },
1944 stop.clone(),
1945 ));
1946
1947 tracing::info!(
1948 "magi serve: queue {} (poll {}s, {} attempts per task, {} run(s) at once{})",
1949 queue.root().display(),
1950 opts.poll.as_secs(),
1951 opts.max_attempts,
1952 concurrency,
1953 if daemon_cfg.pause_for_interrupts {
1954 ", interrupts enabled"
1955 } else {
1956 ""
1957 }
1958 );
1959
1960 janitor(&opts.repo, opts, home, worktrees_root).await;
1963 resweep_superseded_attempts(queue, home);
1964
1965 let outcome = poll(
1966 opts,
1967 queue,
1968 &status,
1969 home,
1970 worktrees_root,
1971 stop,
1972 DispatchLimits {
1973 max_concurrent: concurrency,
1974 pause_for_interrupts: daemon_cfg.pause_for_interrupts,
1975 },
1976 )
1977 .await;
1978
1979 beat.abort();
1980 waiter.abort();
1981 deputies.abort();
1982 fetcher.abort();
1983 release_watch.abort();
1984 clear_status_at(status_file);
1985 outcome
1986}
1987
1988fn release_watch_settings(repo: &Path, opts: &Opts) -> (Vec<PathBuf>, u64) {
1991 let Ok(cfg) = prepare(repo, opts) else {
1992 return (Vec::new(), 0);
1993 };
1994 let mut paths: Vec<PathBuf> = crate::repos::scan(&cfg.repos.roots)
1995 .into_iter()
1996 .map(|r| r.path)
1997 .collect();
1998 let mut known: Vec<PathBuf> = Queue::open().list().into_iter().map(|t| t.repo).collect();
2001 let runs = crate::run::home().join("runs");
2002 for id in crate::run::list_ids_in(&runs) {
2003 let repo = std::fs::read_to_string(runs.join(&id).join("run.json"))
2004 .ok()
2005 .and_then(|s| serde_json::from_str::<serde_json::Value>(&s).ok())
2006 .and_then(|v| v.get("repo")?.as_str().map(PathBuf::from));
2007 known.extend(repo);
2008 }
2009 for k in known {
2010 let k = k.canonicalize().unwrap_or(k);
2011 if k.join(".git").exists() && !paths.contains(&k) {
2012 paths.push(k);
2013 }
2014 }
2015 let own = repo.canonicalize().unwrap_or_else(|_| repo.to_path_buf());
2016 if !paths.contains(&own) {
2017 paths.push(own);
2018 }
2019 (paths, cfg.daemon.release_stall_minutes)
2020}
2021
2022const FETCH_TIMEOUT: Duration = Duration::from_secs(30);
2024
2025async fn fetch_loop(repo: PathBuf, opts: Opts, stop: Stop) {
2029 while !stop.stopped() {
2030 let (interval, roots) = match prepare(&repo, &opts) {
2031 Ok(c) => (c.repos.fetch_interval, c.repos.roots),
2032 Err(_) => (0, Vec::new()),
2033 };
2034 if interval > 0 && !roots.is_empty() {
2035 let r = crate::clean::fetch_origins(&roots, FETCH_TIMEOUT, || stop.stopped()).await;
2036 tracing::debug!("fetch origins: {r:?}");
2037 }
2038 let wait = if interval > 0 { interval } else { 60 };
2040 let mut slept = 0;
2041 while slept < wait && !stop.stopped() {
2042 tokio::time::sleep(Duration::from_secs(1)).await;
2043 slept += 1;
2044 }
2045 }
2046}
2047
2048async fn heartbeat(status: Arc<Mutex<Status>>, path: PathBuf) {
2054 loop {
2055 tokio::time::sleep(HEARTBEAT).await;
2056 let snapshot = {
2057 let mut guard = lock(&status);
2058 guard.updated_at = Timestamp::now();
2059 guard.clone()
2060 };
2061 if let Err(e) = write_status_to(&path, &snapshot) {
2062 tracing::warn!("could not refresh the daemon status file: {e:#}");
2065 }
2066 }
2067}
2068
2069#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2072enum LandResume {
2073 NotLanding,
2076 StillWaiting,
2081 Ready,
2085}
2086
2087fn land_resume_state(task: &Task) -> LandResume {
2091 let Some(run_id) = task.runs.last() else {
2092 return LandResume::NotLanding;
2093 };
2094 let Ok(state) = RunState::load(run_id) else {
2095 return LandResume::NotLanding;
2096 };
2097 if !implementers_match(task, &state) || !state.parked {
2098 return LandResume::NotLanding;
2099 }
2100 let node = if state.status == RunStatus::Landing {
2104 land::APPROVAL_NODE
2105 } else if state.github_text.as_ref().is_some_and(|g| !g.resolved) {
2106 crate::github_text::ASK_NODE
2107 } else {
2108 return LandResume::NotLanding;
2109 };
2110 let store = ask::Questions::open();
2111 let waiting = store
2112 .list()
2113 .into_iter()
2114 .filter(|q| &q.run == run_id && q.node == node)
2115 .max_by(|a, b| a.id.cmp(&b.id));
2116 let Some(mut q) = waiting else {
2117 return LandResume::Ready;
2118 };
2119 if !q.status.open() {
2120 if q.resolution().as_deref() == Some(land::REPAIR) {
2123 return if state.liveness(false) == Liveness::Live {
2124 LandResume::StillWaiting
2125 } else {
2126 LandResume::NotLanding
2127 };
2128 }
2129 return LandResume::Ready;
2130 }
2131 let timeout = Duration::from_secs(state.config.graph.answer_timeout);
2138 let elapsed = Timestamp::now().as_second() - q.asked_at.as_second();
2139 if elapsed >= 0 && elapsed as u64 >= timeout.as_secs() {
2140 q.abandon(format!(
2141 "no answer within {}s of asking",
2142 timeout.as_secs().max(1)
2143 ));
2144 if store.put(&mut q).is_ok() {
2147 return LandResume::Ready;
2148 }
2149 }
2150 LandResume::StillWaiting
2151}
2152
2153const RECHECK_WHILE_BUSY: Duration = Duration::from_millis(200);
2161
2162const CACHE_CHECK_INTERVAL_SECS: u64 = 5 * 60;
2174
2175struct InFlightGuard<'a> {
2188 status: &'a Arc<Mutex<Status>>,
2189 stop: &'a Stop,
2190 task_id: &'a str,
2191}
2192
2193impl Drop for InFlightGuard<'_> {
2194 fn drop(&mut self) {
2195 lock(self.status).current.retain(|c| c.task != self.task_id);
2196 self.stop.exit();
2197 }
2198}
2199
2200#[derive(Debug, Clone, PartialEq, Eq)]
2222enum Interrupt {
2223 Idle,
2225 Parking {
2237 parked: Vec<String>,
2238 interrupt_task: String,
2239 },
2240 Running {
2248 parked: Vec<String>,
2249 interrupt_task: String,
2250 },
2251 Resuming { parked: Vec<String> },
2258}
2259
2260fn advance_interrupt(state: Interrupt, in_flight: &[String], runnable: &[Task]) -> Interrupt {
2281 match state {
2282 Interrupt::Idle => {
2283 if in_flight.len() != 1 {
2295 return Interrupt::Idle;
2296 }
2297 match runnable.iter().find(|t| t.interrupt) {
2298 Some(t) => Interrupt::Parking {
2299 parked: in_flight.to_vec(),
2300 interrupt_task: t.id.clone(),
2301 },
2302 None => Interrupt::Idle,
2303 }
2304 }
2305 Interrupt::Parking {
2306 parked,
2307 interrupt_task,
2308 } => {
2309 if in_flight.iter().any(|id| parked.contains(id)) {
2310 Interrupt::Parking {
2312 parked,
2313 interrupt_task,
2314 }
2315 } else if in_flight.contains(&interrupt_task) {
2316 Interrupt::Running {
2317 parked,
2318 interrupt_task,
2319 }
2320 } else if runnable.iter().any(|t| t.id == interrupt_task) {
2321 Interrupt::Parking {
2325 parked,
2326 interrupt_task,
2327 }
2328 } else {
2329 Interrupt::Resuming { parked }
2334 }
2335 }
2336 Interrupt::Running {
2337 parked,
2338 interrupt_task,
2339 } => {
2340 if in_flight.contains(&interrupt_task) {
2341 Interrupt::Running {
2342 parked,
2343 interrupt_task,
2344 }
2345 } else {
2346 Interrupt::Resuming { parked }
2352 }
2353 }
2354 Interrupt::Resuming { parked } => {
2355 if in_flight.iter().any(|id| parked.contains(id)) {
2356 Interrupt::Idle
2362 } else if runnable.iter().any(|t| parked.contains(&t.id)) {
2363 Interrupt::Resuming { parked }
2364 } else {
2365 Interrupt::Idle
2368 }
2369 }
2370 }
2371}
2372
2373fn advance_interrupt_tick(
2379 enabled: bool,
2380 state: Interrupt,
2381 in_flight: &[String],
2382 runnable: &[Task],
2383) -> Interrupt {
2384 if !enabled {
2385 return Interrupt::Idle;
2386 }
2387 advance_interrupt(state, in_flight, runnable)
2388}
2389
2390fn interrupt_gate(state: &Interrupt, in_flight: &[String], candidates: Vec<Task>) -> Vec<Task> {
2395 match state {
2396 Interrupt::Idle => candidates,
2397 Interrupt::Parking {
2398 parked,
2399 interrupt_task,
2400 } => {
2401 if in_flight.iter().any(|id| parked.contains(id)) {
2402 Vec::new()
2403 } else {
2404 candidates
2405 .into_iter()
2406 .filter(|t| &t.id == interrupt_task)
2407 .collect()
2408 }
2409 }
2410 Interrupt::Running { .. } => Vec::new(),
2411 Interrupt::Resuming { parked } => candidates
2419 .into_iter()
2420 .find(|t| parked.contains(&t.id))
2421 .into_iter()
2422 .collect(),
2423 }
2424}
2425
2426struct DispatchLimits {
2430 max_concurrent: usize,
2433 pause_for_interrupts: bool,
2435}
2436
2437#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2446enum PermitKind {
2447 None,
2452 Urgent,
2459 Ordinary,
2462}
2463
2464fn permit_kind(priority: bool, urgent: bool) -> PermitKind {
2467 if priority {
2468 PermitKind::None
2469 } else if urgent {
2470 PermitKind::Urgent
2471 } else {
2472 PermitKind::Ordinary
2473 }
2474}
2475
2476struct DispatchClaim {
2477 task: Task,
2478 claim: crate::queue::Claim,
2479 permit: Option<tokio::sync::OwnedSemaphorePermit>,
2480}
2481
2482fn claim_dispatch(
2485 queue: &Queue,
2486 candidate: &Task,
2487 cooling_down: bool,
2488 ordinary: &Arc<tokio::sync::Semaphore>,
2489 urgent: &Arc<tokio::sync::Semaphore>,
2490) -> Option<DispatchClaim> {
2491 let Ok(claim) = queue.claim(&candidate.id) else {
2492 tracing::info!("task {} is claimed elsewhere; skipping", candidate.short());
2493 return None;
2494 };
2495 let task = match queue.get(&candidate.id) {
2496 Ok(t) if t.status.runnable() => t,
2497 Ok(_) => return None,
2498 Err(e) => {
2499 tracing::warn!("could not re-read task {}: {e:#}", candidate.short());
2500 return None;
2501 }
2502 };
2503 let resume = land_resume_state(&task);
2504 if resume == LandResume::StillWaiting {
2505 return None;
2506 }
2507 let priority = resume == LandResume::Ready;
2508 if !priority && cooling_down {
2509 return None;
2510 }
2511 let permit = match permit_kind(priority, task.urgent) {
2514 PermitKind::None => None,
2515 PermitKind::Urgent => Some(Arc::clone(urgent).try_acquire_owned().ok()?),
2516 PermitKind::Ordinary => Some(Arc::clone(ordinary).try_acquire_owned().ok()?),
2517 };
2518 Some(DispatchClaim {
2519 task,
2520 claim,
2521 permit,
2522 })
2523}
2524
2525async fn poll(
2544 opts: &Opts,
2545 queue: &Queue,
2546 status: &Arc<Mutex<Status>>,
2547 home: &Path,
2548 worktrees_root: &Path,
2549 stop: &Stop,
2550 limits: DispatchLimits,
2551) -> Result<()> {
2552 let DispatchLimits {
2553 max_concurrent,
2554 pause_for_interrupts,
2555 } = limits;
2556 let mut attempted: Vec<String> = Vec::new();
2561 let sem = Arc::new(tokio::sync::Semaphore::new(max_concurrent));
2562 let urgent_sem = Arc::new(tokio::sync::Semaphore::new(1));
2570 let quota_cooldown_until: Arc<Mutex<Option<Timestamp>>> = Arc::new(Mutex::new(None));
2576 let mut inflight: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
2577 let mut conductor = Conductor::new();
2578 let mut cache_last_checked: Option<Timestamp> = None;
2581 let mut interrupt = Interrupt::Idle;
2583 let mut interrupt_pauses: std::collections::HashMap<String, crate::graph::Pause> =
2589 std::collections::HashMap::new();
2590
2591 while !stop.stopped() {
2592 lock(status).polls += 1;
2593
2594 while let Some(result) = inflight.try_join_next() {
2599 if let Err(e) = result {
2600 tracing::error!("a spawned attempt did not finish cleanly: {e}");
2601 notices::raise(Notice::error(
2602 "loop:attempt",
2603 "A queued attempt ended abnormally; check the task it was running.",
2604 ));
2605 }
2606 }
2607
2608 let swept = sweep_stale_claims(queue, STALE_CLAIM);
2609 if !swept.is_empty() {
2610 tracing::warn!(
2611 "swept {} stale claim(s) left behind by an earlier daemon: {}",
2612 swept.len(),
2613 swept.join(", ")
2614 );
2615 }
2616 let now = Timestamp::now();
2621
2622 if !stop.busy_now() {
2627 maybe_prune_cache_between_runs(
2628 &opts.repo,
2629 opts,
2630 home,
2631 stop,
2632 &mut cache_last_checked,
2633 now,
2634 )
2635 .await;
2636 }
2637
2638 let stalled = stalled_tasks(queue, home, now);
2639 let stalled_ids: std::collections::BTreeSet<_> =
2640 stalled.iter().map(|task| task.id.clone()).collect();
2641 let reclaimed = reclaim_orphaned_running(queue, opts.max_attempts);
2642 if !reclaimed.is_empty() {
2643 tracing::warn!(
2644 "reclaimed {} task(s) left `running` by a daemon that never \
2645 recorded the outcome: {}",
2646 reclaimed.len(),
2647 reclaimed.join(", ")
2648 );
2649 }
2650 let abandoned_runs = reclaim_abandoned_runs(home, now);
2651 if !abandoned_runs.is_empty() {
2652 tracing::warn!(
2653 "failed {} run(s) left behind by a killed process, past every \
2654 active seat's own timeout: {}",
2655 abandoned_runs.len(),
2656 abandoned_runs.join(", ")
2657 );
2658 }
2659
2660 let questions = Questions::at(home.join("questions"));
2665
2666 resolve_blockers(queue, &questions);
2669 apply_choice_actions(queue, &questions, home);
2670 triage::apply_answers(queue, &questions);
2671 reconcile_task_questions(queue, &questions);
2672 {
2676 let talks = crate::talk::Talks::at(home.join("talks"));
2677 crate::chat_notify::sweep(queue, &talks, home);
2678 crate::chat_watch::sweep(home, &talks);
2679 crate::chat_notify::start_turns(&talks, stop.parking(), &mut inflight);
2680 }
2681
2682 let finished: Vec<Task> = finished_tasks(queue)
2688 .into_iter()
2689 .filter(|task| !stalled_ids.contains(&task.id))
2690 .filter(|task| !awaiting_resume_with(task, |id| RunState::load_under(id, home)))
2694 .collect();
2695 let queued = queued_tasks(queue);
2696 if !(queued.is_empty() && stalled.is_empty() && finished.is_empty())
2700 && conductor.worth_a_look(queue, &stalled, &finished)
2701 {
2702 match prepare(&opts.repo, opts) {
2703 Ok(cfg) => {
2704 conductor
2705 .maybe_run(
2706 &cfg,
2707 &opts.repo,
2708 queue,
2709 &questions,
2710 home,
2711 &queued,
2712 &stalled,
2713 &finished,
2714 opts.max_attempts,
2715 )
2716 .await;
2717 }
2718 Err(e) => {
2719 tracing::warn!("conductor: no config: {e:#}");
2720 notices::raise(Notice::warn(
2721 "loop:no-config",
2722 "The loop could not read this repository's config, so held tasks are not being triaged.",
2723 ));
2724 }
2725 }
2726 }
2727
2728 let candidates: Vec<Task> = runnable(queue)
2729 .into_iter()
2730 .filter(|t| !opts.once || !attempted.contains(&t.id))
2731 .collect();
2732
2733 let in_flight: Vec<String> = lock(status)
2738 .current
2739 .iter()
2740 .map(|c| c.task.clone())
2741 .collect();
2742 interrupt_pauses.retain(|id, _| in_flight.contains(id));
2743
2744 interrupt =
2745 advance_interrupt_tick(pause_for_interrupts, interrupt, &in_flight, &candidates);
2746 if let Interrupt::Parking {
2747 parked,
2748 interrupt_task,
2749 } = &interrupt
2750 {
2751 let reason = format!(
2752 "task {} asked to run first",
2753 crate::run::short_of(interrupt_task)
2754 );
2755 for id in parked {
2756 if let Some(pause) = interrupt_pauses.get(id) {
2757 pause.park_because(reason.clone());
2758 }
2759 }
2760 }
2761 let candidates = interrupt_gate(&interrupt, &in_flight, candidates);
2762
2763 let cooling_down =
2764 lock("a_cooldown_until).is_some_and(|until| Timestamp::now() < until);
2765
2766 let mut started_any = false;
2767 for candidate in candidates {
2768 if stop.stopped() {
2769 break;
2770 }
2771
2772 let Some(DispatchClaim {
2773 mut task,
2774 claim,
2775 permit,
2776 }) = claim_dispatch(queue, &candidate, cooling_down, &sem, &urgent_sem)
2777 else {
2778 continue;
2779 };
2780
2781 let task_id = task.id.clone();
2782 attempted.push(task_id.clone());
2783 lock(status).idle = false;
2784 stop.enter();
2787 started_any = true;
2788
2789 let run_pause = crate::graph::Pause::new();
2793 interrupt_pauses.insert(task_id.clone(), run_pause.clone());
2794
2795 let opts = opts.clone();
2796 let queue = queue.clone();
2797 let status = Arc::clone(status);
2798 let stop = stop.clone();
2799 let quota_cooldown_until = Arc::clone("a_cooldown_until);
2800 inflight.spawn(async move {
2801 let _claim = claim;
2805 let _permit = permit;
2806 let _inflight = InFlightGuard {
2808 status: &status,
2809 stop: &stop,
2810 task_id: &task_id,
2811 };
2812 let quota = attempt(&opts, &queue, &status, &stop, run_pause, &mut task).await;
2813 lock(&status).completed += 1;
2814 let now = Timestamp::now();
2820 if let Some(until) = cooldown_until("a, now) {
2821 let wait = until.as_second() - now.as_second();
2822 *lock("a_cooldown_until) = Some(until);
2823 let hint = quota
2824 .iter()
2825 .find(|q| q.reset.is_some())
2826 .and_then(|q| q.reset.as_deref());
2827 match hint {
2828 Some(h) => tracing::warn!(
2829 "quota hit; waiting {wait}s before taking another ordinary task \
2830 (CLI reported reset: {h})"
2831 ),
2832 None => tracing::warn!(
2833 "quota hit; waiting {wait}s before taking another ordinary task \
2834 (no reset hint reported)"
2835 ),
2836 }
2837 }
2838 });
2839 }
2840
2841 if started_any {
2842 continue;
2843 }
2844
2845 if stop.busy_now() {
2846 stop.idle(RECHECK_WHILE_BUSY.min(opts.poll)).await;
2851 continue;
2852 }
2853
2854 lock(status).idle = true;
2856 if opts.once {
2857 janitor(&opts.repo, opts, home, worktrees_root).await;
2861 resweep_superseded_attempts(queue, home);
2862 triage_held(queue, home, opts).await;
2863 break;
2864 }
2865 stop.idle(opts.poll).await;
2866 if stop.stopped() {
2867 continue;
2868 }
2869 janitor(&opts.repo, opts, home, worktrees_root).await;
2875 resweep_superseded_attempts(queue, home);
2876 triage_held(queue, home, opts).await;
2877 }
2878
2879 while let Some(result) = inflight.join_next().await {
2884 if let Err(e) = result {
2885 tracing::error!("a spawned attempt did not finish cleanly: {e}");
2886 notices::raise(Notice::error(
2887 "loop:attempt",
2888 "A queued attempt ended abnormally; check the task it was running.",
2889 ));
2890 }
2891 }
2892 Ok(())
2893}
2894
2895async fn attempt(
2901 opts: &Opts,
2902 queue: &Queue,
2903 status: &Arc<Mutex<Status>>,
2904 stop: &Stop,
2905 interrupt_pause: crate::graph::Pause,
2906 task: &mut Task,
2907) -> Vec<QuotaLoss> {
2908 let repo = repo_for(task, &opts.repo);
2909 tracing::info!(
2910 "task {} — {} (repo {})",
2911 task.short(),
2912 task.title,
2913 repo.display()
2914 );
2915
2916 let explicit = match &task.overrides {
2920 Some(o) if !o.inherit_daemon_options => o.config.as_deref(),
2921 Some(_) => opts.config.as_deref(),
2922 None => opts.config.as_deref(),
2923 };
2924 if let Err(e) = Config::discover_fetched(&repo, explicit).await {
2925 let reason = format!("[config] {e:#}");
2926 task.last_error = Some(reason.clone());
2927 task.hold_machine(Some(reason.clone()));
2928 record(queue, task);
2929 tracing::warn!("holding {}: {reason}", task.short());
2930 return Vec::new();
2931 }
2932 let mut config = match prepare_for(&repo, opts, task) {
2933 Ok(c) => c,
2934 Err(e) => {
2935 task.attempts += 1;
2939 task.fail(format!("config: {e:#}"), opts.max_attempts);
2940 record(queue, task);
2941 return Vec::new();
2942 }
2943 };
2944 apply_task_implementers(&mut config, task);
2945 let p = phrases(&config.graph.language);
2946 let start_failed = p.could_not_start;
2947
2948 if let Some(reason) = disk_gate(&repo, &config) {
2956 task.last_error = Some(reason.clone());
2957 task.hold_machine(Some(reason.clone()));
2958 record(queue, task);
2959 tracing::warn!("holding {} for want of disk space: {reason}", task.short());
2960 notices::raise(
2963 Notice::warn(
2964 &format!("disk:{}", repo.display()),
2965 "A task was held for want of free disk space; free some, then release it from the queue.",
2966 )
2967 .link(Link::Task {
2968 id: task.id.clone(),
2969 }),
2970 );
2971 return Vec::new();
2972 }
2973
2974 let unfinished = (!task.fresh_start)
2994 .then(|| unfinished_run(&task.runs, task.short()))
2995 .flatten()
2996 .filter(|id| RunState::load(id).is_ok_and(|run| may_resume_run(&run, task)));
2997 let previous = task.runs.last().and_then(|id| RunState::load(id).ok());
3003 let review_branch = take_review_branch(task, previous.as_ref());
3004 let branch_exists = match &review_branch {
3005 Some(branch) => crate::git::branch_exists(&repo, branch)
3006 .await
3007 .unwrap_or(false),
3008 None => false,
3009 };
3010 let proposed = match (&review_branch, &unfinished, &task.review_of) {
3014 (None, None, Some(branch)) => Starter::Review(branch.clone()),
3015 _ => choose_starter(
3016 review_branch.as_deref(),
3017 branch_exists,
3018 unfinished.as_deref(),
3019 ),
3020 };
3021 let starter = if let Some(selected) = &task.pr_recovery {
3022 if selected.check_task(&task.id, &task.instruction).is_ok() {
3023 selected_starter(selected, previous.as_ref(), unfinished.as_deref())
3024 } else {
3025 Starter::Review(selected.branch.clone())
3026 }
3027 } else {
3028 starter_with_implementers(task, previous.as_ref(), proposed)
3029 };
3030 if let Some(pin) = &task.resume_override
3033 && let Some(run) = &pin.pinned_run
3034 && Questions::open().get(&pin.question_id).map_or(true, |q| {
3035 q.node == land::APPROVAL_NODE && q.resolution().as_deref() == Some(land::REPAIR)
3036 })
3037 && starter != Starter::Resume(run.clone())
3038 {
3039 task.hold_machine(Some(format!(
3040 "repair owner cannot resume; inspect `magi show {run}` and restore the existing run before `magi task release {}`; no replacement task was started",
3041 task.id
3042 )));
3043 record(queue, task);
3044 return Vec::new();
3045 }
3046 let attachments = match task_attachments(queue, task) {
3051 Ok(a) => a,
3052 Err(e) => {
3053 task.attempts += 1;
3054 task.fail(format!("{start_failed}{e:#}"), opts.max_attempts);
3055 record(queue, task);
3056 return Vec::new();
3057 }
3058 };
3059 let started = match &starter {
3060 Starter::Review(_) if task.pr_recovery.is_some() => {
3061 let selected = task.pr_recovery.clone().expect("selected recovery");
3062 match selected.check_task(&task.id, &task.instruction) {
3063 Ok(()) => Runner::recover_pr(&repo, &task.id, selected, config)
3064 .await
3065 .map(|mut r| {
3066 r.state.attachments = attachments.clone();
3067 r
3068 }),
3069 Err(e) => Err(e),
3070 }
3071 }
3072 Starter::Review(branch) => {
3073 tracing::info!(
3074 "task {} reopens `{branch}` as a review-only pass",
3075 task.short()
3076 );
3077 let takeover = crate::handover::Takeover {
3080 earlier: task.earlier_attempts().to_vec(),
3081 home: crate::run::home(),
3082 choice: take_divergence_answer(branch, &config.merge.remote, task),
3083 task_instruction: Some(instruction_for(task)),
3084 };
3085 Runner::review_taking_over(
3086 &repo,
3087 branch,
3088 config,
3089 Some(takeover),
3090 crate::run::Origin::queue(&task.id),
3091 )
3092 .await
3093 }
3094 Starter::Resume(id) => {
3095 tracing::info!("resuming run {id} rather than competing again");
3096 Runner::resume(id).and_then(|mut r| {
3097 if r.state.prep_failure.is_some() {
3098 let approval = task
3099 .resumption
3100 .as_ref()
3101 .context("preparation retry requires an explicit task release")?;
3102 r.retry_preparation(approval.at)?;
3103 }
3104 if let Some(instruction) =
3105 prepare_instruction(&starter, Some(&r.state.instruction), task)
3106 {
3107 r.state.instruction = instruction;
3108 }
3109 r.state.attachments = attachments.clone();
3110 if let Some(mode) = task
3113 .overrides
3114 .as_ref()
3115 .and_then(|o| o.merge.as_deref())
3116 .and_then(|m| merge_mode(m).ok())
3117 {
3118 r.state.config.merge.mode = mode;
3119 }
3120 Ok(r)
3121 })
3122 }
3123 Starter::Start => {
3124 if let Some(branch) = &review_branch {
3125 tracing::warn!(
3126 "conductor chose review for task {} but branch `{branch}` no longer \
3127 exists; requeuing as a fresh competition instead",
3128 task.short()
3129 );
3130 }
3131 let instruction = prepare_instruction(&starter, None, task)
3132 .unwrap_or_else(|| task.instruction.clone());
3133 Runner::start_naming(
3134 &repo,
3135 instruction,
3136 &task.title,
3137 config,
3138 crate::run::Origin::queue(&task.id),
3139 )
3140 .await
3141 .map(|mut r| {
3142 r.state.attachments = attachments.clone();
3143 r
3144 })
3145 }
3146 };
3147 let mut runner = match started {
3148 Ok(r) => r,
3149 Err(e) if task.pr_recovery.is_some() => {
3153 task.attempts += 1;
3154 if let Some(partial) = e.downcast_ref::<crate::pr_recovery::StartupFailed>() {
3155 task.link_run(&partial.run);
3156 task.record_pass(&partial.run, 0, task.attempts, None);
3157 }
3158 task.last_error = Some(format!("{start_failed}{e:#}"));
3159 task.hold_unrequested_retry(Some(format!(
3160 "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."
3161 )));
3162 record(queue, task);
3163 return Vec::new();
3164 }
3165 Err(e) if e.downcast_ref::<crate::handover::Refused>().is_some() => {
3166 let detail = match e.downcast_ref::<crate::handover::Refused>() {
3167 Some(r) => (p.handover_refused)(r),
3168 None => format!("{e:#}"),
3169 };
3170 let reason = format!("{start_failed}{detail}");
3171 task.last_error = Some(reason.clone());
3172 let branch = match &starter {
3175 Starter::Review(branch) => Some(branch.clone()),
3176 _ => None,
3177 };
3178 task.hold_for_handover(branch, format!("{reason}{}", p.handover_hint));
3182 record(queue, task);
3183 tracing::warn!(
3184 "holding {} for a branch it cannot take over: {e:#}",
3185 task.short()
3186 );
3187 return Vec::new();
3188 }
3189 Err(e) if e.downcast_ref::<crate::reconcile::Diverged>().is_some() => {
3194 let d = e
3195 .downcast_ref::<crate::reconcile::Diverged>()
3196 .expect("checked by the guard");
3197 let mut q = ask::Question::new(
3198 task.id.clone(),
3199 crate::reconcile::NODE.to_owned(),
3200 crate::reconcile::SEAT.to_owned(),
3201 d.summary(),
3202 d.detail(),
3203 d.choices(),
3204 );
3205 match Questions::open().put(&mut q) {
3206 Ok(()) => {
3207 task.last_error = Some(format!("{e:#}"));
3208 task.review_branch = Some(d.branch.clone());
3211 task.review_instruction = Some(task.instruction.clone());
3212 task.block(vec![q.id.clone()], Some(d.summary()));
3213 }
3214 Err(put) => {
3215 tracing::warn!("could not file the divergence question: {put:#}");
3216 task.attempts += 1;
3217 task.fail(format!("{start_failed}{e:#}"), opts.max_attempts);
3218 }
3219 }
3220 record(queue, task);
3221 return Vec::new();
3222 }
3223 Err(e) => {
3224 if e.downcast_ref::<crate::reconcile::Stale>().is_some()
3230 && let Starter::Review(branch) = &starter
3231 {
3232 task.review_branch = Some(branch.clone());
3233 task.review_instruction = Some(task.instruction.clone());
3234 task.last_error = Some(format!("{start_failed}{e:#}"));
3235 task.status = crate::queue::TaskStatus::Failed;
3236 record(queue, task);
3237 return Vec::new();
3238 }
3239 task.attempts += 1;
3240 task.last_error = Some(format!("{start_failed}{e:#}"));
3241 task.hold_unrequested_retry(Some(format!(
3242 "Startup failed: {e:#}. Correct the startup source or configuration, then explicitly release the task; automatic recovery is stopped."
3243 )));
3244 record(queue, task);
3245 return Vec::new();
3246 }
3247 };
3248 runner.state.task_instruction = Some(task.instruction.clone());
3249 runner.state.followup_generation = Some(task.followup.as_ref().map_or(0, |f| f.generation));
3252 if runner.state.origin_chat.is_none() {
3253 runner.state.origin_chat = task.chat_talk().map(str::to_owned);
3254 }
3255 runner.on_pause(stop.pause());
3257 runner.watch_interrupt(interrupt_pause);
3261
3262 let run = runner.state.id.clone();
3265 task.start(run.clone());
3266 record(queue, task);
3267 lock(status).current.push(Current {
3268 task: task.id.clone(),
3269 run,
3270 });
3271
3272 let quota_before = runner.state.quota.clone();
3275 runner.state.next_pass_task_entry = true;
3276 let result = runner.execute().await;
3277 finish_attempt(
3278 opts.max_attempts,
3279 queue,
3280 task,
3281 &runner.state,
3282 "a_before,
3283 result,
3284 )
3285}
3286
3287pub fn finish_attempt(
3292 max_attempts: usize,
3293 queue: &Queue,
3294 task: &mut Task,
3295 state: &RunState,
3296 quota_before: &[QuotaLoss],
3297 result: Result<()>,
3298) -> Vec<QuotaLoss> {
3299 let detail = match result {
3300 Ok(()) => describe(state),
3301 Err(e) => format!("{e:#}"),
3302 };
3303 let fresh = losses_this_attempt(quota_before, &state.quota);
3304 let verdict = Verdict {
3305 status: state.status,
3306 left_pr: state.pr.is_some(),
3309 quota_hit: !fresh.is_empty(),
3315 parked: state.parked,
3319 no_viable_candidates: state.viable().is_empty(),
3322 };
3323 settle_and_diagnose(task, verdict, &detail, max_attempts, state);
3324 if state.status == RunStatus::Blocked && state.prep_failure.is_some() {
3325 task.hold_unrequested_retry(Some(format!(
3326 "Candidate setup failed: {detail}. Select a complete implementation with `magi task recover-pr`, or correct the referenced source before explicitly releasing the task."
3327 )));
3328 }
3329 if task.status == TaskStatus::Done {
3330 supersede_prior_runs(task, &crate::run::home());
3331 }
3332 record(queue, task);
3333 tracing::info!(
3334 "task {} is {} after run {} ({})",
3335 task.short(),
3336 task.status.as_str(),
3337 state.short(),
3338 label(state.status)
3339 );
3340 fresh
3341}
3342
3343pub fn hold_if_runnable(queue: &Queue, task: &mut Task) {
3347 let waits_on_owner = withheld_text_wait(task).is_some();
3352 if task.status.runnable() && !waits_on_owner {
3353 let why = task.last_error.clone().map_or_else(
3354 || "the run did not finish".to_owned(),
3355 |e| format!("the run did not finish: {e}"),
3356 );
3357 task.hold_unrequested_retry(Some(format!(
3358 "{why}. It was started by hand, so it is not retried \
3359 automatically; `magi task release` retries it."
3360 )));
3361 record(queue, task);
3362 }
3363}
3364
3365#[must_use]
3369pub fn loop_would_resume(run: &str) -> bool {
3370 unfinished_run(&[run.to_owned()], crate::run::short_of(run)).is_some()
3371}
3372
3373#[must_use]
3383pub fn foreign_loop(
3384 reading: Option<&Reading>,
3385 now: Timestamp,
3386 own_pid: u32,
3387) -> Option<Option<u32>> {
3388 let reading = reading.filter(|r| r.running(now))?;
3389 match reading.pid {
3390 Some(pid) if pid == own_pid => None,
3391 pid => Some(pid),
3392 }
3393}
3394
3395pub async fn run_claimed(opts: &Opts, queue: &Queue, task: &mut Task) {
3410 let status = Arc::new(Mutex::new(Status::new()));
3411 let stop = Stop::new();
3412 attempt(
3413 opts,
3414 queue,
3415 &status,
3416 &stop,
3417 crate::graph::Pause::new(),
3418 task,
3419 )
3420 .await;
3421 let mut resumes = 0;
3430 while resumes < MAX_WITHHELD_RESUMES {
3431 let Some(timeout) = withheld_text_wait(task) else {
3432 break;
3433 };
3434 eprintln!(
3435 "waiting for the owner's answer on a withheld pull request text \
3436 (up to {timeout}s; Ctrl-C leaves the run parked for a later `magi serve`)"
3437 );
3438 let limit = std::time::Instant::now() + Duration::from_secs(timeout + 60);
3441 let ready = loop {
3442 match land_resume_state(task) {
3443 LandResume::StillWaiting if std::time::Instant::now() < limit => {
3444 tokio::time::sleep(WITHHELD_POLL).await;
3445 }
3446 LandResume::Ready => break true,
3447 _ => break false,
3448 }
3449 };
3450 if !ready {
3451 break;
3452 }
3453 resumes += 1;
3454 attempt(
3455 opts,
3456 queue,
3457 &status,
3458 &stop,
3459 crate::graph::Pause::new(),
3460 task,
3461 )
3462 .await;
3463 }
3464 hold_if_runnable(queue, task);
3465}
3466
3467const MAX_WITHHELD_RESUMES: u32 = 3;
3470
3471const WITHHELD_POLL: Duration = Duration::from_secs(15);
3473
3474fn withheld_text_wait(task: &Task) -> Option<u64> {
3477 let id = task.runs.last()?;
3478 let s = RunState::load(id).ok()?;
3479 (s.parked && s.github_text.as_ref().is_some_and(|g| !g.resolved))
3480 .then_some(s.config.graph.answer_timeout)
3481}
3482
3483fn losses_this_attempt(before: &[QuotaLoss], after: &[QuotaLoss]) -> Vec<QuotaLoss> {
3493 after
3494 .iter()
3495 .filter(|q| !before.contains(q))
3496 .cloned()
3497 .collect()
3498}
3499
3500fn cooldown_until(quota: &[QuotaLoss], now: Timestamp) -> Option<Timestamp> {
3503 if quota.is_empty() {
3504 return None;
3505 }
3506 let with_hint = quota.iter().find(|q| q.reset.is_some());
3507 let reset_at = with_hint.and_then(|q| parse_reset_hint(q.reset.as_deref()?, now, q.at));
3508 let wait = quota_wait(reset_at, now, QUOTA_WAIT_FALLBACK, QUOTA_WAIT_CAP);
3509 let secs = i64::try_from(wait.as_secs()).unwrap_or(i64::MAX);
3510 Some(
3511 now.checked_add(jiff::SignedDuration::from_secs(secs))
3512 .unwrap_or(Timestamp::MAX),
3513 )
3514}
3515
3516fn implementers_match(task: &Task, state: &RunState) -> bool {
3518 task.implementers()
3519 .is_none_or(|n| n == state.config.graph.implementers)
3520}
3521
3522fn starter_with_implementers(
3524 task: &Task,
3525 previous: Option<&RunState>,
3526 proposed: Starter,
3527) -> Starter {
3528 if previous.is_some_and(|s| !implementers_match(task, s)) {
3529 task.review_of
3530 .as_ref()
3531 .map_or(Starter::Start, |branch| Starter::Review(branch.clone()))
3532 } else {
3533 proposed
3534 }
3535}
3536
3537fn apply_task_implementers(config: &mut Config, task: &Task) {
3539 if let Some(n) = task.implementers() {
3540 config.graph.implementers = n;
3541 }
3542}
3543
3544fn prepare_for(repo: &Path, opts: &Opts, task: &Task) -> Result<Config> {
3549 let Some(o) = &task.overrides else {
3550 return prepare(repo, opts);
3551 };
3552 let own = Opts {
3556 config: o.config.clone(),
3557 merge: None,
3558 ..opts.clone()
3559 };
3560 let mut config = prepare(repo, if o.inherit_daemon_options { opts } else { &own })?;
3561 o.apply(&mut config);
3562 Ok(config)
3563}
3564
3565fn prepare(repo: &Path, opts: &Opts) -> Result<Config> {
3567 let (mut config, _layers) = Config::discover(repo, opts.config.as_deref())?;
3568 if let Some(mode) = &opts.merge {
3569 config.merge.mode = merge_mode(mode)?;
3570 }
3571 Ok(config)
3572}
3573
3574async fn maybe_prune_cache_between_runs(
3605 repo: &Path,
3606 opts: &Opts,
3607 home: &Path,
3608 stop: &Stop,
3609 last_checked: &mut Option<Timestamp>,
3610 now: Timestamp,
3611) {
3612 if stop.stopped() || !cache_check_due(*last_checked, now, CACHE_CHECK_INTERVAL_SECS) {
3613 return;
3614 }
3615 *last_checked = Some(now);
3616 let cfg = match prepare(repo, opts) {
3617 Ok(cfg) => cfg,
3618 Err(e) => {
3619 tracing::warn!("cache check: no config: {e:#}");
3620 return;
3621 }
3622 };
3623 match clean::prune_cache_if_over_limit(&cfg, home) {
3624 Ok(Some(pruned)) if pruned.files > 0 => tracing::info!(
3625 "housekeep: pruned {} file(s) ({} bytes) from the shared cache between runs",
3626 pruned.files,
3627 pruned.freed
3628 ),
3629 Ok(_) => {}
3630 Err(e) => {
3631 tracing::warn!("housekeep: prune cache: {e:#}");
3632 notices::raise_in(
3633 home,
3634 Notice::warn(
3635 "housekeep:cache",
3636 "Pruning the shared build cache failed; disk usage may keep growing.",
3637 ),
3638 );
3639 }
3640 }
3641}
3642
3643fn cache_check_due(last_checked: Option<Timestamp>, now: Timestamp, interval_secs: u64) -> bool {
3647 last_checked.is_none_or(|last| clean::due(now, last, interval_secs))
3648}
3649
3650async fn janitor(repo: &Path, opts: &Opts, home: &Path, worktrees_root: &Path) {
3669 let cfg = match prepare(repo, opts) {
3670 Ok(cfg) => cfg,
3671 Err(e) => {
3672 tracing::warn!("housekeep: no config: {e:#}");
3673 return;
3674 }
3675 };
3676 let worktrees_root = cfg.graph.worktree_root.as_deref().unwrap_or(worktrees_root);
3685 let out = clean::housekeep(&cfg, home, worktrees_root, repo, Timestamp::now()).await;
3686 if out.folded > 0 || out.unreadable > 0 || out.orphaned_worktrees > 0 {
3691 let mut extra = Vec::new();
3692 if out.unreadable > 0 {
3693 extra.push(format!("{} unreadable", out.unreadable));
3694 }
3695 if out.orphaned_worktrees > 0 {
3696 extra.push(format!("{} orphaned worktree(s)", out.orphaned_worktrees));
3697 }
3698 let detail = if extra.is_empty() {
3699 String::new()
3700 } else {
3701 format!(" ({})", extra.join(", "))
3702 };
3703 tracing::info!("housekeep: folded {} run(s){detail}", out.folded);
3704 }
3705 if out.external_merges_recorded > 0 {
3706 tracing::info!(
3707 "housekeep: recorded {} run(s) as merged externally",
3708 out.external_merges_recorded
3709 );
3710 }
3711 if out.stale_pr_states_repaired > 0 {
3712 tracing::info!(
3713 "housekeep: rewrote {} run record(s) whose pull request had already settled",
3714 out.stale_pr_states_repaired
3715 );
3716 }
3717 if out.cache_files > 0 {
3718 tracing::info!(
3719 "housekeep: pruned {} file(s) ({} bytes) from the shared cache",
3720 out.cache_files,
3721 out.cache_freed
3722 );
3723 }
3724 if out.questions_abandoned > 0 {
3725 tracing::info!(
3726 "housekeep: abandoned {} question(s) left open by a finished run",
3727 out.questions_abandoned
3728 );
3729 }
3730}
3731
3732async fn triage_held(queue: &Queue, home: &Path, opts: &Opts) {
3741 let questions = Questions::at(home.join("questions"));
3742 let report = triage::run_once(queue, &questions, opts.config.as_deref(), Timestamp::now());
3743 if report.is_empty() {
3744 return;
3745 }
3746 if !report.quarantined.is_empty() {
3747 tracing::info!(
3748 "triage: held {} blocked task(s) whose blocked-on task or \
3749 question no longer exists: {}",
3750 report.quarantined.len(),
3751 report.quarantined.join(", ")
3752 );
3753 }
3754 if !report.resumed.is_empty() {
3755 tracing::info!(
3756 "triage: resumed {} held task(s) whose machine hold had resolved: {}",
3757 report.resumed.len(),
3758 report.resumed.join(", ")
3759 );
3760 }
3761 if !report.asked.is_empty() {
3762 tracing::info!(
3763 "triage: asked about {} held task(s): {}",
3764 report.asked.len(),
3765 report.asked.join(", ")
3766 );
3767 }
3768 if !report.answered.is_empty() {
3769 tracing::info!(
3770 "triage: applied {} operator answer(s): {}",
3771 report.answered.len(),
3772 report.answered.join(", ")
3773 );
3774 }
3775}
3776
3777fn disk_gate(repo: &Path, config: &Config) -> Option<String> {
3784 disk_gate_with(repo, config, crate::disk::free_bytes)
3785}
3786
3787fn disk_gate_with<F: Fn(&Path) -> Result<u64>>(
3791 repo: &Path,
3792 config: &Config,
3793 free_bytes: F,
3794) -> Option<String> {
3795 let min = config.disk.min_free_bytes;
3796 if min == 0 {
3797 return None;
3798 }
3799 match free_bytes(repo) {
3800 Ok(free) => crate::disk::gate_in(free, min, &config.graph.language),
3801 Err(e) => Some(crate::disk::unmeasured_in(repo, &e, &config.graph.language)),
3802 }
3803}
3804
3805const QUOTA_WAIT_FALLBACK: Duration = Duration::from_secs(5 * 60);
3812
3813const QUOTA_WAIT_CAP: Duration = Duration::from_secs(30 * 60);
3817
3818fn quota_wait(
3827 reset_at: Option<Timestamp>,
3828 now: Timestamp,
3829 fallback: Duration,
3830 cap: Duration,
3831) -> Duration {
3832 match reset_at {
3833 Some(at) if at > now => {
3834 let secs = u64::try_from(at.as_second() - now.as_second()).unwrap_or(0);
3835 Duration::from_secs(secs).min(cap)
3836 }
3837 _ => fallback,
3838 }
3839}
3840
3841fn parse_reset_hint(text: &str, now: Timestamp, recorded: Timestamp) -> Option<Timestamp> {
3855 parse_reset_hint_zoned(text, now)
3856 .or_else(|| parse_reset_hint_dated(text))
3857 .or_else(|| parse_reset_hint_relative(text, recorded))
3858}
3859
3860fn parse_reset_hint_relative(text: &str, recorded: Timestamp) -> Option<Timestamp> {
3864 let rest = text.trim().trim_end_matches('.').strip_prefix("in ")?;
3865 let mut rest = rest.trim();
3866 if rest.is_empty() {
3867 return None;
3868 }
3869 let mut total: i64 = 0;
3870 let mut matched = false;
3871 for (unit, secs) in [('h', 3600), ('m', 60), ('s', 1)] {
3872 if let Some((digits, tail)) = rest.split_once(unit)
3873 && !digits.is_empty()
3874 && digits.bytes().all(|b| b.is_ascii_digit())
3875 {
3876 total += digits.parse::<i64>().ok()?.checked_mul(secs)?;
3877 rest = tail;
3878 matched = true;
3879 }
3880 }
3881 if !rest.is_empty() || !matched {
3882 return None;
3883 }
3884 recorded
3885 .checked_add(jiff::SignedDuration::from_secs(total))
3886 .ok()
3887}
3888
3889fn parse_12h_clock(clock: &str) -> Option<(i8, i8)> {
3893 let clock = clock.trim().to_lowercase();
3894 let (digits, pm) = clock
3895 .strip_suffix("am")
3896 .map(|d| (d, false))
3897 .or_else(|| clock.strip_suffix("pm").map(|d| (d, true)))?;
3898 let (h, m) = digits.trim().split_once(':')?;
3899 let mut hour: i8 = h.trim().parse().ok()?;
3900 let minute: i8 = m.trim().parse().ok()?;
3901 if !(1..=12).contains(&hour) || !(0..=59).contains(&minute) {
3902 return None;
3903 }
3904 if pm && hour != 12 {
3905 hour += 12;
3906 } else if !pm && hour == 12 {
3907 hour = 0;
3908 }
3909 Some((hour, minute))
3910}
3911
3912fn parse_reset_hint_zoned(text: &str, now: Timestamp) -> Option<Timestamp> {
3917 let open = text.find('(')?;
3918 let close = text.rfind(')')?;
3919 if close <= open {
3920 return None;
3921 }
3922 let zone = text[open + 1..close].trim();
3923 let (hour, minute) = parse_12h_clock(&text[..open])?;
3924 let tz = jiff::tz::TimeZone::get(zone).ok()?;
3925 let candidate = now
3926 .to_zoned(tz)
3927 .with()
3928 .hour(hour)
3929 .minute(minute)
3930 .second(0)
3931 .millisecond(0)
3932 .microsecond(0)
3933 .nanosecond(0)
3934 .build()
3935 .ok()?;
3936 let mut at = candidate.timestamp();
3937 if at <= now {
3938 at += jiff::SignedDuration::from_hours(24);
3939 }
3940 Some(at)
3941}
3942
3943fn parse_reset_hint_dated(text: &str) -> Option<Timestamp> {
3952 let words: Vec<&str> = text.split_whitespace().collect();
3953 if words.len() < 5 {
3954 return None;
3955 }
3956 (0..=words.len() - 5)
3957 .find_map(|start| parse_dated_window(&words[start..start + 5], words.get(start + 5)))
3958}
3959
3960fn parse_dated_window(window: &[&str], trailing: Option<&&str>) -> Option<Timestamp> {
3966 if trailing.is_some_and(|next| next.starts_with('(')) {
3967 return None;
3968 }
3969 let month = month_number(window[0])?;
3970 let day_token = window[1].strip_suffix(',')?.to_lowercase();
3971 let day_digits = ["st", "nd", "rd", "th"]
3972 .iter()
3973 .find_map(|suffix| day_token.strip_suffix(*suffix))?;
3974 let day: i8 = day_digits.parse().ok()?;
3975 let year_token = window[2];
3976 if year_token.len() != 4 || !year_token.bytes().all(|b| b.is_ascii_digit()) {
3977 return None;
3978 }
3979 let year: i16 = year_token.parse().ok()?;
3980 let ampm = window[4].trim_matches(|c: char| !c.is_ascii_alphabetic());
3984 let (hour, minute) = parse_12h_clock(&format!("{}{}", window[3], ampm))?;
3985 let date = jiff::civil::Date::new(year, month, day).ok()?;
3986 let candidate = date
3987 .at(hour, minute, 0, 0)
3988 .to_zoned(jiff::tz::TimeZone::UTC)
3989 .ok()?;
3990 Some(candidate.timestamp())
3991}
3992
3993fn month_number(name: &str) -> Option<i8> {
3996 const NAMES: [&str; 12] = [
3997 "jan", "feb", "mar", "apr", "may", "jun", "jul", "aug", "sep", "oct", "nov", "dec",
3998 ];
3999 let lower = name.to_lowercase();
4000 NAMES
4001 .iter()
4002 .position(|n| *n == lower.as_str())
4003 .map(|i| i as i8 + 1)
4004}
4005
4006fn exhausted_review_budget(state: &RunState) -> bool {
4018 state.status == RunStatus::Blocked
4021 && !state.pr_publication_pending
4022 && state.reviews.len() >= state.config.graph.review_rounds
4023}
4024
4025fn unfinished_run(runs: &[String], short: &str) -> Option<String> {
4062 unfinished_run_with(runs, short, RunState::load)
4063}
4064
4065fn unfinished_run_with<F>(runs: &[String], short: &str, load: F) -> Option<String>
4068where
4069 F: FnOnce(&str) -> Result<RunState>,
4070{
4071 let id = runs.last()?;
4072 match load(id) {
4073 Ok(s)
4078 if s.status.resumable()
4079 && !s.released()
4080 && !exhausted_review_budget(&s)
4081 && s.liveness(false) != crate::run::Liveness::Live =>
4082 {
4083 Some(id.clone())
4084 }
4085 Ok(_) => None,
4086 Err(e) => {
4087 tracing::warn!("could not read run {id} for task {short}: {e:#}");
4088 None
4089 }
4090 }
4091}
4092
4093fn awaiting_resume_with<F>(task: &Task, load: F) -> bool
4101where
4102 F: FnOnce(&str) -> Result<RunState>,
4103{
4104 if !matches!(task.status, TaskStatus::Parked | TaskStatus::Failed)
4106 || task.fresh_start
4107 || task.review_branch.is_some()
4108 {
4109 return false;
4110 }
4111 let Some(id) = task.runs.last() else {
4112 return false;
4113 };
4114 load(id).is_ok_and(|s| {
4115 s.parked && s.status.resumable() && !s.released() && !exhausted_review_budget(&s)
4116 })
4117}
4118
4119#[derive(Debug, Clone, PartialEq, Eq)]
4122enum Starter {
4123 Review(String),
4126 Resume(String),
4128 Start,
4130}
4131
4132fn selected_starter(
4134 selected: &crate::pr_recovery::Selection,
4135 previous: Option<&RunState>,
4136 unfinished: Option<&str>,
4137) -> Starter {
4138 if let Some(id) = unfinished
4139 && previous.is_some_and(|r| {
4140 r.id == id && r.pr_recovery.as_ref() == Some(selected) && r.reviewed_commits.is_some()
4141 })
4142 {
4143 Starter::Resume(id.to_owned())
4144 } else {
4145 Starter::Review(selected.branch.clone())
4146 }
4147}
4148
4149fn take_divergence_answer(
4153 branch: &str,
4154 remote: &str,
4155 task: &mut Task,
4156) -> Option<crate::reconcile::Choice> {
4157 let summary = crate::reconcile::summary_for(branch, remote);
4158 let (idx, choice) = task.answers.iter().enumerate().rev().find_map(|(i, a)| {
4159 (a.question == summary)
4160 .then(|| crate::reconcile::Choice::from_answer(&a.answer))
4161 .flatten()
4162 .map(|c| (i, c))
4163 })?;
4164 task.answers.remove(idx);
4165 Some(choice)
4166}
4167
4168fn choose_starter(
4180 review_branch: Option<&str>,
4181 branch_exists: bool,
4182 unfinished: Option<&str>,
4183) -> Starter {
4184 match review_branch {
4185 Some(branch) if branch_exists => Starter::Review(branch.to_owned()),
4186 Some(_) => Starter::Start,
4187 None => match unfinished {
4188 Some(id) => Starter::Resume(id.to_owned()),
4189 None => Starter::Start,
4190 },
4191 }
4192}
4193
4194fn take_review_branch(task: &mut Task, previous: Option<&RunState>) -> Option<String> {
4196 let branch = task.review_branch.take();
4197 let instruction = task.review_instruction.take();
4198 branch.filter(|branch| match &instruction {
4199 Some(authored) => *authored == task.instruction,
4200 None => previous.map_or_else(
4203 || task.review_of.as_ref() == Some(branch),
4204 |run| compatible_run_instruction(run, task),
4205 ),
4206 })
4207}
4208
4209fn may_resume_run(run: &RunState, task: &Task) -> bool {
4212 task.resume_override
4213 .as_ref()
4214 .is_some_and(|r| r.pinned_run.as_ref() == Some(&run.id))
4215 || compatible_run_instruction(run, task)
4216}
4217
4218pub(crate) fn compatible_run_instruction(run: &RunState, task: &Task) -> bool {
4220 if let Some(authored) = &run.task_instruction {
4221 return compatible_instruction(authored, task);
4222 }
4223 if compatible_instruction(&run.instruction, task) {
4224 return true;
4225 }
4226 if run.reviewed_commits.is_some() {
4229 if let Some((authored, _)) = run.instruction.split_once("\n\n# Work already on branch `") {
4230 return strip_answers_block(authored, task).trim() == task.instruction.trim();
4231 }
4232 if let Some(branch) = &task.review_of {
4233 let canonical = format!(
4234 "Review the branch `{branch}` (review-only: no implementation, no judging)."
4235 );
4236 let generated = format!(
4237 "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"
4238 );
4239 return task.instruction == canonical && run.instruction.starts_with(&generated);
4240 }
4241 }
4242 false
4243}
4244
4245pub(crate) fn compatible_instruction(old: &str, task: &Task) -> bool {
4247 strip_answers_block(old, task) == task.instruction
4248}
4249
4250fn repo_for(task: &Task, fallback: &Path) -> PathBuf {
4253 if task.repo.as_os_str().is_empty() || task.repo == Path::new(".") {
4254 return fallback.to_path_buf();
4255 }
4256 task.repo.clone()
4257}
4258
4259const ANSWERS_HEADER: &str = "\n\n# Operator answers\n\n";
4263
4264fn answers_block(task: &Task, count: usize) -> String {
4266 let mut s = ANSWERS_HEADER.to_owned();
4267 for a in &task.answers[..count] {
4268 s.push_str(&format!("- {}: {}\n", a.question, a.answer));
4269 }
4270 s
4271}
4272
4273fn resumption_block(r: &crate::queue::Resumption) -> String {
4276 format!(
4277 "\n\n# Explicit operator resumption\n\nGeneration {} at {} approved this instruction:\n{}\n\
4278 Earlier answers (first {}) are audit history; their waiting premises are superseded by this approval.\n",
4279 r.generation, r.at, r.instruction, r.answers_through,
4280 )
4281}
4282
4283fn append_answers(base: &str, task: &Task) -> String {
4285 let mut s = base.to_owned();
4286 if !task.answers.is_empty() {
4287 s.push_str(&answers_block(task, task.answers.len()));
4288 }
4289 if let Some(r) = &task.resumption {
4290 s.push_str(&resumption_block(r));
4291 }
4292 s
4293}
4294
4295fn strip_answers_block<'a>(instruction: &'a str, task: &Task) -> &'a str {
4299 let mut instruction = instruction;
4300 for approval in task
4301 .resumption
4302 .iter()
4303 .chain(task.resumption_history.iter().rev())
4304 {
4305 if let Some(base) = instruction.strip_suffix(&resumption_block(approval)) {
4306 instruction = base;
4307 break;
4308 }
4309 }
4310 for count in (1..=task.answers.len()).rev() {
4311 let block = answers_block(task, count);
4312 if let Some(base) = instruction.strip_suffix(&block) {
4313 return base;
4314 }
4315 }
4316 instruction
4317}
4318
4319fn instruction_for(task: &Task) -> String {
4327 append_answers(&task.instruction, task)
4328}
4329
4330fn resumed_instruction(old_instruction: &str, task: &Task) -> String {
4342 append_answers(strip_answers_block(old_instruction, task), task)
4343}
4344
4345fn task_attachments(queue: &Queue, task: &Task) -> Result<Vec<PathBuf>> {
4347 let paths = queue.attachment_paths(task);
4348 for (name, path) in task.attachments.iter().zip(&paths) {
4349 if !path.is_file() {
4350 bail!(
4351 "attachment `{name}` is recorded on the task but {} is missing",
4352 path.display()
4353 );
4354 }
4355 }
4356 Ok(paths)
4357}
4358
4359fn prepare_instruction(
4370 starter: &Starter,
4371 old_instruction: Option<&str>,
4372 task: &Task,
4373) -> Option<String> {
4374 match starter {
4375 Starter::Start => Some(instruction_for(task)),
4376 Starter::Resume(_) => Some(resumed_instruction(
4377 old_instruction.expect("a resumed run always has a prior instruction"),
4378 task,
4379 )),
4380 Starter::Review(_) => None,
4381 }
4382}
4383
4384fn record(queue: &Queue, task: &mut Task) {
4388 if let Err(e) = queue.put(task) {
4389 tracing::error!("could not record task {}: {e:#}", task.short());
4390 let notice = Notice::error(
4391 "loop:record",
4392 "The loop could not save a task's state; check the disk.",
4393 )
4394 .link(Link::Task {
4395 id: task.id.clone(),
4396 });
4397 match queue.root().parent().filter(|p| !p.as_os_str().is_empty()) {
4399 Some(home) => notices::raise_in(home, notice),
4400 None => notices::raise(notice),
4401 }
4402 }
4403}
4404
4405fn runnable(queue: &Queue) -> Vec<Task> {
4411 let mut tasks: Vec<Task> = queue
4412 .list()
4413 .into_iter()
4414 .filter(|t| t.status.runnable())
4415 .collect();
4416 tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
4417 tasks
4418}
4419
4420fn describe(state: &RunState) -> String {
4434 let p = phrases(&state.config.graph.language);
4435 let mut detail = if state.status == RunStatus::Stalled {
4436 let mut seats: Vec<&str> = state.quota.iter().map(|q| q.seat.as_str()).collect();
4437 seats.sort_unstable();
4438 seats.dedup();
4439 if seats.is_empty() {
4440 p.quorum_lost.to_owned()
4441 } else {
4442 format!("{}{}{}", p.quorum_lost, p.quota_took_out, seats.join(", "))
4443 }
4444 } else {
4445 format!("{}{}", p.run_ended, state.status.display_label())
4446 };
4447 if let Some(last) = state.events.last() {
4448 let unanswered = state
4452 .reviews
4453 .last()
4454 .filter(|r| {
4455 state.status == RunStatus::Blocked
4456 && last.node == "review"
4457 && r.incomplete()
4458 && r.blocking == 0
4459 && r.round == state.config.graph.review_rounds
4460 && r.e2e.iter().all(crate::run::CommandOutcome::ok)
4461 })
4462 .map(|r| (r.expected - r.answered, r.round));
4463 match unanswered {
4464 Some((missing, rounds)) if crate::lang::is_japanese(&state.config.graph.language) => {
4465 detail.push_str(&format!(
4466 " ({}: {})",
4467 last.node,
4468 (p.reviewers_never_answered)(missing, rounds)
4469 ));
4470 }
4471 _ => detail.push_str(&format!(" ({}: {})", last.node, last.message)),
4472 }
4473 }
4474 detail.push_str(&format!(" [run {}]", state.id));
4475 detail
4476}
4477
4478const DIAGNOSTIC_MAX: usize = 4_000;
4484
4485const DIAGNOSTIC_OUTPUT_TAIL: usize = 800;
4490
4491fn diagnostic(state: &RunState) -> Option<String> {
4505 let mut parts: Vec<String> = Vec::new();
4506
4507 for o in state.gate.iter().filter(|o| !o.ok()) {
4509 parts.push(format!(
4510 "gate `{}` failed ({:?}):\n{}",
4511 o.command,
4512 o.code,
4513 crate::run::tail(&o.output_tail, DIAGNOSTIC_OUTPUT_TAIL)
4514 ));
4515 }
4516
4517 if let Some(last) = state
4520 .events
4521 .iter()
4522 .rev()
4523 .find(|e| e.node == "land" && e.message.contains("fixer produced no commit"))
4524 {
4525 parts.push(last.message.clone());
4526 }
4527
4528 if state.viable().is_empty() {
4535 for c in &state.candidates {
4536 if let Some(evidence) = &c.verified_noop {
4537 parts.push(format!(
4538 "candidate {} (agent-verified no-op, unconfirmed by magi): {evidence}",
4539 c.label
4540 ));
4541 } else if !c.summary.trim().is_empty() {
4542 parts.push(format!("candidate {}: {}", c.label, c.summary.trim()));
4543 } else if let Some(why) = &c.failed {
4544 parts.push(format!("candidate {}: {why}", c.label));
4545 }
4546 }
4547 }
4548
4549 if parts.is_empty() {
4550 return None;
4551 }
4552 Some(crate::run::tail(
4557 &parts.join("\n\n"),
4558 DIAGNOSTIC_MAX.saturating_sub(100),
4559 ))
4560}
4561
4562fn label(status: RunStatus) -> &'static str {
4570 status.as_str()
4571}
4572
4573pub(crate) fn merge_mode(mode: &str) -> Result<MergeMode> {
4575 match mode {
4576 "none" => Ok(MergeMode::None),
4577 "local" => Ok(MergeMode::Local),
4578 "pr" => Ok(MergeMode::Pr),
4579 other => bail!("unknown merge mode `{other}`; expected none, local or pr"),
4580 }
4581}
4582
4583fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
4588 mutex
4589 .lock()
4590 .unwrap_or_else(std::sync::PoisonError::into_inner)
4591}
4592
4593#[cfg(test)]
4594mod tests {
4595 use super::*;
4596 use crate::queue::{Source, TaskStatus};
4597 use crate::run::{Candidate, CommandOutcome};
4598 use pretty_assertions::assert_eq;
4599
4600 fn task() -> Task {
4601 Task::new(
4602 "add retries".to_owned(),
4603 "add retries".to_owned(),
4604 PathBuf::from("/repo"),
4605 Source::Human,
4606 )
4607 }
4608
4609 #[test]
4610 fn selected_public_head_overrides_historical_resume_and_pin() {
4611 let mut t = task();
4612 let mut old = run_state(RunStatus::Blocked);
4613 let selection = crate::pr_recovery::Selection {
4614 url: "url".into(),
4615 branch: format!("magi/task-{}", t.id),
4616 head: "a".repeat(40),
4617 instruction: t.instruction.clone(),
4618 approved_at: Timestamp::now(),
4619 };
4620 t.resume_override = Some(crate::queue::OperatorResume {
4621 pinned_run: Some(old.id.clone()),
4622 question_id: "choice".into(),
4623 at: Timestamp::now(),
4624 conductor_rehold: None,
4625 forced: false,
4626 });
4627 assert_eq!(
4628 selected_starter(&selection, Some(&old), Some(&old.id)),
4629 Starter::Review(selection.branch.clone())
4630 );
4631 old.pr_recovery = Some(selection.clone());
4632 assert_eq!(
4633 selected_starter(&selection, Some(&old), Some(&old.id)),
4634 Starter::Review(selection.branch.clone())
4635 );
4636 old.reviewed_commits = Some(vec!["implementation".into()]);
4637 assert_eq!(
4638 selected_starter(&selection, Some(&old), Some(&old.id)),
4639 Starter::Resume(old.id.clone())
4640 );
4641 let mut changed = selection.clone();
4642 changed.head = "b".repeat(40);
4643 assert_eq!(
4644 selected_starter(&changed, Some(&old), Some(&old.id)),
4645 Starter::Review(changed.branch)
4646 );
4647 }
4648
4649 #[test]
4650 fn candidate_transplant_failure_is_one_actionable_hold() {
4651 let dir = tempfile::tempdir().unwrap();
4652 let q = Queue::at(dir.path().join("queue"));
4653 let mut t = task();
4654 let mut state = run_state(RunStatus::Blocked);
4655 state.event("prep", "cannot apply referenced commit: conflict");
4656 state.prep_failure = Some("conflict".into());
4657 t.start(state.id.clone());
4658 finish_attempt(3, &q, &mut t, &state, &[], Err(anyhow::anyhow!("conflict")));
4659 assert_eq!(t.attempts, 1);
4660 assert_eq!(t.status, TaskStatus::Held);
4661 assert!(t.operator_held());
4662 assert!(!t.status.runnable());
4663 assert!(t.hold_reason.as_deref().unwrap().contains("recover-pr"));
4664 }
4665
4666 #[test]
4667 fn revised_resumption_invalidates_old_pins_and_preserves_unchanged_pins() {
4668 for source in ["cli", "web"] {
4669 let mut t = task();
4670 let mut run = run_state(RunStatus::Blocked);
4671 run.instruction = "Run A original criteria".into();
4672 run.task_instruction = Some(run.instruction.clone());
4673 t.start(run.id.clone());
4674 t.release_operator("choice", Some("resume-question"));
4675 t.resume_override = Some(crate::queue::OperatorResume {
4676 question_id: "resume-question".into(),
4677 at: Timestamp::now(),
4678 conductor_rehold: None,
4679 forced: true,
4680 pinned_run: Some(run.id.clone()),
4681 });
4682 t.hold_machine(Some("not enough free space to start a run".into()));
4683 let mut unchanged = t.clone();
4684 unchanged
4685 .edit("title changed".into(), unchanged.instruction.clone())
4686 .unwrap();
4687 unchanged.release_operator(source, None);
4688 assert_eq!(
4689 unchanged
4690 .resume_override
4691 .as_ref()
4692 .unwrap()
4693 .pinned_run
4694 .as_deref(),
4695 Some(run.id.as_str())
4696 );
4697 assert!(!compatible_run_instruction(&run, &unchanged));
4698 assert!(may_resume_run(&run, &unchanged));
4699
4700 t.edit(
4701 "approved recovery".into(),
4702 "Reuse measured PR head B; fix Clippy; review".into(),
4703 )
4704 .unwrap();
4705 assert!(t.resume_override.is_none());
4706 t.release_operator(source, None);
4707 let reloaded: Task = serde_json::from_str(&serde_json::to_string(&t).unwrap()).unwrap();
4708 assert!(reloaded.resume_override.is_none());
4709 assert!(!may_resume_run(&run, &reloaded));
4710 let unfinished =
4711 unfinished_run_with(&reloaded.runs, reloaded.short(), |_| Ok(run.clone()))
4712 .filter(|_| may_resume_run(&run, &reloaded));
4713 assert_eq!(
4714 choose_starter(None, false, unfinished.as_deref()),
4715 Starter::Start
4716 );
4717 assert!(instruction_for(&reloaded).starts_with("Reuse measured PR head B"));
4718 assert_eq!(
4719 reloaded.resumption_history[0].question_id.as_deref(),
4720 Some("resume-question")
4721 );
4722
4723 unchanged.instruction = "Reuse measured PR head B".into();
4726 unchanged.hold_machine(Some("disk hold".into()));
4727 unchanged.release_operator(source, None);
4728 assert!(unchanged.resume_override.is_none());
4729 assert!(!may_resume_run(&run, &unchanged));
4730 }
4731 }
4732
4733 #[test]
4734 fn resumption_preserves_generated_review_runs_and_rejects_revised_targets() {
4735 let mut t = task();
4736 let mut run = run_state(RunStatus::Blocked);
4737 t.review_of = Some("approved-head".into());
4738 t.instruction =
4739 "Review the branch `approved-head` (review-only: no implementation, no judging)."
4740 .into();
4741 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();
4742 run.reviewed_commits = Some(vec!["Commit log".into()]);
4743 t.start(run.id.clone());
4744 t.hold_machine(Some("interrupted review".into()));
4745 t.release_operator("cli", None);
4746 assert!(
4747 compatible_run_instruction(&run, &t),
4748 "legacy explicit review remains resumable"
4749 );
4750 let unfinished = unfinished_run_with(&t.runs, t.short(), |_| Ok(run.clone()))
4751 .filter(|_| compatible_run_instruction(&run, &t));
4752 assert_eq!(
4753 choose_starter(None, false, unfinished.as_deref()),
4754 Starter::Resume(run.id.clone())
4755 );
4756 run.task_instruction = Some(t.instruction.clone());
4757 assert!(compatible_run_instruction(&run, &t));
4758 let reloaded: RunState =
4759 serde_json::from_str(&serde_json::to_string(&run).unwrap()).unwrap();
4760 assert!(compatible_run_instruction(&reloaded, &t));
4761 t.edit(
4762 "new approved head".into(),
4763 "Reuse measured PR head B".into(),
4764 )
4765 .unwrap();
4766 assert!(!compatible_run_instruction(&run, &t));
4767
4768 t.review_of = None;
4770 t.instruction = "Fix Clippy on approved head".into();
4771 run.task_instruction = None;
4772 run.instruction = format!(
4773 "{}\n\n# Work already on branch `approved-head` (commit subjects)\n\nCommit log",
4774 t.instruction
4775 );
4776 assert!(compatible_run_instruction(&run, &t));
4777 t.instruction.push('\n');
4778 assert!(
4779 compatible_run_instruction(&run, &t),
4780 "legacy generated review trimmed the original criteria"
4781 );
4782 t.hold_for_handover(Some("approved-head".into()), "refused".into());
4783 let mut unchanged = t.clone();
4784 assert_eq!(
4785 take_review_branch(&mut unchanged, Some(&run)).as_deref(),
4786 Some("approved-head")
4787 );
4788 t.edit("new plan".into(), "Reuse measured PR head B".into())
4789 .unwrap();
4790 assert!(take_review_branch(&mut t, Some(&run)).is_none());
4791 assert!(!compatible_run_instruction(&run, &t));
4792 t.review_branch = Some("approved-head".into());
4793 t.review_instruction = None;
4794 assert!(
4795 take_review_branch(&mut t, Some(&run)).is_none(),
4796 "legacy retained candidate cannot replace a revised approved head"
4797 );
4798 }
4799
4800 #[test]
4801 fn resumption_actions_survive_automatic_holds_of_hand_started_runs() {
4802 crate::run::pin_test_home();
4803 let dir = tempfile::tempdir().unwrap();
4804 let queue = Queue::at(dir.path().join("queue"));
4805 for resume in [true, false] {
4806 let state = run_state(RunStatus::Blocked);
4807 let mut t = task();
4808 t.start(state.id.clone());
4809 let action = if resume {
4810 ask::ChoiceAction::Resume {
4811 run: state.id.clone(),
4812 }
4813 } else {
4814 ask::ChoiceAction::Requeue
4815 };
4816 let mut q = action_question(&state.id, action);
4817 q.status = ask::QuestionStatus::Open;
4818 q.answer = None;
4819 q.answered_at = None;
4820 t.fail("run stopped with a question", 3);
4821 hold_if_runnable(&queue, &mut t);
4822 assert!(t.operator_decided_at.is_none());
4823 assert!(t.operator_held());
4824 q.answer(ask::Answer::Choice(q.choices[0].clone())).unwrap();
4825 let expected = if resume {
4826 ActionDecision::Resume(state.id.clone())
4827 } else {
4828 ActionDecision::Requeue
4829 };
4830 assert_eq!(
4831 decide_action(&t, &q, phrases("en"), |_| Ok(state.clone())),
4832 expected
4833 );
4834 t.hold_manual(Some("later explicit hold".into()));
4835 assert_eq!(
4836 decide_action(&t, &q, phrases("en"), |_| Ok(state.clone())),
4837 ActionDecision::Stale
4838 );
4839 }
4840 }
4841
4842 #[test]
4843 fn explicit_resumption_keeps_revised_instruction_and_audit_answers() {
4844 let mut t = task();
4845 t.answers.push(crate::queue::AnsweredQuestion {
4846 question: "Inspect the remote?".into(),
4847 answer: "Inspect then wait".into(),
4848 });
4849 let old = instruction_for(&t);
4850 assert!(compatible_instruction(&old, &t));
4851 t.instruction = "Reuse approved PR head and review".into();
4852 t.release_operator("cli", None);
4853 assert!(!compatible_instruction(&old, &t));
4854 let approved = instruction_for(&t);
4855 assert!(approved.contains("waiting premises are superseded"));
4856 assert!(compatible_instruction(&approved, &t));
4857 assert_eq!(resumed_instruction(&approved, &t), approved);
4858 t.release_operator("web", None);
4859 assert!(compatible_instruction(&approved, &t));
4860 assert_eq!(t.resumption_history.len(), 1);
4861 t.start("run".into());
4862 assert!(instruction_for(&t).contains("Explicit operator resumption"));
4863 }
4864
4865 #[test]
4869 fn a_resource_blocked_run_holds_its_task_and_spends_no_attempt() {
4870 let mut state = RunState::new(
4871 PathBuf::from("/repo"),
4872 "main".to_owned(),
4873 "abc1234".to_owned(),
4874 "task".to_owned(),
4875 Config::default(),
4876 );
4877 state.status = RunStatus::Blocked;
4878 state.resource_block = Some(crate::run::ResourceBlock {
4879 node: "graft".to_owned(),
4880 kind: "shared build cache".to_owned(),
4881 detail: "timed out".to_owned(),
4882 at: jiff::Timestamp::now(),
4883 });
4884 let mut t = task();
4885 t.start("20260902-000000-aaaa".to_owned());
4886 assert_eq!(t.attempts, 1);
4887 let verdict = Verdict {
4888 status: RunStatus::Blocked,
4889 left_pr: false,
4890 quota_hit: false,
4891 parked: false,
4892 no_viable_candidates: false,
4893 };
4894 settle_and_diagnose(&mut t, verdict, "blocked", 2, &state);
4895 assert_eq!(t.status, TaskStatus::Held);
4896 assert_eq!(t.attempts, 0, "the attempt is refunded");
4897 assert!(t.hold_reason.as_deref().unwrap().starts_with("[resource] "));
4898 assert!(
4899 !t.operator_held(),
4900 "a machine hold, released without a human"
4901 );
4902
4903 let mut p = task();
4905 p.start("20260902-000000-bbbb".to_owned());
4906 settle_and_diagnose(
4907 &mut p,
4908 Verdict {
4909 parked: true,
4910 ..verdict
4911 },
4912 "parked",
4913 2,
4914 &state,
4915 );
4916 assert_eq!(p.status, TaskStatus::Parked);
4917 }
4918
4919 #[test]
4920 fn record_failure_notice_links_the_task() {
4921 let dir = tempfile::tempdir().unwrap();
4922 std::fs::write(dir.path().join("queue"), b"not a directory").unwrap();
4925 let queue = Queue::at(dir.path().join("queue"));
4926 let mut t = interrupt_task("rec-fail-1");
4927 record(&queue, &mut t);
4928
4929 let found = crate::notices::Notices::at(dir.path().join("notifications")).list();
4930 assert_eq!(found.len(), 1);
4931 assert_eq!(found[0].key, "loop:record");
4932 assert_eq!(
4933 found[0].link,
4934 Some(Link::Task {
4935 id: "rec-fail-1".to_owned()
4936 })
4937 );
4938 }
4939
4940 fn interrupt_task(id: &str) -> Task {
4943 let mut t = task();
4944 t.id = id.to_owned();
4945 t.interrupt = true;
4946 t
4947 }
4948
4949 fn task_with_id(id: &str) -> Task {
4951 let mut t = task();
4952 t.id = id.to_owned();
4953 t
4954 }
4955
4956 fn urgent_task(id: &str) -> Task {
4958 let mut t = task();
4959 t.id = id.to_owned();
4960 t.urgent = true;
4961 t
4962 }
4963
4964 #[test]
4965 fn urgent_dispatch_has_one_extra_slot_and_never_uses_a_stale_flag() {
4966 let dir = tempfile::tempdir().unwrap();
4967 let queue = Queue::at(dir.path().join("queue"));
4968 let ordinary = Arc::new(tokio::sync::Semaphore::new(1));
4969 let urgent = Arc::new(tokio::sync::Semaphore::new(1));
4970 let occupied = Arc::clone(&ordinary).try_acquire_owned().unwrap();
4971 let mut first = task();
4972 queue.put(&mut first).unwrap();
4973 let stale_plain = queue.get(&first.id).unwrap();
4974 first.set_urgent(true).unwrap();
4975 queue.put(&mut first).unwrap();
4976 let dispatched = claim_dispatch(&queue, &stale_plain, false, &ordinary, &urgent).unwrap();
4977 let mut second = urgent_task("second");
4978 queue.put(&mut second).unwrap();
4979 assert!(claim_dispatch(&queue, &second, false, &ordinary, &urgent).is_none());
4980 drop(queue.claim(&second.id).unwrap());
4981 assert_eq!(queue.get(&second.id).unwrap().attempts, 0);
4982 drop(dispatched);
4983 let stale_urgent = queue.get(&first.id).unwrap();
4984 first.set_urgent(false).unwrap();
4985 queue.put(&mut first).unwrap();
4986 assert!(claim_dispatch(&queue, &stale_urgent, false, &ordinary, &urgent).is_none());
4987 assert_eq!(urgent.available_permits(), 1);
4988 drop(queue.claim(&first.id).unwrap());
4989 drop(occupied);
4990 }
4991
4992 #[test]
4993 fn urgent_dispatch_uses_saved_lane_and_releases_unsuccessful_claims() {
4994 let dir = tempfile::tempdir().unwrap();
4995 let queue = Queue::at(dir.path().join("queue"));
4996 let ordinary = Arc::new(tokio::sync::Semaphore::new(1));
4997 let urgent = Arc::new(tokio::sync::Semaphore::new(1));
4998 let mut t = task();
4999 queue.put(&mut t).unwrap();
5000 for enabled in [true, false] {
5001 let stale = queue.get(&t.id).unwrap();
5002 t.set_urgent(enabled).unwrap();
5003 queue.put(&mut t).unwrap();
5004 assert_ne!(stale.urgent, enabled);
5005 let DispatchClaim {
5006 task: fresh,
5007 claim,
5008 permit,
5009 } = claim_dispatch(&queue, &stale, false, &ordinary, &urgent).unwrap();
5010 assert_eq!(fresh.urgent, enabled);
5011 assert_eq!(ordinary.available_permits(), usize::from(enabled));
5012 assert_eq!(urgent.available_permits(), usize::from(!enabled));
5013 assert!(queue.claim(&t.id).is_err());
5014 assert_eq!(fresh.attempts, stale.attempts);
5015 drop((claim, permit));
5016 assert_eq!(ordinary.available_permits(), 1);
5017 assert_eq!(urgent.available_permits(), 1);
5018 }
5019 t.set_urgent(true).unwrap();
5020 queue.put(&mut t).unwrap();
5021 let occupied = Arc::clone(&urgent).try_acquire_owned().unwrap();
5022 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5023 assert_eq!(ordinary.available_permits(), 1, "no ordinary fallback");
5024 drop(queue.claim(&t.id).unwrap());
5025 assert_eq!(queue.get(&t.id).unwrap().attempts, t.attempts);
5026 drop(occupied);
5027 assert!(claim_dispatch(&queue, &t, true, &ordinary, &urgent).is_none());
5028 drop(queue.claim(&t.id).unwrap());
5029 let held = queue.claim(&t.id).unwrap();
5030 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5031 drop(held);
5032 t.status = TaskStatus::Held;
5033 queue.put(&mut t).unwrap();
5034 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5035 drop(queue.claim(&t.id).unwrap());
5036 std::fs::remove_file(queue.path_of(&t.id)).unwrap();
5037 assert!(claim_dispatch(&queue, &t, false, &ordinary, &urgent).is_none());
5038 drop(queue.claim(&t.id).unwrap());
5039 assert_eq!(urgent.available_permits(), 1);
5040 }
5041
5042 #[test]
5047 fn permit_kind_prefers_a_land_resume_over_the_urgent_slot() {
5048 assert_eq!(permit_kind(true, false), PermitKind::None);
5049 assert_eq!(permit_kind(true, true), PermitKind::None);
5050 }
5051
5052 #[test]
5057 fn permit_kind_separates_urgent_from_ordinary() {
5058 assert_eq!(permit_kind(false, true), PermitKind::Urgent);
5059 assert_eq!(permit_kind(false, false), PermitKind::Ordinary);
5060 }
5061
5062 #[test]
5069 fn disk_gate_with_holds_a_task_below_the_threshold_and_names_both_numbers() {
5070 let cfg = Config::default();
5071 let repo = Path::new("/any/repo/path");
5072
5073 let reason =
5074 disk_gate_with(repo, &cfg, |_| Ok(1024)).expect("must hold below the threshold");
5075 assert!(reason.contains("1024"), "{reason}");
5076 assert!(
5077 reason.contains(&cfg.disk.min_free_bytes.to_string()),
5078 "{reason}"
5079 );
5080
5081 assert_eq!(
5082 disk_gate_with(repo, &cfg, |_| Ok(cfg.disk.min_free_bytes)),
5083 None,
5084 "exactly at the floor is open"
5085 );
5086 assert_eq!(
5087 disk_gate_with(repo, &cfg, |_| Ok(cfg.disk.min_free_bytes + 1)),
5088 None,
5089 "comfortably above the floor is open"
5090 );
5091 }
5092
5093 #[test]
5094 fn disk_gate_with_opens_unconditionally_when_the_operator_opted_out() {
5095 let mut cfg = Config::default();
5096 cfg.disk.min_free_bytes = 0;
5097 let repo = Path::new("/any/repo/path");
5098 assert_eq!(
5099 disk_gate_with(repo, &cfg, |_| Ok(0)),
5100 None,
5101 "a zero floor never measures at all"
5102 );
5103 }
5104
5105 #[test]
5106 fn disk_gate_with_closes_rather_than_starts_blind_when_it_cannot_measure() {
5107 let cfg = Config::default();
5108 let repo = Path::new("/any/repo/path");
5109 let reason = disk_gate_with(repo, &cfg, |_| Err(anyhow::anyhow!("no df on this box")))
5110 .expect("a measurement failure must close the gate, not open it");
5111 assert!(reason.contains("could not measure"), "{reason}");
5112 }
5113
5114 #[test]
5115 fn no_interrupt_task_leaves_the_sequence_idle_even_with_something_in_flight() {
5116 let ordinary = task();
5117 let next = advance_interrupt(
5118 Interrupt::Idle,
5119 std::slice::from_ref(&ordinary.id),
5120 std::slice::from_ref(&ordinary),
5121 );
5122 assert_eq!(next, Interrupt::Idle);
5123 }
5124
5125 #[test]
5126 fn an_interrupt_task_with_nothing_in_flight_never_starts_a_sequence() {
5127 let marked = interrupt_task("marked");
5130 let next = advance_interrupt(Interrupt::Idle, &[], std::slice::from_ref(&marked));
5131 assert_eq!(next, Interrupt::Idle);
5132 }
5133
5134 #[test]
5135 fn an_interrupt_task_with_something_in_flight_starts_parking_it() {
5136 let marked = interrupt_task("marked");
5137 let next = advance_interrupt(
5138 Interrupt::Idle,
5139 &["running".to_owned()],
5140 std::slice::from_ref(&marked),
5141 );
5142 assert_eq!(
5143 next,
5144 Interrupt::Parking {
5145 parked: vec!["running".to_owned()],
5146 interrupt_task: "marked".to_owned(),
5147 }
5148 );
5149 }
5150
5151 #[test]
5160 fn more_than_one_run_in_flight_never_starts_an_interrupt_sequence() {
5161 let marked = interrupt_task("marked");
5162
5163 let two = advance_interrupt(
5164 Interrupt::Idle,
5165 &["a".to_owned(), "b".to_owned()],
5166 std::slice::from_ref(&marked),
5167 );
5168 assert_eq!(two, Interrupt::Idle);
5169
5170 let none = advance_interrupt(Interrupt::Idle, &[], std::slice::from_ref(&marked));
5171 assert_eq!(none, Interrupt::Idle, "nothing to interrupt either");
5172 }
5173
5174 #[test]
5175 fn parking_holds_until_every_parked_id_has_actually_left_flight() {
5176 let state = Interrupt::Parking {
5177 parked: vec!["running".to_owned()],
5178 interrupt_task: "marked".to_owned(),
5179 };
5180 let still_going = advance_interrupt(state.clone(), &["running".to_owned()], &[]);
5182 assert_eq!(still_going, state);
5183
5184 let stopped_but_not_yet_dispatched =
5188 advance_interrupt(state.clone(), &[], &[interrupt_task("marked")]);
5189 assert_eq!(stopped_but_not_yet_dispatched, state);
5190
5191 let dispatched = advance_interrupt(state, &["marked".to_owned()], &[]);
5193 assert_eq!(
5194 dispatched,
5195 Interrupt::Running {
5196 parked: vec!["running".to_owned()],
5197 interrupt_task: "marked".to_owned(),
5198 }
5199 );
5200 }
5201
5202 #[test]
5203 fn the_sequence_moves_to_resuming_the_instant_the_interrupt_tasks_own_run_leaves_flight() {
5204 let state = Interrupt::Running {
5205 parked: vec!["running".to_owned()],
5206 interrupt_task: "marked".to_owned(),
5207 };
5208 let still_running = advance_interrupt(state.clone(), &["marked".to_owned()], &[]);
5209 assert_eq!(still_running, state);
5210
5211 let ended = advance_interrupt(state, &[], &[task_with_id("running")]);
5218 assert_eq!(
5219 ended,
5220 Interrupt::Resuming {
5221 parked: vec!["running".to_owned()]
5222 }
5223 );
5224 }
5225
5226 #[test]
5227 fn resuming_ends_the_instant_a_parked_task_is_seen_in_flight() {
5228 let state = Interrupt::Resuming {
5229 parked: vec!["running".to_owned()],
5230 };
5231 let still_waiting = advance_interrupt(state.clone(), &[], &[task_with_id("running")]);
5232 assert_eq!(still_waiting, state);
5233
5234 let dispatched = advance_interrupt(state, &["running".to_owned()], &[]);
5235 assert_eq!(dispatched, Interrupt::Idle);
5236 }
5237
5238 #[test]
5244 fn an_interrupt_task_that_stops_being_runnable_abandons_the_wait_without_losing_the_parked_run()
5245 {
5246 let state = Interrupt::Parking {
5247 parked: vec!["running".to_owned()],
5248 interrupt_task: "marked".to_owned(),
5249 };
5250 let next = advance_interrupt(state, &[], &[]);
5253 assert_eq!(
5254 next,
5255 Interrupt::Resuming {
5256 parked: vec!["running".to_owned()]
5257 },
5258 "abandoning the interrupt must not abandon the resume it owes"
5259 );
5260 }
5261
5262 #[test]
5265 fn resuming_abandons_a_parked_task_that_stops_being_runnable() {
5266 let state = Interrupt::Resuming {
5267 parked: vec!["running".to_owned()],
5268 };
5269 let next = advance_interrupt(state, &[], &[]);
5270 assert_eq!(
5271 next,
5272 Interrupt::Idle,
5273 "nothing is left to wait for; the loop must not stay wedged"
5274 );
5275 }
5276
5277 #[test]
5278 fn disabled_by_config_the_sequence_can_never_leave_idle() {
5279 let marked = interrupt_task("marked");
5280 let next = advance_interrupt_tick(
5281 false,
5282 Interrupt::Idle,
5283 &["running".to_owned()],
5284 std::slice::from_ref(&marked),
5285 );
5286 assert_eq!(
5287 next,
5288 Interrupt::Idle,
5289 "an unmarked, unconfigured daemon must behave exactly as before"
5290 );
5291 }
5292
5293 #[test]
5294 fn the_gate_blocks_everyone_while_something_parked_is_still_in_flight() {
5295 let state = Interrupt::Parking {
5296 parked: vec!["running".to_owned()],
5297 interrupt_task: "marked".to_owned(),
5298 };
5299 let candidates = vec![interrupt_task("marked"), task()];
5300 let allowed = interrupt_gate(&state, &["running".to_owned()], candidates);
5301 assert!(
5302 allowed.is_empty(),
5303 "nothing may dispatch - not even the interrupt task itself - \
5304 until the parked run has actually stopped"
5305 );
5306 }
5307
5308 #[test]
5322 fn urgent_gains_no_exemption_from_an_active_interrupt_sequence() {
5323 for state in [
5324 Interrupt::Parking {
5325 parked: vec!["running".to_owned()],
5326 interrupt_task: "marked".to_owned(),
5327 },
5328 Interrupt::Running {
5329 parked: vec!["running".to_owned()],
5330 interrupt_task: "marked".to_owned(),
5331 },
5332 Interrupt::Resuming {
5333 parked: vec!["running".to_owned()],
5334 },
5335 ] {
5336 let candidates = vec![
5337 interrupt_task("marked"),
5338 urgent_task("hot"),
5339 task_with_id("ordinary"),
5340 ];
5341 let allowed = interrupt_gate(&state, &["running".to_owned()], candidates);
5342 assert!(
5343 !allowed.iter().any(|t| t.id == "hot"),
5344 "an urgent candidate must wait out the same gate as anything \
5345 else while the run it would run alongside has not actually \
5346 left flight, for state {state:?}: {allowed:?}"
5347 );
5348 }
5349 }
5350
5351 #[test]
5357 fn a_task_marked_both_urgent_and_interrupt_is_admitted_once_the_gate_itself_says_so() {
5358 let state = Interrupt::Resuming {
5359 parked: vec!["hot".to_owned()],
5360 };
5361 let candidates = vec![urgent_task("hot"), task()];
5362 let allowed = interrupt_gate(&state, &[], candidates);
5363 assert_eq!(
5364 allowed.iter().filter(|t| t.id == "hot").count(),
5365 1,
5366 "the gate's own decision is unaffected by the urgent flag: {allowed:?}"
5367 );
5368 }
5369
5370 #[test]
5371 fn the_gate_lets_only_the_interrupt_task_through_once_parked_work_has_stopped() {
5372 let state = Interrupt::Parking {
5373 parked: vec!["running".to_owned()],
5374 interrupt_task: "marked".to_owned(),
5375 };
5376 let other = task();
5377 let candidates = vec![interrupt_task("marked"), other.clone()];
5378 let allowed = interrupt_gate(&state, &[], candidates);
5379 assert_eq!(allowed.len(), 1);
5380 assert_eq!(allowed[0].id, "marked");
5381 }
5382
5383 #[test]
5384 fn the_gate_blocks_everyone_while_the_interrupt_task_itself_is_in_flight() {
5385 let state = Interrupt::Running {
5386 parked: vec!["running".to_owned()],
5387 interrupt_task: "marked".to_owned(),
5388 };
5389 let candidates = vec![task(), task()];
5390 let allowed = interrupt_gate(&state, &["marked".to_owned()], candidates);
5391 assert!(allowed.is_empty());
5392 }
5393
5394 #[test]
5401 fn the_gate_offers_at_most_one_candidate_while_resuming_even_with_two_parked() {
5402 let state = Interrupt::Resuming {
5403 parked: vec!["a".to_owned(), "c".to_owned()],
5404 };
5405 let candidates = vec![task_with_id("a"), task_with_id("c"), task_with_id("other")];
5406 let allowed = interrupt_gate(&state, &[], candidates);
5407 assert_eq!(
5408 allowed.len(),
5409 1,
5410 "at most one candidate may be offered while resuming: {allowed:?}"
5411 );
5412 assert_eq!(allowed[0].id, "a");
5413 }
5414
5415 #[test]
5416 fn the_gate_offers_nothing_while_resuming_if_no_parked_task_is_runnable() {
5417 let state = Interrupt::Resuming {
5418 parked: vec!["a".to_owned()],
5419 };
5420 let allowed = interrupt_gate(&state, &[], vec![task_with_id("other")]);
5421 assert!(allowed.is_empty());
5422 }
5423
5424 #[test]
5430 fn a_full_sequence_never_gates_two_runs_through_at_once_and_resumes_exactly_one() {
5431 let running = task(); let marked = interrupt_task("marked");
5433
5434 let mut state = Interrupt::Idle;
5435 let in_flight = vec![running.id.clone()];
5437 state = advance_interrupt_tick(true, state, &in_flight, std::slice::from_ref(&marked));
5438 let gated = interrupt_gate(&state, &in_flight, vec![marked.clone(), running.clone()]);
5439 assert!(gated.is_empty(), "still waiting on `running` to park");
5440
5441 state = advance_interrupt_tick(true, state, &[], &[marked.clone(), running.clone()]);
5443 let gated = interrupt_gate(&state, &[], vec![marked.clone(), running.clone()]);
5444 assert_eq!(
5445 gated.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5446 vec!["marked"],
5447 "only the interrupt task may be offered to the dispatcher now"
5448 );
5449
5450 state = advance_interrupt_tick(
5452 true,
5453 state,
5454 &["marked".to_owned()],
5455 std::slice::from_ref(&running),
5456 );
5457 let gated = interrupt_gate(
5458 &state,
5459 &["marked".to_owned()],
5460 vec![marked.clone(), running.clone()],
5461 );
5462 assert!(
5463 gated.is_empty(),
5464 "the parked run must not be offered back while the interrupt \
5465 task is still running"
5466 );
5467
5468 let other = task_with_id("other");
5472 state = advance_interrupt_tick(true, state, &[], &[running.clone(), other.clone()]);
5473 assert_eq!(
5474 state,
5475 Interrupt::Resuming {
5476 parked: vec![running.id.clone()]
5477 }
5478 );
5479 let gated = interrupt_gate(&state, &[], vec![other.clone(), running.clone()]);
5480 assert_eq!(
5481 gated.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5482 vec![running.id.as_str()],
5483 "exactly the parked run resumes - not the unrelated task, even \
5484 though it was offered first"
5485 );
5486
5487 state = advance_interrupt_tick(
5491 true,
5492 state,
5493 std::slice::from_ref(&running.id),
5494 std::slice::from_ref(&other),
5495 );
5496 assert_eq!(state, Interrupt::Idle);
5497 let gated = interrupt_gate(
5498 &state,
5499 std::slice::from_ref(&running.id),
5500 vec![other.clone()],
5501 );
5502 assert_eq!(
5503 gated.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
5504 vec![other.id.as_str()],
5505 "ordinary dispatch is unrestricted again"
5506 );
5507 }
5508
5509 #[test]
5510 fn every_run_status_settles_the_task_it_came_from() {
5511 let table = [
5513 (RunStatus::Merged, TaskStatus::Done, 1),
5514 (RunStatus::Ready, TaskStatus::Done, 1),
5515 (RunStatus::Stalled, TaskStatus::Failed, 0),
5516 (RunStatus::Blocked, TaskStatus::Failed, 1),
5517 (RunStatus::Failed, TaskStatus::Failed, 1),
5518 (RunStatus::VerifiedNoop, TaskStatus::Held, 1),
5519 (RunStatus::Prep, TaskStatus::Failed, 1),
5520 (RunStatus::Implementing, TaskStatus::Failed, 1),
5521 (RunStatus::Judging, TaskStatus::Failed, 1),
5522 (RunStatus::Deliberating, TaskStatus::Failed, 1),
5523 (RunStatus::Voting, TaskStatus::Failed, 1),
5524 (RunStatus::Reviewing, TaskStatus::Failed, 1),
5525 (RunStatus::Gating, TaskStatus::Failed, 1),
5526 ];
5527 for (run, want, attempts) in table {
5528 let mut t = task();
5529 t.start("20260902-000000-aaaa".to_owned());
5530 settle(
5531 &mut t,
5532 Verdict {
5533 status: run,
5534 left_pr: false,
5535 parked: false,
5536 quota_hit: matches!(run, RunStatus::Stalled),
5537 no_viable_candidates: false,
5538 },
5539 "why",
5540 2,
5541 );
5542 assert_eq!(t.status, want, "task status after {}", label(run));
5543 assert_eq!(t.attempts, attempts, "attempts after {}", label(run));
5544 }
5545 }
5546
5547 #[test]
5548 fn a_quota_stall_costs_the_task_no_attempt_but_a_block_does() {
5549 let mut stalled = task();
5550 stalled.start("20260902-000000-aaaa".to_owned());
5551 settle(
5552 &mut stalled,
5553 Verdict {
5554 status: RunStatus::Stalled,
5555 left_pr: false,
5556 parked: false,
5557 quota_hit: true,
5558 no_viable_candidates: false,
5559 },
5560 "quota",
5561 1,
5562 );
5563 assert_eq!(stalled.attempts, 0);
5564 assert!(
5565 stalled.status.runnable(),
5566 "a machine problem must leave the task in line"
5567 );
5568
5569 let mut blocked = task();
5570 blocked.start("20260902-000000-aaaa".to_owned());
5571 settle(
5572 &mut blocked,
5573 Verdict {
5574 status: RunStatus::Blocked,
5575 left_pr: false,
5576 parked: false,
5577 quota_hit: false,
5578 no_viable_candidates: false,
5579 },
5580 "findings open",
5581 1,
5582 );
5583 assert_eq!(blocked.attempts, 1);
5584 assert_eq!(
5585 blocked.status,
5586 TaskStatus::Held,
5587 "the last attempt hands the task to a human"
5588 );
5589 }
5590
5591 #[test]
5592 fn a_run_that_opened_a_pull_request_is_never_re_competed() {
5593 let mut delivered = task();
5596 delivered.start("20260903-080619-01c2".to_owned());
5597 settle(
5598 &mut delivered,
5599 Verdict {
5600 status: RunStatus::Blocked,
5601 left_pr: true,
5602 parked: false,
5603 quota_hit: false,
5604 no_viable_candidates: false,
5605 },
5606 "no check status",
5607 4,
5608 );
5609 assert_eq!(
5610 delivered.status,
5611 TaskStatus::Held,
5612 "a pull request waiting on CI or a person is not a retryable failure"
5613 );
5614 assert!(
5615 !delivered.status.runnable(),
5616 "the loop must not pick this task up again"
5617 );
5618 assert_eq!(
5619 delivered.last_error.as_deref(),
5620 Some("no check status"),
5621 "the operator needs to be told what the gate was waiting for"
5622 );
5623
5624 let mut empty_handed = task();
5627 empty_handed.start("20260903-080619-01c2".to_owned());
5628 settle(
5629 &mut empty_handed,
5630 Verdict {
5631 status: RunStatus::Blocked,
5632 left_pr: false,
5633 parked: false,
5634 quota_hit: false,
5635 no_viable_candidates: false,
5636 },
5637 "findings open",
5638 4,
5639 );
5640 assert_eq!(empty_handed.status, TaskStatus::Failed);
5641 assert!(empty_handed.status.runnable());
5642 }
5643
5644 #[test]
5645 fn a_verified_noop_run_hands_off_rather_than_closing_or_auto_retrying() {
5646 let mut noop = task();
5652 noop.start("20260912-131304-391f".to_owned());
5653 settle(
5654 &mut noop,
5655 Verdict {
5656 status: RunStatus::VerifiedNoop,
5657 left_pr: false,
5658 parked: false,
5659 quota_hit: false,
5660 no_viable_candidates: true,
5661 },
5662 "candidate A: already fixed by b32cfc4, on main",
5663 4,
5664 );
5665 assert_eq!(
5666 noop.status,
5667 TaskStatus::Held,
5668 "an unverified claim is a request for a human, not a failure"
5669 );
5670 assert!(
5671 !noop.status.runnable(),
5672 "the loop must not requeue this on the same unverified claim"
5673 );
5674 assert_eq!(noop.attempts, 1);
5679 }
5680
5681 #[test]
5682 fn parking_costs_the_task_no_attempt_and_leaves_it_in_line() {
5683 let mut parked = task();
5688 parked.start("20260903-183634-2d98".to_owned());
5689 settle(
5690 &mut parked,
5691 Verdict {
5692 status: RunStatus::Implementing,
5693 left_pr: false,
5694 quota_hit: false,
5695 parked: true,
5696 no_viable_candidates: false,
5697 },
5698 "parked after `implementing`",
5699 2,
5700 );
5701 assert_eq!(parked.attempts, 0, "a park is refunded");
5702 assert!(
5703 parked.status.runnable(),
5704 "and the task stays in line so the next loop resumes its run"
5705 );
5706 assert_eq!(parked.status, TaskStatus::Parked);
5707 assert_eq!(
5708 parked.park_reason.as_deref(),
5709 Some("parked after `implementing`"),
5710 "the card says where it stopped"
5711 );
5712 assert_eq!(parked.last_error, None, "nothing failed");
5713
5714 let mut broken = task();
5718 broken.start("20260903-183634-2d98".to_owned());
5719 settle(
5720 &mut broken,
5721 Verdict {
5722 status: RunStatus::Implementing,
5723 left_pr: false,
5724 quota_hit: false,
5725 parked: false,
5726 no_viable_candidates: false,
5727 },
5728 "returned mid-flight",
5729 2,
5730 );
5731 assert_eq!(broken.attempts, 1);
5732 }
5733
5734 #[test]
5735 fn only_a_rate_limit_buys_the_task_its_attempt_back() {
5736 let mut flaky = task();
5741 flaky.start("20260903-123023-e633".to_owned());
5742 settle(
5743 &mut flaky,
5744 Verdict {
5745 status: RunStatus::Stalled,
5746 left_pr: false,
5747 parked: false,
5748 quota_hit: false,
5749 no_viable_candidates: false,
5750 },
5751 "verdict rests on 1 of 3 judges",
5752 2,
5753 );
5754 assert_eq!(
5755 flaky.attempts, 1,
5756 "flakiness spends an attempt, so `max_attempts` still bounds it"
5757 );
5758 assert!(flaky.status.runnable(), "and it is still worth retrying");
5759
5760 let mut limited = task();
5762 limited.start("20260903-123023-e633".to_owned());
5763 settle(
5764 &mut limited,
5765 Verdict {
5766 status: RunStatus::Stalled,
5767 left_pr: false,
5768 parked: false,
5769 quota_hit: true,
5770 no_viable_candidates: false,
5771 },
5772 "judge-2, judge-3 out of quota",
5773 2,
5774 );
5775 assert_eq!(limited.attempts, 0, "a quota window is refunded");
5776 assert!(limited.status.runnable());
5777
5778 let mut worn = task();
5781 for _ in 0..2 {
5782 worn.release();
5783 }
5784 worn.start("20260903-123023-e633".to_owned());
5785 worn.attempts = 2;
5786 settle(
5787 &mut worn,
5788 Verdict {
5789 status: RunStatus::Stalled,
5790 left_pr: false,
5791 parked: false,
5792 quota_hit: false,
5793 no_viable_candidates: false,
5794 },
5795 "no quorum again",
5796 2,
5797 );
5798 assert_eq!(worn.status, TaskStatus::Held);
5799 assert!(!worn.status.runnable());
5800 }
5801
5802 #[test]
5803 fn a_quota_wipeout_that_leaves_nothing_to_judge_also_costs_no_attempt() {
5804 let mut wiped_out = task();
5811 wiped_out.start("20260907-025000-a1b2".to_owned());
5812 settle(
5813 &mut wiped_out,
5814 Verdict {
5815 status: RunStatus::Failed,
5816 left_pr: false,
5817 parked: false,
5818 quota_hit: true,
5819 no_viable_candidates: true,
5820 },
5821 "no candidate produced a change; nothing to judge",
5822 2,
5823 );
5824 assert_eq!(wiped_out.attempts, 0, "a total quota wipeout is refunded");
5825 assert!(
5826 wiped_out.status.runnable(),
5827 "a machine problem must leave the task in line"
5828 );
5829
5830 let mut partial_progress = task();
5836 partial_progress.start("20260907-025500-c3d4".to_owned());
5837 settle(
5838 &mut partial_progress,
5839 Verdict {
5840 status: RunStatus::Failed,
5841 left_pr: false,
5842 parked: false,
5843 quota_hit: true,
5844 no_viable_candidates: false,
5845 },
5846 "gate failed on the winning candidate",
5847 2,
5848 );
5849 assert_eq!(
5850 partial_progress.attempts, 1,
5851 "a candidate that actually produced a change spends the attempt \
5852 even though some other seat hit its quota"
5853 );
5854 assert!(partial_progress.status.runnable());
5855 }
5856
5857 #[test]
5858 fn reclaim_refunds_a_recovered_quota_wipeout_the_same_way_a_live_settle_does() {
5859 let mut t = task();
5865 t.start("20260907-025000-a1b2".to_owned());
5866 let mut state = run_state(RunStatus::Failed);
5867 state.quota.push(QuotaLoss {
5868 seat: "cand-a".to_owned(),
5869 node: "implement".to_owned(),
5870 at: Timestamp::now(),
5871 reset: None,
5872 });
5873 assert!(
5874 state.viable().is_empty(),
5875 "no candidate was added, so nothing is viable"
5876 );
5877 reclaim(&mut t, Some(state), 2, "en");
5878 assert_eq!(t.attempts, 0, "a recovered quota wipeout is refunded");
5879 assert!(t.status.runnable());
5880 }
5881
5882 #[test]
5883 fn a_held_task_is_never_offered_to_the_loop() {
5884 let dir = tempfile::tempdir().unwrap();
5885 let queue = Queue::at(dir.path().to_path_buf());
5886 for (n, priority) in [(1, 0), (2, 5), (3, 5)] {
5887 let mut t = task();
5888 t.id = format!("2026090{n}-000000-000{n}");
5889 t.priority = priority;
5890 queue.put(&mut t).unwrap();
5891 }
5892 let mut held = task();
5893 held.id = "20260909-000000-9999".to_owned();
5894 held.priority = 99;
5895 held.hold_machine(None);
5896 queue.put(&mut held).unwrap();
5897
5898 let order: Vec<String> = runnable(&queue).into_iter().map(|t| t.id).collect();
5899 assert_eq!(order.len(), 3);
5900 assert!(!order.contains(&held.id));
5901 assert_eq!(
5902 order.first().cloned(),
5903 queue.next_runnable().map(|t| t.id),
5904 "the loop's first candidate is exactly what the queue offers"
5905 );
5906 assert_eq!(
5907 order,
5908 vec![
5909 "20260902-000000-0002".to_owned(),
5910 "20260903-000000-0003".to_owned(),
5911 "20260901-000000-0001".to_owned(),
5912 ],
5913 "priority first, then oldest, so nothing starves"
5914 );
5915 }
5916
5917 #[test]
5918 fn sweep_removes_an_old_unparseable_lock_and_keeps_a_live_one() {
5919 let dir = tempfile::tempdir().unwrap();
5920 let queue = Queue::at(dir.path().to_path_buf());
5921 let mut old = task();
5922 old.id = "20260101-000000-old0".to_owned();
5923 queue.put(&mut old).unwrap();
5924 let mut fresh = task();
5925 fresh.id = "20260101-000000-new0".to_owned();
5926 queue.put(&mut fresh).unwrap();
5927
5928 std::fs::write(dir.path().join(format!("{}.lock", old.id)), "not a pid").unwrap();
5932 std::thread::sleep(Duration::from_millis(60));
5933 let live = queue.claim(&fresh.id).unwrap();
5934
5935 let swept = sweep_stale_claims(&queue, Duration::from_millis(50));
5936 assert_eq!(swept, vec![old.id.clone()]);
5937 assert!(
5938 queue.claim(&old.id).is_ok(),
5939 "an unparseable lock older than the threshold is swept"
5940 );
5941 assert!(
5942 queue.claim(&fresh.id).is_err(),
5943 "a live pid protects its lock regardless of age"
5944 );
5945 drop(live);
5946 }
5947
5948 #[test]
5949 fn an_old_lock_whose_pid_is_still_alive_is_never_swept_by_age_alone() {
5950 let dir = tempfile::tempdir().unwrap();
5960 let queue = Queue::at(dir.path().to_path_buf());
5961 let mut t = task();
5962 t.id = "20260101-000000-live".to_owned();
5963 queue.put(&mut t).unwrap();
5964
5965 let claim = queue.claim(&t.id).unwrap();
5966 std::thread::sleep(Duration::from_millis(60));
5967
5968 let swept = sweep_stale_claims(&queue, Duration::from_millis(50));
5969 assert!(
5970 swept.is_empty(),
5971 "a lock naming a live pid must never be swept by age, no matter how old: {swept:?}"
5972 );
5973 assert!(
5974 queue.claim(&t.id).is_err(),
5975 "the lock still protects its task"
5976 );
5977 drop(claim);
5978 }
5979
5980 fn injected_dead_pid() -> u32 {
5983 std::process::id().checked_add(1).unwrap_or(1)
5984 }
5985
5986 #[test]
5987 fn a_lock_naming_a_dead_pid_is_swept_at_once_regardless_of_age() {
5988 let dir = tempfile::tempdir().unwrap();
5989 let queue = Queue::at(dir.path().to_path_buf());
5990 let mut t = task();
5991 t.id = "20260101-000000-dead".to_owned();
5992 queue.put(&mut t).unwrap();
5993 let dead_pid = injected_dead_pid();
5994
5995 std::fs::write(
6000 dir.path().join(format!("{}.lock", t.id)),
6001 dead_pid.to_string(),
6002 )
6003 .unwrap();
6004
6005 let swept = sweep_stale_claims_with(&queue, Duration::from_secs(6 * 60 * 60), |pid| {
6006 pid != dead_pid
6007 });
6008 assert_eq!(
6009 swept,
6010 vec![t.id.clone()],
6011 "a dead owner is reclaimed immediately, not after STALE_CLAIM"
6012 );
6013 assert!(queue.claim(&t.id).is_ok(), "the task is claimable again");
6014 }
6015
6016 #[test]
6017 fn sweeping_on_every_poll_catches_a_lock_that_appears_after_the_first_sweep() {
6018 let dir = tempfile::tempdir().unwrap();
6019 let queue = Queue::at(dir.path().to_path_buf());
6020 let mut t = task();
6021 t.id = "20260101-000000-late".to_owned();
6022 queue.put(&mut t).unwrap();
6023 let dead_pid = injected_dead_pid();
6024
6025 assert!(
6028 sweep_stale_claims(&queue, Duration::from_secs(6 * 60 * 60)).is_empty(),
6029 "nothing has claimed the task yet"
6030 );
6031
6032 std::fs::write(
6035 dir.path().join(format!("{}.lock", t.id)),
6036 dead_pid.to_string(),
6037 )
6038 .unwrap();
6039
6040 let swept = sweep_stale_claims_with(&queue, Duration::from_secs(6 * 60 * 60), |pid| {
6044 pid != dead_pid
6045 });
6046 assert_eq!(swept, vec![t.id.clone()]);
6047 }
6048
6049 #[test]
6050 fn a_running_task_behind_a_dead_daemons_lock_recovers_once_swept_and_keeps_its_history() {
6051 crate::run::pin_test_home();
6056 let dir = tempfile::tempdir().unwrap();
6057 let queue = Queue::at(dir.path().to_path_buf());
6058 let mut t = task();
6059 t.id = "20260101-000000-crsh".to_owned();
6060 t.status = TaskStatus::Running;
6061 t.attempts = 1;
6062 t.runs.push("20260904-000000-4043".to_owned());
6066 queue.put(&mut t).unwrap();
6067 let dead_pid = injected_dead_pid();
6068
6069 std::fs::write(
6072 dir.path().join(format!("{}.lock", t.id)),
6073 dead_pid.to_string(),
6074 )
6075 .unwrap();
6076
6077 assert!(reclaim_orphaned_running(&queue, 2).is_empty());
6083 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Running);
6084
6085 let swept = sweep_stale_claims_with(&queue, Duration::from_secs(6 * 60 * 60), |pid| {
6086 pid != dead_pid
6087 });
6088 assert_eq!(swept, vec![t.id.clone()]);
6089
6090 let reclaimed = reclaim_orphaned_running(&queue, 2);
6091 assert_eq!(reclaimed, vec![t.id.clone()]);
6092 let after = queue.get(&t.id).unwrap();
6093 assert_eq!(
6094 after.status,
6095 TaskStatus::Held,
6096 "no run.json to recover from, so a human is asked"
6097 );
6098 assert_eq!(
6099 after.runs,
6100 vec!["20260904-000000-4043".to_owned()],
6101 "the crashed run's id is kept as evidence, not discarded"
6102 );
6103 }
6104
6105 #[test]
6106 fn a_lock_is_kept_when_the_process_query_is_unavailable() {
6107 let dir = tempfile::tempdir().unwrap();
6108 let queue = Queue::at(dir.path().to_path_buf());
6109 let mut t = task();
6110 t.id = "20260101-000000-unknown".to_owned();
6111 queue.put(&mut t).unwrap();
6112 let dead_pid = injected_dead_pid();
6113 std::fs::write(
6114 dir.path().join(format!("{}.lock", t.id)),
6115 dead_pid.to_string(),
6116 )
6117 .unwrap();
6118
6119 let swept = sweep_stale_claims_with(&queue, Duration::ZERO, |_| true);
6120 assert!(swept.is_empty(), "an unknown pid must keep its lock");
6121 assert!(queue.claim(&t.id).is_err(), "the lock remains protective");
6122 }
6123
6124 fn run_state_in(status: RunStatus, language: &str) -> RunState {
6125 let mut s = run_state(status);
6126 s.config.graph.language = language.to_owned();
6127 s
6128 }
6129
6130 fn unstarted_verdict(status: RunStatus) -> Verdict {
6131 Verdict {
6132 status,
6133 left_pr: false,
6134 quota_hit: false,
6135 parked: false,
6136 no_viable_candidates: false,
6137 }
6138 }
6139
6140 #[test]
6141 fn an_already_in_base_run_finishes_the_task_without_spending_an_attempt() {
6142 let mut t = task();
6143 t.attempts = 1;
6144 settle_in(
6145 &mut t,
6146 unstarted_verdict(RunStatus::AlreadyInBase),
6147 "already in main",
6148 1,
6149 phrases("en"),
6150 );
6151 assert_eq!(t.status, TaskStatus::Done);
6152 assert_eq!(t.attempts, 0);
6153 }
6154
6155 #[test]
6156 fn settle_renders_the_non_terminal_reason_in_the_configured_language() {
6157 let reason = |language: &str| {
6158 let mut t = task();
6159 settle_in(
6160 &mut t,
6161 unstarted_verdict(RunStatus::Judging),
6162 "boom",
6163 1,
6164 phrases(language),
6165 );
6166 t.last_error.or(t.hold_reason).unwrap_or_default()
6167 };
6168 assert!(
6169 reason("en").starts_with("the graph stopped at `"),
6170 "{}",
6171 reason("en")
6172 );
6173 assert!(reason("ja").starts_with("グラフが終端状態に達しないまま"));
6174 assert!(reason("日本語").contains("boom"));
6175 assert_eq!(reason("fr"), reason("en"));
6176 }
6177
6178 #[test]
6179 fn describe_follows_the_run_language_and_keeps_the_run_id() {
6180 let en = describe(&run_state_in(RunStatus::Stalled, "en"));
6181 assert!(en.starts_with("the judging panel lost its quorum"), "{en}");
6182 let ja = describe(&run_state_in(RunStatus::Stalled, "ja"));
6183 assert!(ja.starts_with("審査パネルが定足数を失いました"), "{ja}");
6184 assert!(ja.contains("[run "), "{ja}");
6185 let ended = describe(&run_state_in(RunStatus::Failed, "jp"));
6186 assert!(ended.starts_with("run 終了: "), "{ended}");
6187 let mut de = run_state_in(RunStatus::Failed, "de");
6188 let mut en = run_state_in(RunStatus::Failed, "en");
6189 de.id = "same".to_owned();
6190 en.id = "same".to_owned();
6191 assert_eq!(describe(&de), describe(&en));
6192 }
6193
6194 #[test]
6195 fn handover_refusals_follow_the_language_and_keep_the_detail_apart() {
6196 use crate::handover::Refused;
6197 let cases = [
6198 Refused::Foreign {
6199 branch: "b".into(),
6200 path: "/w/x".into(),
6201 why: "made by hand".into(),
6202 },
6203 Refused::Unsafe {
6204 branch: "b".into(),
6205 path: "/w/x".into(),
6206 why: "its worktree has uncommitted changes (a.rs)".into(),
6207 },
6208 Refused::ReleaseFailed {
6209 branch: "b".into(),
6210 path: "/w/x".into(),
6211 run: "ab12".into(),
6212 },
6213 ];
6214 for r in &cases {
6215 let en = (phrases("en").handover_refused)(r);
6216 assert_eq!(en, r.to_string());
6217 let ja = (phrases("ja").handover_refused)(r);
6218 assert!(
6219 !ja.contains("is checked out") && !ja.contains("try again"),
6220 "{ja}"
6221 );
6222 assert!(ja.contains("`b`") && ja.contains("/w/x"), "{ja}");
6223 if let Refused::Foreign { why, .. } | Refused::Unsafe { why, .. } = r {
6224 assert!(ja.contains(&format!("(詳細: {why})")), "{ja}");
6225 }
6226 }
6227 assert!(phrases("en").handover_hint.contains("release the task"));
6228 assert!(phrases("ja").handover_hint.contains("解放"));
6229 }
6230
6231 #[test]
6232 fn describe_translates_the_unanswered_reviewer_stop_only_in_ja() {
6233 let build = |lang: &str| {
6234 let mut s = run_state_in(RunStatus::Blocked, lang);
6235 s.id = "same".to_owned();
6236 s.config.graph.review_rounds = 3;
6237 let mut r = review_round(3);
6238 r.expected = 3;
6239 r.answered = 1;
6240 s.reviews.push(r);
6241 s.event(
6242 "review",
6243 "2 reviewer seat(s) never answered after 3 rounds; refusing to call it clean",
6244 );
6245 s
6246 };
6247 let en = describe(&build("en"));
6248 assert!(en.contains("(review: 2 reviewer seat(s) never answered after 3 rounds; refusing to call it clean)"), "{en}");
6249 let ja = describe(&build("ja"));
6250 assert!(ja.contains("2 席のレビュアーが 3 ラウンド"), "{ja}");
6251 assert!(!ja.contains("never answered"), "{ja}");
6252 assert!(ja.contains("[run "), "{ja}");
6253
6254 let mut failed = build("ja");
6257 failed.reviews[0].e2e.push(crate::run::CommandOutcome {
6258 command: "cargo test".to_owned(),
6259 code: Some(1),
6260 output_tail: String::new(),
6261 duration_ms: 0,
6262 resource_blocked: false,
6263 });
6264 failed.event("review", "stopped; e2e failed: cargo test");
6265 let ja = describe(&failed);
6266 assert!(ja.contains("stopped; e2e failed: cargo test"), "{ja}");
6267 assert!(!ja.contains("席のレビュアー"), "{ja}");
6268 }
6269
6270 #[test]
6271 fn refusals_and_recovery_prose_follow_the_language() {
6272 let t = held_task_with("r1");
6273 let q = action_question(
6274 "r1",
6275 ask::ChoiceAction::Resume {
6276 run: "r1".to_owned(),
6277 },
6278 );
6279 let refuse = |p: &Phrases| match decide_action(&t, &q, p, |_: &str| bail!("gone")) {
6280 ActionDecision::Refuse(s) => s,
6281 other => panic!("{other:?}"),
6282 };
6283 assert!(refuse(phrases("en")).contains("could not be read: gone"));
6284 assert!(refuse(phrases("ja")).contains("読み込めませんでした: gone"));
6285 assert_eq!(refuse(phrases("fr")), refuse(phrases("en")));
6286
6287 let mut held = task();
6288 reclaim(&mut held, None, 2, "ja");
6289 assert!(held.hold_reason.unwrap().contains("保留にしました"));
6290 let mut held = task();
6291 reclaim(&mut held, None, 2, "xx");
6292 assert!(held.hold_reason.unwrap().contains("held for a human"));
6293 }
6294
6295 fn run_state(status: RunStatus) -> RunState {
6296 let mut state = RunState::new(
6297 PathBuf::from("/repo"),
6298 "main".to_owned(),
6299 "abc1234def".to_owned(),
6300 "add retries".to_owned(),
6301 Config::default(),
6302 );
6303 state.status = status;
6304 state
6305 }
6306
6307 fn candidate(label: char, summary: &str, empty: bool, failed: Option<&str>) -> Candidate {
6308 Candidate {
6309 index: 0,
6310 label,
6311 agent: "claude".to_owned(),
6312 branch: format!("magi/x/{label}"),
6313 worktree: PathBuf::from("/repo"),
6314 summary: summary.to_owned(),
6315 stat: String::new(),
6316 files: 0,
6317 commits: usize::from(!empty),
6318 empty,
6319 failed: failed.map(str::to_owned),
6320 verified_noop: None,
6321 duration_ms: 0,
6322 folded: false,
6323 }
6324 }
6325
6326 #[test]
6327 fn diagnostic_names_the_failing_gate_checks_and_their_output() {
6328 let mut state = run_state(RunStatus::Blocked);
6329 state.gate = vec![
6330 CommandOutcome {
6331 command: "cargo make check".to_owned(),
6332 code: Some(0),
6333 output_tail: "ok".to_owned(),
6334 duration_ms: 0,
6335 resource_blocked: false,
6336 },
6337 CommandOutcome {
6338 command: "cargo test".to_owned(),
6339 code: Some(101),
6340 output_tail: "thread 'x' panicked: assertion failed".to_owned(),
6341 duration_ms: 0,
6342 resource_blocked: false,
6343 },
6344 ];
6345 let d = diagnostic(&state).expect("a failing gate must produce a diagnostic");
6346 assert!(d.contains("cargo test"), "{d}");
6347 assert!(
6348 !d.contains("cargo make check"),
6349 "a passing check is not a diagnostic: {d}"
6350 );
6351 assert!(d.contains("assertion failed"), "{d}");
6352 }
6353
6354 #[test]
6355 fn diagnostic_names_the_checks_the_fixer_gave_up_in_front_of() {
6356 let mut state = run_state(RunStatus::Blocked);
6357 state.event(
6358 "land",
6359 "stopped: the fixer produced no commit while 2 check(s) were failing \
6360 (build, lint); stopping instead of looping on an unchanged tree",
6361 );
6362 let d = diagnostic(&state).expect("a stalled land loop must produce a diagnostic");
6363 assert!(d.contains("build"), "{d}");
6364 assert!(d.contains("lint"), "{d}");
6365 assert!(d.contains("fixer produced no commit"), "{d}");
6366 }
6367
6368 #[test]
6369 fn describe_never_leaves_a_verified_noop_reading_as_a_bare_status_code() {
6370 let state = run_state(RunStatus::VerifiedNoop);
6375 let d = describe(&state);
6376 assert!(
6377 d.contains("agent-verified no-op"),
6378 "expected the display label, not the wire spelling: {d}"
6379 );
6380 assert!(!d.contains("verified_noop"), "{d}");
6381 }
6382
6383 #[test]
6384 fn diagnostic_carries_a_candidates_own_final_word_when_none_was_viable() {
6385 let mut state = run_state(RunStatus::Failed);
6391 state.candidates = vec![candidate(
6392 'A',
6393 "opened pull request #42, merged it, tagged v1.2.3 and published the release",
6394 true,
6395 None,
6396 )];
6397 let d = diagnostic(&state).expect("an empty candidate with a summary must be surfaced");
6398 assert!(d.contains("candidate A"), "{d}");
6399 assert!(d.contains("tagged v1.2.3"), "{d}");
6400 }
6401
6402 #[test]
6403 fn diagnostic_falls_back_to_a_candidates_failure_reason_when_it_has_no_summary() {
6404 let mut state = run_state(RunStatus::Failed);
6405 state.candidates = vec![candidate('A', "", true, Some("agent timed out"))];
6406 let d = diagnostic(&state).expect("a candidate's own failure reason must be surfaced");
6407 assert!(d.contains("candidate A"), "{d}");
6408 assert!(d.contains("agent timed out"), "{d}");
6409 }
6410
6411 #[test]
6412 fn diagnostic_is_none_when_nothing_recognisable_explains_the_hold() {
6413 let mut state = run_state(RunStatus::Failed);
6416 state.candidates = vec![candidate('A', "did the work", false, None)];
6417 assert!(diagnostic(&state).is_none());
6418 }
6419
6420 #[test]
6421 fn diagnostic_is_bounded_however_much_a_run_printed() {
6422 let mut state = run_state(RunStatus::Blocked);
6423 state.gate = vec![
6424 CommandOutcome {
6425 command: "cargo test".to_owned(),
6426 code: Some(101),
6427 output_tail: "x".repeat(50_000),
6428 duration_ms: 0,
6429 resource_blocked: false,
6430 },
6431 CommandOutcome {
6432 command: "cargo clippy".to_owned(),
6433 code: Some(1),
6434 output_tail: "y".repeat(50_000),
6435 duration_ms: 0,
6436 resource_blocked: false,
6437 },
6438 ];
6439 state.candidates = vec![
6440 candidate('A', &"z".repeat(50_000), true, None),
6441 candidate('B', &"w".repeat(50_000), true, None),
6442 ];
6443 let d = diagnostic(&state).expect("plenty here to diagnose");
6444 assert!(
6445 d.len() <= DIAGNOSTIC_MAX,
6446 "diagnostic grew to {} bytes, unbounded",
6447 d.len()
6448 );
6449 }
6450
6451 #[test]
6452 fn settle_and_diagnose_attaches_a_diagnostic_only_once_the_task_is_held() {
6453 let mut state = run_state(RunStatus::Blocked);
6454 state.gate = vec![CommandOutcome {
6455 command: "cargo test".to_owned(),
6456 code: Some(101),
6457 output_tail: "assertion failed".to_owned(),
6458 duration_ms: 0,
6459 resource_blocked: false,
6460 }];
6461 let verdict = Verdict {
6462 status: RunStatus::Blocked,
6463 left_pr: false,
6464 quota_hit: false,
6465 parked: false,
6466 no_viable_candidates: false,
6467 };
6468
6469 let mut t = task();
6472 t.start("run-1".to_owned());
6473 settle_and_diagnose(&mut t, verdict, "gate failed", 2, &state);
6474 assert_eq!(t.status, TaskStatus::Failed);
6475 assert!(t.diagnostic.is_none());
6476
6477 t.start("run-2".to_owned());
6480 settle_and_diagnose(&mut t, verdict, "gate failed", 2, &state);
6481 assert_eq!(t.status, TaskStatus::Held);
6482 let d = t.diagnostic.expect("a held task must carry its diagnostic");
6483 assert!(d.contains("cargo test"), "{d}");
6484 }
6485
6486 #[test]
6487 fn a_held_task_names_the_open_question_it_is_waiting_on() {
6488 crate::run::pin_test_home();
6493 let home = crate::run::home();
6494 let state = run_state(RunStatus::VerifiedNoop);
6495 let mut q = ask::Question::new(
6496 state.id.clone(),
6497 "implement".to_owned(),
6498 "impl-A".to_owned(),
6499 "is this really a no-op?".to_owned(),
6500 String::new(),
6501 Vec::new(),
6502 );
6503 Questions::at(home.join("questions")).put(&mut q).unwrap();
6504
6505 let verdict = Verdict {
6506 status: RunStatus::VerifiedNoop,
6507 left_pr: false,
6508 quota_hit: false,
6509 parked: false,
6510 no_viable_candidates: false,
6511 };
6512 let mut t = task();
6513 t.start(state.id.clone());
6514 settle_and_diagnose(&mut t, verdict, "run ended agent-verified no-op", 5, &state);
6515
6516 assert_eq!(t.status, TaskStatus::Held);
6517 let reason = t.hold_reason.expect("a held task must record why");
6518 assert!(
6519 reason.starts_with("run ended agent-verified no-op"),
6520 "the original settle reason must survive unchanged: {reason}"
6521 );
6522 assert!(
6523 reason.contains(q.short()),
6524 "the open question's id must be named so the notice is actionable: {reason}"
6525 );
6526 }
6527
6528 #[test]
6529 fn a_held_task_with_no_open_question_keeps_its_plain_reason() {
6530 crate::run::pin_test_home();
6531 let state = run_state(RunStatus::VerifiedNoop);
6532
6533 let verdict = Verdict {
6534 status: RunStatus::VerifiedNoop,
6535 left_pr: false,
6536 quota_hit: false,
6537 parked: false,
6538 no_viable_candidates: false,
6539 };
6540 let mut t = task();
6541 t.start(state.id.clone());
6542 settle_and_diagnose(&mut t, verdict, "run ended agent-verified no-op", 5, &state);
6543
6544 assert_eq!(t.status, TaskStatus::Held);
6545 assert_eq!(
6546 t.hold_reason.as_deref(),
6547 Some("run ended agent-verified no-op"),
6548 "nothing to append when the question was already answered or never asked"
6549 );
6550 }
6551
6552 #[test]
6553 fn supersede_prior_runs_rewrites_an_earlier_blocked_attempt_once_a_later_one_lands() {
6554 crate::run::pin_test_home();
6555 let mut first = run_state(RunStatus::Blocked);
6556 first.id = "20260101-000000-sup1".to_owned();
6557 first.save().unwrap();
6558 let mut second = run_state(RunStatus::Merged);
6559 second.id = "20260101-000000-sup2".to_owned();
6560 second.save().unwrap();
6561
6562 let mut t = task();
6563 t.runs = vec![first.id.clone(), second.id.clone()];
6564 t.status = TaskStatus::Done;
6565
6566 supersede_prior_runs(&t, &crate::run::home());
6567
6568 assert_eq!(
6569 RunState::load(&first.id).unwrap().status,
6570 RunStatus::Superseded,
6571 "the first attempt's Blocked no longer needs anyone's attention"
6572 );
6573 assert_eq!(
6574 RunState::load(&second.id).unwrap().status,
6575 RunStatus::Merged,
6576 "the run that actually succeeded is left exactly as it was"
6577 );
6578 }
6579
6580 #[test]
6581 fn supersede_prior_runs_leaves_a_manually_resumed_attempt_alone() {
6582 crate::run::pin_test_home();
6588 let mut first = run_state(RunStatus::Blocked);
6589 first.id = "20260101-000000-sup9".to_owned();
6590 first.driver_pid = Some(std::process::id());
6593 first.driver_started_at = Some(
6594 crate::proc::process_started_at(std::process::id())
6595 .expect("this test process's own start time must be queryable"),
6596 );
6597 first.save().unwrap();
6598 let mut second = run_state(RunStatus::Merged);
6599 second.id = "20260101-000000-supa".to_owned();
6600 second.save().unwrap();
6601
6602 let mut t = task();
6603 t.runs = vec![first.id.clone(), second.id.clone()];
6604 t.status = TaskStatus::Done;
6605
6606 supersede_prior_runs(&t, &crate::run::home());
6607
6608 assert_eq!(
6609 RunState::load(&first.id).unwrap().status,
6610 RunStatus::Blocked,
6611 "a live driver_pid means something is still actually working this run, \
6612 even though no daemon claims it - rewriting under it would just be \
6613 undone the next time that process saves"
6614 );
6615 }
6616
6617 #[test]
6618 fn resweep_catches_up_a_run_left_live_once_its_manual_process_is_no_longer_driving_it() {
6619 let dir = tempfile::tempdir().unwrap();
6625 let home = dir.path().to_path_buf();
6626 let queue = Queue::at(dir.path().join("queue"));
6627
6628 let mut first = run_state(RunStatus::Blocked);
6629 first.id = "20260101-000000-supd".to_owned();
6630 first.driver_pid = Some(std::process::id());
6631 first.driver_started_at = Some(
6632 crate::proc::process_started_at(std::process::id())
6633 .expect("this test process's own start time must be queryable"),
6634 );
6635 first.save_under(&home).unwrap();
6636 let mut second = run_state(RunStatus::Merged);
6637 second.id = "20260101-000000-supe".to_owned();
6638 second.save_under(&home).unwrap();
6639
6640 let mut t = task();
6641 t.runs = vec![first.id.clone(), second.id.clone()];
6642 t.status = TaskStatus::Done;
6643 queue.put(&mut t).unwrap();
6644
6645 resweep_superseded_attempts(&queue, &home);
6646 assert_eq!(
6647 RunState::load_under(&first.id, &home).unwrap().status,
6648 RunStatus::Blocked,
6649 "still live on the first pass, so still untouched"
6650 );
6651
6652 let mut stale = RunState::load_under(&first.id, &home).unwrap();
6658 stale.driver_started_at = Some("1".to_owned());
6659 stale.save_under(&home).unwrap();
6660
6661 resweep_superseded_attempts(&queue, &home);
6662 assert_eq!(
6663 RunState::load_under(&first.id, &home).unwrap().status,
6664 RunStatus::Superseded,
6665 "the second pass catches up what the first one correctly skipped"
6666 );
6667 }
6668
6669 #[test]
6670 fn supersede_prior_runs_leaves_concurrent_blocked_attempts_alone_while_the_task_is_not_done() {
6671 crate::run::pin_test_home();
6672 let mut first = run_state(RunStatus::Blocked);
6673 first.id = "20260101-000000-sup3".to_owned();
6674 first.save().unwrap();
6675 let mut second = run_state(RunStatus::Blocked);
6676 second.id = "20260101-000000-sup4".to_owned();
6677 second.save().unwrap();
6678
6679 let mut t = task();
6680 t.runs = vec![first.id.clone(), second.id.clone()];
6681 t.status = TaskStatus::Failed;
6685
6686 supersede_prior_runs(&t, &crate::run::home());
6687
6688 assert_eq!(
6689 RunState::load(&first.id).unwrap().status,
6690 RunStatus::Blocked
6691 );
6692 assert_eq!(
6693 RunState::load(&second.id).unwrap().status,
6694 RunStatus::Blocked
6695 );
6696 }
6697
6698 #[test]
6699 fn supersede_prior_runs_does_nothing_when_the_task_was_closed_by_hand() {
6700 crate::run::pin_test_home();
6704 let mut first = run_state(RunStatus::Blocked);
6705 first.id = "20260101-000000-sup5".to_owned();
6706 first.save().unwrap();
6707
6708 let mut t = task();
6709 t.runs = vec![first.id.clone()];
6710 t.status = TaskStatus::Done;
6711
6712 supersede_prior_runs(&t, &crate::run::home());
6713
6714 assert_eq!(
6715 RunState::load(&first.id).unwrap().status,
6716 RunStatus::Blocked,
6717 "a single-attempt task has no earlier run to supersede"
6718 );
6719 }
6720
6721 #[test]
6722 fn supersede_prior_runs_does_nothing_when_the_last_recorded_attempt_never_landed() {
6723 crate::run::pin_test_home();
6730 let mut first = run_state(RunStatus::Blocked);
6731 first.id = "20260101-000000-supb".to_owned();
6732 first.save().unwrap();
6733 let mut second = run_state(RunStatus::Failed);
6734 second.id = "20260101-000000-supc".to_owned();
6735 second.save().unwrap();
6736
6737 let mut t = task();
6738 t.runs = vec![first.id.clone(), second.id.clone()];
6739 t.status = TaskStatus::Done;
6740
6741 supersede_prior_runs(&t, &crate::run::home());
6742
6743 assert_eq!(
6744 RunState::load(&first.id).unwrap().status,
6745 RunStatus::Blocked,
6746 "the task's last attempt never landed, so there is nothing here \
6747 actually superseding it"
6748 );
6749 }
6750
6751 #[test]
6752 fn supersede_prior_runs_leaves_a_failed_or_verified_noop_attempt_as_is() {
6753 crate::run::pin_test_home();
6757 let mut failed = run_state(RunStatus::Failed);
6758 failed.id = "20260101-000000-sup6".to_owned();
6759 failed.save().unwrap();
6760 let mut noop = run_state(RunStatus::VerifiedNoop);
6761 noop.id = "20260101-000000-sup7".to_owned();
6762 noop.save().unwrap();
6763 let mut winner = run_state(RunStatus::Ready);
6764 winner.id = "20260101-000000-sup8".to_owned();
6765 winner.save().unwrap();
6766
6767 let mut t = task();
6768 t.runs = vec![failed.id.clone(), noop.id.clone(), winner.id.clone()];
6769 t.status = TaskStatus::Done;
6770
6771 supersede_prior_runs(&t, &crate::run::home());
6772
6773 assert_eq!(
6774 RunState::load(&failed.id).unwrap().status,
6775 RunStatus::Failed
6776 );
6777 assert_eq!(
6778 RunState::load(&noop.id).unwrap().status,
6779 RunStatus::VerifiedNoop
6780 );
6781 }
6782
6783 fn approval_question(run: &str) -> ask::Question {
6784 ask::Question::new(
6785 run.to_owned(),
6786 land::APPROVAL_NODE.to_owned(),
6787 "land".to_owned(),
6788 "merge?".to_owned(),
6789 String::new(),
6790 vec!["merge".to_owned(), "hold".to_owned()],
6791 )
6792 }
6793
6794 #[test]
6795 fn land_resume_state_leaves_a_fresh_open_question_waiting() {
6796 crate::run::pin_test_home();
6797 let mut state = run_state(RunStatus::Landing);
6798 state.id = "20260101-000000-fre1".to_owned();
6799 state.parked = true;
6800 state.save().unwrap();
6801 ask::Questions::open()
6802 .put(&mut approval_question(&state.id))
6803 .unwrap();
6804
6805 let mut t = task();
6806 t.runs.push(state.id.clone());
6807 assert_eq!(
6808 land_resume_state(&t),
6809 LandResume::StillWaiting,
6810 "nobody has answered and the timeout has not passed"
6811 );
6812 }
6813
6814 #[test]
6815 fn land_resume_state_waits_on_a_withheld_text_question_then_resumes() {
6816 crate::run::pin_test_home();
6817 let mut state = run_state(RunStatus::Gating);
6818 state.id = "20260101-000000-gt01".to_owned();
6819 state.parked = true;
6820 state.config.graph.answer_timeout = 60;
6821 state.github_text = Some(crate::run::GithubTextAsk {
6822 rewrite: None,
6823 fingerprint: "f".into(),
6824 title: true,
6825 body: false,
6826 categories: vec!["title-language".into()],
6827 question: None,
6828 asks: 1,
6829 resolved: false,
6830 chosen_title: None,
6831 choice: None,
6832 });
6833 state.save().unwrap();
6834 let store = ask::Questions::open();
6835 let mut q = ask::Question::new(
6836 state.id.clone(),
6837 crate::github_text::ASK_NODE.to_owned(),
6838 crate::github_text::ASK_SEAT.to_owned(),
6839 "s".into(),
6840 String::new(),
6841 vec!["use fallback".into()],
6842 );
6843 store.put(&mut q).unwrap();
6844 let mut t = task();
6845 t.runs.push(state.id.clone());
6846 assert_eq!(land_resume_state(&t), LandResume::StillWaiting);
6847 q.asked_at = Timestamp::now() - jiff::SignedDuration::from_secs(120);
6848 store.put(&mut q).unwrap();
6849 assert_eq!(land_resume_state(&t), LandResume::Ready);
6850 assert!(!store.get(&q.id).unwrap().status.open());
6851 state.github_text.as_mut().unwrap().resolved = true;
6853 state.save().unwrap();
6854 assert_eq!(land_resume_state(&t), LandResume::NotLanding);
6855 }
6856
6857 #[test]
6858 fn land_resume_state_abandons_a_question_that_outlived_answer_timeout() {
6859 crate::run::pin_test_home();
6864 let mut state = run_state(RunStatus::Landing);
6865 state.id = "20260101-000000-exp1".to_owned();
6866 state.parked = true;
6867 state.config.graph.answer_timeout = 60;
6868 state.save().unwrap();
6869
6870 let store = ask::Questions::open();
6871 let mut q = approval_question(&state.id);
6872 q.asked_at = Timestamp::now() - jiff::SignedDuration::from_secs(120);
6873 store.put(&mut q).unwrap();
6874
6875 let mut t = task();
6876 t.runs.push(state.id.clone());
6877 assert_eq!(
6878 land_resume_state(&t),
6879 LandResume::Ready,
6880 "an expired question must not be waited on forever"
6881 );
6882
6883 let after = store.get(&q.id).unwrap();
6884 assert!(
6885 !after.status.open(),
6886 "the question is abandoned, not silently ignored"
6887 );
6888 assert!(
6889 after.resolution().is_none(),
6890 "an abandoned question is not read as a decision"
6891 );
6892 }
6893
6894 #[test]
6895 fn reclaim_settles_a_running_task_against_its_last_run() {
6896 let mut t = task();
6897 t.start("20260904-000000-4043".to_owned());
6898 reclaim(&mut t, Some(run_state(RunStatus::Ready)), 2, "en");
6899 assert_eq!(
6900 t.status,
6901 TaskStatus::Done,
6902 "a run that actually finished must not stay `running` forever"
6903 );
6904 }
6905
6906 #[test]
6907 fn reclaim_maps_a_parked_run_to_parked_without_touching_last_error() {
6908 let mut t = task();
6909 t.start("20260904-000000-4043".to_owned());
6910 t.last_error = Some("earlier trouble".to_owned());
6911 let mut state = run_state(RunStatus::Judging);
6912 state.parked = true;
6913 reclaim(&mut t, Some(state), 1, "en");
6914 assert_eq!(t.status, TaskStatus::Parked);
6915 assert_eq!(t.attempts, 0, "the park is refunded, even at max_attempts");
6916 assert_eq!(t.last_error.as_deref(), Some("earlier trouble"));
6917 assert!(t.park_reason.is_some());
6918 assert_eq!(t.runs, ["20260904-000000-4043"], "the same run is kept");
6919 assert!(!t.fresh_start);
6920 assert!(t.status.runnable());
6921 }
6922
6923 #[test]
6924 fn settle_parks_a_task_without_a_failure_and_a_new_start_clears_the_reason() {
6925 let mut t = task();
6926 t.start("20260904-000000-4043".to_owned());
6927 t.last_error = Some("earlier trouble".to_owned());
6928 settle(
6929 &mut t,
6930 Verdict {
6931 status: RunStatus::Judging,
6932 left_pr: false,
6933 quota_hit: false,
6934 parked: true,
6935 no_viable_candidates: false,
6936 },
6937 "parked after `judging`",
6938 1,
6939 );
6940 assert_eq!(t.status, TaskStatus::Parked);
6941 assert_eq!(t.attempts, 0);
6942 assert_eq!(t.last_error.as_deref(), Some("earlier trouble"));
6943 assert_eq!(t.park_reason.as_deref(), Some("parked after `judging`"));
6944 assert_eq!(t.schema, crate::queue::SCHEMA);
6945 assert!(
6946 crate::notices::task_held(&t).is_none(),
6947 "a park is not news"
6948 );
6949 t.start("20260904-000000-4043".to_owned());
6950 assert_eq!(t.park_reason, None);
6951 }
6952
6953 #[test]
6954 fn reclaim_reuses_the_same_retry_policy_as_a_live_settle() {
6955 let mut t = task();
6959 t.start("20260904-000000-4043".to_owned());
6960 reclaim(&mut t, Some(run_state(RunStatus::Blocked)), 2, "en");
6961 assert_eq!(t.status, TaskStatus::Failed);
6962 assert!(t.status.runnable());
6963 }
6964
6965 #[test]
6966 fn a_settled_parked_pass_records_its_cause_and_the_refund() {
6967 let mut t = task();
6968 t.start("20260904-000000-4043".to_owned());
6969 assert_eq!(t.attempts, 1);
6970 let mut run = run_state(RunStatus::Implementing);
6971 run.id = "20260904-000000-4043".to_owned();
6972 run.begin_pass();
6973 let cause = crate::run::ParkCause::Upgrade {
6974 from: "0.121.0".to_owned(),
6975 to: Some("0.121.1".to_owned()),
6976 };
6977 run.end_pass_parked(cause.clone());
6978 run.parked = true;
6979 let verdict = Verdict {
6980 status: run.status,
6981 left_pr: false,
6982 quota_hit: false,
6983 parked: true,
6984 no_viable_candidates: false,
6985 };
6986 settle_and_diagnose(&mut t, verdict, "parked", 2, &run);
6987 assert_eq!(t.attempts, 0);
6988 assert_eq!(t.passes.len(), 1);
6989 assert_eq!(t.passes[0].cost, crate::queue::PassCost::Refunded);
6990 assert_eq!(t.passes[0].parked, Some(cause));
6991 assert_eq!(t.passes[0].pass_index, 0);
6992
6993 t.start("20260904-000000-4043".to_owned());
6996 run.begin_pass();
6997 run.status = RunStatus::Failed;
6998 let verdict = Verdict {
6999 status: RunStatus::Failed,
7000 left_pr: false,
7001 quota_hit: false,
7002 parked: false,
7003 no_viable_candidates: false,
7004 };
7005 settle_and_diagnose(&mut t, verdict, "failed", 2, &run);
7006 assert_eq!(t.passes.len(), 2);
7007 assert_eq!(t.passes[0].cost, crate::queue::PassCost::Refunded);
7008 assert_eq!(t.passes[1].cost, crate::queue::PassCost::Spent);
7009 assert_eq!(t.passes[1].pass_index, 1);
7010 assert_eq!(t.attempts, 1);
7011 }
7012
7013 #[test]
7014 fn reclaim_holds_a_running_task_whose_run_cannot_be_found() {
7015 let mut t = task();
7016 t.start("20260904-000000-4043".to_owned());
7017 reclaim(&mut t, None, 2, "en");
7018 assert_eq!(t.status, TaskStatus::Held);
7019 assert!(
7020 t.last_error
7021 .as_deref()
7022 .is_some_and(|e| e.contains("running")),
7023 "the operator needs to know why this task was held"
7024 );
7025 }
7026
7027 #[test]
7028 fn orphaned_running_tasks_are_reclaimed_but_live_ones_are_left_alone() {
7029 let dir = tempfile::tempdir().unwrap();
7030 let queue = Queue::at(dir.path().to_path_buf());
7031
7032 let mut orphaned = task();
7034 orphaned.id = "20260904-000000-orph".to_owned();
7035 orphaned.status = TaskStatus::Running;
7036 orphaned.attempts = 1;
7037 queue.put(&mut orphaned).unwrap();
7038
7039 let mut alive = task();
7040 alive.id = "20260904-000000-live".to_owned();
7041 alive.status = TaskStatus::Running;
7042 alive.attempts = 1;
7043 queue.put(&mut alive).unwrap();
7044 let _held_by_a_live_daemon = queue.claim(&alive.id).unwrap();
7045
7046 let mut queued = task();
7047 queued.id = "20260904-000000-wait".to_owned();
7048 queue.put(&mut queued).unwrap();
7049
7050 let reclaimed = reclaim_orphaned_running(&queue, 2);
7051 assert_eq!(reclaimed, vec![orphaned.id.clone()]);
7052
7053 assert_eq!(
7054 queue.get(&orphaned.id).unwrap().status,
7055 TaskStatus::Held,
7056 "nothing was driving it and there was no run to recover"
7057 );
7058 assert_eq!(
7059 queue.get(&alive.id).unwrap().status,
7060 TaskStatus::Running,
7061 "a live claim must protect the task it belongs to"
7062 );
7063 assert_eq!(queue.get(&queued.id).unwrap().status, TaskStatus::Queued);
7064 }
7065
7066 fn read_run_under(home: &Path, id: &str) -> RunState {
7072 let body = std::fs::read_to_string(home.join("runs").join(id).join("run.json")).unwrap();
7073 serde_json::from_str(&body).unwrap()
7074 }
7075
7076 #[test]
7077 fn reclaim_abandoned_runs_fails_a_run_whose_active_seats_are_all_provably_dead() {
7078 let dir = tempfile::tempdir().unwrap();
7079 let home = dir.path().to_path_buf();
7080 let now = Timestamp::now();
7081 let overrun_seat = || crate::run::ActiveSeat {
7082 node: "implement".to_owned(),
7083 started_at: now - jiff::SignedDuration::new(21_000, 0),
7084 timeout_secs: 3_600,
7085 attempt: 0,
7086 task: None,
7087 command: None,
7088 index: None,
7089 total: None,
7090 };
7091
7092 let mut dead = run_state(RunStatus::Implementing);
7093 dead.id = "20260101-000000-dead".to_owned();
7094 dead.active.insert("impl-A".to_owned(), overrun_seat());
7095 dead.driver_pid = Some(4242);
7098 dead.save_under(&home).unwrap();
7099
7100 let mut alive = run_state(RunStatus::Implementing);
7103 alive.id = "20260101-000000-aliv".to_owned();
7104 alive.active.insert("impl-A".to_owned(), overrun_seat());
7105 alive.save_under(&home).unwrap();
7106 let mut status = Status::new();
7107 status.current = vec![Current {
7108 task: "20260101-000000-task".to_owned(),
7109 run: alive.id.clone(),
7110 }];
7111 write_status_to(&home.join("daemon.json"), &status).unwrap();
7112
7113 let questions = Questions::at(home.join("questions"));
7117 let mut q = ask::Question::new(
7118 dead.id.clone(),
7119 "implement".to_owned(),
7120 "impl-A".to_owned(),
7121 "Which storage backend?".to_owned(),
7122 String::new(),
7123 vec!["SQLite".to_owned(), "Redis".to_owned()],
7124 );
7125 questions.put(&mut q).unwrap();
7126
7127 let abandoned = reclaim_abandoned_runs_with(
7128 &home,
7129 now,
7130 |pid| if pid == 4242 { Some(false) } else { None },
7131 |_| panic!("a query answering Dead outright needs no identity corroboration"),
7132 );
7133 assert_eq!(abandoned, vec![dead.id.clone()]);
7134
7135 let reloaded = read_run_under(&home, &dead.id);
7136 assert_eq!(reloaded.status, RunStatus::Failed);
7137 assert!(reloaded.active.is_empty());
7138 assert!(
7139 !questions.get(&q.id).unwrap().status.open(),
7140 "the failed run's own open question must be settled in the same pass"
7141 );
7142
7143 let still_alive = read_run_under(&home, &alive.id);
7144 assert_eq!(
7145 still_alive.status,
7146 RunStatus::Implementing,
7147 "a live daemon's claim protects it"
7148 );
7149 assert!(!still_alive.active.is_empty());
7150 }
7151
7152 #[test]
7162 fn reclaim_abandoned_runs_leaves_a_live_manual_run_alone_even_though_no_daemon_claims_it() {
7163 let dir = tempfile::tempdir().unwrap();
7164 let home = dir.path().to_path_buf();
7165 let now = Timestamp::now();
7166
7167 let mut manual = run_state(RunStatus::Reviewing);
7168 manual.id = "20260101-000000-manl".to_owned();
7169 manual.active.insert(
7170 "review-1".to_owned(),
7171 crate::run::ActiveSeat {
7172 node: "review".to_owned(),
7173 started_at: now - jiff::SignedDuration::new(21_000, 0),
7174 timeout_secs: 3_600,
7175 attempt: 0,
7176 task: None,
7177 command: None,
7178 index: None,
7179 total: None,
7180 },
7181 );
7182 manual.driver_pid = Some(4242);
7186 manual.driver_started_at = Some("1790000000".to_owned());
7187 manual.save_under(&home).unwrap();
7188
7189 let abandoned = reclaim_abandoned_runs_with(
7190 &home,
7191 now,
7192 |pid| if pid == 4242 { Some(true) } else { None },
7193 |pid| {
7194 if pid == 4242 {
7195 Some("1790000000".to_owned())
7196 } else {
7197 None
7198 }
7199 },
7200 );
7201 assert!(
7202 abandoned.is_empty(),
7203 "a manual run a real process is still driving must never be reclaimed: {abandoned:?}"
7204 );
7205
7206 let reloaded = read_run_under(&home, &manual.id);
7207 assert_eq!(reloaded.status, RunStatus::Reviewing);
7208 assert!(!reloaded.active.is_empty());
7209 }
7210
7211 #[test]
7212 fn an_already_claimed_task_is_skipped_rather_than_failed() {
7213 let dir = tempfile::tempdir().unwrap();
7214 let queue = Queue::at(dir.path().to_path_buf());
7215 let mut only = task();
7216 queue.put(&mut only).unwrap();
7217
7218 let _elsewhere = queue.claim(&only.id).unwrap();
7219 let candidates = runnable(&queue);
7220 assert_eq!(candidates.len(), 1, "the task is still runnable");
7221 assert!(
7222 queue.claim(&candidates[0].id).is_err(),
7223 "the loop cannot take a claim somebody else holds"
7224 );
7225
7226 let after = queue.get(&only.id).unwrap();
7227 assert_eq!(after.status, TaskStatus::Queued);
7228 assert_eq!(
7229 after.attempts, 0,
7230 "losing the race is not an attempt at the task"
7231 );
7232 assert_eq!(after.last_error, None);
7233 }
7234
7235 #[test]
7236 fn the_status_file_round_trips_and_its_heartbeat_advances() {
7237 let dir = tempfile::tempdir().unwrap();
7238 let path = dir.path().join("daemon.json");
7239
7240 let mut status = Status::new();
7241 status.idle = false;
7242 status.completed = 7;
7243 status.current = vec![Current {
7244 task: "20260902-000000-t111".to_owned(),
7245 run: "20260902-000001-r111".to_owned(),
7246 }];
7247 write_status_to(&path, &status).unwrap();
7248 let first: Status = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
7249 assert_eq!(first.schema, SCHEMA);
7250 assert_eq!(first.pid, std::process::id());
7251 assert!(!first.idle);
7252 assert_eq!(first.completed, 7);
7253 assert_eq!(first.current, status.current);
7254 assert!(
7255 !path.with_extension("json.tmp").exists(),
7256 "the temp file is renamed, not left behind"
7257 );
7258
7259 std::thread::sleep(Duration::from_millis(5));
7260 status.updated_at = Timestamp::now();
7261 status.polls = 3;
7262 write_status_to(&path, &status).unwrap();
7263 let second: Status =
7264 serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
7265 assert!(
7266 second.updated_at > first.updated_at,
7267 "a reader can only detect staleness if the heartbeat moves"
7268 );
7269 assert_eq!(
7270 second.started_at, first.started_at,
7271 "the start time is not a heartbeat"
7272 );
7273 assert_eq!(second.polls, 3);
7274 }
7275
7276 #[test]
7277 fn reading_counts_as_running_only_while_its_heartbeat_is_fresh() {
7278 let dir = tempfile::tempdir().unwrap();
7279
7280 assert!(read_status(dir.path()).is_none(), "no file, no daemon");
7281
7282 let mut status = Status::new();
7283 status.updated_at = Timestamp::now() - jiff::SignedDuration::from_secs(60);
7284 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7285 let stale = read_status(dir.path()).unwrap();
7286 assert!(
7287 !stale.running(Timestamp::now()),
7288 "a minute without a heartbeat is a dead daemon, not a busy one"
7289 );
7290 assert!(stale.age_secs(Timestamp::now()).is_some_and(|s| s >= 55));
7291
7292 status.updated_at = Timestamp::now();
7293 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7294 let fresh = read_status(dir.path()).unwrap();
7295 assert!(fresh.running(Timestamp::now()));
7296 }
7297
7298 #[test]
7299 fn only_a_live_daemon_on_this_very_run_counts_as_working_on_it() {
7300 let dir = tempfile::tempdir().unwrap();
7301 let now = Timestamp::now();
7302 let mine = "20260903-080619-01c2";
7303
7304 assert!(
7305 !is_working_on(dir.path(), mine, now),
7306 "no status file means nobody is working on anything"
7307 );
7308
7309 let mut status = Status::new();
7310 status.current = vec![Current {
7311 task: "20260903-080340-0167".to_owned(),
7312 run: mine.to_owned(),
7313 }];
7314 status.updated_at = now;
7315 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7316 assert!(is_working_on(dir.path(), mine, now));
7317 assert!(
7318 !is_working_on(dir.path(), "20260903-105039-3cbf", now),
7319 "a daemon busy with one run is not working on another"
7320 );
7321
7322 status.updated_at = now - jiff::SignedDuration::from_secs(600);
7325 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7326 assert!(
7327 !is_working_on(dir.path(), mine, now),
7328 "a stale heartbeat is a dead daemon, so its run is a leftover"
7329 );
7330 }
7331
7332 #[test]
7333 fn is_working_on_short_matches_by_the_worktree_bays_own_name() {
7334 let dir = tempfile::tempdir().unwrap();
7335 let now = Timestamp::now();
7336
7337 assert!(
7338 !is_working_on_short(dir.path(), "01c2", now),
7339 "no status file means nobody is working on anything"
7340 );
7341
7342 let mut status = Status::new();
7343 status.current = vec![Current {
7344 task: "20260903-080340-0167".to_owned(),
7345 run: "20260903-080619-01c2".to_owned(),
7346 }];
7347 status.updated_at = now;
7348 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
7349 assert!(
7350 is_working_on_short(dir.path(), "01c2", now),
7351 "the run's short id is the last block of its full id"
7352 );
7353 assert!(
7354 !is_working_on_short(dir.path(), "3cbf", now),
7355 "a daemon busy with one worktree bay is not working on another"
7356 );
7357 }
7358
7359 #[test]
7360 fn a_newer_status_file_still_yields_a_reading() {
7361 let dir = tempfile::tempdir().unwrap();
7362 std::fs::write(
7365 dir.path().join("daemon.json"),
7366 serde_json::json!({
7367 "schema": 2,
7368 "updated_at": Timestamp::now().to_string(),
7369 "idle": true,
7370 "surprise": { "nested": [1, 2, 3] },
7371 })
7372 .to_string(),
7373 )
7374 .unwrap();
7375
7376 let reading = read_status(dir.path()).expect("a forward-compatible read");
7377 assert!(reading.running(Timestamp::now()));
7378 assert!(reading.idle);
7379 assert!(reading.current.is_empty());
7380 }
7381
7382 #[test]
7383 fn an_older_daemons_single_object_current_still_reads_as_a_one_item_list() {
7384 let dir = tempfile::tempdir().unwrap();
7390 std::fs::write(
7391 dir.path().join("daemon.json"),
7392 serde_json::json!({
7393 "schema": 1,
7394 "pid": 4242,
7395 "updated_at": Timestamp::now().to_string(),
7396 "idle": false,
7397 "current": {"task": "20260902-140501-aaaa", "run": "20260902-140502-bbbb"},
7398 "completed": 3,
7399 "polls": 9,
7400 })
7401 .to_string(),
7402 )
7403 .unwrap();
7404
7405 let reading = read_status(dir.path()).expect("an older shape must still parse");
7406 assert!(reading.running(Timestamp::now()));
7407 assert_eq!(
7408 reading.current,
7409 vec![Current {
7410 task: "20260902-140501-aaaa".to_owned(),
7411 run: "20260902-140502-bbbb".to_owned(),
7412 }]
7413 );
7414 }
7415
7416 #[test]
7417 fn an_absent_or_null_current_reads_as_idle_not_a_parse_failure() {
7418 let dir = tempfile::tempdir().unwrap();
7419 std::fs::write(
7420 dir.path().join("daemon.json"),
7421 serde_json::json!({
7422 "schema": 1,
7423 "updated_at": Timestamp::now().to_string(),
7424 "idle": true,
7425 "current": null,
7426 })
7427 .to_string(),
7428 )
7429 .unwrap();
7430 let with_null = read_status(dir.path()).expect("null must still parse");
7431 assert!(with_null.current.is_empty());
7432
7433 std::fs::write(
7434 dir.path().join("daemon.json"),
7435 serde_json::json!({
7436 "schema": 1,
7437 "updated_at": Timestamp::now().to_string(),
7438 "idle": true,
7439 })
7440 .to_string(),
7441 )
7442 .unwrap();
7443 let absent = read_status(dir.path()).expect("a missing field must still parse");
7444 assert!(absent.current.is_empty());
7445 }
7446
7447 #[test]
7448 fn a_task_without_a_repository_runs_in_the_daemons_default() {
7449 let fallback = Path::new("/default");
7450 let mut blank = task();
7451 blank.repo = PathBuf::new();
7452 assert_eq!(repo_for(&blank, fallback), PathBuf::from("/default"));
7453 let mut dot = task();
7454 dot.repo = PathBuf::from(".");
7455 assert_eq!(repo_for(&dot, fallback), PathBuf::from("/default"));
7456 assert_eq!(
7457 repo_for(&task(), fallback),
7458 PathBuf::from("/repo"),
7459 "a task that names a repository keeps it"
7460 );
7461 }
7462
7463 #[test]
7464 fn task_implementers_edit_inherits_daemon_config_and_merge() {
7465 let dir = tempfile::tempdir().unwrap();
7466 let path = dir.path().join("explicit.toml");
7467 std::fs::write(&path, "[graph]\nimplementers = 3\njudges = 1\n").unwrap();
7468 let opts = Opts {
7469 config: Some(path),
7470 merge: Some("none".into()),
7471 ..Default::default()
7472 };
7473 let mut t = task();
7474 t.set_implementers(
7475 2,
7476 &toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7477 .unwrap(),
7478 )
7479 .unwrap();
7480 let cfg = prepare_for(dir.path(), &opts, &t).unwrap();
7481 assert_eq!(cfg.graph.implementers, 2);
7482 assert_eq!(cfg.graph.judges, 1);
7483 assert_eq!(cfg.merge.mode, MergeMode::None);
7484 t.set_implementers(1, &cfg).unwrap();
7485 let mut cfg = prepare_for(dir.path(), &opts, &t).unwrap();
7486 apply_task_implementers(&mut cfg, &t);
7487 assert_eq!(cfg.graph.implementers, 1);
7488 assert_eq!(cfg.graph.judges, 1);
7489 assert_eq!(cfg.merge.mode, MergeMode::None);
7490 }
7491
7492 #[test]
7493 fn task_implementers_direct_overrides_stay_isolated_after_edits() {
7494 let dir = tempfile::tempdir().unwrap();
7495 let repo = dir.path().join("repo");
7496 std::fs::create_dir(&repo).unwrap();
7497 std::fs::write(
7498 repo.join("magi.toml"),
7499 "[graph]\njudges = 2\n[merge]\nmode = 'none'\n",
7500 )
7501 .unwrap();
7502 let daemon_config = dir.path().join("daemon.toml");
7503 std::fs::write(
7504 &daemon_config,
7505 "[graph]\njudges = 1\n[merge]\nmode = 'pr'\n",
7506 )
7507 .unwrap();
7508 let opts = Opts {
7509 config: Some(daemon_config),
7510 merge: Some("pr".into()),
7511 ..Default::default()
7512 };
7513 let q = Queue::at(dir.path().join("queue"));
7514 let validation: Config =
7515 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7516 .unwrap();
7517 for record in ["{}", r#"{"candidates":2}"#] {
7519 let overrides = serde_json::from_str(record).unwrap();
7520 let filing = crate::direct::Filing {
7521 instruction: "work".into(),
7522 title: "direct".into(),
7523 repo: repo.clone(),
7524 source: crate::queue::Source::Human,
7525 solo: false,
7526 overrides,
7527 review_of: None,
7528 };
7529 let filed =
7530 crate::direct::file(&q, dir.path(), Timestamp::now(), 1, filing, true).unwrap();
7531 let crate::direct::Filed::Standalone {
7532 mut task,
7533 claim: _claim,
7534 } = filed
7535 else {
7536 panic!("local filing")
7537 };
7538 let before = prepare_for(&repo, &opts, &task).unwrap();
7539 assert_eq!(before.graph.judges, 2, "daemon --config must not leak");
7540 assert_eq!(
7541 before.merge.mode,
7542 MergeMode::None,
7543 "daemon --merge must not leak"
7544 );
7545 for n in [2, 1, 2] {
7546 task.set_implementers(n, &validation).unwrap();
7547 q.put(&mut task).unwrap();
7548 task = q.get(&task.id).unwrap();
7549 assert!(!task.overrides.as_ref().unwrap().inherit_daemon_options);
7550 let mut cfg = prepare_for(&repo, &opts, &task).unwrap();
7551 apply_task_implementers(&mut cfg, &task);
7552 assert_eq!(cfg.graph.implementers, usize::from(n));
7553 assert_eq!(cfg.graph.judges, 2);
7554 assert_eq!(cfg.merge.mode, MergeMode::None);
7555 }
7556 }
7557 }
7558
7559 #[test]
7560 fn task_implementers_mismatch_keeps_explicit_reviews_review_only() {
7561 let cfg: Config =
7562 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7563 .unwrap();
7564 let mut t = task();
7565 t.review_of = Some("review-target".into());
7566 t.hold_manual(Some("retry".into()));
7567 let mut previous = RunState::new(
7568 PathBuf::from("/repo"),
7569 "main".into(),
7570 "abc1234".into(),
7571 "review".into(),
7572 cfg.clone(),
7573 );
7574 previous.config.graph.implementers = 2;
7575 for count in [1, 3] {
7576 t.set_implementers(count, &cfg).unwrap();
7577 t.release();
7578 assert_eq!(t.review_of.as_deref(), Some("review-target"));
7579 for proposed in [
7580 Starter::Resume(previous.id.clone()),
7581 Starter::Review("recovery-target".into()),
7582 ] {
7583 assert_eq!(
7584 starter_with_implementers(&t, Some(&previous), proposed),
7585 Starter::Review("review-target".into())
7586 );
7587 }
7588 let mut configured = cfg.clone();
7589 apply_task_implementers(&mut configured, &t);
7590 assert_eq!(configured.graph.implementers, usize::from(count));
7591 assert_eq!(previous.config.graph.implementers, 2);
7592 }
7593 }
7594
7595 #[test]
7596 fn task_implementers_dispatch_and_resume_agree() {
7597 let mut t = task();
7598 let cfg: crate::config::Config =
7599 toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
7600 .unwrap();
7601 let mut run = RunState::new(
7602 PathBuf::from("/repo"),
7603 "main".into(),
7604 "abc1234".into(),
7605 "work".into(),
7606 cfg.clone(),
7607 );
7608 run.config.graph.implementers = 2;
7609 for n in [2, 1, 3, 2] {
7610 t.set_implementers(n, &cfg).unwrap();
7611 let mut dispatched = cfg.clone();
7612 apply_task_implementers(&mut dispatched, &t);
7613 assert_eq!(dispatched.graph.implementers, usize::from(n));
7614 assert_eq!(implementers_match(&t, &run), n == 2);
7615 for proposed in [
7616 Starter::Resume(run.id.clone()),
7617 Starter::Review("old-branch".into()),
7618 ] {
7619 let expected = if n == 2 {
7620 proposed.clone()
7621 } else {
7622 Starter::Start
7623 };
7624 assert_eq!(
7625 starter_with_implementers(&t, Some(&run), proposed),
7626 expected
7627 );
7628 }
7629 assert_eq!(
7630 run.config.graph.implementers, 2,
7631 "history must not be rewritten"
7632 );
7633 }
7634 t.solo = false;
7635 t.overrides = None;
7636 assert!(implementers_match(&t, &run));
7637 }
7638
7639 #[test]
7640 fn a_solo_task_runs_with_one_candidate_and_a_plain_task_keeps_the_configs() {
7641 let mut solo_cfg = Config::default();
7647 solo_cfg.graph.implementers = 3;
7648 let mut solo_task = task();
7649 solo_task.solo = true;
7650 apply_task_implementers(&mut solo_cfg, &solo_task);
7651 assert_eq!(solo_cfg.graph.implementers, 1);
7652
7653 let mut plain_cfg = Config::default();
7654 plain_cfg.graph.implementers = 3;
7655 let plain_task = task();
7656 assert!(!plain_task.solo);
7657 apply_task_implementers(&mut plain_cfg, &plain_task);
7658 assert_eq!(
7659 plain_cfg.graph.implementers, 3,
7660 "a task that did not ask to run alone keeps the config's candidates"
7661 );
7662 }
7663
7664 fn loss(seat: &str, at: &str, reset: Option<&str>) -> QuotaLoss {
7665 QuotaLoss {
7666 seat: seat.into(),
7667 node: "judge".into(),
7668 at: at.parse().unwrap(),
7669 reset: reset.map(str::to_string),
7670 }
7671 }
7672
7673 #[test]
7674 fn a_resumed_run_with_only_old_quota_losses_arms_no_cooldown() {
7675 let old: Vec<QuotaLoss> = (1..=4)
7676 .map(|i| {
7677 loss(
7678 &format!("judge-{i}"),
7679 "2026-09-23T05:23:00Z",
7680 Some("2:40pm (Asia/Tokyo)"),
7681 )
7682 })
7683 .collect();
7684 let fresh = losses_this_attempt(&old, &old);
7685 assert!(fresh.is_empty());
7686 assert_eq!(cooldown_until(&fresh, Timestamp::now()), None);
7687 }
7689
7690 #[test]
7691 fn a_new_quota_loss_during_the_attempt_still_arms_the_cooldown() {
7692 let old = vec![loss("judge-1", "2026-09-23T05:23:00Z", None)];
7693 let now = Timestamp::now();
7694 let mut after = old.clone();
7695 after.push(loss("judge-2", &now.to_string(), None));
7696 let fresh = losses_this_attempt(&old, &after);
7697 assert_eq!(fresh, vec![after[1].clone()]);
7698 let until = cooldown_until(&fresh, now).expect("a fresh loss arms the cooldown");
7699 assert_eq!(
7700 until,
7701 now + jiff::SignedDuration::from_secs(QUOTA_WAIT_FALLBACK.as_secs() as i64)
7702 );
7703 }
7704
7705 #[test]
7706 fn a_recovered_seat_dropping_out_of_the_history_does_not_hide_a_new_loss() {
7707 let before = vec![
7710 loss("judge-1", "2026-09-23T05:23:00Z", None),
7711 loss("judge-2", "2026-09-23T05:24:00Z", None),
7712 ];
7713 let after = vec![
7714 loss("judge-2", "2026-09-23T05:24:00Z", None),
7715 loss("judge-1", "2026-09-24T01:00:00Z", None),
7716 ];
7717 assert_eq!(losses_this_attempt(&before, &after), vec![after[1].clone()]);
7718 }
7719
7720 #[test]
7721 fn merge_overrides_are_parsed_or_refused() {
7722 assert_eq!(merge_mode("none").unwrap(), MergeMode::None);
7723 assert_eq!(merge_mode("local").unwrap(), MergeMode::Local);
7724 assert_eq!(merge_mode("pr").unwrap(), MergeMode::Pr);
7725 assert!(merge_mode("squash").is_err());
7726 }
7727
7728 #[test]
7729 fn quota_wait_uses_a_future_reset_time_capped_and_falls_back_otherwise() {
7730 let now = Timestamp::now();
7731 let fallback = Duration::from_secs(300);
7732 let cap = Duration::from_secs(1800);
7733
7734 assert_eq!(quota_wait(None, now, fallback, cap), fallback);
7736
7737 let soon = now + jiff::SignedDuration::from_secs(600);
7739 assert_eq!(
7740 quota_wait(Some(soon), now, fallback, cap),
7741 Duration::from_secs(600)
7742 );
7743
7744 let past = now - jiff::SignedDuration::from_secs(60);
7747 assert_eq!(quota_wait(Some(past), now, fallback, cap), fallback);
7748
7749 let far = now + jiff::SignedDuration::from_secs(3 * 3600);
7752 assert_eq!(quota_wait(Some(far), now, fallback, cap), cap);
7753 }
7754
7755 #[test]
7756 fn parse_reset_hint_reads_the_claude_cli_shape_and_rolls_a_past_clock_to_tomorrow() {
7757 let now = "2026-09-07T02:50:00Z".parse::<Timestamp>().unwrap();
7758
7759 let at = parse_reset_hint("4:50am (UTC)", now, now).expect("a recognised shape parses");
7760 assert_eq!(at.to_string(), "2026-09-07T04:50:00Z");
7761
7762 let already_past =
7766 parse_reset_hint("1:00am (UTC)", now, now).expect("a recognised shape parses");
7767 assert_eq!(already_past.to_string(), "2026-09-08T01:00:00Z");
7768
7769 assert!(
7770 parse_reset_hint("session limit reached", now, now).is_none(),
7771 "free text with no recognised shape is not guessed at"
7772 );
7773 assert!(
7774 parse_reset_hint("4:50am (Nowhere/Fake)", now, now).is_none(),
7775 "an unresolvable zone name is not guessed at either"
7776 );
7777 }
7778
7779 #[test]
7780 fn parse_reset_hint_reads_the_codex_cli_shape_with_no_year_rollover_needed() {
7781 let now = "2026-09-07T02:50:00Z".parse::<Timestamp>().unwrap();
7782
7783 let at = parse_reset_hint(
7784 "You've hit your usage limit. Visit \
7785 https://chatgpt.com/codex/settings/usage to purchase more \
7786 credits or try again at Sep 19th, 2026 5:10 PM.",
7787 now,
7788 now,
7789 )
7790 .expect("the codex reset wording is a recognised shape");
7791 assert_eq!(at.to_string(), "2026-09-19T17:10:00Z");
7792
7793 let earlier = parse_reset_hint("try again at Jan 2nd, 2026 1:00 AM.", now, now)
7798 .expect("an explicit year needs no rollover");
7799 assert_eq!(earlier.to_string(), "2026-01-02T01:00:00Z");
7800
7801 assert!(
7802 parse_reset_hint("try again at Sep 19th, 26 5:10 PM.", now, now).is_none(),
7803 "a two-digit year is not the documented shape and is not guessed at"
7804 );
7805 assert!(
7806 parse_reset_hint("try again at Sept 19th, 2026 5:10 PM.", now, now).is_none(),
7807 "a four-letter month name is not the documented three-letter abbreviation"
7808 );
7809 assert!(
7810 parse_reset_hint("try again at Sep 19th, 2026 5:10 PM (UTC).", now, now).is_none(),
7811 "an explicit zone on the dated shape is a format nobody has \
7812 documented, and is refused rather than guessed at as UTC"
7813 );
7814 }
7815
7816 #[test]
7817 fn parse_reset_hint_reads_agys_relative_shape_from_when_the_loss_was_recorded() {
7818 let now = "2026-09-24T12:00:00Z".parse::<Timestamp>().unwrap();
7819 let recorded = "2026-09-24T08:00:00Z".parse::<Timestamp>().unwrap();
7820
7821 let at = parse_reset_hint("in 1h2m49s", now, recorded).expect("agy's shape parses");
7822 assert_eq!(at.as_second() - recorded.as_second(), 3769);
7823
7824 let partial = parse_reset_hint("in 45m", now, recorded).expect("units are optional");
7825 assert_eq!(partial.as_second() - recorded.as_second(), 45 * 60);
7826
7827 for bad in ["in ", "in 45", "in 3x", "in m", "in 1h junk", "1h2m"] {
7828 assert!(
7829 parse_reset_hint(bad, now, recorded).is_none(),
7830 "{bad:?} must not be guessed at"
7831 );
7832 }
7833 }
7834
7835 fn idle_loop(dir: &Path) -> (Opts, Queue, PathBuf, PathBuf, PathBuf) {
7839 let config = dir.join("magi.toml");
7840 std::fs::write(
7841 &config,
7842 "[disk]\nmin_free_bytes = 0\nauto_fold = false\ncache_limit_bytes = 0\n",
7843 )
7844 .unwrap();
7845 let opts = Opts {
7846 poll: Duration::from_secs(30),
7847 config: Some(config),
7848 repo: dir.join("repo"),
7852 ..Opts::default()
7853 };
7854 let home = dir.join("home");
7863 let worktrees = dir.join("wt");
7864 (
7865 opts,
7866 Queue::at(dir.join("queue")),
7867 home.join("daemon.json"),
7868 home,
7869 worktrees,
7870 )
7871 }
7872
7873 #[test]
7874 fn a_stop_is_idempotent_and_once_set_stays_set() {
7875 let stop = Stop::new();
7876 assert!(!stop.stopped());
7877
7878 stop.stop();
7879 assert!(stop.stopped());
7880 stop.stop();
7881 assert!(stop.stopped(), "a second stop is not a toggle");
7882
7883 let shared = stop.clone();
7884 assert!(
7885 shared.stopped(),
7886 "a clone is the same stop; that is how the loop and its caller share one"
7887 );
7888 }
7889
7890 #[test]
7891 fn only_a_stop_with_a_run_in_flight_reads_as_finishing() {
7892 let stop = Stop::new();
7893 stop.enter();
7894 assert!(
7895 !stop.finishing(),
7896 "a busy loop nobody has asked to stop is just running"
7897 );
7898
7899 stop.stop();
7900 assert!(
7901 stop.finishing(),
7902 "a stop asked for mid-run has not landed until the run is settled"
7903 );
7904
7905 stop.exit();
7906 assert!(
7907 !stop.finishing(),
7908 "once the run is settled the stop has landed and there is nothing to finish"
7909 );
7910 }
7911
7912 #[test]
7913 fn finishing_stays_true_until_the_last_of_several_runs_exits() {
7914 let stop = Stop::new();
7915 stop.enter();
7916 stop.enter();
7917 stop.stop();
7918 assert!(stop.finishing(), "two runs still in flight");
7919
7920 stop.exit();
7921 assert!(
7922 stop.finishing(),
7923 "one run finished, but a sibling is still working"
7924 );
7925
7926 stop.exit();
7927 assert!(
7928 !stop.finishing(),
7929 "the last run out is what actually lands the stop"
7930 );
7931 }
7932
7933 #[tokio::test]
7934 async fn a_loop_already_asked_to_stop_returns_without_waiting_out_a_poll() {
7935 let dir = tempfile::tempdir().unwrap();
7936 let (opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
7937 let stop = Stop::new();
7938 stop.stop();
7939
7940 let began = std::time::Instant::now();
7941 tokio::time::timeout(
7942 Duration::from_secs(2),
7943 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
7944 )
7945 .await
7946 .expect("a stopped loop must return, not sit out its poll interval")
7947 .expect("the loop's own setup and teardown must not fail");
7948 assert!(
7949 began.elapsed() < opts.poll,
7950 "returned only after {:?}, which is a poll interval, not a stop",
7951 began.elapsed()
7952 );
7953 }
7954
7955 #[tokio::test]
7956 async fn a_stop_while_idle_wakes_the_wait_instead_of_sleeping_it_out() {
7957 let dir = tempfile::tempdir().unwrap();
7958 let (opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
7959 let stop = Stop::new();
7960
7961 let asker = {
7964 let stop = stop.clone();
7965 tokio::spawn(async move {
7966 tokio::time::sleep(Duration::from_millis(20)).await;
7967 stop.stop();
7968 })
7969 };
7970
7971 let began = std::time::Instant::now();
7972 tokio::time::timeout(
7973 Duration::from_secs(2),
7974 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
7975 )
7976 .await
7977 .expect("a stop asked for while idle must wake the wait")
7978 .expect("the loop's own setup and teardown must not fail");
7979 asker.await.unwrap();
7980 assert!(
7981 began.elapsed() < opts.poll,
7982 "returned only after {:?}, so the stop waited on the sleep",
7983 began.elapsed()
7984 );
7985 }
7986
7987 #[tokio::test]
7988 async fn a_stopped_loop_leaves_no_status_file_claiming_it_is_running() {
7989 let dir = tempfile::tempdir().unwrap();
7990 let (opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
7991 let stop = Stop::new();
7992 stop.stop();
7993
7994 tokio::time::timeout(
7995 Duration::from_secs(2),
7996 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
7997 )
7998 .await
7999 .expect("a stopped loop must return")
8000 .expect("the loop's own setup and teardown must not fail");
8001
8002 assert!(
8003 home.is_dir(),
8004 "the loop did publish a status file, so its removal is the teardown and not an absence"
8005 );
8006 assert!(
8007 !status_file.exists(),
8008 "a stopped loop clears its status file"
8009 );
8010 assert!(
8011 read_status(&home).is_none(),
8012 "a reader must see no daemon at all, not a heartbeat that merely stopped"
8013 );
8014 }
8015
8016 #[tokio::test]
8017 async fn once_runs_startup_housekeeping_before_an_empty_queue_exits() {
8018 let dir = tempfile::tempdir().unwrap();
8019 let (mut opts, queue, status_file, home, worktrees) = idle_loop(dir.path());
8020 opts.once = true;
8021
8022 let mut settled = RunState::new(
8023 dir.path().join("repo"),
8024 "main".to_owned(),
8025 "abc1234".to_owned(),
8026 "fixture".to_owned(),
8027 Config::default(),
8028 );
8029 settled.status = RunStatus::Ready;
8030 let run_dir = home.join("runs").join(&settled.id);
8031 std::fs::create_dir_all(&run_dir).unwrap();
8032 std::fs::write(
8033 run_dir.join("run.json"),
8034 serde_json::to_string_pretty(&settled).unwrap(),
8035 )
8036 .unwrap();
8037 let questions = Questions::at(home.join("questions"));
8038 let mut question = ask::Question::new(
8039 settled.id.clone(),
8040 "review".to_owned(),
8041 "reviewer-1".to_owned(),
8042 "Continue?".to_owned(),
8043 String::new(),
8044 Vec::new(),
8045 );
8046 questions.put(&mut question).unwrap();
8047
8048 drive(&opts, &queue, &status_file, &home, &worktrees, &Stop::new())
8049 .await
8050 .unwrap();
8051
8052 assert_eq!(
8053 questions.get(&question.id).unwrap().status,
8054 ask::QuestionStatus::Abandoned,
8055 "an empty --once drain still performs startup question cleanup"
8056 );
8057 }
8058
8059 #[test]
8060 fn cache_check_due_fires_immediately_then_waits_out_its_own_interval() {
8061 let t0 = "2026-09-15T00:00:00Z".parse::<Timestamp>().unwrap();
8062
8063 assert!(
8064 cache_check_due(None, t0, CACHE_CHECK_INTERVAL_SECS),
8065 "never checked before: due at once"
8066 );
8067
8068 let one_sec_later = t0 + jiff::SignedDuration::from_secs(1);
8069 assert!(
8070 !cache_check_due(Some(t0), one_sec_later, CACHE_CHECK_INTERVAL_SECS),
8071 "well inside the interval: not due yet"
8072 );
8073
8074 let at_the_edge = t0 + jiff::SignedDuration::from_secs(CACHE_CHECK_INTERVAL_SECS as i64);
8075 assert!(
8076 !cache_check_due(Some(t0), at_the_edge, CACHE_CHECK_INTERVAL_SECS),
8077 "exactly at the edge: not yet due, same convention as `clean::due`"
8078 );
8079
8080 let past_it = t0 + jiff::SignedDuration::from_secs(CACHE_CHECK_INTERVAL_SECS as i64 + 1);
8081 assert!(
8082 cache_check_due(Some(t0), past_it, CACHE_CHECK_INTERVAL_SECS),
8083 "past the interval: due again"
8084 );
8085 }
8086
8087 fn cache_check_opts(dir: &Path, cache_dir: &Path, limit_bytes: u64) -> Opts {
8092 let config = dir.join("magi.toml");
8093 std::fs::write(
8099 &config,
8100 format!(
8101 "[disk]\nmin_free_bytes = 0\nauto_fold = false\ncache_limit_bytes = {limit_bytes}\n\n\
8102 [verify]\ngate = ['CARGO_TARGET_DIR={} cargo make check']\n",
8103 cache_dir.display()
8104 ),
8105 )
8106 .unwrap();
8107 Opts {
8108 config: Some(config),
8109 repo: dir.join("repo"),
8110 ..Opts::default()
8111 }
8112 }
8113
8114 #[tokio::test]
8115 async fn maybe_prune_cache_between_runs_reprunes_only_once_its_own_interval_elapses() {
8116 let dir = tempfile::tempdir().unwrap();
8117 let home = dir.path().join("home");
8118 let cache_dir = dir.path().join("cache");
8119 std::fs::create_dir_all(&cache_dir).unwrap();
8120 std::fs::write(cache_dir.join("a"), vec![0u8; 10]).unwrap();
8121 let opts = cache_check_opts(dir.path(), &cache_dir, 1);
8122
8123 let running = Stop::new();
8126 let mut last_checked = None;
8127 let t0 = "2026-09-15T00:00:00Z".parse::<Timestamp>().unwrap();
8128 maybe_prune_cache_between_runs(&opts.repo, &opts, &home, &running, &mut last_checked, t0)
8129 .await;
8130 assert_eq!(
8131 crate::disk::dir_size(&cache_dir),
8132 0,
8133 "over the cap on the first check ever: pruned at once, no idle queue required"
8134 );
8135 assert_eq!(last_checked, Some(t0));
8136
8137 std::fs::write(cache_dir.join("b"), vec![0u8; 10]).unwrap();
8139 let too_soon = t0 + jiff::SignedDuration::from_secs(1);
8140 maybe_prune_cache_between_runs(
8141 &opts.repo,
8142 &opts,
8143 &home,
8144 &running,
8145 &mut last_checked,
8146 too_soon,
8147 )
8148 .await;
8149 assert_eq!(
8150 crate::disk::dir_size(&cache_dir),
8151 10,
8152 "too soon since the last check: left alone rather than rescanned every call"
8153 );
8154 assert_eq!(
8155 last_checked,
8156 Some(t0),
8157 "an idle check does not reset the clock"
8158 );
8159
8160 let due_again = t0 + jiff::SignedDuration::from_secs(CACHE_CHECK_INTERVAL_SECS as i64 + 1);
8162 maybe_prune_cache_between_runs(
8163 &opts.repo,
8164 &opts,
8165 &home,
8166 &running,
8167 &mut last_checked,
8168 due_again,
8169 )
8170 .await;
8171 assert_eq!(
8172 crate::disk::dir_size(&cache_dir),
8173 0,
8174 "due again: pruned back under the cap"
8175 );
8176 }
8177
8178 #[tokio::test]
8186 async fn a_stop_already_asked_for_skips_the_between_runs_cache_walk() {
8187 let dir = tempfile::tempdir().unwrap();
8188 let home = dir.path().join("home");
8189 let cache_dir = dir.path().join("cache");
8190 std::fs::create_dir_all(&cache_dir).unwrap();
8191 std::fs::write(cache_dir.join("a"), vec![0u8; 10]).unwrap();
8192 let opts = cache_check_opts(dir.path(), &cache_dir, 1);
8193
8194 let stop = Stop::new();
8195 stop.stop();
8196 assert!(
8197 !stop.finishing(),
8198 "no run is in flight at a between-runs boundary, so nothing else \
8199 would tell the operator this stop had not taken effect yet"
8200 );
8201
8202 let mut last_checked = None;
8203 let t0 = "2026-09-15T00:00:00Z".parse::<Timestamp>().unwrap();
8204 maybe_prune_cache_between_runs(&opts.repo, &opts, &home, &stop, &mut last_checked, t0)
8205 .await;
8206 assert_eq!(
8207 crate::disk::dir_size(&cache_dir),
8208 10,
8209 "over its cap, and due for the first check ever, but a stop outranks \
8210 it: the cap is a standing policy the next start measures again"
8211 );
8212 assert_eq!(
8213 last_checked, None,
8214 "a check that never happened must not claim the interval"
8215 );
8216 }
8217
8218 #[tokio::test]
8229 async fn cache_prune_reaches_a_queue_that_never_goes_idle() {
8230 let dir = tempfile::tempdir().unwrap();
8231 let cache_dir = dir.path().join("cache");
8232 std::fs::create_dir_all(&cache_dir).unwrap();
8233 std::fs::write(cache_dir.join("stale"), vec![0u8; 4096]).unwrap();
8234
8235 let mut opts = cache_check_opts(dir.path(), &cache_dir, 1);
8236 opts.poll = Duration::from_millis(20);
8237 opts.max_attempts = 1_000;
8238
8239 let queue = Queue::at(dir.path().join("queue"));
8240 for i in 0..128 {
8241 let mut task = Task::new(
8242 format!("queued work {i}"),
8243 format!("queued work {i}"),
8244 opts.repo.clone(),
8245 Source::Human,
8246 );
8247 queue.put(&mut task).unwrap();
8248 }
8249
8250 let home = dir.path().join("home");
8251 let worktrees = dir.path().join("wt");
8252 let status_file = home.join("daemon.json");
8253 let stop = Stop::new();
8254 let stopper = {
8255 let stop = stop.clone();
8256 let queue = Queue::at(dir.path().join("queue"));
8257 tokio::spawn(async move {
8258 for _ in 0..2_000 {
8262 if queue.list().iter().filter(|t| t.attempts == 1).count() >= 2 {
8263 break;
8264 }
8265 tokio::time::sleep(Duration::from_millis(5)).await;
8266 }
8267 stop.stop();
8268 })
8269 };
8270
8271 tokio::time::timeout(
8272 Duration::from_secs(20),
8273 drive(&opts, &queue, &status_file, &home, &worktrees, &stop),
8274 )
8275 .await
8276 .expect("the loop must not hang on a queue that keeps producing failing work")
8277 .expect("the loop's own setup and teardown must not fail");
8278 stopper.await.unwrap();
8279
8280 let after = queue.list();
8281 let completed = after.iter().filter(|t| t.attempts == 1).count();
8282 assert!(
8283 completed >= 2,
8284 "the harness must have processed multiple tasks to exercise a busy queue \
8285 (got {completed} completed task(s))"
8286 );
8287 assert!(
8288 after.iter().any(|t| t.status.runnable()),
8289 "unprocessed tasks must remain: the external stop, rather than an \
8290 exhausted backlog, must end the test"
8291 );
8292 assert!(
8293 after
8294 .iter()
8295 .filter(|t| t.attempts > 0)
8296 .all(|t| { t.attempts == 1 && t.status == TaskStatus::Held }),
8297 "a busy backlog must not retry a task's unchanged startup failure"
8298 );
8299
8300 assert_eq!(
8301 crate::disk::dir_size(&cache_dir),
8302 0,
8303 "an oversized cache must not be left to grow unboundedly just because the \
8304 queue kept the loop busy the whole time"
8305 );
8306 }
8307
8308 #[test]
8309 fn task_question_reconciliation_keeps_references_and_retires_manual_releases() {
8310 let dir = tempfile::tempdir().unwrap();
8311 let queue = Queue::at(dir.path().join("queue"));
8312 let questions = Questions::at(dir.path().join("questions"));
8313 let mut task = task();
8314 queue.put(&mut task).unwrap();
8315
8316 let mut task_question = ask::Question::new(
8317 task.id.clone(),
8318 crate::conduct::NODE.to_owned(),
8319 "conduct".to_owned(),
8320 "Which backend?".to_owned(),
8321 String::new(),
8322 Vec::new(),
8323 );
8324 questions.put(&mut task_question).unwrap();
8325 task.block(vec![task_question.id.clone()], None);
8326 queue.put(&mut task).unwrap();
8327
8328 let mut run_question = ask::Question::new(
8329 "20260101-000000-run1".to_owned(),
8330 "review".to_owned(),
8331 "reviewer-1".to_owned(),
8332 "Run question".to_owned(),
8333 String::new(),
8334 Vec::new(),
8335 );
8336 questions.put(&mut run_question).unwrap();
8337
8338 let mut coincidental = ask::Question::new(
8343 task.id.clone(),
8344 "review".to_owned(),
8345 "reviewer-1".to_owned(),
8346 "Unrelated review question".to_owned(),
8347 String::new(),
8348 Vec::new(),
8349 );
8350 questions.put(&mut coincidental).unwrap();
8351
8352 reconcile_task_questions(&queue, &questions);
8353 assert!(questions.get(&task_question.id).unwrap().status.open());
8354 assert!(questions.get(&run_question.id).unwrap().status.open());
8355 assert!(questions.get(&coincidental.id).unwrap().status.open());
8356
8357 task.release();
8358 queue.put(&mut task).unwrap();
8359 reconcile_task_questions(&queue, &questions);
8360 assert_eq!(
8361 questions.get(&task_question.id).unwrap().status,
8362 ask::QuestionStatus::Abandoned
8363 );
8364 assert!(
8365 questions.get(&run_question.id).unwrap().status.open(),
8366 "run questions remain the run janitor's responsibility"
8367 );
8368 assert!(
8369 questions.get(&coincidental.id).unwrap().status.open(),
8370 "a non-conductor question must not be abandoned just because its \
8371 run id coincides with a task id"
8372 );
8373 }
8374
8375 #[test]
8376 fn a_freshly_started_running_task_is_never_stalled() {
8377 let dir = tempfile::tempdir().unwrap();
8378 let mut t = task();
8379 t.start("run-1".to_owned());
8380 assert!(!is_stalled(&t, dir.path(), Timestamp::now()));
8383 }
8384
8385 #[test]
8386 fn a_long_running_task_with_no_live_daemon_is_stalled() {
8387 let dir = tempfile::tempdir().unwrap();
8388 let mut t = task();
8389 t.start("run-1".to_owned());
8390 t.updated_at = Timestamp::now()
8391 - jiff::SignedDuration::from_secs(STALLED_RUNNING.as_secs() as i64 + 60);
8392 assert!(is_stalled(&t, dir.path(), Timestamp::now()));
8393 assert_eq!(
8394 stalled_tasks(
8395 &Queue::at(dir.path().join("q")),
8396 dir.path(),
8397 Timestamp::now()
8398 )
8399 .len(),
8400 0,
8401 "the task was never written to this queue"
8402 );
8403 }
8404
8405 #[test]
8406 fn a_long_running_task_a_live_daemon_still_names_is_not_stalled() {
8407 let dir = tempfile::tempdir().unwrap();
8408 let mut t = task();
8409 t.id = "20260903-080340-0167".to_owned();
8410 t.start("20260903-080619-01c2".to_owned());
8411 t.updated_at = Timestamp::now()
8412 - jiff::SignedDuration::from_secs(STALLED_RUNNING.as_secs() as i64 + 60);
8413
8414 let mut status = Status::new();
8415 status.current = vec![Current {
8416 task: t.id.clone(),
8417 run: "20260903-080619-01c2".to_owned(),
8418 }];
8419 write_status_to(&dir.path().join("daemon.json"), &status).unwrap();
8420
8421 assert!(
8422 !is_stalled(&t, dir.path(), Timestamp::now()),
8423 "a live daemon's own heartbeat rules out stalled, however long the task has run"
8424 );
8425 }
8426
8427 fn backdate_task(queue: &Queue, id: &str, seconds_ago: i64) {
8431 let path = queue.path_of(id);
8432 let body = std::fs::read_to_string(&path).unwrap();
8433 let mut v: serde_json::Value = serde_json::from_str(&body).unwrap();
8434 let old = Timestamp::now() - jiff::SignedDuration::from_secs(seconds_ago);
8435 v["updated_at"] = serde_json::Value::String(old.to_string());
8436 std::fs::write(&path, serde_json::to_string_pretty(&v).unwrap()).unwrap();
8437 }
8438
8439 #[test]
8440 fn stalled_tasks_still_reaches_a_task_reclaim_could_not_claim_yet() {
8441 let dir = tempfile::tempdir().unwrap();
8454 let queue = Queue::at(dir.path().join("queue"));
8455 let home = dir.path().join("home");
8456
8457 let mut t = task();
8458 t.id = "20260101-000001-lock".to_owned();
8459 t.start("run-1".to_owned());
8460 queue.put(&mut t).unwrap();
8461 backdate_task(&queue, &t.id, STALLED_RUNNING.as_secs() as i64 + 60);
8462 std::fs::write(
8463 dir.path().join("queue").join(format!("{}.lock", t.id)),
8464 "not a pid",
8465 )
8466 .unwrap();
8467
8468 let now = Timestamp::now();
8469 assert!(
8470 reclaim_orphaned_running(&queue, 2).is_empty(),
8471 "the unparseable lock is still well within STALE_CLAIM, so the claim fails \
8472 and reclaim must leave the task alone"
8473 );
8474 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Running);
8475
8476 let stalled = stalled_tasks(&queue, &home, now);
8477 assert_eq!(
8478 stalled.len(),
8479 1,
8480 "reclaim's inability to claim it yet must not hide it from the conductor"
8481 );
8482 assert_eq!(stalled[0].id, t.id);
8483 }
8484
8485 #[test]
8486 fn ordinary_dead_daemon_task_is_shown_stalled_before_reclaim_and_can_be_requeued() {
8487 let dir = tempfile::tempdir().unwrap();
8488 crate::run::set_home(dir.path().join("run-home"));
8489 let queue = Queue::at(dir.path().join("queue"));
8490 let home = dir.path().join("home");
8491 let questions = Questions::at(dir.path().join("questions"));
8492
8493 let mut t = task();
8494 t.id = "20260101-000003-dead".to_owned();
8495 t.start("missing-run".to_owned());
8496 queue.put(&mut t).unwrap();
8497 backdate_task(&queue, &t.id, STALLED_RUNNING.as_secs() as i64 + 60);
8498
8499 let stalled = stalled_tasks(&queue, &home, Timestamp::now());
8502 assert_eq!(
8503 stalled.iter().map(|task| &task.id).collect::<Vec<_>>(),
8504 [&t.id]
8505 );
8506 assert_eq!(reclaim_orphaned_running(&queue, 2), [t.id.clone()]);
8507 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
8508
8509 crate::conduct::apply(
8512 &queue,
8513 &questions,
8514 &crate::conduct::Verdict {
8515 decisions: vec![crate::conduct::Decision {
8516 id: t.id.clone(),
8517 recovery: Some(crate::conduct::Recovery::Requeue),
8518 ..crate::conduct::Decision::default()
8519 }],
8520 },
8521 )
8522 .unwrap();
8523 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
8524 }
8525
8526 #[test]
8527 fn stalled_tasks_reports_exactly_the_tasks_is_stalled_agrees_on() {
8528 let dir = tempfile::tempdir().unwrap();
8529 let queue = Queue::at(dir.path().join("queue"));
8530 let home = dir.path().join("home");
8531
8532 let mut fresh = task();
8533 fresh.id = "20260101-000001-aaaa".to_owned();
8534 fresh.start("run-1".to_owned());
8535 queue.put(&mut fresh).unwrap();
8536
8537 let mut old = task();
8538 old.id = "20260101-000002-bbbb".to_owned();
8539 old.start("run-2".to_owned());
8540 queue.put(&mut old).unwrap();
8541 backdate_task(&queue, &old.id, STALLED_RUNNING.as_secs() as i64 + 60);
8542
8543 let stalled = stalled_tasks(&queue, &home, Timestamp::now());
8544 assert_eq!(stalled.len(), 1);
8545 assert_eq!(stalled[0].id, old.id);
8546 }
8547
8548 #[test]
8549 fn queued_and_finished_task_views_partition_by_status() {
8550 let dir = tempfile::tempdir().unwrap();
8551 let queue = Queue::at(dir.path().join("queue"));
8552
8553 let mut queued = task();
8554 queued.id = "20260101-000001-aaaa".to_owned();
8555 queue.put(&mut queued).unwrap();
8556
8557 let mut failed = task();
8558 failed.id = "20260101-000002-bbbb".to_owned();
8559 failed.start("run-1".to_owned());
8560 failed.fail("gate red", 5);
8561 queue.put(&mut failed).unwrap();
8562
8563 let mut held = task();
8564 held.id = "20260101-000003-cccc".to_owned();
8565 held.hold_machine(None);
8566 queue.put(&mut held).unwrap();
8567
8568 let mut running = task();
8569 running.id = "20260101-000004-dddd".to_owned();
8570 running.start("run-2".to_owned());
8571 queue.put(&mut running).unwrap();
8572
8573 let queued_ids: Vec<String> = queued_tasks(&queue).into_iter().map(|t| t.id).collect();
8574 assert_eq!(queued_ids, [queued.id.clone()]);
8575
8576 let mut finished_ids: Vec<String> =
8577 finished_tasks(&queue).into_iter().map(|t| t.id).collect();
8578 finished_ids.sort_unstable();
8579 let mut want = vec![failed.id.clone(), held.id.clone()];
8580 want.sort_unstable();
8581 assert_eq!(finished_ids, want);
8582 }
8583
8584 #[test]
8585 fn dependency_resolution_preserves_existing_short_and_full_running_chains() {
8586 for short in [false, true] {
8587 let dir = tempfile::tempdir().unwrap();
8588 let queue = Queue::at(dir.path().join("queue"));
8589 let questions = ask::Questions::at(dir.path().join("questions"));
8590 let mut running = task();
8591 running.id = "20261013-192213-feb2".into();
8592 running.start("existing-run".into());
8593 queue.put(&mut running).unwrap();
8594 let mut middle = task();
8595 middle.id = "20261013-193611-ca5b".into();
8596 middle.block(
8597 vec![if short {
8598 "feb2".into()
8599 } else {
8600 running.id.clone()
8601 }],
8602 None,
8603 );
8604 queue.put(&mut middle).unwrap();
8605 let mut waiting = task();
8606 waiting.id = "20261013-202849-ee37".into();
8607 waiting.block(
8608 vec![if short {
8609 "ca5b".into()
8610 } else {
8611 middle.id.clone()
8612 }],
8613 Some("wait for implementation".into()),
8614 );
8615 queue.put(&mut waiting).unwrap();
8616 assert_eq!(queue.get("ca5b").unwrap().id, middle.id);
8617 let inv = crate::blockers::Inventory::load(&queue, &questions);
8618 assert_eq!(inv.waits_on(&waiting), ["ca5b (blocked → feb2 running)"]);
8619 assert!(inv.stuck().is_empty());
8620 let report = crate::triage::run_once(&queue, &questions, None, Timestamp::now());
8621 assert!(report.quarantined.is_empty());
8622 assert!(report.asked.is_empty());
8623 resolve_blockers(&queue, &questions);
8624 assert_eq!(queue.get(&waiting.id).unwrap().status, TaskStatus::Blocked);
8625 assert!(questions.list().is_empty());
8626 running.succeed();
8629 queue.put(&mut running).unwrap();
8630 resolve_blockers(&queue, &questions);
8631 assert_eq!(queue.get(&middle.id).unwrap().status, TaskStatus::Queued);
8632 assert_eq!(queue.get(&waiting.id).unwrap().status, TaskStatus::Blocked);
8633 middle = queue.get(&middle.id).unwrap();
8634 middle.succeed();
8635 queue.put(&mut middle).unwrap();
8636 resolve_blockers(&queue, &questions);
8637 let after = queue.get(&waiting.id).unwrap();
8638 assert_eq!(after.status, TaskStatus::Queued);
8639 assert!(after.blocked_by.is_empty());
8640 assert_eq!(after.instruction, waiting.instruction);
8641 }
8642 }
8643
8644 #[test]
8645 fn dependency_resolution_unknown_records_never_hold_ask_or_release() {
8646 for future_schema in [false, true] {
8647 let name = "20261013-193611-ca5b";
8648 let mut future = task();
8649 future.id = name.into();
8650 future.schema = crate::queue::SCHEMA + 1;
8651 let contents = if future_schema {
8652 serde_json::to_string(&future).unwrap()
8653 } else {
8654 "corrupt".into()
8655 };
8656 let dir = tempfile::tempdir().unwrap();
8657 let queue = Queue::at(dir.path().join("queue"));
8658 let questions = ask::Questions::at(dir.path().join("questions"));
8659 let mut waiting = task();
8660 waiting.id = "20261013-202849-ee37".into();
8661 waiting.block(vec!["ca5b".into()], None);
8662 queue.put(&mut waiting).unwrap();
8663 std::fs::write(queue.path_of(name), contents).unwrap();
8664 assert_eq!(queue.resolve_id("ca5b").unwrap(), name);
8665 assert!(
8666 crate::queue::missing_blockers(&queue, &questions, &waiting.blocked_by).is_empty()
8667 );
8668 let report = crate::triage::run_once(&queue, &questions, None, Timestamp::now());
8669 assert!(report.asked.is_empty());
8670 resolve_blockers(&queue, &questions);
8671 assert_eq!(queue.get(&waiting.id).unwrap().status, TaskStatus::Blocked);
8672 assert!(questions.list().is_empty());
8673 }
8674 let dir = tempfile::tempdir().unwrap();
8675 let queue = Queue::at(dir.path().join("queue"));
8676 let questions = ask::Questions::at(dir.path().join("questions"));
8677 let mut waiting = task();
8678 waiting.id = "20261013-202849-ee37".into();
8679 waiting.block(vec!["ca5b".into()], None);
8680 queue.put(&mut waiting).unwrap();
8681 let mut done = task();
8682 done.id = "20261013-193611-ca5b".into();
8683 done.succeed();
8684 queue.put(&mut done).unwrap();
8685 std::fs::write(queue.path_of("20261014-193611-ca5b"), "bad").unwrap();
8688 assert!(queue.get("ca5b").is_err());
8689 assert_eq!(
8690 crate::queue::dependency(&queue, &questions, "ca5b"),
8691 crate::queue::Dependency::Ambiguous
8692 );
8693 resolve_blockers(&queue, &questions);
8694 let report = crate::triage::run_once(&queue, &questions, None, Timestamp::now());
8695 assert!(report.asked.is_empty());
8696 assert_eq!(queue.get(&waiting.id).unwrap().status, TaskStatus::Blocked);
8697 assert_eq!(queue.get(&done.id).unwrap().status, TaskStatus::Done);
8698 }
8699
8700 #[test]
8701 fn dependency_resolution_preserves_independent_consent_and_holds_after_completion() {
8702 for held in [false, true] {
8703 for question in [false, true] {
8704 for reference in ["ca5b", "20261013-193611-ca5b"] {
8705 let dir = tempfile::tempdir().unwrap();
8706 let queue = Queue::at(dir.path().join("queue"));
8707 let questions = ask::Questions::at(dir.path().join("questions"));
8708 let mut prerequisite = task();
8709 prerequisite.id = "20261013-193611-ca5b".into();
8710 prerequisite.start("existing-run".into());
8711 queue.put(&mut prerequisite).unwrap();
8712 let mut waiting = task();
8713 waiting.id = "20261013-202849-ee37".into();
8714 if held {
8715 waiting.hold_machine(Some("waiting for separate consent".into()));
8716 }
8717 queue.put(&mut waiting).unwrap();
8718 let reason = "Deleting saved data requires separate owner consent";
8719 crate::conduct::apply(
8720 &queue,
8721 &questions,
8722 &crate::conduct::Verdict {
8723 decisions: vec![crate::conduct::Decision {
8724 id: waiting.id.clone(),
8725 blocked_by: vec![reference.into()],
8726 question: question.then(|| "May the saved data be deleted?".into()),
8727 recovery: (!question).then_some(crate::conduct::Recovery::Hold),
8728 reason: Some(reason.into()),
8729 ..crate::conduct::Decision::default()
8730 }],
8731 },
8732 )
8733 .unwrap();
8734 let before = queue.get(&waiting.id).unwrap();
8735 if question {
8736 assert_eq!(before.status, TaskStatus::Blocked);
8737 let q = questions.get(&before.blocked_by[0]).unwrap();
8738 assert_eq!(q.summary, "May the saved data be deleted?");
8739 assert!(q.status.open());
8740 } else {
8741 assert_eq!(before.status, TaskStatus::Held);
8742 let recorded_reason = before.hold_reason.as_deref().unwrap();
8743 assert!(recorded_reason.starts_with(reason));
8744 if held {
8745 assert!(recorded_reason.contains("waiting for separate consent"));
8746 }
8747 }
8748 prerequisite.succeed();
8749 queue.put(&mut prerequisite).unwrap();
8750 resolve_blockers(&queue, &questions);
8751 let after = queue.get(&waiting.id).unwrap();
8752 assert_eq!(after.status, before.status);
8753 assert_eq!(after.blocked_by, before.blocked_by);
8754 assert_eq!(after.hold_reason, before.hold_reason);
8755 assert_eq!(after.instruction, waiting.instruction);
8756 if question {
8757 assert!(questions.get(&after.blocked_by[0]).unwrap().status.open());
8758 }
8759 assert!(
8760 queue.next_runnable().is_none(),
8761 "dependency completion is not consent"
8762 );
8763 }
8764 }
8765 }
8766 }
8767
8768 #[test]
8769 fn resolve_blockers_clears_a_done_dependency_and_keeps_an_unresolved_one() {
8770 let dir = tempfile::tempdir().unwrap();
8771 let queue = Queue::at(dir.path().join("queue"));
8772 let questions = ask::Questions::at(dir.path().join("questions"));
8773
8774 let mut dep = task();
8775 dep.id = "20260101-000001-dep0".to_owned();
8776 dep.succeed();
8777 queue.put(&mut dep).unwrap();
8778
8779 let mut still_going = task();
8780 still_going.id = "20260101-000002-dep1".to_owned();
8781 queue.put(&mut still_going).unwrap();
8782
8783 let mut blocked = task();
8784 blocked.id = "20260101-000003-main".to_owned();
8785 blocked.block(
8786 vec![dep.id.clone(), still_going.id.clone()],
8787 Some("waits on both".to_owned()),
8788 );
8789 queue.put(&mut blocked).unwrap();
8790
8791 resolve_blockers(&queue, &questions);
8792
8793 let after = queue.get(&blocked.id).unwrap();
8794 assert_eq!(
8795 after.status,
8796 TaskStatus::Blocked,
8797 "one dependency is still outstanding"
8798 );
8799 assert_eq!(after.blocked_by, [still_going.id.clone()]);
8800 }
8801
8802 #[test]
8803 fn resolve_blockers_carries_an_answers_content_onto_the_task_and_unblocks_it() {
8804 let dir = tempfile::tempdir().unwrap();
8805 let queue = Queue::at(dir.path().join("queue"));
8806 let questions = ask::Questions::at(dir.path().join("questions"));
8807
8808 let mut q = crate::ask::Question::new(
8809 "20260101-000001-main".to_owned(),
8810 crate::conduct::NODE.to_owned(),
8811 "conduct".to_owned(),
8812 "Which backend?".to_owned(),
8813 String::new(),
8814 Vec::new(),
8815 );
8816 questions.put(&mut q).unwrap();
8817 q.answer(crate::ask::Answer::Text("SQLite".to_owned()))
8818 .unwrap();
8819 questions.put(&mut q).unwrap();
8820
8821 let mut blocked = task();
8822 blocked.id = "20260101-000001-main".to_owned();
8823 blocked.block(vec![q.id.clone()], Some("which backend?".to_owned()));
8824 queue.put(&mut blocked).unwrap();
8825
8826 resolve_blockers(&queue, &questions);
8827
8828 let after = queue.get(&blocked.id).unwrap();
8829 assert_eq!(
8830 after.status,
8831 TaskStatus::Queued,
8832 "the only blocker resolved"
8833 );
8834 assert_eq!(after.answers.len(), 1);
8835 assert_eq!(after.answers[0].question, "Which backend?");
8836 assert_eq!(after.answers[0].answer, "SQLite");
8837
8838 let instruction = instruction_for(&after);
8840 assert!(instruction.contains("Which backend?"));
8841 assert!(instruction.contains("SQLite"));
8842 }
8843
8844 #[test]
8845 fn resolve_blockers_holds_a_task_whose_conductor_question_was_abandoned() {
8846 let dir = tempfile::tempdir().unwrap();
8847 let queue = Queue::at(dir.path().join("queue"));
8848 let questions = ask::Questions::at(dir.path().join("questions"));
8849
8850 let mut q = crate::ask::Question::new(
8851 "20260101-000001-main".to_owned(),
8852 crate::conduct::NODE.to_owned(),
8853 "conduct".to_owned(),
8854 "Is the setup done?".to_owned(),
8855 String::new(),
8856 Vec::new(),
8857 );
8858 q.abandon("no answer within 60s of asking");
8859 questions.put(&mut q).unwrap();
8860
8861 let mut blocked = task();
8862 blocked.id = "20260101-000001-main".to_owned();
8863 blocked.block(vec![q.id.clone()], Some("setup?".to_owned()));
8864 queue.put(&mut blocked).unwrap();
8865
8866 resolve_blockers(&queue, &questions);
8867
8868 let after = queue.get(&blocked.id).unwrap();
8869 assert_eq!(
8870 after.status,
8871 TaskStatus::Held,
8872 "never left blocked on nothing"
8873 );
8874 assert!(!after.operator_held(), "a machine hold, for triage");
8875 assert!(
8876 after
8877 .hold_reason
8878 .as_deref()
8879 .unwrap_or_default()
8880 .contains("went unanswered")
8881 );
8882 }
8883
8884 #[test]
8885 fn resolve_blockers_restores_a_held_task_to_held_instead_of_queuing_it() {
8886 let dir = tempfile::tempdir().unwrap();
8892 let queue = Queue::at(dir.path().join("queue"));
8893 let questions = ask::Questions::at(dir.path().join("questions"));
8894
8895 let mut q = crate::ask::Question::new(
8896 "20260101-000001-main".to_owned(),
8897 crate::conduct::NODE.to_owned(),
8898 "conduct".to_owned(),
8899 "How should this be handled?".to_owned(),
8900 String::new(),
8901 Vec::new(),
8902 );
8903 questions.put(&mut q).unwrap();
8904 q.answer(crate::ask::Answer::Text(
8905 "leave it held, a human will look at it later".to_owned(),
8906 ))
8907 .unwrap();
8908 questions.put(&mut q).unwrap();
8909
8910 let mut held = task();
8911 held.id = "20260101-000001-main".to_owned();
8912 held.hold_machine(Some("out of attempts".to_owned()));
8913 held.block(vec![q.id.clone()], Some("what now?".to_owned()));
8914 queue.put(&mut held).unwrap();
8915
8916 resolve_blockers(&queue, &questions);
8917
8918 let after = queue.get(&held.id).unwrap();
8919 assert_eq!(after.status, TaskStatus::Held);
8920 assert_eq!(after.hold_reason.as_deref(), Some("out of attempts"));
8921 assert_eq!(
8922 after.answers[0].answer,
8923 "leave it held, a human will look at it later"
8924 );
8925 }
8926
8927 #[test]
8928 fn resolve_blockers_releases_a_task_whose_dependency_was_deleted_on_purpose() {
8929 let dir = tempfile::tempdir().unwrap();
8930 let queue = Queue::at(dir.path().join("queue"));
8931 let questions = ask::Questions::at(dir.path().join("questions"));
8932 let mut dep = task();
8933 dep.id = "20260101-000001-gone".to_owned();
8934 queue.put(&mut dep).unwrap();
8935 let mut blocked = task();
8936 blocked.id = "20260101-000003-main".to_owned();
8937 blocked.block(vec![dep.id.clone()], None);
8938 queue.put(&mut blocked).unwrap();
8939
8940 let claim = queue.claim(&blocked.id).unwrap();
8942 queue.remove(&dep.id, false, &questions).unwrap();
8943 drop(claim);
8944 resolve_blockers(&queue, &questions);
8945
8946 let after = queue.get(&blocked.id).unwrap();
8947 assert_eq!(after.status, TaskStatus::Queued);
8948 assert!(after.blocked_by.is_empty());
8949 }
8950
8951 #[test]
8952 fn resolve_blockers_holds_a_task_whose_dependency_was_deleted() {
8953 let dir = tempfile::tempdir().unwrap();
8959 let queue = Queue::at(dir.path().join("queue"));
8960 let questions = ask::Questions::at(dir.path().join("questions"));
8961
8962 let mut still_going = task();
8963 still_going.id = "20260101-000002-dep1".to_owned();
8964 queue.put(&mut still_going).unwrap();
8965
8966 let mut blocked = task();
8967 blocked.id = "20260101-000003-main".to_owned();
8968 blocked.block(
8969 vec!["20260101-000001-gone".to_owned(), still_going.id.clone()],
8970 Some("waits on both".to_owned()),
8971 );
8972 queue.put(&mut blocked).unwrap();
8973
8974 resolve_blockers(&queue, &questions);
8975
8976 let after = queue.get(&blocked.id).unwrap();
8977 assert_eq!(
8978 after.status,
8979 TaskStatus::Held,
8980 "a missing dependency must not leave the task blocked forever"
8981 );
8982 assert_eq!(after.hold_source, Some(crate::queue::HoldSource::Machine));
8983 assert!(after.blocked_by.is_empty());
8984 let reason = after.hold_reason.as_deref().unwrap_or_default();
8985 assert!(
8986 reason.contains("20260101-000001-gone"),
8987 "the missing id must be named so an operator can tell what happened: {reason}"
8988 );
8989 assert!(
8990 reason.contains(&still_going.id),
8991 "the still-valid dependency must not silently vanish from the record: {reason}"
8992 );
8993 }
8994
8995 #[test]
8996 fn instruction_for_is_unchanged_without_any_answers() {
8997 let t = task();
8998 assert_eq!(instruction_for(&t), t.instruction);
8999 }
9000
9001 #[test]
9002 fn task_attachments_are_absolute_and_a_missing_file_is_an_error() {
9003 let dir = tempfile::tempdir().unwrap();
9004 let q = Queue::at(dir.path().join("queue"));
9005 let src = dir.path().join("shot.png");
9006 std::fs::write(&src, "x").unwrap();
9007 let mut t = task();
9008 q.attach(&mut t, &[src]).unwrap();
9009 let paths = task_attachments(&q, &t).unwrap();
9010 assert_eq!(paths.len(), 1);
9011 assert!(paths[0].is_absolute() && paths[0].is_file());
9012 std::fs::remove_file(&paths[0]).unwrap();
9013 let err = task_attachments(&q, &t).unwrap_err().to_string();
9014 assert!(err.contains("shot.png"), "{err}");
9015 }
9016
9017 #[test]
9018 fn resumed_instruction_is_unchanged_without_any_answers() {
9019 let t = task();
9020 assert_eq!(resumed_instruction(&t.instruction, &t), t.instruction);
9021 }
9022
9023 #[test]
9024 fn resumed_instruction_carries_a_new_answer_onto_the_old_run() {
9025 let mut t = task();
9026 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
9027 let old = t.instruction.clone();
9031
9032 let refreshed = resumed_instruction(&old, &t);
9033 assert!(refreshed.starts_with(&old), "the original text is kept");
9034 assert!(refreshed.contains("Which backend?"));
9035 assert!(refreshed.contains("SQLite"));
9036 }
9037
9038 #[test]
9039 fn resumed_instruction_keeps_an_original_answers_heading() {
9040 let mut t = task();
9041 t.instruction = "Context\n\n# Operator answers\n\nThis is part of the task.".to_owned();
9042 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
9043
9044 let refreshed = resumed_instruction(&t.instruction, &t);
9045
9046 assert!(
9047 refreshed.starts_with(&t.instruction),
9048 "an answers heading in the original instruction is not the appended block"
9049 );
9050 assert_eq!(refreshed.matches(ANSWERS_HEADER).count(), 2);
9051 assert!(refreshed.contains("Which backend?"));
9052 assert!(refreshed.contains("SQLite"));
9053
9054 let repeated = resumed_instruction(&refreshed, &t);
9055 assert_eq!(
9056 repeated, refreshed,
9057 "only the final appended block is refreshed"
9058 );
9059 }
9060
9061 #[test]
9062 fn resumed_instruction_does_not_duplicate_across_repeated_resumes() {
9063 let mut t = task();
9064 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
9065
9066 let once = resumed_instruction(&t.instruction, &t);
9070 let twice = resumed_instruction(&once, &t);
9071 assert_eq!(once, twice);
9072 assert_eq!(once.matches("Which backend?").count(), 1);
9073
9074 t.record_answer("Which cache?".to_owned(), "Redis".to_owned());
9076 let refreshed = resumed_instruction(&once, &t);
9077 assert_eq!(refreshed.matches(ANSWERS_HEADER).count(), 1);
9078 assert!(refreshed.contains("Which backend?"));
9079 assert!(refreshed.contains("Which cache?"));
9080 }
9081
9082 #[test]
9083 fn prepare_instruction_covers_all_three_starters() {
9084 let mut t = task();
9085 t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
9086
9087 assert_eq!(
9090 prepare_instruction(&Starter::Start, None, &t),
9091 Some(instruction_for(&t))
9092 );
9093
9094 let old = t.instruction.clone();
9097 assert_eq!(
9098 prepare_instruction(&Starter::Resume("some-run".to_owned()), Some(&old), &t),
9099 Some(resumed_instruction(&old, &t))
9100 );
9101
9102 assert_eq!(
9106 prepare_instruction(&Starter::Review("magi/eba2/A".to_owned()), Some(&old), &t),
9107 None
9108 );
9109 }
9110
9111 #[test]
9112 fn choose_starter_prefers_review_over_resume_when_the_branch_survived() {
9113 assert_eq!(
9114 choose_starter(Some("magi/eba2/A"), true, Some("some-run")),
9115 Starter::Review("magi/eba2/A".to_owned())
9116 );
9117 }
9118
9119 #[test]
9120 fn choose_starter_falls_back_to_start_when_the_review_branch_is_gone() {
9121 assert_eq!(
9122 choose_starter(Some("magi/eba2/A"), false, Some("some-run")),
9123 Starter::Start,
9124 "a vanished review branch must not fall back to resuming the old run either"
9125 );
9126 }
9127
9128 #[test]
9129 fn a_refused_handover_retries_as_a_review_of_the_same_branch() {
9130 let mut t = task();
9131 t.start("old-run".to_owned());
9132 t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
9133 t.release();
9134 let branch = t.review_branch.take();
9135 assert_eq!(
9136 choose_starter(branch.as_deref(), true, Some("old-run")),
9137 Starter::Review("magi/eba2/A".to_owned()),
9138 "a review wins over resuming the old run"
9139 );
9140 }
9141
9142 #[test]
9143 fn choose_starter_resumes_or_starts_when_there_is_no_review_choice_at_all() {
9144 assert_eq!(
9145 choose_starter(None, false, Some("some-run")),
9146 Starter::Resume("some-run".to_owned())
9147 );
9148 assert_eq!(choose_starter(None, false, None), Starter::Start);
9149 }
9150
9151 #[test]
9152 fn an_explicit_release_forces_a_fresh_competition_even_with_a_resumable_run() {
9153 let mut released = task();
9154 released.start("stalled-run".to_owned());
9155 released.requeue();
9156 let unfinished = (!released.fresh_start)
9157 .then(|| Some("stalled-run".to_owned()))
9158 .flatten();
9159 assert_eq!(
9160 choose_starter(None, false, unfinished.as_deref()),
9161 Starter::Start,
9162 "release keeps run history but must not resume it"
9163 );
9164 assert_eq!(released.runs, ["stalled-run"]);
9165 }
9166
9167 #[test]
9168 fn an_ordinary_release_keeps_a_resumable_run_available() {
9169 let mut released = task();
9170 released.start("stalled-run".to_owned());
9171 released.release();
9172 let unfinished = (!released.fresh_start)
9173 .then(|| Some("stalled-run".to_owned()))
9174 .flatten();
9175 assert_eq!(
9176 choose_starter(None, false, unfinished.as_deref()),
9177 Starter::Resume("stalled-run".to_owned()),
9178 "manual release must preserve the normal resume path"
9179 );
9180 }
9181
9182 #[test]
9183 fn a_blocked_run_that_spent_every_review_round_has_exhausted_its_budget() {
9184 let mut state = run_state(RunStatus::Blocked);
9185 state.config.graph.review_rounds = 3;
9186 state.reviews = vec![review_round(1), review_round(2), review_round(3)];
9187 assert!(exhausted_review_budget(&state));
9188
9189 state.reviews.pop();
9191 assert!(!exhausted_review_budget(&state));
9192
9193 let mut stalled = run_state(RunStatus::Stalled);
9196 stalled.config.graph.review_rounds = 1;
9197 stalled.reviews = vec![review_round(1)];
9198 assert!(!exhausted_review_budget(&stalled));
9199 }
9200
9201 fn review_round(round: usize) -> crate::run::ReviewRound {
9202 crate::run::ReviewRound {
9203 round,
9204 head: "deadbeef".to_owned(),
9205 verified_head: None,
9206 verified_at: None,
9207 reviews: Vec::new(),
9208 e2e: Vec::new(),
9209 verify_retried: false,
9210 e2e_deferred: false,
9211 e2e_defer_reason: None,
9212 fix: None,
9213 blocking: 0,
9214 answered: 1,
9215 expected: 1,
9216 clean: false,
9217 progressed: true,
9218 vote_split: false,
9219 reconsideration: Vec::new(),
9220 verdict: None,
9221 }
9222 }
9223
9224 #[test]
9225 fn awaiting_resume_is_a_failed_task_whose_last_run_parked_and_can_be_resumed() {
9226 let mut run = RunState::new(
9227 PathBuf::from("/repo"),
9228 "main".to_owned(),
9229 "abc1234def".to_owned(),
9230 "add retries".to_owned(),
9231 Config::default(),
9232 );
9233 run.status = RunStatus::Judging;
9234 run.parked = true;
9235 let mut task = Task::new(
9236 "add retries".to_owned(),
9237 "add retries".to_owned(),
9238 PathBuf::from("/repo"),
9239 crate::queue::Source::Human,
9240 );
9241 task.status = TaskStatus::Failed;
9242 task.runs = vec![run.id.clone()];
9243 let with = |t: &Task, r: &RunState| awaiting_resume_with(t, |_| Ok(r.clone()));
9244 assert!(with(&task, &run), "parked after judging is the case");
9245 let mut parked_task = task.clone();
9246 parked_task.status = TaskStatus::Parked;
9247 assert!(with(&parked_task, &run), "the Parked status is the case");
9248
9249 let mut not_parked = run.clone();
9250 not_parked.parked = false;
9251 not_parked.status = RunStatus::Stalled;
9252 assert!(!with(&task, ¬_parked), "a stall is the conductor's");
9253
9254 let mut fresh = task.clone();
9255 fresh.fresh_start = true;
9256 assert!(!with(&fresh, &run), "a requeue asked for a new competition");
9257
9258 let mut review = task.clone();
9259 review.review_branch = Some("magi/x/A".to_owned());
9260 assert!(!with(&review, &run), "review is ranked before resume");
9261
9262 let mut held = task.clone();
9263 held.status = TaskStatus::Held;
9264 assert!(!with(&held, &run), "a hold stays visible to the conductor");
9265
9266 let mut released = run.clone();
9267 released.released_to = Some("20260901-000000-new1".to_owned());
9268 assert!(!with(&task, &released), "nothing left to resume into");
9269
9270 assert!(!awaiting_resume_with(&task, |_| anyhow::bail!(
9271 "unreadable"
9272 )));
9273 }
9274
9275 #[test]
9276 fn unfinished_run_never_offers_a_run_whose_worktree_was_released() {
9277 let mut released = RunState::new(
9278 PathBuf::from("/repo"),
9279 "main".to_owned(),
9280 "abc1234def".to_owned(),
9281 "add retries".to_owned(),
9282 Config::default(),
9283 );
9284 released.status = RunStatus::Blocked;
9285 assert_eq!(
9286 unfinished_run_with(&[released.id.clone()], "t", |_| Ok(released.clone())),
9287 Some(released.id.clone())
9288 );
9289 released.released_to = Some("20260901-000000-new1".to_owned());
9290 assert_eq!(
9291 unfinished_run_with(&[released.id.clone()], "t", |_| Ok(released.clone())),
9292 None,
9293 "there is nothing left to resume it into"
9294 );
9295 }
9296
9297 #[test]
9298 fn unfinished_run_skips_a_round_exhausted_blocked_run_so_requeue_means_a_fresh_competition() {
9299 let mut exhausted = RunState::new(
9308 PathBuf::from("/repo"),
9309 "main".to_owned(),
9310 "abc1234def".to_owned(),
9311 "add retries".to_owned(),
9312 Config::default(),
9313 );
9314 exhausted.status = RunStatus::Blocked;
9315 exhausted.config.graph.review_rounds = 1;
9316 exhausted.reviews = vec![review_round(1)];
9317
9318 assert_eq!(
9319 unfinished_run_with(&[exhausted.id.clone()], "t", |_| Ok(exhausted.clone())),
9320 None,
9321 "an exhausted `Blocked` run must not be offered as resumable"
9322 );
9323
9324 let mut has_budget_left = RunState::new(
9327 PathBuf::from("/repo"),
9328 "main".to_owned(),
9329 "abc1234def".to_owned(),
9330 "add retries".to_owned(),
9331 Config::default(),
9332 );
9333 has_budget_left.status = RunStatus::Blocked;
9334 has_budget_left.config.graph.review_rounds = 3;
9335 has_budget_left.reviews = vec![review_round(1)];
9336
9337 assert_eq!(
9338 unfinished_run_with(&[has_budget_left.id.clone()], "t", |_| {
9339 Ok(has_budget_left.clone())
9340 }),
9341 Some(has_budget_left.id.clone())
9342 );
9343 }
9344
9345 #[test]
9346 fn unfinished_run_resumes_a_run_whose_task_head_publication_is_pending() {
9347 let mut run = RunState::new(
9348 PathBuf::from("/repo"),
9349 "main".to_owned(),
9350 "abc1234def".to_owned(),
9351 "add retries".to_owned(),
9352 Config::default(),
9353 );
9354 run.status = RunStatus::Blocked;
9355 run.config.graph.review_rounds = 1;
9356 run.reviews = vec![review_round(1)];
9357 run.pr_publication_pending = true;
9358 assert_eq!(
9359 unfinished_run_with(&[run.id.clone()], "t", |_| Ok(run.clone())),
9360 Some(run.id.clone()),
9361 "a spent review budget must not turn a failed publication into a re-competition"
9362 );
9363 }
9364
9365 #[test]
9366 fn unfinished_run_never_falls_back_to_an_older_resumable_run() {
9367 let mut older_stalled = RunState::new(
9375 PathBuf::from("/repo"),
9376 "main".to_owned(),
9377 "abc1234def".to_owned(),
9378 "add retries".to_owned(),
9379 Config::default(),
9380 );
9381 older_stalled.status = RunStatus::Stalled;
9382
9383 let mut newest_exhausted = RunState::new(
9384 PathBuf::from("/repo"),
9385 "main".to_owned(),
9386 "abc1234def".to_owned(),
9387 "add retries".to_owned(),
9388 Config::default(),
9389 );
9390 newest_exhausted.status = RunStatus::Blocked;
9391 newest_exhausted.config.graph.review_rounds = 1;
9392 newest_exhausted.reviews = vec![review_round(1)];
9393
9394 assert_eq!(
9395 unfinished_run_with(
9396 &[older_stalled.id.clone(), newest_exhausted.id.clone()],
9397 "t",
9398 |_| Ok(newest_exhausted.clone())
9399 ),
9400 None,
9401 "the newest run is exhausted, so nothing here is worth resuming - \
9402 least of all the older, already-superseded run"
9403 );
9404 }
9405
9406 #[test]
9407 fn unfinished_run_warns_and_skips_a_run_it_cannot_read() {
9408 assert_eq!(
9409 unfinished_run_with(&["20260101-000000-gone".to_owned()], "t", |_| {
9410 Err(anyhow::anyhow!("fixture is absent"))
9411 }),
9412 None
9413 );
9414 }
9415
9416 fn action_question(run: &str, action: ask::ChoiceAction) -> ask::Question {
9417 let mut q = ask::Question::new(
9418 run.to_owned(),
9419 "implement".to_owned(),
9420 "impl-A".to_owned(),
9421 "continue?".to_owned(),
9422 String::new(),
9423 vec!["resume で続行する".to_owned(), "other".to_owned()],
9424 );
9425 q.actions.insert("resume で続行する".to_owned(), action);
9426 q.answer(ask::Answer::Choice("resume で続行する".to_owned()))
9427 .unwrap();
9428 q
9429 }
9430
9431 fn held_task_with(run: &str) -> Task {
9432 let mut t = task();
9433 t.runs = vec![run.to_owned()];
9434 t.hold_machine(Some("waiting for magi resume to be executed".to_owned()));
9435 t
9436 }
9437
9438 fn resume_action(run: &str) -> ask::ChoiceAction {
9439 ask::ChoiceAction::Resume { run: run.into() }
9440 }
9441
9442 #[test]
9443 fn explicitly_resumed_older_run_remains_the_release_and_question_target() {
9444 for parked in [false, true] {
9445 let mut t = task();
9446 t.start("A".into());
9447 t.start("B".into());
9448 assert_eq!(t.runs, vec!["A", "B"]);
9449 t.start("A".into());
9450 let mut resumed = run_state(RunStatus::Blocked);
9451 resumed.id = "A".into();
9452 resumed.parked = parked;
9453 resumed.driver_exited = true;
9454 t.hold_machine(Some("resumed run stopped".into()));
9455 let q = action_question("A", resume_action("A"));
9456 assert_eq!(action_standing(&t, &q), ActionStanding::Pending);
9457 assert_eq!(
9458 decide_action(&t, &q, &PHRASES_EN, |id| {
9459 assert_eq!(id, "A");
9460 Ok(resumed.clone())
9461 }),
9462 ActionDecision::Resume("A".into())
9463 );
9464 assert_eq!(
9465 action_standing(&t, &action_question("B", resume_action("B"))),
9466 ActionStanding::Stale
9467 );
9468 t.release_operator("cli", None);
9469 let unfinished = unfinished_run_with(&t.runs, t.short(), |id| {
9470 assert_eq!(id, "A");
9471 Ok(resumed.clone())
9472 });
9473 assert_eq!(
9474 choose_starter(None, false, unfinished.as_deref()),
9475 Starter::Resume("A".into())
9476 );
9477 assert_eq!(t.runs, vec!["B", "A"]);
9478 }
9479 }
9480
9481 #[test]
9482 fn merge_repair_dispatch_refuses_obsolete_bindings_and_unsupported_owners() {
9483 let mut state = crate::land::tests::repair_state(true, true);
9484 state.status = RunStatus::Landing;
9485 state.parked = true;
9486 state.driver_exited = true;
9487 let mut q = ask::Question::new(
9488 state.id.clone(),
9489 land::APPROVAL_NODE.into(),
9490 "land".into(),
9491 "Merge?".into(),
9492 String::new(),
9493 vec![land::APPROVE.into(), land::HOLD.into()],
9494 );
9495 q.id = state.id.clone();
9496 state.land_approval = Some(crate::run::LandApproval {
9497 question: q.id.clone(),
9498 head: "current-head".into(),
9499 });
9500 assert!(land::upgrade_repair_choice(&mut q, &state));
9501 q.answer(ask::Answer::Choice(land::REPAIR.into())).unwrap();
9502 let mut task = held_task_with(&state.id);
9503 task.status = TaskStatus::Parked;
9504 assert_eq!(
9505 decide_action(&task, &q, &PHRASES_EN, |_| Ok(state.clone())),
9506 ActionDecision::Resume(state.id.clone())
9507 );
9508 q.asked_at = "2026-01-01T00:00:00Z".parse().unwrap();
9509 q.answered_at = Some("2026-01-01T00:00:20Z".parse().unwrap());
9510 task.operator_decided_at = Some("2026-01-01T00:00:10Z".parse().unwrap());
9511 assert_eq!(
9512 decide_action(&task, &q, &PHRASES_EN, |_| Ok(state.clone())),
9513 ActionDecision::Resume(state.id.clone()),
9514 "a repair answer newer than the manual hold resumes the existing owner"
9515 );
9516 task.operator_decided_at = Some("2026-01-01T00:00:30Z".parse().unwrap());
9517 for answered_at in [q.answered_at, None] {
9518 let mut older = q.clone();
9519 older.answered_at = answered_at;
9520 let refused = decide_action(&task, &older, &PHRASES_EN, |_| {
9521 panic!("a later manual hold must prevent dispatch")
9522 });
9523 let ActionDecision::Refuse(reason) = refused else {
9524 panic!("expected an actionable refusal, got {refused:?}");
9525 };
9526 assert!(reason.contains("later operator decision"));
9527 assert!(reason.contains(&format!("magi task release {}", task.id)));
9528 }
9529 task.operator_decided_at = None;
9530 let mut superseded = state.clone();
9531 superseded.land_approval.as_mut().unwrap().question = "fresh-decision".into();
9532 assert_eq!(
9533 decide_action(&task, &q, &PHRASES_EN, |_| Ok(superseded)),
9534 ActionDecision::Stale
9535 );
9536 for case in 1..4 {
9537 let mut changed = state.clone();
9538 match case {
9539 1 => changed.status = RunStatus::Blocked,
9540 2 => changed.released_to = Some("new-owner".into()),
9541 _ => changed.land_approval.as_mut().unwrap().head.clear(),
9542 }
9543 assert!(matches!(
9544 decide_action(&task, &q, &PHRASES_EN, |_| Ok(changed)),
9545 ActionDecision::Refuse(_)
9546 ));
9547 }
9548 assert!(matches!(
9549 decide_action(&task, &q, &PHRASES_EN, |_| anyhow::bail!("missing owner")),
9550 ActionDecision::Refuse(_)
9551 ));
9552 task.runs.push("replacement".into());
9553 task.operator_decided_at = Some("2026-01-01T00:00:30Z".parse().unwrap());
9554 assert_eq!(
9555 decide_action(&task, &q, &PHRASES_EN, |_| panic!(
9556 "old owner must not load"
9557 )),
9558 ActionDecision::Stale
9559 );
9560 }
9561
9562 #[test]
9563 fn merge_repair_dispatch_obeys_capacity_and_quota_while_hold_stays_decision_only() {
9564 let mut state = crate::land::tests::repair_state(true, true);
9565 state.status = RunStatus::Landing;
9566 state.parked = true;
9567 state.driver_exited = true;
9568 let mut q = ask::Question::new(
9569 state.id.clone(),
9570 land::APPROVAL_NODE.into(),
9571 "land".into(),
9572 "Merge?".into(),
9573 String::new(),
9574 vec![land::APPROVE.into(), land::HOLD.into()],
9575 );
9576 q.id = state.id.clone();
9577 state.land_approval = Some(crate::run::LandApproval {
9578 question: q.id.clone(),
9579 head: "current-head".into(),
9580 });
9581 state.save().unwrap();
9582 land::upgrade_repair_choice(&mut q, &state);
9583 q.answer(ask::Answer::Choice(land::REPAIR.into())).unwrap();
9584 Questions::open().put(&mut q).unwrap();
9585 let tmp = tempfile::tempdir().unwrap();
9586 let queue = Queue::at(tmp.path().join("queue"));
9587 let mut task = held_task_with(&state.id);
9588 task.status = TaskStatus::Parked;
9589 queue.put(&mut task).unwrap();
9590 let empty = Arc::new(tokio::sync::Semaphore::new(0));
9591 let free = Arc::new(tokio::sync::Semaphore::new(1));
9592 assert!(claim_dispatch(&queue, &task, false, &empty, &free).is_none());
9593 assert!(claim_dispatch(&queue, &task, true, &free, &free).is_none());
9594 let claim = claim_dispatch(&queue, &task, false, &free, &free).unwrap();
9595 assert!(claim.permit.is_some());
9596 assert!(claim_dispatch(&queue, &task, false, &free, &free).is_none());
9597 drop(claim);
9598 Questions::open()
9599 .update(&q.id, |q| {
9600 q.answer = Some(ask::Answer::Choice(land::HOLD.into()));
9601 Ok(())
9602 })
9603 .unwrap();
9604 let claim = claim_dispatch(&queue, &task, true, &empty, &empty).unwrap();
9605 assert!(claim.permit.is_none());
9606 assert!(RunState::load(&state.id).unwrap().merge_repairs.is_empty());
9607 }
9608
9609 #[test]
9610 fn decide_action_resumes_only_the_latest_resumable_run() {
9611 let t = held_task_with("r1");
9612 let q = action_question("r1", resume_action("r1"));
9613 let load = |s: RunState| move |_: &str| Ok(s);
9614 assert_eq!(
9615 decide_action(&t, &q, &PHRASES_EN, load(run_state(RunStatus::Blocked))),
9616 ActionDecision::Resume("r1".into())
9617 );
9618 let q_other = action_question("r1", resume_action("r0"));
9620 assert!(matches!(
9621 decide_action(
9622 &t,
9623 &q_other,
9624 &PHRASES_EN,
9625 load(run_state(RunStatus::Blocked))
9626 ),
9627 ActionDecision::Refuse(_)
9628 ));
9629 assert!(matches!(
9631 decide_action(&t, &q, &PHRASES_EN, load(run_state(RunStatus::Ready))),
9632 ActionDecision::Refuse(_)
9633 ));
9634 let mut released = run_state(RunStatus::Blocked);
9636 released.released_to = Some("elsewhere".into());
9637 assert!(matches!(
9638 decide_action(&t, &q, &PHRASES_EN, load(released)),
9639 ActionDecision::Refuse(_)
9640 ));
9641 assert!(matches!(
9643 decide_action(&t, &q, &PHRASES_EN, |_: &str| bail!("gone")),
9644 ActionDecision::Refuse(_)
9645 ));
9646 }
9647
9648 #[test]
9649 fn decide_action_ignores_a_question_about_an_earlier_run() {
9650 let mut t = held_task_with("r1");
9651 t.runs.push("r2".to_owned());
9652 let never = |_: &str| -> Result<RunState> { bail!("not read") };
9653 assert_eq!(
9654 decide_action(
9655 &t,
9656 &action_question("r1", ask::ChoiceAction::Done),
9657 &PHRASES_EN,
9658 never
9659 ),
9660 ActionDecision::Stale
9661 );
9662 }
9663
9664 #[test]
9665 fn decide_action_maps_requeue_and_done_and_never_acts_twice() {
9666 let mut t = held_task_with("r1");
9667 let never = |_: &str| -> Result<RunState> { bail!("not read") };
9668 assert_eq!(
9669 decide_action(
9670 &t,
9671 &action_question("r1", ask::ChoiceAction::Requeue),
9672 &PHRASES_EN,
9673 never
9674 ),
9675 ActionDecision::Requeue
9676 );
9677 let done_q = action_question("r1", ask::ChoiceAction::Done);
9678 assert_eq!(
9679 decide_action(&t, &done_q, &PHRASES_EN, never),
9680 ActionDecision::Done
9681 );
9682 t.mark_action_applied(&done_q.id);
9683 assert_eq!(
9684 decide_action(&t, &done_q, &PHRASES_EN, never),
9685 ActionDecision::Skip
9686 );
9687
9688 let mut plain = action_question("r1", ask::ChoiceAction::Done);
9690 plain.actions.clear();
9691 assert_eq!(
9692 decide_action(&held_task_with("r1"), &plain, &PHRASES_EN, never),
9693 ActionDecision::Skip
9694 );
9695 let mut running = held_task_with("r1");
9697 running.status = TaskStatus::Running;
9698 assert_eq!(
9699 decide_action(
9700 &running,
9701 &action_question("r1", ask::ChoiceAction::Done),
9702 &PHRASES_EN,
9703 never
9704 ),
9705 ActionDecision::Skip
9706 );
9707 }
9708
9709 #[test]
9710 fn merge_repair_rejected_after_manual_hold_keeps_the_conductor_guard() {
9711 crate::run::pin_test_home();
9712 for legacy in [false, true] {
9713 let dir = tempfile::tempdir().unwrap();
9714 let queue = Queue::at(dir.path().join("queue"));
9715 let questions = Questions::at(dir.path().join("questions"));
9716 let run = "20260101-000000-hold";
9717 let mut task = held_task_with(run);
9718 task.status = TaskStatus::Parked;
9719 let mut q = ask::Question::new(
9720 run.into(),
9721 land::APPROVAL_NODE.into(),
9722 "land".into(),
9723 "Repair?".into(),
9724 String::new(),
9725 vec![land::REPAIR.into()],
9726 );
9727 q.actions.insert(land::REPAIR.into(), resume_action(run));
9728 q.answer(ask::Answer::Choice(land::REPAIR.into())).unwrap();
9729 q.asked_at = "2026-01-01T00:00:00Z".parse().unwrap();
9731 q.answered_at = Some("2026-01-01T00:00:01Z".parse().unwrap());
9732 questions.put(&mut q).unwrap();
9733 task.hold_manual(Some("Keep this PR paused".into()));
9734 if legacy {
9735 task.hold_source = None;
9736 }
9737 queue.put(&mut task).unwrap();
9738
9739 apply_choice_actions(&queue, &questions, dir.path());
9740 let after = queue.get(&task.id).unwrap();
9741 assert!(after.operator_held());
9742 assert_eq!(after.hold_source, task.hold_source);
9743 assert_eq!(after.operator_decided_at, task.operator_decided_at);
9744 assert_eq!(after.runs, task.runs);
9745 assert_eq!(after.instruction, task.instruction);
9746 assert!(after.resume_override.is_none());
9747 assert!(after.action_applied(&q.id));
9748 let reason = after.hold_reason.as_deref().unwrap();
9749 assert!(reason.starts_with("Keep this PR paused"));
9750 assert!(reason.contains("later operator decision"));
9751 assert!(reason.contains(&format!("magi task release {}", task.id)));
9752
9753 for recovery in [
9754 crate::conduct::Recovery::Review,
9755 crate::conduct::Recovery::Requeue,
9756 ] {
9757 crate::conduct::apply(
9758 &queue,
9759 &questions,
9760 &crate::conduct::Verdict {
9761 decisions: vec![crate::conduct::Decision {
9762 id: task.id.clone(),
9763 recovery: Some(recovery),
9764 ..Default::default()
9765 }],
9766 },
9767 )
9768 .unwrap();
9769 }
9770 apply_choice_actions(&queue, &questions, dir.path());
9773 let still_held = queue.get(&task.id).unwrap();
9774 assert!(still_held.operator_held());
9775 assert_eq!(still_held.hold_reason, after.hold_reason);
9776 assert_eq!(still_held.runs, task.runs);
9777 assert!(still_held.review_branch.is_none());
9778 assert!(!still_held.fresh_start);
9779 assert!(queue.next_runnable().is_none());
9780 }
9781 }
9782
9783 #[test]
9784 fn apply_choice_actions_releases_the_task_pinned_to_its_run_and_only_once() {
9785 let dir = tempfile::tempdir().unwrap();
9786 let queue = Queue::at(dir.path().join("queue"));
9787 let questions = Questions::at(dir.path().join("questions"));
9788 let home = dir.path().join("home");
9789 let mut state = run_state(RunStatus::Blocked);
9790 state.id = "20260101-000000-act1".to_owned();
9791 state.save_under(&home).unwrap();
9792
9793 let mut t = held_task_with(&state.id);
9794 queue.put(&mut t).unwrap();
9795 let mut q = action_question(&state.id, resume_action(&state.id));
9796 questions.put(&mut q).unwrap();
9797
9798 apply_choice_actions(&queue, &questions, &home);
9799 let after = queue.get(&t.id).unwrap();
9800 assert_eq!(after.status, TaskStatus::Queued);
9801 assert!(!after.fresh_start);
9802 assert!(after.action_applied(&q.id));
9803 let pin = after.resume_override.clone().unwrap();
9804 assert_eq!(pin.pinned_run.as_deref(), Some(state.id.as_str()));
9805 assert!(pin.forced);
9806
9807 let mut again = queue.get(&t.id).unwrap();
9809 again.hold_machine(Some("later".into()));
9810 queue.put(&mut again).unwrap();
9811 apply_choice_actions(&queue, &questions, &home);
9812 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9813 }
9814
9815 #[test]
9816 fn a_delivered_answer_is_still_acted_on_exactly_once() {
9817 let dir = tempfile::tempdir().unwrap();
9818 let queue = Queue::at(dir.path().join("queue"));
9819 let questions = Questions::at(dir.path().join("questions"));
9820 let home = dir.path().join("home");
9821 let mut state = run_state(RunStatus::Blocked);
9822 state.id = "20260101-000000-act2".to_owned();
9823 state.save_under(&home).unwrap();
9824
9825 let mut t = held_task_with(&state.id);
9826 queue.put(&mut t).unwrap();
9827 let mut q = action_question(&state.id, resume_action(&state.id));
9828 q.answer_delivered = true;
9829 questions.put(&mut q).unwrap();
9830
9831 apply_choice_actions(&queue, &questions, &home);
9834 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
9835 assert!(queue.get(&t.id).unwrap().action_applied(&q.id));
9836
9837 let mut again = queue.get(&t.id).unwrap();
9839 again.hold_machine(Some("later".into()));
9840 queue.put(&mut again).unwrap();
9841 apply_choice_actions(&queue, &questions, &home);
9842 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9843 }
9844
9845 #[test]
9846 fn a_fresh_asker_defers_the_action_until_it_goes_quiet() {
9847 let dir = tempfile::tempdir().unwrap();
9848 let queue = Queue::at(dir.path().join("queue"));
9849 let questions = Questions::at(dir.path().join("questions"));
9850 let home = dir.path().join("home");
9851 let mut state = run_state(RunStatus::Blocked);
9852 state.id = "20260101-000000-act3".to_owned();
9853 state.save_under(&home).unwrap();
9854 let mut t = held_task_with(&state.id);
9855 queue.put(&mut t).unwrap();
9856 let mut q = action_question(&state.id, ask::ChoiceAction::Requeue);
9857 questions.put(&mut q).unwrap();
9858
9859 questions.beat(&q.id, crate::ask::WaiterKind::Asker);
9860 apply_choice_actions(&queue, &questions, &home);
9861 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9862
9863 std::fs::remove_file(questions.root().join(format!("{}.lease", q.id))).unwrap();
9864 apply_choice_actions(&queue, &questions, &home);
9865 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Queued);
9866 }
9867
9868 #[test]
9869 fn action_standing_tells_the_reasons_apart() {
9870 let mut t = held_task_with("r1");
9871 let q = action_question("r1", ask::ChoiceAction::Requeue);
9872 assert_eq!(action_standing(&t, &q), ActionStanding::Pending);
9873 let mut plain = q.clone();
9874 plain.actions.clear();
9875 assert_eq!(action_standing(&t, &plain), ActionStanding::NoAction);
9876
9877 t.mark_action_applied(&q.id);
9879 assert_eq!(action_standing(&t, &q), ActionStanding::Applied);
9880
9881 let mut t = held_task_with("r1");
9883 t.start("r2".to_owned());
9884 assert_eq!(action_standing(&t, &q), ActionStanding::Stale);
9885 let q2 = action_question("r2", ask::ChoiceAction::Requeue);
9887 assert_eq!(action_standing(&t, &q2), ActionStanding::Busy);
9888 t.status = TaskStatus::Blocked;
9889 assert_eq!(action_standing(&t, &q2), ActionStanding::Busy);
9890
9891 let mut c = action_question("r0", ask::ChoiceAction::Requeue);
9893 c.node = crate::conduct::NODE.to_owned();
9894 assert_eq!(action_standing(&t, &c), ActionStanding::Busy);
9895 assert!(!ActionStanding::Busy.daemon_owns());
9896 }
9897
9898 #[test]
9899 fn a_running_task_waits_and_a_dead_asker_with_a_cwd_is_still_actioned() {
9900 let dir = tempfile::tempdir().unwrap();
9901 let queue = Queue::at(dir.path().join("queue"));
9902 let questions = Questions::at(dir.path().join("questions"));
9903 let home = dir.path().join("home");
9904 let mut state = run_state(RunStatus::Blocked);
9905 state.id = "20260101-000000-act4".to_owned();
9906 state.save_under(&home).unwrap();
9907 let mut t = held_task_with(&state.id);
9908 t.status = TaskStatus::Running;
9909 queue.put(&mut t).unwrap();
9910 let mut q = action_question(&state.id, ask::ChoiceAction::Done);
9911 q.cwd = Some(dir.path().display().to_string());
9912 questions.put(&mut q).unwrap();
9913
9914 let running = queue.get(&t.id).unwrap();
9917 assert_eq!(action_standing(&running, &q), ActionStanding::Busy);
9918 assert!(matches!(
9919 crate::waiter::decide_owned(&q, None, false, 86_400, Timestamp::now(), false),
9920 crate::waiter::Action::Deliver(_)
9921 ));
9922 apply_choice_actions(&queue, &questions, &home);
9924 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Running);
9925 let mut back = queue.get(&t.id).unwrap();
9926 back.hold_machine(Some("later".into()));
9927 queue.put(&mut back).unwrap();
9928 apply_choice_actions(&queue, &questions, &home);
9929 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Done);
9930 let mut again = queue.get(&t.id).unwrap();
9931 again.hold_machine(Some("again".into()));
9932 queue.put(&mut again).unwrap();
9933 apply_choice_actions(&queue, &questions, &home);
9934 assert_eq!(queue.get(&t.id).unwrap().status, TaskStatus::Held);
9935 }
9936}