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 if task
404 .hold_reason
405 .as_deref()
406 .is_some_and(|r| r.starts_with(crate::cache::RESOURCE_HOLD))
407 {
408 return Some(
411 cfg.cache_dir()
412 .is_none_or(|dir| !crate::cache::in_use(&crate::run::home(), &dir)),
413 );
414 }
415 if !is_disk_hold(task) {
416 return None;
417 }
418 let min = cfg.disk.min_free_bytes;
419 if min == 0 {
420 return Some(true);
423 }
424 Some(crate::daemon::disk_gate(&repo_for(task), cfg).is_none())
425}
426
427fn manual_is_stale(task: &Task, now: Timestamp) -> bool {
431 now.as_second() - task.updated_at.as_second() > MANUAL_STALE_AFTER.as_secs() as i64
432}
433
434fn marker_for(q: &Question) -> String {
437 format!("[triage:{}]", q.short())
438}
439
440fn already_applied(task: &Task, q: &Question) -> bool {
443 if task.triage_applied(&q.id) {
444 return true;
445 }
446 let marker = marker_for(q);
449 task.hold_reason
450 .as_deref()
451 .is_some_and(|r| r.contains(marker.as_str()))
452}
453
454fn latest_triage_question(questions: &Questions, task_id: &str) -> Option<Question> {
459 latest_question(questions, NODE, task_id)
460}
461
462fn latest_question(questions: &Questions, node: &str, task_id: &str) -> Option<Question> {
464 questions
465 .list()
466 .into_iter()
467 .filter(|q| q.node == node && q.run == task_id)
468 .max_by(|a, b| a.asked_at.cmp(&b.asked_at).then_with(|| a.id.cmp(&b.id)))
471}
472
473#[derive(Debug, Clone, Copy, PartialEq, Eq)]
476enum AnswerAction {
477 Resume,
481 Discard,
484 KeepHeld,
489}
490
491fn interpret_answer(q: &Question) -> AnswerAction {
504 let resolution = q.resolution().unwrap_or_default();
505 match q.choices.iter().position(|c| *c == resolution) {
506 Some(0) => AnswerAction::Resume,
507 Some(2) => AnswerAction::Discard,
508 _ => AnswerAction::KeepHeld,
509 }
510}
511
512pub(crate) fn contextual_keep_held(q: &Question, label: &str) -> bool {
515 q.node == NODE
516 && q.seat == SEAT
517 && q.actions.is_empty()
518 && q.choices.get(1).is_some_and(|c| c == label)
519}
520
521fn keep_held_note(task: &Task, q: &Question, resolution: &str) -> String {
526 let marker = format!("{} operator: {resolution}", marker_for(q));
527 match task.hold_reason.as_deref() {
528 Some(existing) if !existing.is_empty() => format!("{existing}\n{marker}"),
529 _ => marker,
530 }
531}
532
533fn file_question(
537 questions: &Questions,
538 task: &Task,
539 bucket: Bucket,
540 w: &Wording,
541 now: Timestamp,
542) -> Option<Question> {
543 let (summary, why, choices) = match bucket {
544 Bucket::MachineUnknown => (
545 w.summary_machine_unknown(task),
546 w.why_machine().to_owned(),
547 w.choices3(),
548 ),
549 Bucket::ConductorOverride => (
550 w.summary_conductor_override(task),
551 task.resume_override
552 .as_ref()
553 .map(|o| w.why_conductor_override(o))
554 .unwrap_or_default(),
555 w.choices_conductor_override(),
556 ),
557 Bucket::Legacy => (
558 w.summary_legacy(task),
559 w.why_legacy().to_owned(),
560 w.choices3(),
561 ),
562 Bucket::ManualStale => {
563 let days = (now.as_second() - task.updated_at.as_second()) / (24 * 60 * 60);
564 (
565 w.summary_manual_stale(task, days),
566 w.why_manual().to_owned(),
567 w.choices2(),
568 )
569 }
570 };
571 let mut q = Question::new(
572 task.id.clone(),
573 NODE.to_owned(),
574 SEAT.to_owned(),
575 summary,
576 w.detail(task, &why),
577 choices,
578 );
579 q.triage_generation = Some(task.resumption.as_ref().map_or(0, |r| r.generation));
580 questions.put_executable(&mut q).ok()?;
581 Some(q)
582}
583
584pub(crate) fn deputy_brief(q: &Question, queue: &Queue) -> String {
589 let task = queue.get(&q.run).ok();
590 let mut s = match &task {
591 Some(t) => format!(
592 "Task {} (`magi task show {}`), \"{}\", is {}. Hold source: {}. Reason: {}.",
593 t.id,
594 t.id,
595 t.title,
596 t.status.as_str(),
597 match t.hold_source {
598 Some(HoldSource::Machine) => "machine (an automatic hold)",
599 Some(HoldSource::Manual) => "manual (an operator held it)",
600 None => "unknown (an old record, treated as the operator's)",
601 },
602 t.hold_reason
603 .as_deref()
604 .or(t.last_error.as_deref())
605 .unwrap_or("(none recorded)")
606 ),
607 None => format!("Task {} no longer exists (its record is gone).", q.run),
608 };
609 s.push_str("\n\nWhat each option does, by position (the owner taps a button; you only read their words):");
610 let n = q.choices.len();
611 for (i, c) in q.choices.iter().enumerate() {
612 let effect = if q.node == DEPS_NODE {
613 match i {
614 0 => "the dependency is released so the loop runs it",
615 1 => {
616 "the dependants are detached and the dependency task is DELETED; this cannot be undone"
617 }
618 2 => "the dependants are detached; the dependency stays as it is",
619 _ => "nothing is applied",
620 }
621 } else {
622 match i {
623 0 => "the task is released and requeued",
624 2 if n > 2 => "the task is DELETED; this cannot be undone",
625 _ => "the task stays held, with the answer noted",
626 }
627 };
628 s.push_str(&format!("\n- `{c}`: {effect}"));
629 }
630 s.push_str(
631 "\n\nThe daemon applies the answer on its next pass; you apply nothing. \
632 `magi ask --settle` is refused while the task is held by the operator (a \
633 manual hold, or an old record): an answer must not release or delete \
634 something a person parked. In that case do not retry it - ask the owner \
635 with `--thread` to tap the button themselves. The destructive option is \
636 accepted only for an unhedged, verbatim instruction.",
637 );
638 s
639}
640
641fn quarantine_orphaned_blocked(queue: &Queue, questions: &Questions) -> Vec<String> {
659 let mut quarantined = Vec::new();
660 for listed in queue.list() {
661 if listed.status != TaskStatus::Blocked || listed.blocked_by.is_empty() {
662 continue;
663 }
664 let Ok(_claim) = queue.claim(&listed.id) else {
665 continue;
666 };
667 let Ok(mut task) = queue.get(&listed.id) else {
668 continue;
669 };
670 if task.status != TaskStatus::Blocked {
671 continue;
672 }
673 let deleted = queue.apply_deleted_blockers(&mut task, questions);
674 if !deleted.is_empty() && queue.put(&mut task).is_ok() {
675 for id in &deleted {
676 queue.note_dependency_deleted(&task, id);
677 }
678 }
679 if task.status != TaskStatus::Blocked {
680 continue;
681 }
682 let missing = crate::queue::missing_blockers(queue, questions, &task.blocked_by);
683 if missing.is_empty() {
684 continue;
685 }
686 let language = crate::lang::of_repo(&repo_for(&task));
687 let mut cause = missing.clone();
688 cause.sort();
689 cause.dedup();
690 task.notice_cause(&format!("missing-blockers:{}", cause.join(",")), None, None);
691 task.hold_machine(Some(crate::queue::missing_blocker_hold_reason_in(
692 &task.blocked_by,
693 &missing,
694 &language,
695 )));
696 if queue.put(&mut task).is_ok() {
697 quarantined.push(task.id.clone());
698 }
699 }
700 quarantined
701}
702
703#[derive(Debug, Clone, Copy, PartialEq, Eq)]
706enum DepsAction {
707 Release,
708 Discard,
709 Detach,
710 Nothing,
712}
713
714fn interpret_deps_answer(q: &Question) -> DepsAction {
715 let resolution = q.resolution().unwrap_or_default();
716 match q.choices.iter().position(|c| *c == resolution) {
717 Some(0) => DepsAction::Release,
718 Some(1) => DepsAction::Discard,
719 Some(2) => DepsAction::Detach,
720 _ => DepsAction::Nothing,
721 }
722}
723
724fn deps_texts(w: &Wording, root: &Task, dependants: &[&Task]) -> (String, String, Vec<String>) {
726 let ja = w.lang == "ja";
727 let reason = root
728 .hold_reason
729 .as_deref()
730 .or(root.last_error.as_deref())
731 .unwrap_or(if ja {
732 "(記録なし)"
733 } else {
734 "(none recorded)"
735 });
736 let list: String = dependants
737 .iter()
738 .map(|t| format!("- {} {}\n", t.short(), t.title))
739 .collect();
740 if ja {
741 (
742 format!(
743 "{} ({}) が {} 件のタスクを止めています",
744 root.short(),
745 root.status.as_str(),
746 dependants.len()
747 ),
748 format!(
749 "task: {} ({})\ntitle: {}\n状態: {}\n理由: {reason}\n\n\
750 このタスクは自動では実行されないため、依存している次のタスクは永遠に待ち続けます:\n{list}",
751 root.id,
752 root.short(),
753 root.title,
754 root.status.as_str()
755 ),
756 vec![
757 "依存先を再開する".to_owned(),
758 "依存先を捨てる(依存タスクは切り離して実行)".to_owned(),
759 "依存タスクを切り離す(依存先はそのまま)".to_owned(),
760 ],
761 )
762 } else {
763 (
764 format!(
765 "{} ({}) is freezing {} blocked task(s)",
766 root.short(),
767 root.status.as_str(),
768 dependants.len()
769 ),
770 format!(
771 "task: {} ({})\ntitle: {}\nstatus: {}\nreason: {reason}\n\n\
772 Nothing in the loop will ever run this task, so these dependants wait \
773 forever:\n{list}",
774 root.id,
775 root.short(),
776 root.title,
777 root.status.as_str()
778 ),
779 vec![
780 "release the dependency".to_owned(),
781 "discard the dependency (dependants are detached and run)".to_owned(),
782 "detach the dependants (the dependency stays as it is)".to_owned(),
783 ],
784 )
785 }
786}
787
788fn detach_dependants(queue: &Queue, questions: &Questions, root: &str) {
792 for listed in queue.list() {
793 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)) {
794 continue;
795 }
796 let Ok(_claim) = queue.claim(&listed.id) else {
797 continue;
798 };
799 let Ok(mut t) = queue.get(&listed.id) else {
800 continue;
801 };
802 if t.status != TaskStatus::Blocked {
803 continue;
804 }
805 for id in t.blocked_by.clone() {
806 if matches!(crate::queue::dependency(queue, questions, &id), crate::queue::Dependency::Task(full) if full == root)
807 {
808 t.unblock(&id);
809 }
810 }
811 let _ = queue.put(&mut t);
812 }
813}
814
815fn apply_deps_answer(queue: &Queue, questions: &Questions, q: &Question, now: Timestamp) -> bool {
819 let Ok(_claim) = queue.claim(&q.run) else {
820 return false;
821 };
822 let Ok(mut root) = queue.get(&q.run) else {
823 return false;
824 };
825 if root.triage_applied(&q.id) {
826 return false;
827 }
828 if root.operator_decided_at.is_some_and(|at| at > q.asked_at)
829 || q.triage_generation
830 .is_some_and(|g| g != root.resumption.as_ref().map_or(0, |r| r.generation))
831 {
832 root.mark_triage_applied(&q.id);
833 return queue.put(&mut root).is_ok();
834 }
835 match interpret_deps_answer(q) {
836 DepsAction::Release => {
837 if root.status == TaskStatus::Running {
838 return false;
839 }
840 root.answers.push(crate::queue::AnsweredQuestion {
841 question: q.summary.clone(),
842 answer: q.resolution().unwrap_or_default(),
843 });
844 root.release_operator("triage-dependencies", Some(&q.id));
845 root.resume_override = Some(OperatorResume {
846 question_id: q.id.clone(),
847 at: now,
848 conductor_rehold: None,
849 forced: false,
850 pinned_run: None,
851 });
852 }
853 DepsAction::Discard => {
854 detach_dependants(queue, questions, &root.id);
855 return queue.remove(&root.id, false, questions).is_ok();
856 }
857 DepsAction::Detach => detach_dependants(queue, questions, &root.id),
858 DepsAction::Nothing => return false,
863 }
864 root.mark_triage_applied(&q.id);
865 queue.put(&mut root).is_ok()
866}
867
868fn ask_about_stuck_roots(
877 queue: &Queue,
878 questions: &Questions,
879 config_override: Option<&Path>,
880 report: &mut Report,
881) {
882 let all = questions.list();
883 let inv = crate::blockers::Inventory::load(queue, questions);
884 let mut frozen: std::collections::BTreeMap<String, Vec<String>> = Default::default();
885 for (id, roots) in inv.stuck() {
886 for root in roots {
887 if root != id {
888 frozen.entry(root).or_default().push(id.clone());
889 }
890 }
891 }
892 for mut q in all
896 .iter()
897 .filter(|q| q.node == DEPS_NODE && q.status.open())
898 .cloned()
899 {
900 if !frozen.contains_key(&q.run) {
901 q.abandon("nothing waits on this task anymore");
902 let _ = questions.put_executable(&mut q);
903 }
904 }
905
906 for (root_id, dependant_ids) in &frozen {
907 let Ok(_claim) = queue.claim(root_id) else {
908 continue;
909 };
910 let current = crate::blockers::Inventory::load(queue, questions);
913 let Some(root) = current.task(root_id) else {
914 continue;
915 };
916 let dependants: Vec<&Task> = dependant_ids
917 .iter()
918 .filter_map(|d| current.task(d))
919 .filter(|t| current.stuck_roots(t).contains(root_id))
920 .collect();
921 if dependants.is_empty() {
922 continue;
923 }
924 let latest = all
925 .iter()
926 .filter(|q| q.node == DEPS_NODE && q.run == *root_id)
927 .max_by(|a, b| a.asked_at.cmp(&b.asked_at).then_with(|| a.id.cmp(&b.id)));
928 match latest {
929 Some(q) if q.status.open() => continue,
930 Some(q) if q.status == QuestionStatus::Answered && !root.triage_applied(&q.id) => {
931 continue;
932 }
933 Some(q) if q.status == QuestionStatus::Abandoned => {
934 let changed = root.updated_at > q.asked_at
935 || dependants.iter().any(|t| t.updated_at > q.asked_at);
936 if !changed {
937 continue;
938 }
939 }
940 _ => {}
941 }
942 if pending_for(questions, root) {
943 continue;
944 }
945 let cfg = Config::discover(&repo_for(root), config_override)
946 .ok()
947 .map(|(c, _)| c);
948 let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
949 let (summary, detail, choices) = deps_texts(w, root, &dependants);
950 let mut question = Question::new(
951 root.id.clone(),
952 DEPS_NODE.to_owned(),
953 SEAT.to_owned(),
954 summary,
955 detail,
956 choices,
957 );
958 question.triage_generation = Some(root.resumption.as_ref().map_or(0, |r| r.generation));
959 if questions.put_executable(&mut question).is_ok() {
960 report.asked.push(root.id.clone());
961 }
962 }
963}
964
965pub fn apply_answers(queue: &Queue, questions: &Questions) -> Vec<String> {
968 let mut applied = Vec::new();
969 for q in questions.list() {
970 if q.node == DEPS_NODE
971 && q.status == QuestionStatus::Answered
972 && apply_deps_answer(queue, questions, &q, Timestamp::now())
973 {
974 applied.push(q.run.clone());
975 }
976 }
977 for listed in queue.list() {
978 let Some(q) = latest_triage_question(questions, &listed.id) else {
979 continue;
980 };
981 if q.status != QuestionStatus::Answered || already_applied(&listed, &q) {
982 continue;
983 }
984 let Ok(_claim) = queue.claim(&listed.id) else {
985 continue;
986 };
987 let Ok(mut task) = queue.get(&listed.id) else {
988 continue;
989 };
990 if already_applied(&task, &q)
991 || !matches!(task.status, TaskStatus::Held | TaskStatus::Queued)
992 {
993 continue;
994 }
995 let generation = task.resumption.as_ref().map_or(0, |r| r.generation);
996 let action = interpret_answer(&q);
997 let mut newer_decision = q.triage_generation.is_some_and(|g| g != generation)
998 || task.resumption.as_ref().is_some_and(|r| r.at > q.asked_at)
999 || task.held_at.is_some_and(|at| at > q.asked_at)
1000 || task.operator_decided_at.is_some_and(|at| at > q.asked_at);
1001 if matches!(action, AnswerAction::KeepHeld)
1005 && q.answered_at.is_some_and(|at| {
1006 task.operator_decided_at
1007 .is_none_or(|decision| at > decision)
1008 && task.resumption.as_ref().is_none_or(|r| at > r.at)
1009 })
1010 {
1011 newer_decision = false;
1012 }
1013 let resolution = q.resolution().unwrap_or_default();
1015 task.answers.push(crate::queue::AnsweredQuestion {
1016 question: q.summary.clone(),
1017 answer: if newer_decision {
1018 format!("{resolution} (superseded by a later operator decision)")
1019 } else {
1020 resolution
1021 },
1022 });
1023 if !newer_decision {
1024 match action {
1025 AnswerAction::Resume => {
1026 task.release_operator("triage", Some(&q.id));
1027 task.resume_override = Some(OperatorResume {
1028 question_id: q.id.clone(),
1029 at: Timestamp::now(),
1030 conductor_rehold: None,
1031 forced: true,
1032 pinned_run: None,
1033 });
1034 }
1035 AnswerAction::KeepHeld => {
1036 let note = keep_held_note(&task, &q, &q.resolution().unwrap_or_default());
1037 task.hold_manual(Some(note));
1038 }
1039 AnswerAction::Discard => {
1040 if queue.remove(&task.id, false, questions).is_ok() {
1041 applied.push(task.id);
1042 }
1043 continue;
1044 }
1045 }
1046 }
1047 task.mark_triage_applied(&q.id);
1048 if queue.put(&mut task).is_ok() {
1049 applied.push(task.id);
1050 }
1051 }
1052 applied
1053}
1054
1055pub fn run_once(
1073 queue: &Queue,
1074 questions: &Questions,
1075 config_override: Option<&Path>,
1076 now: Timestamp,
1077) -> Report {
1078 let mut report = Report {
1079 answered: apply_answers(queue, questions),
1080 quarantined: quarantine_orphaned_blocked(queue, questions),
1081 ..Report::default()
1082 };
1083 ask_about_stuck_roots(queue, questions, config_override, &mut report);
1084 for listed in queue.list() {
1085 if listed.status != TaskStatus::Held {
1086 continue;
1087 }
1088 let Ok(_claim) = queue.claim(&listed.id) else {
1089 continue;
1090 };
1091 let Ok(mut task) = queue.get(&listed.id) else {
1092 continue;
1093 };
1094 if task.status != TaskStatus::Held {
1098 continue;
1099 }
1100 if deps_pending(questions, &task) {
1104 continue;
1105 }
1106
1107 let cfg = Config::discover(&repo_for(&task), config_override)
1108 .ok()
1109 .map(|(c, _)| c);
1110 let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
1111
1112 if let Some(q) = latest_triage_question(questions, &task.id) {
1113 if q.status.open() {
1114 continue;
1116 }
1117 }
1123
1124 match task.hold_source {
1125 Some(HoldSource::Machine) => {
1126 let overridden = task
1127 .resume_override
1128 .as_ref()
1129 .is_some_and(|o| o.conductor_rehold.is_some() && !o.forced);
1130 if overridden {
1131 if file_question(questions, &task, Bucket::ConductorOverride, w, now).is_some()
1132 {
1133 report.asked.push(task.id.clone());
1134 }
1135 } else if cfg.as_ref().and_then(|c| machine_cause_resolved(&task, c)) == Some(true)
1136 {
1137 task.release();
1138 if queue.put(&mut task).is_ok() {
1139 report.resumed.push(task.id.clone());
1140 }
1141 } else if file_question(questions, &task, Bucket::MachineUnknown, w, now).is_some()
1142 {
1143 report.asked.push(task.id.clone());
1144 }
1145 }
1146 None => {
1147 if file_question(questions, &task, Bucket::Legacy, w, now).is_some() {
1148 report.asked.push(task.id.clone());
1149 }
1150 }
1151 Some(HoldSource::Manual) => {
1152 if manual_is_stale(&task, now)
1153 && file_question(questions, &task, Bucket::ManualStale, w, now).is_some()
1154 {
1155 report.asked.push(task.id.clone());
1156 }
1157 }
1158 }
1159 }
1160 report
1161}
1162
1163pub fn open_question_for(questions: &Questions, task_id: &str) -> Option<Question> {
1168 latest_triage_question(questions, task_id).filter(|q| q.status.open())
1169}
1170
1171pub fn pending_for(questions: &Questions, task: &Task) -> bool {
1185 let own = match latest_triage_question(questions, &task.id) {
1186 Some(q) if q.status.open() => true,
1187 Some(q) if q.status == QuestionStatus::Answered => !already_applied(task, &q),
1188 _ => false,
1189 };
1190 own || deps_pending(questions, task)
1191}
1192
1193fn deps_pending(questions: &Questions, task: &Task) -> bool {
1197 match latest_question(questions, DEPS_NODE, &task.id) {
1198 Some(q) if q.status.open() => true,
1199 Some(q) if q.status == QuestionStatus::Answered => !task.triage_applied(&q.id),
1200 _ => false,
1201 }
1202}
1203
1204pub fn open_task_ids(questions: &Questions) -> std::collections::BTreeSet<String> {
1209 questions
1210 .list()
1211 .into_iter()
1212 .filter(|q| (q.node == NODE || q.node == DEPS_NODE) && q.status.open())
1213 .map(|q| q.run)
1214 .collect()
1215}
1216
1217#[cfg(test)]
1218mod tests {
1219 use super::*;
1220 use crate::ask::Answer;
1221 use crate::queue::Source;
1222 use jiff::SignedDuration;
1223
1224 fn store() -> (tempfile::TempDir, Queue, Questions) {
1225 let dir = tempfile::tempdir().unwrap();
1226 let q = Queue::at(dir.path().join("queue"));
1227 let s = Questions::at(dir.path().join("questions"));
1228 (dir, q, s)
1229 }
1230
1231 fn task(title: &str, repo: PathBuf) -> Task {
1232 Task::new(title.to_owned(), format!("do {title}"), repo, Source::Human)
1233 }
1234
1235 fn gate_disabled_config(dir: &std::path::Path) -> PathBuf {
1239 let config = dir.join("magi.toml");
1240 std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
1241 config
1242 }
1243
1244 #[test]
1245 fn resumption_later_keep_held_answer_wins_but_an_older_answer_cannot_undo_release() {
1246 for source in ["cli", "web"] {
1247 for answer_after_release in [true, false] {
1248 let (dir, queue, questions) = store();
1249 let mut t = task("recovery", dir.path().join("repo"));
1250 t.hold_machine(Some("inspect remote first".into()));
1251 queue.put(&mut t).unwrap();
1252 let mut q = file_question(
1253 &questions,
1254 &t,
1255 Bucket::MachineUnknown,
1256 &JA,
1257 Timestamp::now(),
1258 )
1259 .unwrap();
1260 if !answer_after_release {
1261 q.answer(Answer::Choice(q.choices[1].clone())).unwrap();
1262 questions.put(&mut q).unwrap();
1263 }
1264 t.release_operator(source, None);
1265 queue.put(&mut t).unwrap();
1266 if answer_after_release {
1267 q.answer(Answer::Choice(q.choices[1].clone())).unwrap();
1268 questions.put(&mut q).unwrap();
1269 }
1270 apply_answers(&queue, &questions);
1271 let after = queue.get(&t.id).unwrap();
1272 assert!(after.triage_applied(&q.id));
1273 assert_eq!(after.operator_held(), answer_after_release);
1274 assert_eq!(
1275 after.status,
1276 if answer_after_release {
1277 TaskStatus::Held
1278 } else {
1279 TaskStatus::Queued
1280 }
1281 );
1282 assert!(apply_answers(&queue, &questions).is_empty());
1283 }
1284 }
1285 }
1286
1287 #[test]
1288 fn explicit_resumption_applies_legacy_queued_triage_once_and_preserves_later_holds() {
1289 let (dir, queue, questions) = store();
1290 let mut t = task("recover", dir.path().join("repo"));
1291 t.hold_machine(Some("inspect first".into()));
1292 queue.put(&mut t).unwrap();
1293 let mut q = file_question(
1294 &questions,
1295 &t,
1296 Bucket::MachineUnknown,
1297 &JA,
1298 Timestamp::now(),
1299 )
1300 .unwrap();
1301 assert!(q.actions.is_empty());
1302 q.answer(Answer::Choice(q.choices[0].clone())).unwrap();
1303 questions.put(&mut q).unwrap();
1304 t.release();
1306 t.instruction = "Reuse approved PR head".into();
1307 queue.put(&mut t).unwrap();
1308 assert_eq!(apply_answers(&queue, &questions), [t.id.clone()]);
1309 let mut after = queue.get(&t.id).unwrap();
1310 assert_eq!(
1311 after.resumption.as_ref().unwrap().question_id.as_deref(),
1312 Some(q.id.as_str())
1313 );
1314 assert_eq!(
1315 after.resumption.as_ref().unwrap().instruction,
1316 t.instruction
1317 );
1318 assert_eq!(after.answers[0].answer, JA.resume);
1319 assert!(after.triage_applied(&q.id));
1320 assert!(apply_answers(&Queue::at(dir.path().join("queue")), &questions).is_empty());
1321 after.hold_machine(Some("new blocker".into()));
1322 queue.put(&mut after).unwrap();
1323 let mut late = file_question(
1324 &questions,
1325 &after,
1326 Bucket::MachineUnknown,
1327 &EN,
1328 Timestamp::now(),
1329 )
1330 .unwrap();
1331 after.hold_manual(Some("keep held now".into()));
1332 queue.put(&mut after).unwrap();
1333 late.answer(Answer::Choice(late.choices[0].clone()))
1334 .unwrap();
1335 questions.put(&mut late).unwrap();
1336 apply_answers(&queue, &questions);
1337 let held = queue.get(&t.id).unwrap();
1338 assert!(held.operator_held());
1339 assert_eq!(held.hold_reason.as_deref(), Some("keep held now"));
1340 assert_eq!(held.resumption.as_ref().unwrap().generation, 1);
1341 }
1342
1343 #[test]
1344 fn a_resolved_machine_hold_is_requeued_automatically() {
1345 let (dir, q, questions) = store();
1346 let config = gate_disabled_config(dir.path());
1347 let mut t = task("disk pressure", dir.path().join("repo"));
1348 t.hold_machine(Some(
1349 "not enough free space to start a run: 10 bytes free, 100 required by \
1350 `[disk] min_free_bytes`"
1351 .to_owned(),
1352 ));
1353 q.put(&mut t).unwrap();
1354
1355 let report = run_once(&q, &questions, Some(&config), Timestamp::now());
1356 assert_eq!(report.resumed, [t.id.clone()]);
1357 assert!(report.asked.is_empty());
1358
1359 let back = q.get(&t.id).unwrap();
1360 assert_eq!(back.status, TaskStatus::Queued);
1361 assert!(back.hold_source.is_none());
1362 assert!(questions.list().is_empty(), "nothing needed asking");
1363 }
1364
1365 #[test]
1366 fn a_machine_hold_with_no_recognised_cause_gets_one_question_not_two() {
1367 let (dir, q, questions) = store();
1368 let mut t = task("gate went red", dir.path().join("repo"));
1369 t.hold_machine(Some("gate red".to_owned()));
1370 q.put(&mut t).unwrap();
1371
1372 let first = run_once(&q, &questions, None, Timestamp::now());
1373 assert_eq!(first.asked, [t.id.clone()]);
1374 assert!(first.resumed.is_empty());
1375
1376 let open: Vec<_> = questions
1377 .list()
1378 .into_iter()
1379 .filter(|q| q.status.open())
1380 .collect();
1381 assert_eq!(open.len(), 1);
1382 assert_eq!(open[0].run, t.id);
1383 assert_eq!(open[0].node, NODE);
1384 assert_eq!(open[0].choices.len(), 3);
1385
1386 let second = run_once(&q, &questions, None, Timestamp::now());
1388 assert!(second.asked.is_empty());
1389 assert_eq!(
1390 questions
1391 .list()
1392 .into_iter()
1393 .filter(|q| q.status.open())
1394 .count(),
1395 1
1396 );
1397 }
1398
1399 #[test]
1400 fn a_legacy_hold_with_no_recorded_source_gets_exactly_one_question() {
1401 let (dir, q, questions) = store();
1402 let mut t = task("schema 1 record", dir.path().join("repo"));
1403 t.status = TaskStatus::Held;
1404 assert!(t.hold_source.is_none(), "the case this test is about");
1405 q.put(&mut t).unwrap();
1406
1407 let first = run_once(&q, &questions, None, Timestamp::now());
1408 assert_eq!(first.asked, [t.id.clone()]);
1409
1410 let second = run_once(&q, &questions, None, Timestamp::now());
1411 assert!(
1412 second.asked.is_empty(),
1413 "the same legacy hold must not be asked about twice"
1414 );
1415 assert_eq!(
1416 questions
1417 .list()
1418 .into_iter()
1419 .filter(|q| q.status.open())
1420 .count(),
1421 1
1422 );
1423 }
1424
1425 #[test]
1426 fn dependency_resolution_detaches_legacy_short_refs_without_dropping_other_waits() {
1427 let (dir, queue, questions) = store();
1428 let mut root = task("root", dir.path().join("repo"));
1429 root.id = "20261013-193611-ca5b".into();
1430 root.hold_machine(Some("needs owner decision".into()));
1431 queue.put(&mut root).unwrap();
1432 let mut other = task("other", dir.path().join("repo"));
1433 other.id = "20261013-192213-feb2".into();
1434 queue.put(&mut other).unwrap();
1435 let mut waiting = task("waiting", dir.path().join("repo"));
1436 waiting.block(vec!["ca5b".into(), other.id.clone()], None);
1437 queue.put(&mut waiting).unwrap();
1438 detach_dependants(&queue, &questions, &root.id);
1439 let after = queue.get(&waiting.id).unwrap();
1440 assert_eq!(after.status, TaskStatus::Blocked);
1441 assert_eq!(after.blocked_by, [other.id]);
1442 }
1443
1444 #[test]
1445 fn a_manual_hold_is_never_auto_resumed() {
1446 let (dir, q, questions) = store();
1447 let mut t = task("operator stopped this", dir.path().join("repo"));
1448 t.hold_manual(Some("waiting on a decision".to_owned()));
1449 q.put(&mut t).unwrap();
1450
1451 let report = run_once(&q, &questions, None, Timestamp::now());
1452 assert!(report.resumed.is_empty());
1453 assert!(report.asked.is_empty());
1455
1456 let back = q.get(&t.id).unwrap();
1457 assert_eq!(back.status, TaskStatus::Held);
1458 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1459 assert!(questions.list().is_empty());
1460 }
1461
1462 #[test]
1463 fn a_blocked_task_on_a_deleted_dependency_is_held_and_asked_about_in_one_pass() {
1464 let (dir, q, questions) = store();
1469 let mut still_going = task("still valid", dir.path().join("repo"));
1470 q.put(&mut still_going).unwrap();
1471
1472 let mut t = task("orphaned", dir.path().join("repo"));
1473 t.block(
1474 vec!["20260101-000000-gone".to_owned(), still_going.id.clone()],
1475 Some("waits on both".to_owned()),
1476 );
1477 q.put(&mut t).unwrap();
1478
1479 let report = run_once(&q, &questions, None, Timestamp::now());
1480 assert_eq!(report.quarantined, [t.id.clone()]);
1481 assert_eq!(
1482 report.asked,
1483 [t.id.clone()],
1484 "the fresh machine hold must earn a question in the same pass"
1485 );
1486
1487 let after = q.get(&t.id).unwrap();
1488 assert_eq!(after.status, TaskStatus::Held);
1489 assert_eq!(after.hold_source, Some(HoldSource::Machine));
1490 assert!(after.blocked_by.is_empty());
1491
1492 assert!(!is_disk_hold(&after));
1495
1496 let open: Vec<_> = questions
1497 .list()
1498 .into_iter()
1499 .filter(|q| q.status.open())
1500 .collect();
1501 assert_eq!(open.len(), 1);
1502 assert_eq!(open[0].run, t.id);
1503
1504 let second = run_once(&q, &questions, None, Timestamp::now());
1506 assert!(second.quarantined.is_empty());
1507 assert!(second.asked.is_empty());
1508 }
1509
1510 fn stuck_chain(q: &Queue, dir: &std::path::Path) -> (Task, Task, Task) {
1513 let mut root = task("root", dir.join("repo"));
1514 root.hold_manual(Some("waiting".to_owned()));
1515 q.put(&mut root).unwrap();
1516 let mut mid = task("mid", dir.join("repo"));
1517 mid.block(vec![root.id.clone()], None);
1518 q.put(&mut mid).unwrap();
1519 let mut leaf = task("leaf", dir.join("repo"));
1520 leaf.block(vec![mid.id.clone()], None);
1521 q.put(&mut leaf).unwrap();
1522 (root, mid, leaf)
1523 }
1524
1525 fn deps_questions(questions: &Questions) -> Vec<Question> {
1526 questions
1527 .list()
1528 .into_iter()
1529 .filter(|q| q.node == DEPS_NODE)
1530 .collect()
1531 }
1532
1533 fn answer_deps(questions: &Questions, choice: usize) {
1534 let mut asked = deps_questions(questions).remove(0);
1535 let c = asked.choices[choice].clone();
1536 asked.answer(Answer::Choice(c)).unwrap();
1537 questions.put(&mut asked).unwrap();
1538 }
1539
1540 #[test]
1541 fn a_stuck_chain_earns_one_question_naming_every_dependant_and_is_not_reasked() {
1542 let (dir, q, questions) = store();
1543 let (root, mid, leaf) = stuck_chain(&q, dir.path());
1544
1545 let first = run_once(&q, &questions, None, Timestamp::now());
1546 assert_eq!(first.asked, std::slice::from_ref(&root.id));
1547 let asked = deps_questions(&questions);
1548 assert_eq!(asked.len(), 1, "one question per root, not per dependant");
1549 assert_eq!(asked[0].run, root.id);
1550 assert_eq!(asked[0].choices.len(), 3);
1551 assert!(asked[0].detail.contains(mid.short()), "{}", asked[0].detail);
1552 assert!(
1553 asked[0].detail.contains(leaf.short()),
1554 "{}",
1555 asked[0].detail
1556 );
1557
1558 let second = run_once(&q, &questions, None, Timestamp::now());
1559 assert!(second.asked.is_empty());
1560 assert_eq!(deps_questions(&questions).len(), 1);
1561 assert!(pending_for(&questions, &q.get(&root.id).unwrap()));
1562 }
1563
1564 #[test]
1565 fn a_stuck_root_with_a_machine_hold_gets_only_the_dependency_question() {
1566 let (dir, q, questions) = store();
1567 let mut root = task("root", dir.path().join("repo"));
1568 root.hold_machine(Some("gate red".to_owned()));
1569 q.put(&mut root).unwrap();
1570 let mut dep = task("dep", dir.path().join("repo"));
1571 dep.block(vec![root.id.clone()], None);
1572 q.put(&mut dep).unwrap();
1573
1574 run_once(&q, &questions, None, Timestamp::now());
1575 run_once(&q, &questions, None, Timestamp::now());
1576 assert_eq!(deps_questions(&questions).len(), 1);
1577 assert_eq!(
1578 questions.list().len(),
1579 1,
1580 "no second question about the hold"
1581 );
1582 }
1583
1584 #[test]
1585 fn a_dependency_that_can_still_run_asks_nothing() {
1586 let (dir, q, questions) = store();
1587 let mut live = task("live", dir.path().join("repo"));
1588 q.put(&mut live).unwrap();
1589 let mut dep = task("dep", dir.path().join("repo"));
1590 dep.block(vec![live.id.clone()], None);
1591 q.put(&mut dep).unwrap();
1592 let report = run_once(&q, &questions, None, Timestamp::now());
1593 assert!(report.asked.is_empty());
1594 assert!(questions.list().is_empty());
1595 }
1596
1597 #[test]
1598 fn answering_release_requeues_the_root_and_is_not_reasked() {
1599 let (dir, q, questions) = store();
1600 let (root, _mid, _leaf) = stuck_chain(&q, dir.path());
1601 run_once(&q, &questions, None, Timestamp::now());
1602 answer_deps(&questions, 0);
1603
1604 let report = run_once(&q, &questions, None, Timestamp::now());
1605 assert_eq!(report.answered, std::slice::from_ref(&root.id));
1606 assert!(report.asked.is_empty(), "applying must not re-ask");
1607 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Queued);
1608 let again = run_once(&q, &questions, None, Timestamp::now());
1609 assert!(again.asked.is_empty() && again.answered.is_empty());
1610 assert_eq!(deps_questions(&questions).len(), 1);
1611 }
1612
1613 #[test]
1614 fn answering_detach_frees_the_direct_dependant_and_keeps_the_root() {
1615 let (dir, q, questions) = store();
1616 let (root, mid, leaf) = stuck_chain(&q, dir.path());
1617 run_once(&q, &questions, None, Timestamp::now());
1618 answer_deps(&questions, 2);
1619
1620 let report = run_once(&q, &questions, None, Timestamp::now());
1621 assert!(report.asked.is_empty());
1622 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1623 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1624 let leaf = q.get(&leaf.id).unwrap();
1625 assert_eq!(leaf.status, TaskStatus::Blocked, "still waits on mid");
1626 assert_eq!(leaf.blocked_by, std::slice::from_ref(&mid.id));
1627 }
1628
1629 #[test]
1630 fn a_dependant_that_can_progress_through_another_blocker_is_not_stuck() {
1631 let (dir, q, questions) = store();
1632 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1633 let mut live = task("live", dir.path().join("repo"));
1634 q.put(&mut live).unwrap();
1635 let mut both = q.get(&mid.id).unwrap();
1636 both.block(vec![root.id.clone(), live.id.clone()], None);
1637 q.put(&mut both).unwrap();
1638 assert!(
1640 run_once(&q, &questions, None, Timestamp::now())
1641 .asked
1642 .is_empty()
1643 );
1644 }
1645
1646 #[test]
1647 fn answering_discard_detaches_then_removes_the_root() {
1648 let (dir, q, questions) = store();
1649 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1650 run_once(&q, &questions, None, Timestamp::now());
1651 answer_deps(&questions, 1);
1652
1653 let report = run_once(&q, &questions, None, Timestamp::now());
1654 assert!(q.get(&root.id).is_err());
1655 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1656 assert!(
1657 report.asked.is_empty(),
1658 "no per-dependant quarantine question"
1659 );
1660 assert!(report.quarantined.is_empty());
1661 }
1662
1663 #[test]
1664 fn an_unmatched_deps_answer_changes_nothing_and_is_not_reasked() {
1665 let (dir, q, questions) = store();
1666 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1667 run_once(&q, &questions, None, Timestamp::now());
1668 let mut asked = deps_questions(&questions).remove(0);
1669 asked.choices.clear();
1670 asked.answer(Answer::Text("dunno".to_owned())).unwrap();
1671 questions.put(&mut asked).unwrap();
1672
1673 for _ in 0..2 {
1674 let report = run_once(&q, &questions, None, Timestamp::now());
1675 assert!(report.asked.is_empty() && report.answered.is_empty());
1676 }
1677 assert_eq!(deps_questions(&questions).len(), 1);
1678 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1679 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Blocked);
1680 }
1681
1682 #[test]
1683 fn a_dependency_cycle_terminates_and_is_asked_about_once() {
1684 let (dir, q, questions) = store();
1685 let mut a = task("a", dir.path().join("repo"));
1686 let mut b = task("b", dir.path().join("repo"));
1687 a.block(vec![b.id.clone()], None);
1688 b.block(vec![a.id.clone()], None);
1689 q.put(&mut a).unwrap();
1690 q.put(&mut b).unwrap();
1691
1692 let first = run_once(&q, &questions, None, Timestamp::now());
1693 assert_eq!(first.asked.len(), 1);
1694 let second = run_once(&q, &questions, None, Timestamp::now());
1695 assert!(second.asked.is_empty());
1696 assert_eq!(deps_questions(&questions).len(), 1);
1697 }
1698
1699 #[test]
1700 fn a_missing_dependency_is_still_quarantined_not_asked_about_as_stuck() {
1701 let (dir, q, questions) = store();
1702 let mut t = task("orphan", dir.path().join("repo"));
1703 t.block(vec!["20260101-000000-gone".to_owned()], None);
1704 q.put(&mut t).unwrap();
1705 let report = run_once(&q, &questions, None, Timestamp::now());
1706 assert_eq!(report.quarantined, [t.id.clone()]);
1707 assert!(deps_questions(&questions).is_empty());
1708 }
1709
1710 #[test]
1711 fn a_stale_manual_hold_earns_a_two_choice_question() {
1712 let (dir, q, questions) = store();
1713 let mut t = task("been sitting a while", dir.path().join("repo"));
1714 t.hold_manual(Some("waiting on a decision".to_owned()));
1715 q.put(&mut t).unwrap();
1716 let mut back = q.get(&t.id).unwrap();
1718 back.updated_at = Timestamp::now() - SignedDuration::new(8 * 24 * 60 * 60, 0);
1719 std::fs::write(
1720 q.path_of(&back.id),
1721 serde_json::to_string_pretty(&back).unwrap(),
1722 )
1723 .unwrap();
1724
1725 let report = run_once(&q, &questions, None, Timestamp::now());
1726 assert_eq!(report.asked, [t.id.clone()]);
1727 let open: Vec<_> = questions
1728 .list()
1729 .into_iter()
1730 .filter(|q| q.status.open())
1731 .collect();
1732 assert_eq!(open.len(), 1);
1733 assert_eq!(open[0].choices.len(), 2);
1734 }
1735
1736 #[test]
1737 fn answering_resume_releases_the_task() {
1738 let (dir, q, questions) = store();
1739 let mut t = task("gate went red", dir.path().join("repo"));
1740 t.hold_machine(Some("gate red".to_owned()));
1741 q.put(&mut t).unwrap();
1742 run_once(&q, &questions, None, Timestamp::now());
1743
1744 let mut asked = questions
1745 .list()
1746 .into_iter()
1747 .find(|q| q.run == t.id)
1748 .unwrap();
1749 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1750 questions.put(&mut asked).unwrap();
1751
1752 let report = run_once(&q, &questions, None, Timestamp::now());
1753 assert_eq!(report.answered, [t.id.clone()]);
1754 let back = q.get(&t.id).unwrap();
1755 assert_eq!(back.status, TaskStatus::Queued);
1756 assert!(back.hold_source.is_none());
1757 }
1758
1759 fn resumed_then_failed(q: &Queue, questions: &Questions, dir: &std::path::Path) -> Task {
1761 let mut t = task("gate went red", dir.join("repo"));
1762 t.hold_machine(Some("gate red".to_owned()));
1763 q.put(&mut t).unwrap();
1764 run_once(q, questions, None, Timestamp::now());
1765 let mut asked = questions
1766 .list()
1767 .into_iter()
1768 .find(|q| q.run == t.id)
1769 .unwrap();
1770 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1771 questions.put(&mut asked).unwrap();
1772
1773 let report = run_once(q, questions, None, Timestamp::now());
1774 assert_eq!(report.answered, [t.id.clone()]);
1775 let mut back = q.get(&t.id).unwrap();
1776 assert_eq!(back.status, TaskStatus::Queued);
1777 assert_eq!(back.attempts, 0);
1778
1779 back.start("run-1".to_owned());
1780 back.fail("rebase conflict", 1);
1781 assert_eq!(back.status, TaskStatus::Held);
1782 q.put(&mut back).unwrap();
1783 back
1784 }
1785
1786 #[test]
1787 fn a_resume_answer_is_applied_once_and_a_new_machine_hold_is_asked_about() {
1788 let (dir, q, questions) = store();
1789 let t = resumed_then_failed(&q, &questions, dir.path());
1790
1791 let report = run_once(&q, &questions, None, Timestamp::now());
1792 assert!(report.answered.is_empty(), "the old answer must not replay");
1793 assert_eq!(report.asked, std::slice::from_ref(&t.id));
1794 let back = q.get(&t.id).unwrap();
1795 assert_eq!(back.status, TaskStatus::Held);
1796 assert_eq!(
1797 questions
1798 .list()
1799 .into_iter()
1800 .filter(|q| q.status.open())
1801 .count(),
1802 1
1803 );
1804 }
1805
1806 #[test]
1807 fn a_manual_hold_placed_after_a_resume_is_not_undone_by_the_old_answer() {
1808 let (dir, q, questions) = store();
1809 let mut t = resumed_then_failed(&q, &questions, dir.path());
1810 t.hold_manual(Some("operator stopped this".to_owned()));
1811 q.put(&mut t).unwrap();
1812 let before = questions.list().len();
1813
1814 let report = run_once(&q, &questions, None, Timestamp::now());
1815 assert!(report.answered.is_empty());
1816 assert!(report.asked.is_empty());
1817 let back = q.get(&t.id).unwrap();
1818 assert_eq!(back.status, TaskStatus::Held);
1819 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1820 assert_eq!(questions.list().len(), before);
1821 }
1822
1823 #[test]
1824 fn answering_not_yet_keeps_it_held_as_a_manual_hold_and_does_not_reapply() {
1825 let (dir, q, questions) = store();
1826 let mut t = task("gate went red", dir.path().join("repo"));
1827 t.hold_machine(Some("gate red".to_owned()));
1828 q.put(&mut t).unwrap();
1829 run_once(&q, &questions, None, Timestamp::now());
1830
1831 let mut asked = questions
1832 .list()
1833 .into_iter()
1834 .find(|q| q.run == t.id)
1835 .unwrap();
1836 asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
1837 questions.put(&mut asked).unwrap();
1838
1839 let report = run_once(&q, &questions, None, Timestamp::now());
1840 assert_eq!(report.answered, [t.id.clone()]);
1841 let back = q.get(&t.id).unwrap();
1842 assert_eq!(back.status, TaskStatus::Held);
1843 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1844 assert!(
1845 back.hold_reason
1846 .as_deref()
1847 .is_some_and(|r| r.contains("gate red")),
1848 "the original cause must survive a \"not yet\" answer, not just the \
1849 triage marker: {:?}",
1850 back.hold_reason
1851 );
1852
1853 let third = run_once(&q, &questions, None, Timestamp::now());
1857 assert!(third.answered.is_empty());
1858 assert!(third.asked.is_empty());
1859 }
1860
1861 #[test]
1862 fn answering_discard_removes_the_task_entirely() {
1863 let (dir, q, questions) = store();
1864 let mut t = task("gate went red", dir.path().join("repo"));
1865 t.hold_machine(Some("gate red".to_owned()));
1866 q.put(&mut t).unwrap();
1867 run_once(&q, &questions, None, Timestamp::now());
1868
1869 let mut asked = questions
1870 .list()
1871 .into_iter()
1872 .find(|q| q.run == t.id)
1873 .unwrap();
1874 asked.answer(Answer::Choice(EN.discard.to_owned())).unwrap();
1875 questions.put(&mut asked).unwrap();
1876
1877 let report = run_once(&q, &questions, None, Timestamp::now());
1878 assert_eq!(report.answered, [t.id.clone()]);
1879 assert!(
1880 q.get(&t.id).is_err(),
1881 "\"discard it\" (捨ててよい) must actually discard the task, not \
1882 just leave it sitting held forever"
1883 );
1884 }
1885
1886 #[test]
1887 fn an_answer_is_read_by_its_position_in_choices_not_by_the_callers_current_language() {
1888 let (dir, q, questions) = store();
1890 let ja_config = dir.path().join("ja.toml");
1891 std::fs::write(&ja_config, "[graph]\nlanguage = \"ja\"\n").unwrap();
1892 let mut t = task("gate went red", dir.path().join("repo"));
1893 t.hold_machine(Some("gate red".to_owned()));
1894 q.put(&mut t).unwrap();
1895 run_once(&q, &questions, Some(&ja_config), Timestamp::now());
1896
1897 let mut asked = questions
1898 .list()
1899 .into_iter()
1900 .find(|q| q.run == t.id)
1901 .unwrap();
1902 assert_eq!(asked.choices[0], JA.resume, "filed in Japanese");
1903 asked.answer(Answer::Choice(JA.resume.to_owned())).unwrap();
1904 questions.put(&mut asked).unwrap();
1905
1906 let en_config = dir.path().join("en.toml");
1911 std::fs::write(&en_config, "[graph]\nlanguage = \"en\"\n").unwrap();
1912 let report = run_once(&q, &questions, Some(&en_config), Timestamp::now());
1913 assert_eq!(report.answered, [t.id.clone()]);
1914 let back = q.get(&t.id).unwrap();
1915 assert_eq!(
1916 back.status,
1917 TaskStatus::Queued,
1918 "a resume answer must resume the task regardless of which \
1919 language it is read back in"
1920 );
1921 }
1922
1923 #[test]
1924 fn a_question_falls_back_to_last_error_when_hold_reason_was_never_set() {
1925 let (dir, q, questions) = store();
1930 let mut t = task("kept failing the gate", dir.path().join("repo"));
1931 t.start("run-1".to_owned());
1932 t.fail("gate red three times running", 1);
1933 assert_eq!(t.status, TaskStatus::Held);
1934 t.hold_reason = None; q.put(&mut t).unwrap();
1936
1937 run_once(&q, &questions, None, Timestamp::now());
1938 let asked = questions
1939 .list()
1940 .into_iter()
1941 .find(|q| q.run == t.id)
1942 .unwrap();
1943 assert!(
1944 asked.detail.contains("gate red three times running"),
1945 "the question must surface `last_error` when there is no \
1946 `hold_reason` to show instead: {}",
1947 asked.detail
1948 );
1949 }
1950
1951 #[test]
1952 fn open_question_for_and_open_task_ids_reflect_only_what_is_still_waiting() {
1953 let (dir, q, questions) = store();
1954 let mut t = task("gate went red", dir.path().join("repo"));
1955 t.hold_machine(Some("gate red".to_owned()));
1956 q.put(&mut t).unwrap();
1957
1958 assert!(open_question_for(&questions, &t.id).is_none());
1959 assert!(!open_task_ids(&questions).contains(&t.id));
1960
1961 run_once(&q, &questions, None, Timestamp::now());
1962 assert!(open_question_for(&questions, &t.id).is_some());
1963 assert!(open_task_ids(&questions).contains(&t.id));
1964
1965 let mut asked = questions
1966 .list()
1967 .into_iter()
1968 .find(|q| q.run == t.id)
1969 .unwrap();
1970 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1971 questions.put(&mut asked).unwrap();
1972
1973 assert!(
1974 open_question_for(&questions, &t.id).is_none(),
1975 "an answered question is no longer open"
1976 );
1977 assert!(!open_task_ids(&questions).contains(&t.id));
1978 }
1979
1980 #[test]
1981 fn pending_for_stays_true_between_an_answer_and_the_next_run_once_pass() {
1982 let (dir, q, questions) = store();
1988 let mut t = task("gate went red", dir.path().join("repo"));
1989 t.hold_machine(Some("gate red".to_owned()));
1990 q.put(&mut t).unwrap();
1991
1992 assert!(!pending_for(&questions, &t));
1993
1994 run_once(&q, &questions, None, Timestamp::now());
1995 let held = q.get(&t.id).unwrap();
1996 assert!(pending_for(&questions, &held), "still waiting on an answer");
1997
1998 let mut asked = questions
1999 .list()
2000 .into_iter()
2001 .find(|q| q.run == t.id)
2002 .unwrap();
2003 asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
2004 questions.put(&mut asked).unwrap();
2005 assert!(!asked.status.open());
2006
2007 let still_held = q.get(&t.id).unwrap();
2009 assert!(
2010 pending_for(&questions, &still_held),
2011 "answered but not yet applied is still pending"
2012 );
2013
2014 run_once(&q, &questions, None, Timestamp::now());
2015 let after = q.get(&t.id).unwrap();
2016 assert!(
2017 !pending_for(&questions, &after),
2018 "the answer is applied now, nothing left pending"
2019 );
2020 }
2021}