1use std::path::{Path, PathBuf};
86use std::time::Duration;
87
88use jiff::Timestamp;
89
90use crate::ask::{Question, QuestionStatus, Questions};
91use crate::config::Config;
92use crate::disk;
93use crate::queue::{HoldSource, Queue, Task, TaskStatus};
94
95pub const NODE: &str = "triage";
98
99const SEAT: &str = "triage";
104
105const MANUAL_STALE_AFTER: Duration = Duration::from_secs(7 * 24 * 60 * 60);
117
118struct Wording {
122 lang: &'static str,
123 resume: &'static str,
124 wait: &'static str,
125 discard: &'static str,
126 resume_now: &'static str,
127 keep_held: &'static str,
128}
129
130const EN: Wording = Wording {
131 lang: "en",
132 resume: "resume it",
133 wait: "not yet",
134 discard: "discard it",
135 resume_now: "resume it",
136 keep_held: "keep it held",
137};
138
139const JA: Wording = Wording {
140 lang: "ja",
141 resume: "再開してよい",
142 wait: "まだ待って",
143 discard: "捨ててよい",
144 resume_now: "再開する",
145 keep_held: "まだ止めておく",
146};
147
148fn wording(language: &str) -> &'static Wording {
151 let l = language.trim();
152 if l.eq_ignore_ascii_case("ja")
153 || l.eq_ignore_ascii_case("jp")
154 || l.eq_ignore_ascii_case("japanese")
155 || l.eq_ignore_ascii_case("日本語")
156 {
157 &JA
158 } else {
159 &EN
160 }
161}
162
163impl Wording {
164 fn choices3(&self) -> Vec<String> {
165 vec![
166 self.resume.to_owned(),
167 self.wait.to_owned(),
168 self.discard.to_owned(),
169 ]
170 }
171
172 fn choices2(&self) -> Vec<String> {
173 vec![self.resume_now.to_owned(), self.keep_held.to_owned()]
174 }
175
176 fn source_label(&self, source: Option<HoldSource>) -> &'static str {
177 match (self.lang, source) {
178 ("ja", Some(HoldSource::Machine)) => "machine(機械による自動保留)",
179 ("ja", Some(HoldSource::Manual)) => "manual(操作者による手動保留)",
180 ("ja", None) => "unknown(schema 3 未満の旧レコード、または理由未記録)",
181 (_, Some(HoldSource::Machine)) => "machine (automatic recovery hold)",
182 (_, Some(HoldSource::Manual)) => "manual (an operator held this)",
183 (_, None) => "unknown (pre-schema-3 record, or never recorded)",
184 }
185 }
186
187 fn detail(&self, task: &Task, why: &str) -> String {
196 let none = if self.lang == "ja" {
197 "(記録なし)"
198 } else {
199 "(none recorded)"
200 };
201 let reason = task
202 .hold_reason
203 .as_deref()
204 .or(task.last_error.as_deref())
205 .unwrap_or(none);
206 format!(
207 "task: {} ({})\ntitle: {}\nhold source: {}\nhold reason: {reason}\n\n{why}",
208 task.id,
209 task.short(),
210 task.title,
211 self.source_label(task.hold_source),
212 )
213 }
214
215 fn summary_machine_unknown(&self, task: &Task) -> String {
216 if self.lang == "ja" {
217 format!("保留タスク {} の再開可否を判断してください", task.short())
218 } else {
219 format!("decide whether to resume held task {}", task.short())
220 }
221 }
222
223 fn why_machine(&self) -> &'static str {
224 if self.lang == "ja" {
225 "機械的な保留(machine hold)ですが、原因がすでに解消しているかを自動では判断できませんでした。"
226 } else {
227 "This is a machine hold, but whether its cause has resolved could not be \
228 checked automatically."
229 }
230 }
231
232 fn summary_legacy(&self, task: &Task) -> String {
233 if self.lang == "ja" {
234 format!(
235 "hold_source が不明な保留タスク {} を確認してください",
236 task.short()
237 )
238 } else {
239 format!(
240 "held task {} has no recorded hold source - please take a look",
241 task.short()
242 )
243 }
244 }
245
246 fn why_legacy(&self) -> &'static str {
247 if self.lang == "ja" {
248 "hold_source が記録されていません。schema 3 より前のレコードか、理由が記録されなかった \
249 holdです。人が意図して止めたのか、クラッシュや強制再起動で宙に浮いただけなのか、\
250 このデータからは区別できません。"
251 } else {
252 "No hold_source was recorded - either a pre-schema-3 record, or a hold whose \
253 reason was never written down. Whether this was a deliberate hold or the \
254 leftover of a crash cannot be told from the data alone."
255 }
256 }
257
258 fn summary_manual_stale(&self, task: &Task, days: i64) -> String {
259 if self.lang == "ja" {
260 format!(
261 "{days}日間 保留されたままの手動保留タスク {} を確認してください",
262 task.short()
263 )
264 } else {
265 format!(
266 "held task {} has been on a manual hold for {days} day(s)",
267 task.short()
268 )
269 }
270 }
271
272 fn why_manual(&self) -> &'static str {
273 if self.lang == "ja" {
274 "操作者が明示的に止めた保留ですが、長期間そのままになっています。まだ止めておくか、\
275 再開するか教えてください。"
276 } else {
277 "An operator held this on purpose, but it has sat untouched for a while. Say \
278 whether to keep holding it or resume it."
279 }
280 }
281}
282
283#[derive(Debug, Clone, Copy, PartialEq, Eq)]
285enum Bucket {
286 MachineUnknown,
288 Legacy,
290 ManualStale,
292}
293
294#[derive(Debug, Clone, Default)]
296pub struct Report {
297 pub resumed: Vec<String>,
300 pub asked: Vec<String>,
302 pub answered: Vec<String>,
304}
305
306impl Report {
307 pub fn is_empty(&self) -> bool {
310 self.resumed.is_empty() && self.asked.is_empty() && self.answered.is_empty()
311 }
312}
313
314fn repo_for(task: &Task) -> PathBuf {
318 if task.repo.as_os_str().is_empty() {
319 PathBuf::from(".")
320 } else {
321 task.repo.clone()
322 }
323}
324
325fn is_disk_hold(task: &Task) -> bool {
336 task.hold_reason.as_deref().is_some_and(|r| {
337 r.starts_with("not enough free space to start a run:")
338 || r.starts_with("could not measure free space on ")
339 })
340}
341
342fn machine_cause_resolved(task: &Task, cfg: &Config) -> Option<bool> {
347 if !is_disk_hold(task) {
348 return None;
349 }
350 let min = cfg.disk.min_free_bytes;
351 if min == 0 {
352 return Some(true);
355 }
356 let free = disk::free_bytes(&repo_for(task)).ok()?;
357 Some(disk::gate(free, min).is_none())
358}
359
360fn manual_is_stale(task: &Task, now: Timestamp) -> bool {
364 now.as_second() - task.updated_at.as_second() > MANUAL_STALE_AFTER.as_secs() as i64
365}
366
367fn marker_for(q: &Question) -> String {
370 format!("[triage:{}]", q.short())
371}
372
373fn already_applied(task: &Task, q: &Question) -> bool {
375 let marker = marker_for(q);
376 task.hold_reason
377 .as_deref()
378 .is_some_and(|r| r.contains(marker.as_str()))
379}
380
381fn latest_triage_question(questions: &Questions, task_id: &str) -> Option<Question> {
386 questions
387 .list()
388 .into_iter()
389 .filter(|q| q.node == NODE && q.run == task_id)
390 .max_by(|a, b| a.id.cmp(&b.id))
391}
392
393#[derive(Debug, Clone, Copy, PartialEq, Eq)]
396enum AnswerAction {
397 Resume,
401 Discard,
404 KeepHeld,
409}
410
411fn interpret_answer(q: &Question) -> AnswerAction {
424 let resolution = q.resolution().unwrap_or_default();
425 match q.choices.iter().position(|c| *c == resolution) {
426 Some(0) => AnswerAction::Resume,
427 Some(2) => AnswerAction::Discard,
428 _ => AnswerAction::KeepHeld,
429 }
430}
431
432fn keep_held_note(task: &Task, q: &Question, resolution: &str) -> String {
437 let marker = format!("{} operator: {resolution}", marker_for(q));
438 match task.hold_reason.as_deref() {
439 Some(existing) if !existing.is_empty() => format!("{existing}\n{marker}"),
440 _ => marker,
441 }
442}
443
444fn file_question(
448 questions: &Questions,
449 task: &Task,
450 bucket: Bucket,
451 w: &Wording,
452 now: Timestamp,
453) -> Option<Question> {
454 let (summary, why, choices) = match bucket {
455 Bucket::MachineUnknown => (
456 w.summary_machine_unknown(task),
457 w.why_machine(),
458 w.choices3(),
459 ),
460 Bucket::Legacy => (w.summary_legacy(task), w.why_legacy(), w.choices3()),
461 Bucket::ManualStale => {
462 let days = (now.as_second() - task.updated_at.as_second()) / (24 * 60 * 60);
463 (
464 w.summary_manual_stale(task, days),
465 w.why_manual(),
466 w.choices2(),
467 )
468 }
469 };
470 let mut q = Question::new(
471 task.id.clone(),
472 NODE.to_owned(),
473 SEAT.to_owned(),
474 summary,
475 w.detail(task, why),
476 choices,
477 );
478 questions.put(&mut q).ok()?;
479 Some(q)
480}
481
482pub fn run_once(
496 queue: &Queue,
497 questions: &Questions,
498 config_override: Option<&Path>,
499 now: Timestamp,
500) -> Report {
501 let mut report = Report::default();
502 for listed in queue.list() {
503 if listed.status != TaskStatus::Held {
504 continue;
505 }
506 let Ok(_claim) = queue.claim(&listed.id) else {
507 continue;
508 };
509 let Ok(mut task) = queue.get(&listed.id) else {
510 continue;
511 };
512 if task.status != TaskStatus::Held {
516 continue;
517 }
518
519 let cfg = Config::discover(&repo_for(&task), config_override)
520 .ok()
521 .map(|(c, _)| c);
522 let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
523
524 if let Some(q) = latest_triage_question(questions, &task.id) {
525 if q.status.open() {
526 continue;
528 }
529 if q.status == QuestionStatus::Answered && !already_applied(&task, &q) {
530 match interpret_answer(&q) {
531 AnswerAction::Resume => {
532 task.release();
533 if queue.put(&mut task).is_ok() {
534 report.answered.push(task.id.clone());
535 }
536 }
537 AnswerAction::Discard => {
538 if queue.remove(&task.id, false).is_ok() {
539 report.answered.push(task.id.clone());
540 }
541 }
542 AnswerAction::KeepHeld => {
543 let resolution = q.resolution().unwrap_or_default();
544 let note = keep_held_note(&task, &q, &resolution);
545 task.hold_manual(Some(note));
546 if queue.put(&mut task).is_ok() {
547 report.answered.push(task.id.clone());
548 }
549 }
550 }
551 continue;
552 }
553 }
559
560 match task.hold_source {
561 Some(HoldSource::Machine) => {
562 if cfg.as_ref().and_then(|c| machine_cause_resolved(&task, c)) == Some(true) {
563 task.release();
564 if queue.put(&mut task).is_ok() {
565 report.resumed.push(task.id.clone());
566 }
567 } else if file_question(questions, &task, Bucket::MachineUnknown, w, now).is_some()
568 {
569 report.asked.push(task.id.clone());
570 }
571 }
572 None => {
573 if file_question(questions, &task, Bucket::Legacy, w, now).is_some() {
574 report.asked.push(task.id.clone());
575 }
576 }
577 Some(HoldSource::Manual) => {
578 if manual_is_stale(&task, now)
579 && file_question(questions, &task, Bucket::ManualStale, w, now).is_some()
580 {
581 report.asked.push(task.id.clone());
582 }
583 }
584 }
585 }
586 report
587}
588
589pub fn open_question_for(questions: &Questions, task_id: &str) -> Option<Question> {
594 latest_triage_question(questions, task_id).filter(|q| q.status.open())
595}
596
597pub fn open_task_ids(questions: &Questions) -> std::collections::BTreeSet<String> {
602 questions
603 .list()
604 .into_iter()
605 .filter(|q| q.node == NODE && q.status.open())
606 .map(|q| q.run)
607 .collect()
608}
609
610#[cfg(test)]
611mod tests {
612 use super::*;
613 use crate::ask::Answer;
614 use crate::queue::Source;
615 use jiff::SignedDuration;
616
617 fn store() -> (tempfile::TempDir, Queue, Questions) {
618 let dir = tempfile::tempdir().unwrap();
619 let q = Queue::at(dir.path().join("queue"));
620 let s = Questions::at(dir.path().join("questions"));
621 (dir, q, s)
622 }
623
624 fn task(title: &str, repo: PathBuf) -> Task {
625 Task::new(title.to_owned(), format!("do {title}"), repo, Source::Human)
626 }
627
628 fn gate_disabled_config(dir: &std::path::Path) -> PathBuf {
632 let config = dir.join("magi.toml");
633 std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
634 config
635 }
636
637 #[test]
638 fn a_resolved_machine_hold_is_requeued_automatically() {
639 let (dir, q, questions) = store();
640 let config = gate_disabled_config(dir.path());
641 let mut t = task("disk pressure", dir.path().join("repo"));
642 t.hold_machine(Some(
643 "not enough free space to start a run: 10 bytes free, 100 required by \
644 `[disk] min_free_bytes`"
645 .to_owned(),
646 ));
647 q.put(&mut t).unwrap();
648
649 let report = run_once(&q, &questions, Some(&config), Timestamp::now());
650 assert_eq!(report.resumed, [t.id.clone()]);
651 assert!(report.asked.is_empty());
652
653 let back = q.get(&t.id).unwrap();
654 assert_eq!(back.status, TaskStatus::Queued);
655 assert!(back.hold_source.is_none());
656 assert!(questions.list().is_empty(), "nothing needed asking");
657 }
658
659 #[test]
660 fn a_machine_hold_with_no_recognised_cause_gets_one_question_not_two() {
661 let (dir, q, questions) = store();
662 let mut t = task("gate went red", dir.path().join("repo"));
663 t.hold_machine(Some("gate red".to_owned()));
664 q.put(&mut t).unwrap();
665
666 let first = run_once(&q, &questions, None, Timestamp::now());
667 assert_eq!(first.asked, [t.id.clone()]);
668 assert!(first.resumed.is_empty());
669
670 let open: Vec<_> = questions
671 .list()
672 .into_iter()
673 .filter(|q| q.status.open())
674 .collect();
675 assert_eq!(open.len(), 1);
676 assert_eq!(open[0].run, t.id);
677 assert_eq!(open[0].node, NODE);
678 assert_eq!(open[0].choices.len(), 3);
679
680 let second = run_once(&q, &questions, None, Timestamp::now());
682 assert!(second.asked.is_empty());
683 assert_eq!(
684 questions
685 .list()
686 .into_iter()
687 .filter(|q| q.status.open())
688 .count(),
689 1
690 );
691 }
692
693 #[test]
694 fn a_legacy_hold_with_no_recorded_source_gets_exactly_one_question() {
695 let (dir, q, questions) = store();
696 let mut t = task("schema 1 record", dir.path().join("repo"));
697 t.status = TaskStatus::Held;
698 assert!(t.hold_source.is_none(), "the case this test is about");
699 q.put(&mut t).unwrap();
700
701 let first = run_once(&q, &questions, None, Timestamp::now());
702 assert_eq!(first.asked, [t.id.clone()]);
703
704 let second = run_once(&q, &questions, None, Timestamp::now());
705 assert!(
706 second.asked.is_empty(),
707 "the same legacy hold must not be asked about twice"
708 );
709 assert_eq!(
710 questions
711 .list()
712 .into_iter()
713 .filter(|q| q.status.open())
714 .count(),
715 1
716 );
717 }
718
719 #[test]
720 fn a_manual_hold_is_never_auto_resumed() {
721 let (dir, q, questions) = store();
722 let mut t = task("operator stopped this", dir.path().join("repo"));
723 t.hold_manual(Some("waiting on a decision".to_owned()));
724 q.put(&mut t).unwrap();
725
726 let report = run_once(&q, &questions, None, Timestamp::now());
727 assert!(report.resumed.is_empty());
728 assert!(report.asked.is_empty());
730
731 let back = q.get(&t.id).unwrap();
732 assert_eq!(back.status, TaskStatus::Held);
733 assert_eq!(back.hold_source, Some(HoldSource::Manual));
734 assert!(questions.list().is_empty());
735 }
736
737 #[test]
738 fn a_stale_manual_hold_earns_a_two_choice_question() {
739 let (dir, q, questions) = store();
740 let mut t = task("been sitting a while", dir.path().join("repo"));
741 t.hold_manual(Some("waiting on a decision".to_owned()));
742 q.put(&mut t).unwrap();
743 let mut back = q.get(&t.id).unwrap();
745 back.updated_at = Timestamp::now() - SignedDuration::new(8 * 24 * 60 * 60, 0);
746 std::fs::write(
747 q.path_of(&back.id),
748 serde_json::to_string_pretty(&back).unwrap(),
749 )
750 .unwrap();
751
752 let report = run_once(&q, &questions, None, Timestamp::now());
753 assert_eq!(report.asked, [t.id.clone()]);
754 let open: Vec<_> = questions
755 .list()
756 .into_iter()
757 .filter(|q| q.status.open())
758 .collect();
759 assert_eq!(open.len(), 1);
760 assert_eq!(open[0].choices.len(), 2);
761 }
762
763 #[test]
764 fn answering_resume_releases_the_task() {
765 let (dir, q, questions) = store();
766 let mut t = task("gate went red", dir.path().join("repo"));
767 t.hold_machine(Some("gate red".to_owned()));
768 q.put(&mut t).unwrap();
769 run_once(&q, &questions, None, Timestamp::now());
770
771 let mut asked = questions
772 .list()
773 .into_iter()
774 .find(|q| q.run == t.id)
775 .unwrap();
776 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
777 questions.put(&mut asked).unwrap();
778
779 let report = run_once(&q, &questions, None, Timestamp::now());
780 assert_eq!(report.answered, [t.id.clone()]);
781 let back = q.get(&t.id).unwrap();
782 assert_eq!(back.status, TaskStatus::Queued);
783 assert!(back.hold_source.is_none());
784 }
785
786 #[test]
787 fn answering_not_yet_keeps_it_held_as_a_manual_hold_and_does_not_reapply() {
788 let (dir, q, questions) = store();
789 let mut t = task("gate went red", dir.path().join("repo"));
790 t.hold_machine(Some("gate red".to_owned()));
791 q.put(&mut t).unwrap();
792 run_once(&q, &questions, None, Timestamp::now());
793
794 let mut asked = questions
795 .list()
796 .into_iter()
797 .find(|q| q.run == t.id)
798 .unwrap();
799 asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
800 questions.put(&mut asked).unwrap();
801
802 let report = run_once(&q, &questions, None, Timestamp::now());
803 assert_eq!(report.answered, [t.id.clone()]);
804 let back = q.get(&t.id).unwrap();
805 assert_eq!(back.status, TaskStatus::Held);
806 assert_eq!(back.hold_source, Some(HoldSource::Manual));
807 assert!(
808 back.hold_reason
809 .as_deref()
810 .is_some_and(|r| r.contains("gate red")),
811 "the original cause must survive a \"not yet\" answer, not just the \
812 triage marker: {:?}",
813 back.hold_reason
814 );
815
816 let third = run_once(&q, &questions, None, Timestamp::now());
820 assert!(third.answered.is_empty());
821 assert!(third.asked.is_empty());
822 }
823
824 #[test]
825 fn answering_discard_removes_the_task_entirely() {
826 let (dir, q, questions) = store();
827 let mut t = task("gate went red", dir.path().join("repo"));
828 t.hold_machine(Some("gate red".to_owned()));
829 q.put(&mut t).unwrap();
830 run_once(&q, &questions, None, Timestamp::now());
831
832 let mut asked = questions
833 .list()
834 .into_iter()
835 .find(|q| q.run == t.id)
836 .unwrap();
837 asked.answer(Answer::Choice(EN.discard.to_owned())).unwrap();
838 questions.put(&mut asked).unwrap();
839
840 let report = run_once(&q, &questions, None, Timestamp::now());
841 assert_eq!(report.answered, [t.id.clone()]);
842 assert!(
843 q.get(&t.id).is_err(),
844 "\"discard it\" (捨ててよい) must actually discard the task, not \
845 just leave it sitting held forever"
846 );
847 }
848
849 #[test]
850 fn an_answer_is_read_by_its_position_in_choices_not_by_the_callers_current_language() {
851 let (dir, q, questions) = store();
853 let ja_config = dir.path().join("ja.toml");
854 std::fs::write(&ja_config, "[graph]\nlanguage = \"ja\"\n").unwrap();
855 let mut t = task("gate went red", dir.path().join("repo"));
856 t.hold_machine(Some("gate red".to_owned()));
857 q.put(&mut t).unwrap();
858 run_once(&q, &questions, Some(&ja_config), Timestamp::now());
859
860 let mut asked = questions
861 .list()
862 .into_iter()
863 .find(|q| q.run == t.id)
864 .unwrap();
865 assert_eq!(asked.choices[0], JA.resume, "filed in Japanese");
866 asked.answer(Answer::Choice(JA.resume.to_owned())).unwrap();
867 questions.put(&mut asked).unwrap();
868
869 let en_config = dir.path().join("en.toml");
874 std::fs::write(&en_config, "[graph]\nlanguage = \"en\"\n").unwrap();
875 let report = run_once(&q, &questions, Some(&en_config), Timestamp::now());
876 assert_eq!(report.answered, [t.id.clone()]);
877 let back = q.get(&t.id).unwrap();
878 assert_eq!(
879 back.status,
880 TaskStatus::Queued,
881 "a resume answer must resume the task regardless of which \
882 language it is read back in"
883 );
884 }
885
886 #[test]
887 fn a_question_falls_back_to_last_error_when_hold_reason_was_never_set() {
888 let (dir, q, questions) = store();
892 let mut t = task("kept failing the gate", dir.path().join("repo"));
893 t.start("run-1".to_owned());
894 t.fail("gate red three times running", 1);
895 assert_eq!(t.status, TaskStatus::Held);
896 assert!(t.hold_reason.is_none(), "the case this test is about");
897 q.put(&mut t).unwrap();
898
899 run_once(&q, &questions, None, Timestamp::now());
900 let asked = questions
901 .list()
902 .into_iter()
903 .find(|q| q.run == t.id)
904 .unwrap();
905 assert!(
906 asked.detail.contains("gate red three times running"),
907 "the question must surface `last_error` when there is no \
908 `hold_reason` to show instead: {}",
909 asked.detail
910 );
911 }
912
913 #[test]
914 fn open_question_for_and_open_task_ids_reflect_only_what_is_still_waiting() {
915 let (dir, q, questions) = store();
916 let mut t = task("gate went red", dir.path().join("repo"));
917 t.hold_machine(Some("gate red".to_owned()));
918 q.put(&mut t).unwrap();
919
920 assert!(open_question_for(&questions, &t.id).is_none());
921 assert!(!open_task_ids(&questions).contains(&t.id));
922
923 run_once(&q, &questions, None, Timestamp::now());
924 assert!(open_question_for(&questions, &t.id).is_some());
925 assert!(open_task_ids(&questions).contains(&t.id));
926
927 let mut asked = questions
928 .list()
929 .into_iter()
930 .find(|q| q.run == t.id)
931 .unwrap();
932 asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
933 questions.put(&mut asked).unwrap();
934
935 assert!(
936 open_question_for(&questions, &t.id).is_none(),
937 "an answered question is no longer open"
938 );
939 assert!(!open_task_ids(&questions).contains(&t.id));
940 }
941}