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() || awaiting_task_chat(q, root) => 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() || awaiting_task_chat(&q, &task) {
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() || awaiting_task_chat(&q, task) => 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 awaiting_task_chat(q: &Question, task: &Task) -> bool {
1194 q.status != QuestionStatus::Abandoned
1195 && q.routed_chat.is_some()
1196 && !already_applied(task, q)
1197 && task.resumption.as_ref().map_or(0, |r| r.generation) == q.triage_generation.unwrap_or(0)
1198 && task.operator_decided_at.is_none_or(|at| at <= q.asked_at)
1199}
1200
1201fn deps_pending(questions: &Questions, task: &Task) -> bool {
1205 match latest_question(questions, DEPS_NODE, &task.id) {
1206 Some(q) if q.status.open() || awaiting_task_chat(&q, task) => true,
1207 Some(q) if q.status == QuestionStatus::Answered => !task.triage_applied(&q.id),
1208 _ => false,
1209 }
1210}
1211
1212pub fn open_task_ids(questions: &Questions) -> std::collections::BTreeSet<String> {
1217 questions
1218 .list()
1219 .into_iter()
1220 .filter(|q| (q.node == NODE || q.node == DEPS_NODE) && q.status.open())
1221 .map(|q| q.run)
1222 .collect()
1223}
1224
1225#[cfg(test)]
1226mod tests {
1227 use super::*;
1228 use crate::ask::Answer;
1229 use crate::queue::Source;
1230 use jiff::SignedDuration;
1231
1232 fn store() -> (tempfile::TempDir, Queue, Questions) {
1233 let dir = tempfile::tempdir().unwrap();
1234 let q = Queue::at(dir.path().join("queue"));
1235 let s = Questions::at(dir.path().join("questions"));
1236 (dir, q, s)
1237 }
1238
1239 fn task(title: &str, repo: PathBuf) -> Task {
1240 Task::new(title.to_owned(), format!("do {title}"), repo, Source::Human)
1241 }
1242
1243 fn gate_disabled_config(dir: &std::path::Path) -> PathBuf {
1247 let config = dir.join("magi.toml");
1248 std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
1249 config
1250 }
1251
1252 #[test]
1253 fn resumption_later_keep_held_answer_wins_but_an_older_answer_cannot_undo_release() {
1254 for source in ["cli", "web"] {
1255 for answer_after_release in [true, false] {
1256 let (dir, queue, questions) = store();
1257 let mut t = task("recovery", dir.path().join("repo"));
1258 t.hold_machine(Some("inspect remote first".into()));
1259 queue.put(&mut t).unwrap();
1260 let mut q = file_question(
1261 &questions,
1262 &t,
1263 Bucket::MachineUnknown,
1264 &JA,
1265 Timestamp::now(),
1266 )
1267 .unwrap();
1268 if !answer_after_release {
1269 q.answer(Answer::Choice(q.choices[1].clone())).unwrap();
1270 questions.put(&mut q).unwrap();
1271 }
1272 t.release_operator(source, None);
1273 queue.put(&mut t).unwrap();
1274 if answer_after_release {
1275 q.answer(Answer::Choice(q.choices[1].clone())).unwrap();
1276 questions.put(&mut q).unwrap();
1277 }
1278 apply_answers(&queue, &questions);
1279 let after = queue.get(&t.id).unwrap();
1280 assert!(after.triage_applied(&q.id));
1281 assert_eq!(after.operator_held(), answer_after_release);
1282 assert_eq!(
1283 after.status,
1284 if answer_after_release {
1285 TaskStatus::Held
1286 } else {
1287 TaskStatus::Queued
1288 }
1289 );
1290 assert!(apply_answers(&queue, &questions).is_empty());
1291 }
1292 }
1293 }
1294
1295 #[test]
1296 fn explicit_resumption_applies_legacy_queued_triage_once_and_preserves_later_holds() {
1297 let (dir, queue, questions) = store();
1298 let mut t = task("recover", dir.path().join("repo"));
1299 t.hold_machine(Some("inspect first".into()));
1300 queue.put(&mut t).unwrap();
1301 let mut q = file_question(
1302 &questions,
1303 &t,
1304 Bucket::MachineUnknown,
1305 &JA,
1306 Timestamp::now(),
1307 )
1308 .unwrap();
1309 assert!(q.actions.is_empty());
1310 q.answer(Answer::Choice(q.choices[0].clone())).unwrap();
1311 questions.put(&mut q).unwrap();
1312 t.release();
1314 t.instruction = "Reuse approved PR head".into();
1315 queue.put(&mut t).unwrap();
1316 assert_eq!(apply_answers(&queue, &questions), [t.id.clone()]);
1317 let mut after = queue.get(&t.id).unwrap();
1318 assert_eq!(
1319 after.resumption.as_ref().unwrap().question_id.as_deref(),
1320 Some(q.id.as_str())
1321 );
1322 assert_eq!(
1323 after.resumption.as_ref().unwrap().instruction,
1324 t.instruction
1325 );
1326 assert_eq!(after.answers[0].answer, JA.resume);
1327 assert!(after.triage_applied(&q.id));
1328 assert!(apply_answers(&Queue::at(dir.path().join("queue")), &questions).is_empty());
1329 after.hold_machine(Some("new blocker".into()));
1330 queue.put(&mut after).unwrap();
1331 let mut late = file_question(
1332 &questions,
1333 &after,
1334 Bucket::MachineUnknown,
1335 &EN,
1336 Timestamp::now(),
1337 )
1338 .unwrap();
1339 after.hold_manual(Some("keep held now".into()));
1340 queue.put(&mut after).unwrap();
1341 late.answer(Answer::Choice(late.choices[0].clone()))
1342 .unwrap();
1343 questions.put(&mut late).unwrap();
1344 apply_answers(&queue, &questions);
1345 let held = queue.get(&t.id).unwrap();
1346 assert!(held.operator_held());
1347 assert_eq!(held.hold_reason.as_deref(), Some("keep held now"));
1348 assert_eq!(held.resumption.as_ref().unwrap().generation, 1);
1349 }
1350
1351 #[test]
1352 fn a_resolved_machine_hold_is_requeued_automatically() {
1353 let (dir, q, questions) = store();
1354 let config = gate_disabled_config(dir.path());
1355 let mut t = task("disk pressure", dir.path().join("repo"));
1356 t.hold_machine(Some(
1357 "not enough free space to start a run: 10 bytes free, 100 required by \
1358 `[disk] min_free_bytes`"
1359 .to_owned(),
1360 ));
1361 q.put(&mut t).unwrap();
1362
1363 let report = run_once(&q, &questions, Some(&config), Timestamp::now());
1364 assert_eq!(report.resumed, [t.id.clone()]);
1365 assert!(report.asked.is_empty());
1366
1367 let back = q.get(&t.id).unwrap();
1368 assert_eq!(back.status, TaskStatus::Queued);
1369 assert!(back.hold_source.is_none());
1370 assert!(questions.list().is_empty(), "nothing needed asking");
1371 }
1372
1373 #[test]
1374 fn a_machine_hold_with_no_recognised_cause_gets_one_question_not_two() {
1375 let (dir, q, questions) = store();
1376 let mut t = task("gate went red", dir.path().join("repo"));
1377 t.hold_machine(Some("gate red".to_owned()));
1378 q.put(&mut t).unwrap();
1379
1380 let first = run_once(&q, &questions, None, Timestamp::now());
1381 assert_eq!(first.asked, [t.id.clone()]);
1382 assert!(first.resumed.is_empty());
1383
1384 let open: Vec<_> = questions
1385 .list()
1386 .into_iter()
1387 .filter(|q| q.status.open())
1388 .collect();
1389 assert_eq!(open.len(), 1);
1390 assert_eq!(open[0].run, t.id);
1391 assert_eq!(open[0].node, NODE);
1392 assert_eq!(open[0].choices.len(), 3);
1393
1394 let second = run_once(&q, &questions, None, Timestamp::now());
1396 assert!(second.asked.is_empty());
1397 assert_eq!(
1398 questions
1399 .list()
1400 .into_iter()
1401 .filter(|q| q.status.open())
1402 .count(),
1403 1
1404 );
1405 }
1406
1407 #[test]
1408 fn a_legacy_hold_with_no_recorded_source_gets_exactly_one_question() {
1409 let (dir, q, questions) = store();
1410 let mut t = task("schema 1 record", dir.path().join("repo"));
1411 t.status = TaskStatus::Held;
1412 assert!(t.hold_source.is_none(), "the case this test is about");
1413 q.put(&mut t).unwrap();
1414
1415 let first = run_once(&q, &questions, None, Timestamp::now());
1416 assert_eq!(first.asked, [t.id.clone()]);
1417
1418 let second = run_once(&q, &questions, None, Timestamp::now());
1419 assert!(
1420 second.asked.is_empty(),
1421 "the same legacy hold must not be asked about twice"
1422 );
1423 assert_eq!(
1424 questions
1425 .list()
1426 .into_iter()
1427 .filter(|q| q.status.open())
1428 .count(),
1429 1
1430 );
1431 }
1432
1433 #[test]
1434 fn dependency_resolution_detaches_legacy_short_refs_without_dropping_other_waits() {
1435 let (dir, queue, questions) = store();
1436 let mut root = task("root", dir.path().join("repo"));
1437 root.id = "20261013-193611-ca5b".into();
1438 root.hold_machine(Some("needs owner decision".into()));
1439 queue.put(&mut root).unwrap();
1440 let mut other = task("other", dir.path().join("repo"));
1441 other.id = "20261013-192213-feb2".into();
1442 queue.put(&mut other).unwrap();
1443 let mut waiting = task("waiting", dir.path().join("repo"));
1444 waiting.block(vec!["ca5b".into(), other.id.clone()], None);
1445 queue.put(&mut waiting).unwrap();
1446 detach_dependants(&queue, &questions, &root.id);
1447 let after = queue.get(&waiting.id).unwrap();
1448 assert_eq!(after.status, TaskStatus::Blocked);
1449 assert_eq!(after.blocked_by, [other.id]);
1450 }
1451
1452 #[test]
1453 fn a_manual_hold_is_never_auto_resumed() {
1454 let (dir, q, questions) = store();
1455 let mut t = task("operator stopped this", dir.path().join("repo"));
1456 t.hold_manual(Some("waiting on a decision".to_owned()));
1457 q.put(&mut t).unwrap();
1458
1459 let report = run_once(&q, &questions, None, Timestamp::now());
1460 assert!(report.resumed.is_empty());
1461 assert!(report.asked.is_empty());
1463
1464 let back = q.get(&t.id).unwrap();
1465 assert_eq!(back.status, TaskStatus::Held);
1466 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1467 assert!(questions.list().is_empty());
1468 }
1469
1470 #[test]
1471 fn a_blocked_task_on_a_deleted_dependency_is_held_and_asked_about_in_one_pass() {
1472 let (dir, q, questions) = store();
1477 let mut still_going = task("still valid", dir.path().join("repo"));
1478 q.put(&mut still_going).unwrap();
1479
1480 let mut t = task("orphaned", dir.path().join("repo"));
1481 t.block(
1482 vec!["20260101-000000-gone".to_owned(), still_going.id.clone()],
1483 Some("waits on both".to_owned()),
1484 );
1485 q.put(&mut t).unwrap();
1486
1487 let report = run_once(&q, &questions, None, Timestamp::now());
1488 assert_eq!(report.quarantined, [t.id.clone()]);
1489 assert_eq!(
1490 report.asked,
1491 [t.id.clone()],
1492 "the fresh machine hold must earn a question in the same pass"
1493 );
1494
1495 let after = q.get(&t.id).unwrap();
1496 assert_eq!(after.status, TaskStatus::Held);
1497 assert_eq!(after.hold_source, Some(HoldSource::Machine));
1498 assert!(after.blocked_by.is_empty());
1499
1500 assert!(!is_disk_hold(&after));
1503
1504 let open: Vec<_> = questions
1505 .list()
1506 .into_iter()
1507 .filter(|q| q.status.open())
1508 .collect();
1509 assert_eq!(open.len(), 1);
1510 assert_eq!(open[0].run, t.id);
1511
1512 let second = run_once(&q, &questions, None, Timestamp::now());
1514 assert!(second.quarantined.is_empty());
1515 assert!(second.asked.is_empty());
1516 }
1517
1518 fn stuck_chain(q: &Queue, dir: &std::path::Path) -> (Task, Task, Task) {
1521 let mut root = task("root", dir.join("repo"));
1522 root.hold_manual(Some("waiting".to_owned()));
1523 q.put(&mut root).unwrap();
1524 let mut mid = task("mid", dir.join("repo"));
1525 mid.block(vec![root.id.clone()], None);
1526 q.put(&mut mid).unwrap();
1527 let mut leaf = task("leaf", dir.join("repo"));
1528 leaf.block(vec![mid.id.clone()], None);
1529 q.put(&mut leaf).unwrap();
1530 (root, mid, leaf)
1531 }
1532
1533 fn deps_questions(questions: &Questions) -> Vec<Question> {
1534 questions
1535 .list()
1536 .into_iter()
1537 .filter(|q| q.node == DEPS_NODE)
1538 .collect()
1539 }
1540
1541 fn answer_deps(questions: &Questions, choice: usize) {
1542 let mut asked = deps_questions(questions).remove(0);
1543 let c = asked.choices[choice].clone();
1544 asked.answer(Answer::Choice(c)).unwrap();
1545 questions.put(&mut asked).unwrap();
1546 }
1547
1548 #[test]
1549 fn a_stuck_chain_earns_one_question_naming_every_dependant_and_is_not_reasked() {
1550 let (dir, q, questions) = store();
1551 let (root, mid, leaf) = stuck_chain(&q, dir.path());
1552
1553 let first = run_once(&q, &questions, None, Timestamp::now());
1554 assert_eq!(first.asked, std::slice::from_ref(&root.id));
1555 let asked = deps_questions(&questions);
1556 assert_eq!(asked.len(), 1, "one question per root, not per dependant");
1557 assert_eq!(asked[0].run, root.id);
1558 assert_eq!(asked[0].choices.len(), 3);
1559 assert!(asked[0].detail.contains(mid.short()), "{}", asked[0].detail);
1560 assert!(
1561 asked[0].detail.contains(leaf.short()),
1562 "{}",
1563 asked[0].detail
1564 );
1565
1566 let second = run_once(&q, &questions, None, Timestamp::now());
1567 assert!(second.asked.is_empty());
1568 assert_eq!(deps_questions(&questions).len(), 1);
1569 assert!(pending_for(&questions, &q.get(&root.id).unwrap()));
1570 }
1571
1572 #[test]
1573 fn a_stuck_root_with_a_machine_hold_gets_only_the_dependency_question() {
1574 let (dir, q, questions) = store();
1575 let mut root = task("root", dir.path().join("repo"));
1576 root.hold_machine(Some("gate red".to_owned()));
1577 q.put(&mut root).unwrap();
1578 let mut dep = task("dep", dir.path().join("repo"));
1579 dep.block(vec![root.id.clone()], None);
1580 q.put(&mut dep).unwrap();
1581
1582 run_once(&q, &questions, None, Timestamp::now());
1583 run_once(&q, &questions, None, Timestamp::now());
1584 assert_eq!(deps_questions(&questions).len(), 1);
1585 assert_eq!(
1586 questions.list().len(),
1587 1,
1588 "no second question about the hold"
1589 );
1590 }
1591
1592 #[test]
1593 fn a_dependency_that_can_still_run_asks_nothing() {
1594 let (dir, q, questions) = store();
1595 let mut live = task("live", dir.path().join("repo"));
1596 q.put(&mut live).unwrap();
1597 let mut dep = task("dep", dir.path().join("repo"));
1598 dep.block(vec![live.id.clone()], None);
1599 q.put(&mut dep).unwrap();
1600 let report = run_once(&q, &questions, None, Timestamp::now());
1601 assert!(report.asked.is_empty());
1602 assert!(questions.list().is_empty());
1603 }
1604
1605 #[test]
1606 fn answering_release_requeues_the_root_and_is_not_reasked() {
1607 let (dir, q, questions) = store();
1608 let (root, _mid, _leaf) = stuck_chain(&q, dir.path());
1609 run_once(&q, &questions, None, Timestamp::now());
1610 answer_deps(&questions, 0);
1611
1612 let report = run_once(&q, &questions, None, Timestamp::now());
1613 assert_eq!(report.answered, std::slice::from_ref(&root.id));
1614 assert!(report.asked.is_empty(), "applying must not re-ask");
1615 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Queued);
1616 let again = run_once(&q, &questions, None, Timestamp::now());
1617 assert!(again.asked.is_empty() && again.answered.is_empty());
1618 assert_eq!(deps_questions(&questions).len(), 1);
1619 }
1620
1621 #[test]
1622 fn answering_detach_frees_the_direct_dependant_and_keeps_the_root() {
1623 let (dir, q, questions) = store();
1624 let (root, mid, leaf) = stuck_chain(&q, dir.path());
1625 run_once(&q, &questions, None, Timestamp::now());
1626 answer_deps(&questions, 2);
1627
1628 let report = run_once(&q, &questions, None, Timestamp::now());
1629 assert!(report.asked.is_empty());
1630 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1631 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1632 let leaf = q.get(&leaf.id).unwrap();
1633 assert_eq!(leaf.status, TaskStatus::Blocked, "still waits on mid");
1634 assert_eq!(leaf.blocked_by, std::slice::from_ref(&mid.id));
1635 }
1636
1637 #[test]
1638 fn a_dependant_that_can_progress_through_another_blocker_is_not_stuck() {
1639 let (dir, q, questions) = store();
1640 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1641 let mut live = task("live", dir.path().join("repo"));
1642 q.put(&mut live).unwrap();
1643 let mut both = q.get(&mid.id).unwrap();
1644 both.block(vec![root.id.clone(), live.id.clone()], None);
1645 q.put(&mut both).unwrap();
1646 assert!(
1648 run_once(&q, &questions, None, Timestamp::now())
1649 .asked
1650 .is_empty()
1651 );
1652 }
1653
1654 #[test]
1655 fn answering_discard_detaches_then_removes_the_root() {
1656 let (dir, q, questions) = store();
1657 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1658 run_once(&q, &questions, None, Timestamp::now());
1659 answer_deps(&questions, 1);
1660
1661 let report = run_once(&q, &questions, None, Timestamp::now());
1662 assert!(q.get(&root.id).is_err());
1663 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1664 assert!(
1665 report.asked.is_empty(),
1666 "no per-dependant quarantine question"
1667 );
1668 assert!(report.quarantined.is_empty());
1669 }
1670
1671 #[test]
1672 fn an_unmatched_deps_answer_changes_nothing_and_is_not_reasked() {
1673 let (dir, q, questions) = store();
1674 let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1675 run_once(&q, &questions, None, Timestamp::now());
1676 let mut asked = deps_questions(&questions).remove(0);
1677 asked.choices.clear();
1678 asked.answer(Answer::Text("dunno".to_owned())).unwrap();
1679 questions.put(&mut asked).unwrap();
1680
1681 for _ in 0..2 {
1682 let report = run_once(&q, &questions, None, Timestamp::now());
1683 assert!(report.asked.is_empty() && report.answered.is_empty());
1684 }
1685 assert_eq!(deps_questions(&questions).len(), 1);
1686 assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1687 assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Blocked);
1688 }
1689
1690 #[test]
1691 fn a_dependency_cycle_terminates_and_is_asked_about_once() {
1692 let (dir, q, questions) = store();
1693 let mut a = task("a", dir.path().join("repo"));
1694 let mut b = task("b", dir.path().join("repo"));
1695 a.block(vec![b.id.clone()], None);
1696 b.block(vec![a.id.clone()], None);
1697 q.put(&mut a).unwrap();
1698 q.put(&mut b).unwrap();
1699
1700 let first = run_once(&q, &questions, None, Timestamp::now());
1701 assert_eq!(first.asked.len(), 1);
1702 let second = run_once(&q, &questions, None, Timestamp::now());
1703 assert!(second.asked.is_empty());
1704 assert_eq!(deps_questions(&questions).len(), 1);
1705 }
1706
1707 #[test]
1708 fn a_missing_dependency_is_still_quarantined_not_asked_about_as_stuck() {
1709 let (dir, q, questions) = store();
1710 let mut t = task("orphan", dir.path().join("repo"));
1711 t.block(vec!["20260101-000000-gone".to_owned()], None);
1712 q.put(&mut t).unwrap();
1713 let report = run_once(&q, &questions, None, Timestamp::now());
1714 assert_eq!(report.quarantined, [t.id.clone()]);
1715 assert!(deps_questions(&questions).is_empty());
1716 }
1717
1718 #[test]
1719 fn a_stale_manual_hold_earns_a_two_choice_question() {
1720 let (dir, q, questions) = store();
1721 let mut t = task("been sitting a while", dir.path().join("repo"));
1722 t.hold_manual(Some("waiting on a decision".to_owned()));
1723 q.put(&mut t).unwrap();
1724 let mut back = q.get(&t.id).unwrap();
1726 back.updated_at = Timestamp::now() - SignedDuration::new(8 * 24 * 60 * 60, 0);
1727 std::fs::write(
1728 q.path_of(&back.id),
1729 serde_json::to_string_pretty(&back).unwrap(),
1730 )
1731 .unwrap();
1732
1733 let report = run_once(&q, &questions, None, Timestamp::now());
1734 assert_eq!(report.asked, [t.id.clone()]);
1735 let open: Vec<_> = questions
1736 .list()
1737 .into_iter()
1738 .filter(|q| q.status.open())
1739 .collect();
1740 assert_eq!(open.len(), 1);
1741 assert_eq!(open[0].choices.len(), 2);
1742 }
1743
1744 #[test]
1745 fn answering_resume_releases_the_task() {
1746 let (dir, q, questions) = store();
1747 let mut t = task("gate went red", dir.path().join("repo"));
1748 t.hold_machine(Some("gate red".to_owned()));
1749 q.put(&mut t).unwrap();
1750 run_once(&q, &questions, None, Timestamp::now());
1751
1752 let mut asked = questions
1753 .list()
1754 .into_iter()
1755 .find(|q| q.run == t.id)
1756 .unwrap();
1757 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1758 questions.put(&mut asked).unwrap();
1759
1760 let report = run_once(&q, &questions, None, Timestamp::now());
1761 assert_eq!(report.answered, [t.id.clone()]);
1762 let back = q.get(&t.id).unwrap();
1763 assert_eq!(back.status, TaskStatus::Queued);
1764 assert!(back.hold_source.is_none());
1765 }
1766
1767 fn resumed_then_failed(q: &Queue, questions: &Questions, dir: &std::path::Path) -> Task {
1769 let mut t = task("gate went red", dir.join("repo"));
1770 t.hold_machine(Some("gate red".to_owned()));
1771 q.put(&mut t).unwrap();
1772 run_once(q, questions, None, Timestamp::now());
1773 let mut asked = questions
1774 .list()
1775 .into_iter()
1776 .find(|q| q.run == t.id)
1777 .unwrap();
1778 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1779 questions.put(&mut asked).unwrap();
1780
1781 let report = run_once(q, questions, None, Timestamp::now());
1782 assert_eq!(report.answered, [t.id.clone()]);
1783 let mut back = q.get(&t.id).unwrap();
1784 assert_eq!(back.status, TaskStatus::Queued);
1785 assert_eq!(back.attempts, 0);
1786
1787 back.start("run-1".to_owned());
1788 back.fail("rebase conflict", 1);
1789 assert_eq!(back.status, TaskStatus::Held);
1790 q.put(&mut back).unwrap();
1791 back
1792 }
1793
1794 #[test]
1795 fn a_resume_answer_is_applied_once_and_a_new_machine_hold_is_asked_about() {
1796 let (dir, q, questions) = store();
1797 let t = resumed_then_failed(&q, &questions, dir.path());
1798
1799 let report = run_once(&q, &questions, None, Timestamp::now());
1800 assert!(report.answered.is_empty(), "the old answer must not replay");
1801 assert_eq!(report.asked, std::slice::from_ref(&t.id));
1802 let back = q.get(&t.id).unwrap();
1803 assert_eq!(back.status, TaskStatus::Held);
1804 assert_eq!(
1805 questions
1806 .list()
1807 .into_iter()
1808 .filter(|q| q.status.open())
1809 .count(),
1810 1
1811 );
1812 }
1813
1814 #[test]
1815 fn a_manual_hold_placed_after_a_resume_is_not_undone_by_the_old_answer() {
1816 let (dir, q, questions) = store();
1817 let mut t = resumed_then_failed(&q, &questions, dir.path());
1818 t.hold_manual(Some("operator stopped this".to_owned()));
1819 q.put(&mut t).unwrap();
1820 let before = questions.list().len();
1821
1822 let report = run_once(&q, &questions, None, Timestamp::now());
1823 assert!(report.answered.is_empty());
1824 assert!(report.asked.is_empty());
1825 let back = q.get(&t.id).unwrap();
1826 assert_eq!(back.status, TaskStatus::Held);
1827 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1828 assert_eq!(questions.list().len(), before);
1829 }
1830
1831 #[test]
1832 fn answering_not_yet_keeps_it_held_as_a_manual_hold_and_does_not_reapply() {
1833 let (dir, q, questions) = store();
1834 let mut t = task("gate went red", dir.path().join("repo"));
1835 t.hold_machine(Some("gate red".to_owned()));
1836 q.put(&mut t).unwrap();
1837 run_once(&q, &questions, None, Timestamp::now());
1838
1839 let mut asked = questions
1840 .list()
1841 .into_iter()
1842 .find(|q| q.run == t.id)
1843 .unwrap();
1844 asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
1845 questions.put(&mut asked).unwrap();
1846
1847 let report = run_once(&q, &questions, None, Timestamp::now());
1848 assert_eq!(report.answered, [t.id.clone()]);
1849 let back = q.get(&t.id).unwrap();
1850 assert_eq!(back.status, TaskStatus::Held);
1851 assert_eq!(back.hold_source, Some(HoldSource::Manual));
1852 assert!(
1853 back.hold_reason
1854 .as_deref()
1855 .is_some_and(|r| r.contains("gate red")),
1856 "the original cause must survive a \"not yet\" answer, not just the \
1857 triage marker: {:?}",
1858 back.hold_reason
1859 );
1860
1861 let third = run_once(&q, &questions, None, Timestamp::now());
1865 assert!(third.answered.is_empty());
1866 assert!(third.asked.is_empty());
1867 }
1868
1869 #[test]
1870 fn answering_discard_removes_the_task_entirely() {
1871 let (dir, q, questions) = store();
1872 let mut t = task("gate went red", dir.path().join("repo"));
1873 t.hold_machine(Some("gate red".to_owned()));
1874 q.put(&mut t).unwrap();
1875 run_once(&q, &questions, None, Timestamp::now());
1876
1877 let mut asked = questions
1878 .list()
1879 .into_iter()
1880 .find(|q| q.run == t.id)
1881 .unwrap();
1882 asked.answer(Answer::Choice(EN.discard.to_owned())).unwrap();
1883 questions.put(&mut asked).unwrap();
1884
1885 let report = run_once(&q, &questions, None, Timestamp::now());
1886 assert_eq!(report.answered, [t.id.clone()]);
1887 assert!(
1888 q.get(&t.id).is_err(),
1889 "\"discard it\" (捨ててよい) must actually discard the task, not \
1890 just leave it sitting held forever"
1891 );
1892 }
1893
1894 #[test]
1895 fn an_answer_is_read_by_its_position_in_choices_not_by_the_callers_current_language() {
1896 let (dir, q, questions) = store();
1898 let ja_config = dir.path().join("ja.toml");
1899 std::fs::write(&ja_config, "[graph]\nlanguage = \"ja\"\n").unwrap();
1900 let mut t = task("gate went red", dir.path().join("repo"));
1901 t.hold_machine(Some("gate red".to_owned()));
1902 q.put(&mut t).unwrap();
1903 run_once(&q, &questions, Some(&ja_config), Timestamp::now());
1904
1905 let mut asked = questions
1906 .list()
1907 .into_iter()
1908 .find(|q| q.run == t.id)
1909 .unwrap();
1910 assert_eq!(asked.choices[0], JA.resume, "filed in Japanese");
1911 asked.answer(Answer::Choice(JA.resume.to_owned())).unwrap();
1912 questions.put(&mut asked).unwrap();
1913
1914 let en_config = dir.path().join("en.toml");
1919 std::fs::write(&en_config, "[graph]\nlanguage = \"en\"\n").unwrap();
1920 let report = run_once(&q, &questions, Some(&en_config), Timestamp::now());
1921 assert_eq!(report.answered, [t.id.clone()]);
1922 let back = q.get(&t.id).unwrap();
1923 assert_eq!(
1924 back.status,
1925 TaskStatus::Queued,
1926 "a resume answer must resume the task regardless of which \
1927 language it is read back in"
1928 );
1929 }
1930
1931 #[test]
1932 fn a_question_falls_back_to_last_error_when_hold_reason_was_never_set() {
1933 let (dir, q, questions) = store();
1938 let mut t = task("kept failing the gate", dir.path().join("repo"));
1939 t.start("run-1".to_owned());
1940 t.fail("gate red three times running", 1);
1941 assert_eq!(t.status, TaskStatus::Held);
1942 t.hold_reason = None; q.put(&mut t).unwrap();
1944
1945 run_once(&q, &questions, None, Timestamp::now());
1946 let asked = questions
1947 .list()
1948 .into_iter()
1949 .find(|q| q.run == t.id)
1950 .unwrap();
1951 assert!(
1952 asked.detail.contains("gate red three times running"),
1953 "the question must surface `last_error` when there is no \
1954 `hold_reason` to show instead: {}",
1955 asked.detail
1956 );
1957 }
1958
1959 #[test]
1960 fn open_question_for_and_open_task_ids_reflect_only_what_is_still_waiting() {
1961 let (dir, q, questions) = store();
1962 let mut t = task("gate went red", dir.path().join("repo"));
1963 t.hold_machine(Some("gate red".to_owned()));
1964 q.put(&mut t).unwrap();
1965
1966 assert!(open_question_for(&questions, &t.id).is_none());
1967 assert!(!open_task_ids(&questions).contains(&t.id));
1968
1969 run_once(&q, &questions, None, Timestamp::now());
1970 assert!(open_question_for(&questions, &t.id).is_some());
1971 assert!(open_task_ids(&questions).contains(&t.id));
1972
1973 let mut asked = questions
1974 .list()
1975 .into_iter()
1976 .find(|q| q.run == t.id)
1977 .unwrap();
1978 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1979 questions.put(&mut asked).unwrap();
1980
1981 assert!(
1982 open_question_for(&questions, &t.id).is_none(),
1983 "an answered question is no longer open"
1984 );
1985 assert!(!open_task_ids(&questions).contains(&t.id));
1986 }
1987
1988 #[test]
1989 fn pending_for_stays_true_between_an_answer_and_the_next_run_once_pass() {
1990 let (dir, q, questions) = store();
1996 let mut t = task("gate went red", dir.path().join("repo"));
1997 t.hold_machine(Some("gate red".to_owned()));
1998 q.put(&mut t).unwrap();
1999
2000 assert!(!pending_for(&questions, &t));
2001
2002 run_once(&q, &questions, None, Timestamp::now());
2003 let held = q.get(&t.id).unwrap();
2004 assert!(pending_for(&questions, &held), "still waiting on an answer");
2005
2006 let mut asked = questions
2007 .list()
2008 .into_iter()
2009 .find(|q| q.run == t.id)
2010 .unwrap();
2011 asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
2012 questions.put(&mut asked).unwrap();
2013 assert!(!asked.status.open());
2014
2015 let still_held = q.get(&t.id).unwrap();
2017 assert!(
2018 pending_for(&questions, &still_held),
2019 "answered but not yet applied is still pending"
2020 );
2021
2022 run_once(&q, &questions, None, Timestamp::now());
2023 let after = q.get(&t.id).unwrap();
2024 assert!(
2025 !pending_for(&questions, &after),
2026 "the answer is applied now, nothing left pending"
2027 );
2028 }
2029}