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