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