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    /// 回答された Question の ID。Question を伴わない再開では空文字列。
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(
2054    path: &Path,
2055    token: &str,
2056    depth: u8,
2057    retry: crate::lock_file::Retry,
2058) -> Option<(PathBuf, String)> {
2059    use std::io::Write;
2060    let marker = break_marker(path, token);
2061    for _ in 0..2 {
2062        match retry.create(&marker) {
2063            Ok(mut f) => {
2064                let mine = owner_token();
2065                if f.write_all(mine.as_bytes()).is_err() {
2066                    drop(f);
2067                    let _ = std::fs::remove_file(&marker);
2068                    return None;
2069                }
2070                return Some((marker, mine));
2071            }
2072            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2073                let judged = std::fs::read_to_string(&marker).ok();
2074                match judged.filter(|_| older_than_stale(&marker)) {
2075                    Some(judged) if depth < MAX_MARKER_DEPTH => {
2076                        remove_lock_if(&marker, &judged, true, depth + 1, retry)?;
2077                    }
2078                    _ => return None,
2079                }
2080            }
2081            Err(_) => return None,
2082        }
2083    }
2084    None
2085}
2086
2087/// Remove the lock at `path` if it still carries `token` (and, when
2088/// `require_stale`, is still stale). Removal is serialized per token by the
2089/// break marker, so a remover that judged an old lock cannot delete the fresh
2090/// one that replaced it. `None` means the marker was busy: try again.
2091fn remove_lock_if(
2092    path: &Path,
2093    token: &str,
2094    require_stale: bool,
2095    depth: u8,
2096    retry: crate::lock_file::Retry,
2097) -> Option<bool> {
2098    let (marker, mine) = take_marker(path, token, depth, retry)?;
2099    let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
2100        && (!require_stale || older_than_stale(path) && writer_may_be_gone(token));
2101    if still {
2102        let _ = std::fs::remove_file(path);
2103    }
2104    // Release the marker only if it is still ours.
2105    if depth >= MAX_MARKER_DEPTH {
2106        if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
2107            let _ = std::fs::remove_file(&marker);
2108        }
2109    } else {
2110        let _ = remove_lock_if(&marker, &mine, false, depth + 1, retry);
2111    }
2112    Some(still)
2113}
2114
2115/// Break the lock judged stale while it held `judged`. Returns whether it was
2116/// removed; a lock that has since been replaced is left alone.
2117#[cfg(test)]
2118fn break_stale(path: &Path, judged: &str) -> bool {
2119    break_stale_with(path, judged, crate::lock_file::Retry::new(None))
2120}
2121
2122fn break_stale_with(path: &Path, judged: &str, retry: crate::lock_file::Retry) -> bool {
2123    remove_lock_if(path, judged, true, 0, retry).unwrap_or(false)
2124}
2125
2126impl Drop for TaskLock {
2127    fn drop(&mut self) {
2128        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
2129        let retry = crate::lock_file::Retry::new(Some(deadline));
2130        loop {
2131            match remove_lock_if(&self.path, &self.token, false, 0, retry) {
2132                Some(_) => return,
2133                None if std::time::Instant::now() > deadline => return,
2134                None => std::thread::sleep(std::time::Duration::from_millis(5)),
2135            }
2136        }
2137    }
2138}
2139
2140/// A queue on disk.
2141#[derive(Debug, Clone)]
2142pub struct Queue {
2143    root: PathBuf,
2144}
2145
2146impl Queue {
2147    /// The operator's queue, `<home>/queue`.
2148    pub fn open() -> Self {
2149        Self::at(crate::run::home().join("queue"))
2150    }
2151
2152    /// A queue at an explicit root. Tests use this; so could an operator who
2153    /// wants a queue per project.
2154    pub fn at(root: PathBuf) -> Self {
2155        Self { root }
2156    }
2157
2158    /// Directory holding the task files.
2159    pub fn root(&self) -> &Path {
2160        &self.root
2161    }
2162
2163    /// Path for one task id.
2164    pub fn path_of(&self, id: &str) -> PathBuf {
2165        self.root.join(format!("{id}.json"))
2166    }
2167
2168    /// Directory holding a task's attachments. A sibling of the task file,
2169    /// not a `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
2170    pub fn attachments_dir(&self, id: &str) -> PathBuf {
2171        self.root.join(format!("{id}.attachments"))
2172    }
2173
2174    /// Copy each of `sources` into `task`'s attachment directory and record
2175    /// the stored names on `task` (persist with [`Queue::put`]). Returns the
2176    /// names, in order.
2177    ///
2178    /// Every source is checked (a readable file, a name passing
2179    /// [`crate::ask::valid_asset_name`]) before anything is copied, and a
2180    /// copy that fails part-way removes only what this call created. A name
2181    /// already taken is never overwritten: it becomes `stem-2.ext`,
2182    /// `stem-3.ext`, ... A name that does not validate is refused rather than
2183    /// sanitised - the operator renames the file.
2184    pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2185        let mut wanted = Vec::new();
2186        for src in sources {
2187            let name = src
2188                .file_name()
2189                .and_then(|n| n.to_str())
2190                .with_context(|| format!("`{}` has no usable file name", src.display()))?;
2191            if !crate::ask::valid_asset_name(name) {
2192                bail!(
2193                    "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
2194                     with no `..`; rename the file and try again"
2195                );
2196            }
2197            if !src.is_file() {
2198                bail!("attachment `{}` is not a file", src.display());
2199            }
2200            wanted.push((src, name));
2201        }
2202        let dir = self.attachments_dir(&task.id);
2203        let existed = dir.is_dir();
2204        std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
2205        let mut created: Vec<PathBuf> = Vec::new();
2206        let mut names = Vec::new();
2207        let mut copy_all = || -> Result<()> {
2208            for (src, name) in &wanted {
2209                let (stored, path) = copy_new(&dir, src, name)?;
2210                created.push(path);
2211                names.push(stored);
2212            }
2213            Ok(())
2214        };
2215        if let Err(e) = copy_all() {
2216            for path in &created {
2217                let _ = std::fs::remove_file(path);
2218            }
2219            if !existed {
2220                let _ = std::fs::remove_dir(&dir);
2221            }
2222            return Err(e);
2223        }
2224        task.attachments.extend(names.iter().cloned());
2225        Ok(names)
2226    }
2227
2228    /// [`Queue::attach`] then [`Queue::put`], undoing the copies when the
2229    /// record cannot be written. Without the undo the files sit in
2230    /// `<id>.attachments/` with no task pointing at them.
2231    ///
2232    /// Only what this call copied is removed (the names `attach` returned,
2233    /// and the directory itself only when it did not exist before and is
2234    /// empty), never an attachment an earlier save recorded. `task` is rolled
2235    /// back too, and the error the caller sees is `put`'s own. This relies on
2236    /// a failed `put` leaving the stored record untouched, which its
2237    /// write-then-rename guarantees.
2238    pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
2239        let dir = self.attachments_dir(&task.id);
2240        let existed = dir.is_dir();
2241        let before = task.attachments.len();
2242        let names = self.attach(task, sources)?;
2243        if let Err(e) = self.put(task) {
2244            for name in &names {
2245                let _ = std::fs::remove_file(dir.join(name));
2246            }
2247            if !existed {
2248                let _ = std::fs::remove_dir(&dir);
2249            }
2250            task.attachments.truncate(before);
2251            return Err(e);
2252        }
2253        Ok(names)
2254    }
2255
2256    /// Absolute paths of `task`'s attachments, whatever shape this queue's
2257    /// root has. Absolute because the prompt hands them to an agent whose
2258    /// working directory is somewhere else entirely.
2259    pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
2260        let dir = self.attachments_dir(&task.id);
2261        task.attachments
2262            .iter()
2263            .map(|n| {
2264                let p = dir.join(n);
2265                std::path::absolute(&p).unwrap_or(p)
2266            })
2267            .collect()
2268    }
2269
2270    /// Write a task, atomically, so a daemon killed mid-write leaves the
2271    /// previous state readable rather than a truncated file.
2272    ///
2273    /// Runs already recorded on disk that `task` does not carry are kept, not
2274    /// dropped: `magi run --task` links a run into a task from another
2275    /// process ([`Queue::link_run`]) while a daemon may hold a snapshot taken
2276    /// before it, and writing that snapshot back whole would erase the link.
2277    /// Nothing ever removes a run from a task, so the union loses nothing.
2278    pub fn put(&self, task: &mut Task) -> Result<()> {
2279        let _lock = self.lock_task(&task.id)?;
2280        self.put_unlocked(task)
2281    }
2282
2283    /// Write `task` only if no task with its id exists yet; `Ok(false)` when
2284    /// one does. Unlike [`Queue::put`] it never replaces a record, so a
2285    /// producer with a deterministic id cannot rewind a task that has since
2286    /// started running or finished.
2287    pub fn create_new(&self, task: &mut Task) -> Result<bool> {
2288        let _lock = self.lock_task(&task.id)?;
2289        if self.path_of(&task.id).exists() {
2290            return Ok(false);
2291        }
2292        self.put_unlocked(task)?;
2293        Ok(true)
2294    }
2295
2296    /// [`Queue::put`] for a caller already holding [`Queue::lock_task`].
2297    fn put_unlocked(&self, task: &mut Task) -> Result<()> {
2298        if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
2299            stored.import_chat_notice();
2300            for key in stored.chat_notices {
2301                if !task.chat_notices.contains(&key) {
2302                    task.chat_notices.push(key);
2303                }
2304            }
2305            for run in stored.runs {
2306                if !task.runs.contains(&run) {
2307                    task.runs.push(run);
2308                }
2309            }
2310            for p in stored.passes {
2311                if !task
2312                    .passes
2313                    .iter()
2314                    .any(|q| q.run == p.run && q.pass_index == p.pass_index)
2315                {
2316                    task.passes.push(p);
2317                }
2318            }
2319            // A completion notice already recorded is never rewound by an
2320            // older snapshot: that would report the same state twice.
2321            if let Some(n) = stored.chat_notice
2322                && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
2323            {
2324                task.chat_notice = Some(n);
2325            }
2326        }
2327        task.import_chat_notice();
2328        task.updated_at = Timestamp::now();
2329        std::fs::create_dir_all(&self.root)
2330            .with_context(|| format!("create {}", self.root.display()))?;
2331        let body = serde_json::to_string_pretty(task).context("serialize task")?;
2332        let path = self.path_of(&task.id);
2333        let tmp = path.with_extension("json.tmp");
2334        std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
2335        std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
2336        // A machine hold is news for the notification centre, filed beside
2337        // this queue (`<home>/queue` -> `<home>/notifications`) rather than
2338        // through a process-global, so a queue in a temp directory notifies
2339        // into that directory.
2340        if let (Some(notice), Some(home)) = (
2341            crate::notices::task_held(task),
2342            self.root.parent().filter(|p| !p.as_os_str().is_empty()),
2343        ) {
2344            crate::notices::raise_in(home, notice);
2345        }
2346        Ok(())
2347    }
2348
2349    /// Change the task `id` in place: read it inside its write lock, let `f`
2350    /// edit it, and write it back only when `f` returns `true`. The check a
2351    /// caller makes in `f` therefore sees the record as stored now, not a
2352    /// snapshot taken before another writer got in. Returns whether it wrote.
2353    pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
2354        let id = self.resolve_id(id)?;
2355        let _lock = self.lock_task(&id)?;
2356        let mut task = self.get(&id)?;
2357        if !f(&mut task) {
2358            return Ok(false);
2359        }
2360        self.put_unlocked(&mut task)?;
2361        Ok(true)
2362    }
2363
2364    /// Append `run` to the runs of the task `id` (or an unambiguous prefix),
2365    /// leaving everything else about it alone — see [`Task::link_run`]. Read
2366    /// and written back in one breath, because [`Queue::put`] replaces the
2367    /// whole record. Returns the task as stored.
2368    pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
2369        let id = self.resolve_id(id)?;
2370        // Read inside the lock, so the record written back is the current one
2371        // and a daemon's write cannot slip between the read and the write.
2372        let _lock = self.lock_task(&id)?;
2373        let mut task = self.get(&id)?;
2374        if task.link_run(run) {
2375            self.put_unlocked(&mut task)?;
2376        }
2377        Ok(task)
2378    }
2379
2380    /// Exclusive right to rewrite task `id`'s record, held until the guard
2381    /// drops. Every writer ([`Queue::put`], [`Queue::link_run`]) takes it, so
2382    /// a process linking a run and the daemon saving a transition are
2383    /// serialized instead of last-writer-wins. An old lock may be broken only
2384    /// after its writer is confirmed dead (legacy tokens without a PID retain
2385    /// age-based recovery). Not the claim lock (`<id>.lock`), held for a
2386    /// whole run.
2387    pub(crate) fn lock_task(&self, id: &str) -> Result<TaskLock> {
2388        std::fs::create_dir_all(&self.root)
2389            .with_context(|| format!("create {}", self.root.display()))?;
2390        let path = self.root.join(format!("{id}.write-lock"));
2391        let started = std::time::Instant::now();
2392        let retry = crate::lock_file::Retry::task_write(started + TASK_LOCK_STALE);
2393        let marker_retry = crate::lock_file::Retry::new(Some(started + TASK_LOCK_STALE));
2394        loop {
2395            match retry.create(&path) {
2396                Ok(mut file) => {
2397                    use std::io::Write;
2398                    let token = owner_token();
2399                    if let Err(e) = file.write_all(token.as_bytes()) {
2400                        drop(file);
2401                        let _ = std::fs::remove_file(&path);
2402                        return Err(e).with_context(|| format!("lock {}", path.display()));
2403                    }
2404                    return Ok(TaskLock { path, token });
2405                }
2406                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2407                    // Content first, then age: a lock replaced in between
2408                    // reads fresh, never stale.
2409                    let judged = std::fs::read_to_string(&path).ok();
2410                    let broken = judged
2411                        .filter(|_| older_than_stale(&path))
2412                        .is_some_and(|judged| break_stale_with(&path, &judged, marker_retry));
2413                    // A break that does not happen (marker busy, recovery
2414                    // exhausted) is a wait like any other, so it meets the
2415                    // same deadline instead of spinning forever.
2416                    if !broken {
2417                        if started.elapsed() > TASK_LOCK_STALE {
2418                            bail!("could not lock task {id}");
2419                        }
2420                        std::thread::sleep(std::time::Duration::from_millis(15));
2421                    }
2422                }
2423                Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
2424            }
2425        }
2426    }
2427
2428    /// Load a task by id or unambiguous id prefix.
2429    pub fn get(&self, id: &str) -> Result<Task> {
2430        let resolved = self.resolve_id(id)?;
2431        read_path(&self.path_of(&resolved))
2432    }
2433
2434    /// Remove a task, and the claim lock that belongs to it.
2435    ///
2436    /// `in_flight` comes from the caller — a live daemon's heartbeat naming
2437    /// this task — because the task's own `running` status cannot answer the
2438    /// question. A daemon killed mid-competition leaves the status at
2439    /// `running` and an orphaned `.lock` behind, and a guard that trusted
2440    /// either would make the task undeletable for good: the phone showed
2441    /// exactly that, refusing a task whose daemon had been gone for an hour.
2442    ///
2443    /// So the lock is removed with the task rather than respected. Any lock
2444    /// still there once no live daemon claims the task is by definition stale,
2445    /// and leaving it would make a deleted task look claimed to
2446    /// [`Queue::claim`] and to whoever reads the directory.
2447    ///
2448    /// A deletion is deliberate, so whatever was `blocked` on the id is
2449    /// resolved, not held: a `<id>.removed` tombstone is written first (it
2450    /// is what tells [`missing_blockers`] and the daemon's resolver that this
2451    /// id was deleted on purpose, so a pass that races this call cannot
2452    /// mistake it for a vanished dependency), the record is deleted (the
2453    /// tombstone is taken back if that fails), and then each dependent is
2454    /// rewritten under its claim - see [`Queue::release_dependents_of`].
2455    /// Best-effort for the dependents: one claimed by something else right
2456    /// now, or whose write fails, is released silently by
2457    /// `crate::daemon::resolve_blockers` (or `crate::triage::run_once`) on
2458    /// its next pass, because the tombstone is still there. It does not fail
2459    /// this removal.
2460    ///
2461    /// `questions` is the store [`missing_blockers`] checks a `blocked_by` id
2462    /// against - the same store the caller already resolves `id`'s own home
2463    /// from, passed in rather than reopened here so a test queue at an
2464    /// explicit root never reads the operator's real questions directory.
2465    pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
2466        let resolved = self.resolve_id(id)?;
2467        if in_flight {
2468            bail!("task {resolved} is being run by a live daemon right now");
2469        }
2470        self.write_tombstone(&resolved)?;
2471        if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
2472        {
2473            // A record that is still there was not deleted: take the marker
2474            // back. One already gone keeps it, so the deletion still reads
2475            // as deliberate.
2476            if self.path_of(&resolved).exists() {
2477                let _ = std::fs::remove_file(self.tombstone_path(&resolved));
2478            }
2479            return Err(e);
2480        }
2481        let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
2482        Ok(Removal {
2483            id: resolved,
2484            released,
2485            still_blocked,
2486        })
2487    }
2488
2489    /// Path of the marker saying task `id` was deleted on purpose. Not a
2490    /// `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
2491    fn tombstone_path(&self, id: &str) -> PathBuf {
2492        self.root.join(format!("{id}.removed"))
2493    }
2494
2495    fn write_tombstone(&self, id: &str) -> Result<()> {
2496        let path = self.tombstone_path(id);
2497        let tmp = self.root.join(format!("{id}.removed.tmp"));
2498        std::fs::write(&tmp, b"")
2499            .and_then(|()| std::fs::rename(&tmp, &path))
2500            .with_context(|| format!("write {}", path.display()))
2501    }
2502
2503    /// Was `id` deleted through [`Queue::remove`] (and is its record gone)?
2504    /// A live record always wins: a tombstone never makes an existing task
2505    /// look deleted.
2506    fn was_deleted(&self, id: &str) -> bool {
2507        self.tombstone_path(id).is_file() && !self.path_of(id).exists()
2508    }
2509
2510    /// Apply [`Task::dependency_deleted`] to `task` for every `blocked_by` id
2511    /// that [`Queue::remove`] deleted. Returns those ids. The caller holds
2512    /// the task's claim and saves it.
2513    pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
2514        let deleted = deleted_blockers(self, questions, &task.blocked_by);
2515        deleted
2516            .into_iter()
2517            .filter(|id| task.dependency_deleted(id))
2518            .collect()
2519    }
2520
2521    /// Record, as a fixed-wording info notice, that `dependent` stopped
2522    /// waiting on the deleted task `deleted`. Best-effort and beside the
2523    /// queue, like the hold notice [`Queue::put`] files.
2524    pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
2525        let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
2526            return;
2527        };
2528        let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
2529        let waiting = dependent.status == TaskStatus::Blocked;
2530        let message = match (ja, waiting) {
2531            (true, false) => format!(
2532                "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
2533                dependent.short(),
2534                short(deleted)
2535            ),
2536            (true, true) => format!(
2537                "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
2538                dependent.short(),
2539                short(deleted)
2540            ),
2541            (false, false) => format!(
2542                "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
2543                dependent.short(),
2544                short(deleted)
2545            ),
2546            (false, true) => format!(
2547                "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
2548                dependent.short(),
2549                short(deleted)
2550            ),
2551        };
2552        crate::notices::raise_in(
2553            home,
2554            crate::notices::Notice::info(
2555                &format!("unblocked:{}:{}", dependent.id, deleted),
2556                message,
2557            )
2558            .link(crate::notices::Link::Task {
2559                id: dependent.id.clone(),
2560            }),
2561        );
2562    }
2563
2564    /// Resolve every `blocked` task naming `dependency` (just deleted) with
2565    /// [`Task::dependency_deleted`]. Returns the ids released to another
2566    /// status and those that remain `blocked` on something else. Each
2567    /// dependent is re-read inside its claim, never written from a stale copy.
2568    fn release_dependents_of(
2569        &self,
2570        dependency: &str,
2571        questions: &Questions,
2572    ) -> (Vec<String>, Vec<String>) {
2573        let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
2574        for listed in self.list() {
2575            if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
2576                matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
2577            }) {
2578                continue;
2579            }
2580            let Ok(_claim) = self.claim(&listed.id) else {
2581                continue;
2582            };
2583            let Ok(mut task) = self.get(&listed.id) else {
2584                continue;
2585            };
2586            let references: Vec<String> = task.blocked_by.iter().filter(|id| {
2587                matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
2588            }).cloned().collect();
2589            if references.is_empty() {
2590                continue;
2591            }
2592            for reference in references {
2593                task.dependency_deleted(&reference);
2594            }
2595            if self.put(&mut task).is_err() {
2596                continue;
2597            }
2598            self.note_dependency_deleted(&task, dependency);
2599            if task.status == TaskStatus::Blocked {
2600                still_blocked.push(task.id.clone());
2601            } else {
2602                released.push(task.id.clone());
2603            }
2604        }
2605        (released, still_blocked)
2606    }
2607
2608    /// The attachment-and-record half of [`Queue::remove`]. `remove_record` is
2609    /// how the record file is deleted; production passes `remove_file`, and a
2610    /// test passes a failing one to prove the attachments come back.
2611    fn remove_record_with_attachments(
2612        &self,
2613        resolved: &str,
2614        remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
2615    ) -> Result<()> {
2616        // Neither step may strand the other. The attachments are set aside by a
2617        // rename (cheap, and undone if the record cannot be removed), then the
2618        // record goes, and only then is the set-aside directory deleted for
2619        // real. A failure before the record is gone loses nothing; a failure
2620        // after it leaves at most an orphan `.removing` directory, which the
2621        // next removal sweeps.
2622        self.sweep_removed_attachments();
2623        let attachments = self.attachments_dir(resolved);
2624        let aside = self.root.join(format!("{resolved}.attachments.removing"));
2625        let moved = match std::fs::rename(&attachments, &aside) {
2626            Ok(()) => true,
2627            Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
2628            Err(e) => {
2629                return Err(e).with_context(|| format!("remove {}", attachments.display()));
2630            }
2631        };
2632        let path = self.path_of(resolved);
2633        if let Err(e) = remove_record(&path) {
2634            if moved {
2635                let _ = std::fs::rename(&aside, &attachments);
2636            }
2637            return Err(e).with_context(|| format!("remove {}", path.display()));
2638        }
2639        if moved {
2640            if let Err(e) = std::fs::remove_dir_all(&aside) {
2641                tracing::warn!("leftover attachments {}: {e}", aside.display());
2642            }
2643        }
2644        let lock = self.lock_path(resolved);
2645        if let Err(e) = std::fs::remove_file(&lock) {
2646            if e.kind() != std::io::ErrorKind::NotFound {
2647                return Err(e).with_context(|| format!("remove {}", lock.display()));
2648            }
2649        }
2650        Ok(())
2651    }
2652
2653    /// Delete `*.attachments.removing` directories a previous [`Queue::remove`]
2654    /// could not finish deleting. The task record is already gone by then, so
2655    /// the id no longer resolves and the removal cannot be retried by name;
2656    /// every later removal sweeps them instead. A directory whose record still
2657    /// exists belongs to a removal in progress and is left alone. Best-effort.
2658    fn sweep_removed_attachments(&self) {
2659        let Ok(entries) = std::fs::read_dir(&self.root) else {
2660            return;
2661        };
2662        for entry in entries.flatten() {
2663            let name = entry.file_name();
2664            let name = name.to_string_lossy();
2665            let Some(id) = name.strip_suffix(".attachments.removing") else {
2666                continue;
2667            };
2668            // A record still on disk means a removal is mid-flight (or about
2669            // to roll back): the directory is its to delete or restore.
2670            if !self.path_of(id).exists() {
2671                if let Err(e) = std::fs::remove_dir_all(entry.path()) {
2672                    tracing::warn!("leftover attachments {}: {e}", entry.path().display());
2673                }
2674            }
2675        }
2676        self.sweep_tombstones();
2677    }
2678
2679    /// Delete `*.removed` markers that are no longer needed: the record is
2680    /// gone, no task's `blocked_by` names the id, and the marker is older than
2681    /// [`TOMBSTONE_GRACE`]. The age is what makes this safe against a
2682    /// concurrent writer: the task list read here can be stale, so a dependent
2683    /// that was just blocked on an id (a conductor applying a decision while a
2684    /// removal is under way) would otherwise lose the marker and be held as a
2685    /// vanished dependency. A marker whose record still exists belongs to a
2686    /// removal in progress and is never touched.
2687    fn sweep_tombstones(&self) {
2688        let Ok(entries) = std::fs::read_dir(&self.root) else {
2689            return;
2690        };
2691        let tasks = self.list();
2692        for entry in entries.flatten() {
2693            let name = entry.file_name();
2694            let name = name.to_string_lossy();
2695            let Some(id) = name.strip_suffix(".removed") else {
2696                continue;
2697            };
2698            if self.path_of(id).exists()
2699                || tasks.iter().any(|t| {
2700                    t.blocked_by
2701                        .iter()
2702                        .any(|b| id.starts_with(b) || id.ends_with(b))
2703                })
2704            {
2705                continue;
2706            }
2707            let old = entry
2708                .metadata()
2709                .and_then(|m| m.modified())
2710                .ok()
2711                .and_then(|m| m.elapsed().ok())
2712                .is_some_and(|age| age >= TOMBSTONE_GRACE);
2713            if old {
2714                let _ = std::fs::remove_file(entry.path());
2715            }
2716        }
2717    }
2718
2719    /// Path of the claim lock for a task. One definition, so `claim` and
2720    /// `remove` cannot end up naming different files.
2721    fn lock_path(&self, id: &str) -> PathBuf {
2722        self.root.join(format!("{id}.lock"))
2723    }
2724
2725    /// Every task on disk, highest priority first and newest first within a
2726    /// priority. This is what `magi task list` and `GET /api/queue` print, so
2727    /// a raised priority has to move a task here the moment it is saved, not
2728    /// only in [`Queue::next_runnable`]'s own ordering - the operator reading
2729    /// the backlog and the loop about to drain it must agree on what "first"
2730    /// means. Every existing task defaults to priority 0, so this is a no-op
2731    /// change from the old newest-first order for a queue nobody has
2732    /// reprioritised.
2733    ///
2734    /// Unreadable files are skipped rather than fatal: one corrupt task must
2735    /// not take the queue - or the web UI, or an unattended daemon - down
2736    /// with it.
2737    pub fn list(&self) -> Vec<Task> {
2738        let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2739            .into_iter()
2740            .flatten()
2741            .flatten()
2742            .map(|e| e.path())
2743            .filter(|p| p.extension().is_some_and(|x| x == "json"))
2744            .filter_map(|p| read_path(&p).ok())
2745            .collect();
2746        tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2747        tasks
2748    }
2749
2750    /// Runs that a later attempt at the same task replaced, mapped to the id
2751    /// of the attempt that replaced them.
2752    ///
2753    /// A task keeps its attempts in order, and the deck showed them as two
2754    /// cards with the same title and no hint which was which: yukimemi asked
2755    /// why `stalled` and `blocked` appeared twice for one task, and the
2756    /// answer - "those are two tries, and the second one exists because of a
2757    /// bug since fixed" - was not on the screen anywhere.
2758    ///
2759    /// Read from the queue rather than stored on the run, because the
2760    /// ordering is the queue's fact: a `RunState` has no idea another attempt
2761    /// happened after it.
2762    pub fn superseded(&self) -> HashMap<String, String> {
2763        let mut by = HashMap::new();
2764        for task in self.list() {
2765            for earlier in &task.runs {
2766                if let Some(later) = task.successor_of(earlier) {
2767                    by.insert(earlier.clone(), later.clone());
2768                }
2769            }
2770        }
2771        by
2772    }
2773
2774    /// Whether `run` is an earlier attempt a later one replaced, and if so
2775    /// the id of that later attempt.
2776    ///
2777    /// Same walk as [`Queue::superseded`], narrowed to one run: a run detail
2778    /// page asks about exactly one run at a time, and this keeps that call
2779    /// site from building (and discarding) the whole map's `HashMap` just to
2780    /// read one entry out of it.
2781    pub fn superseded_by(&self, run: &str) -> Option<String> {
2782        for task in self.list() {
2783            if task.runs.iter().any(|r| r == run) {
2784                return task.successor_of(run).cloned();
2785            }
2786        }
2787        None
2788    }
2789
2790    /// The task's own most recent attempt, when `run` belongs to that task
2791    /// but is not already that attempt.
2792    ///
2793    /// Distinct from [`Queue::superseded_by`], which names only the very
2794    /// next attempt: a chain of retries (A superseded by B superseded by C)
2795    /// leaves an older run pointing at an intermediate one that may itself
2796    /// be unresolved, and a run's own detail page needs to know where the
2797    /// task's story currently stands - the chain's current head, C - not an
2798    /// attempt in the middle of it that a client would otherwise have to
2799    /// walk to by hand.
2800    pub fn latest_attempt(&self, run: &str) -> Option<String> {
2801        for task in self.list() {
2802            if task.runs.iter().any(|r| r == run) {
2803                return task.runs.last().filter(|last| **last != run).cloned();
2804            }
2805        }
2806        None
2807    }
2808
2809    /// The task a daemon should run next, or `None` when the queue is idle.
2810    ///
2811    /// Highest priority first, oldest first within a priority, so a burst of
2812    /// agent-filed work cannot starve the task a human filed this morning.
2813    pub fn next_runnable(&self) -> Option<Task> {
2814        let mut runnable: Vec<Task> = self
2815            .list()
2816            .into_iter()
2817            .filter(|t| t.status.runnable())
2818            .collect();
2819        runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2820        runnable.into_iter().next()
2821    }
2822
2823    /// Take exclusive ownership of a task.
2824    ///
2825    /// The lock is a `create_new` file next to the task, which is atomic on
2826    /// every platform magi targets. It exists so two daemons - or a daemon and
2827    /// a human running `magi run` - cannot drive one task into two competing
2828    /// runs. The returned guard releases on drop, including on panic.
2829    pub fn claim(&self, id: &str) -> Result<Claim> {
2830        std::fs::create_dir_all(&self.root)
2831            .with_context(|| format!("create {}", self.root.display()))?;
2832        let path = self.lock_path(id);
2833        let retry = crate::lock_file::Retry::new(None);
2834        match retry.create(&path) {
2835            Ok(mut f) => {
2836                use std::io::Write as _;
2837                // Best effort: the pid is for the human looking at a stale lock.
2838                let _ = writeln!(f, "{}", std::process::id());
2839                Ok(Claim { path })
2840            }
2841            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2842                bail!("task {id} is already claimed ({} exists)", path.display())
2843            }
2844            Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2845        }
2846    }
2847
2848    /// Expand an id prefix to exactly one task id.
2849    pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2850        if !prefix.is_empty()
2851            && prefix != "."
2852            && prefix != ".."
2853            && !prefix.contains(['/', '\\'])
2854            && self.path_of(prefix).is_file()
2855        {
2856            return Ok(prefix.to_owned());
2857        }
2858        let ids = record_ids(&self.root, "json")?;
2859        match resolve_reference(prefix, ids.iter().map(String::as_str)) {
2860            Reference::Found(id) => Ok(id),
2861            Reference::Missing => bail!("no task matches `{prefix}`"),
2862            Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
2863        }
2864    }
2865
2866    /// Change detection token for the queue.
2867    ///
2868    /// Combines file names and modification times of all task files in the
2869    /// queue, so adding, modifying, or deleting any task — even an older one —
2870    /// moves the revision and notifies connected clients via the change stream.
2871    /// Returns 0 when the queue is completely empty.
2872    pub fn revision(&self) -> u64 {
2873        self.revision_excluding(&std::collections::BTreeSet::new())
2874    }
2875
2876    /// [`Queue::revision`] ignoring the files of the named task ids, for a
2877    /// caller that tracks those tasks' content some other way.
2878    pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2879        use std::hash::{Hash as _, Hasher as _};
2880
2881        let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2882            .into_iter()
2883            .flatten()
2884            .flatten()
2885            .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2886            .filter(|e| {
2887                let path = e.path();
2888                !path
2889                    .file_stem()
2890                    .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2891            })
2892            .filter_map(|e| {
2893                let name = e.file_name().to_string_lossy().into_owned();
2894                let mtime = e
2895                    .metadata()
2896                    .ok()?
2897                    .modified()
2898                    .ok()?
2899                    .duration_since(std::time::UNIX_EPOCH)
2900                    .ok()?
2901                    .as_millis() as u64;
2902                Some((name, mtime))
2903            })
2904            .collect();
2905
2906        if entries.is_empty() {
2907            return 0;
2908        }
2909
2910        entries.sort_unstable();
2911        let mut hasher = std::hash::DefaultHasher::new();
2912        for (name, mtime) in &entries {
2913            name.hash(&mut hasher);
2914            mtime.hash(&mut hasher);
2915        }
2916        let h = hasher.finish();
2917        if h == 0 { 1 } else { h }
2918    }
2919}
2920
2921/// How long a deletion marker is kept after its record is gone, even with
2922/// nothing referencing it. See [`Queue::sweep_tombstones`].
2923const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2924
2925/// What [`Queue::remove`] did, beyond deleting the named task's own file.
2926#[derive(Debug, Clone)]
2927pub struct Removal {
2928    /// The id actually removed - `id` expanded from a prefix, if it was one.
2929    pub id: String,
2930    /// Every task that was `blocked` on [`Removal::id`] and left the block
2931    /// (back to the status it interrupted) because that was its only
2932    /// remaining dependency.
2933    pub released: Vec<String>,
2934    /// Every task that named [`Removal::id`] and is still `blocked` on
2935    /// another dependency.
2936    pub still_blocked: Vec<String>,
2937}
2938
2939/// Exclusive ownership of a task, released on drop.
2940#[derive(Debug)]
2941pub struct Claim {
2942    path: PathBuf,
2943}
2944
2945impl Drop for Claim {
2946    fn drop(&mut self) {
2947        let _ = std::fs::remove_file(&self.path);
2948    }
2949}
2950
2951/// The first line of a task, trimmed to a title. Used when the caller gives a
2952/// body but no title, which is the normal case for an agent piping a file in.
2953pub fn title_from(instruction: &str, max: usize) -> String {
2954    let Some(line) = first_line(instruction) else {
2955        return "(empty task)".to_owned();
2956    };
2957    if line.chars().count() <= max {
2958        return line.to_owned();
2959    }
2960    let head: String = line.chars().take(max.saturating_sub(1)).collect();
2961    format!("{head}…")
2962}
2963
2964/// The first non-blank line, whatever it is. A markdown heading is the
2965/// task's own summary - agents pipe in `# Rework the config loader` and mean
2966/// exactly that - so it is preferred over the prose beneath it rather than
2967/// skipped as decoration. Leading list and heading markers are stripped
2968/// because they are syntax, not words. `None` when nothing is left.
2969pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2970    let line = instruction
2971        .lines()
2972        .map(str::trim)
2973        .find(|l| !l.is_empty())?
2974        .trim_start_matches(['#', '-', '*', '>', ' '])
2975        .trim();
2976    (!line.is_empty()).then_some(line)
2977}
2978
2979fn read_path(path: &Path) -> Result<Task> {
2980    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2981    let task: Task =
2982        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2983    // Greater-than, not not-equal: every field added since schema 1 carries
2984    // `#[serde(default)]`, so an older task has nothing to say about it and
2985    // defaulting is exactly as good a reading as a value that build never had
2986    // a chance to write. Only a schema *ahead* of this build - a meaning it
2987    // cannot possibly know - is refused rather than guessed at.
2988    if task.schema > SCHEMA {
2989        bail!(
2990            "task {} was written by a different magi (schema {}, this build \
2991             speaks {SCHEMA})",
2992            task.id,
2993            task.schema
2994        );
2995    }
2996    Ok(task)
2997}
2998
2999/// Resolution shared by queue lookups and dependency inventories. Exact ids
3000/// win; otherwise both the CLI prefix and displayed suffix are supported.
3001#[derive(Debug, PartialEq, Eq)]
3002pub(crate) enum Reference {
3003    Found(String),
3004    Missing,
3005    Ambiguous,
3006}
3007
3008pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
3009    let mut hits = Vec::new();
3010    for candidate in ids {
3011        if candidate == id {
3012            return Reference::Found(candidate.to_owned());
3013        }
3014        if candidate.starts_with(id) || candidate.ends_with(id) {
3015            hits.push(candidate);
3016        }
3017    }
3018    match hits.as_slice() {
3019        [] => Reference::Missing,
3020        [one] => Reference::Found((*one).to_owned()),
3021        _ => Reference::Ambiguous,
3022    }
3023}
3024
3025/// Read names, not deserialized records: list() intentionally skips corrupt
3026/// and future-schema records, which still exist and must keep their waiters.
3027pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
3028    let entries = match std::fs::read_dir(root) {
3029        Ok(entries) => entries,
3030        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
3031        Err(e) => return Err(e.into()),
3032    };
3033    let mut ids = Vec::new();
3034    for entry in entries {
3035        let path = entry?.path();
3036        if path.extension().is_some_and(|e| e == extension)
3037            && let Some(id) = path.file_stem().and_then(|s| s.to_str())
3038        {
3039            ids.push(id.to_owned());
3040        }
3041    }
3042    Ok(ids)
3043}
3044
3045/// Unknown and ambiguous dependencies stay waiting; neither is evidence that
3046/// the operator deleted a prerequisite or needs to supply its replacement.
3047#[derive(Debug, PartialEq, Eq)]
3048pub enum Dependency {
3049    /// Existing task record, with its canonical id; its contents may be unreadable.
3050    Task(String),
3051    /// Existing question record, with its canonical id.
3052    Question(String),
3053    /// Explicitly removed task, identified by its tombstone.
3054    Deleted(String),
3055    /// No matching record or tombstone after a successful inventory read.
3056    Missing,
3057    /// More than one record matches the reference.
3058    Ambiguous,
3059    /// The inventory cannot currently be read, so existence is unknown.
3060    Unknown,
3061}
3062
3063/// Resolve against authoritative task, question and deletion record names.
3064pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
3065    let (Ok(tasks), Ok(qs), Ok(removed)) = (
3066        record_ids(queue.root(), "json"),
3067        record_ids(questions.root(), "json"),
3068        record_ids(queue.root(), "removed"),
3069    ) else {
3070        return Dependency::Unknown;
3071    };
3072    // Include tombstones in ambiguity checks: a reused suffix must never
3073    // release a waiter on a different, still-live task.
3074    let mut ids = tasks
3075        .iter()
3076        .chain(qs.iter())
3077        .chain(removed.iter())
3078        .cloned()
3079        .collect::<Vec<_>>();
3080    ids.sort();
3081    ids.dedup();
3082    match resolve_reference(id, ids.iter().map(String::as_str)) {
3083        Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
3084        Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
3085        Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
3086        Reference::Found(_) => Dependency::Unknown,
3087        Reference::Missing => Dependency::Missing,
3088        Reference::Ambiguous => Dependency::Ambiguous,
3089    }
3090}
3091
3092/// Canonicalize only proven dependencies, preserving unknown legacy references.
3093pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
3094    for id in ids {
3095        if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
3096            dependency(queue, questions, id)
3097        {
3098            *id = full;
3099        }
3100    }
3101}
3102
3103/// Live queue facts for decision and deputy prompts. Historical hold reasons
3104/// and question text are claims, never an inventory of existing prerequisites.
3105pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
3106    let inventory = crate::blockers::Inventory::load(queue, questions);
3107    let mut text = String::from(
3108        "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
3109    );
3110    match record_ids(queue.root(), "json") {
3111        Ok(ids) => {
3112            for id in ids {
3113                match queue.get(&id) {
3114                    Ok(task) => text.push_str(&format!(
3115                        "{}: {}; blocked_by={:?}; chain={:?}\n",
3116                        task.id,
3117                        task.status.as_str(),
3118                        task.blocked_by,
3119                        inventory.waits_on(&task)
3120                    )),
3121                    Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
3122                }
3123            }
3124        }
3125        Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
3126    }
3127    text
3128}
3129
3130/// Only a successful directory scan with no matching record proves absence.
3131pub fn missing_blockers(
3132    queue: &Queue,
3133    questions: &Questions,
3134    blocked_by: &[String],
3135) -> Vec<String> {
3136    blocked_by
3137        .iter()
3138        .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
3139        .cloned()
3140        .collect()
3141}
3142
3143/// Return the original references, so legacy short ids are removed by the
3144/// same Task transition as full ids.
3145pub fn deleted_blockers(
3146    queue: &Queue,
3147    questions: &Questions,
3148    blocked_by: &[String],
3149) -> Vec<String> {
3150    blocked_by
3151        .iter()
3152        .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
3153        .cloned()
3154        .collect()
3155}
3156
3157/// The `hold_reason` text for a task quarantined because one or more of its
3158/// `blocked_by` ids no longer exist. Shared by `crate::daemon::resolve_blockers`,
3159/// `crate::triage::run_once`, and [`Queue::remove`]'s own dependent
3160/// quarantine, so the three call sites read as the same event to an operator
3161/// looking at `magi task show` rather than three different wordings for it.
3162///
3163/// Names the full original `blocked_by` list, not just `missing` - a task
3164/// quarantined here can also have named a dependency that was still
3165/// perfectly valid, and [`Task::hold_machine`] clears `blocked_by` on the way
3166/// in, so this text is the only place that information survives for an
3167/// operator deciding whether to release the task outright.
3168pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
3169    missing_blocker_hold_reason_in(blocked_by, missing, "en")
3170}
3171
3172/// [`missing_blocker_hold_reason`] in `language` (Japanese, else English).
3173pub fn missing_blocker_hold_reason_in(
3174    blocked_by: &[String],
3175    missing: &[String],
3176    language: &str,
3177) -> String {
3178    if crate::lang::is_japanese(language) {
3179        format!(
3180            "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
3181            blocked_by.join(", "),
3182            missing.join(", "),
3183        )
3184    } else {
3185        format!(
3186            "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
3187            blocked_by.join(", "),
3188            missing.join(", "),
3189        )
3190    }
3191}
3192
3193/// The last dash-separated part of an id, the form people say aloud.
3194pub fn short(id: &str) -> &str {
3195    id.split('-').next_back().unwrap_or(id)
3196}
3197
3198fn new_id() -> String {
3199    let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
3200    let seed = crate::rng::entropy();
3201    format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
3202}
3203
3204#[cfg(test)]
3205mod tests {
3206    #[test]
3207    fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
3208        let t = Task::new(
3209            "retry".into(),
3210            "do it".into(),
3211            PathBuf::from("."),
3212            Source::Human,
3213        );
3214        let mut value = serde_json::to_value(&t).unwrap();
3215        for field in ["resumption", "resumption_history", "operator_decided_at"] {
3216            value.as_object_mut().unwrap().remove(field);
3217        }
3218        value["schema"] = serde_json::json!(15);
3219        let mut legacy: Task = serde_json::from_value(value).unwrap();
3220        legacy.hold_machine(Some("automatic retry".into()));
3221        legacy.release();
3222        assert!(legacy.resumption.is_none());
3223        assert!(legacy.resumption_history.is_empty());
3224        assert!(legacy.operator_decided_at.is_none());
3225        legacy.release_operator("cli", None);
3226        legacy.start("run".into());
3227        let reloaded: Task =
3228            serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
3229        assert!(reloaded.resume_override.is_none());
3230        assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
3231    }
3232
3233    #[test]
3234    fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
3235        let mut base = RunOverrides {
3236            merge: Some("pr".to_owned()),
3237            candidates: Some(3),
3238            ..RunOverrides::default()
3239        };
3240        base.merge_over(&RunOverrides {
3241            merge: Some("none".to_owned()),
3242            seed: Some(7),
3243            ..RunOverrides::default()
3244        });
3245        assert_eq!(base.merge.as_deref(), Some("none"));
3246        assert_eq!(base.candidates, Some(3));
3247        assert_eq!(base.seed, Some(7));
3248    }
3249
3250    #[test]
3251    fn an_already_landed_task_is_done_with_its_attempt_refunded() {
3252        let mut t = task("relanded");
3253        t.attempts = 1;
3254        t.status = TaskStatus::Running;
3255        t.already_landed("already in main as 0e368de");
3256        assert_eq!(t.status, TaskStatus::Done);
3257        assert_eq!(t.attempts, 0);
3258        assert!(t.hold_reason.is_none());
3259        assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
3260    }
3261
3262    #[test]
3263    fn missing_blocker_reason_follows_the_language() {
3264        let b = vec!["a".to_owned()];
3265        let en = missing_blocker_hold_reason_in(&b, &b, "en");
3266        assert_eq!(en, missing_blocker_hold_reason(&b, &b));
3267        assert!(en.starts_with("blocked on a"));
3268        assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
3269        assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
3270    }
3271
3272    use super::*;
3273
3274    #[test]
3275    fn triage_applied_survives_release_and_old_records_read_as_empty() {
3276        let mut t = Task::new(
3277            "t".to_owned(),
3278            "i".to_owned(),
3279            PathBuf::from("r"),
3280            Source::Human,
3281        );
3282        t.mark_triage_applied("q1");
3283        t.mark_triage_applied("q1");
3284        t.hold_machine(Some("x".to_owned()));
3285        t.release();
3286        assert_eq!(t.triage_applied, ["q1"]);
3287        assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
3288
3289        let mut v = serde_json::to_value(&t).unwrap();
3290        v.as_object_mut().unwrap().remove("triage_applied");
3291        let old: Task = serde_json::from_value(v).unwrap();
3292        assert!(old.triage_applied.is_empty());
3293    }
3294
3295    #[test]
3296    fn task_counts_of_empty_is_all_zero() {
3297        assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
3298    }
3299
3300    #[test]
3301    fn task_counts_of_tallies_every_status() {
3302        let mut queued = Task::new(
3303            "q".to_owned(),
3304            "i".to_owned(),
3305            PathBuf::from("."),
3306            Source::Human,
3307        );
3308        queued.status = TaskStatus::Queued;
3309        let mut running = queued.clone();
3310        running.status = TaskStatus::Running;
3311        let mut done = queued.clone();
3312        done.status = TaskStatus::Done;
3313        let mut failed = queued.clone();
3314        failed.status = TaskStatus::Failed;
3315        let mut held = queued.clone();
3316        held.status = TaskStatus::Held;
3317        let mut blocked = queued.clone();
3318        blocked.status = TaskStatus::Blocked;
3319
3320        let mut parked = queued.clone();
3321        parked.status = TaskStatus::Parked;
3322
3323        let counts = TaskCounts::of(&[
3324            queued,
3325            running,
3326            done.clone(),
3327            done,
3328            failed,
3329            held,
3330            blocked,
3331            parked,
3332        ]);
3333        assert_eq!(
3334            counts,
3335            TaskCounts {
3336                queued: 1,
3337                running: 1,
3338                done: 2,
3339                failed: 1,
3340                held: 1,
3341                blocked: 1,
3342                parked: 1,
3343            }
3344        );
3345    }
3346
3347    #[test]
3348    fn task_implementers_persist_normalized_without_losing_history() {
3349        let (_dir, q) = queue();
3350        let mut t = task("keep me");
3351        t.hold_manual(Some("waiting".into()));
3352        t.runs.push("earlier-run".into());
3353        t.attachments.push("reference.txt".into());
3354        let cfg: crate::config::Config =
3355            toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
3356                .unwrap();
3357        q.put(&mut t).unwrap();
3358        let original = serde_json::to_value(&t).unwrap();
3359        for n in [2, 1, 2] {
3360            t.set_implementers(n, &cfg).unwrap();
3361            q.put(&mut t).unwrap();
3362            t = q.get(&t.id).unwrap();
3363            assert_eq!(t.implementers(), Some(usize::from(n)));
3364            assert_eq!(t.solo, n == 1);
3365            if n == 1 {
3366                assert!(t.overrides.is_none());
3367            }
3368            let mut actual = serde_json::to_value(&t).unwrap();
3369            for field in ["solo", "overrides", "updated_at"] {
3370                actual[field] = original[field].clone();
3371            }
3372            assert_eq!(actual, original);
3373        }
3374        t.overrides.as_mut().unwrap().judges = Some(2);
3375        t.set_implementers(1, &cfg).unwrap();
3376        assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
3377        assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
3378    }
3379
3380    #[test]
3381    fn task_implementers_old_records_keep_inheritance_until_changed() {
3382        let mut value = serde_json::to_value(task("old task")).unwrap();
3383        value.as_object_mut().unwrap().remove("solo");
3384        value.as_object_mut().unwrap().remove("overrides");
3385        let mut old: Task = serde_json::from_value(value).unwrap();
3386        assert_eq!(old.implementers(), None);
3387        old.set_implementers(1, &crate::config::Config::default())
3388            .unwrap();
3389        assert!(old.solo);
3390        assert!(old.overrides.is_none());
3391    }
3392
3393    #[test]
3394    fn task_implementers_reject_invalid_counts_and_states() {
3395        let cfg = crate::config::Config::default();
3396        let mut t = task("unchanged");
3397        for n in [0, 4, 255] {
3398            assert!(t.set_implementers(n, &cfg).is_err());
3399        }
3400        for status in [
3401            TaskStatus::Running,
3402            TaskStatus::Done,
3403            TaskStatus::Failed,
3404            TaskStatus::Blocked,
3405            TaskStatus::Parked,
3406        ] {
3407            t.status = status;
3408            let before = serde_json::to_value(&t).unwrap();
3409            assert!(t.set_implementers(1, &cfg).is_err());
3410            assert_eq!(serde_json::to_value(&t).unwrap(), before);
3411        }
3412    }
3413
3414    /// A queue of its own, with no process-global state - which is the point of
3415    /// `Queue::at`, and why these can run in parallel.
3416    fn queue() -> (tempfile::TempDir, Queue) {
3417        let dir = tempfile::tempdir().unwrap();
3418        let q = Queue::at(dir.path().join("queue"));
3419        (dir, q)
3420    }
3421
3422    #[test]
3423    fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
3424        let dir = tempfile::tempdir().unwrap();
3425        let q = Queue::at(dir.path().join("queue"));
3426        let mut t = task("held");
3427        q.put(&mut t).unwrap();
3428        assert_eq!(
3429            crate::notices::Notices::at(dir.path().join("notifications"))
3430                .list()
3431                .len(),
3432            0
3433        );
3434        t.hold_machine(Some("out of attempts".to_owned()));
3435        q.put(&mut t).unwrap();
3436        let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
3437        assert_eq!(listed.len(), 1);
3438        assert!(listed[0].message.contains("out of attempts"));
3439    }
3440
3441    fn task(title: &str) -> Task {
3442        Task::new(
3443            title.to_owned(),
3444            format!("do {title}"),
3445            PathBuf::from("."),
3446            Source::Human,
3447        )
3448    }
3449
3450    #[test]
3451    fn attempt_limit_inherits_old_records_and_controls_failure() {
3452        let mut t = task("budget");
3453        let mut old = serde_json::to_value(&t).unwrap();
3454        old.as_object_mut().unwrap().remove("attempt_limit");
3455        old["schema"] = serde_json::json!(15);
3456        let legacy: Task = serde_json::from_value(old).unwrap();
3457        assert_eq!(legacy.attempt_limit, None);
3458        assert_eq!(legacy.effective_attempt_limit(2), 2);
3459        t.attempts = 2;
3460        t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
3461        t.fail("retry", 1);
3462        assert_eq!(t.status, TaskStatus::Failed);
3463        t.attempts = 3;
3464        t.fail("stop", 10);
3465        assert_eq!(t.status, TaskStatus::Held);
3466        t.change_attempt_limit(None).unwrap();
3467        assert_eq!(t.status, TaskStatus::Held);
3468        assert_eq!(t.attempts, 3);
3469        t.fail("inherit", 4);
3470        assert_eq!(t.status, TaskStatus::Failed);
3471        for status in [
3472            TaskStatus::Running,
3473            TaskStatus::Done,
3474            TaskStatus::Failed,
3475            TaskStatus::Blocked,
3476            TaskStatus::Parked,
3477        ] {
3478            t.status = status;
3479            assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
3480            assert_eq!(t.attempt_limit, None);
3481        }
3482    }
3483
3484    #[test]
3485    fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
3486        let mut t = task("retried");
3487        assert!(
3488            t.earlier_attempts().is_empty(),
3489            "a first attempt takes nothing over"
3490        );
3491        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3492        assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
3493        // What the web note shows and what a takeover acts on are one rule.
3494        assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
3495        assert_eq!(t.successor_of("bbbb"), None);
3496        assert_eq!(t.successor_of("zzzz"), None);
3497    }
3498
3499    #[test]
3500    fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
3501        let (_dir, q) = queue();
3502        let mut t = task("resumed");
3503        for (runs, successor) in [
3504            (vec!["aaaa", "aaaa"], None),
3505            (vec!["aaaa", "bbbb", "aaaa"], None),
3506            (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
3507            (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
3508        ] {
3509            t.runs = runs.into_iter().map(str::to_owned).collect();
3510            q.put(&mut t).unwrap();
3511
3512            assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
3513            assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
3514            assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
3515        }
3516    }
3517
3518    #[test]
3519    fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
3520        let (_dir, q) = queue();
3521        let mut t = task("retried");
3522        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3523        q.put(&mut t).unwrap();
3524
3525        assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
3526        assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
3527        assert_eq!(
3528            q.superseded_by("cccc"),
3529            None,
3530            "the latest attempt replaces nothing"
3531        );
3532        assert_eq!(
3533            q.superseded_by("never-heard-of-it"),
3534            None,
3535            "a run belonging to no task on this queue is not superseded"
3536        );
3537
3538        let mut by = HashMap::new();
3539        by.insert("aaaa".to_owned(), "bbbb".to_owned());
3540        by.insert("bbbb".to_owned(), "cccc".to_owned());
3541        assert_eq!(
3542            q.superseded(),
3543            by,
3544            "the whole-map and single-run forms must agree"
3545        );
3546    }
3547
3548    #[test]
3549    fn superseded_attempts_is_empty_until_the_task_is_done() {
3550        let mut t = task("retried");
3551        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3552        t.status = TaskStatus::Failed;
3553        assert_eq!(
3554            t.superseded_attempts(true),
3555            &[] as &[String],
3556            "a task still retrying has no attempt yet that a later one made moot"
3557        );
3558
3559        t.status = TaskStatus::Running;
3560        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3561    }
3562
3563    #[test]
3564    fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
3565        let mut t = task("retried");
3566        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3567        t.status = TaskStatus::Done;
3568        assert_eq!(
3569            t.superseded_attempts(true),
3570            &["aaaa".to_owned(), "bbbb".to_owned()],
3571            "cccc is the attempt whose success made the task done, and stays out"
3572        );
3573    }
3574
3575    #[test]
3576    fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
3577        let mut t = task("first try landed");
3578        t.runs = vec!["aaaa".to_owned()];
3579        t.status = TaskStatus::Done;
3580        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
3581    }
3582
3583    #[test]
3584    fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
3585        // `succeed` is reachable directly - `magi task done`, its web
3586        // equivalent, and the conductor's `Recovery::Done` - on a task in
3587        // any status, including one whose last recorded attempt is itself
3588        // `Blocked`/`Failed`/anything but `Merged`/`Ready`. `runs.last()`
3589        // alone cannot tell that apart from the loop's own settle path, so
3590        // the caller's own read of that run's status is what decides this.
3591        let mut t = task("closed by hand after a manual merge");
3592        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
3593        t.status = TaskStatus::Done;
3594        assert_eq!(
3595            t.superseded_attempts(false),
3596            &[] as &[String],
3597            "nothing here is provably why the task is done, so nothing is superseded"
3598        );
3599    }
3600
3601    #[test]
3602    fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
3603        let (_dir, q) = queue();
3604        let mut t = task("retried twice");
3605        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
3606        q.put(&mut t).unwrap();
3607
3608        assert_eq!(
3609            q.latest_attempt("aaaa"),
3610            Some("cccc".to_owned()),
3611            "an old attempt points straight at the chain's current head, not the \
3612             next attempt in the middle of it"
3613        );
3614        assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
3615        assert_eq!(
3616            q.latest_attempt("cccc"),
3617            None,
3618            "the latest attempt is not superseded by anything"
3619        );
3620        assert_eq!(
3621            q.latest_attempt("never-heard-of-it"),
3622            None,
3623            "a run belonging to no task on this queue is not superseded"
3624        );
3625    }
3626
3627    #[test]
3628    fn a_markdown_heading_is_the_title_not_decoration() {
3629        // A task file's heading is the summary its author already wrote, so it
3630        // beats the prose underneath. Getting this backwards was visible in the
3631        // first smoke test: a task titled "# Rework the config loader" listed
3632        // as "It re-reads the file on every lookup".
3633        assert_eq!(
3634            title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
3635            "Rework the config loader"
3636        );
3637        assert_eq!(title_from("- fix the thing", 40), "fix the thing");
3638        assert_eq!(title_from("> quoted task", 40), "quoted task");
3639        // Nothing usable at all still has to produce something printable.
3640        assert_eq!(title_from("   \n\n", 40), "(empty task)");
3641        assert_eq!(title_from("###\n", 40), "(empty task)");
3642    }
3643
3644    #[test]
3645    fn a_long_title_is_elided_by_characters_not_bytes() {
3646        // Byte truncation would split a multi-byte character and panic.
3647        let long = "課題".repeat(30);
3648        let title = title_from(&long, 10);
3649        assert_eq!(title.chars().count(), 10);
3650        assert!(title.ends_with('…'));
3651    }
3652
3653    #[test]
3654    fn priority_wins_and_ties_break_oldest_first() {
3655        let (_dir, q) = queue();
3656        let mut a = task("first");
3657        let mut b = task("second");
3658        let mut c = task("urgent");
3659        // Ids carry a timestamp, so force a known order.
3660        a.id = "20260101-000001-aaaa".to_owned();
3661        b.id = "20260101-000002-bbbb".to_owned();
3662        c.id = "20260101-000003-cccc".to_owned();
3663        c.priority = 5;
3664        for t in [&mut a, &mut b, &mut c] {
3665            q.put(t).unwrap();
3666        }
3667
3668        // Priority first...
3669        assert_eq!(q.next_runnable().unwrap().id, c.id);
3670        c.hold_machine(None);
3671        q.put(&mut c).unwrap();
3672        // ...then oldest, so a burst of new work cannot starve older work.
3673        assert_eq!(q.next_runnable().unwrap().id, a.id);
3674        assert_eq!(q.list().len(), 3, "b is still waiting its turn");
3675    }
3676
3677    #[test]
3678    fn a_blocked_task_never_starves_another_runnable_one() {
3679        let (_dir, q) = queue();
3680        let mut blocked = task("blocked");
3681        blocked.block(vec!["something".to_owned()], None);
3682        q.put(&mut blocked).unwrap();
3683
3684        let mut runnable = task("free to go");
3685        q.put(&mut runnable).unwrap();
3686
3687        let next = q.next_runnable().expect("a runnable task is still offered");
3688        assert_eq!(next.id, runnable.id);
3689    }
3690
3691    #[test]
3692    fn a_held_task_is_never_offered_to_the_loop() {
3693        let (_dir, q) = queue();
3694        let mut t = task("held");
3695        q.put(&mut t).unwrap();
3696        assert!(q.next_runnable().is_some());
3697
3698        t.hold_machine(None);
3699        q.put(&mut t).unwrap();
3700        assert!(
3701            q.next_runnable().is_none(),
3702            "a held task must wait for a human"
3703        );
3704
3705        // A failed task, by contrast, is exactly what the loop should retry.
3706        t.status = TaskStatus::Failed;
3707        q.put(&mut t).unwrap();
3708        assert!(q.next_runnable().is_some());
3709    }
3710
3711    #[test]
3712    fn attempts_are_capped_and_then_the_task_is_held() {
3713        let mut t = task("doomed");
3714
3715        t.start("run-1".to_owned());
3716        t.fail("gate red", 2);
3717        assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
3718
3719        t.start("run-2".to_owned());
3720        t.fail("gate red", 2);
3721        assert_eq!(
3722            t.status,
3723            TaskStatus::Held,
3724            "out of attempts: stop spending money on it"
3725        );
3726        assert_eq!(t.runs, ["run-1", "run-2"]);
3727        assert_eq!(t.last_error.as_deref(), Some("gate red"));
3728        assert_eq!(
3729            t.hold_reason.as_deref(),
3730            Some("gate red"),
3731            "the hold must say why, not leave hold_reason null next to a \
3732             populated last_error"
3733        );
3734    }
3735
3736    #[test]
3737    fn handing_off_a_task_records_a_hold_reason_too() {
3738        let mut t = task("left a pull request");
3739        t.start("run-1".to_owned());
3740        t.handed_off("run ended with a pull request open [run run-1]");
3741        assert_eq!(t.status, TaskStatus::Held);
3742        assert_eq!(t.hold_source, Some(HoldSource::Machine));
3743        assert_eq!(
3744            t.hold_reason.as_deref(),
3745            Some("run ended with a pull request open [run run-1]")
3746        );
3747        assert_eq!(t.hold_reason, t.last_error);
3748    }
3749
3750    #[test]
3751    fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
3752        let mut t = task("stalled by quota");
3753
3754        t.start("run-1".to_owned());
3755        assert_eq!(t.attempts, 1);
3756        t.stall("judge-1, judge-2 out of quota");
3757        assert_eq!(
3758            t.attempts, 0,
3759            "a closed quota window must not spend the task's retry budget"
3760        );
3761        assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
3762        assert_eq!(
3763            t.last_error.as_deref(),
3764            Some("judge-1, judge-2 out of quota")
3765        );
3766
3767        // A task can therefore stall all night and still get its real attempts
3768        // once the quota resets - which is the whole point.
3769        for _ in 0..20 {
3770            t.start("run-n".to_owned());
3771            t.stall("still out of quota");
3772        }
3773        t.start("run-real".to_owned());
3774        t.fail("gate red", 2);
3775        assert_eq!(
3776            t.status,
3777            TaskStatus::Failed,
3778            "the first attempt that was really judged is attempt one"
3779        );
3780    }
3781
3782    #[test]
3783    fn releasing_a_held_task_gives_it_a_real_second_chance() {
3784        let mut t = task("retry me");
3785        t.start("run-1".to_owned());
3786        t.fail("gate red", 1);
3787        assert_eq!(t.status, TaskStatus::Held);
3788
3789        t.release();
3790        assert_eq!(t.status, TaskStatus::Queued);
3791        // Without resetting attempts the next failure would re-hold at once,
3792        // and a release would be a no-op the operator cannot see.
3793        assert_eq!(t.attempts, 0);
3794        assert!(t.last_error.is_none());
3795        assert_eq!(
3796            t.runs.len(),
3797            1,
3798            "history is kept: attempts reset, evidence does not"
3799        );
3800    }
3801
3802    #[test]
3803    fn a_hold_reason_survives_and_a_release_clears_it() {
3804        let mut t = task("waiting on something else");
3805        t.hold_manual(Some(
3806            "waiting for 20260101-000000-aaaa to land first".to_owned(),
3807        ));
3808        assert_eq!(t.status, TaskStatus::Held);
3809        assert_eq!(
3810            t.hold_reason.as_deref(),
3811            Some("waiting for 20260101-000000-aaaa to land first")
3812        );
3813
3814        // Holding again with no reason must not erase the one already there.
3815        t.hold_manual(None);
3816        assert_eq!(
3817            t.hold_reason.as_deref(),
3818            Some("waiting for 20260101-000000-aaaa to land first"),
3819            "a bare re-hold keeps whatever a human already wrote down"
3820        );
3821
3822        // A hold with no reason at all is still an ordinary, allowed hold.
3823        let mut plain = task("no reason given");
3824        plain.hold_manual(None);
3825        assert_eq!(plain.status, TaskStatus::Held);
3826        assert!(plain.hold_reason.is_none());
3827
3828        t.release();
3829        assert_eq!(t.status, TaskStatus::Queued);
3830        assert!(
3831            t.hold_reason.is_none(),
3832            "a stale reason must not greet the next person who holds this task"
3833        );
3834    }
3835
3836    #[test]
3837    fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
3838        // `done` can close a held task directly - neither `magi task done`
3839        // nor `POST /api/queue/{id}/done` requires a release first - so a
3840        // task held for "waiting on 3ed9" and then closed without ever being
3841        // released must not still read as waiting on it afterwards.
3842        let mut t = task("landed by hand while held");
3843        t.hold_manual(Some("waiting on 3ed9".to_owned()));
3844        assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
3845
3846        t.succeed();
3847        assert_eq!(t.status, TaskStatus::Done);
3848        assert!(
3849            t.hold_reason.is_none(),
3850            "a done task cannot still be waiting on something"
3851        );
3852    }
3853
3854    #[test]
3855    fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
3856        // The web UI's "Hold" and "Mark done" buttons are both reachable on a
3857        // `blocked` task, not just on `queued`/`held` ones - neither requires
3858        // a release first. A task moved off `Blocked` that way must not still
3859        // carry the dependency it was waiting on: a dependency graph built
3860        // from `blocked_by` would otherwise keep drawing an edge for a task
3861        // that is not blocked on anything any more.
3862        let mut held = task("held straight out of blocked");
3863        held.block(
3864            vec!["20260101-000000-dead".to_owned()],
3865            Some("waiting on the migration script".to_owned()),
3866        );
3867        assert_eq!(held.status, TaskStatus::Blocked);
3868
3869        held.hold_manual(None);
3870        assert_eq!(held.status, TaskStatus::Held);
3871        assert!(
3872            held.blocked_by.is_empty(),
3873            "hold overrides the wait, same as release"
3874        );
3875        assert!(held.block_reason.is_none());
3876
3877        let mut done = task("closed straight out of blocked");
3878        done.block(
3879            vec!["20260101-000000-dead".to_owned()],
3880            Some("waiting on the migration script".to_owned()),
3881        );
3882        done.succeed();
3883        assert_eq!(done.status, TaskStatus::Done);
3884        assert!(
3885            done.blocked_by.is_empty(),
3886            "a done task cannot still be waiting on a dependency"
3887        );
3888        assert!(done.block_reason.is_none());
3889    }
3890
3891    #[test]
3892    fn a_blocked_task_is_never_offered_to_the_loop() {
3893        let mut t = task("blocked");
3894        assert!(t.status.runnable());
3895        t.block(
3896            vec!["dep-id".to_owned()],
3897            Some("waits on dep-id".to_owned()),
3898        );
3899        assert_eq!(t.status, TaskStatus::Blocked);
3900        assert!(!t.status.runnable());
3901        assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
3902    }
3903
3904    #[test]
3905    fn unblocking_the_last_dependency_returns_the_task_to_queued() {
3906        let mut t = task("blocked on two");
3907        t.block(
3908            vec!["a".to_owned(), "b".to_owned()],
3909            Some("waits on a and b".to_owned()),
3910        );
3911
3912        t.unblock("a");
3913        assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
3914        assert_eq!(t.blocked_by, ["b"]);
3915
3916        t.unblock("b");
3917        assert_eq!(t.status, TaskStatus::Queued);
3918        assert!(t.blocked_by.is_empty());
3919        assert!(t.block_reason.is_none());
3920    }
3921
3922    #[test]
3923    fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
3924        let mut t = task("never blocked");
3925        t.unblock("whatever");
3926        assert_eq!(t.status, TaskStatus::Queued);
3927    }
3928
3929    #[test]
3930    fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
3931        // The bug this guards: a task an operator (or `crate::triage`) has
3932        // deliberately held, once `crate::conduct` blocks it on a follow-up
3933        // question, must not silently re-enter the competition queue the
3934        // moment that question is answered - whatever the answer said.
3935        let mut t = task("held, then asked about");
3936        t.hold_machine(Some("out of attempts".to_owned()));
3937        assert_eq!(t.status, TaskStatus::Held);
3938
3939        t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3940        assert_eq!(t.status, TaskStatus::Blocked);
3941
3942        t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3943        t.unblock("q1");
3944        assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3945        assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3946        assert_eq!(t.hold_source, Some(HoldSource::Machine));
3947        assert!(t.blocked_from.is_none(), "consumed once restored");
3948    }
3949
3950    #[test]
3951    fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3952        let mut t = task("manually held, then asked about");
3953        t.hold_manual(Some("waiting on a dependency".to_owned()));
3954
3955        t.block(vec!["q1".to_owned()], None);
3956        t.unblock("q1");
3957
3958        assert_eq!(t.status, TaskStatus::Held);
3959        assert_eq!(t.hold_source, Some(HoldSource::Manual));
3960    }
3961
3962    #[test]
3963    fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3964        // A second `Task::block` call - `crate::conduct` adding a question on
3965        // top of an existing block - must not overwrite `blocked_from` with
3966        // `Blocked` itself, or the task would restore into itself.
3967        let mut t = task("held, blocked twice");
3968        t.hold_machine(None);
3969        t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3970        t.block(
3971            vec!["q1".to_owned(), "q2".to_owned()],
3972            Some("second".to_owned()),
3973        );
3974
3975        t.unblock("q1");
3976        assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3977        t.unblock("q2");
3978        assert_eq!(t.status, TaskStatus::Held);
3979    }
3980
3981    #[test]
3982    fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3983        // Whatever process was driving the run is gone by the time a
3984        // conductor's question about it gets answered - there is nothing left
3985        // to resume into.
3986        let mut t = task("blocked mid-run");
3987        t.start("run-1".to_owned());
3988        assert_eq!(t.status, TaskStatus::Running);
3989
3990        t.block(vec!["q1".to_owned()], None);
3991        t.unblock("q1");
3992        assert_eq!(t.status, TaskStatus::Queued);
3993    }
3994
3995    #[test]
3996    fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3997        // `blocked_from` is `None` for a record written before schema 4 (or,
3998        // equivalently, deserialized straight from an on-disk file that never
3999        // had the field). Held evidence surviving on the task - never cleared
4000        // by `block` - is the only way left to tell such a record apart from
4001        // one blocked straight out of `Queued`.
4002        let mut t = task("legacy record, held before it was blocked");
4003        t.hold_source = Some(HoldSource::Machine);
4004        t.hold_reason = Some("legacy hold reason".to_owned());
4005        t.status = TaskStatus::Blocked;
4006        t.blocked_by = vec!["q1".to_owned()];
4007        t.blocked_from = None;
4008
4009        t.unblock("q1");
4010        assert_eq!(t.status, TaskStatus::Held);
4011    }
4012
4013    #[test]
4014    fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
4015        let mut t = task("legacy record, ordinary dependency block");
4016        t.status = TaskStatus::Blocked;
4017        t.blocked_by = vec!["dep".to_owned()];
4018        t.blocked_from = None;
4019
4020        t.unblock("dep");
4021        assert_eq!(t.status, TaskStatus::Queued);
4022    }
4023
4024    #[test]
4025    fn answering_a_question_is_recorded_and_survives_a_release() {
4026        let mut t = task("asked something");
4027        t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
4028        t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
4029        t.unblock("q1");
4030        assert_eq!(t.status, TaskStatus::Queued);
4031        assert_eq!(t.answers.len(), 1);
4032        assert_eq!(t.answers[0].answer, "SQLite");
4033
4034        // A release resets attempts, not evidence - the same rule
4035        // `releasing_a_held_task_gives_it_a_real_second_chance` asserts for
4036        // `runs`.
4037        t.release();
4038        assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
4039    }
4040
4041    #[test]
4042    fn resumption_review_branch_requires_unchanged_authored_criteria() {
4043        for source in ["cli", "web"] {
4044            let mut t = task("review old candidate");
4045            t.request_review("candidate-A".into());
4046            t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
4047            let mut unchanged = t.clone();
4048            unchanged.release_operator(source, None);
4049            assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
4050            t.edit(
4051                "review approved head".into(),
4052                "Reuse measured PR head B; fix Clippy; review".into(),
4053            )
4054            .unwrap();
4055            t.release_operator(source, None);
4056            assert!(t.review_branch.is_none());
4057            assert!(t.review_instruction.is_none());
4058            // Legacy records capture their old criteria at the supported edit.
4059            let mut legacy = unchanged;
4060            legacy.review_instruction = None;
4061            legacy.hold_for_handover(None, "still checked out".into());
4062            legacy
4063                .edit("new plan".into(), "Reuse measured PR head B".into())
4064                .unwrap();
4065            legacy.release_operator(source, None);
4066            assert!(legacy.review_branch.is_none());
4067        }
4068    }
4069
4070    #[test]
4071    fn a_refused_handover_keeps_the_review_branch_across_release() {
4072        let mut t = task("refused takeover");
4073        t.start("run-1".to_owned());
4074        t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
4075        assert_eq!(t.status, TaskStatus::Held);
4076        assert_eq!(t.attempts, 1);
4077        t.release();
4078        assert_eq!(t.status, TaskStatus::Queued);
4079        assert_eq!(t.attempts, 0);
4080        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4081
4082        // A fresh competition chosen afterwards is not a review.
4083        t.hold_for_handover(None, "again".to_owned());
4084        t.requeue();
4085        assert!(t.review_branch.is_none());
4086
4087        // A manual hold never keeps one.
4088        let mut m = task("manual");
4089        m.review_branch = Some("magi/x/A".to_owned());
4090        m.hold_manual(None);
4091        m.release();
4092        assert!(m.review_branch.is_none());
4093    }
4094
4095    #[test]
4096    fn requesting_review_requeues_the_task_and_remembers_the_branch() {
4097        let mut t = task("blocked run with a surviving branch");
4098        t.start("run-1".to_owned());
4099        t.fail("blocked with major findings", 5);
4100        assert_eq!(t.status, TaskStatus::Failed);
4101
4102        t.request_review("magi/eba2/A".to_owned());
4103        assert_eq!(t.status, TaskStatus::Queued);
4104        assert_eq!(t.attempts, 0);
4105        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
4106
4107        // An ordinary release (a human overriding the choice) drops it again.
4108        t.release();
4109        assert!(t.review_branch.is_none());
4110    }
4111
4112    #[test]
4113    fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
4114        let mut t = task("retry");
4115        t.start("run-1".to_owned());
4116        t.requeue();
4117        assert!(t.fresh_start);
4118
4119        t.release();
4120        assert!(!t.fresh_start);
4121    }
4122
4123    #[test]
4124    fn urgent_only_changes_queued_or_held_and_preserves_settings() {
4125        for status in [
4126            TaskStatus::Queued,
4127            TaskStatus::Held,
4128            TaskStatus::Running,
4129            TaskStatus::Done,
4130            TaskStatus::Failed,
4131            TaskStatus::Blocked,
4132            TaskStatus::Parked,
4133        ] {
4134            let mut t = task("urgent settings");
4135            t.status = status;
4136            t.attempts = 2;
4137            t.hold_reason = Some("waiting".into());
4138            t.priority = 7;
4139            t.interrupt = true;
4140            let before = serde_json::to_value(&t).unwrap();
4141            for urgent in [true, true, false, false] {
4142                let result = t.set_urgent(urgent);
4143                let mut expected = before.clone();
4144                if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
4145                    result.unwrap();
4146                    expected["urgent"] = urgent.into();
4147                } else {
4148                    let error = result.unwrap_err().to_string();
4149                    assert!(error.contains(status.as_str()), "{error}");
4150                    assert!(error.contains("queued or held"), "{error}");
4151                }
4152                assert_eq!(serde_json::to_value(&t).unwrap(), expected);
4153            }
4154        }
4155    }
4156
4157    #[test]
4158    fn priority_can_be_changed_while_queued_but_not_while_running() {
4159        let mut t = task("reprioritise me");
4160        t.set_priority(5).unwrap();
4161        assert_eq!(t.priority, 5);
4162
4163        t.start("run-1".to_owned());
4164        let err = t.set_priority(9).unwrap_err().to_string();
4165        assert!(err.contains("running"), "{err}");
4166        assert_eq!(t.priority, 5, "the rejected write must not partially apply");
4167    }
4168
4169    #[test]
4170    fn interrupt_can_be_marked_while_queued_but_not_while_running() {
4171        let mut t = task("interrupt me");
4172        assert!(!t.interrupt, "off unless asked, same as any other task");
4173
4174        t.set_interrupt(true).unwrap();
4175        assert!(t.interrupt);
4176
4177        t.start("run-1".to_owned());
4178        assert!(
4179            !t.interrupt,
4180            "the mark is one-shot: dispatching the task fulfils it, \
4181             whatever the run that follows ends up doing"
4182        );
4183        let err = t.set_interrupt(true).unwrap_err().to_string();
4184        assert!(err.contains("running"), "{err}");
4185        // Clearing is always allowed, even on a running task - there is
4186        // nothing left for it to interrupt once it has been claimed.
4187        t.set_interrupt(false).unwrap();
4188        assert!(!t.interrupt);
4189    }
4190
4191    /// R2-1-1: a task whose run fails and requeues must not go on
4192    /// re-triggering `crate::daemon`'s interrupt scheduler on every later
4193    /// boundary, attempt after attempt, until it exhausts its budget.
4194    #[test]
4195    fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
4196        let mut t = task("interrupt me");
4197        t.set_interrupt(true).unwrap();
4198        t.start("run-1".to_owned());
4199        t.fail("mock failure", 5);
4200        assert_eq!(t.status, TaskStatus::Failed);
4201        assert!(
4202            !t.interrupt,
4203            "one attempt already spent the mark; a retry is an ordinary \
4204             requeue, not a fresh interrupt request"
4205        );
4206    }
4207
4208    #[test]
4209    fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
4210        let (_dir, q) = queue();
4211        let mut a = task("first filed");
4212        let mut b = task("second filed");
4213        a.id = "20260101-000001-aaaa".to_owned();
4214        b.id = "20260101-000002-bbbb".to_owned();
4215        q.put(&mut a).unwrap();
4216        q.put(&mut b).unwrap();
4217
4218        assert_eq!(
4219            q.next_runnable().unwrap().id,
4220            a.id,
4221            "with equal priority the older task goes first, so a burst of \
4222             new work cannot starve it"
4223        );
4224        assert_eq!(
4225            q.list()[0].id,
4226            b.id,
4227            "but the list an operator reads is newest first, the same as \
4228             before priority existed - a's turn to run does not make it the \
4229             newest task"
4230        );
4231
4232        let mut a = q.get(&a.id).unwrap();
4233        a.set_priority(10).unwrap();
4234        q.put(&mut a).unwrap();
4235
4236        assert_eq!(
4237            q.next_runnable().unwrap().id,
4238            a.id,
4239            "a raised priority must be reflected the moment it is saved"
4240        );
4241        // `magi task list` and `GET /api/queue` both print `Queue::list()`
4242        // directly, so the raised task has to lead there too - not only in
4243        // what the loop would claim next.
4244        assert_eq!(
4245            q.list()[0].id,
4246            a.id,
4247            "the raised task must sort first in the list an operator reads, \
4248             not only in next_runnable's own ordering"
4249        );
4250    }
4251
4252    #[test]
4253    fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
4254        let mut t = Task::new(
4255            "old title".to_owned(),
4256            "old instruction".to_owned(),
4257            PathBuf::from("/repo"),
4258            Source::Agent {
4259                run: "20260101-000000-beef".to_owned(),
4260                node: "implement".to_owned(),
4261            },
4262        );
4263        let id = t.id.clone();
4264        let created_at = t.created_at;
4265        t.runs.push("20260101-000000-beef".to_owned());
4266
4267        t.edit("new title".to_owned(), "new instruction".to_owned())
4268            .unwrap();
4269
4270        assert_eq!(t.title, "new title");
4271        assert_eq!(t.instruction, "new instruction");
4272        assert_eq!(t.id, id, "editing must not mint a new id");
4273        assert_eq!(t.created_at, created_at);
4274        assert_eq!(
4275            t.source,
4276            Source::Agent {
4277                run: "20260101-000000-beef".to_owned(),
4278                node: "implement".to_owned(),
4279            },
4280            "editing must not turn agent attribution into human"
4281        );
4282        assert_eq!(t.runs, ["20260101-000000-beef"]);
4283    }
4284
4285    #[test]
4286    fn editing_is_refused_once_a_task_is_running_or_finished() {
4287        let mut running = task("in flight");
4288        running.start("run-1".to_owned());
4289        let err = running
4290            .edit("x".to_owned(), "y".to_owned())
4291            .unwrap_err()
4292            .to_string();
4293        assert!(err.contains("running"), "{err}");
4294
4295        let mut done = task("finished");
4296        done.succeed();
4297        let err = done
4298            .edit("x".to_owned(), "y".to_owned())
4299            .unwrap_err()
4300            .to_string();
4301        assert!(err.contains("done"), "{err}");
4302
4303        // Both queued and held are the point of the feature and must work.
4304        let mut queued = task("waiting");
4305        queued.edit("x".to_owned(), "y".to_owned()).unwrap();
4306        let mut held = task("parked");
4307        held.hold_machine(None);
4308        held.edit("x".to_owned(), "y".to_owned()).unwrap();
4309    }
4310
4311    #[test]
4312    fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
4313        let (_dir, q) = queue();
4314        let path = q.path_of("20260101-000000-aaaa");
4315        std::fs::create_dir_all(q.root()).unwrap();
4316        std::fs::write(
4317            &path,
4318            serde_json::json!({
4319                "schema": SCHEMA,
4320                "id": "20260101-000000-aaaa",
4321                "title": "from before hold reasons existed",
4322                "instruction": "from before hold reasons existed",
4323                "repo": ".",
4324                "source": { "kind": "human" },
4325                "status": "held",
4326                "created_at": Timestamp::now().to_string(),
4327                "updated_at": Timestamp::now().to_string(),
4328            })
4329            .to_string(),
4330        )
4331        .unwrap();
4332
4333        let task = q.get("20260101-000000-aaaa").expect("must still read");
4334        assert!(task.hold_reason.is_none());
4335        assert!(task.operator_held());
4336    }
4337
4338    #[test]
4339    fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
4340        let (_dir, q) = queue();
4341        let path = q.path_of("20260101-000000-bbbb");
4342        std::fs::create_dir_all(q.root()).unwrap();
4343        std::fs::write(
4344            &path,
4345            serde_json::json!({
4346                "schema": 2,
4347                "id": "20260101-000000-bbbb",
4348                "title": "old manual recovery",
4349                "instruction": "old manual recovery",
4350                "repo": ".",
4351                "source": { "kind": "human" },
4352                "status": "held",
4353                "hold_reason": "active manual recovery run20260912-224242-daf5",
4354                "created_at": Timestamp::now().to_string(),
4355                "updated_at": Timestamp::now().to_string(),
4356            })
4357            .to_string(),
4358        )
4359        .unwrap();
4360
4361        let task = q.get("20260101-000000-bbbb").expect("must still read");
4362        assert_eq!(task.hold_source, None);
4363        assert!(task.operator_held());
4364    }
4365
4366    #[test]
4367    fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
4368        let (_dir, q) = queue();
4369        let path = q.path_of("20260101-000000-aaaa");
4370        std::fs::create_dir_all(q.root()).unwrap();
4371        std::fs::write(
4372            &path,
4373            serde_json::json!({
4374                "schema": SCHEMA,
4375                "id": "20260101-000000-aaaa",
4376                "title": "from before diagnostics existed",
4377                "instruction": "from before diagnostics existed",
4378                "repo": ".",
4379                "source": { "kind": "human" },
4380                "status": "held",
4381                "created_at": Timestamp::now().to_string(),
4382                "updated_at": Timestamp::now().to_string(),
4383            })
4384            .to_string(),
4385        )
4386        .unwrap();
4387
4388        let task = q.get("20260101-000000-aaaa").expect("must still read");
4389        assert!(task.diagnostic.is_none());
4390    }
4391
4392    #[test]
4393    fn a_schema_1_task_with_no_blocking_fields_still_reads() {
4394        // Written by a build that predates `blocked_by`, `block_reason`,
4395        // `answers` and `review_branch` entirely - literal `"schema": 1`,
4396        // not `SCHEMA`, since the whole point is a build older than this one.
4397        let (_dir, q) = queue();
4398        let path = q.path_of("20260101-000000-aaaa");
4399        std::fs::create_dir_all(q.root()).unwrap();
4400        std::fs::write(
4401            &path,
4402            serde_json::json!({
4403                "schema": 1,
4404                "id": "20260101-000000-aaaa",
4405                "title": "from before blocking existed",
4406                "instruction": "from before blocking existed",
4407                "repo": ".",
4408                "source": { "kind": "human" },
4409                "status": "queued",
4410                "created_at": Timestamp::now().to_string(),
4411                "updated_at": Timestamp::now().to_string(),
4412            })
4413            .to_string(),
4414        )
4415        .unwrap();
4416
4417        let task = q.get("20260101-000000-aaaa").expect("must still read");
4418        assert!(task.blocked_by.is_empty());
4419        assert!(task.block_reason.is_none());
4420        assert!(task.answers.is_empty());
4421        assert!(task.review_branch.is_none());
4422    }
4423
4424    #[test]
4425    fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
4426        // A diagnostic belongs to the run that produced it. Left in place
4427        // across a release, an unrelated later failure - a config error, say -
4428        // would go on showing evidence for a problem that is no longer why the
4429        // task is stuck.
4430        let mut held = task("diagnosed");
4431        held.start("run-1".to_owned());
4432        held.fail("gate red", 1);
4433        held.diagnostic = Some("cargo test failed: ...".to_owned());
4434        assert_eq!(held.status, TaskStatus::Held);
4435
4436        held.release();
4437        assert!(held.diagnostic.is_none());
4438
4439        held.diagnostic = Some("cargo test failed: ...".to_owned());
4440        held.succeed();
4441        assert!(held.diagnostic.is_none());
4442    }
4443
4444    #[test]
4445    fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
4446        let mut t = task("retried");
4447        t.start("run-1".to_owned());
4448        t.diagnostic = Some("stale evidence from a previous hold".to_owned());
4449        t.fail("unrelated config error", 5);
4450        assert_eq!(t.status, TaskStatus::Failed);
4451        assert!(
4452            t.diagnostic.is_none(),
4453            "fail() must not let an old diagnostic outlive the run that produced it"
4454        );
4455    }
4456
4457    #[test]
4458    fn a_claim_is_exclusive_and_releases_on_drop() {
4459        let (_dir, q) = queue();
4460        let mut t = task("contended");
4461        q.put(&mut t).unwrap();
4462
4463        let held = q.claim(&t.id).unwrap();
4464        assert!(
4465            q.claim(&t.id).is_err(),
4466            "two daemons must not drive one task into two runs"
4467        );
4468        drop(held);
4469        assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
4470    }
4471
4472    #[test]
4473    fn a_round_trip_survives_disk() {
4474        let (_dir, q) = queue();
4475        let mut t = Task::new(
4476            "titled".to_owned(),
4477            "body".to_owned(),
4478            PathBuf::from("/repo"),
4479            Source::Agent {
4480                run: "20260101-000000-beef".to_owned(),
4481                node: "implement".to_owned(),
4482            },
4483        );
4484        t.priority = 3;
4485        q.put(&mut t).unwrap();
4486
4487        let back = q.get(&t.id).unwrap();
4488        assert_eq!(back.id, t.id);
4489        assert_eq!(back.priority, 3);
4490        assert_eq!(back.source.label(), "implement@beef");
4491        // A prefix is enough, the way run ids work everywhere else.
4492        assert_eq!(q.get(t.short()).unwrap().id, t.id);
4493    }
4494
4495    #[test]
4496    fn an_unreadable_task_does_not_take_the_queue_down() {
4497        let (_dir, q) = queue();
4498        let mut t = task("fine");
4499        q.put(&mut t).unwrap();
4500        std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
4501
4502        let listed = q.list();
4503        assert_eq!(listed.len(), 1, "the readable task still lists");
4504        assert_eq!(listed[0].id, t.id);
4505    }
4506
4507    #[test]
4508    fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
4509        let (_dir, q) = queue();
4510        let path = q.path_of("20260101-000000-aaaa");
4511        std::fs::create_dir_all(q.root()).unwrap();
4512        std::fs::write(
4513            &path,
4514            serde_json::json!({
4515                "schema": SCHEMA,
4516                "id": "20260101-000000-aaaa",
4517                "title": "from before solo existed",
4518                "instruction": "from before solo existed",
4519                "repo": ".",
4520                "source": { "kind": "human" },
4521                "status": "queued",
4522                "created_at": Timestamp::now().to_string(),
4523                "updated_at": Timestamp::now().to_string(),
4524            })
4525            .to_string(),
4526        )
4527        .unwrap();
4528
4529        let task = q.get("20260101-000000-aaaa").expect("must still read");
4530        assert!(!task.solo, "a queue file with no `solo` field means false");
4531    }
4532
4533    #[test]
4534    fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
4535        let (_dir, q) = queue();
4536        let path = q.path_of("20260101-000000-bbbb");
4537        std::fs::create_dir_all(q.root()).unwrap();
4538        std::fs::write(
4539            &path,
4540            serde_json::json!({
4541                "schema": SCHEMA,
4542                "id": "20260101-000000-bbbb",
4543                "title": "from before urgent existed",
4544                "instruction": "from before urgent existed",
4545                "repo": ".",
4546                "source": { "kind": "human" },
4547                "status": "queued",
4548                "created_at": Timestamp::now().to_string(),
4549                "updated_at": Timestamp::now().to_string(),
4550            })
4551            .to_string(),
4552        )
4553        .unwrap();
4554
4555        let task = q.get("20260101-000000-bbbb").expect("must still read");
4556        assert!(
4557            !task.urgent,
4558            "a queue file with no `urgent` field means false, same as `solo`"
4559        );
4560    }
4561
4562    #[test]
4563    fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
4564        let (_dir, q) = queue();
4565        let mut t = task("from the future");
4566        q.put(&mut t).unwrap();
4567        let path = q.path_of(&t.id);
4568        let body = std::fs::read_to_string(&path)
4569            .unwrap()
4570            .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
4571        std::fs::write(&path, body).unwrap();
4572
4573        let err = q.get(&t.id).unwrap_err().to_string();
4574        assert!(err.contains("schema 99"), "{err}");
4575    }
4576
4577    #[test]
4578    fn revision_moves_when_the_queue_changes() {
4579        let (_dir, q) = queue();
4580        assert_eq!(q.revision(), 0, "an empty queue has no revision");
4581        let mut t = task("first");
4582        q.put(&mut t).unwrap();
4583        assert!(q.revision() > 0, "a written task moves the revision");
4584    }
4585
4586    #[test]
4587    fn revision_moves_when_deleting_an_older_task() {
4588        let (dir, q) = queue();
4589        let questions = Questions::at(dir.path().join("questions"));
4590        let mut t1 = task("older");
4591        q.put(&mut t1).unwrap();
4592        // Ensure mtime ticks forward.
4593        std::thread::sleep(std::time::Duration::from_millis(10));
4594        let mut t2 = task("newer");
4595        q.put(&mut t2).unwrap();
4596
4597        let rev_before = q.revision();
4598        q.remove(&t1.id, false, &questions).unwrap();
4599        let rev_after = q.revision();
4600
4601        assert_ne!(
4602            rev_before, rev_after,
4603            "deleting an older task must change the revision so other clients see the deletion"
4604        );
4605    }
4606
4607    #[test]
4608    fn removing_a_task_takes_it_out_of_the_listing() {
4609        let (dir, q) = queue();
4610        let questions = Questions::at(dir.path().join("questions"));
4611        let mut t = task("delete me");
4612        q.put(&mut t).unwrap();
4613        let removed = q.remove(t.short(), false, &questions).unwrap();
4614        assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
4615        assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
4616        assert!(q.list().is_empty());
4617        assert!(
4618            q.remove(&t.id, false, &questions).is_err(),
4619            "removing twice is an error"
4620        );
4621    }
4622
4623    #[test]
4624    fn removing_a_task_takes_its_stale_lock_with_it() {
4625        let (dir, q) = queue();
4626        let questions = Questions::at(dir.path().join("questions"));
4627        let mut t = task("interrupted");
4628        q.put(&mut t).unwrap();
4629
4630        // A daemon killed mid-run leaves this behind. Nothing holds it: the
4631        // process that would have dropped the guard is gone.
4632        let claim = q.claim(&t.id).unwrap();
4633        std::mem::forget(claim);
4634        assert!(
4635            q.claim(&t.id).is_err(),
4636            "the orphaned lock is what makes the task look claimed"
4637        );
4638
4639        // A live daemon on this task is refused, whatever the lock says.
4640        let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
4641        assert!(err.contains("live daemon"), "{err}");
4642        assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
4643
4644        // With no daemon behind it, the lock is stale and goes with the task.
4645        q.remove(&t.id, false, &questions).unwrap();
4646        assert!(q.list().is_empty());
4647        let mut again = task("interrupted");
4648        again.id = t.id.clone();
4649        q.put(&mut again).unwrap();
4650        assert!(
4651            q.claim(&t.id).is_ok(),
4652            "a task that comes back must be claimable, which a left-behind lock would prevent"
4653        );
4654    }
4655
4656    fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
4657        crate::notices::Notices::at(dir.join("notifications")).list()
4658    }
4659
4660    #[test]
4661    fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
4662        let (dir, q) = queue();
4663        let questions = Questions::at(dir.path().join("questions"));
4664        let mut dep = task("dependency");
4665        q.put(&mut dep).unwrap();
4666        let mut blocked = task("waiting");
4667        blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
4668        q.put(&mut blocked).unwrap();
4669
4670        let removed = q.remove(&dep.id, false, &questions).unwrap();
4671        assert_eq!(removed.released, [blocked.id.clone()]);
4672        assert!(removed.still_blocked.is_empty());
4673
4674        let after = q.get(&blocked.id).unwrap();
4675        assert_eq!(after.status, TaskStatus::Queued);
4676        assert!(after.blocked_by.is_empty());
4677        assert!(after.block_reason.is_none());
4678        let notes = notices_of(dir.path());
4679        assert_eq!(notes.len(), 1, "{notes:?}");
4680        assert_eq!(notes[0].severity, crate::notices::Severity::Info);
4681        assert_eq!(
4682            notes[0].link,
4683            Some(crate::notices::Link::Task {
4684                id: blocked.id.clone(),
4685            })
4686        );
4687    }
4688
4689    #[test]
4690    fn removing_one_of_two_dependencies_keeps_the_other() {
4691        let (dir, q) = queue();
4692        let questions = Questions::at(dir.path().join("questions"));
4693        let mut dep = task("dependency");
4694        q.put(&mut dep).unwrap();
4695        let mut other = task("other");
4696        q.put(&mut other).unwrap();
4697        let mut blocked = task("waiting");
4698        blocked.block(
4699            vec![dep.id.clone(), other.id.clone()],
4700            Some("waits on both".to_owned()),
4701        );
4702        q.put(&mut blocked).unwrap();
4703
4704        let removed = q.remove(&dep.id, false, &questions).unwrap();
4705        assert!(removed.released.is_empty());
4706        assert_eq!(removed.still_blocked, [blocked.id.clone()]);
4707
4708        let after = q.get(&blocked.id).unwrap();
4709        assert_eq!(after.status, TaskStatus::Blocked);
4710        assert_eq!(after.blocked_by, [other.id.clone()]);
4711        // The tombstone stays while a task still names the id... only this
4712        // one's dependency is gone, so nothing references it and it may go.
4713        assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
4714    }
4715
4716    #[test]
4717    fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
4718        let (dir, q) = queue();
4719        let questions = Questions::at(dir.path().join("questions"));
4720        let mut dep = task("dependency");
4721        q.put(&mut dep).unwrap();
4722        let mut blocked = task("waiting");
4723        blocked.block(vec![dep.id.clone()], None);
4724        q.put(&mut blocked).unwrap();
4725
4726        // The dependent is claimed elsewhere, so `remove` has to skip it.
4727        let claim = q.claim(&blocked.id).unwrap();
4728        let removed = q.remove(&dep.id, false, &questions).unwrap();
4729        assert!(removed.released.is_empty());
4730        drop(claim);
4731
4732        let mut task = q.get(&blocked.id).unwrap();
4733        assert_eq!(task.status, TaskStatus::Blocked);
4734        assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
4735        assert_eq!(
4736            q.apply_deleted_blockers(&mut task, &questions),
4737            [dep.id.clone()]
4738        );
4739        assert_eq!(task.status, TaskStatus::Queued);
4740    }
4741
4742    #[test]
4743    fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
4744        let (dir, q) = queue();
4745        let questions = Questions::at(dir.path().join("questions"));
4746        let mut dep = task("dependency");
4747        dep.id = "20261013-193611-ca5b".into();
4748        q.put(&mut dep).unwrap();
4749        let mut waiting = task("waiting");
4750        waiting.block(vec!["ca5b".into()], None);
4751        q.put(&mut waiting).unwrap();
4752        q.remove(&dep.id, false, &questions).unwrap();
4753        assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
4754        assert_eq!(
4755            dependency(&q, &questions, "ca5b"),
4756            Dependency::Deleted(dep.id)
4757        );
4758    }
4759
4760    #[test]
4761    fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
4762        let (dir, queue) = queue();
4763        let questions = Questions::at(dir.path().join("questions"));
4764        let mut dep = task("dependency");
4765        dep.id = "20261013-193611-ca5b".into();
4766        queue.put(&mut dep).unwrap();
4767        std::fs::create_dir_all(questions.root()).unwrap();
4768        let question_id = "20261014-193611-ca5b";
4769        std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
4770        assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
4771        assert_eq!(
4772            dependency(&queue, &questions, "ca5b"),
4773            Dependency::Ambiguous
4774        );
4775        assert_eq!(
4776            dependency(&queue, &questions, &dep.id),
4777            Dependency::Task(dep.id)
4778        );
4779        assert_eq!(
4780            dependency(&queue, &questions, question_id),
4781            Dependency::Question(question_id.into())
4782        );
4783        assert_eq!(
4784            dependency(&queue, &questions, "missing"),
4785            Dependency::Missing
4786        );
4787    }
4788
4789    #[test]
4790    fn dependency_resolution_directory_errors_are_unknown_not_missing() {
4791        let dir = tempfile::tempdir().unwrap();
4792        let q = Queue::at(dir.path().join("not-a-directory"));
4793        std::fs::write(q.root(), "file").unwrap();
4794        let questions = Questions::at(dir.path().join("questions"));
4795        assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
4796        assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
4797    }
4798
4799    #[test]
4800    fn a_dependency_without_a_tombstone_is_still_missing() {
4801        let (dir, q) = queue();
4802        let questions = Questions::at(dir.path().join("questions"));
4803        let mut dep = task("dependency");
4804        q.put(&mut dep).unwrap();
4805        std::fs::remove_file(q.path_of(&dep.id)).unwrap();
4806        assert_eq!(
4807            missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
4808            [dep.id.clone()]
4809        );
4810        assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
4811    }
4812
4813    #[test]
4814    fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
4815        let mut t = task("held");
4816        t.hold_machine(Some("because".to_owned()));
4817        t.blocked_by = vec!["gone".to_owned()];
4818        assert!(!t.dependency_deleted("gone"));
4819        assert_eq!(t.status, TaskStatus::Held);
4820    }
4821
4822    #[test]
4823    fn a_failed_record_removal_takes_the_tombstone_back() {
4824        let (_dir, q) = queue();
4825        let mut t = task("stays");
4826        q.put(&mut t).unwrap();
4827        q.write_tombstone(&t.id).unwrap();
4828        let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
4829        assert!(err.is_err());
4830        // `remove` is what withdraws the marker; a live record is never
4831        // treated as deleted whatever is on disk.
4832        assert!(!q.was_deleted(&t.id));
4833    }
4834
4835    fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
4836        let p = dir.join(name);
4837        std::fs::write(&p, body).unwrap();
4838        p
4839    }
4840
4841    #[test]
4842    fn an_attachment_copy_survives_deleting_its_source() {
4843        let (dir, q) = queue();
4844        let src = source_file(dir.path(), "shot.png", "pixels");
4845        let mut t = task("with a picture");
4846        let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
4847        q.put(&mut t).unwrap();
4848        std::fs::remove_file(&src).unwrap();
4849        assert_eq!(names, ["shot.png"]);
4850        let loaded = q.get(&t.id).unwrap();
4851        let paths = q.attachment_paths(&loaded);
4852        assert_eq!(paths.len(), 1);
4853        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
4854    }
4855
4856    #[test]
4857    fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
4858        let (dir, q) = queue();
4859        let mut t = task("bad names");
4860        for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
4861            let src = source_file(dir.path(), name, "x");
4862            assert!(
4863                q.attach(&mut t, &[src]).is_err(),
4864                "`{name}` must be refused"
4865            );
4866        }
4867        // A drive-qualified name cannot exist as a file on Windows (joining it
4868        // to a directory yields a drive-relative path to `foo.png` instead),
4869        // so the rule is asserted on the name itself, everywhere.
4870        assert!(!crate::ask::valid_asset_name("C:foo.png"));
4871        #[cfg(not(windows))]
4872        {
4873            let src = source_file(dir.path(), "C:foo.png", "x");
4874            assert!(q.attach(&mut t, &[src]).is_err());
4875        }
4876        let long = format!("{}.png", "a".repeat(70));
4877        let src = source_file(dir.path(), &long, "x");
4878        assert!(q.attach(&mut t, &[src]).is_err());
4879        assert!(t.attachments.is_empty());
4880        assert!(!q.attachments_dir(&t.id).exists());
4881    }
4882
4883    #[test]
4884    fn a_taken_attachment_name_is_numbered_not_overwritten() {
4885        let (dir, q) = queue();
4886        let a = source_file(dir.path(), "shot.png", "one");
4887        let sub = dir.path().join("other");
4888        std::fs::create_dir_all(&sub).unwrap();
4889        let b = source_file(&sub, "shot.png", "two");
4890        let mut t = task("collision");
4891        q.attach(&mut t, &[a]).unwrap();
4892        q.attach(&mut t, &[b]).unwrap();
4893        assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
4894        let paths = q.attachment_paths(&t);
4895        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
4896        assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
4897    }
4898
4899    #[test]
4900    fn a_renumbered_name_stays_inside_the_length_bound() {
4901        let (dir, q) = queue();
4902        let name = format!("{}.png", "a".repeat(60));
4903        assert_eq!(name.len(), 64);
4904        let a = source_file(dir.path(), &name, "one");
4905        let sub = dir.path().join("other");
4906        std::fs::create_dir_all(&sub).unwrap();
4907        let b = source_file(&sub, &name, "two");
4908        let mut t = task("long");
4909        q.attach(&mut t, &[a, b]).unwrap();
4910        assert_eq!(t.attachments.len(), 2);
4911        assert!(
4912            t.attachments
4913                .iter()
4914                .all(|n| crate::ask::valid_asset_name(n))
4915        );
4916        assert!(t.attachments[1].ends_with("-2.png"));
4917    }
4918
4919    #[test]
4920    fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
4921        let (dir, q) = queue();
4922        let good = source_file(dir.path(), "good.png", "ok");
4923        let mut t = task("partial");
4924        q.attach(&mut t, &[good]).unwrap();
4925        let more = source_file(dir.path(), "more.png", "ok");
4926        let missing = dir.path().join("missing.png");
4927        assert!(q.attach(&mut t, &[more, missing]).is_err());
4928        assert_eq!(t.attachments, ["good.png"]);
4929        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4930            .unwrap()
4931            .flatten()
4932            .collect();
4933        assert_eq!(on_disk.len(), 1);
4934    }
4935
4936    /// Make the next `put` of `t` fail: its temp file's path is a directory.
4937    fn block_put(q: &Queue, t: &Task) -> PathBuf {
4938        let tmp = q.path_of(&t.id).with_extension("json.tmp");
4939        std::fs::create_dir_all(&tmp).unwrap();
4940        tmp
4941    }
4942
4943    #[test]
4944    fn a_failed_put_leaves_no_new_attachment_directory() {
4945        let (dir, q) = queue();
4946        let mut t = task("fresh");
4947        let tmp = block_put(&q, &t);
4948        let src = source_file(dir.path(), "shot.png", "x");
4949        assert!(q.attach_and_put(&mut t, &[src]).is_err());
4950        assert!(t.attachments.is_empty());
4951        assert!(!q.attachments_dir(&t.id).exists());
4952        assert!(!q.path_of(&t.id).exists());
4953        std::fs::remove_dir(tmp).unwrap();
4954    }
4955
4956    #[test]
4957    fn a_failed_put_removes_only_the_copy_it_just_made() {
4958        let (dir, q) = queue();
4959        let mut t = task("edited");
4960        let first = source_file(dir.path(), "first.png", "1");
4961        q.attach_and_put(&mut t, &[first]).unwrap();
4962        block_put(&q, &t);
4963        let second = source_file(dir.path(), "second.png", "2");
4964        assert!(q.attach_and_put(&mut t, &[second]).is_err());
4965        assert_eq!(t.attachments, ["first.png"]);
4966        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
4967            .unwrap()
4968            .flatten()
4969            .map(|e| e.file_name().to_string_lossy().into_owned())
4970            .collect();
4971        assert_eq!(on_disk, ["first.png"]);
4972        assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
4973    }
4974
4975    #[test]
4976    fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
4977        let (dir, q) = queue();
4978        let questions = Questions::at(dir.path().join("questions"));
4979        let gone = task("gone");
4980        let mut other = task("other");
4981        let mut live = task("live");
4982        q.put(&mut other).unwrap();
4983        q.put(&mut live).unwrap();
4984        // `gone` has no record: its removal finished deleting the record but
4985        // not the set-aside attachments.
4986        let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
4987        std::fs::create_dir_all(&orphan).unwrap();
4988        std::fs::write(orphan.join("shot.png"), "x").unwrap();
4989        // `live` still has a record, so its `.removing` belongs to a removal
4990        // in progress and must be left alone.
4991        let busy = q.root.join(format!("{}.attachments.removing", live.id));
4992        std::fs::create_dir_all(&busy).unwrap();
4993
4994        q.remove(&other.id, false, &questions).unwrap();
4995        assert!(!orphan.exists(), "an orphan is swept");
4996        assert!(busy.exists(), "a removal in progress is left alone");
4997    }
4998
4999    #[test]
5000    fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
5001        let (dir, q) = queue();
5002        let questions = Questions::at(dir.path().join("questions"));
5003        let mut t = task("stuck");
5004        let src = source_file(dir.path(), "shot.png", "x");
5005        q.attach_and_put(&mut t, &[src]).unwrap();
5006        // A non-empty directory already at the set-aside name makes the rename
5007        // fail; its record is present, so the sweep leaves it in place.
5008        let aside = q.root.join(format!("{}.attachments.removing", t.id));
5009        std::fs::create_dir_all(&aside).unwrap();
5010        std::fs::write(aside.join("old.png"), "o").unwrap();
5011        assert!(q.remove(&t.id, false, &questions).is_err());
5012        assert!(q.path_of(&t.id).exists());
5013        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5014    }
5015
5016    #[test]
5017    fn a_failed_record_removal_puts_the_attachments_back() {
5018        let (dir, q) = queue();
5019        let mut t = task("rollback");
5020        let src = source_file(dir.path(), "shot.png", "x");
5021        q.attach_and_put(&mut t, &[src]).unwrap();
5022        let err = q
5023            .remove_record_with_attachments(&t.id, |_| {
5024                Err(std::io::Error::other("injected failure"))
5025            })
5026            .unwrap_err();
5027        assert!(format!("{err:#}").contains("injected failure"));
5028        assert!(q.path_of(&t.id).exists());
5029        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
5030        assert!(
5031            !q.root
5032                .join(format!("{}.attachments.removing", t.id))
5033                .exists()
5034        );
5035    }
5036
5037    #[test]
5038    fn editing_a_task_keeps_its_attachments() {
5039        let (dir, q) = queue();
5040        let src = source_file(dir.path(), "shot.png", "x");
5041        let mut t = task("editable");
5042        q.attach(&mut t, &[src]).unwrap();
5043        t.edit("new".to_owned(), "new text".to_owned()).unwrap();
5044        q.put(&mut t).unwrap();
5045        assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
5046    }
5047
5048    #[test]
5049    fn removing_a_task_deletes_its_attachments() {
5050        let (dir, q) = queue();
5051        let questions = Questions::at(dir.path().join("questions"));
5052        let src = source_file(dir.path(), "shot.png", "x");
5053        let mut t = task("doomed");
5054        q.attach(&mut t, &[src]).unwrap();
5055        q.put(&mut t).unwrap();
5056        assert!(q.attachments_dir(&t.id).is_dir());
5057        q.remove(&t.id, false, &questions).unwrap();
5058        assert!(!q.attachments_dir(&t.id).exists());
5059        assert!(q.list().is_empty());
5060    }
5061
5062    #[test]
5063    fn a_task_written_before_attachments_still_reads() {
5064        let (_dir, q) = queue();
5065        let mut t = task("old");
5066        q.put(&mut t).unwrap();
5067        let path = q.path_of(&t.id);
5068        let mut v: serde_json::Value =
5069            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
5070        v.as_object_mut().unwrap().remove("attachments");
5071        std::fs::write(&path, v.to_string()).unwrap();
5072        assert!(q.get(&t.id).unwrap().attachments.is_empty());
5073    }
5074
5075    #[test]
5076    fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
5077        let q = Queue::at(PathBuf::from("relative-queue"));
5078        let mut t = task("rel");
5079        t.attachments.push("shot.png".to_owned());
5080        let paths = q.attachment_paths(&t);
5081        assert!(paths[0].is_absolute(), "{}", paths[0].display());
5082        assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
5083    }
5084
5085    #[test]
5086    fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
5087        let dir = tempfile::tempdir().unwrap();
5088        let q = Queue::at(dir.path().join("queue"));
5089        let mut t = task("resume");
5090        t.start("A".into());
5091        t.record_pass("A", 0, 1, None);
5092        t.start("B".into());
5093        q.put(&mut t).unwrap();
5094        assert_eq!(t.runs, vec!["A", "B"]);
5095        t.start("A".into());
5096        t.record_pass("A", 1, 3, None);
5097        q.put(&mut t).unwrap();
5098        let saved = q.get(&t.id).unwrap();
5099        assert_eq!(saved.runs, vec!["B", "A"]);
5100        assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
5101        assert_eq!(saved.attempts, 3);
5102        assert_eq!(saved.passes.len(), 2);
5103        assert_eq!(saved.passes[0].run, "A");
5104        assert_eq!(saved.passes[1].run, "A");
5105    }
5106
5107    #[test]
5108    fn repeated_start_records_one_run_and_separate_passes() {
5109        let mut t = task("resume");
5110        t.start("same-run".into());
5111        t.record_pass("same-run", 0, 1, None);
5112        t.start("same-run".into());
5113        t.record_pass("same-run", 1, 2, None);
5114        assert_eq!(t.runs, vec!["same-run"]);
5115        assert_eq!(t.attempts, 2);
5116        assert_eq!(t.passes.len(), 2);
5117    }
5118
5119    #[test]
5120    fn link_run_adds_a_run_once_and_touches_nothing_else() {
5121        let dir = tempfile::tempdir().unwrap();
5122        let queue = Queue::at(dir.path().join("queue"));
5123        let mut t = Task::new(
5124            "t".to_owned(),
5125            "do it".to_owned(),
5126            PathBuf::from("."),
5127            Source::Human,
5128        );
5129        queue.put(&mut t).unwrap();
5130        let before = queue.get(&t.id).unwrap();
5131
5132        let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
5133        assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
5134        assert_eq!(linked.status, before.status);
5135        assert_eq!(linked.attempts, before.attempts);
5136        assert_eq!(linked.interrupt, before.interrupt);
5137
5138        let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5139        assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
5140        assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
5141        assert!(queue.link_run("no-such-task", "r").is_err());
5142    }
5143
5144    #[test]
5145    fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
5146        let dir = tempfile::tempdir().unwrap();
5147        let queue = Queue::at(dir.path().join("queue"));
5148        let mut t = Task::new(
5149            "t".to_owned(),
5150            "do it".to_owned(),
5151            PathBuf::from("."),
5152            Source::Human,
5153        );
5154        queue.put(&mut t).unwrap();
5155        // The daemon's copy, taken before the link.
5156        let mut snapshot = queue.get(&t.id).unwrap();
5157        queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
5158
5159        snapshot.start("20260930-000000-aaaa".to_owned());
5160        queue.put(&mut snapshot).unwrap();
5161
5162        let stored = queue.get(&t.id).unwrap();
5163        assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
5164        assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
5165        assert_eq!(stored.attempts, 1);
5166    }
5167
5168    #[test]
5169    fn concurrent_links_and_daemon_saves_lose_nothing() {
5170        let dir = tempfile::tempdir().unwrap();
5171        let queue = Queue::at(dir.path().join("queue"));
5172        let mut t = Task::new(
5173            "t".to_owned(),
5174            "do it".to_owned(),
5175            PathBuf::from("."),
5176            Source::Human,
5177        );
5178        queue.put(&mut t).unwrap();
5179        let id = t.id.clone();
5180
5181        let linkers: Vec<_> = (0..4)
5182            .map(|n| {
5183                let (queue, id) = (queue.clone(), id.clone());
5184                std::thread::spawn(move || {
5185                    for k in 0..10 {
5186                        queue
5187                            .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
5188                            .unwrap();
5189                    }
5190                })
5191            })
5192            .collect();
5193        // The daemon's side: transitions saved from its own, ever staler,
5194        // copy of the task.
5195        let mut mine = queue.get(&id).unwrap();
5196        for k in 0..10 {
5197            mine.start(format!("20260930-000009-d{k:03}"));
5198            queue.put(&mut mine).unwrap();
5199        }
5200        for l in linkers {
5201            l.join().unwrap();
5202        }
5203
5204        let stored = queue.get(&id).unwrap();
5205        assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
5206        assert_eq!(
5207            stored.attempts, 10,
5208            "linking never rewinds the daemon's work"
5209        );
5210    }
5211
5212    fn age_lock(path: &Path) {
5213        let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
5214        f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
5215            .unwrap();
5216    }
5217
5218    #[test]
5219    fn concurrent_stale_takeover_yields_one_holder() {
5220        use std::sync::atomic::{AtomicUsize, Ordering};
5221        use std::sync::{Arc, Barrier};
5222        for _ in 0..5 {
5223            let dir = tempfile::tempdir().unwrap();
5224            let q = Queue::at(dir.path().to_path_buf());
5225            let lock = dir.path().join("t.write-lock");
5226            std::fs::write(&lock, "dead-0000").unwrap();
5227            age_lock(&lock);
5228            let n = 6;
5229            let barrier = Arc::new(Barrier::new(n));
5230            let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
5231            let handles: Vec<_> = (0..n)
5232                .map(|_| {
5233                    let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
5234                    std::thread::spawn(move || {
5235                        b.wait();
5236                        let g = q.lock_task("t").unwrap();
5237                        let held = now.fetch_add(1, Ordering::SeqCst) + 1;
5238                        max.fetch_max(held, Ordering::SeqCst);
5239                        std::thread::sleep(std::time::Duration::from_millis(20));
5240                        now.fetch_sub(1, Ordering::SeqCst);
5241                        drop(g);
5242                    })
5243                })
5244                .collect();
5245            for h in handles {
5246                h.join().unwrap();
5247            }
5248            assert_eq!(max.load(Ordering::SeqCst), 1);
5249            assert!(!lock.exists());
5250        }
5251    }
5252
5253    #[test]
5254    fn dropping_a_replaced_lock_leaves_the_new_holders_token() {
5255        let dir = tempfile::tempdir().unwrap();
5256        let q = Queue::at(dir.path().to_path_buf());
5257        let lock = dir.path().join("t.write-lock");
5258        let a = q.lock_task("t").unwrap();
5259        // Simulate a foreign/legacy writer replacing the file. A modern
5260        // writer cannot acquire it while this guard's process is alive.
5261        let replacement = owner_token();
5262        std::fs::write(&lock, &replacement).unwrap();
5263        drop(a);
5264        assert_eq!(std::fs::read_to_string(&lock).unwrap(), replacement);
5265    }
5266
5267    #[test]
5268    fn aged_live_task_lock_excludes_updates_until_retirement_releases_it() {
5269        let dir = tempfile::tempdir().unwrap();
5270        let q = Queue::at(dir.path().to_path_buf());
5271        let held = q.lock_task("t").unwrap();
5272        let path = dir.path().join("t.write-lock");
5273        age_lock(&path);
5274        assert!(!break_stale(&path, &held.token));
5275        let (started_tx, started_rx) = std::sync::mpsc::channel();
5276        let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
5277        let updater = std::thread::spawn(move || {
5278            started_tx.send(()).unwrap();
5279            let _write = q.lock_task("t").unwrap();
5280            acquired_tx.send(()).unwrap();
5281        });
5282        started_rx.recv().unwrap();
5283        assert!(
5284            acquired_rx
5285                .recv_timeout(std::time::Duration::from_millis(100))
5286                .is_err()
5287        );
5288        assert_eq!(std::fs::read_to_string(&path).unwrap(), held.token);
5289        drop(held);
5290        acquired_rx
5291            .recv_timeout(std::time::Duration::from_secs(2))
5292            .unwrap();
5293        updater.join().unwrap();
5294        assert!(!path.exists());
5295    }
5296
5297    #[test]
5298    fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
5299        let dir = tempfile::tempdir().unwrap();
5300        let lock = dir.path().join("t.write-lock");
5301        std::fs::write(&lock, "new-token").unwrap();
5302        assert!(!break_stale(&lock, "old-token"));
5303        assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
5304        // Same token but fresh: not stale any more, so also left alone.
5305        assert!(!break_stale(&lock, "new-token"));
5306        assert!(lock.exists());
5307        age_lock(&lock);
5308        assert!(break_stale(&lock, "new-token"));
5309        assert!(!lock.exists());
5310    }
5311
5312    #[test]
5313    fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
5314        let dir = tempfile::tempdir().unwrap();
5315        let lock = dir.path().join("t.write-lock");
5316        std::fs::write(&lock, "dead-1").unwrap();
5317        age_lock(&lock);
5318        let marker = break_marker(&lock, "dead-1");
5319        std::fs::write(&marker, "crashed-remover").unwrap();
5320        // Fresh marker: another remover is working, so nothing is broken.
5321        assert!(!break_stale(&lock, "dead-1"));
5322        assert!(lock.exists());
5323        // Stale marker: recovered, then the lock is broken.
5324        age_lock(&marker);
5325        assert!(break_stale(&lock, "dead-1"));
5326        assert!(!lock.exists());
5327        assert!(!marker.exists());
5328    }
5329
5330    fn chat_task() -> Task {
5331        Task::new(
5332            "t".to_owned(),
5333            "do t".to_owned(),
5334            PathBuf::from("."),
5335            Source::Agent {
5336                run: "20260902-000000-beef".to_owned(),
5337                node: CHAT_NODE.to_owned(),
5338            },
5339        )
5340    }
5341
5342    #[test]
5343    fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
5344        let mut t = chat_task();
5345        t.hold_machine(Some("waiting".into()));
5346        let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
5347        t.held_at = Some(old);
5348        t.block(vec!["decision".into()], None);
5349        t.hold_machine(Some("same waiting decision".into()));
5350        assert_eq!(t.held_at, Some(old));
5351        t.start("real-new-run".into());
5352        t.fail("new failure", 1);
5353        assert_ne!(
5354            t.held_at,
5355            Some(old),
5356            "old answers cannot approve a new hold"
5357        );
5358    }
5359
5360    #[test]
5361    fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
5362        let (_dir, queue) = queue();
5363        let mut t = chat_task();
5364        t.start("old-run".into());
5365        t.hold_machine(Some("operator is considering recovery".into()));
5366        t.chat_notice = Some(ChatNotice {
5367            key: format!("held#{}", t.held_at.unwrap()),
5368            at: Timestamp::now(),
5369            outcome: ChatNoticeOutcome::Posted,
5370        });
5371        let mut raw = serde_json::to_value(&t).unwrap();
5372        raw.as_object_mut().unwrap().remove("chat_notices");
5373        raw.as_object_mut().unwrap().remove("hold_notice_cause");
5374        std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
5375        std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
5376        assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5377        queue
5378            .modify(&t.id, |task| {
5379                task.requeue();
5380                task.link_run("unprepared-run");
5381                task.startup_notice_cause("startup-source");
5382                task.fail("same unresolved preflight", 0);
5383                true
5384            })
5385            .unwrap();
5386        assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
5387        queue
5388            .modify(&t.id, |task| {
5389                task.succeed();
5390                true
5391            })
5392            .unwrap();
5393        assert_eq!(
5394            queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
5395            Some("done")
5396        );
5397    }
5398
5399    #[test]
5400    fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
5401        let mut t = chat_task();
5402        t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
5403        let key = t.chat_notice_key().unwrap();
5404        t.chat_notices.push(key);
5405        t.release();
5406        t.block(
5407            vec!["b".into(), "a".into(), "a".into()],
5408            Some("still waiting on prerequisites".into()),
5409        );
5410        assert_eq!(t.chat_notice_due(), None);
5411        t.block(
5412            vec!["new-question".into()],
5413            Some("different decision".into()),
5414        );
5415        assert!(t.chat_notice_due().is_some());
5416    }
5417
5418    #[test]
5419    fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
5420        let mut t = chat_task();
5421        t.start("existing-run".into());
5422        t.fail("implementation failed", 1);
5423        let failed = t.chat_notice_key().unwrap();
5424        t.chat_notices.push(failed.clone());
5425        t.release_operator("test", None);
5426        t.start("existing-run".into());
5427        t.fail_startup("startup-source", "same unresolved recovery source", 1);
5428        assert_eq!(
5429            t.chat_notice_due(),
5430            None,
5431            "no-progress recovery stays quiet"
5432        );
5433        t.release_operator("test", None);
5434        t.start("existing-run".into());
5435        t.fail_startup("startup-attachments", "attachment missing", 1);
5436        let attachments = t.chat_notice_due().expect("new attachment blocker");
5437        assert_ne!(attachments, failed);
5438        assert_eq!(
5439            t.hold_notice_cause.as_ref().unwrap().stage,
5440            "startup-attachments"
5441        );
5442        t.chat_notices.push(attachments);
5443        t.release_operator("test", None);
5444        t.start("existing-run".into());
5445        t.fail_startup(
5446            "startup-attachments",
5447            "same missing attachment, reworded",
5448            1,
5449        );
5450        assert_eq!(
5451            t.chat_notice_due(),
5452            None,
5453            "same stage after attempt counter reset"
5454        );
5455        t.release_operator("test", None);
5456        t.start("existing-run".into());
5457        t.fail_startup("startup-config", "different preflight stage", 1);
5458        assert!(t.chat_notice_due().is_some());
5459        assert_eq!(
5460            t.hold_notice_cause.as_ref().unwrap().stage,
5461            "startup-config"
5462        );
5463    }
5464
5465    #[test]
5466    fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
5467        let mut t = chat_task();
5468        t.hold_machine(Some("startup source missing".into()));
5469        t.startup_notice_cause("startup-source");
5470        let first = t.chat_notice_key().unwrap();
5471        t.chat_notices.push(first.clone());
5472        t.release_operator("test", None);
5473        t.startup_notice_cause("startup-source");
5474        t.hold_machine(Some("same source cannot be read".into()));
5475        assert_eq!(t.chat_notice_due(), None);
5476        t.startup_notice_cause("startup-config");
5477        assert!(t.chat_notice_due().is_some(), "different startup stage");
5478        t.notice_cause("failed", Some("existing-run"), Some(0));
5479        let run_failure = t.chat_notice_key().unwrap();
5480        t.chat_notices.push(run_failure.clone());
5481        t.release_operator("test", None);
5482        t.startup_notice_cause("startup-source");
5483        t.hold_machine(Some("no new work happened".into()));
5484        assert_eq!(t.chat_notice_due(), None);
5485        t.notice_cause("failed", Some("existing-run"), Some(1));
5486        assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
5487        assert_ne!(t.chat_notice_key().unwrap(), run_failure);
5488    }
5489
5490    #[test]
5491    fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
5492        let mut t = chat_task();
5493        t.hold_machine(Some("still waiting".into()));
5494        t.chat_notice = Some(ChatNotice {
5495            key: "held#2026-10-14T00:00:00Z".into(),
5496            at: Timestamp::now(),
5497            outcome: ChatNoticeOutcome::Posted,
5498        });
5499        assert_eq!(t.chat_notice_due(), None, "old hold baseline");
5500        t.import_chat_notice();
5501        t.notice_cause("failed", Some("new-run"), Some(0));
5502        assert!(t.chat_notice_due().is_some(), "new producer evidence");
5503        t.succeed();
5504        assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
5505    }
5506
5507    #[test]
5508    fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
5509        let mut t = chat_task();
5510        assert_eq!(t.chat_notice_due(), None, "queued");
5511        t.start("r1".to_owned());
5512        assert_eq!(t.chat_notice_due(), None, "running");
5513        t.fail("boom", 3);
5514        assert_eq!(t.status, TaskStatus::Failed);
5515        assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
5516        t.start("r2".to_owned());
5517        t.park("upgrade");
5518        assert_eq!(t.chat_notice_due(), None, "parked");
5519        t.fail("boom", 1);
5520        assert_eq!(t.status, TaskStatus::Held);
5521        let key = t
5522            .chat_notice_due()
5523            .expect("held on the last attempt is final");
5524        t.chat_notice = Some(ChatNotice {
5525            key,
5526            at: Timestamp::now(),
5527            outcome: ChatNoticeOutcome::Posted,
5528        });
5529        assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
5530        t.release();
5531        assert_eq!(t.chat_notice_due(), None, "requeued after a release");
5532        t.start("r3".to_owned());
5533        t.succeed();
5534        assert!(t.chat_notice_due().is_some(), "done is a new final state");
5535    }
5536
5537    #[test]
5538    fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
5539        for key in ["done", "done#0", "done#1", "done#12"] {
5540            let mut t = chat_task();
5541            t.start("r1".to_owned());
5542            t.succeed();
5543            t.chat_notice = Some(ChatNotice {
5544                key: key.to_owned(),
5545                at: Timestamp::now(),
5546                outcome: ChatNoticeOutcome::Posted,
5547            });
5548            t.requeue();
5549            t.start("r2".to_owned());
5550            t.succeed();
5551            assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
5552            assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
5553        }
5554    }
5555
5556    #[test]
5557    fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
5558        let mut t = chat_task();
5559        t.hold_manual(Some("wait".to_owned()));
5560        let first = t.chat_notice_due().expect("held");
5561        t.chat_notice = Some(ChatNotice {
5562            key: first.clone(),
5563            at: Timestamp::now(),
5564            outcome: ChatNoticeOutcome::Posted,
5565        });
5566        t.hold_manual(Some("still waiting".to_owned()));
5567        assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
5568        t.release();
5569        std::thread::sleep(std::time::Duration::from_millis(5));
5570        t.hold_manual(None);
5571        assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
5572    }
5573
5574    #[test]
5575    fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
5576        let mut human = task("h");
5577        human.succeed();
5578        assert_eq!(human.chat_notice_due(), None);
5579        // A follow-up only inherits the chat; it did not file the task.
5580        let mut inherited = task("i");
5581        inherited.origin_chat = Some("20260902-000000-beef".to_owned());
5582        inherited.succeed();
5583        assert_eq!(inherited.chat_notice_due(), None);
5584        let mut old = chat_task();
5585        old.schema = 12;
5586        old.status = TaskStatus::Done;
5587        assert_eq!(old.chat_notice_due(), None, "history is not replayed");
5588        let mut live = chat_task();
5589        live.schema = 12;
5590        live.start("r1".to_owned());
5591        live.succeed();
5592        assert!(live.chat_notice_due().is_some(), "started by this build");
5593        let mut held = chat_task();
5594        held.schema = 12;
5595        held.hold_machine(Some("config".to_owned()));
5596        assert!(held.chat_notice_due().is_some(), "held before any start");
5597    }
5598
5599    #[test]
5600    fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
5601        let tmp = tempfile::tempdir().expect("tempdir");
5602        let q = Queue::at(tmp.path().join("queue"));
5603        let mut t = chat_task();
5604        t.succeed();
5605        q.put(&mut t).expect("put");
5606        let mut stale = q.get(&t.id).expect("get");
5607        let key = t.chat_notice_due().expect("due");
5608        q.modify(&t.id, |x| {
5609            x.chat_notice = Some(ChatNotice {
5610                key: key.clone(),
5611                at: Timestamp::now(),
5612                outcome: ChatNoticeOutcome::Posted,
5613            });
5614            true
5615        })
5616        .expect("modify");
5617        q.put(&mut stale).expect("stale put");
5618        assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
5619    }
5620}