use std::path::{Path, PathBuf};
use std::time::Duration;
use jiff::Timestamp;
use crate::ask::{Question, QuestionStatus, Questions};
use crate::config::Config;
use crate::disk;
use crate::queue::{HoldSource, Queue, Task, TaskStatus};
pub const NODE: &str = "triage";
const SEAT: &str = "triage";
const MANUAL_STALE_AFTER: Duration = Duration::from_secs(7 * 24 * 60 * 60);
struct Wording {
lang: &'static str,
resume: &'static str,
wait: &'static str,
discard: &'static str,
resume_now: &'static str,
keep_held: &'static str,
}
const EN: Wording = Wording {
lang: "en",
resume: "resume it",
wait: "not yet",
discard: "discard it",
resume_now: "resume it",
keep_held: "keep it held",
};
const JA: Wording = Wording {
lang: "ja",
resume: "再開してよい",
wait: "まだ待って",
discard: "捨ててよい",
resume_now: "再開する",
keep_held: "まだ止めておく",
};
fn wording(language: &str) -> &'static Wording {
let l = language.trim();
if l.eq_ignore_ascii_case("ja")
|| l.eq_ignore_ascii_case("jp")
|| l.eq_ignore_ascii_case("japanese")
|| l.eq_ignore_ascii_case("日本語")
{
&JA
} else {
&EN
}
}
impl Wording {
fn choices3(&self) -> Vec<String> {
vec![
self.resume.to_owned(),
self.wait.to_owned(),
self.discard.to_owned(),
]
}
fn choices2(&self) -> Vec<String> {
vec![self.resume_now.to_owned(), self.keep_held.to_owned()]
}
fn source_label(&self, source: Option<HoldSource>) -> &'static str {
match (self.lang, source) {
("ja", Some(HoldSource::Machine)) => "machine(機械による自動保留)",
("ja", Some(HoldSource::Manual)) => "manual(操作者による手動保留)",
("ja", None) => "unknown(schema 3 未満の旧レコード、または理由未記録)",
(_, Some(HoldSource::Machine)) => "machine (automatic recovery hold)",
(_, Some(HoldSource::Manual)) => "manual (an operator held this)",
(_, None) => "unknown (pre-schema-3 record, or never recorded)",
}
}
fn detail(&self, task: &Task, why: &str) -> String {
let none = if self.lang == "ja" {
"(記録なし)"
} else {
"(none recorded)"
};
let reason = task
.hold_reason
.as_deref()
.or(task.last_error.as_deref())
.unwrap_or(none);
format!(
"task: {} ({})\ntitle: {}\nhold source: {}\nhold reason: {reason}\n\n{why}",
task.id,
task.short(),
task.title,
self.source_label(task.hold_source),
)
}
fn summary_machine_unknown(&self, task: &Task) -> String {
if self.lang == "ja" {
format!("保留タスク {} の再開可否を判断してください", task.short())
} else {
format!("decide whether to resume held task {}", task.short())
}
}
fn why_machine(&self) -> &'static str {
if self.lang == "ja" {
"機械的な保留(machine hold)ですが、原因がすでに解消しているかを自動では判断できませんでした。"
} else {
"This is a machine hold, but whether its cause has resolved could not be \
checked automatically."
}
}
fn summary_legacy(&self, task: &Task) -> String {
if self.lang == "ja" {
format!(
"hold_source が不明な保留タスク {} を確認してください",
task.short()
)
} else {
format!(
"held task {} has no recorded hold source - please take a look",
task.short()
)
}
}
fn why_legacy(&self) -> &'static str {
if self.lang == "ja" {
"hold_source が記録されていません。schema 3 より前のレコードか、理由が記録されなかった \
holdです。人が意図して止めたのか、クラッシュや強制再起動で宙に浮いただけなのか、\
このデータからは区別できません。"
} else {
"No hold_source was recorded - either a pre-schema-3 record, or a hold whose \
reason was never written down. Whether this was a deliberate hold or the \
leftover of a crash cannot be told from the data alone."
}
}
fn summary_manual_stale(&self, task: &Task, days: i64) -> String {
if self.lang == "ja" {
format!(
"{days}日間 保留されたままの手動保留タスク {} を確認してください",
task.short()
)
} else {
format!(
"held task {} has been on a manual hold for {days} day(s)",
task.short()
)
}
}
fn why_manual(&self) -> &'static str {
if self.lang == "ja" {
"操作者が明示的に止めた保留ですが、長期間そのままになっています。まだ止めておくか、\
再開するか教えてください。"
} else {
"An operator held this on purpose, but it has sat untouched for a while. Say \
whether to keep holding it or resume it."
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Bucket {
MachineUnknown,
Legacy,
ManualStale,
}
#[derive(Debug, Clone, Default)]
pub struct Report {
pub resumed: Vec<String>,
pub asked: Vec<String>,
pub answered: Vec<String>,
}
impl Report {
pub fn is_empty(&self) -> bool {
self.resumed.is_empty() && self.asked.is_empty() && self.answered.is_empty()
}
}
fn repo_for(task: &Task) -> PathBuf {
if task.repo.as_os_str().is_empty() {
PathBuf::from(".")
} else {
task.repo.clone()
}
}
fn is_disk_hold(task: &Task) -> bool {
task.hold_reason.as_deref().is_some_and(|r| {
r.starts_with("not enough free space to start a run:")
|| r.starts_with("could not measure free space on ")
})
}
fn machine_cause_resolved(task: &Task, cfg: &Config) -> Option<bool> {
if !is_disk_hold(task) {
return None;
}
let min = cfg.disk.min_free_bytes;
if min == 0 {
return Some(true);
}
let free = disk::free_bytes(&repo_for(task)).ok()?;
Some(disk::gate(free, min).is_none())
}
fn manual_is_stale(task: &Task, now: Timestamp) -> bool {
now.as_second() - task.updated_at.as_second() > MANUAL_STALE_AFTER.as_secs() as i64
}
fn marker_for(q: &Question) -> String {
format!("[triage:{}]", q.short())
}
fn already_applied(task: &Task, q: &Question) -> bool {
let marker = marker_for(q);
task.hold_reason
.as_deref()
.is_some_and(|r| r.contains(marker.as_str()))
}
fn latest_triage_question(questions: &Questions, task_id: &str) -> Option<Question> {
questions
.list()
.into_iter()
.filter(|q| q.node == NODE && q.run == task_id)
.max_by(|a, b| a.id.cmp(&b.id))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AnswerAction {
Resume,
Discard,
KeepHeld,
}
fn interpret_answer(q: &Question) -> AnswerAction {
let resolution = q.resolution().unwrap_or_default();
match q.choices.iter().position(|c| *c == resolution) {
Some(0) => AnswerAction::Resume,
Some(2) => AnswerAction::Discard,
_ => AnswerAction::KeepHeld,
}
}
fn keep_held_note(task: &Task, q: &Question, resolution: &str) -> String {
let marker = format!("{} operator: {resolution}", marker_for(q));
match task.hold_reason.as_deref() {
Some(existing) if !existing.is_empty() => format!("{existing}\n{marker}"),
_ => marker,
}
}
fn file_question(
questions: &Questions,
task: &Task,
bucket: Bucket,
w: &Wording,
now: Timestamp,
) -> Option<Question> {
let (summary, why, choices) = match bucket {
Bucket::MachineUnknown => (
w.summary_machine_unknown(task),
w.why_machine(),
w.choices3(),
),
Bucket::Legacy => (w.summary_legacy(task), w.why_legacy(), w.choices3()),
Bucket::ManualStale => {
let days = (now.as_second() - task.updated_at.as_second()) / (24 * 60 * 60);
(
w.summary_manual_stale(task, days),
w.why_manual(),
w.choices2(),
)
}
};
let mut q = Question::new(
task.id.clone(),
NODE.to_owned(),
SEAT.to_owned(),
summary,
w.detail(task, why),
choices,
);
questions.put(&mut q).ok()?;
Some(q)
}
pub fn run_once(
queue: &Queue,
questions: &Questions,
config_override: Option<&Path>,
now: Timestamp,
) -> Report {
let mut report = Report::default();
for listed in queue.list() {
if listed.status != TaskStatus::Held {
continue;
}
let Ok(_claim) = queue.claim(&listed.id) else {
continue;
};
let Ok(mut task) = queue.get(&listed.id) else {
continue;
};
if task.status != TaskStatus::Held {
continue;
}
let cfg = Config::discover(&repo_for(&task), config_override)
.ok()
.map(|(c, _)| c);
let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
if let Some(q) = latest_triage_question(questions, &task.id) {
if q.status.open() {
continue;
}
if q.status == QuestionStatus::Answered && !already_applied(&task, &q) {
match interpret_answer(&q) {
AnswerAction::Resume => {
task.release();
if queue.put(&mut task).is_ok() {
report.answered.push(task.id.clone());
}
}
AnswerAction::Discard => {
if queue.remove(&task.id, false).is_ok() {
report.answered.push(task.id.clone());
}
}
AnswerAction::KeepHeld => {
let resolution = q.resolution().unwrap_or_default();
let note = keep_held_note(&task, &q, &resolution);
task.hold_manual(Some(note));
if queue.put(&mut task).is_ok() {
report.answered.push(task.id.clone());
}
}
}
continue;
}
}
match task.hold_source {
Some(HoldSource::Machine) => {
if cfg.as_ref().and_then(|c| machine_cause_resolved(&task, c)) == Some(true) {
task.release();
if queue.put(&mut task).is_ok() {
report.resumed.push(task.id.clone());
}
} else if file_question(questions, &task, Bucket::MachineUnknown, w, now).is_some()
{
report.asked.push(task.id.clone());
}
}
None => {
if file_question(questions, &task, Bucket::Legacy, w, now).is_some() {
report.asked.push(task.id.clone());
}
}
Some(HoldSource::Manual) => {
if manual_is_stale(&task, now)
&& file_question(questions, &task, Bucket::ManualStale, w, now).is_some()
{
report.asked.push(task.id.clone());
}
}
}
}
report
}
pub fn open_question_for(questions: &Questions, task_id: &str) -> Option<Question> {
latest_triage_question(questions, task_id).filter(|q| q.status.open())
}
pub fn open_task_ids(questions: &Questions) -> std::collections::BTreeSet<String> {
questions
.list()
.into_iter()
.filter(|q| q.node == NODE && q.status.open())
.map(|q| q.run)
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ask::Answer;
use crate::queue::Source;
use jiff::SignedDuration;
fn store() -> (tempfile::TempDir, Queue, Questions) {
let dir = tempfile::tempdir().unwrap();
let q = Queue::at(dir.path().join("queue"));
let s = Questions::at(dir.path().join("questions"));
(dir, q, s)
}
fn task(title: &str, repo: PathBuf) -> Task {
Task::new(title.to_owned(), format!("do {title}"), repo, Source::Human)
}
fn gate_disabled_config(dir: &std::path::Path) -> PathBuf {
let config = dir.join("magi.toml");
std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
config
}
#[test]
fn a_resolved_machine_hold_is_requeued_automatically() {
let (dir, q, questions) = store();
let config = gate_disabled_config(dir.path());
let mut t = task("disk pressure", dir.path().join("repo"));
t.hold_machine(Some(
"not enough free space to start a run: 10 bytes free, 100 required by \
`[disk] min_free_bytes`"
.to_owned(),
));
q.put(&mut t).unwrap();
let report = run_once(&q, &questions, Some(&config), Timestamp::now());
assert_eq!(report.resumed, [t.id.clone()]);
assert!(report.asked.is_empty());
let back = q.get(&t.id).unwrap();
assert_eq!(back.status, TaskStatus::Queued);
assert!(back.hold_source.is_none());
assert!(questions.list().is_empty(), "nothing needed asking");
}
#[test]
fn a_machine_hold_with_no_recognised_cause_gets_one_question_not_two() {
let (dir, q, questions) = store();
let mut t = task("gate went red", dir.path().join("repo"));
t.hold_machine(Some("gate red".to_owned()));
q.put(&mut t).unwrap();
let first = run_once(&q, &questions, None, Timestamp::now());
assert_eq!(first.asked, [t.id.clone()]);
assert!(first.resumed.is_empty());
let open: Vec<_> = questions
.list()
.into_iter()
.filter(|q| q.status.open())
.collect();
assert_eq!(open.len(), 1);
assert_eq!(open[0].run, t.id);
assert_eq!(open[0].node, NODE);
assert_eq!(open[0].choices.len(), 3);
let second = run_once(&q, &questions, None, Timestamp::now());
assert!(second.asked.is_empty());
assert_eq!(
questions
.list()
.into_iter()
.filter(|q| q.status.open())
.count(),
1
);
}
#[test]
fn a_legacy_hold_with_no_recorded_source_gets_exactly_one_question() {
let (dir, q, questions) = store();
let mut t = task("schema 1 record", dir.path().join("repo"));
t.status = TaskStatus::Held;
assert!(t.hold_source.is_none(), "the case this test is about");
q.put(&mut t).unwrap();
let first = run_once(&q, &questions, None, Timestamp::now());
assert_eq!(first.asked, [t.id.clone()]);
let second = run_once(&q, &questions, None, Timestamp::now());
assert!(
second.asked.is_empty(),
"the same legacy hold must not be asked about twice"
);
assert_eq!(
questions
.list()
.into_iter()
.filter(|q| q.status.open())
.count(),
1
);
}
#[test]
fn a_manual_hold_is_never_auto_resumed() {
let (dir, q, questions) = store();
let mut t = task("operator stopped this", dir.path().join("repo"));
t.hold_manual(Some("waiting on a decision".to_owned()));
q.put(&mut t).unwrap();
let report = run_once(&q, &questions, None, Timestamp::now());
assert!(report.resumed.is_empty());
assert!(report.asked.is_empty());
let back = q.get(&t.id).unwrap();
assert_eq!(back.status, TaskStatus::Held);
assert_eq!(back.hold_source, Some(HoldSource::Manual));
assert!(questions.list().is_empty());
}
#[test]
fn a_stale_manual_hold_earns_a_two_choice_question() {
let (dir, q, questions) = store();
let mut t = task("been sitting a while", dir.path().join("repo"));
t.hold_manual(Some("waiting on a decision".to_owned()));
q.put(&mut t).unwrap();
let mut back = q.get(&t.id).unwrap();
back.updated_at = Timestamp::now() - SignedDuration::new(8 * 24 * 60 * 60, 0);
std::fs::write(
q.path_of(&back.id),
serde_json::to_string_pretty(&back).unwrap(),
)
.unwrap();
let report = run_once(&q, &questions, None, Timestamp::now());
assert_eq!(report.asked, [t.id.clone()]);
let open: Vec<_> = questions
.list()
.into_iter()
.filter(|q| q.status.open())
.collect();
assert_eq!(open.len(), 1);
assert_eq!(open[0].choices.len(), 2);
}
#[test]
fn answering_resume_releases_the_task() {
let (dir, q, questions) = store();
let mut t = task("gate went red", dir.path().join("repo"));
t.hold_machine(Some("gate red".to_owned()));
q.put(&mut t).unwrap();
run_once(&q, &questions, None, Timestamp::now());
let mut asked = questions
.list()
.into_iter()
.find(|q| q.run == t.id)
.unwrap();
asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
questions.put(&mut asked).unwrap();
let report = run_once(&q, &questions, None, Timestamp::now());
assert_eq!(report.answered, [t.id.clone()]);
let back = q.get(&t.id).unwrap();
assert_eq!(back.status, TaskStatus::Queued);
assert!(back.hold_source.is_none());
}
#[test]
fn answering_not_yet_keeps_it_held_as_a_manual_hold_and_does_not_reapply() {
let (dir, q, questions) = store();
let mut t = task("gate went red", dir.path().join("repo"));
t.hold_machine(Some("gate red".to_owned()));
q.put(&mut t).unwrap();
run_once(&q, &questions, None, Timestamp::now());
let mut asked = questions
.list()
.into_iter()
.find(|q| q.run == t.id)
.unwrap();
asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
questions.put(&mut asked).unwrap();
let report = run_once(&q, &questions, None, Timestamp::now());
assert_eq!(report.answered, [t.id.clone()]);
let back = q.get(&t.id).unwrap();
assert_eq!(back.status, TaskStatus::Held);
assert_eq!(back.hold_source, Some(HoldSource::Manual));
assert!(
back.hold_reason
.as_deref()
.is_some_and(|r| r.contains("gate red")),
"the original cause must survive a \"not yet\" answer, not just the \
triage marker: {:?}",
back.hold_reason
);
let third = run_once(&q, &questions, None, Timestamp::now());
assert!(third.answered.is_empty());
assert!(third.asked.is_empty());
}
#[test]
fn answering_discard_removes_the_task_entirely() {
let (dir, q, questions) = store();
let mut t = task("gate went red", dir.path().join("repo"));
t.hold_machine(Some("gate red".to_owned()));
q.put(&mut t).unwrap();
run_once(&q, &questions, None, Timestamp::now());
let mut asked = questions
.list()
.into_iter()
.find(|q| q.run == t.id)
.unwrap();
asked.answer(Answer::Choice(EN.discard.to_owned())).unwrap();
questions.put(&mut asked).unwrap();
let report = run_once(&q, &questions, None, Timestamp::now());
assert_eq!(report.answered, [t.id.clone()]);
assert!(
q.get(&t.id).is_err(),
"\"discard it\" (捨ててよい) must actually discard the task, not \
just leave it sitting held forever"
);
}
#[test]
fn an_answer_is_read_by_its_position_in_choices_not_by_the_callers_current_language() {
let (dir, q, questions) = store();
let ja_config = dir.path().join("ja.toml");
std::fs::write(&ja_config, "[graph]\nlanguage = \"ja\"\n").unwrap();
let mut t = task("gate went red", dir.path().join("repo"));
t.hold_machine(Some("gate red".to_owned()));
q.put(&mut t).unwrap();
run_once(&q, &questions, Some(&ja_config), Timestamp::now());
let mut asked = questions
.list()
.into_iter()
.find(|q| q.run == t.id)
.unwrap();
assert_eq!(asked.choices[0], JA.resume, "filed in Japanese");
asked.answer(Answer::Choice(JA.resume.to_owned())).unwrap();
questions.put(&mut asked).unwrap();
let en_config = dir.path().join("en.toml");
std::fs::write(&en_config, "[graph]\nlanguage = \"en\"\n").unwrap();
let report = run_once(&q, &questions, Some(&en_config), Timestamp::now());
assert_eq!(report.answered, [t.id.clone()]);
let back = q.get(&t.id).unwrap();
assert_eq!(
back.status,
TaskStatus::Queued,
"a resume answer must resume the task regardless of which \
language it is read back in"
);
}
#[test]
fn a_question_falls_back_to_last_error_when_hold_reason_was_never_set() {
let (dir, q, questions) = store();
let mut t = task("kept failing the gate", dir.path().join("repo"));
t.start("run-1".to_owned());
t.fail("gate red three times running", 1);
assert_eq!(t.status, TaskStatus::Held);
assert!(t.hold_reason.is_none(), "the case this test is about");
q.put(&mut t).unwrap();
run_once(&q, &questions, None, Timestamp::now());
let asked = questions
.list()
.into_iter()
.find(|q| q.run == t.id)
.unwrap();
assert!(
asked.detail.contains("gate red three times running"),
"the question must surface `last_error` when there is no \
`hold_reason` to show instead: {}",
asked.detail
);
}
#[test]
fn open_question_for_and_open_task_ids_reflect_only_what_is_still_waiting() {
let (dir, q, questions) = store();
let mut t = task("gate went red", dir.path().join("repo"));
t.hold_machine(Some("gate red".to_owned()));
q.put(&mut t).unwrap();
assert!(open_question_for(&questions, &t.id).is_none());
assert!(!open_task_ids(&questions).contains(&t.id));
run_once(&q, &questions, None, Timestamp::now());
assert!(open_question_for(&questions, &t.id).is_some());
assert!(open_task_ids(&questions).contains(&t.id));
let mut asked = questions
.list()
.into_iter()
.find(|q| q.run == t.id)
.unwrap();
asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
questions.put(&mut asked).unwrap();
assert!(
open_question_for(&questions, &t.id).is_none(),
"an answered question is no longer open"
);
assert!(!open_task_ids(&questions).contains(&t.id));
}
}