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