Skip to main content

magi/
queue.rs

1//! The task queue: what magi should do next, and who asked for it.
2//!
3//! The queue is what lets magi run unattended. `magi serve` takes the next
4//! task, runs the graph on it, records the outcome, and takes the next one.
5//!
6//! It is also the reason an agent can ask for work. `magi task add` is the
7//! whole interface, and it is the same command whether a human types it at a
8//! prompt, a phone posts it through the web UI, or an implementer inside a run
9//! shells out to it because it noticed something worth doing but out of scope.
10//! magi's CLI is the operating surface for both kinds of user; the queue is
11//! where their intentions meet.
12//!
13//! One task is one JSON file under [`Queue`]'s root. Files rather than a
14//! database because the operator has to be able to read, edit, and delete the
15//! backlog with the tools already on the machine, and because a crashed daemon
16//! must leave a queue the next one can pick up without recovery ceremony.
17//!
18//! # Shape
19//!
20//! [`Task`] is data plus *pure* state transitions - [`Task::fail`] decides
21//! whether an attempt was the last one, and touches no disk. [`Queue`] owns all
22//! I/O and is constructed with its root, so a test drives a real queue in a
23//! temp directory without setting a process-global home. Splitting them this
24//! way is why the retry policy below can be asserted directly.
25//!
26//! # Bounded by construction
27//!
28//! An autonomous loop that retries forever is a way to spend money on a task
29//! that cannot succeed. Every claim increments [`Task::attempts`]; a task that
30//! has burned its attempts becomes [`TaskStatus::Held`] and waits for a human
31//! rather than for another agent.
32
33use std::collections::HashMap;
34use std::num::NonZeroUsize;
35use std::path::{Path, PathBuf};
36
37use anyhow::{Context, Result, bail};
38use jiff::Timestamp;
39use serde::{Deserialize, Serialize};
40
41use crate::ask::Questions;
42
43/// On-disk format for a queued task. Bumped when a field's meaning changes.
44///
45/// 18: retained review criteria snapshots survive operator edits.
46///
47/// 17: durable explicit operator resumption, independent of run startup.
48///
49/// 7: added [`Task::attachments`], names of files copied under
50/// `<id>.attachments/` beside the task file (a field only; `#[serde(default)]`,
51/// so an older record reads as empty and [`read_path`] still accepts it).
52///
53/// 6: added [`Task::resume_override`] (a field only; `#[serde(default)]`, so
54/// an older record reads as `None` and [`read_path`] still accepts it).
55///
56/// 5: added [`Task::triage_applied`], the ids of triage questions whose
57/// answer has already been applied to this task. A "resume" answer used to
58/// leave no trace ([`Task::release`] clears [`Task::hold_reason`], which was
59/// the only place the applied marker lived), so when the released task failed
60/// its attempts and went back to `held`, the next idle pass found the same
61/// answered question "not applied" and released it again with `attempts` reset
62/// to 0 - the `max_attempts` bound never held. `#[serde(default)]` so an older
63/// record reads as empty. A task already looping when this build arrives has
64/// no record, so it is released once more, recorded, and then stays held.
65///
66/// 8: added [`Task::followup`], the origin of a task `crate::followup` filed
67/// from a merged run's leftover findings. Field-only bump, `#[serde(default)]`.
68///
69/// 4: added [`Task::blocked_from`], the status a task had the moment it
70/// became [`TaskStatus::Blocked`], so [`Task::unblock`] restores it instead
71/// of always landing on [`TaskStatus::Queued`]. Without it, a task a human
72/// or `crate::triage` had deliberately left [`TaskStatus::Held`] — machine
73/// or manual — would lose that the instant `crate::conduct` blocked it on a
74/// follow-up question, and come back `Queued` the moment the question was
75/// answered, regardless of what the answer said: exactly the loop where a
76/// task the operator told to stay held instead re-enters the competition
77/// queue every time someone answers a question about it. `#[serde(default)]`
78/// so an older record reads as `None`; [`Task::unblock`] then falls back to
79/// inferring `Held` from surviving hold evidence ([`Task::hold_reason`] /
80/// [`Task::hold_source`], never cleared by [`Task::block`]) rather than
81/// guessing `Queued` outright — see [`Task::unblock`]'s own doc.
82///
83/// 9: added [`Task::held_at`], the start of the current hold, so a notice can
84/// tell a question filed for this hold from one about an older cause.
85/// Field-only, `#[serde(default)]`.
86///
87/// 10: added [`Task::overrides`] and [`Task::review_of`], so a run started by
88/// hand (`magi run` / `magi review`) is carried by a task without losing its
89/// command-line choices. Field-only.
90///
91/// 11: added [`TaskStatus::Parked`] and [`Task::park_reason`], so a run that
92/// stopped at a node boundary (an upgrade, an operator's park) is no longer
93/// recorded as a failed task. Variant plus `#[serde(default)]` field; an older
94/// build must not half-read the new status.
95///
96/// 12: added [`Task::origin_chat`], the talk a task descends from, so a
97/// follow-up still knows its chat after an ancestor task is deleted.
98/// Field-only, `#[serde(default)]`.
99///
100/// 13: added [`Task::chat_notice`], the record that a chat-filed task's final
101/// state was reported to its conversation. Field-only, `#[serde(default)]`;
102/// a record below this schema is never notified (an upgrade must not replay
103/// the whole history into chats).
104///
105/// 3: added [`HoldSource`] so conductor recovery cannot release a hold an
106/// operator deliberately placed. Old records default to `None` and are
107/// protected as operator-held until an explicit release; the safe direction
108/// when their author was never recorded.
109///
110/// 2: added [`TaskStatus::Blocked`], [`Task::blocked_by`] and
111/// [`Task::block_reason`] (`crate::conduct`'s decisions) and
112/// [`Task::answers`] (operator answers carried forward to the next
113/// conductor prompt and the next run's instruction). All three are
114/// `#[serde(default)]`, so [`read_path`] accepts anything up to and
115/// including this schema rather than only an exact match — a task written
116/// by a build that only knew about schema 1 has nothing to say about
117/// blocking or answers, and defaulting those fields is exactly as good a
118/// reading as a value that build never had a chance to write.
119///
120/// 15: added [`Task::passes`], what each settled pass over a run did to the
121/// attempt budget and why it stopped. Field-only, `#[serde(default)]`; a task
122/// from before this has no entry for its earlier passes, which stay unknown.
123///
124/// 14: record whether a task count edit's overrides inherit daemon options.
125/// Old overrides default to isolated command-line settings, including empty ones.
126///
127/// 16: added [`Task::attempt_limit`], an optional task-specific retry budget.
128/// Field-only, `#[serde(default)]`; old records inherit the daemon limit.
129pub const SCHEMA: u32 = 18;
130
131/// Who placed the current hold.
132#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
133#[serde(rename_all = "lowercase")]
134pub enum HoldSource {
135    /// An operator used the CLI or web UI.
136    Manual,
137    /// The daemon or conductor placed the hold as part of its own recovery.
138    Machine,
139}
140
141impl HoldSource {
142    /// Short human-facing label for reports and the CLI.
143    pub fn label(self) -> &'static str {
144        match self {
145            Self::Manual => "manual",
146            Self::Machine => "machine",
147        }
148    }
149}
150
151/// Where a task came from. Recorded because "who asked for this" is the first
152/// question about an autonomous run, and the answer is not recoverable later.
153#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(tag = "kind", rename_all = "lowercase")]
155pub enum Source {
156    /// A person, at a terminal or through the web UI.
157    Human,
158    /// An agent inside a run, via `magi task add`. Both ids are recorded so a
159    /// task can be traced back to the exact seat that asked for it.
160    Agent {
161        /// Run the asking agent belonged to.
162        run: String,
163        /// Node it was working in, e.g. `implement` or `review`.
164        node: String,
165    },
166    /// A GitHub issue, imported by number.
167    Issue {
168        /// Issue number.
169        number: u64,
170        /// `owner/repo`, as `gh` reports it.
171        repo: String,
172    },
173}
174
175/// The `node` a [`Source::Agent`] carries when the asker is a chat
176/// conversation (its `run` is then the conversation's id).
177pub const CHAT_NODE: &str = "chat";
178
179impl Source {
180    /// Short human-facing label, for lists and the web UI.
181    pub fn label(&self) -> String {
182        match self {
183            Self::Human => "human".to_owned(),
184            Self::Agent { run, node } => format!("{node}@{}", short(run)),
185            Self::Issue { number, .. } => format!("issue #{number}"),
186        }
187    }
188}
189
190/// Where a task is in its life.
191#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
192#[serde(rename_all = "lowercase")]
193pub enum TaskStatus {
194    /// Waiting to be claimed.
195    Queued,
196    /// Claimed by a daemon; a run is in flight.
197    Running,
198    /// A run finished and its gate passed.
199    Done,
200    /// A run finished without passing, and attempts remain.
201    Failed,
202    /// Out of attempts, or held by hand. The loop will not pick it up.
203    Held,
204    /// Waiting on another task or an unanswered question. See
205    /// [`Task::blocked_by`]. Set and cleared by `crate::conduct` and
206    /// `crate::daemon`'s deterministic resolver, never by hand.
207    Blocked,
208    /// The run stopped at a node boundary because the process was asked to
209    /// give itself back. Nothing failed: the run's work is intact and the next
210    /// loop resumes it. Runnable like [`TaskStatus::Queued`]; the reason is in
211    /// [`Task::park_reason`], never [`Task::last_error`].
212    Parked,
213}
214
215impl TaskStatus {
216    /// Is this task eligible for a daemon to claim?
217    pub fn runnable(self) -> bool {
218        matches!(self, Self::Queued | Self::Failed | Self::Parked)
219    }
220
221    /// Lowercase name, as it appears on disk and in the API.
222    pub fn as_str(self) -> &'static str {
223        match self {
224            Self::Queued => "queued",
225            Self::Running => "running",
226            Self::Done => "done",
227            Self::Failed => "failed",
228            Self::Held => "held",
229            Self::Blocked => "blocked",
230            Self::Parked => "parked",
231        }
232    }
233}
234
235/// How many tasks sit in each [`TaskStatus`], for a dashboard tile — never a
236/// per-task view, so it carries no ids.
237#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
238pub struct TaskCounts {
239    /// Waiting to be claimed.
240    pub queued: usize,
241    /// Claimed; a run is in flight.
242    pub running: usize,
243    /// A run finished and its gate passed.
244    pub done: usize,
245    /// A run finished without passing, and attempts remain.
246    pub failed: usize,
247    /// Out of attempts, or held by hand.
248    pub held: usize,
249    /// Waiting on another task or an unanswered question.
250    pub blocked: usize,
251    /// Run parked at a node boundary; resumable, not a failure.
252    pub parked: usize,
253}
254
255impl TaskCounts {
256    /// Tally `tasks` by status. A pure function over whatever [`Queue::list`]
257    /// already read, so it needs no root of its own and stays trivially
258    /// testable against a hand-built slice.
259    pub fn of(tasks: &[Task]) -> Self {
260        Self::of_statuses(tasks.iter().map(|task| task.status))
261    }
262
263    /// Count current statuses from a reconciled display projection.
264    pub(crate) fn of_statuses(statuses: impl IntoIterator<Item = TaskStatus>) -> Self {
265        let mut counts = Self::default();
266        for status in statuses {
267            match status {
268                TaskStatus::Queued => counts.queued += 1,
269                TaskStatus::Running => counts.running += 1,
270                TaskStatus::Done => counts.done += 1,
271                TaskStatus::Failed => counts.failed += 1,
272                TaskStatus::Held => counts.held += 1,
273                TaskStatus::Blocked => counts.blocked += 1,
274                TaskStatus::Parked => counts.parked += 1,
275            }
276        }
277        counts
278    }
279}
280
281/// One unit of work.
282#[derive(Debug, Clone, Serialize, Deserialize)]
283#[serde(deny_unknown_fields)]
284pub struct Task {
285    /// On-disk format version.
286    pub schema: u32,
287    /// Task id, e.g. `20260902-140501-a1b2`.
288    pub id: String,
289    /// One line, for lists and notifications.
290    pub title: String,
291    /// The task itself, handed to the graph verbatim.
292    pub instruction: String,
293    /// Repository to work in.
294    pub repo: PathBuf,
295    /// Who asked.
296    pub source: Source,
297    /// Higher runs first; ties break oldest-first so nothing starves.
298    #[serde(default)]
299    pub priority: i32,
300    /// Run this task alone: one implementer, no panel of judges to convince.
301    ///
302    /// `#[serde(default)]` so a queue file written before this field existed
303    /// still reads, as `false` - the ordinary multi-candidate competition,
304    /// unchanged. A task set to `solo` still runs the whole graph; only the
305    /// candidate count the daemon builds it with changes, and
306    /// [`crate::graph::Runner`] already collapses a single-candidate run to
307    /// implement → review → gate → merge on its own (see
308    /// [`crate::graph::Runner::review`]'s doc), so nothing about judging,
309    /// deliberation or voting had to change to support this.
310    #[serde(default)]
311    pub solo: bool,
312    /// Current state.
313    pub status: TaskStatus,
314    /// How many times this task has been claimed.
315    #[serde(default)]
316    pub attempts: usize,
317    /// Total attempt budget; `None` inherits the daemon limit.
318    #[serde(default)]
319    pub attempt_limit: Option<NonZeroUsize>,
320    /// Distinct runs, ordered by their most recent claim. Older records may
321    /// contain repeated ids; execution history lives in the run pass records.
322    #[serde(default)]
323    pub runs: Vec<String>,
324    /// Why the last attempt did not land.
325    #[serde(default)]
326    pub last_error: Option<String>,
327    /// What a human hold is waiting on.
328    ///
329    /// `None` covers both the ordinary cases: a hold the loop makes itself
330    /// (out of attempts, or the disk gate closed) explains itself through
331    /// [`Task::last_error`] instead, and a human hold nobody bothered to
332    /// explain is still a valid hold. The queue has no way to express a
333    /// dependency between two tasks, so on the occasions a hold really is
334    /// "wait for that other task first", this is the only place that reason
335    /// survives - see [`Task::hold_manual`] and [`Task::release`].
336    ///
337    /// `#[serde(default)]` so a queue file written before this field existed
338    /// still reads, with no reason recorded rather than a parse error.
339    #[serde(default)]
340    pub hold_reason: Option<String>,
341    /// Who placed [`Task::hold_reason`].  `None` is a compatible old record;
342    /// see [`Task::operator_held`] for its deliberately conservative meaning.
343    #[serde(default)]
344    pub hold_source: Option<HoldSource>,
345    /// Diagnostic detail excerpted from the run that led to a hold - what a
346    /// human would have found opening `artifacts/` by hand, not the one-line
347    /// reason in [`Task::last_error`]. Set only when a run's own attempts are
348    /// exhausted and the task becomes [`TaskStatus::Held`]; `daemon` computes
349    /// it from the run's own record, since this module has no notion of a
350    /// run's internals. Bounded in length by the writer - see
351    /// `daemon::diagnostic` - so a verbose run cannot make this file grow
352    /// without limit.
353    ///
354    /// `#[serde(default)]` so a queue file written before this field existed
355    /// still reads, with no diagnostic recorded rather than a parse error.
356    #[serde(default)]
357    pub diagnostic: Option<String>,
358    /// What this task is waiting on: other task ids, unanswered
359    /// `crate::ask::Question` ids, or both. Non-empty exactly when
360    /// [`TaskStatus::Blocked`]; emptying it — see [`Task::unblock`] — is what
361    /// puts the task back at [`TaskStatus::Queued`].
362    ///
363    /// Set by `crate::conduct`'s decisions and cleared deterministically by
364    /// `crate::daemon` as each dependency resolves, never by a person. Never
365    /// `#[serde(default)]` is skipped: a queue file from before this field
366    /// existed has nothing to report here, and an empty list is exactly that.
367    #[serde(default)]
368    pub blocked_by: Vec<String>,
369    /// One line explaining the current [`Task::blocked_by`], written by
370    /// `crate::conduct`. Cleared whenever `blocked_by` empties.
371    #[serde(default)]
372    pub block_reason: Option<String>,
373    /// The status this task had the moment [`Task::block`] most recently
374    /// moved it to [`TaskStatus::Blocked`] — what [`Task::unblock`] restores
375    /// once nothing is left in `blocked_by`, instead of always landing on
376    /// [`TaskStatus::Queued`]. See [`SCHEMA`]'s doc for schema 4 on why this
377    /// exists: an answer to a question `crate::conduct` filed about a
378    /// [`TaskStatus::Held`] task must not itself be what puts the task back
379    /// in the competition queue.
380    ///
381    /// `#[serde(default)]` so a queue file written before this field existed
382    /// reads as `None`; [`Task::unblock`] treats that the same as a task
383    /// blocked straight from `Queued`, unless surviving hold evidence says
384    /// otherwise.
385    #[serde(default)]
386    pub blocked_from: Option<TaskStatus>,
387    /// Last change of blocking targets, independent of explanation or delivery writes.
388    #[serde(default)]
389    pub blocked_at: Option<Timestamp>,
390    /// Questions `crate::conduct` asked about this task that the operator has
391    /// since answered, oldest first — what was asked, and what they said.
392    ///
393    /// A blocking question's id leaves [`Task::blocked_by`] the moment
394    /// [`crate::ask::QuestionStatus::Answered`] is observed, but the id alone
395    /// tells nobody what was decided. This is what carries the answer's
396    /// *content* forward: into the next conductor prompt for this task, and
397    /// into the instruction handed to the next run — see `crate::daemon`'s
398    /// deterministic blocker resolution. Kept for the task's whole life, the
399    /// same as [`Task::runs`]: a release resets attempts, not evidence.
400    #[serde(default)]
401    pub answers: Vec<AnsweredQuestion>,
402    /// Ids of the `crate::triage` questions whose answer has been applied to
403    /// this task. Unlike [`Task::hold_reason`], [`Task::release`] and every
404    /// hold transition leave it alone, so an answer is applied at most once
405    /// however many times the task is held again. See [`SCHEMA`]'s doc for
406    /// schema 5. `#[serde(default)]` so an older record reads as empty.
407    #[serde(default)]
408    pub triage_applied: Vec<String>,
409    /// Ids of the `magi ask` questions whose [`crate::ask::ChoiceAction`] has
410    /// been applied to this task. Separate from [`Task::triage_applied`]
411    /// (a different producer) and, like it, untouched by [`Task::release`],
412    /// so one answer acts at most once however often the task is held again.
413    /// `#[serde(default)]` so an older record reads as empty.
414    #[serde(default)]
415    pub actions_applied: Vec<String>,
416    /// The operator's "resume" answer to a triage question, kept until the
417    /// task actually runs (or is done) so `crate::conduct` cannot silently
418    /// undo it and `crate::triage` can tell that a hold it sees now came
419    /// *after* the answer. See [`OperatorResume`]. `#[serde(default)]`.
420    #[serde(default)]
421    pub resume_override: Option<OperatorResume>,
422    /// Latest explicit operator release. Historical answers remain audit evidence;
423    /// this approval takes precedence over their earlier waiting premises.
424    #[serde(default)]
425    pub resumption: Option<Resumption>,
426    /// Earlier explicit approvals retained for audit and instruction suffix matching.
427    #[serde(default)]
428    pub resumption_history: Vec<Resumption>,
429    /// Latest explicit release or manual hold, used to reject delayed answers.
430    #[serde(default)]
431    pub operator_decided_at: Option<Timestamp>,
432    /// Set by `crate::conduct` when it chooses `Review` recovery for a task
433    /// whose branch survived a blocked run: the branch to reopen with
434    /// `crate::graph::Runner::review` instead of competing from scratch.
435    ///
436    /// Requeues the task the same way [`Task::release`] does, so it is
437    /// picked up by the ordinary loop; `crate::daemon` reads this field once,
438    /// when it actually starts the run, and clears it either way — consumed
439    /// on success, dropped if the branch no longer exists by then. Never set
440    /// from the conductor's own words: `crate::daemon` derives the branch
441    /// name itself from the task's last run, so a hallucinated branch can
442    /// never reach here.
443    #[serde(default)]
444    pub review_branch: Option<String>,
445    /// Authored criteria when the retained recovery review was selected.
446    #[serde(default)]
447    pub review_instruction: Option<String>,
448    /// Explicit public implementation selected by the operator; never inferred from prose.
449    #[serde(default)]
450    pub pr_recovery: Option<crate::pr_recovery::Selection>,
451    /// A release deliberately starts a new competition instead of resuming
452    /// the prior run. History remains as evidence in `runs`.
453    #[serde(default)]
454    pub fresh_start: bool,
455    /// Explicit replacement selection; persists through dispatch and resume.
456    #[serde(default)]
457    pub replacement: Option<crate::merge_decision::Replacement>,
458    /// Marked by an operator (`magi task interrupt`) to ask `magi serve` to
459    /// run this one ahead of whatever it already has in flight, once
460    /// `[daemon] pause_for_interrupts` is on - see
461    /// `crate::daemon::advance_interrupt`. Never set by the loop itself, and
462    /// deliberately a different operation from [`Task::set_priority`]: a
463    /// priority only reorders the queue a claim has not reached yet, while
464    /// this asks a run already in flight to park at its next safe boundary
465    /// and step aside. `#[serde(default)]` so a queue file written before
466    /// this field existed still reads, as `false` - no task interrupts
467    /// anything unless asked to, exactly as before.
468    #[serde(default)]
469    pub interrupt: bool,
470    /// Marked by `magi task add --urgent` or `magi task urgent`.
471    /// `crate::daemon::poll` dispatches this task through its own one-slot `urgent_sem` the moment it is
472    /// runnable, in addition to whatever is already running under the
473    /// ordinary `[daemon] max_concurrent_runs` pool - never instead of it,
474    /// and never by pausing or otherwise touching that run. This is the
475    /// opposite direction from [`Task::interrupt`]: that one asks a run
476    /// already in flight to step aside; this one never asks anything to
477    /// step aside, it only spends one additional, temporary concurrency
478    /// slot. The two are independent and may both be set on the same task,
479    /// but this exemption stops at `[daemon] pause_for_interrupts`'s own
480    /// park/resume handoff (75dd): while an interrupt sequence is actively
481    /// parking, running, or resuming - its own, or an unrelated task's -
482    /// `crate::daemon::interrupt_gate` withholds an urgent candidate exactly
483    /// like an ordinary one, never exempted. 75dd's "at most one run, ever,
484    /// at once" guarantee takes precedence, because the alternative is a run
485    /// still genuinely in flight (only *asked* to park, not yet gone) ending
486    /// up alongside a second one this feature let through - the very thing
487    /// that guarantee exists to rule out.
488    ///
489    /// `#[serde(default)]` so a queue file written before this field existed
490    /// still reads, as `false` - no task claims the urgent slot unless asked
491    /// to, exactly as before.
492    #[serde(default)]
493    pub urgent: bool,
494    /// Files (typically screenshots) copied into `<id>.attachments/` beside
495    /// this task's file by [`Queue::attach`], as bare validated names. The
496    /// copy is what makes them reach the implementer: the original may be
497    /// cleaned up long before the task runs. Untouched by [`Task::edit`].
498    /// `#[serde(default)]` so an older record reads as empty.
499    #[serde(default)]
500    pub attachments: Vec<String>,
501    /// Set on a task `crate::followup` filed from the findings a merged run
502    /// left open: which run and findings it carries, and how many follow-ups
503    /// deep it is. `None` for every ordinary task. `#[serde(default)]`.
504    #[serde(default)]
505    pub followup: Option<FollowUp>,
506    /// Id of the chat this task descends from: set by [`Task::new`] when a
507    /// chat files it, inherited by the follow-ups `crate::followup` files, so
508    /// the provenance survives deleting an ancestor. `None` for a task written
509    /// before the field existed (`crate::consult` then walks the ancestry).
510    /// `#[serde(default)]`.
511    #[serde(default)]
512    pub origin_chat: Option<String>,
513    /// The command-line choices (`--merge`, `-c`, `--seed`, ...) of the
514    /// `magi run` / `magi review` that filed this task, applied on top of the
515    /// repository's config by whoever executes it - a daemon included - so
516    /// handing the task over does not change what it does. A count edit on an
517    /// ordinary queue task explicitly inherits daemon options instead.
518    /// `#[serde(default)]`.
519    #[serde(default)]
520    pub overrides: Option<RunOverrides>,
521    /// The branch an explicit `magi review <branch>` asked to review. Unlike
522    /// [`Task::review_branch`] (a recovery choice the daemon consumes and
523    /// drops back to a competition when the branch is gone), this is the whole
524    /// request: it is never consumed, and a missing branch fails the attempt
525    /// rather than paying for an implementation nobody asked for.
526    /// `#[serde(default)]`.
527    #[serde(default)]
528    pub review_of: Option<String>,
529    /// When the current hold began. Set by the transitions into
530    /// [`TaskStatus::Held`] (a re-hold keeps it), cleared by
531    /// [`Task::release`]. `crate::notices` uses it as the identity of the
532    /// cause: a question filed before it is about something older.
533    #[serde(default)]
534    pub held_at: Option<Timestamp>,
535    /// Why the run parked, while the task is [`TaskStatus::Parked`]. Kept apart
536    /// from [`Task::last_error`] because nothing failed. Cleared by every
537    /// transition out of the parked state. `#[serde(default)]`.
538    #[serde(default)]
539    pub park_reason: Option<String>,
540    /// The last final state reported to the originating chat. See
541    /// [`Task::chat_notice_due`]. Never cleared by a release or requeue.
542    #[serde(default)]
543    pub chat_notice: Option<ChatNotice>,
544    /// Delivery ledger, unioned under the task lock even on stale writes.
545    #[serde(default)]
546    pub chat_notices: Vec<String>,
547    /// Structured producer identity of the unresolved hold. Requeues and prose
548    /// updates retain it; a settled run or a distinct startup stage replaces it.
549    #[serde(default)]
550    pub hold_notice_cause: Option<NoticeCause>,
551    /// Append-only: one entry per pass over a run that the loop settled. A
552    /// resumed run is one run with several passes, and `attempts` counts
553    /// claims net of refunds, so the two are not the same number; this says
554    /// which pass cost what. Kept across a release.
555    #[serde(default)]
556    pub passes: Vec<TaskPass>,
557    /// Durable failed-input budget, unaffected by automatic release/requeue.
558    #[serde(default)]
559    pub recovery_budget: Option<RecoveryBudget>,
560    /// Inputs resolved before starting the current automatic attempt. The
561    /// settlement uses this same identity even for review-only runs.
562    #[serde(default)]
563    pub recovery_admission: Option<RecoveryAdmission>,
564    /// Previously seen input generations. Switching inputs away and back does
565    /// not refund an exhausted generation without explicit operator authority.
566    #[serde(default)]
567    pub recovery_generations: Vec<RecoveryBudget>,
568    /// When the task was filed.
569    pub created_at: Timestamp,
570    /// Last change to this file.
571    pub updated_at: Timestamp,
572}
573
574/// Admission identity shared by start and settlement.
575#[derive(Debug, Clone, Serialize, Deserialize)]
576pub struct RecoveryAdmission {
577    /// Fetched upstream commit before any run-local rebase.
578    pub base: String,
579    /// Resolved instruction references and explicit review source.
580    pub inputs: String,
581}
582
583/// Failures against one recovery input generation. An owner release explicitly
584/// starts a new generation; a machine requeue never replenishes it.
585#[derive(Debug, Clone, Serialize, Deserialize)]
586pub struct RecoveryBudget {
587    /// Authored instructions, not generated review scaffolding.
588    pub instruction: String,
589    /// Upstream commit used to prepare this generation.
590    pub base: String,
591    /// Referenced input commits, including their resolved identities.
592    pub inputs: String,
593    /// Stable processing stage of the last failure.
594    pub stage: String,
595    /// Failed passes, independent of the resettable claim counter.
596    pub failures: usize,
597    /// Fingerprint of actual candidate patches; equivalent rebased commits do
598    /// not count as progress. Unknown inputs never replenish the budget.
599    #[serde(default)]
600    pub progress: Option<String>,
601}
602
603fn recovery_progress(state: &crate::run::RunState) -> Option<String> {
604    use crate::proc::Quiet as _;
605    use std::hash::{Hash, Hasher};
606    let mut patches = Vec::new();
607    for candidate in state.candidates.iter().filter(|c| c.commits > 0) {
608        let output = std::process::Command::new("git")
609            .current_dir(&state.repo)
610            .args(["diff", "--binary", &state.base_commit, &candidate.branch])
611            .quiet()
612            .output()
613            .ok()?;
614        if !output.status.success() {
615            return None;
616        }
617        patches.push(output.stdout);
618    }
619    patches.sort();
620    patches.dedup();
621    let mut hash = std::collections::hash_map::DefaultHasher::new();
622    patches.hash(&mut hash);
623    Some(format!("{:x}", hash.finish()))
624}
625
626pub(crate) fn recovery_inputs(seeds: &[crate::refs::Seed]) -> String {
627    serde_json::to_string(
628        &seeds
629            .iter()
630            .map(|s| (&s.token, &s.sha, s.kind))
631            .collect::<Vec<_>>(),
632    )
633    .unwrap_or_default()
634}
635
636/// Facts a producer can independently attest about a held outcome.
637#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
638pub struct NoticeCause {
639    /// Processing stage or verified external evidence.
640    pub stage: String,
641    /// Producer-verified blocker object, independent of the displayed error.
642    /// Omitted when absent to preserve existing persisted notice keys.
643    #[serde(default, skip_serializing_if = "Option::is_none")]
644    pub target: Option<String>,
645    /// An actual run that reached settlement; absent for preflight failures.
646    pub run: Option<String>,
647    /// The settled pass, so a resumed run can report a new failure.
648    pub pass: Option<usize>,
649}
650
651/// Current and legacy completion keys; legacy run counts contain only digits.
652pub(crate) fn is_done_notice_key(key: &str) -> bool {
653    key == "done"
654        || key
655            .strip_prefix("done#")
656            .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
657}
658
659/// What a settled pass did to the attempt budget.
660#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
661#[serde(rename_all = "snake_case")]
662pub enum PassCost {
663    /// The attempt stays counted.
664    Spent,
665    /// The attempt was handed back.
666    Refunded,
667}
668
669/// One settled pass over a run. See [`Task::passes`].
670#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
671pub struct TaskPass {
672    /// The run the pass belongs to.
673    pub run: String,
674    /// [`crate::run::RunPass::index`] in that run.
675    pub pass_index: usize,
676    /// What it did to the budget.
677    pub cost: PassCost,
678    /// `attempts` once this pass was settled.
679    pub attempts_after: usize,
680    /// When the loop settled it.
681    pub settled_at: Timestamp,
682    /// Why the pass parked, when it did.
683    #[serde(default)]
684    pub parked: Option<crate::run::ParkCause>,
685}
686
687/// What became of a completion notice. See [`Task::chat_notice`].
688#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
689#[serde(rename_all = "lowercase")]
690pub enum ChatNoticeOutcome {
691    /// Queued as a draft of the originating talk.
692    Posted,
693    /// The talk is gone or closed; nothing was queued, and nothing will be.
694    Skipped,
695}
696
697/// The final state of a chat-filed task that was already reported (or skipped).
698#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
699pub struct ChatNotice {
700    /// Identity of the cause, see [`Task::chat_notice_key`].
701    pub key: String,
702    /// When it was recorded.
703    pub at: Timestamp,
704    /// Posted or skipped.
705    pub outcome: ChatNoticeOutcome,
706}
707
708/// Command-line choices carried by a task. See [`Task::overrides`].
709#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
710pub struct RunOverrides {
711    /// Count edits on ordinary queued tasks inherit the loop's config and merge.
712    /// False for direct-run overrides and old records, even when otherwise empty.
713    #[serde(default)]
714    pub inherit_daemon_options: bool,
715    /// `--merge`: `none`, `local` or `pr`.
716    #[serde(default)]
717    pub merge: Option<String>,
718    /// `-c` / `--implementers`: parallel implementations (persisted as `candidates`).
719    #[serde(default)]
720    pub candidates: Option<usize>,
721    /// `-j`: independent judges.
722    #[serde(default)]
723    pub judges: Option<usize>,
724    /// `--reviewers`: reviewers per round.
725    #[serde(default)]
726    pub reviewers: Option<usize>,
727    /// `--review-rounds`.
728    #[serde(default)]
729    pub review_rounds: Option<usize>,
730    /// `--seed`.
731    #[serde(default)]
732    pub seed: Option<u64>,
733    /// `--config`, absolute: the daemon's own working directory differs.
734    #[serde(default)]
735    pub config: Option<PathBuf>,
736}
737
738impl RunOverrides {
739    /// Lay `other` over these: every choice `other` made wins, the rest stay.
740    pub fn merge_over(&mut self, other: &RunOverrides) {
741        macro_rules! take {
742            ($($f:ident),*) => {$(
743                if other.$f.is_some() {
744                    self.$f.clone_from(&other.$f);
745                }
746            )*};
747        }
748        take!(
749            merge,
750            candidates,
751            judges,
752            reviewers,
753            review_rounds,
754            seed,
755            config
756        );
757    }
758
759    /// Apply these on top of `config`.
760    pub fn apply(&self, config: &mut crate::config::Config) {
761        if let Some(n) = self.candidates {
762            config.graph.implementers = n;
763        }
764        if let Some(n) = self.judges {
765            config.graph.judges = n;
766        }
767        if let Some(n) = self.reviewers {
768            config.graph.reviewers = n;
769        }
770        if let Some(n) = self.review_rounds {
771            config.graph.review_rounds = n;
772        }
773        if let Some(m) = &self.merge
774            && let Ok(mode) = crate::daemon::merge_mode(m)
775        {
776            config.merge.mode = mode;
777        }
778        if let Some(s) = self.seed {
779            config.blind.seed = Some(s);
780        }
781    }
782}
783
784/// What a follow-up task came from. See [`Task::followup`].
785#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
786pub struct FollowUp {
787    /// The merged run whose open findings this task carries.
788    pub run: String,
789    /// The queue task that run served, when it had one.
790    #[serde(default)]
791    pub origin_task: Option<String>,
792    /// The merged pull request.
793    pub pr: String,
794    /// Ids of the findings (e.g. `R3-1-1`) this task covers.
795    pub findings: Vec<String>,
796    /// Follow-up depth: 1 for a follow-up of an ordinary task's run, 2 for a
797    /// follow-up of that, and so on. An ordinary task is generation 0.
798    pub generation: u32,
799}
800
801/// Durable scope of an explicit operator resumption, surviving run startup.
802#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
803pub struct Resumption {
804    /// Monotonic count of explicit resumptions.
805    pub generation: u64,
806    /// Mutation path that applied the approval.
807    pub source: String,
808    /// Answered question, if the approval came from one.
809    pub question_id: Option<String>,
810    /// When this approval was applied.
811    pub at: Timestamp,
812    /// Authored instruction approved at release.
813    pub instruction: String,
814    /// Number of historical answers superseded as waiting premises.
815    pub answers_through: usize,
816    /// Legacy run-list boundary retained for record compatibility. Evidence
817    /// uses approval time and run passes because resumes retain the same id.
818    pub runs_through: usize,
819    /// Previous hold or dependency reason cleared by this release.
820    pub cleared_reason: Option<String>,
821    /// Verified evidence versions already used to block after this approval.
822    #[serde(default)]
823    pub blocking_evidence: Vec<String>,
824}
825
826/// A triage "resume" answer and what became of it. See
827/// [`Task::resume_override`].
828#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
829pub struct OperatorResume {
830    /// The triage question the operator answered.
831    pub question_id: String,
832    /// When the answer was applied.
833    pub at: Timestamp,
834    /// The reason `crate::conduct` gave for holding the task again after the
835    /// answer, if it did. The conductor may do this once.
836    #[serde(default)]
837    pub conductor_rehold: Option<String>,
838    /// The operator answered "resume" a second time, to the question about
839    /// that contradiction: the conductor may no longer hold this task.
840    #[serde(default)]
841    pub forced: bool,
842    /// Set when the answer was a structured `resume` action on a `magi ask`
843    /// choice ([`crate::ask::ChoiceAction::Resume`]): the run the operator
844    /// chose to continue. Its presence also stops `crate::conduct` from
845    /// requeuing the task into a fresh competition, which would discard
846    /// exactly the run that was named. Cleared with the rest of the record
847    /// once the task runs. `#[serde(default)]` so older records read as `None`.
848    #[serde(default)]
849    pub pinned_run: Option<String>,
850}
851
852/// One question `crate::conduct` asked about a task, and what the operator
853/// said back. See [`Task::answers`].
854#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
855pub struct AnsweredQuestion {
856    /// The question as asked, e.g. [`crate::ask::Question::summary`].
857    pub question: String,
858    /// What the operator answered.
859    pub answer: String,
860}
861
862impl Task {
863    /// File a new task. Persist it with [`Queue::put`].
864    pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
865        let now = Timestamp::now();
866        let origin_chat = match &source {
867            Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
868            _ => None,
869        };
870        Self {
871            schema: SCHEMA,
872            id: new_id(),
873            title,
874            instruction,
875            repo,
876            source,
877            priority: 0,
878            solo: false,
879            status: TaskStatus::Queued,
880            attempts: 0,
881            attempt_limit: None,
882            runs: Vec::new(),
883            last_error: None,
884            hold_reason: None,
885            hold_source: None,
886            diagnostic: None,
887            blocked_by: Vec::new(),
888            block_reason: None,
889            blocked_from: None,
890            blocked_at: None,
891            answers: Vec::new(),
892            triage_applied: Vec::new(),
893            actions_applied: Vec::new(),
894            resume_override: None,
895            resumption: None,
896            resumption_history: Vec::new(),
897            operator_decided_at: None,
898            review_branch: None,
899            review_instruction: None,
900            pr_recovery: None,
901            fresh_start: false,
902            replacement: None,
903            interrupt: false,
904            urgent: false,
905            attachments: Vec::new(),
906            followup: None,
907            origin_chat,
908            overrides: None,
909            review_of: None,
910            held_at: None,
911            park_reason: None,
912            chat_notice: None,
913            chat_notices: Vec::new(),
914            hold_notice_cause: None,
915            passes: Vec::new(),
916            recovery_budget: None,
917            recovery_admission: None,
918            recovery_generations: Vec::new(),
919            created_at: now,
920            updated_at: now,
921        }
922    }
923
924    /// The chat this task itself says it came from: the recorded
925    /// [`Task::origin_chat`], else its own source when a chat filed it.
926    pub fn chat_talk(&self) -> Option<&str> {
927        if let Some(id) = &self.origin_chat {
928            return Some(id);
929        }
930        match &self.source {
931            Source::Agent { run, node } if node == CHAT_NODE => Some(run),
932            _ => None,
933        }
934    }
935
936    /// The talk that filed this task *itself* (`Source::Agent` with
937    /// [`CHAT_NODE`]); unlike [`Task::chat_talk`] a follow-up's inherited
938    /// `origin_chat` does not count.
939    pub fn filed_by_chat(&self) -> Option<&str> {
940        match &self.source {
941            Source::Agent { run, node } if node == CHAT_NODE => Some(run),
942            _ => None,
943        }
944    }
945
946    /// Identity of a material outcome. Prose and wall-clock times are never
947    /// evidence. Dependencies are a set; held causes survive intermediate states.
948    pub fn chat_notice_key(&self) -> Option<String> {
949        match self.status {
950            TaskStatus::Done => Some("done".to_owned()),
951            TaskStatus::Held => {
952                let cause = self
953                    .hold_notice_cause
954                    .clone()
955                    .unwrap_or_else(|| self.unresolved_hold_cause());
956                Some(format!(
957                    "cause:{}",
958                    serde_json::to_string(&cause).expect("cause")
959                ))
960            }
961            TaskStatus::Blocked => {
962                let ids: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
963                // Only new verified blocking evidence versions the dependency
964                // set. Freezing an unrelated hold on release must not do so.
965                let evidence = self
966                    .hold_notice_cause
967                    .as_ref()
968                    .filter(|c| c.stage.starts_with("conductor-evidence:"));
969                Some(format!(
970                    "dependencies:{}",
971                    serde_json::to_string(&(ids, evidence)).expect("ids")
972                ))
973            }
974            _ => None,
975        }
976    }
977
978    fn unresolved_hold_cause(&self) -> NoticeCause {
979        NoticeCause {
980            stage: "hold".into(),
981            target: None,
982            run: self.runs.last().cloned(),
983            pass: None,
984        }
985    }
986
987    /// Exact old delivery identity, only for reconciling an existing record.
988    pub(crate) fn legacy_chat_notice_key(&self) -> Option<String> {
989        let key = match self.status {
990            TaskStatus::Held if self.hold_notice_cause.is_none() => Some(format!(
991                "held#{}",
992                self.held_at.map(|t| t.to_string()).unwrap_or_default()
993            )),
994            TaskStatus::Blocked => Some(format!(
995                "blocked#{}#{}",
996                self.blocked_by.join(","),
997                self.block_reason.as_deref().unwrap_or_default()
998            )),
999            _ => None,
1000        };
1001        key.filter(|key| self.legacy_notice_matches(key))
1002    }
1003
1004    pub(crate) fn legacy_notice_matches(&self, key: &str) -> bool {
1005        match self.status {
1006            TaskStatus::Held if self.hold_notice_cause.is_none() => key
1007                .strip_prefix("held#")
1008                .is_some_and(|time| time.is_empty() || time.parse::<Timestamp>().is_ok()),
1009            TaskStatus::Blocked
1010                if self
1011                    .hold_notice_cause
1012                    .as_ref()
1013                    .is_none_or(|c| !c.stage.starts_with("conductor-evidence:")) =>
1014            {
1015                key.strip_prefix("blocked#").is_some_and(|tail| {
1016                    let Some((ids, _)) = tail.split_once('#') else {
1017                        return false;
1018                    };
1019                    let old: std::collections::BTreeSet<_> =
1020                        ids.split(',').filter(|id| !id.is_empty()).collect();
1021                    let current: std::collections::BTreeSet<_> =
1022                        self.blocked_by.iter().map(String::as_str).collect();
1023                    old == current
1024                })
1025            }
1026            _ => false,
1027        }
1028    }
1029
1030    fn import_chat_notice(&mut self) {
1031        if let Some(n) = &self.chat_notice {
1032            let key = if is_done_notice_key(&n.key) {
1033                "done".to_owned()
1034            } else if self.legacy_notice_matches(&n.key) {
1035                self.chat_notice_key().unwrap_or_else(|| n.key.clone())
1036            } else {
1037                n.key.clone()
1038            };
1039            if !self.chat_notices.contains(&key) {
1040                self.chat_notices.push(key);
1041            }
1042        }
1043    }
1044
1045    /// A producer records the processing stage and actual run/pass, never the
1046    /// error's rendered prose. No run means startup made no implementation progress.
1047    pub(crate) fn notice_cause(&mut self, stage: &str, run: Option<&str>, pass: Option<usize>) {
1048        self.import_chat_notice();
1049        self.hold_notice_cause = Some(NoticeCause {
1050            stage: stage.to_owned(),
1051            target: None,
1052            run: run.map(str::to_owned),
1053            pass,
1054        });
1055    }
1056
1057    /// Startup retries cannot replace an unresolved run outcome without doing
1058    /// work. Distinct startup stages can still explain a newly encountered blocker.
1059    pub(crate) fn startup_notice_cause(&mut self, stage: &str) {
1060        let recovery_only = matches!(
1061            stage,
1062            "startup-source" | "startup-candidates" | "startup-public-head"
1063        );
1064        match self.hold_notice_cause.as_ref() {
1065            Some(cause) if recovery_only && !cause.stage.starts_with("startup-") => return,
1066            None if recovery_only
1067                && self
1068                    .chat_notice
1069                    .as_ref()
1070                    .is_some_and(|n| !is_done_notice_key(&n.key)) =>
1071            {
1072                return;
1073            }
1074            _ => {}
1075        }
1076        self.notice_cause(stage, None, None);
1077    }
1078
1079    /// Apply a preflight failure without allowing Running's generic failure
1080    /// identity to overwrite the producer's stage or its unresolved recovery anchor.
1081    pub(crate) fn fail_startup(
1082        &mut self,
1083        stage: &str,
1084        why: impl Into<String>,
1085        max_attempts: usize,
1086    ) {
1087        let unresolved = self.hold_notice_cause.clone();
1088        self.fail(why, max_attempts);
1089        self.hold_notice_cause = unresolved;
1090        self.startup_notice_cause(stage);
1091    }
1092
1093    /// A startup failure tied to a concrete object, such as a missing attachment.
1094    pub(crate) fn fail_startup_target(
1095        &mut self,
1096        stage: &str,
1097        target: String,
1098        why: impl Into<String>,
1099        max_attempts: usize,
1100    ) {
1101        self.fail_startup(stage, why, max_attempts);
1102        if let Some(cause) = self.hold_notice_cause.as_mut() {
1103            cause.target = Some(target);
1104        }
1105    }
1106
1107    /// A chat-filed task owes each material final outcome at most one notice.
1108    pub fn chat_notice_due(&self) -> Option<String> {
1109        self.filed_by_chat()?;
1110        if self.schema < 13 {
1111            return None;
1112        }
1113        let key = self.chat_notice_key()?;
1114        if self.chat_notices.contains(&key) {
1115            return None;
1116        }
1117        match &self.chat_notice {
1118            Some(n)
1119                if n.key == key
1120                    || (key == "done" && is_done_notice_key(&n.key))
1121                    || self.legacy_notice_matches(&n.key) =>
1122            {
1123                None
1124            }
1125            _ => Some(key),
1126        }
1127    }
1128
1129    /// Short form used in reports, matching a run's short id.
1130    pub fn short(&self) -> &str {
1131        short(&self.id)
1132    }
1133
1134    /// Record that triage question `question_id`'s answer has been applied.
1135    pub fn mark_triage_applied(&mut self, question_id: &str) {
1136        if !self.triage_applied(question_id) {
1137            self.triage_applied.push(question_id.to_owned());
1138        }
1139    }
1140
1141    /// Has the action of `magi ask` question `question_id` already been
1142    /// applied, or is it being applied now? See [`Task::actions_applied`].
1143    pub fn action_applied(&self, question_id: &str) -> bool {
1144        self.actions_applied.iter().any(|id| id == question_id)
1145    }
1146
1147    /// Record that `magi ask` question `question_id`'s action was applied.
1148    pub fn mark_action_applied(&mut self, question_id: &str) {
1149        if !self.action_applied(question_id) {
1150            self.actions_applied.push(question_id.to_owned());
1151        }
1152    }
1153
1154    /// Has triage question `question_id`'s answer already been applied?
1155    pub fn triage_applied(&self, question_id: &str) -> bool {
1156        self.triage_applied.iter().any(|id| id == question_id)
1157    }
1158
1159    /// Record that a run has started for this task.
1160    /// Move an already linked run to the end so it remains the current attempt,
1161    /// with one history entry per run and separate pass records for each resume.
1162    ///
1163    /// Clears [`Task::interrupt`]: a mark to run ahead of whatever else is
1164    /// in flight is fulfilled the moment this task actually gets its turn,
1165    /// dispatched same as any other. Without this, a task whose run fails
1166    /// and requeues - still `runnable`, still carrying the mark from its
1167    /// first attempt - would keep re-triggering `crate::daemon`'s interrupt
1168    /// scheduler and re-parking whatever it interrupted on every later
1169    /// boundary, for as long as its attempts hold out, instead of the
1170    /// one-shot "let this go next" the mark is meant to be.
1171    pub fn start(&mut self, run: String) {
1172        self.stamp_schema();
1173        self.status = TaskStatus::Running;
1174        self.attempts += 1;
1175        self.runs.retain(|id| id != &run);
1176        self.runs.push(run);
1177        self.last_error = None;
1178        self.park_reason = None;
1179        self.fresh_start = false;
1180        self.interrupt = false;
1181        // The answer has been honoured: the task got its turn.
1182        self.resume_override = None;
1183    }
1184
1185    /// Note that `run` serves this task without otherwise changing its life:
1186    /// no status, attempt or interrupt change, unlike [`Task::start`]. For a
1187    /// run somebody else started on the task's behalf (`magi run --task`);
1188    /// idempotent. Returns whether anything was added.
1189    pub fn link_run(&mut self, run: &str) -> bool {
1190        if self.runs.iter().any(|r| r == run) {
1191            return false;
1192        }
1193        self.runs.push(run.to_owned());
1194        true
1195    }
1196
1197    /// Record a successful run.
1198    ///
1199    /// Both `magi task done` and `POST /api/queue/{id}/done` can close a held
1200    /// *or blocked* task directly, with no release in between, so this clears
1201    /// `hold_reason` and `blocked_by`/`block_reason` the same way
1202    /// [`Task::release`] does. Otherwise a task held for "waiting on 3ed9", or
1203    /// blocked on a dependency that never actually finished, and then closed
1204    /// as done without ever being released would still read as waiting on
1205    /// something in `magi task show` and on its card, after it no longer is.
1206    pub fn succeed(&mut self) {
1207        self.stamp_schema();
1208        self.status = TaskStatus::Done;
1209        self.resume_override = None;
1210        self.last_error = None;
1211        self.park_reason = None;
1212        self.hold_reason = None;
1213        self.hold_source = None;
1214        self.diagnostic = None;
1215        self.blocked_by.clear();
1216        self.block_reason = None;
1217        self.blocked_from = None;
1218    }
1219
1220    /// Finish the task because the run's whole change turned out to be on the
1221    /// base already ([`crate::run::RunStatus::AlreadyInBase`]): `Done`, with
1222    /// `note` kept in [`Task::last_error`] so `magi task show` says why, and
1223    /// the attempt refunded. Nothing went wrong and nothing was spent
1224    /// misbehaving, so this is neither [`Task::fail`] nor
1225    /// [`Task::handed_off`]; and unlike [`Task::succeed`] it does not pretend
1226    /// the run landed anything.
1227    pub fn already_landed(&mut self, note: impl Into<String>) {
1228        self.succeed();
1229        self.attempts = self.attempts.saturating_sub(1);
1230        self.last_error = Some(note.into());
1231    }
1232
1233    /// Earlier attempts at this task that a later one has since made moot —
1234    /// empty unless the task is [`TaskStatus::Done`] *and* `last_run_succeeded`
1235    /// says `runs.last()` is actually why.
1236    ///
1237    /// `runs` is oldest first, and [`Task::start`] is the only thing that
1238    /// pushes to it, always right before the attempt it names either succeeds
1239    /// or fails; `succeed` itself never touches `runs`. So whenever `status`
1240    /// is `Done` *because* the loop itself saw that last attempt land
1241    /// (`Merged`/`Ready`), everything before it in the same list is a retry
1242    /// this task no longer needs. But `succeed` is also reachable directly —
1243    /// `magi task done`, the web UI's equivalent, and the conductor's
1244    /// `Recovery::Done` all call it on a task in *any* status, including one
1245    /// whose last recorded attempt never landed at all (closed by hand after
1246    /// a merge magi's own loop never saw). `runs.last()` alone cannot tell
1247    /// those two cases apart — that requires the caller to have actually
1248    /// looked at that run's own `status`, which this module has no way to
1249    /// do — so `last_run_succeeded` is the caller's answer to exactly that
1250    /// question, not something this function can derive from `Task` alone.
1251    ///
1252    /// Deliberately narrower than [`Queue::superseded`]'s "every earlier
1253    /// attempt has a later one" walk, which fires the moment a retry starts
1254    /// even though the retry itself might still fail: that reading is right
1255    /// for the web UI's "a newer attempt exists, go look at that one
1256    /// instead" note, but wrong for deciding a run no longer needs a human's
1257    /// attention, which is only true once the task's story has actually
1258    /// ended well. Two Blocked runs sitting side by side while a third
1259    /// attempt is still in flight must not be touched by this — see
1260    /// `daemon::supersede_prior_runs`, the caller that turns this list into
1261    /// rewritten `run.json` files.
1262    pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
1263        if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
1264            return &[];
1265        }
1266        &self.runs[..self.runs.len() - 1]
1267    }
1268
1269    /// The attempt right after the last occurrence of `run`, if there is one.
1270    /// Resuming records the same run again; it cannot supersede itself.
1271    ///
1272    /// The one place "a later attempt replaced this run" is decided:
1273    /// [`Queue::superseded`] and [`Queue::superseded_by`] (the web UI's
1274    /// note) read it, and so does [`Task::earlier_attempts`], so what is shown
1275    /// and what `crate::handover` acts on cannot disagree.
1276    pub fn successor_of(&self, run: &str) -> Option<&String> {
1277        let pos = self.runs.iter().rposition(|r| r == run)?;
1278        self.runs.get(pos + 1)
1279    }
1280
1281    /// Every run already recorded for this task, i.e. every run that an
1282    /// attempt being minted *right now* supersedes.
1283    ///
1284    /// The new run is not in [`Task::runs`] yet when it decides what to take
1285    /// over (`Task::start` pushes it afterwards), so "has a later attempt" is
1286    /// true of everything listed here by the time the new run exists.
1287    pub fn earlier_attempts(&self) -> &[String] {
1288        &self.runs
1289    }
1290
1291    /// Stamp [`Task::held_at`] on the way into [`TaskStatus::Held`]; a task
1292    /// that is already held keeps the time its hold began.
1293    /// A transition this build makes (start, hold, block, done) makes the task
1294    /// eligible for a chat notice even when it was filed under an older
1295    /// schema; only records no build has touched since stay silent
1296    /// ([`Task::chat_notice_due`]).
1297    fn stamp_schema(&mut self) {
1298        self.schema = self.schema.max(SCHEMA);
1299    }
1300
1301    fn note_held(&mut self) {
1302        self.stamp_schema();
1303        let continuing = self.status == TaskStatus::Held
1304            || (self.status == TaskStatus::Blocked
1305                && (self.blocked_from == Some(TaskStatus::Held)
1306                    || (self.blocked_from.is_none()
1307                        && (self.hold_reason.is_some() || self.hold_source.is_some()))));
1308        if !continuing || self.held_at.is_none() {
1309            self.held_at = Some(Timestamp::now());
1310        }
1311    }
1312
1313    /// Record a failed attempt. Out of attempts means held for a human, rather
1314    /// than retried until the money runs out.
1315    ///
1316    /// Clears [`Task::diagnostic`] unconditionally: it belongs to whatever run
1317    /// produced it, and a caller that has one for *this* attempt sets it
1318    /// itself right after calling this, once it knows the task actually ended
1319    /// up [`TaskStatus::Held`] - see `daemon::diagnostic`. Without the clear, a
1320    /// task released after a diagnosed hold and then failed again for an
1321    /// unrelated, undiagnosed reason (a config error, say) would go on
1322    /// showing the previous run's diagnostic as if it explained the new one.
1323    ///
1324    /// Also records `why` into [`Task::hold_reason`] when this ends up
1325    /// [`TaskStatus::Held`], so the notification and `magi task show` say the
1326    /// same thing as [`Task::last_error`] instead of leaving the hold's own
1327    /// reason blank.
1328    pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
1329        let why = why.into();
1330        if self.status == TaskStatus::Running {
1331            let run = self.runs.last().cloned();
1332            self.notice_cause("run-failed", run.as_deref(), Some(self.attempts));
1333        }
1334        self.diagnostic = None;
1335        self.park_reason = None;
1336        self.status = if self.attempts >= self.effective_attempt_limit(max_attempts) {
1337            self.note_held();
1338            self.hold_source = Some(HoldSource::Machine);
1339            self.hold_reason = Some(why.clone());
1340            TaskStatus::Held
1341        } else {
1342            TaskStatus::Failed
1343        };
1344        self.last_error = Some(why);
1345    }
1346
1347    /// Record an attempt that failed for a reason the task is not responsible
1348    /// for - the agent CLIs ran out of quota and the judging panel collapsed.
1349    ///
1350    /// This refunds the attempt on purpose. A quota window closing at 4am must
1351    /// not spend the backlog's retry budget: the operator would come back to a
1352    /// queue of held tasks that were never actually judged, and would have to
1353    /// release every one by hand to find out which had a real problem. The task
1354    /// goes back to `Failed`, which the loop retries, so a reset quota picks the
1355    /// work up where it stopped.
1356    pub fn stall(&mut self, why: impl Into<String>) {
1357        self.last_error = Some(why.into());
1358        self.diagnostic = None;
1359        self.attempts = self.attempts.saturating_sub(1);
1360        self.status = TaskStatus::Failed;
1361    }
1362
1363    /// Record a run that parked at a node boundary. Nothing failed: the attempt
1364    /// is refunded, [`Task::last_error`] is left as it was, the run history and
1365    /// `fresh_start` are kept so the next loop resumes the same run, and the
1366    /// reason goes to [`Task::park_reason`]. Runnable like a requeued task and
1367    /// never counted against `max_attempts`.
1368    pub fn park(&mut self, why: impl Into<String>) {
1369        self.diagnostic = None;
1370        self.attempts = self.attempts.saturating_sub(1);
1371        self.status = TaskStatus::Parked;
1372        self.park_reason = Some(why.into());
1373        // The variant and field are schema 11; stamp it so an older build
1374        // refuses the record instead of half-reading it.
1375        self.schema = self.schema.max(SCHEMA);
1376    }
1377
1378    /// Record a settled pass. `attempts_before` is the count just before the
1379    /// settle, so a refund is read off what actually changed, not inferred.
1380    pub fn record_pass(
1381        &mut self,
1382        run: &str,
1383        pass_index: usize,
1384        attempts_before: usize,
1385        parked: Option<crate::run::ParkCause>,
1386    ) {
1387        if self
1388            .passes
1389            .iter()
1390            .any(|p| p.run == run && p.pass_index == pass_index)
1391        {
1392            return;
1393        }
1394        let cost = if self.attempts < attempts_before {
1395            PassCost::Refunded
1396        } else {
1397            PassCost::Spent
1398        };
1399        self.passes.push(TaskPass {
1400            run: run.to_owned(),
1401            pass_index,
1402            cost,
1403            attempts_after: self.attempts,
1404            settled_at: Timestamp::now(),
1405            parked,
1406        });
1407        self.schema = self.schema.max(SCHEMA);
1408    }
1409
1410    /// Whether this held task may only be released by an operator.
1411    ///
1412    /// Old files did not record a source. Preserve every such hold rather
1413    /// than guessing that it was automatic and risking duplicate work. New
1414    /// automatic holds record [`HoldSource::Machine`] and remain recoverable.
1415    pub fn operator_held(&self) -> bool {
1416        self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1417    }
1418
1419    /// Take this task out of the loop's reach by an operator action.
1420    ///
1421    /// Clears `blocked_by`/`block_reason` unconditionally, the same as
1422    /// [`Task::release`] and for the same reason its own comment already
1423    /// gives: a human choosing to hold a *blocked* task overrides its wait
1424    /// outright, the same as it overrides an ordinary hold. Without this, a
1425    /// task held straight out of [`TaskStatus::Blocked`] - the web UI's "Hold"
1426    /// button is reachable on a blocked task, same as "Mark done" - kept
1427    /// reading as still waiting on a dependency it no longer had any claim on.
1428    pub fn hold_manual(&mut self, reason: Option<String>) {
1429        // A later explicit hold supersedes outstanding triage answers.
1430        self.operator_decided_at = Some(Timestamp::now());
1431        self.hold_unrequested_retry(reason);
1432    }
1433
1434    /// Keep a hand-started run from retrying automatically, without recording
1435    /// an operator decision that could invalidate its outstanding questions.
1436    pub(crate) fn hold_unrequested_retry(&mut self, reason: Option<String>) {
1437        self.park_reason = None;
1438        self.note_held();
1439        self.status = TaskStatus::Held;
1440        if reason.is_some() {
1441            self.hold_reason = reason;
1442        }
1443        self.hold_source = Some(HoldSource::Manual);
1444        self.blocked_by.clear();
1445        self.block_reason = None;
1446        self.blocked_from = None;
1447    }
1448
1449    /// Take this task out of the loop's reach during automatic recovery.
1450    ///
1451    /// Clears `blocked_by`/`block_reason` for the same reason
1452    /// [`Task::hold_manual`] does.
1453    pub fn hold_machine(&mut self, reason: Option<String>) {
1454        self.park_reason = None;
1455        self.note_held();
1456        self.status = TaskStatus::Held;
1457        if reason.is_some() {
1458            self.hold_reason = reason;
1459        }
1460        self.hold_source = Some(HoldSource::Machine);
1461        self.blocked_by.clear();
1462        self.block_reason = None;
1463        self.blocked_from = None;
1464    }
1465
1466    /// Block this task on other task ids and/or open question ids, chosen by
1467    /// `crate::conduct`. Pure: the caller still owns writing it back with
1468    /// [`Queue::put`].
1469    ///
1470    /// Records [`Task::blocked_from`] the first time this moves the task into
1471    /// [`TaskStatus::Blocked`], and leaves it alone on a later call that adds
1472    /// or replaces `blocked_by` while the task is already `Blocked` - a
1473    /// second question about an already-blocked task must not overwrite the
1474    /// status it should eventually return to with `Blocked` itself.
1475    pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1476        self.stamp_schema();
1477        let before: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
1478        let after: std::collections::BTreeSet<_> = blocked_by.iter().collect();
1479        if self.status != TaskStatus::Blocked || before != after {
1480            self.blocked_at = Some(Timestamp::now());
1481        }
1482        if self.status != TaskStatus::Blocked {
1483            self.blocked_from = Some(self.status);
1484        }
1485        self.status = TaskStatus::Blocked;
1486        self.blocked_by = blocked_by;
1487        self.block_reason = reason;
1488    }
1489
1490    /// Remove one resolved dependency (a task id that became [`TaskStatus::Done`],
1491    /// or a question id that became [`crate::ask::QuestionStatus::Answered`]).
1492    /// Once nothing is left in [`Task::blocked_by`], the task returns to
1493    /// whatever [`Task::blocked_from`] recorded - deciding *why* a task was
1494    /// blocked was `crate::conduct`'s job, but noticing a dependency resolved
1495    /// needs no model at all, and restoring the status it interrupted needs
1496    /// nothing more than what `block` already wrote down.
1497    ///
1498    /// A task blocked while `Running` restores to [`TaskStatus::Queued`]
1499    /// instead: whatever process was running it is gone by the time this
1500    /// runs, so there is nothing left to resume. A task with no recorded
1501    /// `blocked_from` - a pre-schema-4 record, or one blocked before this
1502    /// field existed - falls back to [`TaskStatus::Held`] when it still
1503    /// carries hold evidence ([`Task::hold_reason`] or [`Task::hold_source`],
1504    /// neither ever cleared by `block`), and to `Queued` otherwise: the same
1505    /// choice `block` itself would have recorded, reconstructed from what
1506    /// survived.
1507    ///
1508    /// A no-op, on purpose, for a task that is not [`TaskStatus::Blocked`]:
1509    /// `crate::daemon`'s deterministic resolver runs over every task on every
1510    /// poll, and a task that moved on for some other reason must not be
1511    /// dragged back by a stale id it still happens to carry.
1512    pub fn unblock(&mut self, resolved_id: &str) {
1513        if self.status != TaskStatus::Blocked {
1514            return;
1515        }
1516        self.blocked_by.retain(|id| id != resolved_id);
1517        self.restore_if_unblocked();
1518    }
1519
1520    /// A dependency this task waited on was deleted on purpose
1521    /// ([`Queue::remove`]): drop just that id from [`Task::blocked_by`] and,
1522    /// once nothing is left, return to the status the block interrupted, the
1523    /// same way [`Task::unblock`] does. Returns whether anything changed.
1524    ///
1525    /// Never a hold: the operator chose the deletion, so the dependent is not
1526    /// in trouble. Touches nothing but a `Blocked` task that actually names
1527    /// `deleted_id`, so a `Held` task (or a manual hold) is left exactly as it
1528    /// was. A dependency that vanished *without* going through a deletion is
1529    /// not this method's case - that stays a machine hold.
1530    pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1531        if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1532            return false;
1533        }
1534        self.blocked_by.retain(|id| id != deleted_id);
1535        self.restore_if_unblocked();
1536        true
1537    }
1538
1539    /// The shared tail of [`Task::unblock`] and [`Task::dependency_deleted`]:
1540    /// with nothing left in `blocked_by`, restore the interrupted status and
1541    /// clear the now-stale block bookkeeping.
1542    fn restore_if_unblocked(&mut self) {
1543        if self.blocked_by.is_empty() {
1544            self.status = match self.blocked_from {
1545                Some(TaskStatus::Running) => TaskStatus::Queued,
1546                Some(other) => other,
1547                None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1548                    TaskStatus::Held
1549                }
1550                None => TaskStatus::Queued,
1551            };
1552            self.block_reason = None;
1553            self.blocked_from = None;
1554        }
1555    }
1556
1557    /// Record that a question `crate::conduct` asked about this task has been
1558    /// answered, so the answer's content — not just the fact that the
1559    /// question is gone — reaches the next conductor prompt and the next
1560    /// run's instruction. See [`Task::answers`].
1561    pub fn record_answer(&mut self, question: String, answer: String) {
1562        self.answers.push(AnsweredQuestion { question, answer });
1563    }
1564
1565    /// Requeue this task to reopen its last run as a review-only pass against
1566    /// `branch` (`crate::graph::Runner::review`) rather than competing from
1567    /// scratch. See [`Task::review_branch`].
1568    pub fn request_review(&mut self, branch: String) {
1569        self.release();
1570        self.review_branch = Some(branch);
1571        self.review_instruction = Some(self.instruction.clone());
1572    }
1573
1574    /// Requeue after a conductor chose a new competition. Unlike an ordinary
1575    /// operator release, this deliberately does not resume the old run.
1576    pub fn requeue(&mut self) {
1577        self.release();
1578        // A fresh competition is the opposite of a review of the old branch.
1579        self.review_branch = None;
1580        self.review_instruction = None;
1581        self.fresh_start = true;
1582    }
1583
1584    /// Hold this task because a takeover of `branch` was refused, keeping
1585    /// [`Task::review_branch`] so that once the operator has cleaned up and
1586    /// released it, the retry is a review of the same branch rather than a
1587    /// resume or a fresh competition. Spends no attempt.
1588    ///
1589    /// This is the only way a machine hold keeps a `review_branch`: the
1590    /// daemon takes it at the start of every attempt, so [`Task::release`]
1591    /// relies on that to tell this hold from every other.
1592    pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1593        if branch.is_some() {
1594            self.review_branch = branch;
1595            self.review_instruction = Some(self.instruction.clone());
1596        }
1597        self.hold_machine(Some(reason));
1598    }
1599
1600    /// Change the urgent lane setting without releasing a held task.
1601    pub fn set_urgent(&mut self, urgent: bool) -> Result<()> {
1602        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1603            bail!(
1604                "task {} is {}; urgent can only change while queued or held; \
1605                 wait for the active attempt to finish and requeue or hold the task first",
1606                self.short(),
1607                self.status.as_str()
1608            );
1609        }
1610        self.urgent = urgent;
1611        Ok(())
1612    }
1613
1614    /// Change how urgently this task should run next.
1615    ///
1616    /// Refused once the task is `running`: priority only feeds the sort
1617    /// [`Queue::next_runnable`] does over tasks waiting to be claimed, and a
1618    /// running task has already left that pool. Accepting the write anyway
1619    /// would look like it worked while changing nothing until - and unless -
1620    /// this attempt fails and the task becomes runnable again, which is a
1621    /// surprise the phone should not hand back as a success.
1622    pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1623        if self.status == TaskStatus::Running {
1624            bail!(
1625                "task {} is running; its priority cannot be changed until \
1626                 this attempt finishes",
1627                self.short()
1628            );
1629        }
1630        self.priority = priority;
1631        Ok(())
1632    }
1633
1634    /// Mark (or unmark) this task to interrupt whatever `magi serve` already
1635    /// has in flight, once `[daemon] pause_for_interrupts` is on. See
1636    /// [`Task::interrupt`].
1637    ///
1638    /// Setting it is restricted to a task the loop could pick up on its own
1639    /// right now - [`TaskStatus::runnable`] - for the same reason as
1640    /// [`Task::set_priority`]: a task already `running` has been claimed, and
1641    /// a task that is `done`, `held`, or `blocked` is not going to compete
1642    /// for the daemon's attention regardless of this flag. Unlike priority,
1643    /// this is never silently inert while `running` - it is refused outright,
1644    /// because the entire feature this flag drives (`crate::daemon`'s
1645    /// interrupt scheduler) is scoped to tasks still waiting to be claimed.
1646    /// Clearing it back to `false` carries no such risk and is always
1647    /// allowed, including on a task that moved on since it was set.
1648    pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1649        if interrupt && !self.status.runnable() {
1650            bail!(
1651                "task {} is {}; only a queued or failed task can be marked \
1652                 to interrupt",
1653                self.short(),
1654                self.status.as_str()
1655            );
1656        }
1657        self.interrupt = interrupt;
1658        Ok(())
1659    }
1660
1661    /// Resolve the task override against the running daemon's limit.
1662    pub fn effective_attempt_limit(&self, daemon_limit: usize) -> usize {
1663        self.attempt_limit.map_or(daemon_limit, NonZeroUsize::get)
1664    }
1665
1666    /// Change the budget without resetting attempts or releasing a hold.
1667    /// Callers must hold the queue claim and reload the task before calling.
1668    pub fn change_attempt_limit(&mut self, limit: Option<NonZeroUsize>) -> Result<()> {
1669        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1670            bail!("only queued or held tasks can change the attempt limit");
1671        }
1672        self.stamp_schema();
1673        self.attempt_limit = limit;
1674        Ok(())
1675    }
1676
1677    /// Explicit implementer choice, with Solo taking precedence for old records.
1678    pub fn implementers(&self) -> Option<usize> {
1679        if self.solo {
1680            Some(1)
1681        } else {
1682            self.overrides.as_ref().and_then(|o| o.candidates)
1683        }
1684    }
1685
1686    /// Validate against this task's configuration and change only its seat count.
1687    /// Callers must hold the queue claim and reload the task before calling.
1688    pub fn change_implementers(&mut self, count: u8) -> Result<()> {
1689        self.check_implementers_editable()?;
1690        let explicit = self.overrides.as_ref().and_then(|o| o.config.as_deref());
1691        let (mut config, _) = crate::config::Config::discover(&self.repo, explicit)?;
1692        if let Some(o) = &self.overrides {
1693            o.apply(&mut config);
1694        }
1695        self.set_implementers(count, &config)
1696    }
1697
1698    fn check_implementers_editable(&self) -> Result<()> {
1699        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1700            bail!(
1701                "task {} is {}; only queued or held tasks can change implementers",
1702                self.short(),
1703                self.status.as_str()
1704            );
1705        }
1706        Ok(())
1707    }
1708
1709    /// Normalize Solo and candidate overrides without altering task history.
1710    pub fn set_implementers(&mut self, count: u8, config: &crate::config::Config) -> Result<()> {
1711        self.check_implementers_editable()?;
1712        crate::talk::check_implementers(count, config).map_err(anyhow::Error::msg)?;
1713        self.stamp_schema();
1714        self.solo = count == 1;
1715        let mut overrides = self.overrides.clone().unwrap_or_else(|| RunOverrides {
1716            inherit_daemon_options: true,
1717            ..Default::default()
1718        });
1719        overrides.candidates = (count > 1).then_some(usize::from(count));
1720        let mut settings = overrides.clone();
1721        settings.inherit_daemon_options = false;
1722        // Some(empty) is a direct-run setting too: retain its isolation on Solo.
1723        self.overrides = if overrides.inherit_daemon_options && settings == RunOverrides::default()
1724        {
1725            None
1726        } else {
1727            Some(overrides)
1728        };
1729        Ok(())
1730    }
1731
1732    /// Replace this task's title and instruction wholesale.
1733    ///
1734    /// Restricted to `queued` and `held`. A `running` task's instruction has
1735    /// already been handed to the graph, so a run in flight and the file on
1736    /// disk must not be allowed to disagree about what was asked; a `done` or
1737    /// `failed` task is a record of what actually happened and editing it
1738    /// after the fact would falsify that record. `id`, `created_at`,
1739    /// `source`, and `runs` are left untouched on purpose - an edit stands in
1740    /// for "delete and refile", and keeping the id, the timestamp, the
1741    /// attribution, and the run history is the entire reason it exists
1742    /// instead.
1743    pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1744        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1745            bail!(
1746                "task {} is {}; only a queued or held task's instruction can \
1747                 be edited",
1748                self.short(),
1749                self.status.as_str()
1750            );
1751        }
1752        // Older retained reviews have no criteria snapshot yet. Capture the
1753        // old instruction before this supported edit can change the target.
1754        if self.review_branch.is_some() && self.review_instruction.is_none() {
1755            self.review_instruction = Some(self.instruction.clone());
1756        }
1757        if self.instruction != instruction {
1758            // A run selected for earlier criteria is no longer authorized by
1759            // this revised instruction. Title-only edits preserve the pin.
1760            self.resume_override = None;
1761        }
1762        self.title = title;
1763        self.instruction = instruction;
1764        Ok(())
1765    }
1766
1767    /// Record a run that produced a pull request without merging it.
1768    ///
1769    /// The task is held rather than retried, and it costs no further attempt
1770    /// either way. The work the task asked for exists: it is sitting on a
1771    /// branch, in a pull request, waiting for CI or for a person. Retrying
1772    /// would spend the whole competition budget a second time and then race a
1773    /// second branch against the pull request the first one opened - which is
1774    /// exactly what happened to run 01c2, whose finished and green pull request
1775    /// was re-competed from scratch four seconds after it opened.
1776    ///
1777    /// A pull request nobody merged is a request for a person, not a failure.
1778    ///
1779    /// Records `why` into [`Task::hold_reason`] as well as
1780    /// [`Task::last_error`], so the notification centre and `magi task show`
1781    /// say why the task is held rather than "no reason recorded". This is
1782    /// also the path a verified no-op with an outstanding `magi ask` question
1783    /// settles through - see `daemon::settle` and `daemon::settle_and_diagnose`,
1784    /// which append the question id to `hold_reason` when one is still open
1785    /// for the run.
1786    pub fn handed_off(&mut self, why: impl Into<String>) {
1787        let why = why.into();
1788        self.diagnostic = None;
1789        self.note_held();
1790        self.status = TaskStatus::Held;
1791        self.hold_source = Some(HoldSource::Machine);
1792        self.hold_reason = Some(why.clone());
1793        self.last_error = Some(why);
1794    }
1795
1796    /// Release on explicit operator authority, rather than automatic recovery.
1797    pub fn release_operator(&mut self, source: &str, question_id: Option<&str>) {
1798        self.recovery_budget = None;
1799        self.recovery_generations.clear();
1800        self.recovery_admission = None;
1801        if self
1802            .resumption
1803            .as_ref()
1804            .is_some_and(|r| r.instruction != self.instruction)
1805        {
1806            // Also cover persisted revisions whose earlier approval still
1807            // carries a transient pin when this release is applied.
1808            self.resume_override = None;
1809        }
1810        let approval = Resumption {
1811            generation: self.resumption.as_ref().map_or(1, |r| r.generation + 1),
1812            source: source.to_owned(),
1813            question_id: question_id.map(str::to_owned),
1814            at: Timestamp::now(),
1815            instruction: self.instruction.clone(),
1816            answers_through: self.answers.len(),
1817            runs_through: self.runs.len(),
1818            blocking_evidence: Vec::new(),
1819            cleared_reason: self
1820                .hold_reason
1821                .clone()
1822                .or_else(|| self.block_reason.clone()),
1823        };
1824        self.release();
1825        self.operator_decided_at = Some(approval.at);
1826        if let Some(previous) = self.resumption.replace(approval) {
1827            self.resumption_history.push(previous);
1828        }
1829    }
1830
1831    /// Record a failed graph pass without allowing automatic requeues to refund it.
1832    pub(crate) fn recovery_failed(&mut self, state: &crate::run::RunState) {
1833        let (base, inputs) = self.recovery_admission.as_ref().map_or_else(
1834            || (state.base_commit.clone(), recovery_inputs(&state.seeds)),
1835            |a| (a.base.clone(), a.inputs.clone()),
1836        );
1837        let progress = recovery_progress(state);
1838        let stage = if state.prep_failure.is_some() {
1839            "preparation"
1840        } else {
1841            "execution"
1842        };
1843        let matches = |b: &RecoveryBudget| {
1844            b.instruction == self.instruction
1845                && b.base == base
1846                && b.inputs == inputs
1847                && (progress.is_none() || b.progress == progress)
1848        };
1849        if !self.recovery_budget.as_ref().is_some_and(matches) {
1850            let previous = self
1851                .recovery_generations
1852                .iter()
1853                .position(matches)
1854                .map(|index| self.recovery_generations.remove(index));
1855            if let Some(old) = self.recovery_budget.take() {
1856                self.recovery_generations.push(old);
1857            }
1858            self.recovery_budget = previous;
1859        }
1860        if let Some(budget) = &mut self.recovery_budget {
1861            budget.failures = budget.failures.saturating_add(1);
1862            budget.stage = stage.to_owned();
1863        } else {
1864            self.recovery_budget = Some(RecoveryBudget {
1865                instruction: self.instruction.clone(),
1866                base,
1867                inputs,
1868                stage: stage.to_owned(),
1869                failures: 1,
1870                progress,
1871            });
1872        }
1873    }
1874
1875    pub(crate) fn recovery_exhausted(
1876        &self,
1877        base: &str,
1878        inputs: &str,
1879        limit: usize,
1880    ) -> Option<String> {
1881        let same_input = |b: &&RecoveryBudget| {
1882            b.instruction == self.instruction && b.base == base && b.inputs == inputs
1883        };
1884        let budget = self
1885            .recovery_budget
1886            .as_ref()
1887            .filter(same_input)
1888            .or_else(|| self.recovery_generations.iter().rev().find(same_input))?;
1889        (budget.instruction == self.instruction && budget.base == base && budget.inputs == inputs
1890            && budget.failures >= limit.max(1)).then(|| format!(
1891                "[recovery-budget] {} failed passes against unchanged inputs ({}). Inspect the retained recovery sources and correct the cause, change the inputs, or explicitly release the task to approve another generation. Automatic requeue does not reset this budget.",
1892                budget.failures, budget.stage))
1893    }
1894
1895    /// Put a held or finished task back in line, with its attempt count reset
1896    /// so a release is a real second chance rather than an instant re-hold.
1897    /// The run history is kept: attempts reset, evidence does not.
1898    pub fn release(&mut self) {
1899        // Freeze the unresolved cause before a recovery can mint a run without
1900        // doing work. Approval and timestamps alone are not new outcomes.
1901        self.import_chat_notice();
1902        if self.hold_notice_cause.is_none()
1903            && matches!(self.status, TaskStatus::Held | TaskStatus::Blocked)
1904        {
1905            self.hold_notice_cause = Some(self.unresolved_hold_cause());
1906        }
1907        let refused_handover = self.status == TaskStatus::Held
1908            && self.hold_source == Some(HoldSource::Machine)
1909            && self.review_branch.is_some()
1910            && self
1911                .review_instruction
1912                .as_ref()
1913                .is_none_or(|i| *i == self.instruction);
1914        self.status = TaskStatus::Queued;
1915        self.held_at = None;
1916        self.attempts = 0;
1917        self.last_error = None;
1918        self.park_reason = None;
1919        // Otherwise the next person who holds this task reads a reason that
1920        // belonged to whatever it was waiting on last time.
1921        self.hold_reason = None;
1922        self.hold_source = None;
1923        self.diagnostic = None;
1924        // A release also un-blocks: the dependency or question `blocked_by`
1925        // named may still be unresolved, but a human (or `crate::conduct`)
1926        // choosing to release the task overrides that wait outright, the same
1927        // as it overrides an ordinary hold.
1928        self.blocked_by.clear();
1929        self.block_reason = None;
1930        self.blocked_from = None;
1931        // A machine hold that still names a review branch is a refused
1932        // handover (see [`Task::hold_for_handover`]): the retry must review
1933        // that same branch. Any other release drops it.
1934        if !refused_handover {
1935            self.review_branch = None;
1936            self.review_instruction = None;
1937        }
1938        self.fresh_start = false;
1939    }
1940}
1941
1942/// Copy `src` into `dir` as `name`, or `stem-2.ext`, `stem-3.ext`, ... when
1943/// that is taken. `create_new` makes the collision check atomic and also
1944/// catches a case-insensitive filesystem. Returns the stored name and path.
1945fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
1946    use std::io::ErrorKind;
1947    let (stem, ext) = match name.rfind('.') {
1948        Some(i) if i > 0 => (&name[..i], &name[i..]),
1949        _ => (name, ""),
1950    };
1951    for n in 1u32.. {
1952        let candidate = if n == 1 {
1953            name.to_owned()
1954        } else {
1955            let suffix = format!("-{n}");
1956            let room = 64usize.saturating_sub(suffix.len() + ext.len());
1957            let stem: String = stem.chars().take(room).collect();
1958            format!("{stem}{suffix}{ext}")
1959        };
1960        if !crate::ask::valid_asset_name(&candidate) {
1961            bail!("no valid attachment name is left for `{name}`");
1962        }
1963        let path = dir.join(&candidate);
1964        match std::fs::OpenOptions::new()
1965            .write(true)
1966            .create_new(true)
1967            .open(&path)
1968        {
1969            Ok(mut out) => {
1970                let copied = std::fs::File::open(src)
1971                    .and_then(|mut input| std::io::copy(&mut input, &mut out));
1972                if let Err(e) = copied {
1973                    drop(out);
1974                    let _ = std::fs::remove_file(&path);
1975                    return Err(e).with_context(|| format!("copy {}", src.display()));
1976                }
1977                return Ok((candidate, path));
1978            }
1979            Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
1980            Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
1981        }
1982    }
1983    unreachable!("the counter never runs out")
1984}
1985
1986/// Minimum age before a dead writer's lock may be recovered, and maximum wait
1987/// for a contending writer. Age alone never revokes a live writer's ownership.
1988const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
1989
1990/// Guard for [`Queue::lock_task`]; removes the lock file on drop, but only
1991/// while the file still carries this guard's owner token.
1992pub(crate) struct TaskLock {
1993    path: PathBuf,
1994    token: String,
1995}
1996
1997/// A token no other holder shares: pid, then a mix of entropy and a process
1998/// counter so threads born in the same nanosecond still differ.
1999fn owner_token() -> String {
2000    static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
2001    let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
2002    let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
2003    format!("{}-{:016x}", std::process::id(), r.next_u64())
2004}
2005
2006/// The marker that serializes everyone removing the lock that carries `token`.
2007/// Hashed (FNV-1a) so an empty or foreign token is still a valid file name.
2008fn break_marker(path: &Path, token: &str) -> PathBuf {
2009    let mut h: u64 = 0xcbf29ce484222325;
2010    for b in token.bytes() {
2011        h ^= u64::from(b);
2012        h = h.wrapping_mul(0x100000001b3);
2013    }
2014    let mut name = path.as_os_str().to_owned();
2015    name.push(format!(".break-{h:016x}"));
2016    PathBuf::from(name)
2017}
2018
2019fn older_than_stale(path: &Path) -> bool {
2020    std::fs::metadata(path)
2021        .and_then(|m| m.modified())
2022        .ok()
2023        .and_then(|t| t.elapsed().ok())
2024        .is_some_and(|age| age > TASK_LOCK_STALE)
2025}
2026
2027fn writer_may_be_gone(token: &str) -> bool {
2028    let Some(pid) = token
2029        .split_once('-')
2030        .and_then(|(pid, _)| pid.parse::<u32>().ok())
2031    else {
2032        // Legacy/incomplete tokens have no valid writer identity. They remain
2033        // subject to the age and compare/remove checks.
2034        return true;
2035    };
2036    if pid == std::process::id() {
2037        return false;
2038    }
2039    // Unknown process-table access must not steal a long-running retirement.
2040    crate::proc::pid_status(pid) == Some(false)
2041}
2042
2043/// How many levels of "a stale marker guarding a stale marker" are recovered
2044/// before giving up. Each level needs its own crashed remover, so the cap is
2045/// never reached without several independent crashes.
2046const MAX_MARKER_DEPTH: u8 = 3;
2047
2048/// Take the break marker for `token`, or `None` if another remover holds it.
2049/// The marker carries its own owner token. One older than [`TASK_LOCK_STALE`]
2050/// was left by a remover that died; it is cleared by the same compare-and-
2051/// remove discipline as a lock, guarded by a marker of its own, so a recoverer
2052/// that judged it stale cannot delete a fresh marker that replaced it.
2053fn take_marker(path: &Path, token: &str, depth: u8) -> Option<(PathBuf, String)> {
2054    use std::io::Write;
2055    let marker = break_marker(path, token);
2056    for _ in 0..2 {
2057        match std::fs::OpenOptions::new()
2058            .write(true)
2059            .create_new(true)
2060            .open(&marker)
2061        {
2062            Ok(mut f) => {
2063                let mine = owner_token();
2064                if f.write_all(mine.as_bytes()).is_err() {
2065                    drop(f);
2066                    let _ = std::fs::remove_file(&marker);
2067                    return None;
2068                }
2069                return Some((marker, mine));
2070            }
2071            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2072                let judged = std::fs::read_to_string(&marker).ok();
2073                match judged.filter(|_| older_than_stale(&marker)) {
2074                    Some(judged) if depth < MAX_MARKER_DEPTH => {
2075                        remove_lock_if(&marker, &judged, true, depth + 1)?;
2076                    }
2077                    _ => return None,
2078                }
2079            }
2080            Err(_) => return None,
2081        }
2082    }
2083    None
2084}
2085
2086/// Remove the lock at `path` if it still carries `token` (and, when
2087/// `require_stale`, is still stale). Removal is serialized per token by the
2088/// break marker, so a remover that judged an old lock cannot delete the fresh
2089/// one that replaced it. `None` means the marker was busy: try again.
2090fn remove_lock_if(path: &Path, token: &str, require_stale: bool, depth: u8) -> Option<bool> {
2091    let (marker, mine) = take_marker(path, token, depth)?;
2092    let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
2093        && (!require_stale || older_than_stale(path) && writer_may_be_gone(token));
2094    if still {
2095        let _ = std::fs::remove_file(path);
2096    }
2097    // Release the marker only if it is still ours.
2098    if depth >= MAX_MARKER_DEPTH {
2099        if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
2100            let _ = std::fs::remove_file(&marker);
2101        }
2102    } else {
2103        let _ = remove_lock_if(&marker, &mine, false, depth + 1);
2104    }
2105    Some(still)
2106}
2107
2108/// Break the lock judged stale while it held `judged`. Returns whether it was
2109/// removed; a lock that has since been replaced is left alone.
2110fn break_stale(path: &Path, judged: &str) -> bool {
2111    remove_lock_if(path, judged, true, 0).unwrap_or(false)
2112}
2113
2114impl Drop for TaskLock {
2115    fn drop(&mut self) {
2116        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
2117        loop {
2118            match remove_lock_if(&self.path, &self.token, false, 0) {
2119                Some(_) => return,
2120                None if std::time::Instant::now() > deadline => return,
2121                None => std::thread::sleep(std::time::Duration::from_millis(5)),
2122            }
2123        }
2124    }
2125}
2126
2127/// A queue on disk.
2128#[derive(Debug, Clone)]
2129pub struct Queue {
2130    root: PathBuf,
2131}
2132
2133impl Queue {
2134    /// The operator's queue, `<home>/queue`.
2135    pub fn open() -> Self {
2136        Self::at(crate::run::home().join("queue"))
2137    }
2138
2139    /// A queue at an explicit root. Tests use this; so could an operator who
2140    /// wants a queue per project.
2141    pub fn at(root: PathBuf) -> Self {
2142        Self { root }
2143    }
2144
2145    /// Directory holding the task files.
2146    pub fn root(&self) -> &Path {
2147        &self.root
2148    }
2149
2150    /// Path for one task id.
2151    pub fn path_of(&self, id: &str) -> PathBuf {
2152        self.root.join(format!("{id}.json"))
2153    }
2154
2155    /// Directory holding a task's attachments. A sibling of the task file,
2156    /// not a `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
2157    pub fn attachments_dir(&self, id: &str) -> PathBuf {
2158        self.root.join(format!("{id}.attachments"))
2159    }
2160
2161    /// Copy each of `sources` into `task`'s attachment directory and record
2162    /// the stored names on `task` (persist with [`Queue::put`]). Returns the
2163    /// names, in order.
2164    ///
2165    /// Every source is checked (a readable file, a name passing
2166    /// [`crate::ask::valid_asset_name`]) before anything is copied, and a
2167    /// copy that fails part-way removes only what this call created. A name
2168    /// already taken is never overwritten: it becomes `stem-2.ext`,
2169    /// `stem-3.ext`, ... A name that does not validate is refused rather than
2170    /// sanitised - the operator renames the file.
2171    pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2172        let mut wanted = Vec::new();
2173        for src in sources {
2174            let name = src
2175                .file_name()
2176                .and_then(|n| n.to_str())
2177                .with_context(|| format!("`{}` has no usable file name", src.display()))?;
2178            if !crate::ask::valid_asset_name(name) {
2179                bail!(
2180                    "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
2181                     with no `..`; rename the file and try again"
2182                );
2183            }
2184            if !src.is_file() {
2185                bail!("attachment `{}` is not a file", src.display());
2186            }
2187            wanted.push((src, name));
2188        }
2189        let dir = self.attachments_dir(&task.id);
2190        let existed = dir.is_dir();
2191        std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
2192        let mut created: Vec<PathBuf> = Vec::new();
2193        let mut names = Vec::new();
2194        let mut copy_all = || -> Result<()> {
2195            for (src, name) in &wanted {
2196                let (stored, path) = copy_new(&dir, src, name)?;
2197                created.push(path);
2198                names.push(stored);
2199            }
2200            Ok(())
2201        };
2202        if let Err(e) = copy_all() {
2203            for path in &created {
2204                let _ = std::fs::remove_file(path);
2205            }
2206            if !existed {
2207                let _ = std::fs::remove_dir(&dir);
2208            }
2209            return Err(e);
2210        }
2211        task.attachments.extend(names.iter().cloned());
2212        Ok(names)
2213    }
2214
2215    /// [`Queue::attach`] then [`Queue::put`], undoing the copies when the
2216    /// record cannot be written. Without the undo the files sit in
2217    /// `<id>.attachments/` with no task pointing at them.
2218    ///
2219    /// Only what this call copied is removed (the names `attach` returned,
2220    /// and the directory itself only when it did not exist before and is
2221    /// empty), never an attachment an earlier save recorded. `task` is rolled
2222    /// back too, and the error the caller sees is `put`'s own. This relies on
2223    /// a failed `put` leaving the stored record untouched, which its
2224    /// write-then-rename guarantees.
2225    pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2226        let dir = self.attachments_dir(&task.id);
2227        let existed = dir.is_dir();
2228        let before = task.attachments.len();
2229        let names = self.attach(task, sources)?;
2230        if let Err(e) = self.put(task) {
2231            for name in &names {
2232                let _ = std::fs::remove_file(dir.join(name));
2233            }
2234            if !existed {
2235                let _ = std::fs::remove_dir(&dir);
2236            }
2237            task.attachments.truncate(before);
2238            return Err(e);
2239        }
2240        Ok(names)
2241    }
2242
2243    /// Absolute paths of `task`'s attachments, whatever shape this queue's
2244    /// root has. Absolute because the prompt hands them to an agent whose
2245    /// working directory is somewhere else entirely.
2246    pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
2247        let dir = self.attachments_dir(&task.id);
2248        task.attachments
2249            .iter()
2250            .map(|n| {
2251                let p = dir.join(n);
2252                std::path::absolute(&p).unwrap_or(p)
2253            })
2254            .collect()
2255    }
2256
2257    /// Write a task, atomically, so a daemon killed mid-write leaves the
2258    /// previous state readable rather than a truncated file.
2259    ///
2260    /// Runs already recorded on disk that `task` does not carry are kept, not
2261    /// dropped: `magi run --task` links a run into a task from another
2262    /// process ([`Queue::link_run`]) while a daemon may hold a snapshot taken
2263    /// before it, and writing that snapshot back whole would erase the link.
2264    /// Nothing ever removes a run from a task, so the union loses nothing.
2265    pub fn put(&self, task: &mut Task) -> Result<()> {
2266        let _lock = self.lock_task(&task.id)?;
2267        self.put_unlocked(task)
2268    }
2269
2270    /// Write `task` only if no task with its id exists yet; `Ok(false)` when
2271    /// one does. Unlike [`Queue::put`] it never replaces a record, so a
2272    /// producer with a deterministic id cannot rewind a task that has since
2273    /// started running or finished.
2274    pub fn create_new(&self, task: &mut Task) -> Result<bool> {
2275        let _lock = self.lock_task(&task.id)?;
2276        if self.path_of(&task.id).exists() {
2277            return Ok(false);
2278        }
2279        self.put_unlocked(task)?;
2280        Ok(true)
2281    }
2282
2283    /// [`Queue::put`] for a caller already holding [`Queue::lock_task`].
2284    fn put_unlocked(&self, task: &mut Task) -> Result<()> {
2285        if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
2286            stored.import_chat_notice();
2287            for key in stored.chat_notices {
2288                if !task.chat_notices.contains(&key) {
2289                    task.chat_notices.push(key);
2290                }
2291            }
2292            for run in stored.runs {
2293                if !task.runs.contains(&run) {
2294                    task.runs.push(run);
2295                }
2296            }
2297            for p in stored.passes {
2298                if !task
2299                    .passes
2300                    .iter()
2301                    .any(|q| q.run == p.run && q.pass_index == p.pass_index)
2302                {
2303                    task.passes.push(p);
2304                }
2305            }
2306            // A completion notice already recorded is never rewound by an
2307            // older snapshot: that would report the same state twice.
2308            if let Some(n) = stored.chat_notice
2309                && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
2310            {
2311                task.chat_notice = Some(n);
2312            }
2313        }
2314        task.import_chat_notice();
2315        task.updated_at = Timestamp::now();
2316        std::fs::create_dir_all(&self.root)
2317            .with_context(|| format!("create {}", self.root.display()))?;
2318        let body = serde_json::to_string_pretty(task).context("serialize task")?;
2319        let path = self.path_of(&task.id);
2320        let tmp = path.with_extension("json.tmp");
2321        std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
2322        std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
2323        // A machine hold is news for the notification centre, filed beside
2324        // this queue (`<home>/queue` -> `<home>/notifications`) rather than
2325        // through a process-global, so a queue in a temp directory notifies
2326        // into that directory.
2327        if let (Some(notice), Some(home)) = (
2328            crate::notices::task_held(task),
2329            self.root.parent().filter(|p| !p.as_os_str().is_empty()),
2330        ) {
2331            crate::notices::raise_in(home, notice);
2332        }
2333        Ok(())
2334    }
2335
2336    /// Change the task `id` in place: read it inside its write lock, let `f`
2337    /// edit it, and write it back only when `f` returns `true`. The check a
2338    /// caller makes in `f` therefore sees the record as stored now, not a
2339    /// snapshot taken before another writer got in. Returns whether it wrote.
2340    pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
2341        let id = self.resolve_id(id)?;
2342        let _lock = self.lock_task(&id)?;
2343        let mut task = self.get(&id)?;
2344        if !f(&mut task) {
2345            return Ok(false);
2346        }
2347        self.put_unlocked(&mut task)?;
2348        Ok(true)
2349    }
2350
2351    /// Append `run` to the runs of the task `id` (or an unambiguous prefix),
2352    /// leaving everything else about it alone — see [`Task::link_run`]. Read
2353    /// and written back in one breath, because [`Queue::put`] replaces the
2354    /// whole record. Returns the task as stored.
2355    pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
2356        let id = self.resolve_id(id)?;
2357        // Read inside the lock, so the record written back is the current one
2358        // and a daemon's write cannot slip between the read and the write.
2359        let _lock = self.lock_task(&id)?;
2360        let mut task = self.get(&id)?;
2361        if task.link_run(run) {
2362            self.put_unlocked(&mut task)?;
2363        }
2364        Ok(task)
2365    }
2366
2367    /// Exclusive right to rewrite task `id`'s record, held until the guard
2368    /// drops. Every writer ([`Queue::put`], [`Queue::link_run`]) takes it, so
2369    /// a process linking a run and the daemon saving a transition are
2370    /// serialized instead of last-writer-wins. An old lock may be broken only
2371    /// after its writer is confirmed dead (legacy tokens without a PID retain
2372    /// age-based recovery). Not the claim lock (`<id>.lock`), held for a
2373    /// whole run.
2374    pub(crate) fn lock_task(&self, id: &str) -> Result<TaskLock> {
2375        std::fs::create_dir_all(&self.root)
2376            .with_context(|| format!("create {}", self.root.display()))?;
2377        let path = self.root.join(format!("{id}.write-lock"));
2378        let started = std::time::Instant::now();
2379        loop {
2380            match std::fs::OpenOptions::new()
2381                .write(true)
2382                .create_new(true)
2383                .open(&path)
2384            {
2385                Ok(mut file) => {
2386                    use std::io::Write;
2387                    let token = owner_token();
2388                    if let Err(e) = file.write_all(token.as_bytes()) {
2389                        drop(file);
2390                        let _ = std::fs::remove_file(&path);
2391                        return Err(e).with_context(|| format!("lock {}", path.display()));
2392                    }
2393                    return Ok(TaskLock { path, token });
2394                }
2395                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2396                    // Content first, then age: a lock replaced in between
2397                    // reads fresh, never stale.
2398                    let judged = std::fs::read_to_string(&path).ok();
2399                    let broken = judged
2400                        .filter(|_| older_than_stale(&path))
2401                        .is_some_and(|judged| break_stale(&path, &judged));
2402                    // A break that does not happen (marker busy, recovery
2403                    // exhausted) is a wait like any other, so it meets the
2404                    // same deadline instead of spinning forever.
2405                    if !broken {
2406                        if started.elapsed() > TASK_LOCK_STALE {
2407                            bail!("could not lock task {id}");
2408                        }
2409                        std::thread::sleep(std::time::Duration::from_millis(15));
2410                    }
2411                }
2412                // On Windows, `create_new` on a file whose delete is still
2413                // pending fails with ERROR_ACCESS_DENIED, not `AlreadyExists`.
2414                // Treat it as a race and retry until the deadline.
2415                Err(e)
2416                    if e.kind() == std::io::ErrorKind::PermissionDenied
2417                        && started.elapsed() <= TASK_LOCK_STALE =>
2418                {
2419                    std::thread::sleep(std::time::Duration::from_millis(15));
2420                }
2421                Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2422            }
2423        }
2424    }
2425
2426    /// Load a task by id or unambiguous id prefix.
2427    pub fn get(&self, id: &str) -> Result<Task> {
2428        let resolved = self.resolve_id(id)?;
2429        read_path(&self.path_of(&resolved))
2430    }
2431
2432    /// Remove a task, and the claim lock that belongs to it.
2433    ///
2434    /// `in_flight` comes from the caller — a live daemon's heartbeat naming
2435    /// this task — because the task's own `running` status cannot answer the
2436    /// question. A daemon killed mid-competition leaves the status at
2437    /// `running` and an orphaned `.lock` behind, and a guard that trusted
2438    /// either would make the task undeletable for good: the phone showed
2439    /// exactly that, refusing a task whose daemon had been gone for an hour.
2440    ///
2441    /// So the lock is removed with the task rather than respected. Any lock
2442    /// still there once no live daemon claims the task is by definition stale,
2443    /// and leaving it would make a deleted task look claimed to
2444    /// [`Queue::claim`] and to whoever reads the directory.
2445    ///
2446    /// A deletion is deliberate, so whatever was `blocked` on the id is
2447    /// resolved, not held: a `<id>.removed` tombstone is written first (it
2448    /// is what tells [`missing_blockers`] and the daemon's resolver that this
2449    /// id was deleted on purpose, so a pass that races this call cannot
2450    /// mistake it for a vanished dependency), the record is deleted (the
2451    /// tombstone is taken back if that fails), and then each dependent is
2452    /// rewritten under its claim - see [`Queue::release_dependents_of`].
2453    /// Best-effort for the dependents: one claimed by something else right
2454    /// now, or whose write fails, is released silently by
2455    /// `crate::daemon::resolve_blockers` (or `crate::triage::run_once`) on
2456    /// its next pass, because the tombstone is still there. It does not fail
2457    /// this removal.
2458    ///
2459    /// `questions` is the store [`missing_blockers`] checks a `blocked_by` id
2460    /// against - the same store the caller already resolves `id`'s own home
2461    /// from, passed in rather than reopened here so a test queue at an
2462    /// explicit root never reads the operator's real questions directory.
2463    pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2464        let resolved = self.resolve_id(id)?;
2465        if in_flight {
2466            bail!("task {resolved} is being run by a live daemon right now");
2467        }
2468        self.write_tombstone(&resolved)?;
2469        if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2470        {
2471            // A record that is still there was not deleted: take the marker
2472            // back. One already gone keeps it, so the deletion still reads
2473            // as deliberate.
2474            if self.path_of(&resolved).exists() {
2475                let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2476            }
2477            return Err(e);
2478        }
2479        let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2480        Ok(Removal {
2481            id: resolved,
2482            released,
2483            still_blocked,
2484        })
2485    }
2486
2487    /// Path of the marker saying task `id` was deleted on purpose. Not a
2488    /// `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
2489    fn tombstone_path(&self, id: &str) -> PathBuf {
2490        self.root.join(format!("{id}.removed"))
2491    }
2492
2493    fn write_tombstone(&self, id: &str) -> Result<()> {
2494        let path = self.tombstone_path(id);
2495        let tmp = self.root.join(format!("{id}.removed.tmp"));
2496        std::fs::write(&tmp, b"")
2497            .and_then(|()| std::fs::rename(&tmp, &path))
2498            .with_context(|| format!("write {}", path.display()))
2499    }
2500
2501    /// Was `id` deleted through [`Queue::remove`] (and is its record gone)?
2502    /// A live record always wins: a tombstone never makes an existing task
2503    /// look deleted.
2504    fn was_deleted(&self, id: &str) -> bool {
2505        self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2506    }
2507
2508    /// Apply [`Task::dependency_deleted`] to `task` for every `blocked_by` id
2509    /// that [`Queue::remove`] deleted. Returns those ids. The caller holds
2510    /// the task's claim and saves it.
2511    pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2512        let deleted = deleted_blockers(self, questions, &task.blocked_by);
2513        deleted
2514            .into_iter()
2515            .filter(|id| task.dependency_deleted(id))
2516            .collect()
2517    }
2518
2519    /// Record, as a fixed-wording info notice, that `dependent` stopped
2520    /// waiting on the deleted task `deleted`. Best-effort and beside the
2521    /// queue, like the hold notice [`Queue::put`] files.
2522    pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2523        let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2524            return;
2525        };
2526        let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2527        let waiting = dependent.status == TaskStatus::Blocked;
2528        let message = match (ja, waiting) {
2529            (true, false) => format!(
2530                "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2531                dependent.short(),
2532                short(deleted)
2533            ),
2534            (true, true) => format!(
2535                "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2536                dependent.short(),
2537                short(deleted)
2538            ),
2539            (false, false) => format!(
2540                "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2541                dependent.short(),
2542                short(deleted)
2543            ),
2544            (false, true) => format!(
2545                "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2546                dependent.short(),
2547                short(deleted)
2548            ),
2549        };
2550        crate::notices::raise_in(
2551            home,
2552            crate::notices::Notice::info(
2553                &format!("unblocked:{}:{}", dependent.id, deleted),
2554                message,
2555            )
2556            .link(crate::notices::Link::Task {
2557                id: dependent.id.clone(),
2558            }),
2559        );
2560    }
2561
2562    /// Resolve every `blocked` task naming `dependency` (just deleted) with
2563    /// [`Task::dependency_deleted`]. Returns the ids released to another
2564    /// status and those that remain `blocked` on something else. Each
2565    /// dependent is re-read inside its claim, never written from a stale copy.
2566    fn release_dependents_of(
2567        &self,
2568        dependency: &str,
2569        questions: &Questions,
2570    ) -> (Vec<String>, Vec<String>) {
2571        let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2572        for listed in self.list() {
2573            if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2574                matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2575            }) {
2576                continue;
2577            }
2578            let Ok(_claim) = self.claim(&listed.id) else {
2579                continue;
2580            };
2581            let Ok(mut task) = self.get(&listed.id) else {
2582                continue;
2583            };
2584            let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2585                matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2586            }).cloned().collect();
2587            if references.is_empty() {
2588                continue;
2589            }
2590            for reference in references {
2591                task.dependency_deleted(&reference);
2592            }
2593            if self.put(&mut task).is_err() {
2594                continue;
2595            }
2596            self.note_dependency_deleted(&task, dependency);
2597            if task.status == TaskStatus::Blocked {
2598                still_blocked.push(task.id.clone());
2599            } else {
2600                released.push(task.id.clone());
2601            }
2602        }
2603        (released, still_blocked)
2604    }
2605
2606    /// The attachment-and-record half of [`Queue::remove`]. `remove_record` is
2607    /// how the record file is deleted; production passes `remove_file`, and a
2608    /// test passes a failing one to prove the attachments come back.
2609    fn remove_record_with_attachments(
2610        &self,
2611        resolved: &str,
2612        remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2613    ) -> Result<()> {
2614        // Neither step may strand the other. The attachments are set aside by a
2615        // rename (cheap, and undone if the record cannot be removed), then the
2616        // record goes, and only then is the set-aside directory deleted for
2617        // real. A failure before the record is gone loses nothing; a failure
2618        // after it leaves at most an orphan `.removing` directory, which the
2619        // next removal sweeps.
2620        self.sweep_removed_attachments();
2621        let attachments = self.attachments_dir(resolved);
2622        let aside = self.root.join(format!("{resolved}.attachments.removing"));
2623        let moved = match std::fs::rename(&attachments, &aside) {
2624            Ok(()) => true,
2625            Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2626            Err(e) => {
2627                return Err(e).with_context(|| format!("remove {}", attachments.display()));
2628            }
2629        };
2630        let path = self.path_of(resolved);
2631        if let Err(e) = remove_record(&path) {
2632            if moved {
2633                let _ = std::fs::rename(&aside, &attachments);
2634            }
2635            return Err(e).with_context(|| format!("remove {}", path.display()));
2636        }
2637        if moved {
2638            if let Err(e) = std::fs::remove_dir_all(&aside) {
2639                tracing::warn!("leftover attachments {}: {e}", aside.display());
2640            }
2641        }
2642        let lock = self.lock_path(resolved);
2643        if let Err(e) = std::fs::remove_file(&lock) {
2644            if e.kind() != std::io::ErrorKind::NotFound {
2645                return Err(e).with_context(|| format!("remove {}", lock.display()));
2646            }
2647        }
2648        Ok(())
2649    }
2650
2651    /// Delete `*.attachments.removing` directories a previous [`Queue::remove`]
2652    /// could not finish deleting. The task record is already gone by then, so
2653    /// the id no longer resolves and the removal cannot be retried by name;
2654    /// every later removal sweeps them instead. A directory whose record still
2655    /// exists belongs to a removal in progress and is left alone. Best-effort.
2656    fn sweep_removed_attachments(&self) {
2657        let Ok(entries) = std::fs::read_dir(&self.root) else {
2658            return;
2659        };
2660        for entry in entries.flatten() {
2661            let name = entry.file_name();
2662            let name = name.to_string_lossy();
2663            let Some(id) = name.strip_suffix(".attachments.removing") else {
2664                continue;
2665            };
2666            // A record still on disk means a removal is mid-flight (or about
2667            // to roll back): the directory is its to delete or restore.
2668            if !self.path_of(id).exists() {
2669                if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2670                    tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2671                }
2672            }
2673        }
2674        self.sweep_tombstones();
2675    }
2676
2677    /// Delete `*.removed` markers that are no longer needed: the record is
2678    /// gone, no task's `blocked_by` names the id, and the marker is older than
2679    /// [`TOMBSTONE_GRACE`]. The age is what makes this safe against a
2680    /// concurrent writer: the task list read here can be stale, so a dependent
2681    /// that was just blocked on an id (a conductor applying a decision while a
2682    /// removal is under way) would otherwise lose the marker and be held as a
2683    /// vanished dependency. A marker whose record still exists belongs to a
2684    /// removal in progress and is never touched.
2685    fn sweep_tombstones(&self) {
2686        let Ok(entries) = std::fs::read_dir(&self.root) else {
2687            return;
2688        };
2689        let tasks = self.list();
2690        for entry in entries.flatten() {
2691            let name = entry.file_name();
2692            let name = name.to_string_lossy();
2693            let Some(id) = name.strip_suffix(".removed") else {
2694                continue;
2695            };
2696            if self.path_of(id).exists()
2697                || tasks.iter().any(|t| {
2698                    t.blocked_by
2699                        .iter()
2700                        .any(|b| id.starts_with(b) || id.ends_with(b))
2701                })
2702            {
2703                continue;
2704            }
2705            let old = entry
2706                .metadata()
2707                .and_then(|m| m.modified())
2708                .ok()
2709                .and_then(|m| m.elapsed().ok())
2710                .is_some_and(|age| age >= TOMBSTONE_GRACE);
2711            if old {
2712                let _ = std::fs::remove_file(entry.path());
2713            }
2714        }
2715    }
2716
2717    /// Path of the claim lock for a task. One definition, so `claim` and
2718    /// `remove` cannot end up naming different files.
2719    fn lock_path(&self, id: &str) -> PathBuf {
2720        self.root.join(format!("{id}.lock"))
2721    }
2722
2723    /// Every task on disk, highest priority first and newest first within a
2724    /// priority. This is what `magi task list` and `GET /api/queue` print, so
2725    /// a raised priority has to move a task here the moment it is saved, not
2726    /// only in [`Queue::next_runnable`]'s own ordering - the operator reading
2727    /// the backlog and the loop about to drain it must agree on what "first"
2728    /// means. Every existing task defaults to priority 0, so this is a no-op
2729    /// change from the old newest-first order for a queue nobody has
2730    /// reprioritised.
2731    ///
2732    /// Unreadable files are skipped rather than fatal: one corrupt task must
2733    /// not take the queue - or the web UI, or an unattended daemon - down
2734    /// with it.
2735    pub fn list(&self) -> Vec<Task> {
2736        let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2737            .into_iter()
2738            .flatten()
2739            .flatten()
2740            .map(|e| e.path())
2741            .filter(|p| p.extension().is_some_and(|x| x == "json"))
2742            .filter_map(|p| read_path(&p).ok())
2743            .collect();
2744        tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2745        tasks
2746    }
2747
2748    /// Runs that a later attempt at the same task replaced, mapped to the id
2749    /// of the attempt that replaced them.
2750    ///
2751    /// A task keeps its attempts in order, and the deck showed them as two
2752    /// cards with the same title and no hint which was which: yukimemi asked
2753    /// why `stalled` and `blocked` appeared twice for one task, and the
2754    /// answer - "those are two tries, and the second one exists because of a
2755    /// bug since fixed" - was not on the screen anywhere.
2756    ///
2757    /// Read from the queue rather than stored on the run, because the
2758    /// ordering is the queue's fact: a `RunState` has no idea another attempt
2759    /// happened after it.
2760    pub fn superseded(&self) -> HashMap<String, String> {
2761        let mut by = HashMap::new();
2762        for task in self.list() {
2763            for earlier in &task.runs {
2764                if let Some(later) = task.successor_of(earlier) {
2765                    by.insert(earlier.clone(), later.clone());
2766                }
2767            }
2768        }
2769        by
2770    }
2771
2772    /// Whether `run` is an earlier attempt a later one replaced, and if so
2773    /// the id of that later attempt.
2774    ///
2775    /// Same walk as [`Queue::superseded`], narrowed to one run: a run detail
2776    /// page asks about exactly one run at a time, and this keeps that call
2777    /// site from building (and discarding) the whole map's `HashMap` just to
2778    /// read one entry out of it.
2779    pub fn superseded_by(&self, run: &str) -> Option<String> {
2780        for task in self.list() {
2781            if task.runs.iter().any(|r| r == run) {
2782                return task.successor_of(run).cloned();
2783            }
2784        }
2785        None
2786    }
2787
2788    /// The task's own most recent attempt, when `run` belongs to that task
2789    /// but is not already that attempt.
2790    ///
2791    /// Distinct from [`Queue::superseded_by`], which names only the very
2792    /// next attempt: a chain of retries (A superseded by B superseded by C)
2793    /// leaves an older run pointing at an intermediate one that may itself
2794    /// be unresolved, and a run's own detail page needs to know where the
2795    /// task's story currently stands - the chain's current head, C - not an
2796    /// attempt in the middle of it that a client would otherwise have to
2797    /// walk to by hand.
2798    pub fn latest_attempt(&self, run: &str) -> Option<String> {
2799        for task in self.list() {
2800            if task.runs.iter().any(|r| r == run) {
2801                return task.runs.last().filter(|last| **last != run).cloned();
2802            }
2803        }
2804        None
2805    }
2806
2807    /// The task a daemon should run next, or `None` when the queue is idle.
2808    ///
2809    /// Highest priority first, oldest first within a priority, so a burst of
2810    /// agent-filed work cannot starve the task a human filed this morning.
2811    pub fn next_runnable(&self) -> Option<Task> {
2812        let mut runnable: Vec<Task> = self
2813            .list()
2814            .into_iter()
2815            .filter(|t| t.status.runnable())
2816            .collect();
2817        runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2818        runnable.into_iter().next()
2819    }
2820
2821    /// Take exclusive ownership of a task.
2822    ///
2823    /// The lock is a `create_new` file next to the task, which is atomic on
2824    /// every platform magi targets. It exists so two daemons - or a daemon and
2825    /// a human running `magi run` - cannot drive one task into two competing
2826    /// runs. The returned guard releases on drop, including on panic.
2827    pub fn claim(&self, id: &str) -> Result<Claim> {
2828        std::fs::create_dir_all(&self.root)
2829            .with_context(|| format!("create {}", self.root.display()))?;
2830        let path = self.lock_path(id);
2831        match std::fs::OpenOptions::new()
2832            .write(true)
2833            .create_new(true)
2834            .open(&path)
2835        {
2836            Ok(mut f) => {
2837                use std::io::Write as _;
2838                // Best effort: the pid is for the human looking at a stale lock.
2839                let _ = writeln!(f, "{}", std::process::id());
2840                Ok(Claim { path })
2841            }
2842            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2843                bail!("task {id} is already claimed ({} exists)", path.display())
2844            }
2845            Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2846        }
2847    }
2848
2849    /// Expand an id prefix to exactly one task id.
2850    pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2851        if !prefix.is_empty()
2852            && prefix != "."
2853            && prefix != ".."
2854            && !prefix.contains(['/', '\\'])
2855            && self.path_of(prefix).is_file()
2856        {
2857            return Ok(prefix.to_owned());
2858        }
2859        let ids = record_ids(&self.root, "json")?;
2860        match resolve_reference(prefix, ids.iter().map(String::as_str)) {
2861            Reference::Found(id) => Ok(id),
2862            Reference::Missing => bail!("no task matches `{prefix}`"),
2863            Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
2864        }
2865    }
2866
2867    /// Change detection token for the queue.
2868    ///
2869    /// Combines file names and modification times of all task files in the
2870    /// queue, so adding, modifying, or deleting any task — even an older one —
2871    /// moves the revision and notifies connected clients via the change stream.
2872    /// Returns 0 when the queue is completely empty.
2873    pub fn revision(&self) -> u64 {
2874        self.revision_excluding(&std::collections::BTreeSet::new())
2875    }
2876
2877    /// [`Queue::revision`] ignoring the files of the named task ids, for a
2878    /// caller that tracks those tasks' content some other way.
2879    pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2880        use std::hash::{Hash as _, Hasher as _};
2881
2882        let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2883            .into_iter()
2884            .flatten()
2885            .flatten()
2886            .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2887            .filter(|e| {
2888                let path = e.path();
2889                !path
2890                    .file_stem()
2891                    .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2892            })
2893            .filter_map(|e| {
2894                let name = e.file_name().to_string_lossy().into_owned();
2895                let mtime = e
2896                    .metadata()
2897                    .ok()?
2898                    .modified()
2899                    .ok()?
2900                    .duration_since(std::time::UNIX_EPOCH)
2901                    .ok()?
2902                    .as_millis() as u64;
2903                Some((name, mtime))
2904            })
2905            .collect();
2906
2907        if entries.is_empty() {
2908            return 0;
2909        }
2910
2911        entries.sort_unstable();
2912        let mut hasher = std::hash::DefaultHasher::new();
2913        for (name, mtime) in &entries {
2914            name.hash(&mut hasher);
2915            mtime.hash(&mut hasher);
2916        }
2917        let h = hasher.finish();
2918        if h == 0 { 1 } else { h }
2919    }
2920}
2921
2922/// How long a deletion marker is kept after its record is gone, even with
2923/// nothing referencing it. See [`Queue::sweep_tombstones`].
2924const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2925
2926/// What [`Queue::remove`] did, beyond deleting the named task's own file.
2927#[derive(Debug, Clone)]
2928pub struct Removal {
2929    /// The id actually removed - `id` expanded from a prefix, if it was one.
2930    pub id: String,
2931    /// Every task that was `blocked` on [`Removal::id`] and left the block
2932    /// (back to the status it interrupted) because that was its only
2933    /// remaining dependency.
2934    pub released: Vec<String>,
2935    /// Every task that named [`Removal::id`] and is still `blocked` on
2936    /// another dependency.
2937    pub still_blocked: Vec<String>,
2938}
2939
2940/// Exclusive ownership of a task, released on drop.
2941#[derive(Debug)]
2942pub struct Claim {
2943    path: PathBuf,
2944}
2945
2946impl Drop for Claim {
2947    fn drop(&mut self) {
2948        let _ = std::fs::remove_file(&self.path);
2949    }
2950}
2951
2952/// The first line of a task, trimmed to a title. Used when the caller gives a
2953/// body but no title, which is the normal case for an agent piping a file in.
2954pub fn title_from(instruction: &str, max: usize) -> String {
2955    let Some(line) = first_line(instruction) else {
2956        return "(empty task)".to_owned();
2957    };
2958    if line.chars().count() <= max {
2959        return line.to_owned();
2960    }
2961    let head: String = line.chars().take(max.saturating_sub(1)).collect();
2962    format!("{head}…")
2963}
2964
2965/// The first non-blank line, whatever it is. A markdown heading is the
2966/// task's own summary - agents pipe in `# Rework the config loader` and mean
2967/// exactly that - so it is preferred over the prose beneath it rather than
2968/// skipped as decoration. Leading list and heading markers are stripped
2969/// because they are syntax, not words. `None` when nothing is left.
2970pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2971    let line = instruction
2972        .lines()
2973        .map(str::trim)
2974        .find(|l| !l.is_empty())?
2975        .trim_start_matches(['#', '-', '*', '>', ' '])
2976        .trim();
2977    (!line.is_empty()).then_some(line)
2978}
2979
2980fn read_path(path: &Path) -> Result<Task> {
2981    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2982    let task: Task =
2983        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2984    // Greater-than, not not-equal: every field added since schema 1 carries
2985    // `#[serde(default)]`, so an older task has nothing to say about it and
2986    // defaulting is exactly as good a reading as a value that build never had
2987    // a chance to write. Only a schema *ahead* of this build - a meaning it
2988    // cannot possibly know - is refused rather than guessed at.
2989    if task.schema > SCHEMA {
2990        bail!(
2991            "task {} was written by a different magi (schema {}, this build \
2992             speaks {SCHEMA})",
2993            task.id,
2994            task.schema
2995        );
2996    }
2997    Ok(task)
2998}
2999
3000/// Resolution shared by queue lookups and dependency inventories. Exact ids
3001/// win; otherwise both the CLI prefix and displayed suffix are supported.
3002#[derive(Debug, PartialEq, Eq)]
3003pub(crate) enum Reference {
3004    Found(String),
3005    Missing,
3006    Ambiguous,
3007}
3008
3009pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
3010    let mut hits = Vec::new();
3011    for candidate in ids {
3012        if candidate == id {
3013            return Reference::Found(candidate.to_owned());
3014        }
3015        if candidate.starts_with(id) || candidate.ends_with(id) {
3016            hits.push(candidate);
3017        }
3018    }
3019    match hits.as_slice() {
3020        [] => Reference::Missing,
3021        [one] => Reference::Found((*one).to_owned()),
3022        _ => Reference::Ambiguous,
3023    }
3024}
3025
3026/// Read names, not deserialized records: list() intentionally skips corrupt
3027/// and future-schema records, which still exist and must keep their waiters.
3028pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
3029    let entries = match std::fs::read_dir(root) {
3030        Ok(entries) => entries,
3031        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
3032        Err(e) => return Err(e.into()),
3033    };
3034    let mut ids = Vec::new();
3035    for entry in entries {
3036        let path = entry?.path();
3037        if path.extension().is_some_and(|e| e == extension)
3038            && let Some(id) = path.file_stem().and_then(|s| s.to_str())
3039        {
3040            ids.push(id.to_owned());
3041        }
3042    }
3043    Ok(ids)
3044}
3045
3046/// Unknown and ambiguous dependencies stay waiting; neither is evidence that
3047/// the operator deleted a prerequisite or needs to supply its replacement.
3048#[derive(Debug, PartialEq, Eq)]
3049pub enum Dependency {
3050    /// Existing task record, with its canonical id; its contents may be unreadable.
3051    Task(String),
3052    /// Existing question record, with its canonical id.
3053    Question(String),
3054    /// Explicitly removed task, identified by its tombstone.
3055    Deleted(String),
3056    /// No matching record or tombstone after a successful inventory read.
3057    Missing,
3058    /// More than one record matches the reference.
3059    Ambiguous,
3060    /// The inventory cannot currently be read, so existence is unknown.
3061    Unknown,
3062}
3063
3064/// Resolve against authoritative task, question and deletion record names.
3065pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
3066    let (Ok(tasks), Ok(qs), Ok(removed)) = (
3067        record_ids(queue.root(), "json"),
3068        record_ids(questions.root(), "json"),
3069        record_ids(queue.root(), "removed"),
3070    ) else {
3071        return Dependency::Unknown;
3072    };
3073    // Include tombstones in ambiguity checks: a reused suffix must never
3074    // release a waiter on a different, still-live task.
3075    let mut ids = tasks
3076        .iter()
3077        .chain(qs.iter())
3078        .chain(removed.iter())
3079        .cloned()
3080        .collect::<Vec<_>>();
3081    ids.sort();
3082    ids.dedup();
3083    match resolve_reference(id, ids.iter().map(String::as_str)) {
3084        Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
3085        Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
3086        Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
3087        Reference::Found(_) => Dependency::Unknown,
3088        Reference::Missing => Dependency::Missing,
3089        Reference::Ambiguous => Dependency::Ambiguous,
3090    }
3091}
3092
3093/// Canonicalize only proven dependencies, preserving unknown legacy references.
3094pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
3095    for id in ids {
3096        if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
3097            dependency(queue, questions, id)
3098        {
3099            *id = full;
3100        }
3101    }
3102}
3103
3104/// Live queue facts for decision and deputy prompts. Historical hold reasons
3105/// and question text are claims, never an inventory of existing prerequisites.
3106pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
3107    let inventory = crate::blockers::Inventory::load(queue, questions);
3108    let mut text = String::from(
3109        "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
3110    );
3111    match record_ids(queue.root(), "json") {
3112        Ok(ids) => {
3113            for id in ids {
3114                match queue.get(&id) {
3115                    Ok(task) => text.push_str(&format!(
3116                        "{}: {}; blocked_by={:?}; chain={:?}\n",
3117                        task.id,
3118                        task.status.as_str(),
3119                        task.blocked_by,
3120                        inventory.waits_on(&task)
3121                    )),
3122                    Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
3123                }
3124            }
3125        }
3126        Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
3127    }
3128    text
3129}
3130
3131/// Only a successful directory scan with no matching record proves absence.
3132pub fn missing_blockers(
3133    queue: &Queue,
3134    questions: &Questions,
3135    blocked_by: &[String],
3136) -> Vec<String> {
3137    blocked_by
3138        .iter()
3139        .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
3140        .cloned()
3141        .collect()
3142}
3143
3144/// Return the original references, so legacy short ids are removed by the
3145/// same Task transition as full ids.
3146pub fn deleted_blockers(
3147    queue: &Queue,
3148    questions: &Questions,
3149    blocked_by: &[String],
3150) -> Vec<String> {
3151    blocked_by
3152        .iter()
3153        .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
3154        .cloned()
3155        .collect()
3156}
3157
3158/// The `hold_reason` text for a task quarantined because one or more of its
3159/// `blocked_by` ids no longer exist. Shared by `crate::daemon::resolve_blockers`,
3160/// `crate::triage::run_once`, and [`Queue::remove`]'s own dependent
3161/// quarantine, so the three call sites read as the same event to an operator
3162/// looking at `magi task show` rather than three different wordings for it.
3163///
3164/// Names the full original `blocked_by` list, not just `missing` - a task
3165/// quarantined here can also have named a dependency that was still
3166/// perfectly valid, and [`Task::hold_machine`] clears `blocked_by` on the way
3167/// in, so this text is the only place that information survives for an
3168/// operator deciding whether to release the task outright.
3169pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
3170    missing_blocker_hold_reason_in(blocked_by, missing, "en")
3171}
3172
3173/// [`missing_blocker_hold_reason`] in `language` (Japanese, else English).
3174pub fn missing_blocker_hold_reason_in(
3175    blocked_by: &[String],
3176    missing: &[String],
3177    language: &str,
3178) -> String {
3179    if crate::lang::is_japanese(language) {
3180        format!(
3181            "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
3182            blocked_by.join(", "),
3183            missing.join(", "),
3184        )
3185    } else {
3186        format!(
3187            "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
3188            blocked_by.join(", "),
3189            missing.join(", "),
3190        )
3191    }
3192}
3193
3194/// The last dash-separated part of an id, the form people say aloud.
3195pub fn short(id: &str) -> &str {
3196    id.split('-').next_back().unwrap_or(id)
3197}
3198
3199fn new_id() -> String {
3200    let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
3201    let seed = crate::rng::entropy();
3202    format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
3203}
3204
3205#[cfg(test)]
3206mod tests {
3207    #[test]
3208    fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
3209        let t = Task::new(
3210            "retry".into(),
3211            "do it".into(),
3212            PathBuf::from("."),
3213            Source::Human,
3214        );
3215        let mut value = serde_json::to_value(&t).unwrap();
3216        for field in ["resumption", "resumption_history", "operator_decided_at"] {
3217            value.as_object_mut().unwrap().remove(field);
3218        }
3219        value["schema"] = serde_json::json!(15);
3220        let mut legacy: Task = serde_json::from_value(value).unwrap();
3221        legacy.hold_machine(Some("automatic retry".into()));
3222        legacy.release();
3223        assert!(legacy.resumption.is_none());
3224        assert!(legacy.resumption_history.is_empty());
3225        assert!(legacy.operator_decided_at.is_none());
3226        legacy.release_operator("cli", None);
3227        legacy.start("run".into());
3228        let reloaded: Task =
3229            serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
3230        assert!(reloaded.resume_override.is_none());
3231        assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
3232    }
3233
3234    #[test]
3235    fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
3236        let mut base = RunOverrides {
3237            merge: Some("pr".to_owned()),
3238            candidates: Some(3),
3239            ..RunOverrides::default()
3240        };
3241        base.merge_over(&RunOverrides {
3242            merge: Some("none".to_owned()),
3243            seed: Some(7),
3244            ..RunOverrides::default()
3245        });
3246        assert_eq!(base.merge.as_deref(), Some("none"));
3247        assert_eq!(base.candidates, Some(3));
3248        assert_eq!(base.seed, Some(7));
3249    }
3250
3251    #[test]
3252    fn an_already_landed_task_is_done_with_its_attempt_refunded() {
3253        let mut t = task("relanded");
3254        t.attempts = 1;
3255        t.status = TaskStatus::Running;
3256        t.already_landed("already in main as 0e368de");
3257        assert_eq!(t.status, TaskStatus::Done);
3258        assert_eq!(t.attempts, 0);
3259        assert!(t.hold_reason.is_none());
3260        assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
3261    }
3262
3263    #[test]
3264    fn missing_blocker_reason_follows_the_language() {
3265        let b = vec!["a".to_owned()];
3266        let en = missing_blocker_hold_reason_in(&b, &b, "en");
3267        assert_eq!(en, missing_blocker_hold_reason(&b, &b));
3268        assert!(en.starts_with("blocked on a"));
3269        assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
3270        assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
3271    }
3272
3273    use super::*;
3274
3275    #[test]
3276    fn triage_applied_survives_release_and_old_records_read_as_empty() {
3277        let mut t = Task::new(
3278            "t".to_owned(),
3279            "i".to_owned(),
3280            PathBuf::from("r"),
3281            Source::Human,
3282        );
3283        t.mark_triage_applied("q1");
3284        t.mark_triage_applied("q1");
3285        t.hold_machine(Some("x".to_owned()));
3286        t.release();
3287        assert_eq!(t.triage_applied, ["q1"]);
3288        assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
3289
3290        let mut v = serde_json::to_value(&t).unwrap();
3291        v.as_object_mut().unwrap().remove("triage_applied");
3292        let old: Task = serde_json::from_value(v).unwrap();
3293        assert!(old.triage_applied.is_empty());
3294    }
3295
3296    #[test]
3297    fn task_counts_of_empty_is_all_zero() {
3298        assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
3299    }
3300
3301    #[test]
3302    fn task_counts_of_tallies_every_status() {
3303        let mut queued = Task::new(
3304            "q".to_owned(),
3305            "i".to_owned(),
3306            PathBuf::from("."),
3307            Source::Human,
3308        );
3309        queued.status = TaskStatus::Queued;
3310        let mut running = queued.clone();
3311        running.status = TaskStatus::Running;
3312        let mut done = queued.clone();
3313        done.status = TaskStatus::Done;
3314        let mut failed = queued.clone();
3315        failed.status = TaskStatus::Failed;
3316        let mut held = queued.clone();
3317        held.status = TaskStatus::Held;
3318        let mut blocked = queued.clone();
3319        blocked.status = TaskStatus::Blocked;
3320
3321        let mut parked = queued.clone();
3322        parked.status = TaskStatus::Parked;
3323
3324        let counts = TaskCounts::of(&[
3325            queued,
3326            running,
3327            done.clone(),
3328            done,
3329            failed,
3330            held,
3331            blocked,
3332            parked,
3333        ]);
3334        assert_eq!(
3335            counts,
3336            TaskCounts {
3337                queued: 1,
3338                running: 1,
3339                done: 2,
3340                failed: 1,
3341                held: 1,
3342                blocked: 1,
3343                parked: 1,
3344            }
3345        );
3346    }
3347
3348    #[test]
3349    fn task_implementers_persist_normalized_without_losing_history() {
3350        let (_dir, q) = queue();
3351        let mut t = task("keep me");
3352        t.hold_manual(Some("waiting".into()));
3353        t.runs.push("earlier-run".into());
3354        t.attachments.push("reference.txt".into());
3355        let cfg: crate::config::Config =
3356            toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
3357                .unwrap();
3358        q.put(&mut t).unwrap();
3359        let original = serde_json::to_value(&t).unwrap();
3360        for n in [2, 1, 2] {
3361            t.set_implementers(n, &cfg).unwrap();
3362            q.put(&mut t).unwrap();
3363            t = q.get(&t.id).unwrap();
3364            assert_eq!(t.implementers(), Some(usize::from(n)));
3365            assert_eq!(t.solo, n == 1);
3366            if n == 1 {
3367                assert!(t.overrides.is_none());
3368            }
3369            let mut actual = serde_json::to_value(&t).unwrap();
3370            for field in ["solo", "overrides", "updated_at"] {
3371                actual[field] = original[field].clone();
3372            }
3373            assert_eq!(actual, original);
3374        }
3375        t.overrides.as_mut().unwrap().judges = Some(2);
3376        t.set_implementers(1, &cfg).unwrap();
3377        assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
3378        assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
3379    }
3380
3381    #[test]
3382    fn task_implementers_old_records_keep_inheritance_until_changed() {
3383        let mut value = serde_json::to_value(task("old task")).unwrap();
3384        value.as_object_mut().unwrap().remove("solo");
3385        value.as_object_mut().unwrap().remove("overrides");
3386        let mut old: Task = serde_json::from_value(value).unwrap();
3387        assert_eq!(old.implementers(), None);
3388        old.set_implementers(1, &crate::config::Config::default())
3389            .unwrap();
3390        assert!(old.solo);
3391        assert!(old.overrides.is_none());
3392    }
3393
3394    #[test]
3395    fn task_implementers_reject_invalid_counts_and_states() {
3396        let cfg = crate::config::Config::default();
3397        let mut t = task("unchanged");
3398        for n in [0, 4, 255] {
3399            assert!(t.set_implementers(n, &cfg).is_err());
3400        }
3401        for status in [
3402            TaskStatus::Running,
3403            TaskStatus::Done,
3404            TaskStatus::Failed,
3405            TaskStatus::Blocked,
3406            TaskStatus::Parked,
3407        ] {
3408            t.status = status;
3409            let before = serde_json::to_value(&t).unwrap();
3410            assert!(t.set_implementers(1, &cfg).is_err());
3411            assert_eq!(serde_json::to_value(&t).unwrap(), before);
3412        }
3413    }
3414
3415    /// A queue of its own, with no process-global state - which is the point of
3416    /// `Queue::at`, and why these can run in parallel.
3417    fn queue() -> (tempfile::TempDir, Queue) {
3418        let dir = tempfile::tempdir().unwrap();
3419        let q = Queue::at(dir.path().join("queue"));
3420        (dir, q)
3421    }
3422
3423    #[test]
3424    fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3425        let dir = tempfile::tempdir().unwrap();
3426        let q = Queue::at(dir.path().join("queue"));
3427        let mut t = task("held");
3428        q.put(&mut t).unwrap();
3429        assert_eq!(
3430            crate::notices::Notices::at(dir.path().join("notifications"))
3431                .list()
3432                .len(),
3433            0
3434        );
3435        t.hold_machine(Some("out of attempts".to_owned()));
3436        q.put(&mut t).unwrap();
3437        let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3438        assert_eq!(listed.len(), 1);
3439        assert!(listed[0].message.contains("out of attempts"));
3440    }
3441
3442    fn task(title: &str) -> Task {
3443        Task::new(
3444            title.to_owned(),
3445            format!("do {title}"),
3446            PathBuf::from("."),
3447            Source::Human,
3448        )
3449    }
3450
3451    #[test]
3452    fn attempt_limit_inherits_old_records_and_controls_failure() {
3453        let mut t = task("budget");
3454        let mut old = serde_json::to_value(&t).unwrap();
3455        old.as_object_mut().unwrap().remove("attempt_limit");
3456        old["schema"] = serde_json::json!(15);
3457        let legacy: Task = serde_json::from_value(old).unwrap();
3458        assert_eq!(legacy.attempt_limit, None);
3459        assert_eq!(legacy.effective_attempt_limit(2), 2);
3460        t.attempts = 2;
3461        t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3462        t.fail("retry", 1);
3463        assert_eq!(t.status, TaskStatus::Failed);
3464        t.attempts = 3;
3465        t.fail("stop", 10);
3466        assert_eq!(t.status, TaskStatus::Held);
3467        t.change_attempt_limit(None).unwrap();
3468        assert_eq!(t.status, TaskStatus::Held);
3469        assert_eq!(t.attempts, 3);
3470        t.fail("inherit", 4);
3471        assert_eq!(t.status, TaskStatus::Failed);
3472        for status in [
3473            TaskStatus::Running,
3474            TaskStatus::Done,
3475            TaskStatus::Failed,
3476            TaskStatus::Blocked,
3477            TaskStatus::Parked,
3478        ] {
3479            t.status = status;
3480            assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3481            assert_eq!(t.attempt_limit, None);
3482        }
3483    }
3484
3485    #[test]
3486    fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3487        let mut t = task("retried");
3488        assert!(
3489            t.earlier_attempts().is_empty(),
3490            "a first attempt takes nothing over"
3491        );
3492        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3493        assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3494        // What the web note shows and what a takeover acts on are one rule.
3495        assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3496        assert_eq!(t.successor_of("bbbb"), None);
3497        assert_eq!(t.successor_of("zzzz"), None);
3498    }
3499
3500    #[test]
3501    fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3502        let (_dir, q) = queue();
3503        let mut t = task("resumed");
3504        for (runs, successor) in [
3505            (vec!["aaaa", "aaaa"], None),
3506            (vec!["aaaa", "bbbb", "aaaa"], None),
3507            (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3508            (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3509        ] {
3510            t.runs = runs.into_iter().map(str::to_owned).collect();
3511            q.put(&mut t).unwrap();
3512
3513            assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3514            assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3515            assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3516        }
3517    }
3518
3519    #[test]
3520    fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3521        let (_dir, q) = queue();
3522        let mut t = task("retried");
3523        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3524        q.put(&mut t).unwrap();
3525
3526        assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3527        assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3528        assert_eq!(
3529            q.superseded_by("cccc"),
3530            None,
3531            "the latest attempt replaces nothing"
3532        );
3533        assert_eq!(
3534            q.superseded_by("never-heard-of-it"),
3535            None,
3536            "a run belonging to no task on this queue is not superseded"
3537        );
3538
3539        let mut by = HashMap::new();
3540        by.insert("aaaa".to_owned(), "bbbb".to_owned());
3541        by.insert("bbbb".to_owned(), "cccc".to_owned());
3542        assert_eq!(
3543            q.superseded(),
3544            by,
3545            "the whole-map and single-run forms must agree"
3546        );
3547    }
3548
3549    #[test]
3550    fn superseded_attempts_is_empty_until_the_task_is_done() {
3551        let mut t = task("retried");
3552        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3553        t.status = TaskStatus::Failed;
3554        assert_eq!(
3555            t.superseded_attempts(true),
3556            &[] as &[String],
3557            "a task still retrying has no attempt yet that a later one made moot"
3558        );
3559
3560        t.status = TaskStatus::Running;
3561        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3562    }
3563
3564    #[test]
3565    fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3566        let mut t = task("retried");
3567        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3568        t.status = TaskStatus::Done;
3569        assert_eq!(
3570            t.superseded_attempts(true),
3571            &["aaaa".to_owned(), "bbbb".to_owned()],
3572            "cccc is the attempt whose success made the task done, and stays out"
3573        );
3574    }
3575
3576    #[test]
3577    fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3578        let mut t = task("first try landed");
3579        t.runs = vec!["aaaa".to_owned()];
3580        t.status = TaskStatus::Done;
3581        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3582    }
3583
3584    #[test]
3585    fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3586        // `succeed` is reachable directly - `magi task done`, its web
3587        // equivalent, and the conductor's `Recovery::Done` - on a task in
3588        // any status, including one whose last recorded attempt is itself
3589        // `Blocked`/`Failed`/anything but `Merged`/`Ready`. `runs.last()`
3590        // alone cannot tell that apart from the loop's own settle path, so
3591        // the caller's own read of that run's status is what decides this.
3592        let mut t = task("closed by hand after a manual merge");
3593        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3594        t.status = TaskStatus::Done;
3595        assert_eq!(
3596            t.superseded_attempts(false),
3597            &[] as &[String],
3598            "nothing here is provably why the task is done, so nothing is superseded"
3599        );
3600    }
3601
3602    #[test]
3603    fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3604        let (_dir, q) = queue();
3605        let mut t = task("retried twice");
3606        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3607        q.put(&mut t).unwrap();
3608
3609        assert_eq!(
3610            q.latest_attempt("aaaa"),
3611            Some("cccc".to_owned()),
3612            "an old attempt points straight at the chain's current head, not the \
3613             next attempt in the middle of it"
3614        );
3615        assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3616        assert_eq!(
3617            q.latest_attempt("cccc"),
3618            None,
3619            "the latest attempt is not superseded by anything"
3620        );
3621        assert_eq!(
3622            q.latest_attempt("never-heard-of-it"),
3623            None,
3624            "a run belonging to no task on this queue is not superseded"
3625        );
3626    }
3627
3628    #[test]
3629    fn a_markdown_heading_is_the_title_not_decoration() {
3630        // A task file's heading is the summary its author already wrote, so it
3631        // beats the prose underneath. Getting this backwards was visible in the
3632        // first smoke test: a task titled "# Rework the config loader" listed
3633        // as "It re-reads the file on every lookup".
3634        assert_eq!(
3635            title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3636            "Rework the config loader"
3637        );
3638        assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3639        assert_eq!(title_from("> quoted task", 40), "quoted task");
3640        // Nothing usable at all still has to produce something printable.
3641        assert_eq!(title_from("   \n\n", 40), "(empty task)");
3642        assert_eq!(title_from("###\n", 40), "(empty task)");
3643    }
3644
3645    #[test]
3646    fn a_long_title_is_elided_by_characters_not_bytes() {
3647        // Byte truncation would split a multi-byte character and panic.
3648        let long = "課題".repeat(30);
3649        let title = title_from(&long, 10);
3650        assert_eq!(title.chars().count(), 10);
3651        assert!(title.ends_with('…'));
3652    }
3653
3654    #[test]
3655    fn priority_wins_and_ties_break_oldest_first() {
3656        let (_dir, q) = queue();
3657        let mut a = task("first");
3658        let mut b = task("second");
3659        let mut c = task("urgent");
3660        // Ids carry a timestamp, so force a known order.
3661        a.id = "20260101-000001-aaaa".to_owned();
3662        b.id = "20260101-000002-bbbb".to_owned();
3663        c.id = "20260101-000003-cccc".to_owned();
3664        c.priority = 5;
3665        for t in [&mut a, &mut b, &mut c] {
3666            q.put(t).unwrap();
3667        }
3668
3669        // Priority first...
3670        assert_eq!(q.next_runnable().unwrap().id, c.id);
3671        c.hold_machine(None);
3672        q.put(&mut c).unwrap();
3673        // ...then oldest, so a burst of new work cannot starve older work.
3674        assert_eq!(q.next_runnable().unwrap().id, a.id);
3675        assert_eq!(q.list().len(), 3, "b is still waiting its turn");
3676    }
3677
3678    #[test]
3679    fn a_blocked_task_never_starves_another_runnable_one() {
3680        let (_dir, q) = queue();
3681        let mut blocked = task("blocked");
3682        blocked.block(vec!["something".to_owned()], None);
3683        q.put(&mut blocked).unwrap();
3684
3685        let mut runnable = task("free to go");
3686        q.put(&mut runnable).unwrap();
3687
3688        let next = q.next_runnable().expect("a runnable task is still offered");
3689        assert_eq!(next.id, runnable.id);
3690    }
3691
3692    #[test]
3693    fn a_held_task_is_never_offered_to_the_loop() {
3694        let (_dir, q) = queue();
3695        let mut t = task("held");
3696        q.put(&mut t).unwrap();
3697        assert!(q.next_runnable().is_some());
3698
3699        t.hold_machine(None);
3700        q.put(&mut t).unwrap();
3701        assert!(
3702            q.next_runnable().is_none(),
3703            "a held task must wait for a human"
3704        );
3705
3706        // A failed task, by contrast, is exactly what the loop should retry.
3707        t.status = TaskStatus::Failed;
3708        q.put(&mut t).unwrap();
3709        assert!(q.next_runnable().is_some());
3710    }
3711
3712    #[test]
3713    fn attempts_are_capped_and_then_the_task_is_held() {
3714        let mut t = task("doomed");
3715
3716        t.start("run-1".to_owned());
3717        t.fail("gate red", 2);
3718        assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
3719
3720        t.start("run-2".to_owned());
3721        t.fail("gate red", 2);
3722        assert_eq!(
3723            t.status,
3724            TaskStatus::Held,
3725            "out of attempts: stop spending money on it"
3726        );
3727        assert_eq!(t.runs, ["run-1", "run-2"]);
3728        assert_eq!(t.last_error.as_deref(), Some("gate red"));
3729        assert_eq!(
3730            t.hold_reason.as_deref(),
3731            Some("gate red"),
3732            "the hold must say why, not leave hold_reason null next to a \
3733             populated last_error"
3734        );
3735    }
3736
3737    #[test]
3738    fn handing_off_a_task_records_a_hold_reason_too() {
3739        let mut t = task("left a pull request");
3740        t.start("run-1".to_owned());
3741        t.handed_off("run ended with a pull request open [run run-1]");
3742        assert_eq!(t.status, TaskStatus::Held);
3743        assert_eq!(t.hold_source, Some(HoldSource::Machine));
3744        assert_eq!(
3745            t.hold_reason.as_deref(),
3746            Some("run ended with a pull request open [run run-1]")
3747        );
3748        assert_eq!(t.hold_reason, t.last_error);
3749    }
3750
3751    #[test]
3752    fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
3753        let mut t = task("stalled by quota");
3754
3755        t.start("run-1".to_owned());
3756        assert_eq!(t.attempts, 1);
3757        t.stall("judge-1, judge-2 out of quota");
3758        assert_eq!(
3759            t.attempts, 0,
3760            "a closed quota window must not spend the task's retry budget"
3761        );
3762        assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
3763        assert_eq!(
3764            t.last_error.as_deref(),
3765            Some("judge-1, judge-2 out of quota")
3766        );
3767
3768        // A task can therefore stall all night and still get its real attempts
3769        // once the quota resets - which is the whole point.
3770        for _ in 0..20 {
3771            t.start("run-n".to_owned());
3772            t.stall("still out of quota");
3773        }
3774        t.start("run-real".to_owned());
3775        t.fail("gate red", 2);
3776        assert_eq!(
3777            t.status,
3778            TaskStatus::Failed,
3779            "the first attempt that was really judged is attempt one"
3780        );
3781    }
3782
3783    #[test]
3784    fn releasing_a_held_task_gives_it_a_real_second_chance() {
3785        let mut t = task("retry me");
3786        t.start("run-1".to_owned());
3787        t.fail("gate red", 1);
3788        assert_eq!(t.status, TaskStatus::Held);
3789
3790        t.release();
3791        assert_eq!(t.status, TaskStatus::Queued);
3792        // Without resetting attempts the next failure would re-hold at once,
3793        // and a release would be a no-op the operator cannot see.
3794        assert_eq!(t.attempts, 0);
3795        assert!(t.last_error.is_none());
3796        assert_eq!(
3797            t.runs.len(),
3798            1,
3799            "history is kept: attempts reset, evidence does not"
3800        );
3801    }
3802
3803    #[test]
3804    fn a_hold_reason_survives_and_a_release_clears_it() {
3805        let mut t = task("waiting on something else");
3806        t.hold_manual(Some(
3807            "waiting for 20260101-000000-aaaa to land first".to_owned(),
3808        ));
3809        assert_eq!(t.status, TaskStatus::Held);
3810        assert_eq!(
3811            t.hold_reason.as_deref(),
3812            Some("waiting for 20260101-000000-aaaa to land first")
3813        );
3814
3815        // Holding again with no reason must not erase the one already there.
3816        t.hold_manual(None);
3817        assert_eq!(
3818            t.hold_reason.as_deref(),
3819            Some("waiting for 20260101-000000-aaaa to land first"),
3820            "a bare re-hold keeps whatever a human already wrote down"
3821        );
3822
3823        // A hold with no reason at all is still an ordinary, allowed hold.
3824        let mut plain = task("no reason given");
3825        plain.hold_manual(None);
3826        assert_eq!(plain.status, TaskStatus::Held);
3827        assert!(plain.hold_reason.is_none());
3828
3829        t.release();
3830        assert_eq!(t.status, TaskStatus::Queued);
3831        assert!(
3832            t.hold_reason.is_none(),
3833            "a stale reason must not greet the next person who holds this task"
3834        );
3835    }
3836
3837    #[test]
3838    fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
3839        // `done` can close a held task directly - neither `magi task done`
3840        // nor `POST /api/queue/{id}/done` requires a release first - so a
3841        // task held for "waiting on 3ed9" and then closed without ever being
3842        // released must not still read as waiting on it afterwards.
3843        let mut t = task("landed by hand while held");
3844        t.hold_manual(Some("waiting on 3ed9".to_owned()));
3845        assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
3846
3847        t.succeed();
3848        assert_eq!(t.status, TaskStatus::Done);
3849        assert!(
3850            t.hold_reason.is_none(),
3851            "a done task cannot still be waiting on something"
3852        );
3853    }
3854
3855    #[test]
3856    fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
3857        // The web UI's "Hold" and "Mark done" buttons are both reachable on a
3858        // `blocked` task, not just on `queued`/`held` ones - neither requires
3859        // a release first. A task moved off `Blocked` that way must not still
3860        // carry the dependency it was waiting on: a dependency graph built
3861        // from `blocked_by` would otherwise keep drawing an edge for a task
3862        // that is not blocked on anything any more.
3863        let mut held = task("held straight out of blocked");
3864        held.block(
3865            vec!["20260101-000000-dead".to_owned()],
3866            Some("waiting on the migration script".to_owned()),
3867        );
3868        assert_eq!(held.status, TaskStatus::Blocked);
3869
3870        held.hold_manual(None);
3871        assert_eq!(held.status, TaskStatus::Held);
3872        assert!(
3873            held.blocked_by.is_empty(),
3874            "hold overrides the wait, same as release"
3875        );
3876        assert!(held.block_reason.is_none());
3877
3878        let mut done = task("closed straight out of blocked");
3879        done.block(
3880            vec!["20260101-000000-dead".to_owned()],
3881            Some("waiting on the migration script".to_owned()),
3882        );
3883        done.succeed();
3884        assert_eq!(done.status, TaskStatus::Done);
3885        assert!(
3886            done.blocked_by.is_empty(),
3887            "a done task cannot still be waiting on a dependency"
3888        );
3889        assert!(done.block_reason.is_none());
3890    }
3891
3892    #[test]
3893    fn a_blocked_task_is_never_offered_to_the_loop() {
3894        let mut t = task("blocked");
3895        assert!(t.status.runnable());
3896        t.block(
3897            vec!["dep-id".to_owned()],
3898            Some("waits on dep-id".to_owned()),
3899        );
3900        assert_eq!(t.status, TaskStatus::Blocked);
3901        assert!(!t.status.runnable());
3902        assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
3903    }
3904
3905    #[test]
3906    fn unblocking_the_last_dependency_returns_the_task_to_queued() {
3907        let mut t = task("blocked on two");
3908        t.block(
3909            vec!["a".to_owned(), "b".to_owned()],
3910            Some("waits on a and b".to_owned()),
3911        );
3912
3913        t.unblock("a");
3914        assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
3915        assert_eq!(t.blocked_by, ["b"]);
3916
3917        t.unblock("b");
3918        assert_eq!(t.status, TaskStatus::Queued);
3919        assert!(t.blocked_by.is_empty());
3920        assert!(t.block_reason.is_none());
3921    }
3922
3923    #[test]
3924    fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
3925        let mut t = task("never blocked");
3926        t.unblock("whatever");
3927        assert_eq!(t.status, TaskStatus::Queued);
3928    }
3929
3930    #[test]
3931    fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
3932        // The bug this guards: a task an operator (or `crate::triage`) has
3933        // deliberately held, once `crate::conduct` blocks it on a follow-up
3934        // question, must not silently re-enter the competition queue the
3935        // moment that question is answered - whatever the answer said.
3936        let mut t = task("held, then asked about");
3937        t.hold_machine(Some("out of attempts".to_owned()));
3938        assert_eq!(t.status, TaskStatus::Held);
3939
3940        t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3941        assert_eq!(t.status, TaskStatus::Blocked);
3942
3943        t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3944        t.unblock("q1");
3945        assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3946        assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3947        assert_eq!(t.hold_source, Some(HoldSource::Machine));
3948        assert!(t.blocked_from.is_none(), "consumed once restored");
3949    }
3950
3951    #[test]
3952    fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3953        let mut t = task("manually held, then asked about");
3954        t.hold_manual(Some("waiting on a dependency".to_owned()));
3955
3956        t.block(vec!["q1".to_owned()], None);
3957        t.unblock("q1");
3958
3959        assert_eq!(t.status, TaskStatus::Held);
3960        assert_eq!(t.hold_source, Some(HoldSource::Manual));
3961    }
3962
3963    #[test]
3964    fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3965        // A second `Task::block` call - `crate::conduct` adding a question on
3966        // top of an existing block - must not overwrite `blocked_from` with
3967        // `Blocked` itself, or the task would restore into itself.
3968        let mut t = task("held, blocked twice");
3969        t.hold_machine(None);
3970        t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3971        t.block(
3972            vec!["q1".to_owned(), "q2".to_owned()],
3973            Some("second".to_owned()),
3974        );
3975
3976        t.unblock("q1");
3977        assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3978        t.unblock("q2");
3979        assert_eq!(t.status, TaskStatus::Held);
3980    }
3981
3982    #[test]
3983    fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3984        // Whatever process was driving the run is gone by the time a
3985        // conductor's question about it gets answered - there is nothing left
3986        // to resume into.
3987        let mut t = task("blocked mid-run");
3988        t.start("run-1".to_owned());
3989        assert_eq!(t.status, TaskStatus::Running);
3990
3991        t.block(vec!["q1".to_owned()], None);
3992        t.unblock("q1");
3993        assert_eq!(t.status, TaskStatus::Queued);
3994    }
3995
3996    #[test]
3997    fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3998        // `blocked_from` is `None` for a record written before schema 4 (or,
3999        // equivalently, deserialized straight from an on-disk file that never
4000        // had the field). Held evidence surviving on the task - never cleared
4001        // by `block` - is the only way left to tell such a record apart from
4002        // one blocked straight out of `Queued`.
4003        let mut t = task("legacy record, held before it was blocked");
4004        t.hold_source = Some(HoldSource::Machine);
4005        t.hold_reason = Some("legacy hold reason".to_owned());
4006        t.status = TaskStatus::Blocked;
4007        t.blocked_by = vec!["q1".to_owned()];
4008        t.blocked_from = None;
4009
4010        t.unblock("q1");
4011        assert_eq!(t.status, TaskStatus::Held);
4012    }
4013
4014    #[test]
4015    fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
4016        let mut t = task("legacy record, ordinary dependency block");
4017        t.status = TaskStatus::Blocked;
4018        t.blocked_by = vec!["dep".to_owned()];
4019        t.blocked_from = None;
4020
4021        t.unblock("dep");
4022        assert_eq!(t.status, TaskStatus::Queued);
4023    }
4024
4025    #[test]
4026    fn answering_a_question_is_recorded_and_survives_a_release() {
4027        let mut t = task("asked something");
4028        t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
4029        t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
4030        t.unblock("q1");
4031        assert_eq!(t.status, TaskStatus::Queued);
4032        assert_eq!(t.answers.len(), 1);
4033        assert_eq!(t.answers[0].answer, "SQLite");
4034
4035        // A release resets attempts, not evidence - the same rule
4036        // `releasing_a_held_task_gives_it_a_real_second_chance` asserts for
4037        // `runs`.
4038        t.release();
4039        assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
4040    }
4041
4042    #[test]
4043    fn resumption_review_branch_requires_unchanged_authored_criteria() {
4044        for source in ["cli", "web"] {
4045            let mut t = task("review old candidate");
4046            t.request_review("candidate-A".into());
4047            t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
4048            let mut unchanged = t.clone();
4049            unchanged.release_operator(source, None);
4050            assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
4051            t.edit(
4052                "review approved head".into(),
4053                "Reuse measured PR head B; fix Clippy; review".into(),
4054            )
4055            .unwrap();
4056            t.release_operator(source, None);
4057            assert!(t.review_branch.is_none());
4058            assert!(t.review_instruction.is_none());
4059            // Legacy records capture their old criteria at the supported edit.
4060            let mut legacy = unchanged;
4061            legacy.review_instruction = None;
4062            legacy.hold_for_handover(None, "still checked out".into());
4063            legacy
4064                .edit("new plan".into(), "Reuse measured PR head B".into())
4065                .unwrap();
4066            legacy.release_operator(source, None);
4067            assert!(legacy.review_branch.is_none());
4068        }
4069    }
4070
4071    #[test]
4072    fn a_refused_handover_keeps_the_review_branch_across_release() {
4073        let mut t = task("refused takeover");
4074        t.start("run-1".to_owned());
4075        t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
4076        assert_eq!(t.status, TaskStatus::Held);
4077        assert_eq!(t.attempts, 1);
4078        t.release();
4079        assert_eq!(t.status, TaskStatus::Queued);
4080        assert_eq!(t.attempts, 0);
4081        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4082
4083        // A fresh competition chosen afterwards is not a review.
4084        t.hold_for_handover(None, "again".to_owned());
4085        t.requeue();
4086        assert!(t.review_branch.is_none());
4087
4088        // A manual hold never keeps one.
4089        let mut m = task("manual");
4090        m.review_branch = Some("magi/x/A".to_owned());
4091        m.hold_manual(None);
4092        m.release();
4093        assert!(m.review_branch.is_none());
4094    }
4095
4096    #[test]
4097    fn requesting_review_requeues_the_task_and_remembers_the_branch() {
4098        let mut t = task("blocked run with a surviving branch");
4099        t.start("run-1".to_owned());
4100        t.fail("blocked with major findings", 5);
4101        assert_eq!(t.status, TaskStatus::Failed);
4102
4103        t.request_review("magi/eba2/A".to_owned());
4104        assert_eq!(t.status, TaskStatus::Queued);
4105        assert_eq!(t.attempts, 0);
4106        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4107
4108        // An ordinary release (a human overriding the choice) drops it again.
4109        t.release();
4110        assert!(t.review_branch.is_none());
4111    }
4112
4113    #[test]
4114    fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
4115        let mut t = task("retry");
4116        t.start("run-1".to_owned());
4117        t.requeue();
4118        assert!(t.fresh_start);
4119
4120        t.release();
4121        assert!(!t.fresh_start);
4122    }
4123
4124    #[test]
4125    fn urgent_only_changes_queued_or_held_and_preserves_settings() {
4126        for status in [
4127            TaskStatus::Queued,
4128            TaskStatus::Held,
4129            TaskStatus::Running,
4130            TaskStatus::Done,
4131            TaskStatus::Failed,
4132            TaskStatus::Blocked,
4133            TaskStatus::Parked,
4134        ] {
4135            let mut t = task("urgent settings");
4136            t.status = status;
4137            t.attempts = 2;
4138            t.hold_reason = Some("waiting".into());
4139            t.priority = 7;
4140            t.interrupt = true;
4141            let before = serde_json::to_value(&t).unwrap();
4142            for urgent in [true, true, false, false] {
4143                let result = t.set_urgent(urgent);
4144                let mut expected = before.clone();
4145                if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
4146                    result.unwrap();
4147                    expected["urgent"] = urgent.into();
4148                } else {
4149                    let error = result.unwrap_err().to_string();
4150                    assert!(error.contains(status.as_str()), "{error}");
4151                    assert!(error.contains("queued or held"), "{error}");
4152                }
4153                assert_eq!(serde_json::to_value(&t).unwrap(), expected);
4154            }
4155        }
4156    }
4157
4158    #[test]
4159    fn priority_can_be_changed_while_queued_but_not_while_running() {
4160        let mut t = task("reprioritise me");
4161        t.set_priority(5).unwrap();
4162        assert_eq!(t.priority, 5);
4163
4164        t.start("run-1".to_owned());
4165        let err = t.set_priority(9).unwrap_err().to_string();
4166        assert!(err.contains("running"), "{err}");
4167        assert_eq!(t.priority, 5, "the rejected write must not partially apply");
4168    }
4169
4170    #[test]
4171    fn interrupt_can_be_marked_while_queued_but_not_while_running() {
4172        let mut t = task("interrupt me");
4173        assert!(!t.interrupt, "off unless asked, same as any other task");
4174
4175        t.set_interrupt(true).unwrap();
4176        assert!(t.interrupt);
4177
4178        t.start("run-1".to_owned());
4179        assert!(
4180            !t.interrupt,
4181            "the mark is one-shot: dispatching the task fulfils it, \
4182             whatever the run that follows ends up doing"
4183        );
4184        let err = t.set_interrupt(true).unwrap_err().to_string();
4185        assert!(err.contains("running"), "{err}");
4186        // Clearing is always allowed, even on a running task - there is
4187        // nothing left for it to interrupt once it has been claimed.
4188        t.set_interrupt(false).unwrap();
4189        assert!(!t.interrupt);
4190    }
4191
4192    /// R2-1-1: a task whose run fails and requeues must not go on
4193    /// re-triggering `crate::daemon`'s interrupt scheduler on every later
4194    /// boundary, attempt after attempt, until it exhausts its budget.
4195    #[test]
4196    fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
4197        let mut t = task("interrupt me");
4198        t.set_interrupt(true).unwrap();
4199        t.start("run-1".to_owned());
4200        t.fail("mock failure", 5);
4201        assert_eq!(t.status, TaskStatus::Failed);
4202        assert!(
4203            !t.interrupt,
4204            "one attempt already spent the mark; a retry is an ordinary \
4205             requeue, not a fresh interrupt request"
4206        );
4207    }
4208
4209    #[test]
4210    fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
4211        let (_dir, q) = queue();
4212        let mut a = task("first filed");
4213        let mut b = task("second filed");
4214        a.id = "20260101-000001-aaaa".to_owned();
4215        b.id = "20260101-000002-bbbb".to_owned();
4216        q.put(&mut a).unwrap();
4217        q.put(&mut b).unwrap();
4218
4219        assert_eq!(
4220            q.next_runnable().unwrap().id,
4221            a.id,
4222            "with equal priority the older task goes first, so a burst of \
4223             new work cannot starve it"
4224        );
4225        assert_eq!(
4226            q.list()[0].id,
4227            b.id,
4228            "but the list an operator reads is newest first, the same as \
4229             before priority existed - a's turn to run does not make it the \
4230             newest task"
4231        );
4232
4233        let mut a = q.get(&a.id).unwrap();
4234        a.set_priority(10).unwrap();
4235        q.put(&mut a).unwrap();
4236
4237        assert_eq!(
4238            q.next_runnable().unwrap().id,
4239            a.id,
4240            "a raised priority must be reflected the moment it is saved"
4241        );
4242        // `magi task list` and `GET /api/queue` both print `Queue::list()`
4243        // directly, so the raised task has to lead there too - not only in
4244        // what the loop would claim next.
4245        assert_eq!(
4246            q.list()[0].id,
4247            a.id,
4248            "the raised task must sort first in the list an operator reads, \
4249             not only in next_runnable's own ordering"
4250        );
4251    }
4252
4253    #[test]
4254    fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
4255        let mut t = Task::new(
4256            "old title".to_owned(),
4257            "old instruction".to_owned(),
4258            PathBuf::from("/repo"),
4259            Source::Agent {
4260                run: "20260101-000000-beef".to_owned(),
4261                node: "implement".to_owned(),
4262            },
4263        );
4264        let id = t.id.clone();
4265        let created_at = t.created_at;
4266        t.runs.push("20260101-000000-beef".to_owned());
4267
4268        t.edit("new title".to_owned(), "new instruction".to_owned())
4269            .unwrap();
4270
4271        assert_eq!(t.title, "new title");
4272        assert_eq!(t.instruction, "new instruction");
4273        assert_eq!(t.id, id, "editing must not mint a new id");
4274        assert_eq!(t.created_at, created_at);
4275        assert_eq!(
4276            t.source,
4277            Source::Agent {
4278                run: "20260101-000000-beef".to_owned(),
4279                node: "implement".to_owned(),
4280            },
4281            "editing must not turn agent attribution into human"
4282        );
4283        assert_eq!(t.runs, ["20260101-000000-beef"]);
4284    }
4285
4286    #[test]
4287    fn editing_is_refused_once_a_task_is_running_or_finished() {
4288        let mut running = task("in flight");
4289        running.start("run-1".to_owned());
4290        let err = running
4291            .edit("x".to_owned(), "y".to_owned())
4292            .unwrap_err()
4293            .to_string();
4294        assert!(err.contains("running"), "{err}");
4295
4296        let mut done = task("finished");
4297        done.succeed();
4298        let err = done
4299            .edit("x".to_owned(), "y".to_owned())
4300            .unwrap_err()
4301            .to_string();
4302        assert!(err.contains("done"), "{err}");
4303
4304        // Both queued and held are the point of the feature and must work.
4305        let mut queued = task("waiting");
4306        queued.edit("x".to_owned(), "y".to_owned()).unwrap();
4307        let mut held = task("parked");
4308        held.hold_machine(None);
4309        held.edit("x".to_owned(), "y".to_owned()).unwrap();
4310    }
4311
4312    #[test]
4313    fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
4314        let (_dir, q) = queue();
4315        let path = q.path_of("20260101-000000-aaaa");
4316        std::fs::create_dir_all(q.root()).unwrap();
4317        std::fs::write(
4318            &path,
4319            serde_json::json!({
4320                "schema": SCHEMA,
4321                "id": "20260101-000000-aaaa",
4322                "title": "from before hold reasons existed",
4323                "instruction": "from before hold reasons existed",
4324                "repo": ".",
4325                "source": { "kind": "human" },
4326                "status": "held",
4327                "created_at": Timestamp::now().to_string(),
4328                "updated_at": Timestamp::now().to_string(),
4329            })
4330            .to_string(),
4331        )
4332        .unwrap();
4333
4334        let task = q.get("20260101-000000-aaaa").expect("must still read");
4335        assert!(task.hold_reason.is_none());
4336        assert!(task.operator_held());
4337    }
4338
4339    #[test]
4340    fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
4341        let (_dir, q) = queue();
4342        let path = q.path_of("20260101-000000-bbbb");
4343        std::fs::create_dir_all(q.root()).unwrap();
4344        std::fs::write(
4345            &path,
4346            serde_json::json!({
4347                "schema": 2,
4348                "id": "20260101-000000-bbbb",
4349                "title": "old manual recovery",
4350                "instruction": "old manual recovery",
4351                "repo": ".",
4352                "source": { "kind": "human" },
4353                "status": "held",
4354                "hold_reason": "active manual recovery run20260912-224242-daf5",
4355                "created_at": Timestamp::now().to_string(),
4356                "updated_at": Timestamp::now().to_string(),
4357            })
4358            .to_string(),
4359        )
4360        .unwrap();
4361
4362        let task = q.get("20260101-000000-bbbb").expect("must still read");
4363        assert_eq!(task.hold_source, None);
4364        assert!(task.operator_held());
4365    }
4366
4367    #[test]
4368    fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
4369        let (_dir, q) = queue();
4370        let path = q.path_of("20260101-000000-aaaa");
4371        std::fs::create_dir_all(q.root()).unwrap();
4372        std::fs::write(
4373            &path,
4374            serde_json::json!({
4375                "schema": SCHEMA,
4376                "id": "20260101-000000-aaaa",
4377                "title": "from before diagnostics existed",
4378                "instruction": "from before diagnostics existed",
4379                "repo": ".",
4380                "source": { "kind": "human" },
4381                "status": "held",
4382                "created_at": Timestamp::now().to_string(),
4383                "updated_at": Timestamp::now().to_string(),
4384            })
4385            .to_string(),
4386        )
4387        .unwrap();
4388
4389        let task = q.get("20260101-000000-aaaa").expect("must still read");
4390        assert!(task.diagnostic.is_none());
4391    }
4392
4393    #[test]
4394    fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4395        // Written by a build that predates `blocked_by`, `block_reason`,
4396        // `answers` and `review_branch` entirely - literal `"schema": 1`,
4397        // not `SCHEMA`, since the whole point is a build older than this one.
4398        let (_dir, q) = queue();
4399        let path = q.path_of("20260101-000000-aaaa");
4400        std::fs::create_dir_all(q.root()).unwrap();
4401        std::fs::write(
4402            &path,
4403            serde_json::json!({
4404                "schema": 1,
4405                "id": "20260101-000000-aaaa",
4406                "title": "from before blocking existed",
4407                "instruction": "from before blocking existed",
4408                "repo": ".",
4409                "source": { "kind": "human" },
4410                "status": "queued",
4411                "created_at": Timestamp::now().to_string(),
4412                "updated_at": Timestamp::now().to_string(),
4413            })
4414            .to_string(),
4415        )
4416        .unwrap();
4417
4418        let task = q.get("20260101-000000-aaaa").expect("must still read");
4419        assert!(task.blocked_by.is_empty());
4420        assert!(task.block_reason.is_none());
4421        assert!(task.answers.is_empty());
4422        assert!(task.review_branch.is_none());
4423    }
4424
4425    #[test]
4426    fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4427        // A diagnostic belongs to the run that produced it. Left in place
4428        // across a release, an unrelated later failure - a config error, say -
4429        // would go on showing evidence for a problem that is no longer why the
4430        // task is stuck.
4431        let mut held = task("diagnosed");
4432        held.start("run-1".to_owned());
4433        held.fail("gate red", 1);
4434        held.diagnostic = Some("cargo test failed: ...".to_owned());
4435        assert_eq!(held.status, TaskStatus::Held);
4436
4437        held.release();
4438        assert!(held.diagnostic.is_none());
4439
4440        held.diagnostic = Some("cargo test failed: ...".to_owned());
4441        held.succeed();
4442        assert!(held.diagnostic.is_none());
4443    }
4444
4445    #[test]
4446    fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4447        let mut t = task("retried");
4448        t.start("run-1".to_owned());
4449        t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4450        t.fail("unrelated config error", 5);
4451        assert_eq!(t.status, TaskStatus::Failed);
4452        assert!(
4453            t.diagnostic.is_none(),
4454            "fail() must not let an old diagnostic outlive the run that produced it"
4455        );
4456    }
4457
4458    #[test]
4459    fn a_claim_is_exclusive_and_releases_on_drop() {
4460        let (_dir, q) = queue();
4461        let mut t = task("contended");
4462        q.put(&mut t).unwrap();
4463
4464        let held = q.claim(&t.id).unwrap();
4465        assert!(
4466            q.claim(&t.id).is_err(),
4467            "two daemons must not drive one task into two runs"
4468        );
4469        drop(held);
4470        assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4471    }
4472
4473    #[test]
4474    fn a_round_trip_survives_disk() {
4475        let (_dir, q) = queue();
4476        let mut t = Task::new(
4477            "titled".to_owned(),
4478            "body".to_owned(),
4479            PathBuf::from("/repo"),
4480            Source::Agent {
4481                run: "20260101-000000-beef".to_owned(),
4482                node: "implement".to_owned(),
4483            },
4484        );
4485        t.priority = 3;
4486        q.put(&mut t).unwrap();
4487
4488        let back = q.get(&t.id).unwrap();
4489        assert_eq!(back.id, t.id);
4490        assert_eq!(back.priority, 3);
4491        assert_eq!(back.source.label(), "implement@beef");
4492        // A prefix is enough, the way run ids work everywhere else.
4493        assert_eq!(q.get(t.short()).unwrap().id, t.id);
4494    }
4495
4496    #[test]
4497    fn an_unreadable_task_does_not_take_the_queue_down() {
4498        let (_dir, q) = queue();
4499        let mut t = task("fine");
4500        q.put(&mut t).unwrap();
4501        std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4502
4503        let listed = q.list();
4504        assert_eq!(listed.len(), 1, "the readable task still lists");
4505        assert_eq!(listed[0].id, t.id);
4506    }
4507
4508    #[test]
4509    fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4510        let (_dir, q) = queue();
4511        let path = q.path_of("20260101-000000-aaaa");
4512        std::fs::create_dir_all(q.root()).unwrap();
4513        std::fs::write(
4514            &path,
4515            serde_json::json!({
4516                "schema": SCHEMA,
4517                "id": "20260101-000000-aaaa",
4518                "title": "from before solo existed",
4519                "instruction": "from before solo existed",
4520                "repo": ".",
4521                "source": { "kind": "human" },
4522                "status": "queued",
4523                "created_at": Timestamp::now().to_string(),
4524                "updated_at": Timestamp::now().to_string(),
4525            })
4526            .to_string(),
4527        )
4528        .unwrap();
4529
4530        let task = q.get("20260101-000000-aaaa").expect("must still read");
4531        assert!(!task.solo, "a queue file with no `solo` field means false");
4532    }
4533
4534    #[test]
4535    fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4536        let (_dir, q) = queue();
4537        let path = q.path_of("20260101-000000-bbbb");
4538        std::fs::create_dir_all(q.root()).unwrap();
4539        std::fs::write(
4540            &path,
4541            serde_json::json!({
4542                "schema": SCHEMA,
4543                "id": "20260101-000000-bbbb",
4544                "title": "from before urgent existed",
4545                "instruction": "from before urgent existed",
4546                "repo": ".",
4547                "source": { "kind": "human" },
4548                "status": "queued",
4549                "created_at": Timestamp::now().to_string(),
4550                "updated_at": Timestamp::now().to_string(),
4551            })
4552            .to_string(),
4553        )
4554        .unwrap();
4555
4556        let task = q.get("20260101-000000-bbbb").expect("must still read");
4557        assert!(
4558            !task.urgent,
4559            "a queue file with no `urgent` field means false, same as `solo`"
4560        );
4561    }
4562
4563    #[test]
4564    fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4565        let (_dir, q) = queue();
4566        let mut t = task("from the future");
4567        q.put(&mut t).unwrap();
4568        let path = q.path_of(&t.id);
4569        let body = std::fs::read_to_string(&path)
4570            .unwrap()
4571            .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4572        std::fs::write(&path, body).unwrap();
4573
4574        let err = q.get(&t.id).unwrap_err().to_string();
4575        assert!(err.contains("schema 99"), "{err}");
4576    }
4577
4578    #[test]
4579    fn revision_moves_when_the_queue_changes() {
4580        let (_dir, q) = queue();
4581        assert_eq!(q.revision(), 0, "an empty queue has no revision");
4582        let mut t = task("first");
4583        q.put(&mut t).unwrap();
4584        assert!(q.revision() > 0, "a written task moves the revision");
4585    }
4586
4587    #[test]
4588    fn revision_moves_when_deleting_an_older_task() {
4589        let (dir, q) = queue();
4590        let questions = Questions::at(dir.path().join("questions"));
4591        let mut t1 = task("older");
4592        q.put(&mut t1).unwrap();
4593        // Ensure mtime ticks forward.
4594        std::thread::sleep(std::time::Duration::from_millis(10));
4595        let mut t2 = task("newer");
4596        q.put(&mut t2).unwrap();
4597
4598        let rev_before = q.revision();
4599        q.remove(&t1.id, false, &questions).unwrap();
4600        let rev_after = q.revision();
4601
4602        assert_ne!(
4603            rev_before, rev_after,
4604            "deleting an older task must change the revision so other clients see the deletion"
4605        );
4606    }
4607
4608    #[test]
4609    fn removing_a_task_takes_it_out_of_the_listing() {
4610        let (dir, q) = queue();
4611        let questions = Questions::at(dir.path().join("questions"));
4612        let mut t = task("delete me");
4613        q.put(&mut t).unwrap();
4614        let removed = q.remove(t.short(), false, &questions).unwrap();
4615        assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4616        assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4617        assert!(q.list().is_empty());
4618        assert!(
4619            q.remove(&t.id, false, &questions).is_err(),
4620            "removing twice is an error"
4621        );
4622    }
4623
4624    #[test]
4625    fn removing_a_task_takes_its_stale_lock_with_it() {
4626        let (dir, q) = queue();
4627        let questions = Questions::at(dir.path().join("questions"));
4628        let mut t = task("interrupted");
4629        q.put(&mut t).unwrap();
4630
4631        // A daemon killed mid-run leaves this behind. Nothing holds it: the
4632        // process that would have dropped the guard is gone.
4633        let claim = q.claim(&t.id).unwrap();
4634        std::mem::forget(claim);
4635        assert!(
4636            q.claim(&t.id).is_err(),
4637            "the orphaned lock is what makes the task look claimed"
4638        );
4639
4640        // A live daemon on this task is refused, whatever the lock says.
4641        let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4642        assert!(err.contains("live daemon"), "{err}");
4643        assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4644
4645        // With no daemon behind it, the lock is stale and goes with the task.
4646        q.remove(&t.id, false, &questions).unwrap();
4647        assert!(q.list().is_empty());
4648        let mut again = task("interrupted");
4649        again.id = t.id.clone();
4650        q.put(&mut again).unwrap();
4651        assert!(
4652            q.claim(&t.id).is_ok(),
4653            "a task that comes back must be claimable, which a left-behind lock would prevent"
4654        );
4655    }
4656
4657    fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4658        crate::notices::Notices::at(dir.join("notifications")).list()
4659    }
4660
4661    #[test]
4662    fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4663        let (dir, q) = queue();
4664        let questions = Questions::at(dir.path().join("questions"));
4665        let mut dep = task("dependency");
4666        q.put(&mut dep).unwrap();
4667        let mut blocked = task("waiting");
4668        blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
4669        q.put(&mut blocked).unwrap();
4670
4671        let removed = q.remove(&dep.id, false, &questions).unwrap();
4672        assert_eq!(removed.released, [blocked.id.clone()]);
4673        assert!(removed.still_blocked.is_empty());
4674
4675        let after = q.get(&blocked.id).unwrap();
4676        assert_eq!(after.status, TaskStatus::Queued);
4677        assert!(after.blocked_by.is_empty());
4678        assert!(after.block_reason.is_none());
4679        let notes = notices_of(dir.path());
4680        assert_eq!(notes.len(), 1, "{notes:?}");
4681        assert_eq!(notes[0].severity, crate::notices::Severity::Info);
4682        assert_eq!(
4683            notes[0].link,
4684            Some(crate::notices::Link::Task {
4685                id: blocked.id.clone(),
4686            })
4687        );
4688    }
4689
4690    #[test]
4691    fn removing_one_of_two_dependencies_keeps_the_other() {
4692        let (dir, q) = queue();
4693        let questions = Questions::at(dir.path().join("questions"));
4694        let mut dep = task("dependency");
4695        q.put(&mut dep).unwrap();
4696        let mut other = task("other");
4697        q.put(&mut other).unwrap();
4698        let mut blocked = task("waiting");
4699        blocked.block(
4700            vec![dep.id.clone(), other.id.clone()],
4701            Some("waits on both".to_owned()),
4702        );
4703        q.put(&mut blocked).unwrap();
4704
4705        let removed = q.remove(&dep.id, false, &questions).unwrap();
4706        assert!(removed.released.is_empty());
4707        assert_eq!(removed.still_blocked, [blocked.id.clone()]);
4708
4709        let after = q.get(&blocked.id).unwrap();
4710        assert_eq!(after.status, TaskStatus::Blocked);
4711        assert_eq!(after.blocked_by, [other.id.clone()]);
4712        // The tombstone stays while a task still names the id... only this
4713        // one's dependency is gone, so nothing references it and it may go.
4714        assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
4715    }
4716
4717    #[test]
4718    fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
4719        let (dir, q) = queue();
4720        let questions = Questions::at(dir.path().join("questions"));
4721        let mut dep = task("dependency");
4722        q.put(&mut dep).unwrap();
4723        let mut blocked = task("waiting");
4724        blocked.block(vec![dep.id.clone()], None);
4725        q.put(&mut blocked).unwrap();
4726
4727        // The dependent is claimed elsewhere, so `remove` has to skip it.
4728        let claim = q.claim(&blocked.id).unwrap();
4729        let removed = q.remove(&dep.id, false, &questions).unwrap();
4730        assert!(removed.released.is_empty());
4731        drop(claim);
4732
4733        let mut task = q.get(&blocked.id).unwrap();
4734        assert_eq!(task.status, TaskStatus::Blocked);
4735        assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
4736        assert_eq!(
4737            q.apply_deleted_blockers(&mut task, &questions),
4738            [dep.id.clone()]
4739        );
4740        assert_eq!(task.status, TaskStatus::Queued);
4741    }
4742
4743    #[test]
4744    fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
4745        let (dir, q) = queue();
4746        let questions = Questions::at(dir.path().join("questions"));
4747        let mut dep = task("dependency");
4748        dep.id = "20261013-193611-ca5b".into();
4749        q.put(&mut dep).unwrap();
4750        let mut waiting = task("waiting");
4751        waiting.block(vec!["ca5b".into()], None);
4752        q.put(&mut waiting).unwrap();
4753        q.remove(&dep.id, false, &questions).unwrap();
4754        assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
4755        assert_eq!(
4756            dependency(&q, &questions, "ca5b"),
4757            Dependency::Deleted(dep.id)
4758        );
4759    }
4760
4761    #[test]
4762    fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
4763        let (dir, queue) = queue();
4764        let questions = Questions::at(dir.path().join("questions"));
4765        let mut dep = task("dependency");
4766        dep.id = "20261013-193611-ca5b".into();
4767        queue.put(&mut dep).unwrap();
4768        std::fs::create_dir_all(questions.root()).unwrap();
4769        let question_id = "20261014-193611-ca5b";
4770        std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
4771        assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
4772        assert_eq!(
4773            dependency(&queue, &questions, "ca5b"),
4774            Dependency::Ambiguous
4775        );
4776        assert_eq!(
4777            dependency(&queue, &questions, &dep.id),
4778            Dependency::Task(dep.id)
4779        );
4780        assert_eq!(
4781            dependency(&queue, &questions, question_id),
4782            Dependency::Question(question_id.into())
4783        );
4784        assert_eq!(
4785            dependency(&queue, &questions, "missing"),
4786            Dependency::Missing
4787        );
4788    }
4789
4790    #[test]
4791    fn dependency_resolution_directory_errors_are_unknown_not_missing() {
4792        let dir = tempfile::tempdir().unwrap();
4793        let q = Queue::at(dir.path().join("not-a-directory"));
4794        std::fs::write(q.root(), "file").unwrap();
4795        let questions = Questions::at(dir.path().join("questions"));
4796        assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
4797        assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
4798    }
4799
4800    #[test]
4801    fn a_dependency_without_a_tombstone_is_still_missing() {
4802        let (dir, q) = queue();
4803        let questions = Questions::at(dir.path().join("questions"));
4804        let mut dep = task("dependency");
4805        q.put(&mut dep).unwrap();
4806        std::fs::remove_file(q.path_of(&dep.id)).unwrap();
4807        assert_eq!(
4808            missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
4809            [dep.id.clone()]
4810        );
4811        assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
4812    }
4813
4814    #[test]
4815    fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
4816        let mut t = task("held");
4817        t.hold_machine(Some("because".to_owned()));
4818        t.blocked_by = vec!["gone".to_owned()];
4819        assert!(!t.dependency_deleted("gone"));
4820        assert_eq!(t.status, TaskStatus::Held);
4821    }
4822
4823    #[test]
4824    fn a_failed_record_removal_takes_the_tombstone_back() {
4825        let (_dir, q) = queue();
4826        let mut t = task("stays");
4827        q.put(&mut t).unwrap();
4828        q.write_tombstone(&t.id).unwrap();
4829        let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
4830        assert!(err.is_err());
4831        // `remove` is what withdraws the marker; a live record is never
4832        // treated as deleted whatever is on disk.
4833        assert!(!q.was_deleted(&t.id));
4834    }
4835
4836    fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
4837        let p = dir.join(name);
4838        std::fs::write(&p, body).unwrap();
4839        p
4840    }
4841
4842    #[test]
4843    fn an_attachment_copy_survives_deleting_its_source() {
4844        let (dir, q) = queue();
4845        let src = source_file(dir.path(), "shot.png", "pixels");
4846        let mut t = task("with a picture");
4847        let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
4848        q.put(&mut t).unwrap();
4849        std::fs::remove_file(&src).unwrap();
4850        assert_eq!(names, ["shot.png"]);
4851        let loaded = q.get(&t.id).unwrap();
4852        let paths = q.attachment_paths(&loaded);
4853        assert_eq!(paths.len(), 1);
4854        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
4855    }
4856
4857    #[test]
4858    fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
4859        let (dir, q) = queue();
4860        let mut t = task("bad names");
4861        for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
4862            let src = source_file(dir.path(), name, "x");
4863            assert!(
4864                q.attach(&mut t, &[src]).is_err(),
4865                "`{name}` must be refused"
4866            );
4867        }
4868        // A drive-qualified name cannot exist as a file on Windows (joining it
4869        // to a directory yields a drive-relative path to `foo.png` instead),
4870        // so the rule is asserted on the name itself, everywhere.
4871        assert!(!crate::ask::valid_asset_name("C:foo.png"));
4872        #[cfg(not(windows))]
4873        {
4874            let src = source_file(dir.path(), "C:foo.png", "x");
4875            assert!(q.attach(&mut t, &[src]).is_err());
4876        }
4877        let long = format!("{}.png", "a".repeat(70));
4878        let src = source_file(dir.path(), &long, "x");
4879        assert!(q.attach(&mut t, &[src]).is_err());
4880        assert!(t.attachments.is_empty());
4881        assert!(!q.attachments_dir(&t.id).exists());
4882    }
4883
4884    #[test]
4885    fn a_taken_attachment_name_is_numbered_not_overwritten() {
4886        let (dir, q) = queue();
4887        let a = source_file(dir.path(), "shot.png", "one");
4888        let sub = dir.path().join("other");
4889        std::fs::create_dir_all(&sub).unwrap();
4890        let b = source_file(&sub, "shot.png", "two");
4891        let mut t = task("collision");
4892        q.attach(&mut t, &[a]).unwrap();
4893        q.attach(&mut t, &[b]).unwrap();
4894        assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
4895        let paths = q.attachment_paths(&t);
4896        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
4897        assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
4898    }
4899
4900    #[test]
4901    fn a_renumbered_name_stays_inside_the_length_bound() {
4902        let (dir, q) = queue();
4903        let name = format!("{}.png", "a".repeat(60));
4904        assert_eq!(name.len(), 64);
4905        let a = source_file(dir.path(), &name, "one");
4906        let sub = dir.path().join("other");
4907        std::fs::create_dir_all(&sub).unwrap();
4908        let b = source_file(&sub, &name, "two");
4909        let mut t = task("long");
4910        q.attach(&mut t, &[a, b]).unwrap();
4911        assert_eq!(t.attachments.len(), 2);
4912        assert!(
4913            t.attachments
4914                .iter()
4915                .all(|n| crate::ask::valid_asset_name(n))
4916        );
4917        assert!(t.attachments[1].ends_with("-2.png"));
4918    }
4919
4920    #[test]
4921    fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
4922        let (dir, q) = queue();
4923        let good = source_file(dir.path(), "good.png", "ok");
4924        let mut t = task("partial");
4925        q.attach(&mut t, &[good]).unwrap();
4926        let more = source_file(dir.path(), "more.png", "ok");
4927        let missing = dir.path().join("missing.png");
4928        assert!(q.attach(&mut t, &[more, missing]).is_err());
4929        assert_eq!(t.attachments, ["good.png"]);
4930        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4931            .unwrap()
4932            .flatten()
4933            .collect();
4934        assert_eq!(on_disk.len(), 1);
4935    }
4936
4937    /// Make the next `put` of `t` fail: its temp file's path is a directory.
4938    fn block_put(q: &Queue, t: &Task) -> PathBuf {
4939        let tmp = q.path_of(&t.id).with_extension("json.tmp");
4940        std::fs::create_dir_all(&tmp).unwrap();
4941        tmp
4942    }
4943
4944    #[test]
4945    fn a_failed_put_leaves_no_new_attachment_directory() {
4946        let (dir, q) = queue();
4947        let mut t = task("fresh");
4948        let tmp = block_put(&q, &t);
4949        let src = source_file(dir.path(), "shot.png", "x");
4950        assert!(q.attach_and_put(&mut t, &[src]).is_err());
4951        assert!(t.attachments.is_empty());
4952        assert!(!q.attachments_dir(&t.id).exists());
4953        assert!(!q.path_of(&t.id).exists());
4954        std::fs::remove_dir(tmp).unwrap();
4955    }
4956
4957    #[test]
4958    fn a_failed_put_removes_only_the_copy_it_just_made() {
4959        let (dir, q) = queue();
4960        let mut t = task("edited");
4961        let first = source_file(dir.path(), "first.png", "1");
4962        q.attach_and_put(&mut t, &[first]).unwrap();
4963        block_put(&q, &t);
4964        let second = source_file(dir.path(), "second.png", "2");
4965        assert!(q.attach_and_put(&mut t, &[second]).is_err());
4966        assert_eq!(t.attachments, ["first.png"]);
4967        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4968            .unwrap()
4969            .flatten()
4970            .map(|e| e.file_name().to_string_lossy().into_owned())
4971            .collect();
4972        assert_eq!(on_disk, ["first.png"]);
4973        assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
4974    }
4975
4976    #[test]
4977    fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
4978        let (dir, q) = queue();
4979        let questions = Questions::at(dir.path().join("questions"));
4980        let gone = task("gone");
4981        let mut other = task("other");
4982        let mut live = task("live");
4983        q.put(&mut other).unwrap();
4984        q.put(&mut live).unwrap();
4985        // `gone` has no record: its removal finished deleting the record but
4986        // not the set-aside attachments.
4987        let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
4988        std::fs::create_dir_all(&orphan).unwrap();
4989        std::fs::write(orphan.join("shot.png"), "x").unwrap();
4990        // `live` still has a record, so its `.removing` belongs to a removal
4991        // in progress and must be left alone.
4992        let busy = q.root.join(format!("{}.attachments.removing", live.id));
4993        std::fs::create_dir_all(&busy).unwrap();
4994
4995        q.remove(&other.id, false, &questions).unwrap();
4996        assert!(!orphan.exists(), "an orphan is swept");
4997        assert!(busy.exists(), "a removal in progress is left alone");
4998    }
4999
5000    #[test]
5001    fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
5002        let (dir, q) = queue();
5003        let questions = Questions::at(dir.path().join("questions"));
5004        let mut t = task("stuck");
5005        let src = source_file(dir.path(), "shot.png", "x");
5006        q.attach_and_put(&mut t, &[src]).unwrap();
5007        // A non-empty directory already at the set-aside name makes the rename
5008        // fail; its record is present, so the sweep leaves it in place.
5009        let aside = q.root.join(format!("{}.attachments.removing", t.id));
5010        std::fs::create_dir_all(&aside).unwrap();
5011        std::fs::write(aside.join("old.png"), "o").unwrap();
5012        assert!(q.remove(&t.id, false, &questions).is_err());
5013        assert!(q.path_of(&t.id).exists());
5014        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5015    }
5016
5017    #[test]
5018    fn a_failed_record_removal_puts_the_attachments_back() {
5019        let (dir, q) = queue();
5020        let mut t = task("rollback");
5021        let src = source_file(dir.path(), "shot.png", "x");
5022        q.attach_and_put(&mut t, &[src]).unwrap();
5023        let err = q
5024            .remove_record_with_attachments(&t.id, |_| {
5025                Err(std::io::Error::other("injected failure"))
5026            })
5027            .unwrap_err();
5028        assert!(format!("{err:#}").contains("injected failure"));
5029        assert!(q.path_of(&t.id).exists());
5030        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5031        assert!(
5032            !q.root
5033                .join(format!("{}.attachments.removing", t.id))
5034                .exists()
5035        );
5036    }
5037
5038    #[test]
5039    fn editing_a_task_keeps_its_attachments() {
5040        let (dir, q) = queue();
5041        let src = source_file(dir.path(), "shot.png", "x");
5042        let mut t = task("editable");
5043        q.attach(&mut t, &[src]).unwrap();
5044        t.edit("new".to_owned(), "new text".to_owned()).unwrap();
5045        q.put(&mut t).unwrap();
5046        assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
5047    }
5048
5049    #[test]
5050    fn removing_a_task_deletes_its_attachments() {
5051        let (dir, q) = queue();
5052        let questions = Questions::at(dir.path().join("questions"));
5053        let src = source_file(dir.path(), "shot.png", "x");
5054        let mut t = task("doomed");
5055        q.attach(&mut t, &[src]).unwrap();
5056        q.put(&mut t).unwrap();
5057        assert!(q.attachments_dir(&t.id).is_dir());
5058        q.remove(&t.id, false, &questions).unwrap();
5059        assert!(!q.attachments_dir(&t.id).exists());
5060        assert!(q.list().is_empty());
5061    }
5062
5063    #[test]
5064    fn a_task_written_before_attachments_still_reads() {
5065        let (_dir, q) = queue();
5066        let mut t = task("old");
5067        q.put(&mut t).unwrap();
5068        let path = q.path_of(&t.id);
5069        let mut v: serde_json::Value =
5070            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
5071        v.as_object_mut().unwrap().remove("attachments");
5072        std::fs::write(&path, v.to_string()).unwrap();
5073        assert!(q.get(&t.id).unwrap().attachments.is_empty());
5074    }
5075
5076    #[test]
5077    fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
5078        let q = Queue::at(PathBuf::from("relative-queue"));
5079        let mut t = task("rel");
5080        t.attachments.push("shot.png".to_owned());
5081        let paths = q.attachment_paths(&t);
5082        assert!(paths[0].is_absolute(), "{}", paths[0].display());
5083        assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
5084    }
5085
5086    #[test]
5087    fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
5088        let dir = tempfile::tempdir().unwrap();
5089        let q = Queue::at(dir.path().join("queue"));
5090        let mut t = task("resume");
5091        t.start("A".into());
5092        t.record_pass("A", 0, 1, None);
5093        t.start("B".into());
5094        q.put(&mut t).unwrap();
5095        assert_eq!(t.runs, vec!["A", "B"]);
5096        t.start("A".into());
5097        t.record_pass("A", 1, 3, None);
5098        q.put(&mut t).unwrap();
5099        let saved = q.get(&t.id).unwrap();
5100        assert_eq!(saved.runs, vec!["B", "A"]);
5101        assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
5102        assert_eq!(saved.attempts, 3);
5103        assert_eq!(saved.passes.len(), 2);
5104        assert_eq!(saved.passes[0].run, "A");
5105        assert_eq!(saved.passes[1].run, "A");
5106    }
5107
5108    #[test]
5109    fn repeated_start_records_one_run_and_separate_passes() {
5110        let mut t = task("resume");
5111        t.start("same-run".into());
5112        t.record_pass("same-run", 0, 1, None);
5113        t.start("same-run".into());
5114        t.record_pass("same-run", 1, 2, None);
5115        assert_eq!(t.runs, vec!["same-run"]);
5116        assert_eq!(t.attempts, 2);
5117        assert_eq!(t.passes.len(), 2);
5118    }
5119
5120    #[test]
5121    fn link_run_adds_a_run_once_and_touches_nothing_else() {
5122        let dir = tempfile::tempdir().unwrap();
5123        let queue = Queue::at(dir.path().join("queue"));
5124        let mut t = Task::new(
5125            "t".to_owned(),
5126            "do it".to_owned(),
5127            PathBuf::from("."),
5128            Source::Human,
5129        );
5130        queue.put(&mut t).unwrap();
5131        let before = queue.get(&t.id).unwrap();
5132
5133        let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
5134        assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
5135        assert_eq!(linked.status, before.status);
5136        assert_eq!(linked.attempts, before.attempts);
5137        assert_eq!(linked.interrupt, before.interrupt);
5138
5139        let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5140        assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
5141        assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
5142        assert!(queue.link_run("no-such-task", "r").is_err());
5143    }
5144
5145    #[test]
5146    fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
5147        let dir = tempfile::tempdir().unwrap();
5148        let queue = Queue::at(dir.path().join("queue"));
5149        let mut t = Task::new(
5150            "t".to_owned(),
5151            "do it".to_owned(),
5152            PathBuf::from("."),
5153            Source::Human,
5154        );
5155        queue.put(&mut t).unwrap();
5156        // The daemon's copy, taken before the link.
5157        let mut snapshot = queue.get(&t.id).unwrap();
5158        queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5159
5160        snapshot.start("20260930-000000-aaaa".to_owned());
5161        queue.put(&mut snapshot).unwrap();
5162
5163        let stored = queue.get(&t.id).unwrap();
5164        assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
5165        assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
5166        assert_eq!(stored.attempts, 1);
5167    }
5168
5169    #[test]
5170    fn concurrent_links_and_daemon_saves_lose_nothing() {
5171        let dir = tempfile::tempdir().unwrap();
5172        let queue = Queue::at(dir.path().join("queue"));
5173        let mut t = Task::new(
5174            "t".to_owned(),
5175            "do it".to_owned(),
5176            PathBuf::from("."),
5177            Source::Human,
5178        );
5179        queue.put(&mut t).unwrap();
5180        let id = t.id.clone();
5181
5182        let linkers: Vec<_> = (0..4)
5183            .map(|n| {
5184                let (queue, id) = (queue.clone(), id.clone());
5185                std::thread::spawn(move || {
5186                    for k in 0..10 {
5187                        queue
5188                            .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
5189                            .unwrap();
5190                    }
5191                })
5192            })
5193            .collect();
5194        // The daemon's side: transitions saved from its own, ever staler,
5195        // copy of the task.
5196        let mut mine = queue.get(&id).unwrap();
5197        for k in 0..10 {
5198            mine.start(format!("20260930-000009-d{k:03}"));
5199            queue.put(&mut mine).unwrap();
5200        }
5201        for l in linkers {
5202            l.join().unwrap();
5203        }
5204
5205        let stored = queue.get(&id).unwrap();
5206        assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
5207        assert_eq!(
5208            stored.attempts, 10,
5209            "linking never rewinds the daemon's work"
5210        );
5211    }
5212
5213    fn age_lock(path: &Path) {
5214        let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
5215        f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
5216            .unwrap();
5217    }
5218
5219    #[test]
5220    fn concurrent_stale_takeover_yields_one_holder() {
5221        use std::sync::atomic::{AtomicUsize, Ordering};
5222        use std::sync::{Arc, Barrier};
5223        for _ in 0..5 {
5224            let dir = tempfile::tempdir().unwrap();
5225            let q = Queue::at(dir.path().to_path_buf());
5226            let lock = dir.path().join("t.write-lock");
5227            std::fs::write(&lock, "dead-0000").unwrap();
5228            age_lock(&lock);
5229            let n = 6;
5230            let barrier = Arc::new(Barrier::new(n));
5231            let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
5232            let handles: Vec<_> = (0..n)
5233                .map(|_| {
5234                    let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
5235                    std::thread::spawn(move || {
5236                        b.wait();
5237                        let g = q.lock_task("t").unwrap();
5238                        let held = now.fetch_add(1, Ordering::SeqCst) + 1;
5239                        max.fetch_max(held, Ordering::SeqCst);
5240                        std::thread::sleep(std::time::Duration::from_millis(20));
5241                        now.fetch_sub(1, Ordering::SeqCst);
5242                        drop(g);
5243                    })
5244                })
5245                .collect();
5246            for h in handles {
5247                h.join().unwrap();
5248            }
5249            assert_eq!(max.load(Ordering::SeqCst), 1);
5250            assert!(!lock.exists());
5251        }
5252    }
5253
5254    #[test]
5255    fn dropping_a_replaced_lock_leaves_the_new_holders_token() {
5256        let dir = tempfile::tempdir().unwrap();
5257        let q = Queue::at(dir.path().to_path_buf());
5258        let lock = dir.path().join("t.write-lock");
5259        let a = q.lock_task("t").unwrap();
5260        // Simulate a foreign/legacy writer replacing the file. A modern
5261        // writer cannot acquire it while this guard's process is alive.
5262        let replacement = owner_token();
5263        std::fs::write(&lock, &replacement).unwrap();
5264        drop(a);
5265        assert_eq!(std::fs::read_to_string(&lock).unwrap(), replacement);
5266    }
5267
5268    #[test]
5269    fn aged_live_task_lock_excludes_updates_until_retirement_releases_it() {
5270        let dir = tempfile::tempdir().unwrap();
5271        let q = Queue::at(dir.path().to_path_buf());
5272        let held = q.lock_task("t").unwrap();
5273        let path = dir.path().join("t.write-lock");
5274        age_lock(&path);
5275        assert!(!break_stale(&path, &held.token));
5276        let (started_tx, started_rx) = std::sync::mpsc::channel();
5277        let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
5278        let updater = std::thread::spawn(move || {
5279            started_tx.send(()).unwrap();
5280            let _write = q.lock_task("t").unwrap();
5281            acquired_tx.send(()).unwrap();
5282        });
5283        started_rx.recv().unwrap();
5284        assert!(
5285            acquired_rx
5286                .recv_timeout(std::time::Duration::from_millis(100))
5287                .is_err()
5288        );
5289        assert_eq!(std::fs::read_to_string(&path).unwrap(), held.token);
5290        drop(held);
5291        acquired_rx
5292            .recv_timeout(std::time::Duration::from_secs(2))
5293            .unwrap();
5294        updater.join().unwrap();
5295        assert!(!path.exists());
5296    }
5297
5298    #[test]
5299    fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
5300        let dir = tempfile::tempdir().unwrap();
5301        let lock = dir.path().join("t.write-lock");
5302        std::fs::write(&lock, "new-token").unwrap();
5303        assert!(!break_stale(&lock, "old-token"));
5304        assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
5305        // Same token but fresh: not stale any more, so also left alone.
5306        assert!(!break_stale(&lock, "new-token"));
5307        assert!(lock.exists());
5308        age_lock(&lock);
5309        assert!(break_stale(&lock, "new-token"));
5310        assert!(!lock.exists());
5311    }
5312
5313    #[test]
5314    fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
5315        let dir = tempfile::tempdir().unwrap();
5316        let lock = dir.path().join("t.write-lock");
5317        std::fs::write(&lock, "dead-1").unwrap();
5318        age_lock(&lock);
5319        let marker = break_marker(&lock, "dead-1");
5320        std::fs::write(&marker, "crashed-remover").unwrap();
5321        // Fresh marker: another remover is working, so nothing is broken.
5322        assert!(!break_stale(&lock, "dead-1"));
5323        assert!(lock.exists());
5324        // Stale marker: recovered, then the lock is broken.
5325        age_lock(&marker);
5326        assert!(break_stale(&lock, "dead-1"));
5327        assert!(!lock.exists());
5328        assert!(!marker.exists());
5329    }
5330
5331    fn chat_task() -> Task {
5332        Task::new(
5333            "t".to_owned(),
5334            "do t".to_owned(),
5335            PathBuf::from("."),
5336            Source::Agent {
5337                run: "20260902-000000-beef".to_owned(),
5338                node: CHAT_NODE.to_owned(),
5339            },
5340        )
5341    }
5342
5343    #[test]
5344    fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
5345        let mut t = chat_task();
5346        t.hold_machine(Some("waiting".into()));
5347        let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
5348        t.held_at = Some(old);
5349        t.block(vec!["decision".into()], None);
5350        t.hold_machine(Some("same waiting decision".into()));
5351        assert_eq!(t.held_at, Some(old));
5352        t.start("real-new-run".into());
5353        t.fail("new failure", 1);
5354        assert_ne!(
5355            t.held_at,
5356            Some(old),
5357            "old answers cannot approve a new hold"
5358        );
5359    }
5360
5361    #[test]
5362    fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
5363        let (_dir, queue) = queue();
5364        let mut t = chat_task();
5365        t.start("old-run".into());
5366        t.hold_machine(Some("operator is considering recovery".into()));
5367        t.chat_notice = Some(ChatNotice {
5368            key: format!("held#{}", t.held_at.unwrap()),
5369            at: Timestamp::now(),
5370            outcome: ChatNoticeOutcome::Posted,
5371        });
5372        let mut raw = serde_json::to_value(&t).unwrap();
5373        raw.as_object_mut().unwrap().remove("chat_notices");
5374        raw.as_object_mut().unwrap().remove("hold_notice_cause");
5375        std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
5376        std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
5377        assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5378        queue
5379            .modify(&t.id, |task| {
5380                task.requeue();
5381                task.link_run("unprepared-run");
5382                task.startup_notice_cause("startup-source");
5383                task.fail("same unresolved preflight", 0);
5384                true
5385            })
5386            .unwrap();
5387        assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5388        queue
5389            .modify(&t.id, |task| {
5390                task.succeed();
5391                true
5392            })
5393            .unwrap();
5394        assert_eq!(
5395            queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
5396            Some("done")
5397        );
5398    }
5399
5400    #[test]
5401    fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
5402        let mut t = chat_task();
5403        t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
5404        let key = t.chat_notice_key().unwrap();
5405        t.chat_notices.push(key);
5406        t.release();
5407        t.block(
5408            vec!["b".into(), "a".into(), "a".into()],
5409            Some("still waiting on prerequisites".into()),
5410        );
5411        assert_eq!(t.chat_notice_due(), None);
5412        t.block(
5413            vec!["new-question".into()],
5414            Some("different decision".into()),
5415        );
5416        assert!(t.chat_notice_due().is_some());
5417    }
5418
5419    #[test]
5420    fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
5421        let mut t = chat_task();
5422        t.start("existing-run".into());
5423        t.fail("implementation failed", 1);
5424        let failed = t.chat_notice_key().unwrap();
5425        t.chat_notices.push(failed.clone());
5426        t.release_operator("test", None);
5427        t.start("existing-run".into());
5428        t.fail_startup("startup-source", "same unresolved recovery source", 1);
5429        assert_eq!(
5430            t.chat_notice_due(),
5431            None,
5432            "no-progress recovery stays quiet"
5433        );
5434        t.release_operator("test", None);
5435        t.start("existing-run".into());
5436        t.fail_startup("startup-attachments", "attachment missing", 1);
5437        let attachments = t.chat_notice_due().expect("new attachment blocker");
5438        assert_ne!(attachments, failed);
5439        assert_eq!(
5440            t.hold_notice_cause.as_ref().unwrap().stage,
5441            "startup-attachments"
5442        );
5443        t.chat_notices.push(attachments);
5444        t.release_operator("test", None);
5445        t.start("existing-run".into());
5446        t.fail_startup(
5447            "startup-attachments",
5448            "same missing attachment, reworded",
5449            1,
5450        );
5451        assert_eq!(
5452            t.chat_notice_due(),
5453            None,
5454            "same stage after attempt counter reset"
5455        );
5456        t.release_operator("test", None);
5457        t.start("existing-run".into());
5458        t.fail_startup("startup-config", "different preflight stage", 1);
5459        assert!(t.chat_notice_due().is_some());
5460        assert_eq!(
5461            t.hold_notice_cause.as_ref().unwrap().stage,
5462            "startup-config"
5463        );
5464    }
5465
5466    #[test]
5467    fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
5468        let mut t = chat_task();
5469        t.hold_machine(Some("startup source missing".into()));
5470        t.startup_notice_cause("startup-source");
5471        let first = t.chat_notice_key().unwrap();
5472        t.chat_notices.push(first.clone());
5473        t.release_operator("test", None);
5474        t.startup_notice_cause("startup-source");
5475        t.hold_machine(Some("same source cannot be read".into()));
5476        assert_eq!(t.chat_notice_due(), None);
5477        t.startup_notice_cause("startup-config");
5478        assert!(t.chat_notice_due().is_some(), "different startup stage");
5479        t.notice_cause("failed", Some("existing-run"), Some(0));
5480        let run_failure = t.chat_notice_key().unwrap();
5481        t.chat_notices.push(run_failure.clone());
5482        t.release_operator("test", None);
5483        t.startup_notice_cause("startup-source");
5484        t.hold_machine(Some("no new work happened".into()));
5485        assert_eq!(t.chat_notice_due(), None);
5486        t.notice_cause("failed", Some("existing-run"), Some(1));
5487        assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
5488        assert_ne!(t.chat_notice_key().unwrap(), run_failure);
5489    }
5490
5491    #[test]
5492    fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
5493        let mut t = chat_task();
5494        t.hold_machine(Some("still waiting".into()));
5495        t.chat_notice = Some(ChatNotice {
5496            key: "held#2026-10-14T00:00:00Z".into(),
5497            at: Timestamp::now(),
5498            outcome: ChatNoticeOutcome::Posted,
5499        });
5500        assert_eq!(t.chat_notice_due(), None, "old hold baseline");
5501        t.import_chat_notice();
5502        t.notice_cause("failed", Some("new-run"), Some(0));
5503        assert!(t.chat_notice_due().is_some(), "new producer evidence");
5504        t.succeed();
5505        assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
5506    }
5507
5508    #[test]
5509    fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
5510        let mut t = chat_task();
5511        assert_eq!(t.chat_notice_due(), None, "queued");
5512        t.start("r1".to_owned());
5513        assert_eq!(t.chat_notice_due(), None, "running");
5514        t.fail("boom", 3);
5515        assert_eq!(t.status, TaskStatus::Failed);
5516        assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
5517        t.start("r2".to_owned());
5518        t.park("upgrade");
5519        assert_eq!(t.chat_notice_due(), None, "parked");
5520        t.fail("boom", 1);
5521        assert_eq!(t.status, TaskStatus::Held);
5522        let key = t
5523            .chat_notice_due()
5524            .expect("held on the last attempt is final");
5525        t.chat_notice = Some(ChatNotice {
5526            key,
5527            at: Timestamp::now(),
5528            outcome: ChatNoticeOutcome::Posted,
5529        });
5530        assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
5531        t.release();
5532        assert_eq!(t.chat_notice_due(), None, "requeued after a release");
5533        t.start("r3".to_owned());
5534        t.succeed();
5535        assert!(t.chat_notice_due().is_some(), "done is a new final state");
5536    }
5537
5538    #[test]
5539    fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
5540        for key in ["done", "done#0", "done#1", "done#12"] {
5541            let mut t = chat_task();
5542            t.start("r1".to_owned());
5543            t.succeed();
5544            t.chat_notice = Some(ChatNotice {
5545                key: key.to_owned(),
5546                at: Timestamp::now(),
5547                outcome: ChatNoticeOutcome::Posted,
5548            });
5549            t.requeue();
5550            t.start("r2".to_owned());
5551            t.succeed();
5552            assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
5553            assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
5554        }
5555    }
5556
5557    #[test]
5558    fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
5559        let mut t = chat_task();
5560        t.hold_manual(Some("wait".to_owned()));
5561        let first = t.chat_notice_due().expect("held");
5562        t.chat_notice = Some(ChatNotice {
5563            key: first.clone(),
5564            at: Timestamp::now(),
5565            outcome: ChatNoticeOutcome::Posted,
5566        });
5567        t.hold_manual(Some("still waiting".to_owned()));
5568        assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
5569        t.release();
5570        std::thread::sleep(std::time::Duration::from_millis(5));
5571        t.hold_manual(None);
5572        assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
5573    }
5574
5575    #[test]
5576    fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
5577        let mut human = task("h");
5578        human.succeed();
5579        assert_eq!(human.chat_notice_due(), None);
5580        // A follow-up only inherits the chat; it did not file the task.
5581        let mut inherited = task("i");
5582        inherited.origin_chat = Some("20260902-000000-beef".to_owned());
5583        inherited.succeed();
5584        assert_eq!(inherited.chat_notice_due(), None);
5585        let mut old = chat_task();
5586        old.schema = 12;
5587        old.status = TaskStatus::Done;
5588        assert_eq!(old.chat_notice_due(), None, "history is not replayed");
5589        let mut live = chat_task();
5590        live.schema = 12;
5591        live.start("r1".to_owned());
5592        live.succeed();
5593        assert!(live.chat_notice_due().is_some(), "started by this build");
5594        let mut held = chat_task();
5595        held.schema = 12;
5596        held.hold_machine(Some("config".to_owned()));
5597        assert!(held.chat_notice_due().is_some(), "held before any start");
5598    }
5599
5600    #[test]
5601    fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
5602        let tmp = tempfile::tempdir().expect("tempdir");
5603        let q = Queue::at(tmp.path().join("queue"));
5604        let mut t = chat_task();
5605        t.succeed();
5606        q.put(&mut t).expect("put");
5607        let mut stale = q.get(&t.id).expect("get");
5608        let key = t.chat_notice_due().expect("due");
5609        q.modify(&t.id, |x| {
5610            x.chat_notice = Some(ChatNotice {
5611                key: key.clone(),
5612                at: Timestamp::now(),
5613                outcome: ChatNoticeOutcome::Posted,
5614            });
5615            true
5616        })
5617        .expect("modify");
5618        q.put(&mut stale).expect("stale put");
5619        assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
5620    }
5621}