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