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