Skip to main content

magi/
triage.rs

1//! Held-task triage: walking `held` tasks so a hold left by an accident does
2//! not sit unread forever next to one a human placed on purpose.
3//!
4//! [`crate::queue::HoldSource`] already distinguishes "the daemon or
5//! conductor held this during its own recovery" ([`HoldSource::Machine`],
6//! documented as recoverable) from "an operator held this on purpose"
7//! ([`HoldSource::Manual`]). What was missing was anything that actually acts
8//! on that distinction: nothing walked the held list and asked whether a
9//! machine hold's cause was still true, and a record written before
10//! `hold_source` existed (schema < 3, `None`) was silently protected forever
11//! by [`Task::operator_held`]'s conservative default - never wrong, but also
12//! never looked at again by anything.
13//!
14//! [`run_once`] is that walk. For every `held` task it finds:
15//!
16//! - [`HoldSource::Machine`]: if [`machine_cause_resolved`] can tell the
17//!   cause is gone, the task goes straight back to `queued` - the same effect
18//!   as `magi task release`, just automatic. When it cannot tell, a
19//!   [`Question`] is filed once and the task stays held until answered.
20//! - `None` (a legacy record, or a hold nobody explained): always a question,
21//!   exactly once - the whole point being that "protected forever" must not
22//!   mean "never shown to anyone" either.
23//! - [`HoldSource::Manual`]: never touched automatically. Only once the hold
24//!   has sat untouched past [`MANUAL_STALE_AFTER`] does it earn a question of
25//!   its own, asking whether it is still wanted.
26//!
27//! Every question this module files carries [`NODE`] and uses the task id as
28//! [`Question::run`] - the same convention `crate::conduct`'s own questions
29//! use for a task rather than a run (see `conduct::apply_one`'s own comment
30//! on why the dedupe check there also filters on `node`, not `run` alone: an
31//! ordinary graph question's `run` is a real run id, and a coincidental
32//! equality with some task's id must not be read as "about this task").
33//! [`latest_triage_question`] follows the identical rule.
34//!
35//! # Why answers apply here rather than through `crate::queue::Task::block`
36//!
37//! `crate::conduct` blocks a task on its own question
38//! (`Task::block(vec![question_id], …)`), and `crate::daemon::resolve_blockers`
39//! unblocks it - unconditionally, back to `queued` - the moment that question
40//! is answered, whatever the answer actually said. That is correct for
41//! `conduct`: the *content* of the answer is meant for whoever reads
42//! `Task::answers` next, not for the resolver.
43//!
44//! A triage question's answer is different: "not yet" and "discard it" do two
45//! entirely different, non-resuming things, and only "resume it" may put the
46//! task back in line. Reusing the generic blocked/unblock path would resume
47//! every answer alike, so this module never calls [`Task::block`] and never
48//! leaves a triaged task anything but `held` while its question is open.
49//! [`interpret_answer`] reads [`Question::resolution`] itself and
50//! [`run_once`] acts on it directly: [`Task::release`] for an actual "resume
51//! it" choice, [`Queue::remove`] for "discard it" (捨ててよい really means
52//! "you may throw this away", not "leave it sitting held" - the English
53//! wording must say the same thing, not "leave it held"), and
54//! [`Task::hold_manual`] for anything else - which both keeps the task held
55//! and reclassifies it as a hold an operator has now actually seen, one
56//! `crate::conduct` and a later triage pass leave alone.
57//!
58//! The choice is read by its **position** in [`Question::choices`]
59//! ([`Wording::choices3`]/[`Wording::choices2`] always put "resume" first and
60//! "discard" third), never by comparing the answer text against [`Wording`]'s
61//! own strings picked from whatever config is in force *now* - the language a
62//! question was filed in and the language a later `run_once` call happens to
63//! read back are not guaranteed to be the same call's [`Config`], and a text
64//! comparison would silently misread a real "resume" answer as "keep held"
65//! the moment they disagree.
66//!
67//! # Idempotency
68//!
69//! [`run_once`] runs on every daemon idle tick (see `crate::daemon::poll`)
70//! and on every `magi task triage`, so an answered question must be applied to
71//! a task **at most once**, and a *fresh* question for the same task (once it
72//! is held again, or goes stale) must still be possible. The record is
73//! [`Task::triage_applied`], the question ids already applied. It cannot live
74//! in [`Task::hold_reason`]: [`Task::release`] clears that, so a "resume"
75//! answer left no trace, and a released task that failed back to `held` was
76//! released again by the same old answer with its attempts reset - forever.
77//! A task that comes back to `held` after an applied answer is therefore a
78//! new hold, handled per [`HoldSource`] (a fresh question for a machine hold).
79//! [`already_applied`] also still reads the `[triage:<short>]` marker
80//! [`keep_held_note`] appends to the hold reason, for records that pre-date
81//! the field.
82
83use std::path::{Path, PathBuf};
84use std::time::Duration;
85
86use jiff::Timestamp;
87
88use crate::ask::{Question, QuestionStatus, Questions};
89use crate::config::Config;
90use crate::disk;
91use crate::queue::{HoldSource, OperatorResume, Queue, Task, TaskStatus};
92
93/// Node recorded on every question this module files - `crate::conduct::NODE`
94/// for the same idea applied to a `crate::conduct` decision instead.
95pub const NODE: &str = "triage";
96
97/// Node on the question filed about a stuck dependency root - see
98/// [`ask_about_stuck_roots`]. Separate from [`NODE`] because the choices mean
99/// something else (position 2 is "detach the dependants", not "discard").
100pub const DEPS_NODE: &str = "triage-deps";
101
102/// Seat name on a filed question. Not a real agent seat - there is no model
103/// call anywhere in this module - but every [`Question`] needs one, and every
104/// other deterministic filer (`crate::land`'s merge approval) names itself
105/// the same way.
106const SEAT: &str = "triage";
107
108/// How long a [`HoldSource::Manual`] hold sits untouched before triage asks
109/// whether it is still wanted.
110///
111/// A judgement call, not a `magi.toml` setting - the same reasoning
112/// `ask::REPLY_QUIET_WINDOW` documents for itself: there is no operator
113/// preference for "how long is too long to ignore my own hold" that a
114/// per-repository config could be *right* about. Seven days is long enough
115/// that an ordinary multi-day hold (waiting on a dependency, waiting on the
116/// operator's own schedule) never gets nagged, and short enough that a hold
117/// nobody has looked at in a week surfaces again rather than aging into the
118/// kind of silent backlog this feature exists to prevent.
119const MANUAL_STALE_AFTER: Duration = Duration::from_secs(7 * 24 * 60 * 60);
120
121/// Localised strings for a filed question, the same idea as `crate::land`'s
122/// own `Words`/`words()` - only the languages magi can actually check are
123/// translated, and anything else falls back to English.
124struct Wording {
125    lang: &'static str,
126    resume: &'static str,
127    wait: &'static str,
128    discard: &'static str,
129    resume_now: &'static str,
130    keep_held: &'static str,
131}
132
133const EN: Wording = Wording {
134    lang: "en",
135    resume: "resume it",
136    wait: "not yet",
137    discard: "discard it",
138    resume_now: "resume it",
139    keep_held: "keep it held",
140};
141
142const JA: Wording = Wording {
143    lang: "ja",
144    resume: "再開してよい",
145    wait: "まだ待って",
146    discard: "捨ててよい",
147    resume_now: "再開する",
148    keep_held: "まだ止めておく",
149};
150
151/// Pick the wording. Codes and names both, the same acceptance
152/// `crate::land::words` gives `[graph] language`.
153fn wording(language: &str) -> &'static Wording {
154    if crate::lang::is_japanese(language) {
155        &JA
156    } else {
157        &EN
158    }
159}
160
161impl Wording {
162    fn choices3(&self) -> Vec<String> {
163        vec![
164            self.resume.to_owned(),
165            self.wait.to_owned(),
166            self.discard.to_owned(),
167        ]
168    }
169
170    fn choices2(&self) -> Vec<String> {
171        vec![self.resume_now.to_owned(), self.keep_held.to_owned()]
172    }
173
174    fn source_label(&self, source: Option<HoldSource>) -> &'static str {
175        match (self.lang, source) {
176            ("ja", Some(HoldSource::Machine)) => "machine(機械による自動保留)",
177            ("ja", Some(HoldSource::Manual)) => "manual(操作者による手動保留)",
178            ("ja", None) => "unknown(schema 3 未満の旧レコード、または理由未記録)",
179            (_, Some(HoldSource::Machine)) => "machine (automatic recovery hold)",
180            (_, Some(HoldSource::Manual)) => "manual (an operator held this)",
181            (_, None) => "unknown (pre-schema-3 record, or never recorded)",
182        }
183    }
184
185    /// The body under the summary: everything an operator needs to judge this
186    /// without opening a terminal - id, title, hold reason, hold source.
187    ///
188    /// Falls back to [`Task::last_error`] when [`Task::hold_reason`] is empty:
189    /// a record written before `Task::fail`/`Task::handed_off` started
190    /// copying `why` into `hold_reason` too, or one written by a still older
191    /// build, would otherwise leave the question blank for exactly the case
192    /// requirement 4 exists for.
193    fn detail(&self, task: &Task, why: &str) -> String {
194        let none = if self.lang == "ja" {
195            "(記録なし)"
196        } else {
197            "(none recorded)"
198        };
199        let reason = task
200            .hold_reason
201            .as_deref()
202            .or(task.last_error.as_deref())
203            .unwrap_or(none);
204        format!(
205            "task: {} ({})\ntitle: {}\nhold source: {}\nhold reason: {reason}\n\n{why}",
206            task.id,
207            task.short(),
208            task.title,
209            self.source_label(task.hold_source),
210        )
211    }
212
213    fn summary_machine_unknown(&self, task: &Task) -> String {
214        if self.lang == "ja" {
215            format!("保留タスク {} の再開可否を判断してください", task.short())
216        } else {
217            format!("decide whether to resume held task {}", task.short())
218        }
219    }
220
221    fn why_machine(&self) -> &'static str {
222        if self.lang == "ja" {
223            "機械的な保留(machine hold)ですが、原因がすでに解消しているかを自動では判断できませんでした。"
224        } else {
225            "This is a machine hold, but whether its cause has resolved could not be \
226             checked automatically."
227        }
228    }
229
230    fn summary_conductor_override(&self, task: &Task) -> String {
231        if self.lang == "ja" {
232            format!(
233                "再開と回答済みのタスク {} を conductor が再び保留しました",
234                task.short()
235            )
236        } else {
237            format!(
238                "task {} was resumed at your word, but the conductor held it again",
239                task.short()
240            )
241        }
242    }
243
244    fn why_conductor_override(&self, o: &OperatorResume) -> String {
245        let reason = o.conductor_rehold.as_deref().unwrap_or_default();
246        if self.lang == "ja" {
247            format!(
248                "{} に再開と回答済みですが、conductor が再び hold しました。conductor の理由: \
249                 {reason}\n\n強制再キューを選ぶと、以後 conductor はこのタスクを hold できません。",
250                o.at
251            )
252        } else {
253            format!(
254                "You answered \"resume\" at {}, but the conductor held the task again. \
255                 Its reason: {reason}\n\nForcing a requeue stops the conductor from \
256                 holding this task again.",
257                o.at
258            )
259        }
260    }
261
262    /// Positions match [`AnswerAction`]: 0 resume, 1 keep held, 2 discard.
263    fn choices_conductor_override(&self) -> Vec<String> {
264        if self.lang == "ja" {
265            vec![
266                "強制再キュー(conductor は再 hold 不可)".to_owned(),
267                "手動 hold のまま".to_owned(),
268                "捨ててよい".to_owned(),
269            ]
270        } else {
271            vec![
272                "force requeue (conductor must not hold again)".to_owned(),
273                "keep held (manual)".to_owned(),
274                "discard".to_owned(),
275            ]
276        }
277    }
278
279    fn summary_legacy(&self, task: &Task) -> String {
280        if self.lang == "ja" {
281            format!(
282                "hold_source が不明な保留タスク {} を確認してください",
283                task.short()
284            )
285        } else {
286            format!(
287                "held task {} has no recorded hold source - please take a look",
288                task.short()
289            )
290        }
291    }
292
293    fn why_legacy(&self) -> &'static str {
294        if self.lang == "ja" {
295            "hold_source が記録されていません。schema 3 より前のレコードか、理由が記録されなかった \
296             holdです。人が意図して止めたのか、クラッシュや強制再起動で宙に浮いただけなのか、\
297             このデータからは区別できません。"
298        } else {
299            "No hold_source was recorded - either a pre-schema-3 record, or a hold whose \
300             reason was never written down. Whether this was a deliberate hold or the \
301             leftover of a crash cannot be told from the data alone."
302        }
303    }
304
305    fn summary_manual_stale(&self, task: &Task, days: i64) -> String {
306        if self.lang == "ja" {
307            format!(
308                "{days}日間 保留されたままの手動保留タスク {} を確認してください",
309                task.short()
310            )
311        } else {
312            format!(
313                "held task {} has been on a manual hold for {days} day(s)",
314                task.short()
315            )
316        }
317    }
318
319    fn why_manual(&self) -> &'static str {
320        if self.lang == "ja" {
321            "操作者が明示的に止めた保留ですが、長期間そのままになっています。まだ止めておくか、\
322             再開するか教えてください。"
323        } else {
324            "An operator held this on purpose, but it has sat untouched for a while. Say \
325             whether to keep holding it or resume it."
326        }
327    }
328}
329
330/// Which of the three situations this module recognises a held task is in.
331#[derive(Debug, Clone, Copy, PartialEq, Eq)]
332enum Bucket {
333    /// `HoldSource::Machine`, cause not verifiably resolved.
334    MachineUnknown,
335    /// `HoldSource::Machine`, held by the conductor after the operator
336    /// answered "resume" (see [`Task::resume_override`]).
337    ConductorOverride,
338    /// `hold_source` is `None`.
339    Legacy,
340    /// `HoldSource::Manual`, held past [`MANUAL_STALE_AFTER`].
341    ManualStale,
342}
343
344/// What one [`run_once`] pass did, task ids in each list.
345#[derive(Debug, Clone, Default)]
346pub struct Report {
347    /// A `HoldSource::Machine` hold whose cause was found resolved, put back
348    /// in line automatically.
349    pub resumed: Vec<String>,
350    /// A fresh question was filed this pass.
351    pub asked: Vec<String>,
352    /// An operator's answer to an earlier triage question was applied.
353    pub answered: Vec<String>,
354    /// A `blocked` task whose `blocked_by` named an id that no longer exists
355    /// was moved to a machine hold this pass - see
356    /// [`quarantine_orphaned_blocked`]. Distinct from `asked`: the question
357    /// about it, if any, is filed in the same pass and only counted there.
358    pub quarantined: Vec<String>,
359}
360
361impl Report {
362    /// Is there nothing to report? Callers use this to skip logging an empty
363    /// pass rather than repeating "triaged 0 held task(s)" on every idle tick.
364    pub fn is_empty(&self) -> bool {
365        self.resumed.is_empty()
366            && self.asked.is_empty()
367            && self.answered.is_empty()
368            && self.quarantined.is_empty()
369    }
370}
371
372/// The repository a task's config and disk check should read from. Mirrors
373/// `crate::conduct::repo_for`'s own fallback, duplicated rather than shared
374/// because that one is private to its module and the two are one `if` each.
375fn repo_for(task: &Task) -> PathBuf {
376    if task.repo.as_os_str().is_empty() {
377        PathBuf::from(".")
378    } else {
379        task.repo.clone()
380    }
381}
382
383/// Recognise a `HoldSource::Machine` hold caused by the free-space gate
384/// (`crate::disk::gate`'s message, or `daemon::disk_gate`'s "could not
385/// measure" fallback) from `hold_reason` text alone.
386///
387/// There is no field recording *why* a machine hold happened - `hold_machine`
388/// takes only a reason string - so this is the one signal available, and disk
389/// pressure is the one cause this module can safely re-measure without
390/// touching git, a run, or an agent CLI. Keep the two prefixes here in sync
391/// with `crate::disk::gate`'s formatted string and `daemon::disk_gate`'s own
392/// message if either changes; nothing else ties them together.
393fn is_disk_hold(task: &Task) -> bool {
394    task.hold_reason
395        .as_deref()
396        .is_some_and(crate::disk::is_gate_reason)
397}
398
399/// Has a `HoldSource::Machine` hold's cause resolved? `Some(true)` means yes -
400/// safe to requeue. `Some(false)` means the same cause was checked and is
401/// still in force. `None` means this hold's cause is not one this module
402/// knows how to re-check at all, and a human has to look.
403fn machine_cause_resolved(task: &Task, cfg: &Config) -> Option<bool> {
404    if task
405        .hold_reason
406        .as_deref()
407        .is_some_and(|r| r.starts_with(crate::cache::RESOURCE_HOLD))
408    {
409        // The shared build cache's lease: free again (or no cache at all)
410        // means the same run can resume.
411        return Some(
412            cfg.cache_dir()
413                .is_none_or(|dir| !crate::cache::in_use(&crate::run::home(), &dir)),
414        );
415    }
416    if !is_disk_hold(task) {
417        return None;
418    }
419    let min = cfg.disk.min_free_bytes;
420    if min == 0 {
421        // The operator turned the gate off since this hold was placed - its
422        // one possible cause is gone by construction, no measurement needed.
423        return Some(true);
424    }
425    let free = disk::free_bytes(&repo_for(task)).ok()?;
426    Some(disk::gate(free, min).is_none())
427}
428
429/// Is a `HoldSource::Manual` hold old enough to earn a "still wanted?"
430/// question? Same comparison `crate::clean::due` uses for a run's fold grace,
431/// against [`MANUAL_STALE_AFTER`] instead of a configured one.
432fn manual_is_stale(task: &Task, now: Timestamp) -> bool {
433    now.as_second() - task.updated_at.as_second() > MANUAL_STALE_AFTER.as_secs() as i64
434}
435
436/// The marker [`apply_answer`] writes into [`Task::hold_reason`] and
437/// [`already_applied`] reads back - see this module's own doc on why.
438fn marker_for(q: &Question) -> String {
439    format!("[triage:{}]", q.short())
440}
441
442/// Has `q`'s answer already been applied to `task`? See this module's doc.
443/// Checks [`Task::triage_applied`] first, which survives [`Task::release`].
444fn already_applied(task: &Task, q: &Question) -> bool {
445    if task.triage_applied(&q.id) {
446        return true;
447    }
448    // Records written before `Task::triage_applied` existed carry only the
449    // "keep held" marker in the hold reason.
450    let marker = marker_for(q);
451    task.hold_reason
452        .as_deref()
453        .is_some_and(|r| r.contains(marker.as_str()))
454}
455
456/// The most recent question this module filed for `task_id`, any status -
457/// open (still waiting), answered (may need applying), or abandoned (settled
458/// with nothing decided). Filters on both `node` and `run`, never `run`
459/// alone - see this module's doc on why a bare `run` match is not safe.
460fn latest_triage_question(questions: &Questions, task_id: &str) -> Option<Question> {
461    latest_question(questions, NODE, task_id)
462}
463
464/// [`latest_triage_question`] for any of this module's nodes.
465fn latest_question(questions: &Questions, node: &str, task_id: &str) -> Option<Question> {
466    questions
467        .list()
468        .into_iter()
469        .filter(|q| q.node == node && q.run == task_id)
470        // `asked_at` first: ids carry only whole seconds plus a random
471        // suffix, so two questions filed in the same second order randomly.
472        .max_by(|a, b| a.asked_at.cmp(&b.asked_at).then_with(|| a.id.cmp(&b.id)))
473}
474
475/// What an answered triage question's choice means, independent of which
476/// language it was filed in.
477#[derive(Debug, Clone, Copy, PartialEq, Eq)]
478enum AnswerAction {
479    /// The first choice, always "resume it" / "再開してよい" / "再開する" -
480    /// see [`Wording::choices3`] and [`Wording::choices2`], whose first entry
481    /// is always the resume choice.
482    Resume,
483    /// The third choice, present only in [`Wording::choices3`]: "discard it"
484    /// / "捨ててよい".
485    Discard,
486    /// Anything else: the second choice ("not yet" / "keep it held"), or an
487    /// answer that does not match any offered choice at all (should not
488    /// happen for a multiple-choice question, but the safe default is to
489    /// keep holding rather than guess at "resume").
490    KeepHeld,
491}
492
493/// Read `q`'s answer as an [`AnswerAction`].
494///
495/// Matched by **position in `q.choices`**, never by comparing the answer text
496/// against [`Wording`]'s own strings: [`Wording`] is picked from the task's
497/// *current* repository config, which can differ from whatever language was
498/// in force when the question was filed (a `--config` override on one `magi
499/// task triage` call and not the next, or the repository's config edited in
500/// between). Comparing text would then silently misread an actual "resume"
501/// answer as "keep held" - the choices themselves are fixed at filing time,
502/// in [`file_question`], and never change afterwards, so their position is
503/// the one thing that stays true regardless of which language reads them
504/// back.
505fn interpret_answer(q: &Question) -> AnswerAction {
506    let resolution = q.resolution().unwrap_or_default();
507    match q.choices.iter().position(|c| *c == resolution) {
508        Some(0) => AnswerAction::Resume,
509        Some(2) => AnswerAction::Discard,
510        _ => AnswerAction::KeepHeld,
511    }
512}
513
514/// The note [`already_applied`] looks for, appended to (never replacing)
515/// whatever [`Task::hold_reason`] already said - the original cause is still
516/// worth reading in `magi task show` after the operator answers "not yet",
517/// and [`Task::hold_manual`] would otherwise overwrite it outright.
518fn keep_held_note(task: &Task, q: &Question, resolution: &str) -> String {
519    let marker = format!("{} operator: {resolution}", marker_for(q));
520    match task.hold_reason.as_deref() {
521        Some(existing) if !existing.is_empty() => format!("{existing}\n{marker}"),
522        _ => marker,
523    }
524}
525
526/// File a fresh triage question for `task` and return it. The caller is
527/// responsible for having already established there is no open one - see
528/// [`latest_triage_question`] - so this never checks again.
529fn file_question(
530    questions: &Questions,
531    task: &Task,
532    bucket: Bucket,
533    w: &Wording,
534    now: Timestamp,
535) -> Option<Question> {
536    let (summary, why, choices) = match bucket {
537        Bucket::MachineUnknown => (
538            w.summary_machine_unknown(task),
539            w.why_machine().to_owned(),
540            w.choices3(),
541        ),
542        Bucket::ConductorOverride => (
543            w.summary_conductor_override(task),
544            task.resume_override
545                .as_ref()
546                .map(|o| w.why_conductor_override(o))
547                .unwrap_or_default(),
548            w.choices_conductor_override(),
549        ),
550        Bucket::Legacy => (
551            w.summary_legacy(task),
552            w.why_legacy().to_owned(),
553            w.choices3(),
554        ),
555        Bucket::ManualStale => {
556            let days = (now.as_second() - task.updated_at.as_second()) / (24 * 60 * 60);
557            (
558                w.summary_manual_stale(task, days),
559                w.why_manual().to_owned(),
560                w.choices2(),
561            )
562        }
563    };
564    let mut q = Question::new(
565        task.id.clone(),
566        NODE.to_owned(),
567        SEAT.to_owned(),
568        summary,
569        w.detail(task, &why),
570        choices,
571    );
572    questions.put(&mut q).ok()?;
573    Some(q)
574}
575
576/// What a deputy is told about a triage question: the task, why it is held and
577/// what each position does, in the words [`interpret_answer`] /
578/// [`interpret_deps_answer`] apply them. Written from the stored record; a task
579/// that no longer exists is said to be missing, never reconstructed.
580pub(crate) fn deputy_brief(q: &Question, queue: &Queue) -> String {
581    let task = queue.get(&q.run).ok();
582    let mut s = match &task {
583        Some(t) => format!(
584            "Task {} (`magi task show {}`), \"{}\", is {}. Hold source: {}. Reason: {}.",
585            t.id,
586            t.id,
587            t.title,
588            t.status.as_str(),
589            match t.hold_source {
590                Some(HoldSource::Machine) => "machine (an automatic hold)",
591                Some(HoldSource::Manual) => "manual (an operator held it)",
592                None => "unknown (an old record, treated as the operator's)",
593            },
594            t.hold_reason
595                .as_deref()
596                .or(t.last_error.as_deref())
597                .unwrap_or("(none recorded)")
598        ),
599        None => format!("Task {} no longer exists (its record is gone).", q.run),
600    };
601    s.push_str("\n\nWhat each option does, by position (the owner taps a button; you only read their words):");
602    let n = q.choices.len();
603    for (i, c) in q.choices.iter().enumerate() {
604        let effect = if q.node == DEPS_NODE {
605            match i {
606                0 => "the dependency is released so the loop runs it",
607                1 => {
608                    "the dependants are detached and the dependency task is DELETED; this cannot be undone"
609                }
610                2 => "the dependants are detached; the dependency stays as it is",
611                _ => "nothing is applied",
612            }
613        } else {
614            match i {
615                0 => "the task is released and requeued",
616                2 if n > 2 => "the task is DELETED; this cannot be undone",
617                _ => "the task stays held, with the answer noted",
618            }
619        };
620        s.push_str(&format!("\n- `{c}`: {effect}"));
621    }
622    s.push_str(
623        "\n\nThe daemon applies the answer on its next pass; you apply nothing. \
624         `magi ask --settle` is refused while the task is held by the operator (a \
625         manual hold, or an old record): an answer must not release or delete \
626         something a person parked. In that case do not retry it - ask the owner \
627         with `--thread` to tap the button themselves. The destructive option is \
628         accepted only for an unhedged, verbatim instruction.",
629    );
630    s
631}
632
633/// Move every `blocked` task whose `blocked_by` names a task or question id
634/// that no longer exists to a machine hold, before the per-`held` walk
635/// [`run_once`] does gets a look at it.
636///
637/// `crate::daemon::resolve_blockers` already catches the same situation on
638/// every idle poll, and [`Queue::remove`] already catches it the moment a
639/// dependency is deleted through `magi task rm` - both call the same
640/// [`crate::queue::missing_blockers`]/[`crate::queue::missing_blocker_hold_reason`]
641/// this does. This third copy exists because a dependency can also be deleted
642/// by hand (the file just removed from disk, not through either of those
643/// paths), and because a queue can carry a `blocked` task with a
644/// long-since-deleted dependency from *before* either catch above ever
645/// existed - and such a task is `blocked`, never `held`, so it is invisible
646/// to the rest of this module without this pass. Running it here, first, is
647/// also what makes `magi task triage` alone - with no daemon running at all -
648/// enough to fix one: the task lands `held` in this same call, and the
649/// ordinary loop below files its question in the very same pass.
650fn quarantine_orphaned_blocked(queue: &Queue, questions: &Questions) -> Vec<String> {
651    let mut quarantined = Vec::new();
652    for listed in queue.list() {
653        if listed.status != TaskStatus::Blocked || listed.blocked_by.is_empty() {
654            continue;
655        }
656        let Ok(_claim) = queue.claim(&listed.id) else {
657            continue;
658        };
659        let Ok(mut task) = queue.get(&listed.id) else {
660            continue;
661        };
662        if task.status != TaskStatus::Blocked {
663            continue;
664        }
665        let deleted = queue.apply_deleted_blockers(&mut task);
666        if !deleted.is_empty() && queue.put(&mut task).is_ok() {
667            for id in &deleted {
668                queue.note_dependency_deleted(&task, id);
669            }
670        }
671        if task.status != TaskStatus::Blocked {
672            continue;
673        }
674        let missing = crate::queue::missing_blockers(queue, questions, &task.blocked_by);
675        if missing.is_empty() {
676            continue;
677        }
678        let language = crate::lang::of_repo(&repo_for(&task));
679        task.hold_machine(Some(crate::queue::missing_blocker_hold_reason_in(
680            &task.blocked_by,
681            &missing,
682            &language,
683        )));
684        if queue.put(&mut task).is_ok() {
685            quarantined.push(task.id.clone());
686        }
687    }
688    quarantined
689}
690
691/// Choices of a [`DEPS_NODE`] question, by position: release the root, discard
692/// it, or detach the dependants from it.
693#[derive(Debug, Clone, Copy, PartialEq, Eq)]
694enum DepsAction {
695    Release,
696    Discard,
697    Detach,
698    /// An answer that matches no offered choice: nothing is changed.
699    Nothing,
700}
701
702fn interpret_deps_answer(q: &Question) -> DepsAction {
703    let resolution = q.resolution().unwrap_or_default();
704    match q.choices.iter().position(|c| *c == resolution) {
705        Some(0) => DepsAction::Release,
706        Some(1) => DepsAction::Discard,
707        Some(2) => DepsAction::Detach,
708        _ => DepsAction::Nothing,
709    }
710}
711
712/// Summary, body and choices of the question about one stuck root.
713fn deps_texts(w: &Wording, root: &Task, dependants: &[&Task]) -> (String, String, Vec<String>) {
714    let ja = w.lang == "ja";
715    let reason = root
716        .hold_reason
717        .as_deref()
718        .or(root.last_error.as_deref())
719        .unwrap_or(if ja {
720            "(記録なし)"
721        } else {
722            "(none recorded)"
723        });
724    let list: String = dependants
725        .iter()
726        .map(|t| format!("- {} {}\n", t.short(), t.title))
727        .collect();
728    if ja {
729        (
730            format!(
731                "{} ({}) が {} 件のタスクを止めています",
732                root.short(),
733                root.status.as_str(),
734                dependants.len()
735            ),
736            format!(
737                "task: {} ({})\ntitle: {}\n状態: {}\n理由: {reason}\n\n\
738                 このタスクは自動では実行されないため、依存している次のタスクは永遠に待ち続けます:\n{list}",
739                root.id,
740                root.short(),
741                root.title,
742                root.status.as_str()
743            ),
744            vec![
745                "依存先を再開する".to_owned(),
746                "依存先を捨てる(依存タスクは切り離して実行)".to_owned(),
747                "依存タスクを切り離す(依存先はそのまま)".to_owned(),
748            ],
749        )
750    } else {
751        (
752            format!(
753                "{} ({}) is freezing {} blocked task(s)",
754                root.short(),
755                root.status.as_str(),
756                dependants.len()
757            ),
758            format!(
759                "task: {} ({})\ntitle: {}\nstatus: {}\nreason: {reason}\n\n\
760                 Nothing in the loop will ever run this task, so these dependants wait \
761                 forever:\n{list}",
762                root.id,
763                root.short(),
764                root.title,
765                root.status.as_str()
766            ),
767            vec![
768                "release the dependency".to_owned(),
769                "discard the dependency (dependants are detached and run)".to_owned(),
770                "detach the dependants (the dependency stays as it is)".to_owned(),
771            ],
772        )
773    }
774}
775
776/// Detach every `blocked` task that names `root` directly: drop that one id
777/// from its `blocked_by` (`Task::unblock`), so a dependant with another
778/// unresolved blocker keeps waiting on it. Idempotent.
779fn detach_dependants(queue: &Queue, root: &str) {
780    for listed in queue.list() {
781        if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| b == root) {
782            continue;
783        }
784        let Ok(_claim) = queue.claim(&listed.id) else {
785            continue;
786        };
787        let Ok(mut t) = queue.get(&listed.id) else {
788            continue;
789        };
790        if t.status != TaskStatus::Blocked {
791            continue;
792        }
793        t.unblock(root);
794        let _ = queue.put(&mut t);
795    }
796}
797
798/// Apply an answered [`DEPS_NODE`] question. Every step is idempotent, so a
799/// pass that dies half way is finished by the next one. Returns whether the
800/// answer was consumed.
801fn apply_deps_answer(queue: &Queue, questions: &Questions, q: &Question, now: Timestamp) -> bool {
802    let Ok(_claim) = queue.claim(&q.run) else {
803        return false;
804    };
805    let Ok(mut root) = queue.get(&q.run) else {
806        return false;
807    };
808    if root.triage_applied(&q.id) {
809        return false;
810    }
811    match interpret_deps_answer(q) {
812        DepsAction::Release => {
813            if root.status == TaskStatus::Running {
814                return false;
815            }
816            root.release();
817            root.resume_override = Some(OperatorResume {
818                question_id: q.id.clone(),
819                at: now,
820                conductor_rehold: None,
821                forced: false,
822                pinned_run: None,
823            });
824        }
825        DepsAction::Discard => {
826            detach_dependants(queue, &root.id);
827            return queue.remove(&root.id, false, questions).is_ok();
828        }
829        DepsAction::Detach => detach_dependants(queue, &root.id),
830        // Matches no offered choice, so there is nothing to apply. Left unmarked:
831        // marking it applied would read as "settled", and the next stuck check
832        // would file an identical question at once. Unmarked, it stays an
833        // answered question awaiting application, which suppresses re-asking.
834        DepsAction::Nothing => return false,
835    }
836    root.mark_triage_applied(&q.id);
837    queue.put(&mut root).is_ok()
838}
839
840/// One question per stuck dependency root, and apply the answers to earlier
841/// ones. See [`crate::blockers`] for what "stuck" means.
842///
843/// Not re-asked: a root with an open or answered-unapplied question; one whose
844/// own hold question ([`NODE`]) is still pending; and one whose question was
845/// abandoned unless the root or a task it freezes changed since. An *applied*
846/// answer leaves nothing stuck (released, discarded, or detached), so a root
847/// that is stuck again is a new situation and is asked about afresh.
848fn ask_about_stuck_roots(
849    queue: &Queue,
850    questions: &Questions,
851    config_override: Option<&Path>,
852    now: Timestamp,
853    report: &mut Report,
854) {
855    for q in questions.list() {
856        if q.node == DEPS_NODE
857            && q.status == QuestionStatus::Answered
858            && apply_deps_answer(queue, questions, &q, now)
859        {
860            report.answered.push(q.run.clone());
861        }
862    }
863
864    let all = questions.list();
865    let inv = crate::blockers::Inventory::new(queue.list(), &all);
866    let mut frozen: std::collections::BTreeMap<String, Vec<String>> = Default::default();
867    for (id, roots) in inv.stuck() {
868        for root in roots {
869            if root != id {
870                frozen.entry(root).or_default().push(id.clone());
871            }
872        }
873    }
874    // A cycle root freezes the others but may have no dependants of its own
875    // listed above when it is alone; a root with nothing to name is not asked.
876
877    for mut q in all
878        .iter()
879        .filter(|q| q.node == DEPS_NODE && q.status.open())
880        .cloned()
881    {
882        if !frozen.contains_key(&q.run) {
883            q.abandon("nothing waits on this task anymore");
884            let _ = questions.put(&mut q);
885        }
886    }
887
888    for (root_id, dependant_ids) in &frozen {
889        let Some(root) = inv.task(root_id) else {
890            continue;
891        };
892        let dependants: Vec<&Task> = dependant_ids.iter().filter_map(|d| inv.task(d)).collect();
893        let latest = all
894            .iter()
895            .filter(|q| q.node == DEPS_NODE && q.run == *root_id)
896            .max_by(|a, b| a.asked_at.cmp(&b.asked_at).then_with(|| a.id.cmp(&b.id)));
897        match latest {
898            Some(q) if q.status.open() => continue,
899            Some(q) if q.status == QuestionStatus::Answered && !root.triage_applied(&q.id) => {
900                continue;
901            }
902            Some(q) if q.status == QuestionStatus::Abandoned => {
903                let changed = root.updated_at > q.asked_at
904                    || dependants.iter().any(|t| t.updated_at > q.asked_at);
905                if !changed {
906                    continue;
907                }
908            }
909            _ => {}
910        }
911        if pending_for(questions, root) {
912            continue;
913        }
914        let cfg = Config::discover(&repo_for(root), config_override)
915            .ok()
916            .map(|(c, _)| c);
917        let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
918        let (summary, detail, choices) = deps_texts(w, root, &dependants);
919        let mut question = Question::new(
920            root.id.clone(),
921            DEPS_NODE.to_owned(),
922            SEAT.to_owned(),
923            summary,
924            detail,
925            choices,
926        );
927        if questions.put(&mut question).is_ok() {
928            report.asked.push(root.id.clone());
929        }
930    }
931}
932
933/// Run one deterministic triage pass over every `held` task in `queue`. No
934/// model call anywhere in this function - see this module's own doc for what
935/// each `HoldSource` gets instead.
936///
937/// `config_override` is threaded straight to [`Config::discover`], the same
938/// role `daemon::Opts::config` plays for `daemon::prepare` - an explicit
939/// `--config` from the caller, or `None` to let each task's own repository
940/// pick its layers.
941///
942/// Safe to call on every daemon idle tick and from `magi task triage` alike:
943/// a task already answered and applied is left alone (see
944/// [`already_applied`]), and a task with an open question is left alone too,
945/// so repeated calls with nothing new to say do nothing.
946///
947/// Also runs [`quarantine_orphaned_blocked`] first, so a `blocked` task whose
948/// dependency no longer exists is caught and turned into a fresh `held`
949/// question in this same pass, not left for a later call to notice.
950pub fn run_once(
951    queue: &Queue,
952    questions: &Questions,
953    config_override: Option<&Path>,
954    now: Timestamp,
955) -> Report {
956    let mut report = Report {
957        quarantined: quarantine_orphaned_blocked(queue, questions),
958        ..Report::default()
959    };
960    ask_about_stuck_roots(queue, questions, config_override, now, &mut report);
961    for listed in queue.list() {
962        if listed.status != TaskStatus::Held {
963            continue;
964        }
965        let Ok(_claim) = queue.claim(&listed.id) else {
966            continue;
967        };
968        let Ok(mut task) = queue.get(&listed.id) else {
969            continue;
970        };
971        // Re-read under the claim: a release or a re-hold landed by a human
972        // between the listing above and the claim just taken must not be
973        // clobbered by a decision based on the stale copy.
974        if task.status != TaskStatus::Held {
975            continue;
976        }
977        // A question about what this task freezes is already the one question
978        // it owes the operator; a second, about the hold itself, would ask two
979        // things at once.
980        if deps_pending(questions, &task) {
981            continue;
982        }
983
984        let cfg = Config::discover(&repo_for(&task), config_override)
985            .ok()
986            .map(|(c, _)| c);
987        let w = wording(cfg.as_ref().map_or("en", |c| c.graph.language.as_str()));
988
989        if let Some(q) = latest_triage_question(questions, &task.id) {
990            if q.status.open() {
991                // Already asked, still waiting - nothing to do this pass.
992                continue;
993            }
994            if q.status == QuestionStatus::Answered && !already_applied(&task, &q) {
995                match interpret_answer(&q) {
996                    AnswerAction::Resume => {
997                        // A second "resume", to the question about the
998                        // conductor's re-hold, forces it: the conductor may
999                        // not hold this task again. Any other resume records
1000                        // the answer so a re-hold can be recognised.
1001                        let contradicted = task
1002                            .resume_override
1003                            .as_ref()
1004                            .is_some_and(|o| o.conductor_rehold.is_some());
1005                        let record = match task.resume_override.take() {
1006                            Some(mut o) if contradicted => {
1007                                o.forced = true;
1008                                o
1009                            }
1010                            _ => OperatorResume {
1011                                question_id: q.id.clone(),
1012                                at: now,
1013                                conductor_rehold: None,
1014                                forced: false,
1015                                pinned_run: None,
1016                            },
1017                        };
1018                        task.release();
1019                        task.resume_override = Some(record);
1020                        task.mark_triage_applied(&q.id);
1021                        if queue.put(&mut task).is_ok() {
1022                            report.answered.push(task.id.clone());
1023                        }
1024                    }
1025                    AnswerAction::Discard => {
1026                        if queue.remove(&task.id, false, questions).is_ok() {
1027                            report.answered.push(task.id.clone());
1028                        }
1029                    }
1030                    AnswerAction::KeepHeld => {
1031                        let resolution = q.resolution().unwrap_or_default();
1032                        let note = keep_held_note(&task, &q, &resolution);
1033                        task.hold_manual(Some(note));
1034                        task.mark_triage_applied(&q.id);
1035                        if queue.put(&mut task).is_ok() {
1036                            report.answered.push(task.id.clone());
1037                        }
1038                    }
1039                }
1040                continue;
1041            }
1042            // Abandoned, or an already-applied answer: fall through to the
1043            // ordinary per-source handling below, which is how a stale
1044            // `HoldSource::Manual` re-ask - or a fresh machine/legacy
1045            // question, once a prior one settled the task back into a hold -
1046            // gets filed.
1047        }
1048
1049        match task.hold_source {
1050            Some(HoldSource::Machine) => {
1051                let overridden = task
1052                    .resume_override
1053                    .as_ref()
1054                    .is_some_and(|o| o.conductor_rehold.is_some() && !o.forced);
1055                if overridden {
1056                    if file_question(questions, &task, Bucket::ConductorOverride, w, now).is_some()
1057                    {
1058                        report.asked.push(task.id.clone());
1059                    }
1060                } else if cfg.as_ref().and_then(|c| machine_cause_resolved(&task, c)) == Some(true)
1061                {
1062                    task.release();
1063                    if queue.put(&mut task).is_ok() {
1064                        report.resumed.push(task.id.clone());
1065                    }
1066                } else if file_question(questions, &task, Bucket::MachineUnknown, w, now).is_some()
1067                {
1068                    report.asked.push(task.id.clone());
1069                }
1070            }
1071            None => {
1072                if file_question(questions, &task, Bucket::Legacy, w, now).is_some() {
1073                    report.asked.push(task.id.clone());
1074                }
1075            }
1076            Some(HoldSource::Manual) => {
1077                if manual_is_stale(&task, now)
1078                    && file_question(questions, &task, Bucket::ManualStale, w, now).is_some()
1079                {
1080                    report.asked.push(task.id.clone());
1081                }
1082            }
1083        }
1084    }
1085    report
1086}
1087
1088/// The open triage question about `task_id`, if any - what `magi task show`
1089/// prints so a held task's card names the question waiting on it, not only
1090/// its hold reason. `None` once it is answered or abandoned: nothing is
1091/// waiting on it anymore.
1092pub fn open_question_for(questions: &Questions, task_id: &str) -> Option<Question> {
1093    latest_triage_question(questions, task_id).filter(|q| q.status.open())
1094}
1095
1096/// Does this module still have unfinished business with `task`?
1097///
1098/// True while its latest triage question is still open (waiting on an
1099/// answer), and true for a beat longer than [`open_question_for`] alone
1100/// would say: once answered, the question sits [`QuestionStatus::Answered`]
1101/// but unread until the next [`run_once`] pass actually applies it (see
1102/// [`already_applied`]), and [`run_once`] only ever runs on a fully idle
1103/// daemon tick - far less often than `crate::conduct` polls. A caller that
1104/// only checked "is a question open" would walk straight through that gap
1105/// the moment the operator answers, moving the task out of `held` before
1106/// [`run_once`] gets a turn - orphaning the very answer it was about to
1107/// apply, the same failure mode this function exists to keep `crate::conduct`
1108/// out of. `crate::conduct::apply_one` is exactly that caller.
1109pub fn pending_for(questions: &Questions, task: &Task) -> bool {
1110    let own = match latest_triage_question(questions, &task.id) {
1111        Some(q) if q.status.open() => true,
1112        Some(q) if q.status == QuestionStatus::Answered => !already_applied(task, &q),
1113        _ => false,
1114    };
1115    own || deps_pending(questions, task)
1116}
1117
1118/// Is the latest [`DEPS_NODE`] question about `task` still open, or answered
1119/// but not yet applied? Same reasoning as [`pending_for`]: the gap between an
1120/// answer and the next [`run_once`] pass must not be walked through.
1121fn deps_pending(questions: &Questions, task: &Task) -> bool {
1122    match latest_question(questions, DEPS_NODE, &task.id) {
1123        Some(q) if q.status.open() => true,
1124        Some(q) if q.status == QuestionStatus::Answered => !task.triage_applied(&q.id),
1125        _ => false,
1126    }
1127}
1128
1129/// Every task id with an open triage question right now - what `magi task
1130/// list` uses to mark a held task that is already waiting on an operator
1131/// decision, rather than have it read identically to one nobody has looked
1132/// at yet.
1133pub fn open_task_ids(questions: &Questions) -> std::collections::BTreeSet<String> {
1134    questions
1135        .list()
1136        .into_iter()
1137        .filter(|q| (q.node == NODE || q.node == DEPS_NODE) && q.status.open())
1138        .map(|q| q.run)
1139        .collect()
1140}
1141
1142#[cfg(test)]
1143mod tests {
1144    use super::*;
1145    use crate::ask::Answer;
1146    use crate::queue::Source;
1147    use jiff::SignedDuration;
1148
1149    fn store() -> (tempfile::TempDir, Queue, Questions) {
1150        let dir = tempfile::tempdir().unwrap();
1151        let q = Queue::at(dir.path().join("queue"));
1152        let s = Questions::at(dir.path().join("questions"));
1153        (dir, q, s)
1154    }
1155
1156    fn task(title: &str, repo: PathBuf) -> Task {
1157        Task::new(title.to_owned(), format!("do {title}"), repo, Source::Human)
1158    }
1159
1160    /// `[disk] min_free_bytes = 0` written next to a fictional repo, so
1161    /// `machine_cause_resolved` never has to ask the real disk anything - the
1162    /// same fixture pattern `daemon`'s own idle-loop tests use.
1163    fn gate_disabled_config(dir: &std::path::Path) -> PathBuf {
1164        let config = dir.join("magi.toml");
1165        std::fs::write(&config, "[disk]\nmin_free_bytes = 0\n").unwrap();
1166        config
1167    }
1168
1169    #[test]
1170    fn a_resolved_machine_hold_is_requeued_automatically() {
1171        let (dir, q, questions) = store();
1172        let config = gate_disabled_config(dir.path());
1173        let mut t = task("disk pressure", dir.path().join("repo"));
1174        t.hold_machine(Some(
1175            "not enough free space to start a run: 10 bytes free, 100 required by \
1176             `[disk] min_free_bytes`"
1177                .to_owned(),
1178        ));
1179        q.put(&mut t).unwrap();
1180
1181        let report = run_once(&q, &questions, Some(&config), Timestamp::now());
1182        assert_eq!(report.resumed, [t.id.clone()]);
1183        assert!(report.asked.is_empty());
1184
1185        let back = q.get(&t.id).unwrap();
1186        assert_eq!(back.status, TaskStatus::Queued);
1187        assert!(back.hold_source.is_none());
1188        assert!(questions.list().is_empty(), "nothing needed asking");
1189    }
1190
1191    #[test]
1192    fn a_machine_hold_with_no_recognised_cause_gets_one_question_not_two() {
1193        let (dir, q, questions) = store();
1194        let mut t = task("gate went red", dir.path().join("repo"));
1195        t.hold_machine(Some("gate red".to_owned()));
1196        q.put(&mut t).unwrap();
1197
1198        let first = run_once(&q, &questions, None, Timestamp::now());
1199        assert_eq!(first.asked, [t.id.clone()]);
1200        assert!(first.resumed.is_empty());
1201
1202        let open: Vec<_> = questions
1203            .list()
1204            .into_iter()
1205            .filter(|q| q.status.open())
1206            .collect();
1207        assert_eq!(open.len(), 1);
1208        assert_eq!(open[0].run, t.id);
1209        assert_eq!(open[0].node, NODE);
1210        assert_eq!(open[0].choices.len(), 3);
1211
1212        // A second pass with nothing new must not file a second question.
1213        let second = run_once(&q, &questions, None, Timestamp::now());
1214        assert!(second.asked.is_empty());
1215        assert_eq!(
1216            questions
1217                .list()
1218                .into_iter()
1219                .filter(|q| q.status.open())
1220                .count(),
1221            1
1222        );
1223    }
1224
1225    #[test]
1226    fn a_legacy_hold_with_no_recorded_source_gets_exactly_one_question() {
1227        let (dir, q, questions) = store();
1228        let mut t = task("schema 1 record", dir.path().join("repo"));
1229        t.status = TaskStatus::Held;
1230        assert!(t.hold_source.is_none(), "the case this test is about");
1231        q.put(&mut t).unwrap();
1232
1233        let first = run_once(&q, &questions, None, Timestamp::now());
1234        assert_eq!(first.asked, [t.id.clone()]);
1235
1236        let second = run_once(&q, &questions, None, Timestamp::now());
1237        assert!(
1238            second.asked.is_empty(),
1239            "the same legacy hold must not be asked about twice"
1240        );
1241        assert_eq!(
1242            questions
1243                .list()
1244                .into_iter()
1245                .filter(|q| q.status.open())
1246                .count(),
1247            1
1248        );
1249    }
1250
1251    #[test]
1252    fn a_manual_hold_is_never_auto_resumed() {
1253        let (dir, q, questions) = store();
1254        let mut t = task("operator stopped this", dir.path().join("repo"));
1255        t.hold_manual(Some("waiting on a decision".to_owned()));
1256        q.put(&mut t).unwrap();
1257
1258        let report = run_once(&q, &questions, None, Timestamp::now());
1259        assert!(report.resumed.is_empty());
1260        // Fresh, not stale yet - no question either.
1261        assert!(report.asked.is_empty());
1262
1263        let back = q.get(&t.id).unwrap();
1264        assert_eq!(back.status, TaskStatus::Held);
1265        assert_eq!(back.hold_source, Some(HoldSource::Manual));
1266        assert!(questions.list().is_empty());
1267    }
1268
1269    #[test]
1270    fn a_blocked_task_on_a_deleted_dependency_is_held_and_asked_about_in_one_pass() {
1271        // The five real tasks this whole change exists for are `blocked`, not
1272        // `held`, and no daemon has to be running for `magi task triage` alone
1273        // to reach them - `run_once` must both quarantine and ask in the same
1274        // call.
1275        let (dir, q, questions) = store();
1276        let mut still_going = task("still valid", dir.path().join("repo"));
1277        q.put(&mut still_going).unwrap();
1278
1279        let mut t = task("orphaned", dir.path().join("repo"));
1280        t.block(
1281            vec!["20260101-000000-gone".to_owned(), still_going.id.clone()],
1282            Some("waits on both".to_owned()),
1283        );
1284        q.put(&mut t).unwrap();
1285
1286        let report = run_once(&q, &questions, None, Timestamp::now());
1287        assert_eq!(report.quarantined, [t.id.clone()]);
1288        assert_eq!(
1289            report.asked,
1290            [t.id.clone()],
1291            "the fresh machine hold must earn a question in the same pass"
1292        );
1293
1294        let after = q.get(&t.id).unwrap();
1295        assert_eq!(after.status, TaskStatus::Held);
1296        assert_eq!(after.hold_source, Some(HoldSource::Machine));
1297        assert!(after.blocked_by.is_empty());
1298
1299        // The reason is not disk-pressure wording, so this must not be read
1300        // as a disk hold and silently auto-resumed.
1301        assert!(!is_disk_hold(&after));
1302
1303        let open: Vec<_> = questions
1304            .list()
1305            .into_iter()
1306            .filter(|q| q.status.open())
1307            .collect();
1308        assert_eq!(open.len(), 1);
1309        assert_eq!(open[0].run, t.id);
1310
1311        // A second pass with nothing new files no second question.
1312        let second = run_once(&q, &questions, None, Timestamp::now());
1313        assert!(second.quarantined.is_empty());
1314        assert!(second.asked.is_empty());
1315    }
1316
1317    /// A held root with a chain of dependants: 9db7 <- 4135 <- 6081, and a
1318    /// second direct dependant that also waits on a queued task.
1319    fn stuck_chain(q: &Queue, dir: &std::path::Path) -> (Task, Task, Task) {
1320        let mut root = task("root", dir.join("repo"));
1321        root.hold_manual(Some("waiting".to_owned()));
1322        q.put(&mut root).unwrap();
1323        let mut mid = task("mid", dir.join("repo"));
1324        mid.block(vec![root.id.clone()], None);
1325        q.put(&mut mid).unwrap();
1326        let mut leaf = task("leaf", dir.join("repo"));
1327        leaf.block(vec![mid.id.clone()], None);
1328        q.put(&mut leaf).unwrap();
1329        (root, mid, leaf)
1330    }
1331
1332    fn deps_questions(questions: &Questions) -> Vec<Question> {
1333        questions
1334            .list()
1335            .into_iter()
1336            .filter(|q| q.node == DEPS_NODE)
1337            .collect()
1338    }
1339
1340    fn answer_deps(questions: &Questions, choice: usize) {
1341        let mut asked = deps_questions(questions).remove(0);
1342        let c = asked.choices[choice].clone();
1343        asked.answer(Answer::Choice(c)).unwrap();
1344        questions.put(&mut asked).unwrap();
1345    }
1346
1347    #[test]
1348    fn a_stuck_chain_earns_one_question_naming_every_dependant_and_is_not_reasked() {
1349        let (dir, q, questions) = store();
1350        let (root, mid, leaf) = stuck_chain(&q, dir.path());
1351
1352        let first = run_once(&q, &questions, None, Timestamp::now());
1353        assert_eq!(first.asked, std::slice::from_ref(&root.id));
1354        let asked = deps_questions(&questions);
1355        assert_eq!(asked.len(), 1, "one question per root, not per dependant");
1356        assert_eq!(asked[0].run, root.id);
1357        assert_eq!(asked[0].choices.len(), 3);
1358        assert!(asked[0].detail.contains(mid.short()), "{}", asked[0].detail);
1359        assert!(
1360            asked[0].detail.contains(leaf.short()),
1361            "{}",
1362            asked[0].detail
1363        );
1364
1365        let second = run_once(&q, &questions, None, Timestamp::now());
1366        assert!(second.asked.is_empty());
1367        assert_eq!(deps_questions(&questions).len(), 1);
1368        assert!(pending_for(&questions, &q.get(&root.id).unwrap()));
1369    }
1370
1371    #[test]
1372    fn a_stuck_root_with_a_machine_hold_gets_only_the_dependency_question() {
1373        let (dir, q, questions) = store();
1374        let mut root = task("root", dir.path().join("repo"));
1375        root.hold_machine(Some("gate red".to_owned()));
1376        q.put(&mut root).unwrap();
1377        let mut dep = task("dep", dir.path().join("repo"));
1378        dep.block(vec![root.id.clone()], None);
1379        q.put(&mut dep).unwrap();
1380
1381        run_once(&q, &questions, None, Timestamp::now());
1382        run_once(&q, &questions, None, Timestamp::now());
1383        assert_eq!(deps_questions(&questions).len(), 1);
1384        assert_eq!(
1385            questions.list().len(),
1386            1,
1387            "no second question about the hold"
1388        );
1389    }
1390
1391    #[test]
1392    fn a_dependency_that_can_still_run_asks_nothing() {
1393        let (dir, q, questions) = store();
1394        let mut live = task("live", dir.path().join("repo"));
1395        q.put(&mut live).unwrap();
1396        let mut dep = task("dep", dir.path().join("repo"));
1397        dep.block(vec![live.id.clone()], None);
1398        q.put(&mut dep).unwrap();
1399        let report = run_once(&q, &questions, None, Timestamp::now());
1400        assert!(report.asked.is_empty());
1401        assert!(questions.list().is_empty());
1402    }
1403
1404    #[test]
1405    fn answering_release_requeues_the_root_and_is_not_reasked() {
1406        let (dir, q, questions) = store();
1407        let (root, _mid, _leaf) = stuck_chain(&q, dir.path());
1408        run_once(&q, &questions, None, Timestamp::now());
1409        answer_deps(&questions, 0);
1410
1411        let report = run_once(&q, &questions, None, Timestamp::now());
1412        assert_eq!(report.answered, std::slice::from_ref(&root.id));
1413        assert!(report.asked.is_empty(), "applying must not re-ask");
1414        assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Queued);
1415        let again = run_once(&q, &questions, None, Timestamp::now());
1416        assert!(again.asked.is_empty() && again.answered.is_empty());
1417        assert_eq!(deps_questions(&questions).len(), 1);
1418    }
1419
1420    #[test]
1421    fn answering_detach_frees_the_direct_dependant_and_keeps_the_root() {
1422        let (dir, q, questions) = store();
1423        let (root, mid, leaf) = stuck_chain(&q, dir.path());
1424        run_once(&q, &questions, None, Timestamp::now());
1425        answer_deps(&questions, 2);
1426
1427        let report = run_once(&q, &questions, None, Timestamp::now());
1428        assert!(report.asked.is_empty());
1429        assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1430        assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1431        let leaf = q.get(&leaf.id).unwrap();
1432        assert_eq!(leaf.status, TaskStatus::Blocked, "still waits on mid");
1433        assert_eq!(leaf.blocked_by, std::slice::from_ref(&mid.id));
1434    }
1435
1436    #[test]
1437    fn a_dependant_that_can_progress_through_another_blocker_is_not_stuck() {
1438        let (dir, q, questions) = store();
1439        let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1440        let mut live = task("live", dir.path().join("repo"));
1441        q.put(&mut live).unwrap();
1442        let mut both = q.get(&mid.id).unwrap();
1443        both.block(vec![root.id.clone(), live.id.clone()], None);
1444        q.put(&mut both).unwrap();
1445        // `mid` can still progress through `live`, so nothing is stuck yet.
1446        assert!(
1447            run_once(&q, &questions, None, Timestamp::now())
1448                .asked
1449                .is_empty()
1450        );
1451    }
1452
1453    #[test]
1454    fn answering_discard_detaches_then_removes_the_root() {
1455        let (dir, q, questions) = store();
1456        let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1457        run_once(&q, &questions, None, Timestamp::now());
1458        answer_deps(&questions, 1);
1459
1460        let report = run_once(&q, &questions, None, Timestamp::now());
1461        assert!(q.get(&root.id).is_err());
1462        assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Queued);
1463        assert!(
1464            report.asked.is_empty(),
1465            "no per-dependant quarantine question"
1466        );
1467        assert!(report.quarantined.is_empty());
1468    }
1469
1470    #[test]
1471    fn an_unmatched_deps_answer_changes_nothing_and_is_not_reasked() {
1472        let (dir, q, questions) = store();
1473        let (root, mid, _leaf) = stuck_chain(&q, dir.path());
1474        run_once(&q, &questions, None, Timestamp::now());
1475        let mut asked = deps_questions(&questions).remove(0);
1476        asked.choices.clear();
1477        asked.answer(Answer::Text("dunno".to_owned())).unwrap();
1478        questions.put(&mut asked).unwrap();
1479
1480        for _ in 0..2 {
1481            let report = run_once(&q, &questions, None, Timestamp::now());
1482            assert!(report.asked.is_empty() && report.answered.is_empty());
1483        }
1484        assert_eq!(deps_questions(&questions).len(), 1);
1485        assert_eq!(q.get(&root.id).unwrap().status, TaskStatus::Held);
1486        assert_eq!(q.get(&mid.id).unwrap().status, TaskStatus::Blocked);
1487    }
1488
1489    #[test]
1490    fn a_dependency_cycle_terminates_and_is_asked_about_once() {
1491        let (dir, q, questions) = store();
1492        let mut a = task("a", dir.path().join("repo"));
1493        let mut b = task("b", dir.path().join("repo"));
1494        a.block(vec![b.id.clone()], None);
1495        b.block(vec![a.id.clone()], None);
1496        q.put(&mut a).unwrap();
1497        q.put(&mut b).unwrap();
1498
1499        let first = run_once(&q, &questions, None, Timestamp::now());
1500        assert_eq!(first.asked.len(), 1);
1501        let second = run_once(&q, &questions, None, Timestamp::now());
1502        assert!(second.asked.is_empty());
1503        assert_eq!(deps_questions(&questions).len(), 1);
1504    }
1505
1506    #[test]
1507    fn a_missing_dependency_is_still_quarantined_not_asked_about_as_stuck() {
1508        let (dir, q, questions) = store();
1509        let mut t = task("orphan", dir.path().join("repo"));
1510        t.block(vec!["20260101-000000-gone".to_owned()], None);
1511        q.put(&mut t).unwrap();
1512        let report = run_once(&q, &questions, None, Timestamp::now());
1513        assert_eq!(report.quarantined, [t.id.clone()]);
1514        assert!(deps_questions(&questions).is_empty());
1515    }
1516
1517    #[test]
1518    fn a_stale_manual_hold_earns_a_two_choice_question() {
1519        let (dir, q, questions) = store();
1520        let mut t = task("been sitting a while", dir.path().join("repo"));
1521        t.hold_manual(Some("waiting on a decision".to_owned()));
1522        q.put(&mut t).unwrap();
1523        // Back-date the hold past MANUAL_STALE_AFTER without waiting a week.
1524        let mut back = q.get(&t.id).unwrap();
1525        back.updated_at = Timestamp::now() - SignedDuration::new(8 * 24 * 60 * 60, 0);
1526        std::fs::write(
1527            q.path_of(&back.id),
1528            serde_json::to_string_pretty(&back).unwrap(),
1529        )
1530        .unwrap();
1531
1532        let report = run_once(&q, &questions, None, Timestamp::now());
1533        assert_eq!(report.asked, [t.id.clone()]);
1534        let open: Vec<_> = questions
1535            .list()
1536            .into_iter()
1537            .filter(|q| q.status.open())
1538            .collect();
1539        assert_eq!(open.len(), 1);
1540        assert_eq!(open[0].choices.len(), 2);
1541    }
1542
1543    #[test]
1544    fn answering_resume_releases_the_task() {
1545        let (dir, q, questions) = store();
1546        let mut t = task("gate went red", dir.path().join("repo"));
1547        t.hold_machine(Some("gate red".to_owned()));
1548        q.put(&mut t).unwrap();
1549        run_once(&q, &questions, None, Timestamp::now());
1550
1551        let mut asked = questions
1552            .list()
1553            .into_iter()
1554            .find(|q| q.run == t.id)
1555            .unwrap();
1556        asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1557        questions.put(&mut asked).unwrap();
1558
1559        let report = run_once(&q, &questions, None, Timestamp::now());
1560        assert_eq!(report.answered, [t.id.clone()]);
1561        let back = q.get(&t.id).unwrap();
1562        assert_eq!(back.status, TaskStatus::Queued);
1563        assert!(back.hold_source.is_none());
1564    }
1565
1566    /// A task released by a "resume" answer, run, and failed back to `held`.
1567    fn resumed_then_failed(q: &Queue, questions: &Questions, dir: &std::path::Path) -> Task {
1568        let mut t = task("gate went red", dir.join("repo"));
1569        t.hold_machine(Some("gate red".to_owned()));
1570        q.put(&mut t).unwrap();
1571        run_once(q, questions, None, Timestamp::now());
1572        let mut asked = questions
1573            .list()
1574            .into_iter()
1575            .find(|q| q.run == t.id)
1576            .unwrap();
1577        asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1578        questions.put(&mut asked).unwrap();
1579
1580        let report = run_once(q, questions, None, Timestamp::now());
1581        assert_eq!(report.answered, [t.id.clone()]);
1582        let mut back = q.get(&t.id).unwrap();
1583        assert_eq!(back.status, TaskStatus::Queued);
1584        assert_eq!(back.attempts, 0);
1585
1586        back.start("run-1".to_owned());
1587        back.fail("rebase conflict", 1);
1588        assert_eq!(back.status, TaskStatus::Held);
1589        q.put(&mut back).unwrap();
1590        back
1591    }
1592
1593    #[test]
1594    fn a_resume_answer_is_applied_once_and_a_new_machine_hold_is_asked_about() {
1595        let (dir, q, questions) = store();
1596        let t = resumed_then_failed(&q, &questions, dir.path());
1597
1598        let report = run_once(&q, &questions, None, Timestamp::now());
1599        assert!(report.answered.is_empty(), "the old answer must not replay");
1600        assert_eq!(report.asked, std::slice::from_ref(&t.id));
1601        let back = q.get(&t.id).unwrap();
1602        assert_eq!(back.status, TaskStatus::Held);
1603        assert_eq!(
1604            questions
1605                .list()
1606                .into_iter()
1607                .filter(|q| q.status.open())
1608                .count(),
1609            1
1610        );
1611    }
1612
1613    #[test]
1614    fn a_manual_hold_placed_after_a_resume_is_not_undone_by_the_old_answer() {
1615        let (dir, q, questions) = store();
1616        let mut t = resumed_then_failed(&q, &questions, dir.path());
1617        t.hold_manual(Some("operator stopped this".to_owned()));
1618        q.put(&mut t).unwrap();
1619        let before = questions.list().len();
1620
1621        let report = run_once(&q, &questions, None, Timestamp::now());
1622        assert!(report.answered.is_empty());
1623        assert!(report.asked.is_empty());
1624        let back = q.get(&t.id).unwrap();
1625        assert_eq!(back.status, TaskStatus::Held);
1626        assert_eq!(back.hold_source, Some(HoldSource::Manual));
1627        assert_eq!(questions.list().len(), before);
1628    }
1629
1630    #[test]
1631    fn answering_not_yet_keeps_it_held_as_a_manual_hold_and_does_not_reapply() {
1632        let (dir, q, questions) = store();
1633        let mut t = task("gate went red", dir.path().join("repo"));
1634        t.hold_machine(Some("gate red".to_owned()));
1635        q.put(&mut t).unwrap();
1636        run_once(&q, &questions, None, Timestamp::now());
1637
1638        let mut asked = questions
1639            .list()
1640            .into_iter()
1641            .find(|q| q.run == t.id)
1642            .unwrap();
1643        asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
1644        questions.put(&mut asked).unwrap();
1645
1646        let report = run_once(&q, &questions, None, Timestamp::now());
1647        assert_eq!(report.answered, [t.id.clone()]);
1648        let back = q.get(&t.id).unwrap();
1649        assert_eq!(back.status, TaskStatus::Held);
1650        assert_eq!(back.hold_source, Some(HoldSource::Manual));
1651        assert!(
1652            back.hold_reason
1653                .as_deref()
1654                .is_some_and(|r| r.contains("gate red")),
1655            "the original cause must survive a \"not yet\" answer, not just the \
1656             triage marker: {:?}",
1657            back.hold_reason
1658        );
1659
1660        // A third pass, same instant: the manual hold is fresh (just
1661        // touched), so nothing more happens - in particular the already
1662        // answered question is not re-applied.
1663        let third = run_once(&q, &questions, None, Timestamp::now());
1664        assert!(third.answered.is_empty());
1665        assert!(third.asked.is_empty());
1666    }
1667
1668    #[test]
1669    fn answering_discard_removes_the_task_entirely() {
1670        let (dir, q, questions) = store();
1671        let mut t = task("gate went red", dir.path().join("repo"));
1672        t.hold_machine(Some("gate red".to_owned()));
1673        q.put(&mut t).unwrap();
1674        run_once(&q, &questions, None, Timestamp::now());
1675
1676        let mut asked = questions
1677            .list()
1678            .into_iter()
1679            .find(|q| q.run == t.id)
1680            .unwrap();
1681        asked.answer(Answer::Choice(EN.discard.to_owned())).unwrap();
1682        questions.put(&mut asked).unwrap();
1683
1684        let report = run_once(&q, &questions, None, Timestamp::now());
1685        assert_eq!(report.answered, [t.id.clone()]);
1686        assert!(
1687            q.get(&t.id).is_err(),
1688            "\"discard it\" (捨ててよい) must actually discard the task, not \
1689             just leave it sitting held forever"
1690        );
1691    }
1692
1693    #[test]
1694    fn an_answer_is_read_by_its_position_in_choices_not_by_the_callers_current_language() {
1695        // Filed while the repo's config reads Japanese...
1696        let (dir, q, questions) = store();
1697        let ja_config = dir.path().join("ja.toml");
1698        std::fs::write(&ja_config, "[graph]\nlanguage = \"ja\"\n").unwrap();
1699        let mut t = task("gate went red", dir.path().join("repo"));
1700        t.hold_machine(Some("gate red".to_owned()));
1701        q.put(&mut t).unwrap();
1702        run_once(&q, &questions, Some(&ja_config), Timestamp::now());
1703
1704        let mut asked = questions
1705            .list()
1706            .into_iter()
1707            .find(|q| q.run == t.id)
1708            .unwrap();
1709        assert_eq!(asked.choices[0], JA.resume, "filed in Japanese");
1710        asked.answer(Answer::Choice(JA.resume.to_owned())).unwrap();
1711        questions.put(&mut asked).unwrap();
1712
1713        // ...but applied against an English config (a later `--config`, or an
1714        // edited repository config). The Japanese "再開してよい" answer must
1715        // still be read as a resume, not silently misread as "keep held"
1716        // because it fails a text comparison against the English wording.
1717        let en_config = dir.path().join("en.toml");
1718        std::fs::write(&en_config, "[graph]\nlanguage = \"en\"\n").unwrap();
1719        let report = run_once(&q, &questions, Some(&en_config), Timestamp::now());
1720        assert_eq!(report.answered, [t.id.clone()]);
1721        let back = q.get(&t.id).unwrap();
1722        assert_eq!(
1723            back.status,
1724            TaskStatus::Queued,
1725            "a resume answer must resume the task regardless of which \
1726             language it is read back in"
1727        );
1728    }
1729
1730    #[test]
1731    fn a_question_falls_back_to_last_error_when_hold_reason_was_never_set() {
1732        // A record written before `Task::fail` started copying `why` into
1733        // `hold_reason` too (or one written by a still older build) carries
1734        // only `last_error`. The question detail must still name a cause
1735        // rather than reading "(none recorded)".
1736        let (dir, q, questions) = store();
1737        let mut t = task("kept failing the gate", dir.path().join("repo"));
1738        t.start("run-1".to_owned());
1739        t.fail("gate red three times running", 1);
1740        assert_eq!(t.status, TaskStatus::Held);
1741        t.hold_reason = None; // simulate a pre-fix or pre-schema record
1742        q.put(&mut t).unwrap();
1743
1744        run_once(&q, &questions, None, Timestamp::now());
1745        let asked = questions
1746            .list()
1747            .into_iter()
1748            .find(|q| q.run == t.id)
1749            .unwrap();
1750        assert!(
1751            asked.detail.contains("gate red three times running"),
1752            "the question must surface `last_error` when there is no \
1753             `hold_reason` to show instead: {}",
1754            asked.detail
1755        );
1756    }
1757
1758    #[test]
1759    fn open_question_for_and_open_task_ids_reflect_only_what_is_still_waiting() {
1760        let (dir, q, questions) = store();
1761        let mut t = task("gate went red", dir.path().join("repo"));
1762        t.hold_machine(Some("gate red".to_owned()));
1763        q.put(&mut t).unwrap();
1764
1765        assert!(open_question_for(&questions, &t.id).is_none());
1766        assert!(!open_task_ids(&questions).contains(&t.id));
1767
1768        run_once(&q, &questions, None, Timestamp::now());
1769        assert!(open_question_for(&questions, &t.id).is_some());
1770        assert!(open_task_ids(&questions).contains(&t.id));
1771
1772        let mut asked = questions
1773            .list()
1774            .into_iter()
1775            .find(|q| q.run == t.id)
1776            .unwrap();
1777        asked.answer(Answer::Choice(EN.resume.to_owned())).unwrap();
1778        questions.put(&mut asked).unwrap();
1779
1780        assert!(
1781            open_question_for(&questions, &t.id).is_none(),
1782            "an answered question is no longer open"
1783        );
1784        assert!(!open_task_ids(&questions).contains(&t.id));
1785    }
1786
1787    #[test]
1788    fn pending_for_stays_true_between_an_answer_and_the_next_run_once_pass() {
1789        // `open_question_for` alone goes `None` the instant the operator
1790        // answers, well before `run_once` - idle-tick only - gets a turn to
1791        // actually apply that answer (see `already_applied`). `pending_for`
1792        // exists so a caller polling far more often than `run_once` does -
1793        // `crate::conduct::apply_one` - does not walk through that gap.
1794        let (dir, q, questions) = store();
1795        let mut t = task("gate went red", dir.path().join("repo"));
1796        t.hold_machine(Some("gate red".to_owned()));
1797        q.put(&mut t).unwrap();
1798
1799        assert!(!pending_for(&questions, &t));
1800
1801        run_once(&q, &questions, None, Timestamp::now());
1802        let held = q.get(&t.id).unwrap();
1803        assert!(pending_for(&questions, &held), "still waiting on an answer");
1804
1805        let mut asked = questions
1806            .list()
1807            .into_iter()
1808            .find(|q| q.run == t.id)
1809            .unwrap();
1810        asked.answer(Answer::Choice(EN.wait.to_owned())).unwrap();
1811        questions.put(&mut asked).unwrap();
1812        assert!(!asked.status.open());
1813
1814        // The race window: answered, but `run_once` has not run again yet.
1815        let still_held = q.get(&t.id).unwrap();
1816        assert!(
1817            pending_for(&questions, &still_held),
1818            "answered but not yet applied is still pending"
1819        );
1820
1821        run_once(&q, &questions, None, Timestamp::now());
1822        let after = q.get(&t.id).unwrap();
1823        assert!(
1824            !pending_for(&questions, &after),
1825            "the answer is applied now, nothing left pending"
1826        );
1827    }
1828}