1use std::path::{Path, PathBuf};
84use std::time::Duration;
85
86use jiff::Timestamp;
87
88use crate::ask::{Question, QuestionStatus, Questions};
89use crate::config::Config;
90use crate::queue::{HoldSource, OperatorResume, Queue, Task, TaskStatus};
91
92pub const NODE: &str = "triage";
95
96pub const DEPS_NODE: &str = "triage-deps";
100
101const SEAT: &str = "triage";
106
107const MANUAL_STALE_AFTER: Duration = Duration::from_secs(7 * 24 * 60 * 60);
119
120struct Wording {
124 lang: &'static str,
125 resume: &'static str,
126 wait: &'static str,
127 discard: &'static str,
128 resume_now: &'static str,
129 keep_held: &'static str,
130}
131
132const EN: Wording = Wording {
133 lang: "en",
134 resume: "resume it",
135 wait: "not yet",
136 discard: "discard it",
137 resume_now: "resume it",
138 keep_held: "keep it held",
139};
140
141const JA: Wording = Wording {
142 lang: "ja",
143 resume: "再開してよい",
144 wait: "まだ待って",
145 discard: "捨ててよい",
146 resume_now: "再開する",
147 keep_held: "まだ止めておく",
148};
149
150fn wording(language: &str) -> &'static Wording {
153 if crate::lang::is_japanese(language) {
154 &JA
155 } else {
156 &EN
157 }
158}
159
160impl Wording {
161 fn choices3(&self) -> Vec<String> {
162 vec![
163 self.resume.to_owned(),
164 self.wait.to_owned(),
165 self.discard.to_owned(),
166 ]
167 }
168
169 fn choices2(&self) -> Vec<String> {
170 vec![self.resume_now.to_owned(), self.keep_held.to_owned()]
171 }
172
173 fn source_label(&self, source: Option<HoldSource>) -> &'static str {
174 match (self.lang, source) {
175 ("ja", Some(HoldSource::Machine)) => "machine(機械による自動保留)",
176 ("ja", Some(HoldSource::Manual)) => "manual(操作者による手動保留)",
177 ("ja", None) => "unknown(schema 3 未満の旧レコード、または理由未記録)",
178 (_, Some(HoldSource::Machine)) => "machine (automatic recovery hold)",
179 (_, Some(HoldSource::Manual)) => "manual (an operator held this)",
180 (_, None) => "unknown (pre-schema-3 record, or never recorded)",
181 }
182 }
183
184 fn detail(&self, task: &Task, why: &str) -> String {
193 let none = if self.lang == "ja" {
194 "(記録なし)"
195 } else {
196 "(none recorded)"
197 };
198 let reason = task
199 .hold_reason
200 .as_deref()
201 .or(task.last_error.as_deref())
202 .unwrap_or(none);
203 format!(
204 "task: {} ({})\ntitle: {}\nhold source: {}\nhold reason: {reason}\n\n{why}",
205 task.id,
206 task.short(),
207 task.title,
208 self.source_label(task.hold_source),
209 )
210 }
211
212 fn summary_machine_unknown(&self, task: &Task) -> String {
213 if self.lang == "ja" {
214 format!("保留タスク {} の再開可否を判断してください", task.short())
215 } else {
216 format!("decide whether to resume held task {}", task.short())
217 }
218 }
219
220 fn why_machine(&self) -> &'static str {
221 if self.lang == "ja" {
222 "機械的な保留(machine hold)ですが、原因がすでに解消しているかを自動では判断できませんでした。"
223 } else {
224 "This is a machine hold, but whether its cause has resolved could not be \
225 checked automatically."
226 }
227 }
228
229 fn summary_conductor_override(&self, task: &Task) -> String {
230 if self.lang == "ja" {
231 format!(
232 "再開と回答済みのタスク {} を conductor が再び保留しました",
233 task.short()
234 )
235 } else {
236 format!(
237 "task {} was resumed at your word, but the conductor held it again",
238 task.short()
239 )
240 }
241 }
242
243 fn why_conductor_override(&self, o: &OperatorResume) -> String {
244 let reason = o.conductor_rehold.as_deref().unwrap_or_default();
245 if self.lang == "ja" {
246 format!(
247 "{} に再開と回答済みですが、conductor が再び hold しました。conductor の理由: \
248 {reason}\n\n強制再キューを選ぶと、以後 conductor はこのタスクを hold できません。",
249 o.at
250 )
251 } else {
252 format!(
253 "You answered \"resume\" at {}, but the conductor held the task again. \
254 Its reason: {reason}\n\nForcing a requeue stops the conductor from \
255 holding this task again.",
256 o.at
257 )
258 }
259 }
260
261 fn choices_conductor_override(&self) -> Vec<String> {
263 if self.lang == "ja" {
264 vec![
265 "強制再キュー(conductor は再 hold 不可)".to_owned(),
266 "手動 hold のまま".to_owned(),
267 "捨ててよい".to_owned(),
268 ]
269 } else {
270 vec![
271 "force requeue (conductor must not hold again)".to_owned(),
272 "keep held (manual)".to_owned(),
273 "discard".to_owned(),
274 ]
275 }
276 }
277
278 fn summary_legacy(&self, task: &Task) -> String {
279 if self.lang == "ja" {
280 format!(
281 "hold_source が不明な保留タスク {} を確認してください",
282 task.short()
283 )
284 } else {
285 format!(
286 "held task {} has no recorded hold source - please take a look",
287 task.short()
288 )
289 }
290 }
291
292 fn why_legacy(&self) -> &'static str {
293 if self.lang == "ja" {
294 "hold_source が記録されていません。schema 3 より前のレコードか、理由が記録されなかった \
295 holdです。人が意図して止めたのか、クラッシュや強制再起動で宙に浮いただけなのか、\
296 このデータからは区別できません。"
297 } else {
298 "No hold_source was recorded - either a pre-schema-3 record, or a hold whose \
299 reason was never written down. Whether this was a deliberate hold or the \
300 leftover of a crash cannot be told from the data alone."
301 }
302 }
303
304 fn summary_manual_stale(&self, task: &Task, days: i64) -> String {
305 if self.lang == "ja" {
306 format!(
307 "{days}日間 保留されたままの手動保留タスク {} を確認してください",
308 task.short()
309 )
310 } else {
311 format!(
312 "held task {} has been on a manual hold for {days} day(s)",
313 task.short()
314 )
315 }
316 }
317
318 fn why_manual(&self) -> &'static str {
319 if self.lang == "ja" {
320 "操作者が明示的に止めた保留ですが、長期間そのままになっています。まだ止めておくか、\
321 再開するか教えてください。"
322 } else {
323 "An operator held this on purpose, but it has sat untouched for a while. Say \
324 whether to keep holding it or resume it."
325 }
326 }
327}
328
329#[derive(Debug, Clone, Copy, PartialEq, Eq)]
331enum Bucket {
332 MachineUnknown,
334 ConductorOverride,
337 Legacy,
339 ManualStale,
341}
342
343#[derive(Debug, Clone, Default)]
345pub struct Report {
346 pub resumed: Vec<String>,
349 pub asked: Vec<String>,
351 pub answered: Vec<String>,
353 pub quarantined: Vec<String>,
358}
359
360impl Report {
361 pub fn is_empty(&self) -> bool {
364 self.resumed.is_empty()
365 && self.asked.is_empty()
366 && self.answered.is_empty()
367 && self.quarantined.is_empty()
368 }
369}
370
371fn repo_for(task: &Task) -> PathBuf {
375 if task.repo.as_os_str().is_empty() {
376 PathBuf::from(".")
377 } else {
378 task.repo.clone()
379 }
380}
381
382fn is_disk_hold(task: &Task) -> bool {
393 task.hold_reason
394 .as_deref()
395 .is_some_and(crate::disk::is_gate_reason)
396}
397
398fn machine_cause_resolved(task: &Task, cfg: &Config) -> Option<bool> {
403 machine_cause_resolved_under(task, cfg, &crate::run::home())
404}
405
406fn machine_cause_resolved_under(task: &Task, cfg: &Config, home: &Path) -> Option<bool> {
407 if task
408 .hold_reason
409 .as_deref()
410 .is_some_and(|r| r.starts_with(crate::cache::RESOURCE_HOLD))
411 {
412 let id = task.runs.last()?;
416 let state = crate::run::RunState::load_under(id, home).ok()?;
417 let block = state.resource_block.as_ref()?;
418 if state.id != *id
419 || block.kind != "shared build cache"
420 || !matches!(block.cause, crate::run::ResourceBlockCause::Timeout { .. })
421 {
422 return None;
423 }
424 return Some(
425 cfg.cache_dir()
426 .is_none_or(|dir| !crate::cache::in_use(home, &dir)),
427 );
428 }
429 if !is_disk_hold(task) {
430 return None;
431 }
432 let min = cfg.disk.min_free_bytes;
433 if min == 0 {
434 return Some(true);
437 }
438 Some(crate::daemon::disk_gate(&repo_for(task), cfg).is_none())
439}
440
441fn manual_is_stale(task: &Task, now: Timestamp) -> bool {
445 now.as_second() - task.updated_at.as_second() > MANUAL_STALE_AFTER.as_secs() as i64
446}
447
448fn marker_for(q: &Question) -> String {
451 format!("[triage:{}]", q.short())
452}
453
454fn already_applied(task: &Task, q: &Question) -> bool {
457 if task.triage_applied(&q.id) {
458 return true;
459 }
460 let marker = marker_for(q);
463 task.hold_reason
464 .as_deref()
465 .is_some_and(|r| r.contains(marker.as_str()))
466}
467
468fn latest_triage_question(questions: &Questions, task_id: &str) -> Option<Question> {
473 latest_question(questions, NODE, task_id)
474}
475
476fn latest_question(questions: &Questions, node: &str, task_id: &str) -> Option<Question> {
478 questions
479 .list()
480 .into_iter()
481 .filter(|q| q.node == node && q.run == task_id)
482 .max_by(|a, b| a.asked_at.cmp(&b.asked_at).then_with(|| a.id.cmp(&b.id)))
485}
486
487#[derive(Debug, Clone, Copy, PartialEq, Eq)]
490enum AnswerAction {
491 Resume,
495 Discard,
498 KeepHeld,
503}
504
505fn interpret_answer(q: &Question) -> AnswerAction {
518 let resolution = q.resolution().unwrap_or_default();
519 match q.choices.iter().position(|c| *c == resolution) {
520 Some(0) => AnswerAction::Resume,
521 Some(2) => AnswerAction::Discard,
522 _ => AnswerAction::KeepHeld,
523 }
524}
525
526pub(crate) fn contextual_keep_held(q: &Question, label: &str) -> bool {
529 q.node == NODE
530 && q.seat == SEAT
531 && q.actions.is_empty()
532 && q.choices.get(1).is_some_and(|c| c == label)
533}
534
535fn keep_held_note(task: &Task, q: &Question, resolution: &str) -> String {
540 let marker = format!("{} operator: {resolution}", marker_for(q));
541 match task.hold_reason.as_deref() {
542 Some(existing) if !existing.is_empty() => format!("{existing}\n{marker}"),
543 _ => marker,
544 }
545}
546
547fn file_question(
551 questions: &Questions,
552 task: &Task,
553 bucket: Bucket,
554 w: &Wording,
555 now: Timestamp,
556) -> Option<Question> {
557 let (summary, why, choices) = match bucket {
558 Bucket::MachineUnknown => (
559 w.summary_machine_unknown(task),
560 w.why_machine().to_owned(),
561 w.choices3(),
562 ),
563 Bucket::ConductorOverride => (
564 w.summary_conductor_override(task),
565 task.resume_override
566 .as_ref()
567 .map(|o| w.why_conductor_override(o))
568 .unwrap_or_default(),
569 w.choices_conductor_override(),
570 ),
571 Bucket::Legacy => (
572 w.summary_legacy(task),
573 w.why_legacy().to_owned(),
574 w.choices3(),
575 ),
576 Bucket::ManualStale => {
577 let days = (now.as_second() - task.updated_at.as_second()) / (24 * 60 * 60);
578 (
579 w.summary_manual_stale(task, days),
580 w.why_manual().to_owned(),
581 w.choices2(),
582 )
583 }
584 };
585 let mut q = Question::new(
586 task.id.clone(),
587 NODE.to_owned(),
588 SEAT.to_owned(),
589 summary,
590 w.detail(task, &why),
591 choices,
592 );
593 q.triage_generation = Some(task.resumption.as_ref().map_or(0, |r| r.generation));
594 questions.put_executable(&mut q).ok()?;
595 Some(q)
596}
597
598pub(crate) fn deputy_brief(q: &Question, queue: &Queue) -> String {
603 let task = queue.get(&q.run).ok();
604 let mut s = match &task {
605 Some(t) => format!(
606 "Task {} (`magi task show {}`), \"{}\", is {}. Hold source: {}. Reason: {}.",
607 t.id,
608 t.id,
609 t.title,
610 t.status.as_str(),
611 match t.hold_source {
612 Some(HoldSource::Machine) => "machine (an automatic hold)",
613 Some(HoldSource::Manual) => "manual (an operator held it)",
614 None => "unknown (an old record, treated as the operator's)",
615 },
616 t.hold_reason
617 .as_deref()
618 .or(t.last_error.as_deref())
619 .unwrap_or("(none recorded)")
620 ),
621 None => format!("Task {} no longer exists (its record is gone).", q.run),
622 };
623 s.push_str("\n\nWhat each option does, by position:");
624 let n = q.choices.len();
625 for (i, c) in q.choices.iter().enumerate() {
626 let effect = if q.node == DEPS_NODE {
627 match i {
628 0 => "the dependency is released so the loop runs it",
629 1 => {
630 "the dependants are detached and the dependency task is DELETED; this cannot be undone"
631 }
632 2 => "the dependants are detached; the dependency stays as it is",
633 _ => "nothing is applied",
634 }
635 } else {
636 match i {
637 0 => "the task is released and requeued",
638 2 if n > 2 => "the task is DELETED; this cannot be undone",
639 _ => "the task stays held, with the answer noted",
640 }
641 };
642 s.push_str(&format!("\n- `{c}`: {effect}"));
643 }
644 s.push_str(
645 "\n\nThe daemon applies the answer on its next pass; you apply nothing. When \
646 the owner's words pick an option - including releasing a task they held \
647 themselves, or deleting it - settle it.",
648 );
649 s
650}
651
652fn quarantine_orphaned_blocked(queue: &Queue, questions: &Questions) -> Vec<String> {
670 let mut quarantined = Vec::new();
671 for listed in queue.list() {
672 if listed.status != TaskStatus::Blocked || listed.blocked_by.is_empty() {
673 continue;
674 }
675 let Ok(_claim) = queue.claim(&listed.id) else {
676 continue;
677 };
678 let Ok(mut task) = queue.get(&listed.id) else {
679 continue;
680 };
681 if task.status != TaskStatus::Blocked {
682 continue;
683 }
684 let deleted = queue.apply_deleted_blockers(&mut task, questions);
685 if !deleted.is_empty() && queue.put(&mut task).is_ok() {
686 for id in &deleted {
687 queue.note_dependency_deleted(&task, id);
688 }
689 }
690 if task.status != TaskStatus::Blocked {
691 continue;
692 }
693 let missing = crate::queue::missing_blockers(queue, questions, &task.blocked_by);
694 if missing.is_empty() {
695 continue;
696 }
697 let language = crate::lang::of_repo(&repo_for(&task));
698 let mut cause = missing.clone();
699 cause.sort();
700 cause.dedup();
701 task.notice_cause(&format!("missing-blockers:{}", cause.join(",")), None, None);
702 task.hold_machine(Some(crate::queue::missing_blocker_hold_reason_in(
703 &task.blocked_by,
704 &missing,
705 &language,
706 )));
707 if queue.put(&mut task).is_ok() {
708 quarantined.push(task.id.clone());
709 }
710 }
711 quarantined
712}
713
714#[derive(Debug, Clone, Copy, PartialEq, Eq)]
717enum DepsAction {
718 Release,
719 Discard,
720 Detach,
721 Nothing,
723}
724
725fn interpret_deps_answer(q: &Question) -> DepsAction {
726 let resolution = q.resolution().unwrap_or_default();
727 match q.choices.iter().position(|c| *c == resolution) {
728 Some(0) => DepsAction::Release,
729 Some(1) => DepsAction::Discard,
730 Some(2) => DepsAction::Detach,
731 _ => DepsAction::Nothing,
732 }
733}
734
735fn deps_texts(w: &Wording, root: &Task, dependants: &[&Task]) -> (String, String, Vec<String>) {
737 let ja = w.lang == "ja";
738 let reason = root
739 .hold_reason
740 .as_deref()
741 .or(root.last_error.as_deref())
742 .unwrap_or(if ja {
743 "(記録なし)"
744 } else {
745 "(none recorded)"
746 });
747 let list: String = dependants
748 .iter()
749 .map(|t| format!("- {} {}\n", t.short(), t.title))
750 .collect();
751 if ja {
752 (
753 format!(
754 "{} ({}) が {} 件のタスクを止めています",
755 root.short(),
756 root.status.as_str(),
757 dependants.len()
758 ),
759 format!(
760 "task: {} ({})\ntitle: {}\n状態: {}\n理由: {reason}\n\n\
761 このタスクは自動では実行されないため、依存している次のタスクは永遠に待ち続けます:\n{list}",
762 root.id,
763 root.short(),
764 root.title,
765 root.status.as_str()
766 ),
767 vec![
768 "依存先を再開する".to_owned(),
769 "依存先を捨てる(依存タスクは切り離して実行)".to_owned(),
770 "依存タスクを切り離す(依存先はそのまま)".to_owned(),
771 ],
772 )
773 } else {
774 (
775 format!(
776 "{} ({}) is freezing {} blocked task(s)",
777 root.short(),
778 root.status.as_str(),
779 dependants.len()
780 ),
781 format!(
782 "task: {} ({})\ntitle: {}\nstatus: {}\nreason: {reason}\n\n\
783 Nothing in the loop will ever run this task, so these dependants wait \
784 forever:\n{list}",
785 root.id,
786 root.short(),
787 root.title,
788 root.status.as_str()
789 ),
790 vec![
791 "release the dependency".to_owned(),
792 "discard the dependency (dependants are detached and run)".to_owned(),
793 "detach the dependants (the dependency stays as it is)".to_owned(),
794 ],
795 )
796 }
797}
798
799fn detach_dependants(queue: &Queue, questions: &Questions, root: &str) {
803 for listed in queue.list() {
804 if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| matches!(crate::queue::dependency(queue, questions, b), crate::queue::Dependency::Task(id) if id == root)) {
805 continue;
806 }
807 let Ok(_claim) = queue.claim(&listed.id) else {
808 continue;
809 };
810 let Ok(mut t) = queue.get(&listed.id) else {
811 continue;
812 };
813 if t.status != TaskStatus::Blocked {
814 continue;
815 }
816 for id in t.blocked_by.clone() {
817 if matches!(crate::queue::dependency(queue, questions, &id), crate::queue::Dependency::Task(full) if full == root)
818 {
819 t.unblock(&id);
820 }
821 }
822 let _ = queue.put(&mut t);
823 }
824}
825
826fn apply_deps_answer(queue: &Queue, questions: &Questions, q: &Question, now: Timestamp) -> bool {
830 let Ok(_claim) = queue.claim(&q.run) else {
831 return false;
832 };
833 let Ok(mut root) = queue.get(&q.run) else {
834 return false;
835 };
836 if root.triage_applied(&q.id) {
837 return false;
838 }
839 if root.operator_decided_at.is_some_and(|at| at > q.asked_at)
840 || q.triage_generation
841 .is_some_and(|g| g != root.resumption.as_ref().map_or(0, |r| r.generation))
842 {
843 root.mark_triage_applied(&q.id);
844 return queue.put(&mut root).is_ok();
845 }
846 match interpret_deps_answer(q) {
847 DepsAction::Release => {
848 if root.status == TaskStatus::Running {
849 return false;
850 }
851 root.answers.push(crate::queue::AnsweredQuestion {
852 question: q.summary.clone(),
853 answer: q.resolution().unwrap_or_default(),
854 });
855 root.release_operator("triage-dependencies", Some(&q.id));
856 root.resume_override = Some(OperatorResume {
857 question_id: q.id.clone(),
858 at: now,
859 conductor_rehold: None,
860 forced: false,
861 pinned_run: None,
862 });
863 }
864 DepsAction::Discard => {
865 detach_dependants(queue, questions, &root.id);
866 return queue.remove(&root.id, false, questions).is_ok();
867 }
868 DepsAction::Detach => detach_dependants(queue, questions, &root.id),
869 DepsAction::Nothing => return false,
874 }
875 root.mark_triage_applied(&q.id);
876 queue.put(&mut root).is_ok()
877}
878
879fn ask_about_stuck_roots(
888 queue: &Queue,
889 questions: &Questions,
890 config_override: Option<&Path>,
891 report: &mut Report,
892) {
893 let all = questions.list();
894 let inv = crate::blockers::Inventory::load(queue, questions);
895 let mut frozen: std::collections::BTreeMap<String, Vec<String>> = Default::default();
896 for (id, roots) in inv.stuck() {
897 for root in roots {
898 if root != id {
899 frozen.entry(root).or_default().push(id.clone());
900 }
901 }
902 }
903 for mut q in all
907 .iter()
908 .filter(|q| q.node == DEPS_NODE && q.status.open())
909 .cloned()
910 {
911 if !frozen.contains_key(&q.run) {
912 q.abandon("nothing waits on this task anymore");
913 let _ = questions.put_executable(&mut q);
914 }
915 }
916
917 for (root_id, dependant_ids) in &frozen {
918 let Ok(_claim) = queue.claim(root_id) else {
919 continue;
920 };
921 let current = crate::blockers::Inventory::load(queue, questions);
924 let Some(root) = current.task(root_id) else {
925 continue;
926 };
927 let dependants: Vec<&Task> = dependant_ids
928 .iter()
929 .filter_map(|d| current.task(d))
930 .filter(|t| current.stuck_roots(t).contains(root_id))
931 .collect();
932 if dependants.is_empty() {
933 continue;
934 }
935 let latest = all
936 .iter()
937 .filter(|q| q.node == DEPS_NODE && q.run == *root_id)
938 .max_by(|a, b| a.asked_at.cmp(&b.asked_at).then_with(|| a.id.cmp(&b.id)));
939 match latest {
940 Some(q) if q.status.open() || awaiting_task_chat(q, root) => continue,
941 Some(q) if q.status == QuestionStatus::Answered && !root.triage_applied(&q.id) => {
942 continue;
943 }
944 Some(q) if q.status == QuestionStatus::Abandoned => {
945 let changed = root.updated_at > q.asked_at
946 || dependants.iter().any(|t| t.updated_at > q.asked_at);
947 if !changed {
948 continue;
949 }
950 }
951 _ => {}
952 }
953 if pending_for(questions, root) {
954 continue;
955 }
956 let cfg = Config::discover(&repo_for(root), config_override)
957 .ok()
958 .map(|(c, _)| c);
959 let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
960 let (summary, detail, choices) = deps_texts(w, root, &dependants);
961 let mut question = Question::new(
962 root.id.clone(),
963 DEPS_NODE.to_owned(),
964 SEAT.to_owned(),
965 summary,
966 detail,
967 choices,
968 );
969 question.triage_generation = Some(root.resumption.as_ref().map_or(0, |r| r.generation));
970 if questions.put_executable(&mut question).is_ok() {
971 report.asked.push(root.id.clone());
972 }
973 }
974}
975
976pub fn apply_answers(queue: &Queue, questions: &Questions) -> Vec<String> {
979 let mut applied = Vec::new();
980 for q in questions.list() {
981 if q.node == DEPS_NODE
982 && q.status == QuestionStatus::Answered
983 && apply_deps_answer(queue, questions, &q, Timestamp::now())
984 {
985 applied.push(q.run.clone());
986 }
987 }
988 for listed in queue.list() {
989 let Some(q) = latest_triage_question(questions, &listed.id) else {
990 continue;
991 };
992 if q.status != QuestionStatus::Answered || already_applied(&listed, &q) {
993 continue;
994 }
995 let Ok(_claim) = queue.claim(&listed.id) else {
996 continue;
997 };
998 let Ok(mut task) = queue.get(&listed.id) else {
999 continue;
1000 };
1001 if already_applied(&task, &q)
1002 || !matches!(task.status, TaskStatus::Held | TaskStatus::Queued)
1003 {
1004 continue;
1005 }
1006 let generation = task.resumption.as_ref().map_or(0, |r| r.generation);
1007 let action = interpret_answer(&q);
1008 let mut newer_decision = q.triage_generation.is_some_and(|g| g != generation)
1009 || task.resumption.as_ref().is_some_and(|r| r.at > q.asked_at)
1010 || task.held_at.is_some_and(|at| at > q.asked_at)
1011 || task.operator_decided_at.is_some_and(|at| at > q.asked_at);
1012 if matches!(action, AnswerAction::KeepHeld)
1016 && q.answered_at.is_some_and(|at| {
1017 task.operator_decided_at
1018 .is_none_or(|decision| at > decision)
1019 && task.resumption.as_ref().is_none_or(|r| at > r.at)
1020 })
1021 {
1022 newer_decision = false;
1023 }
1024 let resolution = q.resolution().unwrap_or_default();
1026 task.answers.push(crate::queue::AnsweredQuestion {
1027 question: q.summary.clone(),
1028 answer: if newer_decision {
1029 format!("{resolution} (superseded by a later operator decision)")
1030 } else {
1031 resolution
1032 },
1033 });
1034 if !newer_decision {
1035 match action {
1036 AnswerAction::Resume => {
1037 task.release_operator("triage", Some(&q.id));
1038 task.resume_override = Some(OperatorResume {
1039 question_id: q.id.clone(),
1040 at: Timestamp::now(),
1041 conductor_rehold: None,
1042 forced: true,
1043 pinned_run: None,
1044 });
1045 }
1046 AnswerAction::KeepHeld => {
1047 let note = keep_held_note(&task, &q, &q.resolution().unwrap_or_default());
1048 task.hold_manual(Some(note));
1049 }
1050 AnswerAction::Discard => {
1051 if queue.remove(&task.id, false, questions).is_ok() {
1052 applied.push(task.id);
1053 }
1054 continue;
1055 }
1056 }
1057 }
1058 task.mark_triage_applied(&q.id);
1059 if queue.put(&mut task).is_ok() {
1060 applied.push(task.id);
1061 }
1062 }
1063 applied
1064}
1065
1066pub fn run_once(
1084 queue: &Queue,
1085 questions: &Questions,
1086 config_override: Option<&Path>,
1087 now: Timestamp,
1088) -> Report {
1089 let mut report = Report {
1090 answered: apply_answers(queue, questions),
1091 quarantined: quarantine_orphaned_blocked(queue, questions),
1092 ..Report::default()
1093 };
1094 ask_about_stuck_roots(queue, questions, config_override, &mut report);
1095 for listed in queue.list() {
1096 if listed.status != TaskStatus::Held {
1097 continue;
1098 }
1099 let Ok(_claim) = queue.claim(&listed.id) else {
1100 continue;
1101 };
1102 let Ok(mut task) = queue.get(&listed.id) else {
1103 continue;
1104 };
1105 if task.status != TaskStatus::Held {
1109 continue;
1110 }
1111 if deps_pending(questions, &task) {
1115 continue;
1116 }
1117
1118 let cfg = Config::discover(&repo_for(&task), config_override)
1119 .ok()
1120 .map(|(c, _)| c);
1121 let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
1122
1123 if let Some(q) = latest_triage_question(questions, &task.id) {
1124 if q.status.open() || awaiting_task_chat(&q, &task) {
1125 continue;
1127 }
1128 }
1134
1135 match task.hold_source {
1136 Some(HoldSource::Machine) => {
1137 let overridden = task
1138 .resume_override
1139 .as_ref()
1140 .is_some_and(|o| o.conductor_rehold.is_some() && !o.forced);
1141 if overridden {
1142 if file_question(questions, &task, Bucket::ConductorOverride, w, now).is_some()
1143 {
1144 report.asked.push(task.id.clone());
1145 }
1146 } else if cfg.as_ref().and_then(|c| machine_cause_resolved(&task, c)) == Some(true)
1147 {
1148 task.release();
1149 if queue.put(&mut task).is_ok() {
1150 report.resumed.push(task.id.clone());
1151 }
1152 } else if file_question(questions, &task, Bucket::MachineUnknown, w, now).is_some()
1153 {
1154 report.asked.push(task.id.clone());
1155 }
1156 }
1157 None => {
1158 if file_question(questions, &task, Bucket::Legacy, w, now).is_some() {
1159 report.asked.push(task.id.clone());
1160 }
1161 }
1162 Some(HoldSource::Manual) => {
1163 if manual_is_stale(&task, now)
1164 && file_question(questions, &task, Bucket::ManualStale, w, now).is_some()
1165 {
1166 report.asked.push(task.id.clone());
1167 }
1168 }
1169 }
1170 }
1171 report
1172}
1173
1174pub fn open_question_for(questions: &Questions, task_id: &str) -> Option<Question> {
1179 latest_triage_question(questions, task_id).filter(|q| q.status.open())
1180}
1181
1182pub fn pending_for(questions: &Questions, task: &Task) -> bool {
1196 let own = match latest_triage_question(questions, &task.id) {
1197 Some(q) if q.status.open() || awaiting_task_chat(&q, task) => true,
1198 Some(q) if q.status == QuestionStatus::Answered => !already_applied(task, &q),
1199 _ => false,
1200 };
1201 own || deps_pending(questions, task)
1202}
1203
1204fn awaiting_task_chat(q: &Question, task: &Task) -> bool {
1205 q.status != QuestionStatus::Abandoned
1206 && q.routed_chat.is_some()
1207 && !already_applied(task, q)
1208 && task.resumption.as_ref().map_or(0, |r| r.generation) == q.triage_generation.unwrap_or(0)
1209 && task.operator_decided_at.is_none_or(|at| at <= q.asked_at)
1210}
1211
1212fn deps_pending(questions: &Questions, task: &Task) -> bool {
1216 match latest_question(questions, DEPS_NODE, &task.id) {
1217 Some(q) if q.status.open() || awaiting_task_chat(&q, task) => true,
1218 Some(q) if q.status == QuestionStatus::Answered => !task.triage_applied(&q.id),
1219 _ => false,
1220 }
1221}
1222
1223pub fn open_task_ids(questions: &Questions) -> std::collections::BTreeSet<String> {
1228 questions
1229 .list()
1230 .into_iter()
1231 .filter(|q| (q.node == NODE || q.node == DEPS_NODE) && q.status.open())
1232 .map(|q| q.run)
1233 .collect()
1234}
1235
1236#[cfg(test)]
1237mod tests {
1238 use super::*;
1239 use crate::ask::Answer;
1240 use crate::queue::Source;
1241 use jiff::SignedDuration;
1242
1243 fn store() -> (tempfile::TempDir, Queue, Questions) {
1244 let dir = tempfile::tempdir().unwrap();
1245 let q = Queue::at(dir.path().join("queue"));
1246 let s = Questions::at(dir.path().join("questions"));
1247 (dir, q, s)
1248 }
1249
1250 fn task(title: &str, repo: PathBuf) -> Task {
1251 Task::new(title.to_owned(), format!("do {title}"), repo, Source::Human)
1252 }
1253
1254 #[test]
1255 fn resource_holds_only_auto_release_a_proven_cache_wait_timeout() {
1256 use crate::run::{ResourceBlock, ResourceBlockCause, RunState};
1257 let dir = tempfile::tempdir().unwrap();
1258 let cfg = Config::default(); assert!(cfg.cache_dir().is_none());
1260 let mut state = RunState::new(
1261 dir.path().into(),
1262 "main".into(),
1263 "base".into(),
1264 "verify".into(),
1265 cfg.clone(),
1266 );
1267 let mut task = task("verify", dir.path().into());
1268 task.runs = vec![state.id.clone()];
1269 task.hold_machine(Some("[resource] cache operation stopped".into()));
1270 for cause in [
1271 ResourceBlockCause::Other,
1272 ResourceBlockCause::Unrecorded,
1273 ResourceBlockCause::Cancelled { park: None },
1274 ] {
1275 state.resource_block = Some(ResourceBlock {
1276 node: "gate".into(),
1277 kind: "shared build cache".into(),
1278 detail: "persistent metadata or manifest error".into(),
1279 cause,
1280 at: Timestamp::now(),
1281 });
1282 state.save_under(dir.path()).unwrap();
1283 for _ in 0..3 {
1285 assert_eq!(machine_cause_resolved_under(&task, &cfg, dir.path()), None);
1286 }
1287 }
1288 let block = state.resource_block.as_mut().unwrap();
1289 block.cause = ResourceBlockCause::Timeout {
1290 waited_secs: 10,
1291 limit_secs: 10,
1292 };
1293 state.save_under(dir.path()).unwrap();
1294 assert_eq!(
1295 machine_cause_resolved_under(&task, &cfg, dir.path()),
1296 Some(true)
1297 );
1298 state.resource_block.as_mut().unwrap().kind = "repository coordination".into();
1299 state.save_under(dir.path()).unwrap();
1300 assert_eq!(machine_cause_resolved_under(&task, &cfg, dir.path()), None);
1301 task.runs = vec!["unreadable-run".into()];
1302 assert_eq!(machine_cause_resolved_under(&task, &cfg, dir.path()), None);
1303 }
1304
1305 fn gate_disabled_config(dir: &std::path::Path) -> PathBuf {
1309 let config = dir.join("magi.toml");
1310 std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
1311 config
1312 }
1313
1314 #[test]
1315 fn resumption_later_keep_held_answer_wins_but_an_older_answer_cannot_undo_release() {
1316 for source in ["cli", "web"] {
1317 for answer_after_release in [true, false] {
1318 let (dir, queue, questions) = store();
1319 let mut t = task("recovery", dir.path().join("repo"));
1320 t.hold_machine(Some("inspect remote first".into()));
1321 queue.put(&mut t).unwrap();
1322 let mut q = file_question(
1323 &questions,
1324 &t,
1325 Bucket::MachineUnknown,
1326 &JA,
1327 Timestamp::now(),
1328 )
1329 .unwrap();
1330 if !answer_after_release {
1331 q.answer(Answer::Choice(q.choices[1].clone())).unwrap();
1332 questions.put(&mut q).unwrap();
1333 }
1334 t.release_operator(source, None);
1335 queue.put(&mut t).unwrap();
1336 if answer_after_release {
1337 q.answer(Answer::Choice(q.choices[1].clone())).unwrap();
1338 questions.put(&mut q).unwrap();
1339 }
1340 apply_answers(&queue, &questions);
1341 let after = queue.get(&t.id).unwrap();
1342 assert!(after.triage_applied(&q.id));
1343 assert_eq!(after.operator_held(), answer_after_release);
1344 assert_eq!(
1345 after.status,
1346 if answer_after_release {
1347 TaskStatus::Held
1348 } else {
1349 TaskStatus::Queued
1350 }
1351 );
1352 assert!(apply_answers(&queue, &questions).is_empty());
1353 }
1354 }
1355 }
1356
1357 #[test]
1358 fn explicit_resumption_applies_legacy_queued_triage_once_and_preserves_later_holds() {
1359 let (dir, queue, questions) = store();
1360 let mut t = task("recover", dir.path().join("repo"));
1361 t.hold_machine(Some("inspect first".into()));
1362 queue.put(&mut t).unwrap();
1363 let mut q = file_question(
1364 &questions,
1365 &t,
1366 Bucket::MachineUnknown,
1367 &JA,
1368 Timestamp::now(),
1369 )
1370 .unwrap();
1371 assert!(q.actions.is_empty());
1372 q.answer(Answer::Choice(q.choices[0].clone())).unwrap();
1373 questions.put(&mut q).unwrap();
1374 t.release();
1376 t.instruction = "Reuse approved PR head".into();
1377 queue.put(&mut t).unwrap();
1378 assert_eq!(apply_answers(&queue, &questions), [t.id.clone()]);
1379 let mut after = queue.get(&t.id).unwrap();
1380 assert_eq!(
1381 after.resumption.as_ref().unwrap().question_id.as_deref(),
1382 Some(q.id.as_str())
1383 );
1384 assert_eq!(
1385 after.resumption.as_ref().unwrap().instruction,
1386 t.instruction
1387 );
1388 assert_eq!(after.answers[0].answer, JA.resume);
1389 assert!(after.triage_applied(&q.id));
1390 assert!(apply_answers(&Queue::at(dir.path().join("queue")), &questions).is_empty());
1391 after.hold_machine(Some("new blocker".into()));
1392 queue.put(&mut after).unwrap();
1393 let mut late = file_question(
1394 &questions,
1395 &after,
1396 Bucket::MachineUnknown,
1397 &EN,
1398 Timestamp::now(),
1399 )
1400 .unwrap();
1401 after.hold_manual(Some("keep held now".into()));
1402 queue.put(&mut after).unwrap();
1403 late.answer(Answer::Choice(late.choices[0].clone()))
1404 .unwrap();
1405 questions.put(&mut late).unwrap();
1406 apply_answers(&queue, &questions);
1407 let held = queue.get(&t.id).unwrap();
1408 assert!(held.operator_held());
1409 assert_eq!(held.hold_reason.as_deref(), Some("keep held now"));
1410 assert_eq!(held.resumption.as_ref().unwrap().generation, 1);
1411 }
1412
1413 #[test]
1414 fn a_resolved_machine_hold_is_requeued_automatically() {
1415 let (dir, q, questions) = store();
1416 let config = gate_disabled_config(dir.path());
1417 let mut t = task("disk pressure", dir.path().join("repo"));
1418 t.hold_machine(Some(
1419 "not enough free space to start a run: 10 bytes free, 100 required by \
1420 `[disk] min_free_bytes`"
1421 .to_owned(),
1422 ));
1423 q.put(&mut t).unwrap();
1424
1425 let report = run_once(&q, &questions, Some(&config), Timestamp::now());
1426 assert_eq!(report.resumed, [t.id.clone()]);
1427 assert!(report.asked.is_empty());
1428
1429 let back = q.get(&t.id).unwrap();
1430 assert_eq!(back.status, TaskStatus::Queued);
1431 assert!(back.hold_source.is_none());
1432 assert!(questions.list().is_empty(), "nothing needed asking");
1433 }
1434
1435 #[test]
1436 fn a_machine_hold_with_no_recognised_cause_gets_one_question_not_two() {
1437 let (dir, q, questions) = store();
1438 let mut t = task("gate went red", dir.path().join("repo"));
1439 t.hold_machine(Some("gate red".to_owned()));
1440 q.put(&mut t).unwrap();
1441
1442 let first = run_once(&q, &questions, None, Timestamp::now());
1443 assert_eq!(first.asked, [t.id.clone()]);
1444 assert!(first.resumed.is_empty());
1445
1446 let open: Vec<_> = questions
1447 .list()
1448 .into_iter()
1449 .filter(|q| q.status.open())
1450 .collect();
1451 assert_eq!(open.len(), 1);
1452 assert_eq!(open[0].run, t.id);
1453 assert_eq!(open[0].node, NODE);
1454 assert_eq!(open[0].choices.len(), 3);
1455
1456 let second = run_once(&q, &questions, None, Timestamp::now());
1458 assert!(second.asked.is_empty());
1459 assert_eq!(
1460 questions
1461 .list()
1462 .into_iter()
1463 .filter(|q| q.status.open())
1464 .count(),
1465 1
1466 );
1467 }
1468
1469 #[test]
1470 fn a_legacy_hold_with_no_recorded_source_gets_exactly_one_question() {
1471 let (dir, q, questions) = store();
1472 let mut t = task("schema 1 record", dir.path().join("repo"));
1473 t.status = TaskStatus::Held;
1474 assert!(t.hold_source.is_none(), "the case this test is about");
1475 q.put(&mut t).unwrap();
1476
1477 let first = run_once(&q, &questions, None, Timestamp::now());
1478 assert_eq!(first.asked, [t.id.clone()]);
1479
1480 let second = run_once(&q, &questions, None, Timestamp::now());
1481 assert!(
1482 second.asked.is_empty(),
1483 "the same legacy hold must not be asked about twice"
1484 );
1485 assert_eq!(
1486 questions
1487 .list()
1488 .into_iter()
1489 .filter(|q| q.status.open())
1490 .count(),
1491 1
1492 );
1493 }
1494
1495 #[test]
1496 fn dependency_resolution_detaches_legacy_short_refs_without_dropping_other_waits() {
1497 let (dir, queue, questions) = store();
1498 let mut root = task("root", dir.path().join("repo"));
1499 root.id = "20261013-193611-ca5b".into();
1500 root.hold_machine(Some("needs owner decision".into()));
1501 queue.put(&mut root).unwrap();
1502 let mut other = task("other", dir.path().join("repo"));
1503 other.id = "20261013-192213-feb2".into();
1504 queue.put(&mut other).unwrap();
1505 let mut waiting = task("waiting", dir.path().join("repo"));
1506 waiting.block(vec!["ca5b".into(), other.id.clone()], None);
1507 queue.put(&mut waiting).unwrap();
1508 detach_dependants(&queue, &questions, &root.id);
1509 let after = queue.get(&waiting.id).unwrap();
1510 assert_eq!(after.status, TaskStatus::Blocked);
1511 assert_eq!(after.blocked_by, [other.id]);
1512 }
1513
1514 #[test]
1515 fn a_manual_hold_is_never_auto_resumed() {
1516 let (dir, q, questions) = store();
1517 let mut t = task("operator stopped this", dir.path().join("repo"));
1518 t.hold_manual(Some("waiting on a decision".to_owned()));
1519 q.put(&mut t).unwrap();
1520
1521 let report = run_once(&q, &questions, None, Timestamp::now());
1522 assert!(report.resumed.is_empty());
1523 assert!(report.asked.is_empty());
1525
1526 let back = q.get(&t.id).unwrap();
1527 assert_eq!(back.status, TaskStatus::Held);
1528 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1529 assert!(questions.list().is_empty());
1530 }
1531
1532 #[test]
1533 fn a_blocked_task_on_a_deleted_dependency_is_held_and_asked_about_in_one_pass() {
1534 let (dir, q, questions) = store();
1539 let mut still_going = task("still valid", dir.path().join("repo"));
1540 q.put(&mut still_going).unwrap();
1541
1542 let mut t = task("orphaned", dir.path().join("repo"));
1543 t.block(
1544 vec!["20260101-000000-gone".to_owned(), still_going.id.clone()],
1545 Some("waits on both".to_owned()),
1546 );
1547 q.put(&mut t).unwrap();
1548
1549 let report = run_once(&q, &questions, None, Timestamp::now());
1550 assert_eq!(report.quarantined, [t.id.clone()]);
1551 assert_eq!(
1552 report.asked,
1553 [t.id.clone()],
1554 "the fresh machine hold must earn a question in the same pass"
1555 );
1556
1557 let after = q.get(&t.id).unwrap();
1558 assert_eq!(after.status, TaskStatus::Held);
1559 assert_eq!(after.hold_source, Some(HoldSource::Machine));
1560 assert!(after.blocked_by.is_empty());
1561
1562 assert!(!is_disk_hold(&after));
1565
1566 let open: Vec<_> = questions
1567 .list()
1568 .into_iter()
1569 .filter(|q| q.status.open())
1570 .collect();
1571 assert_eq!(open.len(), 1);
1572 assert_eq!(open[0].run, t.id);
1573
1574 let second = run_once(&q, &questions, None, Timestamp::now());
1576 assert!(second.quarantined.is_empty());
1577 assert!(second.asked.is_empty());
1578 }
1579
1580 fn stuck_chain(q: &Queue, dir: &std::path::Path) -> (Task, Task, Task) {
1583 let mut root = task("root", dir.join("repo"));
1584 root.hold_manual(Some("waiting".to_owned()));
1585 q.put(&mut root).unwrap();
1586 let mut mid = task("mid", dir.join("repo"));
1587 mid.block(vec![root.id.clone()], None);
1588 q.put(&mut mid).unwrap();
1589 let mut leaf = task("leaf", dir.join("repo"));
1590 leaf.block(vec![mid.id.clone()], None);
1591 q.put(&mut leaf).unwrap();
1592 (root, mid, leaf)
1593 }
1594
1595 fn deps_questions(questions: &Questions) -> Vec<Question> {
1596 questions
1597 .list()
1598 .into_iter()
1599 .filter(|q| q.node == DEPS_NODE)
1600 .collect()
1601 }
1602
1603 fn answer_deps(questions: &Questions, choice: usize) {
1604 let mut asked = deps_questions(questions).remove(0);
1605 let c = asked.choices[choice].clone();
1606 asked.answer(Answer::Choice(c)).unwrap();
1607 questions.put(&mut asked).unwrap();
1608 }
1609
1610 #[test]
1611 fn a_stuck_chain_earns_one_question_naming_every_dependant_and_is_not_reasked() {
1612 let (dir, q, questions) = store();
1613 let (root, mid, leaf) = stuck_chain(&q, dir.path());
1614
1615 let first = run_once(&q, &questions, None, Timestamp::now());
1616 assert_eq!(first.asked, std::slice::from_ref(&root.id));
1617 let asked = deps_questions(&questions);
1618 assert_eq!(asked.len(), 1, "one question per root, not per dependant");
1619 assert_eq!(asked[0].run, root.id);
1620 assert_eq!(asked[0].choices.len(), 3);
1621 assert!(asked[0].detail.contains(mid.short()), "{}", asked[0].detail);
1622 assert!(
1623 asked[0].detail.contains(leaf.short()),
1624 "{}",
1625 asked[0].detail
1626 );
1627
1628 let second = run_once(&q, &questions, None, Timestamp::now());
1629 assert!(second.asked.is_empty());
1630 assert_eq!(deps_questions(&questions).len(), 1);
1631 assert!(pending_for(&questions, &q.get(&root.id).unwrap()));
1632 }
1633
1634 #[test]
1635 fn a_stuck_root_with_a_machine_hold_gets_only_the_dependency_question() {
1636 let (dir, q, questions) = store();
1637 let mut root = task("root", dir.path().join("repo"));
1638 root.hold_machine(Some("gate red".to_owned()));
1639 q.put(&mut root).unwrap();
1640 let mut dep = task("dep", dir.path().join("repo"));
1641 dep.block(vec![root.id.clone()], None);
1642 q.put(&mut dep).unwrap();
1643
1644 run_once(&q, &questions, None, Timestamp::now());
1645 run_once(&q, &questions, None, Timestamp::now());
1646 assert_eq!(deps_questions(&questions).len(), 1);
1647 assert_eq!(
1648 questions.list().len(),
1649 1,
1650 "no second question about the hold"
1651 );
1652 }
1653
1654 #[test]
1655 fn a_dependency_that_can_still_run_asks_nothing() {
1656 let (dir, q, questions) = store();
1657 let mut live = task("live", dir.path().join("repo"));
1658 q.put(&mut live).unwrap();
1659 let mut dep = task("dep", dir.path().join("repo"));
1660 dep.block(vec![live.id.clone()], None);
1661 q.put(&mut dep).unwrap();
1662 let report = run_once(&q, &questions, None, Timestamp::now());
1663 assert!(report.asked.is_empty());
1664 assert!(questions.list().is_empty());
1665 }
1666
1667 #[test]
1668 fn answering_release_requeues_the_root_and_is_not_reasked() {
1669 let (dir, q, questions) = store();
1670 let (root, _mid, _leaf) = stuck_chain(&q, dir.path());
1671 run_once(&q, &questions, None, Timestamp::now());
1672 answer_deps(&questions, 0);
1673
1674 let report = run_once(&q, &questions, None, Timestamp::now());
1675 assert_eq!(report.answered, std::slice::from_ref(&root.id));
1676 assert!(report.asked.is_empty(), "applying must not re-ask");
1677 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Queued);
1678 let again = run_once(&q, &questions, None, Timestamp::now());
1679 assert!(again.asked.is_empty() && again.answered.is_empty());
1680 assert_eq!(deps_questions(&questions).len(), 1);
1681 }
1682
1683 #[test]
1684 fn answering_detach_frees_the_direct_dependant_and_keeps_the_root() {
1685 let (dir, q, questions) = store();
1686 let (root, mid, leaf) = stuck_chain(&q, dir.path());
1687 run_once(&q, &questions, None, Timestamp::now());
1688 answer_deps(&questions, 2);
1689
1690 let report = run_once(&q, &questions, None, Timestamp::now());
1691 assert!(report.asked.is_empty());
1692 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1693 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1694 let leaf = q.get(&leaf.id).unwrap();
1695 assert_eq!(leaf.status, TaskStatus::Blocked, "still waits on mid");
1696 assert_eq!(leaf.blocked_by, std::slice::from_ref(&mid.id));
1697 }
1698
1699 #[test]
1700 fn a_dependant_that_can_progress_through_another_blocker_is_not_stuck() {
1701 let (dir, q, questions) = store();
1702 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1703 let mut live = task("live", dir.path().join("repo"));
1704 q.put(&mut live).unwrap();
1705 let mut both = q.get(&mid.id).unwrap();
1706 both.block(vec![root.id.clone(), live.id.clone()], None);
1707 q.put(&mut both).unwrap();
1708 assert!(
1710 run_once(&q, &questions, None, Timestamp::now())
1711 .asked
1712 .is_empty()
1713 );
1714 }
1715
1716 #[test]
1717 fn answering_discard_detaches_then_removes_the_root() {
1718 let (dir, q, questions) = store();
1719 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1720 run_once(&q, &questions, None, Timestamp::now());
1721 answer_deps(&questions, 1);
1722
1723 let report = run_once(&q, &questions, None, Timestamp::now());
1724 assert!(q.get(&root.id).is_err());
1725 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1726 assert!(
1727 report.asked.is_empty(),
1728 "no per-dependant quarantine question"
1729 );
1730 assert!(report.quarantined.is_empty());
1731 }
1732
1733 #[test]
1734 fn an_unmatched_deps_answer_changes_nothing_and_is_not_reasked() {
1735 let (dir, q, questions) = store();
1736 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1737 run_once(&q, &questions, None, Timestamp::now());
1738 let mut asked = deps_questions(&questions).remove(0);
1739 asked.choices.clear();
1740 asked.answer(Answer::Text("dunno".to_owned())).unwrap();
1741 questions.put(&mut asked).unwrap();
1742
1743 for _ in 0..2 {
1744 let report = run_once(&q, &questions, None, Timestamp::now());
1745 assert!(report.asked.is_empty() && report.answered.is_empty());
1746 }
1747 assert_eq!(deps_questions(&questions).len(), 1);
1748 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1749 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Blocked);
1750 }
1751
1752 #[test]
1753 fn a_dependency_cycle_terminates_and_is_asked_about_once() {
1754 let (dir, q, questions) = store();
1755 let mut a = task("a", dir.path().join("repo"));
1756 let mut b = task("b", dir.path().join("repo"));
1757 a.block(vec![b.id.clone()], None);
1758 b.block(vec![a.id.clone()], None);
1759 q.put(&mut a).unwrap();
1760 q.put(&mut b).unwrap();
1761
1762 let first = run_once(&q, &questions, None, Timestamp::now());
1763 assert_eq!(first.asked.len(), 1);
1764 let second = run_once(&q, &questions, None, Timestamp::now());
1765 assert!(second.asked.is_empty());
1766 assert_eq!(deps_questions(&questions).len(), 1);
1767 }
1768
1769 #[test]
1770 fn a_missing_dependency_is_still_quarantined_not_asked_about_as_stuck() {
1771 let (dir, q, questions) = store();
1772 let mut t = task("orphan", dir.path().join("repo"));
1773 t.block(vec!["20260101-000000-gone".to_owned()], None);
1774 q.put(&mut t).unwrap();
1775 let report = run_once(&q, &questions, None, Timestamp::now());
1776 assert_eq!(report.quarantined, [t.id.clone()]);
1777 assert!(deps_questions(&questions).is_empty());
1778 }
1779
1780 #[test]
1781 fn a_stale_manual_hold_earns_a_two_choice_question() {
1782 let (dir, q, questions) = store();
1783 let mut t = task("been sitting a while", dir.path().join("repo"));
1784 t.hold_manual(Some("waiting on a decision".to_owned()));
1785 q.put(&mut t).unwrap();
1786 let mut back = q.get(&t.id).unwrap();
1788 back.updated_at = Timestamp::now() - SignedDuration::new(8 * 24 * 60 * 60, 0);
1789 std::fs::write(
1790 q.path_of(&back.id),
1791 serde_json::to_string_pretty(&back).unwrap(),
1792 )
1793 .unwrap();
1794
1795 let report = run_once(&q, &questions, None, Timestamp::now());
1796 assert_eq!(report.asked, [t.id.clone()]);
1797 let open: Vec<_> = questions
1798 .list()
1799 .into_iter()
1800 .filter(|q| q.status.open())
1801 .collect();
1802 assert_eq!(open.len(), 1);
1803 assert_eq!(open[0].choices.len(), 2);
1804 }
1805
1806 #[test]
1807 fn answering_resume_releases_the_task() {
1808 let (dir, q, questions) = store();
1809 let mut t = task("gate went red", dir.path().join("repo"));
1810 t.hold_machine(Some("gate red".to_owned()));
1811 q.put(&mut t).unwrap();
1812 run_once(&q, &questions, None, Timestamp::now());
1813
1814 let mut asked = questions
1815 .list()
1816 .into_iter()
1817 .find(|q| q.run == t.id)
1818 .unwrap();
1819 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1820 questions.put(&mut asked).unwrap();
1821
1822 let report = run_once(&q, &questions, None, Timestamp::now());
1823 assert_eq!(report.answered, [t.id.clone()]);
1824 let back = q.get(&t.id).unwrap();
1825 assert_eq!(back.status, TaskStatus::Queued);
1826 assert!(back.hold_source.is_none());
1827 }
1828
1829 fn resumed_then_failed(q: &Queue, questions: &Questions, dir: &std::path::Path) -> Task {
1831 let mut t = task("gate went red", dir.join("repo"));
1832 t.hold_machine(Some("gate red".to_owned()));
1833 q.put(&mut t).unwrap();
1834 run_once(q, questions, None, Timestamp::now());
1835 let mut asked = questions
1836 .list()
1837 .into_iter()
1838 .find(|q| q.run == t.id)
1839 .unwrap();
1840 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1841 questions.put(&mut asked).unwrap();
1842
1843 let report = run_once(q, questions, None, Timestamp::now());
1844 assert_eq!(report.answered, [t.id.clone()]);
1845 let mut back = q.get(&t.id).unwrap();
1846 assert_eq!(back.status, TaskStatus::Queued);
1847 assert_eq!(back.attempts, 0);
1848
1849 back.start("run-1".to_owned());
1850 back.fail("rebase conflict", 1);
1851 assert_eq!(back.status, TaskStatus::Held);
1852 q.put(&mut back).unwrap();
1853 back
1854 }
1855
1856 #[test]
1857 fn a_resume_answer_is_applied_once_and_a_new_machine_hold_is_asked_about() {
1858 let (dir, q, questions) = store();
1859 let t = resumed_then_failed(&q, &questions, dir.path());
1860
1861 let report = run_once(&q, &questions, None, Timestamp::now());
1862 assert!(report.answered.is_empty(), "the old answer must not replay");
1863 assert_eq!(report.asked, std::slice::from_ref(&t.id));
1864 let back = q.get(&t.id).unwrap();
1865 assert_eq!(back.status, TaskStatus::Held);
1866 assert_eq!(
1867 questions
1868 .list()
1869 .into_iter()
1870 .filter(|q| q.status.open())
1871 .count(),
1872 1
1873 );
1874 }
1875
1876 #[test]
1877 fn a_manual_hold_placed_after_a_resume_is_not_undone_by_the_old_answer() {
1878 let (dir, q, questions) = store();
1879 let mut t = resumed_then_failed(&q, &questions, dir.path());
1880 t.hold_manual(Some("operator stopped this".to_owned()));
1881 q.put(&mut t).unwrap();
1882 let before = questions.list().len();
1883
1884 let report = run_once(&q, &questions, None, Timestamp::now());
1885 assert!(report.answered.is_empty());
1886 assert!(report.asked.is_empty());
1887 let back = q.get(&t.id).unwrap();
1888 assert_eq!(back.status, TaskStatus::Held);
1889 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1890 assert_eq!(questions.list().len(), before);
1891 }
1892
1893 #[test]
1894 fn answering_not_yet_keeps_it_held_as_a_manual_hold_and_does_not_reapply() {
1895 let (dir, q, questions) = store();
1896 let mut t = task("gate went red", dir.path().join("repo"));
1897 t.hold_machine(Some("gate red".to_owned()));
1898 q.put(&mut t).unwrap();
1899 run_once(&q, &questions, None, Timestamp::now());
1900
1901 let mut asked = questions
1902 .list()
1903 .into_iter()
1904 .find(|q| q.run == t.id)
1905 .unwrap();
1906 asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
1907 questions.put(&mut asked).unwrap();
1908
1909 let report = run_once(&q, &questions, None, Timestamp::now());
1910 assert_eq!(report.answered, [t.id.clone()]);
1911 let back = q.get(&t.id).unwrap();
1912 assert_eq!(back.status, TaskStatus::Held);
1913 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1914 assert!(
1915 back.hold_reason
1916 .as_deref()
1917 .is_some_and(|r| r.contains("gate red")),
1918 "the original cause must survive a \"not yet\" answer, not just the \
1919 triage marker: {:?}",
1920 back.hold_reason
1921 );
1922
1923 let third = run_once(&q, &questions, None, Timestamp::now());
1927 assert!(third.answered.is_empty());
1928 assert!(third.asked.is_empty());
1929 }
1930
1931 #[test]
1932 fn answering_discard_removes_the_task_entirely() {
1933 let (dir, q, questions) = store();
1934 let mut t = task("gate went red", dir.path().join("repo"));
1935 t.hold_machine(Some("gate red".to_owned()));
1936 q.put(&mut t).unwrap();
1937 run_once(&q, &questions, None, Timestamp::now());
1938
1939 let mut asked = questions
1940 .list()
1941 .into_iter()
1942 .find(|q| q.run == t.id)
1943 .unwrap();
1944 asked.answer(Answer::Choice(EN.discard.to_owned())).unwrap();
1945 questions.put(&mut asked).unwrap();
1946
1947 let report = run_once(&q, &questions, None, Timestamp::now());
1948 assert_eq!(report.answered, [t.id.clone()]);
1949 assert!(
1950 q.get(&t.id).is_err(),
1951 "\"discard it\" (捨ててよい) must actually discard the task, not \
1952 just leave it sitting held forever"
1953 );
1954 }
1955
1956 #[test]
1957 fn an_answer_is_read_by_its_position_in_choices_not_by_the_callers_current_language() {
1958 let (dir, q, questions) = store();
1960 let ja_config = dir.path().join("ja.toml");
1961 std::fs::write(&ja_config, "[graph]\nlanguage = \"ja\"\n").unwrap();
1962 let mut t = task("gate went red", dir.path().join("repo"));
1963 t.hold_machine(Some("gate red".to_owned()));
1964 q.put(&mut t).unwrap();
1965 run_once(&q, &questions, Some(&ja_config), Timestamp::now());
1966
1967 let mut asked = questions
1968 .list()
1969 .into_iter()
1970 .find(|q| q.run == t.id)
1971 .unwrap();
1972 assert_eq!(asked.choices[0], JA.resume, "filed in Japanese");
1973 asked.answer(Answer::Choice(JA.resume.to_owned())).unwrap();
1974 questions.put(&mut asked).unwrap();
1975
1976 let en_config = dir.path().join("en.toml");
1981 std::fs::write(&en_config, "[graph]\nlanguage = \"en\"\n").unwrap();
1982 let report = run_once(&q, &questions, Some(&en_config), Timestamp::now());
1983 assert_eq!(report.answered, [t.id.clone()]);
1984 let back = q.get(&t.id).unwrap();
1985 assert_eq!(
1986 back.status,
1987 TaskStatus::Queued,
1988 "a resume answer must resume the task regardless of which \
1989 language it is read back in"
1990 );
1991 }
1992
1993 #[test]
1994 fn a_question_falls_back_to_last_error_when_hold_reason_was_never_set() {
1995 let (dir, q, questions) = store();
2000 let mut t = task("kept failing the gate", dir.path().join("repo"));
2001 t.start("run-1".to_owned());
2002 t.fail("gate red three times running", 1);
2003 assert_eq!(t.status, TaskStatus::Held);
2004 t.hold_reason = None; q.put(&mut t).unwrap();
2006
2007 run_once(&q, &questions, None, Timestamp::now());
2008 let asked = questions
2009 .list()
2010 .into_iter()
2011 .find(|q| q.run == t.id)
2012 .unwrap();
2013 assert!(
2014 asked.detail.contains("gate red three times running"),
2015 "the question must surface `last_error` when there is no \
2016 `hold_reason` to show instead: {}",
2017 asked.detail
2018 );
2019 }
2020
2021 #[test]
2022 fn open_question_for_and_open_task_ids_reflect_only_what_is_still_waiting() {
2023 let (dir, q, questions) = store();
2024 let mut t = task("gate went red", dir.path().join("repo"));
2025 t.hold_machine(Some("gate red".to_owned()));
2026 q.put(&mut t).unwrap();
2027
2028 assert!(open_question_for(&questions, &t.id).is_none());
2029 assert!(!open_task_ids(&questions).contains(&t.id));
2030
2031 run_once(&q, &questions, None, Timestamp::now());
2032 assert!(open_question_for(&questions, &t.id).is_some());
2033 assert!(open_task_ids(&questions).contains(&t.id));
2034
2035 let mut asked = questions
2036 .list()
2037 .into_iter()
2038 .find(|q| q.run == t.id)
2039 .unwrap();
2040 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
2041 questions.put(&mut asked).unwrap();
2042
2043 assert!(
2044 open_question_for(&questions, &t.id).is_none(),
2045 "an answered question is no longer open"
2046 );
2047 assert!(!open_task_ids(&questions).contains(&t.id));
2048 }
2049
2050 #[test]
2051 fn pending_for_stays_true_between_an_answer_and_the_next_run_once_pass() {
2052 let (dir, q, questions) = store();
2058 let mut t = task("gate went red", dir.path().join("repo"));
2059 t.hold_machine(Some("gate red".to_owned()));
2060 q.put(&mut t).unwrap();
2061
2062 assert!(!pending_for(&questions, &t));
2063
2064 run_once(&q, &questions, None, Timestamp::now());
2065 let held = q.get(&t.id).unwrap();
2066 assert!(pending_for(&questions, &held), "still waiting on an answer");
2067
2068 let mut asked = questions
2069 .list()
2070 .into_iter()
2071 .find(|q| q.run == t.id)
2072 .unwrap();
2073 asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
2074 questions.put(&mut asked).unwrap();
2075 assert!(!asked.status.open());
2076
2077 let still_held = q.get(&t.id).unwrap();
2079 assert!(
2080 pending_for(&questions, &still_held),
2081 "answered but not yet applied is still pending"
2082 );
2083
2084 run_once(&q, &questions, None, Timestamp::now());
2085 let after = q.get(&t.id).unwrap();
2086 assert!(
2087 !pending_for(&questions, &after),
2088 "the answer is applied now, nothing left pending"
2089 );
2090 }
2091}