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::path::{Path, PathBuf};
35
36use anyhow::{Context, Result, bail};
37use jiff::Timestamp;
38use serde::{Deserialize, Serialize};
39
40use crate::ask::Questions;
41
42/// On-disk format for a queued task. Bumped when a field's meaning changes.
43///
44/// 7: added [`Task::attachments`], names of files copied under
45/// `<id>.attachments/` beside the task file (a field only; `#[serde(default)]`,
46/// so an older record reads as empty and [`read_path`] still accepts it).
47///
48/// 6: added [`Task::resume_override`] (a field only; `#[serde(default)]`, so
49/// an older record reads as `None` and [`read_path`] still accepts it).
50///
51/// 5: added [`Task::triage_applied`], the ids of triage questions whose
52/// answer has already been applied to this task. A "resume" answer used to
53/// leave no trace ([`Task::release`] clears [`Task::hold_reason`], which was
54/// the only place the applied marker lived), so when the released task failed
55/// its attempts and went back to `held`, the next idle pass found the same
56/// answered question "not applied" and released it again with `attempts` reset
57/// to 0 - the `max_attempts` bound never held. `#[serde(default)]` so an older
58/// record reads as empty. A task already looping when this build arrives has
59/// no record, so it is released once more, recorded, and then stays held.
60///
61/// 8: added [`Task::followup`], the origin of a task `crate::followup` filed
62/// from a merged run's leftover findings. Field-only bump, `#[serde(default)]`.
63///
64/// 4: added [`Task::blocked_from`], the status a task had the moment it
65/// became [`TaskStatus::Blocked`], so [`Task::unblock`] restores it instead
66/// of always landing on [`TaskStatus::Queued`]. Without it, a task a human
67/// or `crate::triage` had deliberately left [`TaskStatus::Held`] — machine
68/// or manual — would lose that the instant `crate::conduct` blocked it on a
69/// follow-up question, and come back `Queued` the moment the question was
70/// answered, regardless of what the answer said: exactly the loop where a
71/// task the operator told to stay held instead re-enters the competition
72/// queue every time someone answers a question about it. `#[serde(default)]`
73/// so an older record reads as `None`; [`Task::unblock`] then falls back to
74/// inferring `Held` from surviving hold evidence ([`Task::hold_reason`] /
75/// [`Task::hold_source`], never cleared by [`Task::block`]) rather than
76/// guessing `Queued` outright — see [`Task::unblock`]'s own doc.
77///
78/// 9: added [`Task::held_at`], the start of the current hold, so a notice can
79/// tell a question filed for this hold from one about an older cause.
80/// Field-only, `#[serde(default)]`.
81///
82/// 10: added [`Task::overrides`] and [`Task::review_of`], so a run started by
83/// hand (`magi run` / `magi review`) is carried by a task without losing its
84/// command-line choices. Field-only.
85///
86/// 11: added [`TaskStatus::Parked`] and [`Task::park_reason`], so a run that
87/// stopped at a node boundary (an upgrade, an operator's park) is no longer
88/// recorded as a failed task. Variant plus `#[serde(default)]` field; an older
89/// build must not half-read the new status.
90///
91/// 12: added [`Task::origin_chat`], the talk a task descends from, so a
92/// follow-up still knows its chat after an ancestor task is deleted.
93/// Field-only, `#[serde(default)]`.
94///
95/// 13: added [`Task::chat_notice`], the record that a chat-filed task's final
96/// state was reported to its conversation. Field-only, `#[serde(default)]`;
97/// a record below this schema is never notified (an upgrade must not replay
98/// the whole history into chats).
99///
100/// 3: added [`HoldSource`] so conductor recovery cannot release a hold an
101/// operator deliberately placed. Old records default to `None` and are
102/// protected as operator-held until an explicit release; the safe direction
103/// when their author was never recorded.
104///
105/// 2: added [`TaskStatus::Blocked`], [`Task::blocked_by`] and
106/// [`Task::block_reason`] (`crate::conduct`'s decisions) and
107/// [`Task::answers`] (operator answers carried forward to the next
108/// conductor prompt and the next run's instruction). All three are
109/// `#[serde(default)]`, so [`read_path`] accepts anything up to and
110/// including this schema rather than only an exact match — a task written
111/// by a build that only knew about schema 1 has nothing to say about
112/// blocking or answers, and defaulting those fields is exactly as good a
113/// reading as a value that build never had a chance to write.
114pub const SCHEMA: u32 = 13;
115
116/// Who placed the current hold.
117#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
118#[serde(rename_all = "lowercase")]
119pub enum HoldSource {
120    /// An operator used the CLI or web UI.
121    Manual,
122    /// The daemon or conductor placed the hold as part of its own recovery.
123    Machine,
124}
125
126impl HoldSource {
127    /// Short human-facing label for reports and the CLI.
128    pub fn label(self) -> &'static str {
129        match self {
130            Self::Manual => "manual",
131            Self::Machine => "machine",
132        }
133    }
134}
135
136/// Where a task came from. Recorded because "who asked for this" is the first
137/// question about an autonomous run, and the answer is not recoverable later.
138#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
139#[serde(tag = "kind", rename_all = "lowercase")]
140pub enum Source {
141    /// A person, at a terminal or through the web UI.
142    Human,
143    /// An agent inside a run, via `magi task add`. Both ids are recorded so a
144    /// task can be traced back to the exact seat that asked for it.
145    Agent {
146        /// Run the asking agent belonged to.
147        run: String,
148        /// Node it was working in, e.g. `implement` or `review`.
149        node: String,
150    },
151    /// A GitHub issue, imported by number.
152    Issue {
153        /// Issue number.
154        number: u64,
155        /// `owner/repo`, as `gh` reports it.
156        repo: String,
157    },
158}
159
160/// The `node` a [`Source::Agent`] carries when the asker is a chat
161/// conversation (its `run` is then the conversation's id).
162pub const CHAT_NODE: &str = "chat";
163
164impl Source {
165    /// Short human-facing label, for lists and the web UI.
166    pub fn label(&self) -> String {
167        match self {
168            Self::Human => "human".to_owned(),
169            Self::Agent { run, node } => format!("{node}@{}", short(run)),
170            Self::Issue { number, .. } => format!("issue #{number}"),
171        }
172    }
173}
174
175/// Where a task is in its life.
176#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
177#[serde(rename_all = "lowercase")]
178pub enum TaskStatus {
179    /// Waiting to be claimed.
180    Queued,
181    /// Claimed by a daemon; a run is in flight.
182    Running,
183    /// A run finished and its gate passed.
184    Done,
185    /// A run finished without passing, and attempts remain.
186    Failed,
187    /// Out of attempts, or held by hand. The loop will not pick it up.
188    Held,
189    /// Waiting on another task or an unanswered question. See
190    /// [`Task::blocked_by`]. Set and cleared by `crate::conduct` and
191    /// `crate::daemon`'s deterministic resolver, never by hand.
192    Blocked,
193    /// The run stopped at a node boundary because the process was asked to
194    /// give itself back. Nothing failed: the run's work is intact and the next
195    /// loop resumes it. Runnable like [`TaskStatus::Queued`]; the reason is in
196    /// [`Task::park_reason`], never [`Task::last_error`].
197    Parked,
198}
199
200impl TaskStatus {
201    /// Is this task eligible for a daemon to claim?
202    pub fn runnable(self) -> bool {
203        matches!(self, Self::Queued | Self::Failed | Self::Parked)
204    }
205
206    /// Lowercase name, as it appears on disk and in the API.
207    pub fn as_str(self) -> &'static str {
208        match self {
209            Self::Queued => "queued",
210            Self::Running => "running",
211            Self::Done => "done",
212            Self::Failed => "failed",
213            Self::Held => "held",
214            Self::Blocked => "blocked",
215            Self::Parked => "parked",
216        }
217    }
218}
219
220/// How many tasks sit in each [`TaskStatus`], for a dashboard tile — never a
221/// per-task view, so it carries no ids.
222#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
223pub struct TaskCounts {
224    /// Waiting to be claimed.
225    pub queued: usize,
226    /// Claimed; a run is in flight.
227    pub running: usize,
228    /// A run finished and its gate passed.
229    pub done: usize,
230    /// A run finished without passing, and attempts remain.
231    pub failed: usize,
232    /// Out of attempts, or held by hand.
233    pub held: usize,
234    /// Waiting on another task or an unanswered question.
235    pub blocked: usize,
236    /// Run parked at a node boundary; resumable, not a failure.
237    pub parked: usize,
238}
239
240impl TaskCounts {
241    /// Tally `tasks` by status. A pure function over whatever [`Queue::list`]
242    /// already read, so it needs no root of its own and stays trivially
243    /// testable against a hand-built slice.
244    pub fn of(tasks: &[Task]) -> Self {
245        let mut counts = Self::default();
246        for t in tasks {
247            match t.status {
248                TaskStatus::Queued => counts.queued += 1,
249                TaskStatus::Running => counts.running += 1,
250                TaskStatus::Done => counts.done += 1,
251                TaskStatus::Failed => counts.failed += 1,
252                TaskStatus::Held => counts.held += 1,
253                TaskStatus::Blocked => counts.blocked += 1,
254                TaskStatus::Parked => counts.parked += 1,
255            }
256        }
257        counts
258    }
259}
260
261/// One unit of work.
262#[derive(Debug, Clone, Serialize, Deserialize)]
263#[serde(deny_unknown_fields)]
264pub struct Task {
265    /// On-disk format version.
266    pub schema: u32,
267    /// Task id, e.g. `20260902-140501-a1b2`.
268    pub id: String,
269    /// One line, for lists and notifications.
270    pub title: String,
271    /// The task itself, handed to the graph verbatim.
272    pub instruction: String,
273    /// Repository to work in.
274    pub repo: PathBuf,
275    /// Who asked.
276    pub source: Source,
277    /// Higher runs first; ties break oldest-first so nothing starves.
278    #[serde(default)]
279    pub priority: i32,
280    /// Run this task alone: one implementer, no panel of judges to convince.
281    ///
282    /// `#[serde(default)]` so a queue file written before this field existed
283    /// still reads, as `false` - the ordinary multi-candidate competition,
284    /// unchanged. A task set to `solo` still runs the whole graph; only the
285    /// candidate count the daemon builds it with changes, and
286    /// [`crate::graph::Runner`] already collapses a single-candidate run to
287    /// implement → review → gate → merge on its own (see
288    /// [`crate::graph::Runner::review`]'s doc), so nothing about judging,
289    /// deliberation or voting had to change to support this.
290    #[serde(default)]
291    pub solo: bool,
292    /// Current state.
293    pub status: TaskStatus,
294    /// How many times this task has been claimed.
295    #[serde(default)]
296    pub attempts: usize,
297    /// Runs this task has produced, oldest first.
298    #[serde(default)]
299    pub runs: Vec<String>,
300    /// Why the last attempt did not land.
301    #[serde(default)]
302    pub last_error: Option<String>,
303    /// What a human hold is waiting on.
304    ///
305    /// `None` covers both the ordinary cases: a hold the loop makes itself
306    /// (out of attempts, or the disk gate closed) explains itself through
307    /// [`Task::last_error`] instead, and a human hold nobody bothered to
308    /// explain is still a valid hold. The queue has no way to express a
309    /// dependency between two tasks, so on the occasions a hold really is
310    /// "wait for that other task first", this is the only place that reason
311    /// survives - see [`Task::hold_manual`] and [`Task::release`].
312    ///
313    /// `#[serde(default)]` so a queue file written before this field existed
314    /// still reads, with no reason recorded rather than a parse error.
315    #[serde(default)]
316    pub hold_reason: Option<String>,
317    /// Who placed [`Task::hold_reason`].  `None` is a compatible old record;
318    /// see [`Task::operator_held`] for its deliberately conservative meaning.
319    #[serde(default)]
320    pub hold_source: Option<HoldSource>,
321    /// Diagnostic detail excerpted from the run that led to a hold - what a
322    /// human would have found opening `artifacts/` by hand, not the one-line
323    /// reason in [`Task::last_error`]. Set only when a run's own attempts are
324    /// exhausted and the task becomes [`TaskStatus::Held`]; `daemon` computes
325    /// it from the run's own record, since this module has no notion of a
326    /// run's internals. Bounded in length by the writer - see
327    /// `daemon::diagnostic` - so a verbose run cannot make this file grow
328    /// without limit.
329    ///
330    /// `#[serde(default)]` so a queue file written before this field existed
331    /// still reads, with no diagnostic recorded rather than a parse error.
332    #[serde(default)]
333    pub diagnostic: Option<String>,
334    /// What this task is waiting on: other task ids, unanswered
335    /// `crate::ask::Question` ids, or both. Non-empty exactly when
336    /// [`TaskStatus::Blocked`]; emptying it — see [`Task::unblock`] — is what
337    /// puts the task back at [`TaskStatus::Queued`].
338    ///
339    /// Set by `crate::conduct`'s decisions and cleared deterministically by
340    /// `crate::daemon` as each dependency resolves, never by a person. Never
341    /// `#[serde(default)]` is skipped: a queue file from before this field
342    /// existed has nothing to report here, and an empty list is exactly that.
343    #[serde(default)]
344    pub blocked_by: Vec<String>,
345    /// One line explaining the current [`Task::blocked_by`], written by
346    /// `crate::conduct`. Cleared whenever `blocked_by` empties.
347    #[serde(default)]
348    pub block_reason: Option<String>,
349    /// The status this task had the moment [`Task::block`] most recently
350    /// moved it to [`TaskStatus::Blocked`] — what [`Task::unblock`] restores
351    /// once nothing is left in `blocked_by`, instead of always landing on
352    /// [`TaskStatus::Queued`]. See [`SCHEMA`]'s doc for schema 4 on why this
353    /// exists: an answer to a question `crate::conduct` filed about a
354    /// [`TaskStatus::Held`] task must not itself be what puts the task back
355    /// in the competition queue.
356    ///
357    /// `#[serde(default)]` so a queue file written before this field existed
358    /// reads as `None`; [`Task::unblock`] treats that the same as a task
359    /// blocked straight from `Queued`, unless surviving hold evidence says
360    /// otherwise.
361    #[serde(default)]
362    pub blocked_from: Option<TaskStatus>,
363    /// Questions `crate::conduct` asked about this task that the operator has
364    /// since answered, oldest first — what was asked, and what they said.
365    ///
366    /// A blocking question's id leaves [`Task::blocked_by`] the moment
367    /// [`crate::ask::QuestionStatus::Answered`] is observed, but the id alone
368    /// tells nobody what was decided. This is what carries the answer's
369    /// *content* forward: into the next conductor prompt for this task, and
370    /// into the instruction handed to the next run — see `crate::daemon`'s
371    /// deterministic blocker resolution. Kept for the task's whole life, the
372    /// same as [`Task::runs`]: a release resets attempts, not evidence.
373    #[serde(default)]
374    pub answers: Vec<AnsweredQuestion>,
375    /// Ids of the `crate::triage` questions whose answer has been applied to
376    /// this task. Unlike [`Task::hold_reason`], [`Task::release`] and every
377    /// hold transition leave it alone, so an answer is applied at most once
378    /// however many times the task is held again. See [`SCHEMA`]'s doc for
379    /// schema 5. `#[serde(default)]` so an older record reads as empty.
380    #[serde(default)]
381    pub triage_applied: Vec<String>,
382    /// Ids of the `magi ask` questions whose [`crate::ask::ChoiceAction`] has
383    /// been applied to this task. Separate from [`Task::triage_applied`]
384    /// (a different producer) and, like it, untouched by [`Task::release`],
385    /// so one answer acts at most once however often the task is held again.
386    /// `#[serde(default)]` so an older record reads as empty.
387    #[serde(default)]
388    pub actions_applied: Vec<String>,
389    /// The operator's "resume" answer to a triage question, kept until the
390    /// task actually runs (or is done) so `crate::conduct` cannot silently
391    /// undo it and `crate::triage` can tell that a hold it sees now came
392    /// *after* the answer. See [`OperatorResume`]. `#[serde(default)]`.
393    #[serde(default)]
394    pub resume_override: Option<OperatorResume>,
395    /// Set by `crate::conduct` when it chooses `Review` recovery for a task
396    /// whose branch survived a blocked run: the branch to reopen with
397    /// `crate::graph::Runner::review` instead of competing from scratch.
398    ///
399    /// Requeues the task the same way [`Task::release`] does, so it is
400    /// picked up by the ordinary loop; `crate::daemon` reads this field once,
401    /// when it actually starts the run, and clears it either way — consumed
402    /// on success, dropped if the branch no longer exists by then. Never set
403    /// from the conductor's own words: `crate::daemon` derives the branch
404    /// name itself from the task's last run, so a hallucinated branch can
405    /// never reach here.
406    #[serde(default)]
407    pub review_branch: Option<String>,
408    /// A release deliberately starts a new competition instead of resuming
409    /// the prior run. History remains as evidence in `runs`.
410    #[serde(default)]
411    pub fresh_start: bool,
412    /// Marked by an operator (`magi task interrupt`) to ask `magi serve` to
413    /// run this one ahead of whatever it already has in flight, once
414    /// `[daemon] pause_for_interrupts` is on - see
415    /// `crate::daemon::advance_interrupt`. Never set by the loop itself, and
416    /// deliberately a different operation from [`Task::set_priority`]: a
417    /// priority only reorders the queue a claim has not reached yet, while
418    /// this asks a run already in flight to park at its next safe boundary
419    /// and step aside. `#[serde(default)]` so a queue file written before
420    /// this field existed still reads, as `false` - no task interrupts
421    /// anything unless asked to, exactly as before.
422    #[serde(default)]
423    pub interrupt: bool,
424    /// Marked by `magi task add --urgent`: `crate::daemon::poll` dispatches
425    /// this task through its own one-slot `urgent_sem` the moment it is
426    /// runnable, in addition to whatever is already running under the
427    /// ordinary `[daemon] max_concurrent_runs` pool - never instead of it,
428    /// and never by pausing or otherwise touching that run. This is the
429    /// opposite direction from [`Task::interrupt`]: that one asks a run
430    /// already in flight to step aside; this one never asks anything to
431    /// step aside, it only spends one additional, temporary concurrency
432    /// slot. The two are independent and may both be set on the same task,
433    /// but this exemption stops at `[daemon] pause_for_interrupts`'s own
434    /// park/resume handoff (75dd): while an interrupt sequence is actively
435    /// parking, running, or resuming - its own, or an unrelated task's -
436    /// `crate::daemon::interrupt_gate` withholds an urgent candidate exactly
437    /// like an ordinary one, never exempted. 75dd's "at most one run, ever,
438    /// at once" guarantee takes precedence, because the alternative is a run
439    /// still genuinely in flight (only *asked* to park, not yet gone) ending
440    /// up alongside a second one this feature let through - the very thing
441    /// that guarantee exists to rule out.
442    ///
443    /// `#[serde(default)]` so a queue file written before this field existed
444    /// still reads, as `false` - no task claims the urgent slot unless asked
445    /// to, exactly as before.
446    #[serde(default)]
447    pub urgent: bool,
448    /// Files (typically screenshots) copied into `<id>.attachments/` beside
449    /// this task's file by [`Queue::attach`], as bare validated names. The
450    /// copy is what makes them reach the implementer: the original may be
451    /// cleaned up long before the task runs. Untouched by [`Task::edit`].
452    /// `#[serde(default)]` so an older record reads as empty.
453    #[serde(default)]
454    pub attachments: Vec<String>,
455    /// Set on a task `crate::followup` filed from the findings a merged run
456    /// left open: which run and findings it carries, and how many follow-ups
457    /// deep it is. `None` for every ordinary task. `#[serde(default)]`.
458    #[serde(default)]
459    pub followup: Option<FollowUp>,
460    /// Id of the chat this task descends from: set by [`Task::new`] when a
461    /// chat files it, inherited by the follow-ups `crate::followup` files, so
462    /// the provenance survives deleting an ancestor. `None` for a task written
463    /// before the field existed (`crate::consult` then walks the ancestry).
464    /// `#[serde(default)]`.
465    #[serde(default)]
466    pub origin_chat: Option<String>,
467    /// The command-line choices (`--merge`, `-c`, `--seed`, ...) of the
468    /// `magi run` / `magi review` that filed this task, applied on top of the
469    /// repository's config by whoever executes it - a daemon included - so
470    /// handing the task over does not change what it does. `None` for every
471    /// task filed any other way. `#[serde(default)]`.
472    #[serde(default)]
473    pub overrides: Option<RunOverrides>,
474    /// The branch an explicit `magi review <branch>` asked to review. Unlike
475    /// [`Task::review_branch`] (a recovery choice the daemon consumes and
476    /// drops back to a competition when the branch is gone), this is the whole
477    /// request: it is never consumed, and a missing branch fails the attempt
478    /// rather than paying for an implementation nobody asked for.
479    /// `#[serde(default)]`.
480    #[serde(default)]
481    pub review_of: Option<String>,
482    /// When the current hold began. Set by the transitions into
483    /// [`TaskStatus::Held`] (a re-hold keeps it), cleared by
484    /// [`Task::release`]. `crate::notices` uses it as the identity of the
485    /// cause: a question filed before it is about something older.
486    #[serde(default)]
487    pub held_at: Option<Timestamp>,
488    /// Why the run parked, while the task is [`TaskStatus::Parked`]. Kept apart
489    /// from [`Task::last_error`] because nothing failed. Cleared by every
490    /// transition out of the parked state. `#[serde(default)]`.
491    #[serde(default)]
492    pub park_reason: Option<String>,
493    /// The last final state reported to the originating chat. See
494    /// [`Task::chat_notice_due`]. Never cleared by a release or requeue.
495    #[serde(default)]
496    pub chat_notice: Option<ChatNotice>,
497    /// When the task was filed.
498    pub created_at: Timestamp,
499    /// Last change to this file.
500    pub updated_at: Timestamp,
501}
502
503/// Current and legacy completion keys; legacy run counts contain only digits.
504pub(crate) fn is_done_notice_key(key: &str) -> bool {
505    key == "done"
506        || key
507            .strip_prefix("done#")
508            .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
509}
510
511/// What became of a completion notice. See [`Task::chat_notice`].
512#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
513#[serde(rename_all = "lowercase")]
514pub enum ChatNoticeOutcome {
515    /// Queued as a draft of the originating talk.
516    Posted,
517    /// The talk is gone or closed; nothing was queued, and nothing will be.
518    Skipped,
519}
520
521/// The final state of a chat-filed task that was already reported (or skipped).
522#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
523pub struct ChatNotice {
524    /// Identity of the cause, see [`Task::chat_notice_key`].
525    pub key: String,
526    /// When it was recorded.
527    pub at: Timestamp,
528    /// Posted or skipped.
529    pub outcome: ChatNoticeOutcome,
530}
531
532/// Command-line choices carried by a task. See [`Task::overrides`].
533#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
534pub struct RunOverrides {
535    /// `--merge`: `none`, `local` or `pr`.
536    #[serde(default)]
537    pub merge: Option<String>,
538    /// `-c` / `--implementers`: parallel implementations (persisted as `candidates`).
539    #[serde(default)]
540    pub candidates: Option<usize>,
541    /// `-j`: independent judges.
542    #[serde(default)]
543    pub judges: Option<usize>,
544    /// `--reviewers`: reviewers per round.
545    #[serde(default)]
546    pub reviewers: Option<usize>,
547    /// `--review-rounds`.
548    #[serde(default)]
549    pub review_rounds: Option<usize>,
550    /// `--seed`.
551    #[serde(default)]
552    pub seed: Option<u64>,
553    /// `--config`, absolute: the daemon's own working directory differs.
554    #[serde(default)]
555    pub config: Option<PathBuf>,
556}
557
558impl RunOverrides {
559    /// Lay `other` over these: every choice `other` made wins, the rest stay.
560    pub fn merge_over(&mut self, other: &RunOverrides) {
561        macro_rules! take {
562            ($($f:ident),*) => {$(
563                if other.$f.is_some() {
564                    self.$f.clone_from(&other.$f);
565                }
566            )*};
567        }
568        take!(
569            merge,
570            candidates,
571            judges,
572            reviewers,
573            review_rounds,
574            seed,
575            config
576        );
577    }
578
579    /// Apply these on top of `config`.
580    pub fn apply(&self, config: &mut crate::config::Config) {
581        if let Some(n) = self.candidates {
582            config.graph.implementers = n;
583        }
584        if let Some(n) = self.judges {
585            config.graph.judges = n;
586        }
587        if let Some(n) = self.reviewers {
588            config.graph.reviewers = n;
589        }
590        if let Some(n) = self.review_rounds {
591            config.graph.review_rounds = n;
592        }
593        if let Some(m) = &self.merge
594            && let Ok(mode) = crate::daemon::merge_mode(m)
595        {
596            config.merge.mode = mode;
597        }
598        if let Some(s) = self.seed {
599            config.blind.seed = Some(s);
600        }
601    }
602}
603
604/// What a follow-up task came from. See [`Task::followup`].
605#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
606pub struct FollowUp {
607    /// The merged run whose open findings this task carries.
608    pub run: String,
609    /// The queue task that run served, when it had one.
610    #[serde(default)]
611    pub origin_task: Option<String>,
612    /// The merged pull request.
613    pub pr: String,
614    /// Ids of the findings (e.g. `R3-1-1`) this task covers.
615    pub findings: Vec<String>,
616    /// Follow-up depth: 1 for a follow-up of an ordinary task's run, 2 for a
617    /// follow-up of that, and so on. An ordinary task is generation 0.
618    pub generation: u32,
619}
620
621/// A triage "resume" answer and what became of it. See
622/// [`Task::resume_override`].
623#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
624pub struct OperatorResume {
625    /// The triage question the operator answered.
626    pub question_id: String,
627    /// When the answer was applied.
628    pub at: Timestamp,
629    /// The reason `crate::conduct` gave for holding the task again after the
630    /// answer, if it did. The conductor may do this once.
631    #[serde(default)]
632    pub conductor_rehold: Option<String>,
633    /// The operator answered "resume" a second time, to the question about
634    /// that contradiction: the conductor may no longer hold this task.
635    #[serde(default)]
636    pub forced: bool,
637    /// Set when the answer was a structured `resume` action on a `magi ask`
638    /// choice ([`crate::ask::ChoiceAction::Resume`]): the run the operator
639    /// chose to continue. Its presence also stops `crate::conduct` from
640    /// requeuing the task into a fresh competition, which would discard
641    /// exactly the run that was named. Cleared with the rest of the record
642    /// once the task runs. `#[serde(default)]` so older records read as `None`.
643    #[serde(default)]
644    pub pinned_run: Option<String>,
645}
646
647/// One question `crate::conduct` asked about a task, and what the operator
648/// said back. See [`Task::answers`].
649#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
650pub struct AnsweredQuestion {
651    /// The question as asked, e.g. [`crate::ask::Question::summary`].
652    pub question: String,
653    /// What the operator answered.
654    pub answer: String,
655}
656
657impl Task {
658    /// File a new task. Persist it with [`Queue::put`].
659    pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
660        let now = Timestamp::now();
661        let origin_chat = match &source {
662            Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
663            _ => None,
664        };
665        Self {
666            schema: SCHEMA,
667            id: new_id(),
668            title,
669            instruction,
670            repo,
671            source,
672            priority: 0,
673            solo: false,
674            status: TaskStatus::Queued,
675            attempts: 0,
676            runs: Vec::new(),
677            last_error: None,
678            hold_reason: None,
679            hold_source: None,
680            diagnostic: None,
681            blocked_by: Vec::new(),
682            block_reason: None,
683            blocked_from: None,
684            answers: Vec::new(),
685            triage_applied: Vec::new(),
686            actions_applied: Vec::new(),
687            resume_override: None,
688            review_branch: None,
689            fresh_start: false,
690            interrupt: false,
691            urgent: false,
692            attachments: Vec::new(),
693            followup: None,
694            origin_chat,
695            overrides: None,
696            review_of: None,
697            held_at: None,
698            park_reason: None,
699            chat_notice: None,
700            created_at: now,
701            updated_at: now,
702        }
703    }
704
705    /// The chat this task itself says it came from: the recorded
706    /// [`Task::origin_chat`], else its own source when a chat filed it.
707    pub fn chat_talk(&self) -> Option<&str> {
708        if let Some(id) = &self.origin_chat {
709            return Some(id);
710        }
711        match &self.source {
712            Source::Agent { run, node } if node == CHAT_NODE => Some(run),
713            _ => None,
714        }
715    }
716
717    /// The talk that filed this task *itself* (`Source::Agent` with
718    /// [`CHAT_NODE`]); unlike [`Task::chat_talk`] a follow-up's inherited
719    /// `origin_chat` does not count.
720    pub fn filed_by_chat(&self) -> Option<&str> {
721        match &self.source {
722            Source::Agent { run, node } if node == CHAT_NODE => Some(run),
723            _ => None,
724        }
725    }
726
727    /// Identity of the final state a chat should hear about, or `None` while
728    /// the task is intermediate (queued, running, parked, a failed attempt the
729    /// loop will retry - the last attempt turns into a hold). A hold is keyed
730    /// by when it began and a block by what it waits on. Completion is keyed
731    /// by the task itself, so requeues never report completion twice.
732    pub fn chat_notice_key(&self) -> Option<String> {
733        match self.status {
734            TaskStatus::Done => Some("done".to_owned()),
735            TaskStatus::Held => Some(format!(
736                "held#{}",
737                self.held_at.map(|t| t.to_string()).unwrap_or_default()
738            )),
739            TaskStatus::Blocked => Some(format!(
740                "blocked#{}#{}",
741                self.blocked_by.join(","),
742                self.block_reason.as_deref().unwrap_or_default()
743            )),
744            _ => None,
745        }
746    }
747
748    /// The key of a completion notice owed to the originating chat now, if
749    /// any: filed by a chat, schema 13 or newer, in a final state, and not the
750    /// state already recorded in [`Task::chat_notice`].
751    pub fn chat_notice_due(&self) -> Option<String> {
752        self.filed_by_chat()?;
753        if self.schema < 13 {
754            return None;
755        }
756        let key = self.chat_notice_key()?;
757        match &self.chat_notice {
758            Some(n) if n.key == key || (key == "done" && is_done_notice_key(&n.key)) => None,
759            _ => Some(key),
760        }
761    }
762
763    /// Short form used in reports, matching a run's short id.
764    pub fn short(&self) -> &str {
765        short(&self.id)
766    }
767
768    /// Record that triage question `question_id`'s answer has been applied.
769    pub fn mark_triage_applied(&mut self, question_id: &str) {
770        if !self.triage_applied(question_id) {
771            self.triage_applied.push(question_id.to_owned());
772        }
773    }
774
775    /// Has the action of `magi ask` question `question_id` already been
776    /// applied, or is it being applied now? See [`Task::actions_applied`].
777    pub fn action_applied(&self, question_id: &str) -> bool {
778        self.actions_applied.iter().any(|id| id == question_id)
779    }
780
781    /// Record that `magi ask` question `question_id`'s action was applied.
782    pub fn mark_action_applied(&mut self, question_id: &str) {
783        if !self.action_applied(question_id) {
784            self.actions_applied.push(question_id.to_owned());
785        }
786    }
787
788    /// Has triage question `question_id`'s answer already been applied?
789    pub fn triage_applied(&self, question_id: &str) -> bool {
790        self.triage_applied.iter().any(|id| id == question_id)
791    }
792
793    /// Record that a run has started for this task.
794    ///
795    /// Clears [`Task::interrupt`]: a mark to run ahead of whatever else is
796    /// in flight is fulfilled the moment this task actually gets its turn,
797    /// dispatched same as any other. Without this, a task whose run fails
798    /// and requeues - still `runnable`, still carrying the mark from its
799    /// first attempt - would keep re-triggering `crate::daemon`'s interrupt
800    /// scheduler and re-parking whatever it interrupted on every later
801    /// boundary, for as long as its attempts hold out, instead of the
802    /// one-shot "let this go next" the mark is meant to be.
803    pub fn start(&mut self, run: String) {
804        self.stamp_schema();
805        self.status = TaskStatus::Running;
806        self.attempts += 1;
807        self.runs.push(run);
808        self.last_error = None;
809        self.park_reason = None;
810        self.fresh_start = false;
811        self.interrupt = false;
812        // The answer has been honoured: the task got its turn.
813        self.resume_override = None;
814    }
815
816    /// Note that `run` serves this task without otherwise changing its life:
817    /// no status, attempt or interrupt change, unlike [`Task::start`]. For a
818    /// run somebody else started on the task's behalf (`magi run --task`);
819    /// idempotent. Returns whether anything was added.
820    pub fn link_run(&mut self, run: &str) -> bool {
821        if self.runs.iter().any(|r| r == run) {
822            return false;
823        }
824        self.runs.push(run.to_owned());
825        true
826    }
827
828    /// Record a successful run.
829    ///
830    /// Both `magi task done` and `POST /api/queue/{id}/done` can close a held
831    /// *or blocked* task directly, with no release in between, so this clears
832    /// `hold_reason` and `blocked_by`/`block_reason` the same way
833    /// [`Task::release`] does. Otherwise a task held for "waiting on 3ed9", or
834    /// blocked on a dependency that never actually finished, and then closed
835    /// as done without ever being released would still read as waiting on
836    /// something in `magi task show` and on its card, after it no longer is.
837    pub fn succeed(&mut self) {
838        self.stamp_schema();
839        self.status = TaskStatus::Done;
840        self.resume_override = None;
841        self.last_error = None;
842        self.park_reason = None;
843        self.hold_reason = None;
844        self.hold_source = None;
845        self.diagnostic = None;
846        self.blocked_by.clear();
847        self.block_reason = None;
848        self.blocked_from = None;
849    }
850
851    /// Finish the task because the run's whole change turned out to be on the
852    /// base already ([`crate::run::RunStatus::AlreadyInBase`]): `Done`, with
853    /// `note` kept in [`Task::last_error`] so `magi task show` says why, and
854    /// the attempt refunded. Nothing went wrong and nothing was spent
855    /// misbehaving, so this is neither [`Task::fail`] nor
856    /// [`Task::handed_off`]; and unlike [`Task::succeed`] it does not pretend
857    /// the run landed anything.
858    pub fn already_landed(&mut self, note: impl Into<String>) {
859        self.succeed();
860        self.attempts = self.attempts.saturating_sub(1);
861        self.last_error = Some(note.into());
862    }
863
864    /// Earlier attempts at this task that a later one has since made moot —
865    /// empty unless the task is [`TaskStatus::Done`] *and* `last_run_succeeded`
866    /// says `runs.last()` is actually why.
867    ///
868    /// `runs` is oldest first, and [`Task::start`] is the only thing that
869    /// pushes to it, always right before the attempt it names either succeeds
870    /// or fails; `succeed` itself never touches `runs`. So whenever `status`
871    /// is `Done` *because* the loop itself saw that last attempt land
872    /// (`Merged`/`Ready`), everything before it in the same list is a retry
873    /// this task no longer needs. But `succeed` is also reachable directly —
874    /// `magi task done`, the web UI's equivalent, and the conductor's
875    /// `Recovery::Done` all call it on a task in *any* status, including one
876    /// whose last recorded attempt never landed at all (closed by hand after
877    /// a merge magi's own loop never saw). `runs.last()` alone cannot tell
878    /// those two cases apart — that requires the caller to have actually
879    /// looked at that run's own `status`, which this module has no way to
880    /// do — so `last_run_succeeded` is the caller's answer to exactly that
881    /// question, not something this function can derive from `Task` alone.
882    ///
883    /// Deliberately narrower than [`Queue::superseded`]'s "every earlier
884    /// attempt has a later one" walk, which fires the moment a retry starts
885    /// even though the retry itself might still fail: that reading is right
886    /// for the web UI's "a newer attempt exists, go look at that one
887    /// instead" note, but wrong for deciding a run no longer needs a human's
888    /// attention, which is only true once the task's story has actually
889    /// ended well. Two Blocked runs sitting side by side while a third
890    /// attempt is still in flight must not be touched by this — see
891    /// `daemon::supersede_prior_runs`, the caller that turns this list into
892    /// rewritten `run.json` files.
893    pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
894        if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
895            return &[];
896        }
897        &self.runs[..self.runs.len() - 1]
898    }
899
900    /// The attempt right after `run` in this task's history, if there is one.
901    ///
902    /// The one place "a later attempt replaced this run" is decided:
903    /// [`Queue::superseded`] and [`Queue::superseded_by`] (the web UI's
904    /// note) read it, and so does [`Task::earlier_attempts`], so what is shown
905    /// and what `crate::handover` acts on cannot disagree.
906    pub fn successor_of(&self, run: &str) -> Option<&String> {
907        let pos = self.runs.iter().position(|r| r == run)?;
908        self.runs.get(pos + 1)
909    }
910
911    /// Every run already recorded for this task, i.e. every run that an
912    /// attempt being minted *right now* supersedes.
913    ///
914    /// The new run is not in [`Task::runs`] yet when it decides what to take
915    /// over (`Task::start` pushes it afterwards), so "has a later attempt" is
916    /// true of everything listed here by the time the new run exists.
917    pub fn earlier_attempts(&self) -> &[String] {
918        &self.runs
919    }
920
921    /// Stamp [`Task::held_at`] on the way into [`TaskStatus::Held`]; a task
922    /// that is already held keeps the time its hold began.
923    /// A transition this build makes (start, hold, block, done) makes the task
924    /// eligible for a chat notice even when it was filed under an older
925    /// schema; only records no build has touched since stay silent
926    /// ([`Task::chat_notice_due`]).
927    fn stamp_schema(&mut self) {
928        self.schema = self.schema.max(SCHEMA);
929    }
930
931    fn note_held(&mut self) {
932        self.stamp_schema();
933        if self.status != TaskStatus::Held || self.held_at.is_none() {
934            self.held_at = Some(Timestamp::now());
935        }
936    }
937
938    /// Record a failed attempt. Out of attempts means held for a human, rather
939    /// than retried until the money runs out.
940    ///
941    /// Clears [`Task::diagnostic`] unconditionally: it belongs to whatever run
942    /// produced it, and a caller that has one for *this* attempt sets it
943    /// itself right after calling this, once it knows the task actually ended
944    /// up [`TaskStatus::Held`] - see `daemon::diagnostic`. Without the clear, a
945    /// task released after a diagnosed hold and then failed again for an
946    /// unrelated, undiagnosed reason (a config error, say) would go on
947    /// showing the previous run's diagnostic as if it explained the new one.
948    ///
949    /// Also records `why` into [`Task::hold_reason`] when this ends up
950    /// [`TaskStatus::Held`], so the notification and `magi task show` say the
951    /// same thing as [`Task::last_error`] instead of leaving the hold's own
952    /// reason blank.
953    pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
954        let why = why.into();
955        self.diagnostic = None;
956        self.park_reason = None;
957        self.status = if self.attempts >= max_attempts {
958            self.note_held();
959            self.hold_source = Some(HoldSource::Machine);
960            self.hold_reason = Some(why.clone());
961            TaskStatus::Held
962        } else {
963            TaskStatus::Failed
964        };
965        self.last_error = Some(why);
966    }
967
968    /// Record an attempt that failed for a reason the task is not responsible
969    /// for - the agent CLIs ran out of quota and the judging panel collapsed.
970    ///
971    /// This refunds the attempt on purpose. A quota window closing at 4am must
972    /// not spend the backlog's retry budget: the operator would come back to a
973    /// queue of held tasks that were never actually judged, and would have to
974    /// release every one by hand to find out which had a real problem. The task
975    /// goes back to `Failed`, which the loop retries, so a reset quota picks the
976    /// work up where it stopped.
977    pub fn stall(&mut self, why: impl Into<String>) {
978        self.last_error = Some(why.into());
979        self.diagnostic = None;
980        self.attempts = self.attempts.saturating_sub(1);
981        self.status = TaskStatus::Failed;
982    }
983
984    /// Record a run that parked at a node boundary. Nothing failed: the attempt
985    /// is refunded, [`Task::last_error`] is left as it was, the run history and
986    /// `fresh_start` are kept so the next loop resumes the same run, and the
987    /// reason goes to [`Task::park_reason`]. Runnable like a requeued task and
988    /// never counted against `max_attempts`.
989    pub fn park(&mut self, why: impl Into<String>) {
990        self.diagnostic = None;
991        self.attempts = self.attempts.saturating_sub(1);
992        self.status = TaskStatus::Parked;
993        self.park_reason = Some(why.into());
994        // The variant and field are schema 11; stamp it so an older build
995        // refuses the record instead of half-reading it.
996        self.schema = self.schema.max(SCHEMA);
997    }
998
999    /// Whether this held task may only be released by an operator.
1000    ///
1001    /// Old files did not record a source. Preserve every such hold rather
1002    /// than guessing that it was automatic and risking duplicate work. New
1003    /// automatic holds record [`HoldSource::Machine`] and remain recoverable.
1004    pub fn operator_held(&self) -> bool {
1005        self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
1006    }
1007
1008    /// Take this task out of the loop's reach by an operator action.
1009    ///
1010    /// Clears `blocked_by`/`block_reason` unconditionally, the same as
1011    /// [`Task::release`] and for the same reason its own comment already
1012    /// gives: a human choosing to hold a *blocked* task overrides its wait
1013    /// outright, the same as it overrides an ordinary hold. Without this, a
1014    /// task held straight out of [`TaskStatus::Blocked`] - the web UI's "Hold"
1015    /// button is reachable on a blocked task, same as "Mark done" - kept
1016    /// reading as still waiting on a dependency it no longer had any claim on.
1017    pub fn hold_manual(&mut self, reason: Option<String>) {
1018        self.park_reason = None;
1019        self.note_held();
1020        self.status = TaskStatus::Held;
1021        if reason.is_some() {
1022            self.hold_reason = reason;
1023        }
1024        self.hold_source = Some(HoldSource::Manual);
1025        self.blocked_by.clear();
1026        self.block_reason = None;
1027        self.blocked_from = None;
1028    }
1029
1030    /// Take this task out of the loop's reach during automatic recovery.
1031    ///
1032    /// Clears `blocked_by`/`block_reason` for the same reason
1033    /// [`Task::hold_manual`] does.
1034    pub fn hold_machine(&mut self, reason: Option<String>) {
1035        self.park_reason = None;
1036        self.note_held();
1037        self.status = TaskStatus::Held;
1038        if reason.is_some() {
1039            self.hold_reason = reason;
1040        }
1041        self.hold_source = Some(HoldSource::Machine);
1042        self.blocked_by.clear();
1043        self.block_reason = None;
1044        self.blocked_from = None;
1045    }
1046
1047    /// Block this task on other task ids and/or open question ids, chosen by
1048    /// `crate::conduct`. Pure: the caller still owns writing it back with
1049    /// [`Queue::put`].
1050    ///
1051    /// Records [`Task::blocked_from`] the first time this moves the task into
1052    /// [`TaskStatus::Blocked`], and leaves it alone on a later call that adds
1053    /// or replaces `blocked_by` while the task is already `Blocked` - a
1054    /// second question about an already-blocked task must not overwrite the
1055    /// status it should eventually return to with `Blocked` itself.
1056    pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
1057        self.stamp_schema();
1058        if self.status != TaskStatus::Blocked {
1059            self.blocked_from = Some(self.status);
1060        }
1061        self.status = TaskStatus::Blocked;
1062        self.blocked_by = blocked_by;
1063        self.block_reason = reason;
1064    }
1065
1066    /// Remove one resolved dependency (a task id that became [`TaskStatus::Done`],
1067    /// or a question id that became [`crate::ask::QuestionStatus::Answered`]).
1068    /// Once nothing is left in [`Task::blocked_by`], the task returns to
1069    /// whatever [`Task::blocked_from`] recorded - deciding *why* a task was
1070    /// blocked was `crate::conduct`'s job, but noticing a dependency resolved
1071    /// needs no model at all, and restoring the status it interrupted needs
1072    /// nothing more than what `block` already wrote down.
1073    ///
1074    /// A task blocked while `Running` restores to [`TaskStatus::Queued`]
1075    /// instead: whatever process was running it is gone by the time this
1076    /// runs, so there is nothing left to resume. A task with no recorded
1077    /// `blocked_from` - a pre-schema-4 record, or one blocked before this
1078    /// field existed - falls back to [`TaskStatus::Held`] when it still
1079    /// carries hold evidence ([`Task::hold_reason`] or [`Task::hold_source`],
1080    /// neither ever cleared by `block`), and to `Queued` otherwise: the same
1081    /// choice `block` itself would have recorded, reconstructed from what
1082    /// survived.
1083    ///
1084    /// A no-op, on purpose, for a task that is not [`TaskStatus::Blocked`]:
1085    /// `crate::daemon`'s deterministic resolver runs over every task on every
1086    /// poll, and a task that moved on for some other reason must not be
1087    /// dragged back by a stale id it still happens to carry.
1088    pub fn unblock(&mut self, resolved_id: &str) {
1089        if self.status != TaskStatus::Blocked {
1090            return;
1091        }
1092        self.blocked_by.retain(|id| id != resolved_id);
1093        self.restore_if_unblocked();
1094    }
1095
1096    /// A dependency this task waited on was deleted on purpose
1097    /// ([`Queue::remove`]): drop just that id from [`Task::blocked_by`] and,
1098    /// once nothing is left, return to the status the block interrupted, the
1099    /// same way [`Task::unblock`] does. Returns whether anything changed.
1100    ///
1101    /// Never a hold: the operator chose the deletion, so the dependent is not
1102    /// in trouble. Touches nothing but a `Blocked` task that actually names
1103    /// `deleted_id`, so a `Held` task (or a manual hold) is left exactly as it
1104    /// was. A dependency that vanished *without* going through a deletion is
1105    /// not this method's case - that stays a machine hold.
1106    pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
1107        if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
1108            return false;
1109        }
1110        self.blocked_by.retain(|id| id != deleted_id);
1111        self.restore_if_unblocked();
1112        true
1113    }
1114
1115    /// The shared tail of [`Task::unblock`] and [`Task::dependency_deleted`]:
1116    /// with nothing left in `blocked_by`, restore the interrupted status and
1117    /// clear the now-stale block bookkeeping.
1118    fn restore_if_unblocked(&mut self) {
1119        if self.blocked_by.is_empty() {
1120            self.status = match self.blocked_from {
1121                Some(TaskStatus::Running) => TaskStatus::Queued,
1122                Some(other) => other,
1123                None if self.hold_reason.is_some() || self.hold_source.is_some() => {
1124                    TaskStatus::Held
1125                }
1126                None => TaskStatus::Queued,
1127            };
1128            self.block_reason = None;
1129            self.blocked_from = None;
1130        }
1131    }
1132
1133    /// Record that a question `crate::conduct` asked about this task has been
1134    /// answered, so the answer's content — not just the fact that the
1135    /// question is gone — reaches the next conductor prompt and the next
1136    /// run's instruction. See [`Task::answers`].
1137    pub fn record_answer(&mut self, question: String, answer: String) {
1138        self.answers.push(AnsweredQuestion { question, answer });
1139    }
1140
1141    /// Requeue this task to reopen its last run as a review-only pass against
1142    /// `branch` (`crate::graph::Runner::review`) rather than competing from
1143    /// scratch. See [`Task::review_branch`].
1144    pub fn request_review(&mut self, branch: String) {
1145        self.release();
1146        self.review_branch = Some(branch);
1147    }
1148
1149    /// Requeue after a conductor chose a new competition. Unlike an ordinary
1150    /// operator release, this deliberately does not resume the old run.
1151    pub fn requeue(&mut self) {
1152        self.release();
1153        // A fresh competition is the opposite of a review of the old branch.
1154        self.review_branch = None;
1155        self.fresh_start = true;
1156    }
1157
1158    /// Hold this task because a takeover of `branch` was refused, keeping
1159    /// [`Task::review_branch`] so that once the operator has cleaned up and
1160    /// released it, the retry is a review of the same branch rather than a
1161    /// resume or a fresh competition. Spends no attempt.
1162    ///
1163    /// This is the only way a machine hold keeps a `review_branch`: the
1164    /// daemon takes it at the start of every attempt, so [`Task::release`]
1165    /// relies on that to tell this hold from every other.
1166    pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
1167        if branch.is_some() {
1168            self.review_branch = branch;
1169        }
1170        self.hold_machine(Some(reason));
1171    }
1172
1173    /// Change how urgently this task should run next.
1174    ///
1175    /// Refused once the task is `running`: priority only feeds the sort
1176    /// [`Queue::next_runnable`] does over tasks waiting to be claimed, and a
1177    /// running task has already left that pool. Accepting the write anyway
1178    /// would look like it worked while changing nothing until - and unless -
1179    /// this attempt fails and the task becomes runnable again, which is a
1180    /// surprise the phone should not hand back as a success.
1181    pub fn set_priority(&mut self, priority: i32) -> Result<()> {
1182        if self.status == TaskStatus::Running {
1183            bail!(
1184                "task {} is running; its priority cannot be changed until \
1185                 this attempt finishes",
1186                self.short()
1187            );
1188        }
1189        self.priority = priority;
1190        Ok(())
1191    }
1192
1193    /// Mark (or unmark) this task to interrupt whatever `magi serve` already
1194    /// has in flight, once `[daemon] pause_for_interrupts` is on. See
1195    /// [`Task::interrupt`].
1196    ///
1197    /// Setting it is restricted to a task the loop could pick up on its own
1198    /// right now - [`TaskStatus::runnable`] - for the same reason as
1199    /// [`Task::set_priority`]: a task already `running` has been claimed, and
1200    /// a task that is `done`, `held`, or `blocked` is not going to compete
1201    /// for the daemon's attention regardless of this flag. Unlike priority,
1202    /// this is never silently inert while `running` - it is refused outright,
1203    /// because the entire feature this flag drives (`crate::daemon`'s
1204    /// interrupt scheduler) is scoped to tasks still waiting to be claimed.
1205    /// Clearing it back to `false` carries no such risk and is always
1206    /// allowed, including on a task that moved on since it was set.
1207    pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
1208        if interrupt && !self.status.runnable() {
1209            bail!(
1210                "task {} is {}; only a queued or failed task can be marked \
1211                 to interrupt",
1212                self.short(),
1213                self.status.as_str()
1214            );
1215        }
1216        self.interrupt = interrupt;
1217        Ok(())
1218    }
1219
1220    /// Replace this task's title and instruction wholesale.
1221    ///
1222    /// Restricted to `queued` and `held`. A `running` task's instruction has
1223    /// already been handed to the graph, so a run in flight and the file on
1224    /// disk must not be allowed to disagree about what was asked; a `done` or
1225    /// `failed` task is a record of what actually happened and editing it
1226    /// after the fact would falsify that record. `id`, `created_at`,
1227    /// `source`, and `runs` are left untouched on purpose - an edit stands in
1228    /// for "delete and refile", and keeping the id, the timestamp, the
1229    /// attribution, and the run history is the entire reason it exists
1230    /// instead.
1231    pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
1232        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
1233            bail!(
1234                "task {} is {}; only a queued or held task's instruction can \
1235                 be edited",
1236                self.short(),
1237                self.status.as_str()
1238            );
1239        }
1240        self.title = title;
1241        self.instruction = instruction;
1242        Ok(())
1243    }
1244
1245    /// Record a run that produced a pull request without merging it.
1246    ///
1247    /// The task is held rather than retried, and it costs no further attempt
1248    /// either way. The work the task asked for exists: it is sitting on a
1249    /// branch, in a pull request, waiting for CI or for a person. Retrying
1250    /// would spend the whole competition budget a second time and then race a
1251    /// second branch against the pull request the first one opened - which is
1252    /// exactly what happened to run 01c2, whose finished and green pull request
1253    /// was re-competed from scratch four seconds after it opened.
1254    ///
1255    /// A pull request nobody merged is a request for a person, not a failure.
1256    ///
1257    /// Records `why` into [`Task::hold_reason`] as well as
1258    /// [`Task::last_error`], so the notification centre and `magi task show`
1259    /// say why the task is held rather than "no reason recorded". This is
1260    /// also the path a verified no-op with an outstanding `magi ask` question
1261    /// settles through - see `daemon::settle` and `daemon::settle_and_diagnose`,
1262    /// which append the question id to `hold_reason` when one is still open
1263    /// for the run.
1264    pub fn handed_off(&mut self, why: impl Into<String>) {
1265        let why = why.into();
1266        self.diagnostic = None;
1267        self.note_held();
1268        self.status = TaskStatus::Held;
1269        self.hold_source = Some(HoldSource::Machine);
1270        self.hold_reason = Some(why.clone());
1271        self.last_error = Some(why);
1272    }
1273
1274    /// Put a held or finished task back in line, with its attempt count reset
1275    /// so a release is a real second chance rather than an instant re-hold.
1276    /// The run history is kept: attempts reset, evidence does not.
1277    pub fn release(&mut self) {
1278        let refused_handover = self.status == TaskStatus::Held
1279            && self.hold_source == Some(HoldSource::Machine)
1280            && self.review_branch.is_some();
1281        self.status = TaskStatus::Queued;
1282        self.held_at = None;
1283        self.attempts = 0;
1284        self.last_error = None;
1285        self.park_reason = None;
1286        // Otherwise the next person who holds this task reads a reason that
1287        // belonged to whatever it was waiting on last time.
1288        self.hold_reason = None;
1289        self.hold_source = None;
1290        self.diagnostic = None;
1291        // A release also un-blocks: the dependency or question `blocked_by`
1292        // named may still be unresolved, but a human (or `crate::conduct`)
1293        // choosing to release the task overrides that wait outright, the same
1294        // as it overrides an ordinary hold.
1295        self.blocked_by.clear();
1296        self.block_reason = None;
1297        self.blocked_from = None;
1298        // A machine hold that still names a review branch is a refused
1299        // handover (see [`Task::hold_for_handover`]): the retry must review
1300        // that same branch. Any other release drops it.
1301        if !refused_handover {
1302            self.review_branch = None;
1303        }
1304        self.fresh_start = false;
1305    }
1306}
1307
1308/// Copy `src` into `dir` as `name`, or `stem-2.ext`, `stem-3.ext`, ... when
1309/// that is taken. `create_new` makes the collision check atomic and also
1310/// catches a case-insensitive filesystem. Returns the stored name and path.
1311fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
1312    use std::io::ErrorKind;
1313    let (stem, ext) = match name.rfind('.') {
1314        Some(i) if i > 0 => (&name[..i], &name[i..]),
1315        _ => (name, ""),
1316    };
1317    for n in 1u32.. {
1318        let candidate = if n == 1 {
1319            name.to_owned()
1320        } else {
1321            let suffix = format!("-{n}");
1322            let room = 64usize.saturating_sub(suffix.len() + ext.len());
1323            let stem: String = stem.chars().take(room).collect();
1324            format!("{stem}{suffix}{ext}")
1325        };
1326        if !crate::ask::valid_asset_name(&candidate) {
1327            bail!("no valid attachment name is left for `{name}`");
1328        }
1329        let path = dir.join(&candidate);
1330        match std::fs::OpenOptions::new()
1331            .write(true)
1332            .create_new(true)
1333            .open(&path)
1334        {
1335            Ok(mut out) => {
1336                let copied = std::fs::File::open(src)
1337                    .and_then(|mut input| std::io::copy(&mut input, &mut out));
1338                if let Err(e) = copied {
1339                    drop(out);
1340                    let _ = std::fs::remove_file(&path);
1341                    return Err(e).with_context(|| format!("copy {}", src.display()));
1342                }
1343                return Ok((candidate, path));
1344            }
1345            Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
1346            Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
1347        }
1348    }
1349    unreachable!("the counter never runs out")
1350}
1351
1352/// How long a task's write lock may stand before it is read as left behind by
1353/// a writer that died.
1354const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);
1355
1356/// Guard for [`Queue::lock_task`]; removes the lock file on drop, but only
1357/// while the file still carries this guard's owner token.
1358struct TaskLock {
1359    path: PathBuf,
1360    token: String,
1361}
1362
1363/// A token no other holder shares: pid, then a mix of entropy and a process
1364/// counter so threads born in the same nanosecond still differ.
1365fn owner_token() -> String {
1366    static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
1367    let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1368    let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
1369    format!("{}-{:016x}", std::process::id(), r.next_u64())
1370}
1371
1372/// The marker that serializes everyone removing the lock that carries `token`.
1373/// Hashed (FNV-1a) so an empty or foreign token is still a valid file name.
1374fn break_marker(path: &Path, token: &str) -> PathBuf {
1375    let mut h: u64 = 0xcbf29ce484222325;
1376    for b in token.bytes() {
1377        h ^= u64::from(b);
1378        h = h.wrapping_mul(0x100000001b3);
1379    }
1380    let mut name = path.as_os_str().to_owned();
1381    name.push(format!(".break-{h:016x}"));
1382    PathBuf::from(name)
1383}
1384
1385fn older_than_stale(path: &Path) -> bool {
1386    std::fs::metadata(path)
1387        .and_then(|m| m.modified())
1388        .ok()
1389        .and_then(|t| t.elapsed().ok())
1390        .is_some_and(|age| age > TASK_LOCK_STALE)
1391}
1392
1393/// How many levels of "a stale marker guarding a stale marker" are recovered
1394/// before giving up. Each level needs its own crashed remover, so the cap is
1395/// never reached without several independent crashes.
1396const MAX_MARKER_DEPTH: u8 = 3;
1397
1398/// Take the break marker for `token`, or `None` if another remover holds it.
1399/// The marker carries its own owner token. One older than [`TASK_LOCK_STALE`]
1400/// was left by a remover that died; it is cleared by the same compare-and-
1401/// remove discipline as a lock, guarded by a marker of its own, so a recoverer
1402/// that judged it stale cannot delete a fresh marker that replaced it.
1403fn take_marker(path: &Path, token: &str, depth: u8) -> Option<(PathBuf, String)> {
1404    use std::io::Write;
1405    let marker = break_marker(path, token);
1406    for _ in 0..2 {
1407        match std::fs::OpenOptions::new()
1408            .write(true)
1409            .create_new(true)
1410            .open(&marker)
1411        {
1412            Ok(mut f) => {
1413                let mine = owner_token();
1414                if f.write_all(mine.as_bytes()).is_err() {
1415                    drop(f);
1416                    let _ = std::fs::remove_file(&marker);
1417                    return None;
1418                }
1419                return Some((marker, mine));
1420            }
1421            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
1422                let judged = std::fs::read_to_string(&marker).ok();
1423                match judged.filter(|_| older_than_stale(&marker)) {
1424                    Some(judged) if depth < MAX_MARKER_DEPTH => {
1425                        remove_lock_if(&marker, &judged, true, depth + 1)?;
1426                    }
1427                    _ => return None,
1428                }
1429            }
1430            Err(_) => return None,
1431        }
1432    }
1433    None
1434}
1435
1436/// Remove the lock at `path` if it still carries `token` (and, when
1437/// `require_stale`, is still stale). Removal is serialized per token by the
1438/// break marker, so a remover that judged an old lock cannot delete the fresh
1439/// one that replaced it. `None` means the marker was busy: try again.
1440fn remove_lock_if(path: &Path, token: &str, require_stale: bool, depth: u8) -> Option<bool> {
1441    let (marker, mine) = take_marker(path, token, depth)?;
1442    let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
1443        && (!require_stale || older_than_stale(path));
1444    if still {
1445        let _ = std::fs::remove_file(path);
1446    }
1447    // Release the marker only if it is still ours.
1448    if depth >= MAX_MARKER_DEPTH {
1449        if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
1450            let _ = std::fs::remove_file(&marker);
1451        }
1452    } else {
1453        let _ = remove_lock_if(&marker, &mine, false, depth + 1);
1454    }
1455    Some(still)
1456}
1457
1458/// Break the lock judged stale while it held `judged`. Returns whether it was
1459/// removed; a lock that has since been replaced is left alone.
1460fn break_stale(path: &Path, judged: &str) -> bool {
1461    remove_lock_if(path, judged, true, 0).unwrap_or(false)
1462}
1463
1464impl Drop for TaskLock {
1465    fn drop(&mut self) {
1466        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
1467        loop {
1468            match remove_lock_if(&self.path, &self.token, false, 0) {
1469                Some(_) => return,
1470                None if std::time::Instant::now() > deadline => return,
1471                None => std::thread::sleep(std::time::Duration::from_millis(5)),
1472            }
1473        }
1474    }
1475}
1476
1477/// A queue on disk.
1478#[derive(Debug, Clone)]
1479pub struct Queue {
1480    root: PathBuf,
1481}
1482
1483impl Queue {
1484    /// The operator's queue, `<home>/queue`.
1485    pub fn open() -> Self {
1486        Self::at(crate::run::home().join("queue"))
1487    }
1488
1489    /// A queue at an explicit root. Tests use this; so could an operator who
1490    /// wants a queue per project.
1491    pub fn at(root: PathBuf) -> Self {
1492        Self { root }
1493    }
1494
1495    /// Directory holding the task files.
1496    pub fn root(&self) -> &Path {
1497        &self.root
1498    }
1499
1500    /// Path for one task id.
1501    pub fn path_of(&self, id: &str) -> PathBuf {
1502        self.root.join(format!("{id}.json"))
1503    }
1504
1505    /// Directory holding a task's attachments. A sibling of the task file,
1506    /// not a `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
1507    pub fn attachments_dir(&self, id: &str) -> PathBuf {
1508        self.root.join(format!("{id}.attachments"))
1509    }
1510
1511    /// Copy each of `sources` into `task`'s attachment directory and record
1512    /// the stored names on `task` (persist with [`Queue::put`]). Returns the
1513    /// names, in order.
1514    ///
1515    /// Every source is checked (a readable file, a name passing
1516    /// [`crate::ask::valid_asset_name`]) before anything is copied, and a
1517    /// copy that fails part-way removes only what this call created. A name
1518    /// already taken is never overwritten: it becomes `stem-2.ext`,
1519    /// `stem-3.ext`, ... A name that does not validate is refused rather than
1520    /// sanitised - the operator renames the file.
1521    pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
1522        let mut wanted = Vec::new();
1523        for src in sources {
1524            let name = src
1525                .file_name()
1526                .and_then(|n| n.to_str())
1527                .with_context(|| format!("`{}` has no usable file name", src.display()))?;
1528            if !crate::ask::valid_asset_name(name) {
1529                bail!(
1530                    "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
1531                     with no `..`; rename the file and try again"
1532                );
1533            }
1534            if !src.is_file() {
1535                bail!("attachment `{}` is not a file", src.display());
1536            }
1537            wanted.push((src, name));
1538        }
1539        let dir = self.attachments_dir(&task.id);
1540        let existed = dir.is_dir();
1541        std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
1542        let mut created: Vec<PathBuf> = Vec::new();
1543        let mut names = Vec::new();
1544        let mut copy_all = || -> Result<()> {
1545            for (src, name) in &wanted {
1546                let (stored, path) = copy_new(&dir, src, name)?;
1547                created.push(path);
1548                names.push(stored);
1549            }
1550            Ok(())
1551        };
1552        if let Err(e) = copy_all() {
1553            for path in &created {
1554                let _ = std::fs::remove_file(path);
1555            }
1556            if !existed {
1557                let _ = std::fs::remove_dir(&dir);
1558            }
1559            return Err(e);
1560        }
1561        task.attachments.extend(names.iter().cloned());
1562        Ok(names)
1563    }
1564
1565    /// [`Queue::attach`] then [`Queue::put`], undoing the copies when the
1566    /// record cannot be written. Without the undo the files sit in
1567    /// `<id>.attachments/` with no task pointing at them.
1568    ///
1569    /// Only what this call copied is removed (the names `attach` returned,
1570    /// and the directory itself only when it did not exist before and is
1571    /// empty), never an attachment an earlier save recorded. `task` is rolled
1572    /// back too, and the error the caller sees is `put`'s own. This relies on
1573    /// a failed `put` leaving the stored record untouched, which its
1574    /// write-then-rename guarantees.
1575    pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
1576        let dir = self.attachments_dir(&task.id);
1577        let existed = dir.is_dir();
1578        let before = task.attachments.len();
1579        let names = self.attach(task, sources)?;
1580        if let Err(e) = self.put(task) {
1581            for name in &names {
1582                let _ = std::fs::remove_file(dir.join(name));
1583            }
1584            if !existed {
1585                let _ = std::fs::remove_dir(&dir);
1586            }
1587            task.attachments.truncate(before);
1588            return Err(e);
1589        }
1590        Ok(names)
1591    }
1592
1593    /// Absolute paths of `task`'s attachments, whatever shape this queue's
1594    /// root has. Absolute because the prompt hands them to an agent whose
1595    /// working directory is somewhere else entirely.
1596    pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
1597        let dir = self.attachments_dir(&task.id);
1598        task.attachments
1599            .iter()
1600            .map(|n| {
1601                let p = dir.join(n);
1602                std::path::absolute(&p).unwrap_or(p)
1603            })
1604            .collect()
1605    }
1606
1607    /// Write a task, atomically, so a daemon killed mid-write leaves the
1608    /// previous state readable rather than a truncated file.
1609    ///
1610    /// Runs already recorded on disk that `task` does not carry are kept, not
1611    /// dropped: `magi run --task` links a run into a task from another
1612    /// process ([`Queue::link_run`]) while a daemon may hold a snapshot taken
1613    /// before it, and writing that snapshot back whole would erase the link.
1614    /// Nothing ever removes a run from a task, so the union loses nothing.
1615    pub fn put(&self, task: &mut Task) -> Result<()> {
1616        let _lock = self.lock_task(&task.id)?;
1617        self.put_unlocked(task)
1618    }
1619
1620    /// Write `task` only if no task with its id exists yet; `Ok(false)` when
1621    /// one does. Unlike [`Queue::put`] it never replaces a record, so a
1622    /// producer with a deterministic id cannot rewind a task that has since
1623    /// started running or finished.
1624    pub fn create_new(&self, task: &mut Task) -> Result<bool> {
1625        let _lock = self.lock_task(&task.id)?;
1626        if self.path_of(&task.id).exists() {
1627            return Ok(false);
1628        }
1629        self.put_unlocked(task)?;
1630        Ok(true)
1631    }
1632
1633    /// [`Queue::put`] for a caller already holding [`Queue::lock_task`].
1634    fn put_unlocked(&self, task: &mut Task) -> Result<()> {
1635        if let Ok(stored) = read_path(&self.path_of(&task.id)) {
1636            for run in stored.runs {
1637                if !task.runs.contains(&run) {
1638                    task.runs.push(run);
1639                }
1640            }
1641            // A completion notice already recorded is never rewound by an
1642            // older snapshot: that would report the same state twice.
1643            if let Some(n) = stored.chat_notice
1644                && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
1645            {
1646                task.chat_notice = Some(n);
1647            }
1648        }
1649        task.updated_at = Timestamp::now();
1650        std::fs::create_dir_all(&self.root)
1651            .with_context(|| format!("create {}", self.root.display()))?;
1652        let body = serde_json::to_string_pretty(task).context("serialize task")?;
1653        let path = self.path_of(&task.id);
1654        let tmp = path.with_extension("json.tmp");
1655        std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
1656        std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
1657        // A machine hold is news for the notification centre, filed beside
1658        // this queue (`<home>/queue` -> `<home>/notifications`) rather than
1659        // through a process-global, so a queue in a temp directory notifies
1660        // into that directory.
1661        if let (Some(notice), Some(home)) = (
1662            crate::notices::task_held(task),
1663            self.root.parent().filter(|p| !p.as_os_str().is_empty()),
1664        ) {
1665            crate::notices::raise_in(home, notice);
1666        }
1667        Ok(())
1668    }
1669
1670    /// Change the task `id` in place: read it inside its write lock, let `f`
1671    /// edit it, and write it back only when `f` returns `true`. The check a
1672    /// caller makes in `f` therefore sees the record as stored now, not a
1673    /// snapshot taken before another writer got in. Returns whether it wrote.
1674    pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
1675        let id = self.resolve_id(id)?;
1676        let _lock = self.lock_task(&id)?;
1677        let mut task = self.get(&id)?;
1678        if !f(&mut task) {
1679            return Ok(false);
1680        }
1681        self.put_unlocked(&mut task)?;
1682        Ok(true)
1683    }
1684
1685    /// Append `run` to the runs of the task `id` (or an unambiguous prefix),
1686    /// leaving everything else about it alone — see [`Task::link_run`]. Read
1687    /// and written back in one breath, because [`Queue::put`] replaces the
1688    /// whole record. Returns the task as stored.
1689    pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
1690        let id = self.resolve_id(id)?;
1691        // Read inside the lock, so the record written back is the current one
1692        // and a daemon's write cannot slip between the read and the write.
1693        let _lock = self.lock_task(&id)?;
1694        let mut task = self.get(&id)?;
1695        if task.link_run(run) {
1696            self.put_unlocked(&mut task)?;
1697        }
1698        Ok(task)
1699    }
1700
1701    /// Exclusive right to rewrite task `id`'s record, held until the guard
1702    /// drops. Every writer ([`Queue::put`], [`Queue::link_run`]) takes it, so
1703    /// a process linking a run and the daemon saving a transition are
1704    /// serialized instead of last-writer-wins. A lock older than
1705    /// [`TASK_LOCK_STALE`] belongs to a writer that died mid-update and is
1706    /// broken. Not the claim lock (`<id>.lock`), which a daemon holds for a
1707    /// whole run.
1708    fn lock_task(&self, id: &str) -> Result<TaskLock> {
1709        std::fs::create_dir_all(&self.root)
1710            .with_context(|| format!("create {}", self.root.display()))?;
1711        let path = self.root.join(format!("{id}.write-lock"));
1712        let started = std::time::Instant::now();
1713        loop {
1714            match std::fs::OpenOptions::new()
1715                .write(true)
1716                .create_new(true)
1717                .open(&path)
1718            {
1719                Ok(mut file) => {
1720                    use std::io::Write;
1721                    let token = owner_token();
1722                    if let Err(e) = file.write_all(token.as_bytes()) {
1723                        drop(file);
1724                        let _ = std::fs::remove_file(&path);
1725                        return Err(e).with_context(|| format!("lock {}", path.display()));
1726                    }
1727                    return Ok(TaskLock { path, token });
1728                }
1729                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
1730                    // Content first, then age: a lock replaced in between
1731                    // reads fresh, never stale.
1732                    let judged = std::fs::read_to_string(&path).ok();
1733                    let broken = judged
1734                        .filter(|_| older_than_stale(&path))
1735                        .is_some_and(|judged| break_stale(&path, &judged));
1736                    // A break that does not happen (marker busy, recovery
1737                    // exhausted) is a wait like any other, so it meets the
1738                    // same deadline instead of spinning forever.
1739                    if !broken {
1740                        if started.elapsed() > TASK_LOCK_STALE {
1741                            bail!("could not lock task {id}");
1742                        }
1743                        std::thread::sleep(std::time::Duration::from_millis(15));
1744                    }
1745                }
1746                // On Windows, `create_new` on a file whose delete is still
1747                // pending fails with ERROR_ACCESS_DENIED, not `AlreadyExists`.
1748                // Treat it as a race and retry until the deadline.
1749                Err(e)
1750                    if e.kind() == std::io::ErrorKind::PermissionDenied
1751                        && started.elapsed() <= TASK_LOCK_STALE =>
1752                {
1753                    std::thread::sleep(std::time::Duration::from_millis(15));
1754                }
1755                Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
1756            }
1757        }
1758    }
1759
1760    /// Load a task by id or unambiguous id prefix.
1761    pub fn get(&self, id: &str) -> Result<Task> {
1762        let resolved = self.resolve_id(id)?;
1763        read_path(&self.path_of(&resolved))
1764    }
1765
1766    /// Remove a task, and the claim lock that belongs to it.
1767    ///
1768    /// `in_flight` comes from the caller — a live daemon's heartbeat naming
1769    /// this task — because the task's own `running` status cannot answer the
1770    /// question. A daemon killed mid-competition leaves the status at
1771    /// `running` and an orphaned `.lock` behind, and a guard that trusted
1772    /// either would make the task undeletable for good: the phone showed
1773    /// exactly that, refusing a task whose daemon had been gone for an hour.
1774    ///
1775    /// So the lock is removed with the task rather than respected. Any lock
1776    /// still there once no live daemon claims the task is by definition stale,
1777    /// and leaving it would make a deleted task look claimed to
1778    /// [`Queue::claim`] and to whoever reads the directory.
1779    ///
1780    /// A deletion is deliberate, so whatever was `blocked` on the id is
1781    /// resolved, not held: a `<id>.removed` tombstone is written first (it
1782    /// is what tells [`missing_blockers`] and the daemon's resolver that this
1783    /// id was deleted on purpose, so a pass that races this call cannot
1784    /// mistake it for a vanished dependency), the record is deleted (the
1785    /// tombstone is taken back if that fails), and then each dependent is
1786    /// rewritten under its claim - see [`Queue::release_dependents_of`].
1787    /// Best-effort for the dependents: one claimed by something else right
1788    /// now, or whose write fails, is released silently by
1789    /// `crate::daemon::resolve_blockers` (or `crate::triage::run_once`) on
1790    /// its next pass, because the tombstone is still there. It does not fail
1791    /// this removal.
1792    ///
1793    /// `questions` is the store [`missing_blockers`] checks a `blocked_by` id
1794    /// against - the same store the caller already resolves `id`'s own home
1795    /// from, passed in rather than reopened here so a test queue at an
1796    /// explicit root never reads the operator's real questions directory.
1797    pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
1798        let _ = questions;
1799        let resolved = self.resolve_id(id)?;
1800        if in_flight {
1801            bail!("task {resolved} is being run by a live daemon right now");
1802        }
1803        self.write_tombstone(&resolved)?;
1804        if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
1805        {
1806            // A record that is still there was not deleted: take the marker
1807            // back. One already gone keeps it, so the deletion still reads
1808            // as deliberate.
1809            if self.path_of(&resolved).exists() {
1810                let _ = std::fs::remove_file(self.tombstone_path(&resolved));
1811            }
1812            return Err(e);
1813        }
1814        let (released, still_blocked) = self.release_dependents_of(&resolved);
1815        Ok(Removal {
1816            id: resolved,
1817            released,
1818            still_blocked,
1819        })
1820    }
1821
1822    /// Path of the marker saying task `id` was deleted on purpose. Not a
1823    /// `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
1824    fn tombstone_path(&self, id: &str) -> PathBuf {
1825        self.root.join(format!("{id}.removed"))
1826    }
1827
1828    fn write_tombstone(&self, id: &str) -> Result<()> {
1829        let path = self.tombstone_path(id);
1830        let tmp = self.root.join(format!("{id}.removed.tmp"));
1831        std::fs::write(&tmp, b"")
1832            .and_then(|()| std::fs::rename(&tmp, &path))
1833            .with_context(|| format!("write {}", path.display()))
1834    }
1835
1836    /// Was `id` deleted through [`Queue::remove`] (and is its record gone)?
1837    /// A live record always wins: a tombstone never makes an existing task
1838    /// look deleted.
1839    fn was_deleted(&self, id: &str) -> bool {
1840        self.tombstone_path(id).is_file() && !self.path_of(id).exists()
1841    }
1842
1843    /// Apply [`Task::dependency_deleted`] to `task` for every `blocked_by` id
1844    /// that [`Queue::remove`] deleted. Returns those ids. The caller holds
1845    /// the task's claim and saves it.
1846    pub fn apply_deleted_blockers(&self, task: &mut Task) -> Vec<String> {
1847        let deleted = deleted_blockers(self, &task.blocked_by);
1848        deleted
1849            .into_iter()
1850            .filter(|id| task.dependency_deleted(id))
1851            .collect()
1852    }
1853
1854    /// Record, as a fixed-wording info notice, that `dependent` stopped
1855    /// waiting on the deleted task `deleted`. Best-effort and beside the
1856    /// queue, like the hold notice [`Queue::put`] files.
1857    pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
1858        let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
1859            return;
1860        };
1861        let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
1862        let waiting = dependent.status == TaskStatus::Blocked;
1863        let message = match (ja, waiting) {
1864            (true, false) => format!(
1865                "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
1866                dependent.short(),
1867                short(deleted)
1868            ),
1869            (true, true) => format!(
1870                "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
1871                dependent.short(),
1872                short(deleted)
1873            ),
1874            (false, false) => format!(
1875                "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
1876                dependent.short(),
1877                short(deleted)
1878            ),
1879            (false, true) => format!(
1880                "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
1881                dependent.short(),
1882                short(deleted)
1883            ),
1884        };
1885        crate::notices::raise_in(
1886            home,
1887            crate::notices::Notice::info(
1888                &format!("unblocked:{}:{}", dependent.id, deleted),
1889                message,
1890            )
1891            .link(crate::notices::Link::Task {
1892                id: dependent.id.clone(),
1893            }),
1894        );
1895    }
1896
1897    /// Resolve every `blocked` task naming `dependency` (just deleted) with
1898    /// [`Task::dependency_deleted`]. Returns the ids released to another
1899    /// status and those that remain `blocked` on something else. Each
1900    /// dependent is re-read inside its claim, never written from a stale copy.
1901    fn release_dependents_of(&self, dependency: &str) -> (Vec<String>, Vec<String>) {
1902        let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
1903        for listed in self.list() {
1904            if listed.status != TaskStatus::Blocked
1905                || !listed.blocked_by.iter().any(|b| b == dependency)
1906            {
1907                continue;
1908            }
1909            let Ok(_claim) = self.claim(&listed.id) else {
1910                continue;
1911            };
1912            let Ok(mut task) = self.get(&listed.id) else {
1913                continue;
1914            };
1915            if !task.dependency_deleted(dependency) {
1916                continue;
1917            }
1918            if self.put(&mut task).is_err() {
1919                continue;
1920            }
1921            self.note_dependency_deleted(&task, dependency);
1922            if task.status == TaskStatus::Blocked {
1923                still_blocked.push(task.id.clone());
1924            } else {
1925                released.push(task.id.clone());
1926            }
1927        }
1928        (released, still_blocked)
1929    }
1930
1931    /// The attachment-and-record half of [`Queue::remove`]. `remove_record` is
1932    /// how the record file is deleted; production passes `remove_file`, and a
1933    /// test passes a failing one to prove the attachments come back.
1934    fn remove_record_with_attachments(
1935        &self,
1936        resolved: &str,
1937        remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
1938    ) -> Result<()> {
1939        // Neither step may strand the other. The attachments are set aside by a
1940        // rename (cheap, and undone if the record cannot be removed), then the
1941        // record goes, and only then is the set-aside directory deleted for
1942        // real. A failure before the record is gone loses nothing; a failure
1943        // after it leaves at most an orphan `.removing` directory, which the
1944        // next removal sweeps.
1945        self.sweep_removed_attachments();
1946        let attachments = self.attachments_dir(resolved);
1947        let aside = self.root.join(format!("{resolved}.attachments.removing"));
1948        let moved = match std::fs::rename(&attachments, &aside) {
1949            Ok(()) => true,
1950            Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
1951            Err(e) => {
1952                return Err(e).with_context(|| format!("remove {}", attachments.display()));
1953            }
1954        };
1955        let path = self.path_of(resolved);
1956        if let Err(e) = remove_record(&path) {
1957            if moved {
1958                let _ = std::fs::rename(&aside, &attachments);
1959            }
1960            return Err(e).with_context(|| format!("remove {}", path.display()));
1961        }
1962        if moved {
1963            if let Err(e) = std::fs::remove_dir_all(&aside) {
1964                tracing::warn!("leftover attachments {}: {e}", aside.display());
1965            }
1966        }
1967        let lock = self.lock_path(resolved);
1968        if let Err(e) = std::fs::remove_file(&lock) {
1969            if e.kind() != std::io::ErrorKind::NotFound {
1970                return Err(e).with_context(|| format!("remove {}", lock.display()));
1971            }
1972        }
1973        Ok(())
1974    }
1975
1976    /// Delete `*.attachments.removing` directories a previous [`Queue::remove`]
1977    /// could not finish deleting. The task record is already gone by then, so
1978    /// the id no longer resolves and the removal cannot be retried by name;
1979    /// every later removal sweeps them instead. A directory whose record still
1980    /// exists belongs to a removal in progress and is left alone. Best-effort.
1981    fn sweep_removed_attachments(&self) {
1982        let Ok(entries) = std::fs::read_dir(&self.root) else {
1983            return;
1984        };
1985        for entry in entries.flatten() {
1986            let name = entry.file_name();
1987            let name = name.to_string_lossy();
1988            let Some(id) = name.strip_suffix(".attachments.removing") else {
1989                continue;
1990            };
1991            // A record still on disk means a removal is mid-flight (or about
1992            // to roll back): the directory is its to delete or restore.
1993            if !self.path_of(id).exists() {
1994                if let Err(e) = std::fs::remove_dir_all(entry.path()) {
1995                    tracing::warn!("leftover attachments {}: {e}", entry.path().display());
1996                }
1997            }
1998        }
1999        self.sweep_tombstones();
2000    }
2001
2002    /// Delete `*.removed` markers that are no longer needed: the record is
2003    /// gone, no task's `blocked_by` names the id, and the marker is older than
2004    /// [`TOMBSTONE_GRACE`]. The age is what makes this safe against a
2005    /// concurrent writer: the task list read here can be stale, so a dependent
2006    /// that was just blocked on an id (a conductor applying a decision while a
2007    /// removal is under way) would otherwise lose the marker and be held as a
2008    /// vanished dependency. A marker whose record still exists belongs to a
2009    /// removal in progress and is never touched.
2010    fn sweep_tombstones(&self) {
2011        let Ok(entries) = std::fs::read_dir(&self.root) else {
2012            return;
2013        };
2014        let tasks = self.list();
2015        for entry in entries.flatten() {
2016            let name = entry.file_name();
2017            let name = name.to_string_lossy();
2018            let Some(id) = name.strip_suffix(".removed") else {
2019                continue;
2020            };
2021            if self.path_of(id).exists()
2022                || tasks.iter().any(|t| t.blocked_by.iter().any(|b| b == id))
2023            {
2024                continue;
2025            }
2026            let old = entry
2027                .metadata()
2028                .and_then(|m| m.modified())
2029                .ok()
2030                .and_then(|m| m.elapsed().ok())
2031                .is_some_and(|age| age >= TOMBSTONE_GRACE);
2032            if old {
2033                let _ = std::fs::remove_file(entry.path());
2034            }
2035        }
2036    }
2037
2038    /// Path of the claim lock for a task. One definition, so `claim` and
2039    /// `remove` cannot end up naming different files.
2040    fn lock_path(&self, id: &str) -> PathBuf {
2041        self.root.join(format!("{id}.lock"))
2042    }
2043
2044    /// Every task on disk, highest priority first and newest first within a
2045    /// priority. This is what `magi task list` and `GET /api/queue` print, so
2046    /// a raised priority has to move a task here the moment it is saved, not
2047    /// only in [`Queue::next_runnable`]'s own ordering - the operator reading
2048    /// the backlog and the loop about to drain it must agree on what "first"
2049    /// means. Every existing task defaults to priority 0, so this is a no-op
2050    /// change from the old newest-first order for a queue nobody has
2051    /// reprioritised.
2052    ///
2053    /// Unreadable files are skipped rather than fatal: one corrupt task must
2054    /// not take the queue - or the web UI, or an unattended daemon - down
2055    /// with it.
2056    pub fn list(&self) -> Vec<Task> {
2057        let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
2058            .into_iter()
2059            .flatten()
2060            .flatten()
2061            .map(|e| e.path())
2062            .filter(|p| p.extension().is_some_and(|x| x == "json"))
2063            .filter_map(|p| read_path(&p).ok())
2064            .collect();
2065        tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
2066        tasks
2067    }
2068
2069    /// Runs that a later attempt at the same task replaced, mapped to the id
2070    /// of the attempt that replaced them.
2071    ///
2072    /// A task keeps its attempts in order, and the deck showed them as two
2073    /// cards with the same title and no hint which was which: yukimemi asked
2074    /// why `stalled` and `blocked` appeared twice for one task, and the
2075    /// answer - "those are two tries, and the second one exists because of a
2076    /// bug since fixed" - was not on the screen anywhere.
2077    ///
2078    /// Read from the queue rather than stored on the run, because the
2079    /// ordering is the queue's fact: a `RunState` has no idea another attempt
2080    /// happened after it.
2081    pub fn superseded(&self) -> HashMap<String, String> {
2082        let mut by = HashMap::new();
2083        for task in self.list() {
2084            for earlier in &task.runs {
2085                if let Some(later) = task.successor_of(earlier) {
2086                    by.insert(earlier.clone(), later.clone());
2087                }
2088            }
2089        }
2090        by
2091    }
2092
2093    /// Whether `run` is an earlier attempt a later one replaced, and if so
2094    /// the id of that later attempt.
2095    ///
2096    /// Same walk as [`Queue::superseded`], narrowed to one run: a run detail
2097    /// page asks about exactly one run at a time, and this keeps that call
2098    /// site from building (and discarding) the whole map's `HashMap` just to
2099    /// read one entry out of it.
2100    pub fn superseded_by(&self, run: &str) -> Option<String> {
2101        for task in self.list() {
2102            if task.runs.iter().any(|r| r == run) {
2103                return task.successor_of(run).cloned();
2104            }
2105        }
2106        None
2107    }
2108
2109    /// The task's own most recent attempt, when `run` belongs to that task
2110    /// but is not already that attempt.
2111    ///
2112    /// Distinct from [`Queue::superseded_by`], which names only the very
2113    /// next attempt: a chain of retries (A superseded by B superseded by C)
2114    /// leaves an older run pointing at an intermediate one that may itself
2115    /// be unresolved, and a run's own detail page needs to know where the
2116    /// task's story currently stands - the chain's current head, C - not an
2117    /// attempt in the middle of it that a client would otherwise have to
2118    /// walk to by hand.
2119    pub fn latest_attempt(&self, run: &str) -> Option<String> {
2120        for task in self.list() {
2121            if task.runs.iter().any(|r| r == run) {
2122                return task.runs.last().filter(|last| **last != run).cloned();
2123            }
2124        }
2125        None
2126    }
2127
2128    /// The task a daemon should run next, or `None` when the queue is idle.
2129    ///
2130    /// Highest priority first, oldest first within a priority, so a burst of
2131    /// agent-filed work cannot starve the task a human filed this morning.
2132    pub fn next_runnable(&self) -> Option<Task> {
2133        let mut runnable: Vec<Task> = self
2134            .list()
2135            .into_iter()
2136            .filter(|t| t.status.runnable())
2137            .collect();
2138        runnable.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)));
2139        runnable.into_iter().next()
2140    }
2141
2142    /// Take exclusive ownership of a task.
2143    ///
2144    /// The lock is a `create_new` file next to the task, which is atomic on
2145    /// every platform magi targets. It exists so two daemons - or a daemon and
2146    /// a human running `magi run` - cannot drive one task into two competing
2147    /// runs. The returned guard releases on drop, including on panic.
2148    pub fn claim(&self, id: &str) -> Result<Claim> {
2149        std::fs::create_dir_all(&self.root)
2150            .with_context(|| format!("create {}", self.root.display()))?;
2151        let path = self.lock_path(id);
2152        match std::fs::OpenOptions::new()
2153            .write(true)
2154            .create_new(true)
2155            .open(&path)
2156        {
2157            Ok(mut f) => {
2158                use std::io::Write as _;
2159                // Best effort: the pid is for the human looking at a stale lock.
2160                let _ = writeln!(f, "{}", std::process::id());
2161                Ok(Claim { path })
2162            }
2163            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
2164                bail!("task {id} is already claimed ({} exists)", path.display())
2165            }
2166            Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
2167        }
2168    }
2169
2170    /// Expand an id prefix to exactly one task id.
2171    pub fn resolve_id(&self, prefix: &str) -> Result<String> {
2172        if self.path_of(prefix).is_file() {
2173            return Ok(prefix.to_owned());
2174        }
2175        let hits: Vec<String> = self
2176            .list()
2177            .into_iter()
2178            .map(|t| t.id)
2179            .filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
2180            .collect();
2181        match hits.len() {
2182            1 => Ok(hits.into_iter().next().expect("exactly one hit")),
2183            0 => bail!("no task matches `{prefix}`"),
2184            _ => bail!(
2185                "`{prefix}` matches {} tasks: {}",
2186                hits.len(),
2187                hits.join(", ")
2188            ),
2189        }
2190    }
2191
2192    /// Change detection token for the queue.
2193    ///
2194    /// Combines file names and modification times of all task files in the
2195    /// queue, so adding, modifying, or deleting any task — even an older one —
2196    /// moves the revision and notifies connected clients via the change stream.
2197    /// Returns 0 when the queue is completely empty.
2198    pub fn revision(&self) -> u64 {
2199        self.revision_excluding(&std::collections::BTreeSet::new())
2200    }
2201
2202    /// [`Queue::revision`] ignoring the files of the named task ids, for a
2203    /// caller that tracks those tasks' content some other way.
2204    pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
2205        use std::hash::{Hash as _, Hasher as _};
2206
2207        let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
2208            .into_iter()
2209            .flatten()
2210            .flatten()
2211            .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
2212            .filter(|e| {
2213                let path = e.path();
2214                !path
2215                    .file_stem()
2216                    .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
2217            })
2218            .filter_map(|e| {
2219                let name = e.file_name().to_string_lossy().into_owned();
2220                let mtime = e
2221                    .metadata()
2222                    .ok()?
2223                    .modified()
2224                    .ok()?
2225                    .duration_since(std::time::UNIX_EPOCH)
2226                    .ok()?
2227                    .as_millis() as u64;
2228                Some((name, mtime))
2229            })
2230            .collect();
2231
2232        if entries.is_empty() {
2233            return 0;
2234        }
2235
2236        entries.sort_unstable();
2237        let mut hasher = std::hash::DefaultHasher::new();
2238        for (name, mtime) in &entries {
2239            name.hash(&mut hasher);
2240            mtime.hash(&mut hasher);
2241        }
2242        let h = hasher.finish();
2243        if h == 0 { 1 } else { h }
2244    }
2245}
2246
2247/// How long a deletion marker is kept after its record is gone, even with
2248/// nothing referencing it. See [`Queue::sweep_tombstones`].
2249const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);
2250
2251/// What [`Queue::remove`] did, beyond deleting the named task's own file.
2252#[derive(Debug, Clone)]
2253pub struct Removal {
2254    /// The id actually removed - `id` expanded from a prefix, if it was one.
2255    pub id: String,
2256    /// Every task that was `blocked` on [`Removal::id`] and left the block
2257    /// (back to the status it interrupted) because that was its only
2258    /// remaining dependency.
2259    pub released: Vec<String>,
2260    /// Every task that named [`Removal::id`] and is still `blocked` on
2261    /// another dependency.
2262    pub still_blocked: Vec<String>,
2263}
2264
2265/// Exclusive ownership of a task, released on drop.
2266#[derive(Debug)]
2267pub struct Claim {
2268    path: PathBuf,
2269}
2270
2271impl Drop for Claim {
2272    fn drop(&mut self) {
2273        let _ = std::fs::remove_file(&self.path);
2274    }
2275}
2276
2277/// The first line of a task, trimmed to a title. Used when the caller gives a
2278/// body but no title, which is the normal case for an agent piping a file in.
2279pub fn title_from(instruction: &str, max: usize) -> String {
2280    let Some(line) = first_line(instruction) else {
2281        return "(empty task)".to_owned();
2282    };
2283    if line.chars().count() <= max {
2284        return line.to_owned();
2285    }
2286    let head: String = line.chars().take(max.saturating_sub(1)).collect();
2287    format!("{head}…")
2288}
2289
2290/// The first non-blank line, whatever it is. A markdown heading is the
2291/// task's own summary - agents pipe in `# Rework the config loader` and mean
2292/// exactly that - so it is preferred over the prose beneath it rather than
2293/// skipped as decoration. Leading list and heading markers are stripped
2294/// because they are syntax, not words. `None` when nothing is left.
2295pub(crate) fn first_line(instruction: &str) -> Option<&str> {
2296    let line = instruction
2297        .lines()
2298        .map(str::trim)
2299        .find(|l| !l.is_empty())?
2300        .trim_start_matches(['#', '-', '*', '>', ' '])
2301        .trim();
2302    (!line.is_empty()).then_some(line)
2303}
2304
2305fn read_path(path: &Path) -> Result<Task> {
2306    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
2307    let task: Task =
2308        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
2309    // Greater-than, not not-equal: every field added since schema 1 carries
2310    // `#[serde(default)]`, so an older task has nothing to say about it and
2311    // defaulting is exactly as good a reading as a value that build never had
2312    // a chance to write. Only a schema *ahead* of this build - a meaning it
2313    // cannot possibly know - is refused rather than guessed at.
2314    if task.schema > SCHEMA {
2315        bail!(
2316            "task {} was written by a different magi (schema {}, this build \
2317             speaks {SCHEMA})",
2318            task.id,
2319            task.schema
2320        );
2321    }
2322    Ok(task)
2323}
2324
2325/// Ids inside a `blocked_by` list that name neither an existing task file nor
2326/// an existing question file - a dependency deleted (`magi task rm`, or by
2327/// hand) while something was still waiting on it.
2328///
2329/// Existence is decided by [`Queue::path_of`]/[`Questions::path_of`]
2330/// `is_file()` alone, never by [`Queue::get`]/[`Questions::get`] succeeding:
2331/// those also fail on a merely unreadable file - mid-write, corrupt, or from
2332/// a schema ahead of this build (see [`read_path`]) - and misreading "cannot
2333/// read it right now" as "it was deleted" would quarantine a task over a
2334/// transient failure. `blocked_by` always carries a full id, written by
2335/// `crate::conduct` or `crate::triage` from a real task's or question's own
2336/// `id`/`short`, never a prefix a caller typed - so the exact-path check is
2337/// complete on its own, with no [`Queue::resolve_id`] fallback needed.
2338pub fn missing_blockers(
2339    queue: &Queue,
2340    questions: &Questions,
2341    blocked_by: &[String],
2342) -> Vec<String> {
2343    blocked_by
2344        .iter()
2345        .filter(|id| {
2346            !queue.path_of(id).is_file()
2347                && !questions.path_of(id).is_file()
2348                && !queue.was_deleted(id)
2349        })
2350        .cloned()
2351        .collect()
2352}
2353
2354/// Ids in `blocked_by` that [`Queue::remove`] deleted on purpose: tombstoned
2355/// and with no record. A live task is never included, tombstone or not. These
2356/// are resolved with [`Task::dependency_deleted`], not held; an id that
2357/// vanished without a tombstone stays [`missing_blockers`]' to report.
2358pub fn deleted_blockers(queue: &Queue, blocked_by: &[String]) -> Vec<String> {
2359    blocked_by
2360        .iter()
2361        .filter(|id| queue.was_deleted(id))
2362        .cloned()
2363        .collect()
2364}
2365
2366/// The `hold_reason` text for a task quarantined because one or more of its
2367/// `blocked_by` ids no longer exist. Shared by `crate::daemon::resolve_blockers`,
2368/// `crate::triage::run_once`, and [`Queue::remove`]'s own dependent
2369/// quarantine, so the three call sites read as the same event to an operator
2370/// looking at `magi task show` rather than three different wordings for it.
2371///
2372/// Names the full original `blocked_by` list, not just `missing` - a task
2373/// quarantined here can also have named a dependency that was still
2374/// perfectly valid, and [`Task::hold_machine`] clears `blocked_by` on the way
2375/// in, so this text is the only place that information survives for an
2376/// operator deciding whether to release the task outright.
2377pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
2378    missing_blocker_hold_reason_in(blocked_by, missing, "en")
2379}
2380
2381/// [`missing_blocker_hold_reason`] in `language` (Japanese, else English).
2382pub fn missing_blocker_hold_reason_in(
2383    blocked_by: &[String],
2384    missing: &[String],
2385    language: &str,
2386) -> String {
2387    if crate::lang::is_japanese(language) {
2388        format!(
2389            "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
2390            blocked_by.join(", "),
2391            missing.join(", "),
2392        )
2393    } else {
2394        format!(
2395            "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
2396            blocked_by.join(", "),
2397            missing.join(", "),
2398        )
2399    }
2400}
2401
2402/// The last dash-separated part of an id, the form people say aloud.
2403pub fn short(id: &str) -> &str {
2404    id.split('-').next_back().unwrap_or(id)
2405}
2406
2407fn new_id() -> String {
2408    let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
2409    let seed = crate::rng::entropy();
2410    format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
2411}
2412
2413#[cfg(test)]
2414mod tests {
2415    #[test]
2416    fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
2417        let mut base = RunOverrides {
2418            merge: Some("pr".to_owned()),
2419            candidates: Some(3),
2420            ..RunOverrides::default()
2421        };
2422        base.merge_over(&RunOverrides {
2423            merge: Some("none".to_owned()),
2424            seed: Some(7),
2425            ..RunOverrides::default()
2426        });
2427        assert_eq!(base.merge.as_deref(), Some("none"));
2428        assert_eq!(base.candidates, Some(3));
2429        assert_eq!(base.seed, Some(7));
2430    }
2431
2432    #[test]
2433    fn an_already_landed_task_is_done_with_its_attempt_refunded() {
2434        let mut t = task("relanded");
2435        t.attempts = 1;
2436        t.status = TaskStatus::Running;
2437        t.already_landed("already in main as 0e368de");
2438        assert_eq!(t.status, TaskStatus::Done);
2439        assert_eq!(t.attempts, 0);
2440        assert!(t.hold_reason.is_none());
2441        assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
2442    }
2443
2444    #[test]
2445    fn missing_blocker_reason_follows_the_language() {
2446        let b = vec!["a".to_owned()];
2447        let en = missing_blocker_hold_reason_in(&b, &b, "en");
2448        assert_eq!(en, missing_blocker_hold_reason(&b, &b));
2449        assert!(en.starts_with("blocked on a"));
2450        assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
2451        assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
2452    }
2453
2454    use super::*;
2455
2456    #[test]
2457    fn triage_applied_survives_release_and_old_records_read_as_empty() {
2458        let mut t = Task::new(
2459            "t".to_owned(),
2460            "i".to_owned(),
2461            PathBuf::from("r"),
2462            Source::Human,
2463        );
2464        t.mark_triage_applied("q1");
2465        t.mark_triage_applied("q1");
2466        t.hold_machine(Some("x".to_owned()));
2467        t.release();
2468        assert_eq!(t.triage_applied, ["q1"]);
2469        assert!(t.triage_applied("q1") && !t.triage_applied("q2"));
2470
2471        let mut v = serde_json::to_value(&t).unwrap();
2472        v.as_object_mut().unwrap().remove("triage_applied");
2473        let old: Task = serde_json::from_value(v).unwrap();
2474        assert!(old.triage_applied.is_empty());
2475    }
2476
2477    #[test]
2478    fn task_counts_of_empty_is_all_zero() {
2479        assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
2480    }
2481
2482    #[test]
2483    fn task_counts_of_tallies_every_status() {
2484        let mut queued = Task::new(
2485            "q".to_owned(),
2486            "i".to_owned(),
2487            PathBuf::from("."),
2488            Source::Human,
2489        );
2490        queued.status = TaskStatus::Queued;
2491        let mut running = queued.clone();
2492        running.status = TaskStatus::Running;
2493        let mut done = queued.clone();
2494        done.status = TaskStatus::Done;
2495        let mut failed = queued.clone();
2496        failed.status = TaskStatus::Failed;
2497        let mut held = queued.clone();
2498        held.status = TaskStatus::Held;
2499        let mut blocked = queued.clone();
2500        blocked.status = TaskStatus::Blocked;
2501
2502        let mut parked = queued.clone();
2503        parked.status = TaskStatus::Parked;
2504
2505        let counts = TaskCounts::of(&[
2506            queued,
2507            running,
2508            done.clone(),
2509            done,
2510            failed,
2511            held,
2512            blocked,
2513            parked,
2514        ]);
2515        assert_eq!(
2516            counts,
2517            TaskCounts {
2518                queued: 1,
2519                running: 1,
2520                done: 2,
2521                failed: 1,
2522                held: 1,
2523                blocked: 1,
2524                parked: 1,
2525            }
2526        );
2527    }
2528
2529    /// A queue of its own, with no process-global state - which is the point of
2530    /// `Queue::at`, and why these can run in parallel.
2531    fn queue() -> (tempfile::TempDir, Queue) {
2532        let dir = tempfile::tempdir().unwrap();
2533        let q = Queue::at(dir.path().join("queue"));
2534        (dir, q)
2535    }
2536
2537    #[test]
2538    fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
2539        let dir = tempfile::tempdir().unwrap();
2540        let q = Queue::at(dir.path().join("queue"));
2541        let mut t = task("held");
2542        q.put(&mut t).unwrap();
2543        assert_eq!(
2544            crate::notices::Notices::at(dir.path().join("notifications"))
2545                .list()
2546                .len(),
2547            0
2548        );
2549        t.hold_machine(Some("out of attempts".to_owned()));
2550        q.put(&mut t).unwrap();
2551        let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
2552        assert_eq!(listed.len(), 1);
2553        assert!(listed[0].message.contains("out of attempts"));
2554    }
2555
2556    fn task(title: &str) -> Task {
2557        Task::new(
2558            title.to_owned(),
2559            format!("do {title}"),
2560            PathBuf::from("."),
2561            Source::Human,
2562        )
2563    }
2564
2565    #[test]
2566    fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
2567        let mut t = task("retried");
2568        assert!(
2569            t.earlier_attempts().is_empty(),
2570            "a first attempt takes nothing over"
2571        );
2572        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
2573        assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
2574        // What the web note shows and what a takeover acts on are one rule.
2575        assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
2576        assert_eq!(t.successor_of("bbbb"), None);
2577        assert_eq!(t.successor_of("zzzz"), None);
2578    }
2579
2580    #[test]
2581    fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
2582        let (_dir, q) = queue();
2583        let mut t = task("retried");
2584        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
2585        q.put(&mut t).unwrap();
2586
2587        assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
2588        assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
2589        assert_eq!(
2590            q.superseded_by("cccc"),
2591            None,
2592            "the latest attempt replaces nothing"
2593        );
2594        assert_eq!(
2595            q.superseded_by("never-heard-of-it"),
2596            None,
2597            "a run belonging to no task on this queue is not superseded"
2598        );
2599
2600        let mut by = HashMap::new();
2601        by.insert("aaaa".to_owned(), "bbbb".to_owned());
2602        by.insert("bbbb".to_owned(), "cccc".to_owned());
2603        assert_eq!(
2604            q.superseded(),
2605            by,
2606            "the whole-map and single-run forms must agree"
2607        );
2608    }
2609
2610    #[test]
2611    fn superseded_attempts_is_empty_until_the_task_is_done() {
2612        let mut t = task("retried");
2613        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
2614        t.status = TaskStatus::Failed;
2615        assert_eq!(
2616            t.superseded_attempts(true),
2617            &[] as &[String],
2618            "a task still retrying has no attempt yet that a later one made moot"
2619        );
2620
2621        t.status = TaskStatus::Running;
2622        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
2623    }
2624
2625    #[test]
2626    fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
2627        let mut t = task("retried");
2628        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
2629        t.status = TaskStatus::Done;
2630        assert_eq!(
2631            t.superseded_attempts(true),
2632            &["aaaa".to_owned(), "bbbb".to_owned()],
2633            "cccc is the attempt whose success made the task done, and stays out"
2634        );
2635    }
2636
2637    #[test]
2638    fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
2639        let mut t = task("first try landed");
2640        t.runs = vec!["aaaa".to_owned()];
2641        t.status = TaskStatus::Done;
2642        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
2643    }
2644
2645    #[test]
2646    fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
2647        // `succeed` is reachable directly - `magi task done`, its web
2648        // equivalent, and the conductor's `Recovery::Done` - on a task in
2649        // any status, including one whose last recorded attempt is itself
2650        // `Blocked`/`Failed`/anything but `Merged`/`Ready`. `runs.last()`
2651        // alone cannot tell that apart from the loop's own settle path, so
2652        // the caller's own read of that run's status is what decides this.
2653        let mut t = task("closed by hand after a manual merge");
2654        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
2655        t.status = TaskStatus::Done;
2656        assert_eq!(
2657            t.superseded_attempts(false),
2658            &[] as &[String],
2659            "nothing here is provably why the task is done, so nothing is superseded"
2660        );
2661    }
2662
2663    #[test]
2664    fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
2665        let (_dir, q) = queue();
2666        let mut t = task("retried twice");
2667        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
2668        q.put(&mut t).unwrap();
2669
2670        assert_eq!(
2671            q.latest_attempt("aaaa"),
2672            Some("cccc".to_owned()),
2673            "an old attempt points straight at the chain's current head, not the \
2674             next attempt in the middle of it"
2675        );
2676        assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
2677        assert_eq!(
2678            q.latest_attempt("cccc"),
2679            None,
2680            "the latest attempt is not superseded by anything"
2681        );
2682        assert_eq!(
2683            q.latest_attempt("never-heard-of-it"),
2684            None,
2685            "a run belonging to no task on this queue is not superseded"
2686        );
2687    }
2688
2689    #[test]
2690    fn a_markdown_heading_is_the_title_not_decoration() {
2691        // A task file's heading is the summary its author already wrote, so it
2692        // beats the prose underneath. Getting this backwards was visible in the
2693        // first smoke test: a task titled "# Rework the config loader" listed
2694        // as "It re-reads the file on every lookup".
2695        assert_eq!(
2696            title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
2697            "Rework the config loader"
2698        );
2699        assert_eq!(title_from("- fix the thing", 40), "fix the thing");
2700        assert_eq!(title_from("> quoted task", 40), "quoted task");
2701        // Nothing usable at all still has to produce something printable.
2702        assert_eq!(title_from("   \n\n", 40), "(empty task)");
2703        assert_eq!(title_from("###\n", 40), "(empty task)");
2704    }
2705
2706    #[test]
2707    fn a_long_title_is_elided_by_characters_not_bytes() {
2708        // Byte truncation would split a multi-byte character and panic.
2709        let long = "課題".repeat(30);
2710        let title = title_from(&long, 10);
2711        assert_eq!(title.chars().count(), 10);
2712        assert!(title.ends_with('…'));
2713    }
2714
2715    #[test]
2716    fn priority_wins_and_ties_break_oldest_first() {
2717        let (_dir, q) = queue();
2718        let mut a = task("first");
2719        let mut b = task("second");
2720        let mut c = task("urgent");
2721        // Ids carry a timestamp, so force a known order.
2722        a.id = "20260101-000001-aaaa".to_owned();
2723        b.id = "20260101-000002-bbbb".to_owned();
2724        c.id = "20260101-000003-cccc".to_owned();
2725        c.priority = 5;
2726        for t in [&mut a, &mut b, &mut c] {
2727            q.put(t).unwrap();
2728        }
2729
2730        // Priority first...
2731        assert_eq!(q.next_runnable().unwrap().id, c.id);
2732        c.hold_machine(None);
2733        q.put(&mut c).unwrap();
2734        // ...then oldest, so a burst of new work cannot starve older work.
2735        assert_eq!(q.next_runnable().unwrap().id, a.id);
2736        assert_eq!(q.list().len(), 3, "b is still waiting its turn");
2737    }
2738
2739    #[test]
2740    fn a_blocked_task_never_starves_another_runnable_one() {
2741        let (_dir, q) = queue();
2742        let mut blocked = task("blocked");
2743        blocked.block(vec!["something".to_owned()], None);
2744        q.put(&mut blocked).unwrap();
2745
2746        let mut runnable = task("free to go");
2747        q.put(&mut runnable).unwrap();
2748
2749        let next = q.next_runnable().expect("a runnable task is still offered");
2750        assert_eq!(next.id, runnable.id);
2751    }
2752
2753    #[test]
2754    fn a_held_task_is_never_offered_to_the_loop() {
2755        let (_dir, q) = queue();
2756        let mut t = task("held");
2757        q.put(&mut t).unwrap();
2758        assert!(q.next_runnable().is_some());
2759
2760        t.hold_machine(None);
2761        q.put(&mut t).unwrap();
2762        assert!(
2763            q.next_runnable().is_none(),
2764            "a held task must wait for a human"
2765        );
2766
2767        // A failed task, by contrast, is exactly what the loop should retry.
2768        t.status = TaskStatus::Failed;
2769        q.put(&mut t).unwrap();
2770        assert!(q.next_runnable().is_some());
2771    }
2772
2773    #[test]
2774    fn attempts_are_capped_and_then_the_task_is_held() {
2775        let mut t = task("doomed");
2776
2777        t.start("run-1".to_owned());
2778        t.fail("gate red", 2);
2779        assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");
2780
2781        t.start("run-2".to_owned());
2782        t.fail("gate red", 2);
2783        assert_eq!(
2784            t.status,
2785            TaskStatus::Held,
2786            "out of attempts: stop spending money on it"
2787        );
2788        assert_eq!(t.runs, ["run-1", "run-2"]);
2789        assert_eq!(t.last_error.as_deref(), Some("gate red"));
2790        assert_eq!(
2791            t.hold_reason.as_deref(),
2792            Some("gate red"),
2793            "the hold must say why, not leave hold_reason null next to a \
2794             populated last_error"
2795        );
2796    }
2797
2798    #[test]
2799    fn handing_off_a_task_records_a_hold_reason_too() {
2800        let mut t = task("left a pull request");
2801        t.start("run-1".to_owned());
2802        t.handed_off("run ended with a pull request open [run run-1]");
2803        assert_eq!(t.status, TaskStatus::Held);
2804        assert_eq!(t.hold_source, Some(HoldSource::Machine));
2805        assert_eq!(
2806            t.hold_reason.as_deref(),
2807            Some("run ended with a pull request open [run run-1]")
2808        );
2809        assert_eq!(t.hold_reason, t.last_error);
2810    }
2811
2812    #[test]
2813    fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
2814        let mut t = task("stalled by quota");
2815
2816        t.start("run-1".to_owned());
2817        assert_eq!(t.attempts, 1);
2818        t.stall("judge-1, judge-2 out of quota");
2819        assert_eq!(
2820            t.attempts, 0,
2821            "a closed quota window must not spend the task's retry budget"
2822        );
2823        assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
2824        assert_eq!(
2825            t.last_error.as_deref(),
2826            Some("judge-1, judge-2 out of quota")
2827        );
2828
2829        // A task can therefore stall all night and still get its real attempts
2830        // once the quota resets - which is the whole point.
2831        for _ in 0..20 {
2832            t.start("run-n".to_owned());
2833            t.stall("still out of quota");
2834        }
2835        t.start("run-real".to_owned());
2836        t.fail("gate red", 2);
2837        assert_eq!(
2838            t.status,
2839            TaskStatus::Failed,
2840            "the first attempt that was really judged is attempt one"
2841        );
2842    }
2843
2844    #[test]
2845    fn releasing_a_held_task_gives_it_a_real_second_chance() {
2846        let mut t = task("retry me");
2847        t.start("run-1".to_owned());
2848        t.fail("gate red", 1);
2849        assert_eq!(t.status, TaskStatus::Held);
2850
2851        t.release();
2852        assert_eq!(t.status, TaskStatus::Queued);
2853        // Without resetting attempts the next failure would re-hold at once,
2854        // and a release would be a no-op the operator cannot see.
2855        assert_eq!(t.attempts, 0);
2856        assert!(t.last_error.is_none());
2857        assert_eq!(
2858            t.runs.len(),
2859            1,
2860            "history is kept: attempts reset, evidence does not"
2861        );
2862    }
2863
2864    #[test]
2865    fn a_hold_reason_survives_and_a_release_clears_it() {
2866        let mut t = task("waiting on something else");
2867        t.hold_manual(Some(
2868            "waiting for 20260101-000000-aaaa to land first".to_owned(),
2869        ));
2870        assert_eq!(t.status, TaskStatus::Held);
2871        assert_eq!(
2872            t.hold_reason.as_deref(),
2873            Some("waiting for 20260101-000000-aaaa to land first")
2874        );
2875
2876        // Holding again with no reason must not erase the one already there.
2877        t.hold_manual(None);
2878        assert_eq!(
2879            t.hold_reason.as_deref(),
2880            Some("waiting for 20260101-000000-aaaa to land first"),
2881            "a bare re-hold keeps whatever a human already wrote down"
2882        );
2883
2884        // A hold with no reason at all is still an ordinary, allowed hold.
2885        let mut plain = task("no reason given");
2886        plain.hold_manual(None);
2887        assert_eq!(plain.status, TaskStatus::Held);
2888        assert!(plain.hold_reason.is_none());
2889
2890        t.release();
2891        assert_eq!(t.status, TaskStatus::Queued);
2892        assert!(
2893            t.hold_reason.is_none(),
2894            "a stale reason must not greet the next person who holds this task"
2895        );
2896    }
2897
2898    #[test]
2899    fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
2900        // `done` can close a held task directly - neither `magi task done`
2901        // nor `POST /api/queue/{id}/done` requires a release first - so a
2902        // task held for "waiting on 3ed9" and then closed without ever being
2903        // released must not still read as waiting on it afterwards.
2904        let mut t = task("landed by hand while held");
2905        t.hold_manual(Some("waiting on 3ed9".to_owned()));
2906        assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));
2907
2908        t.succeed();
2909        assert_eq!(t.status, TaskStatus::Done);
2910        assert!(
2911            t.hold_reason.is_none(),
2912            "a done task cannot still be waiting on something"
2913        );
2914    }
2915
2916    #[test]
2917    fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
2918        // The web UI's "Hold" and "Mark done" buttons are both reachable on a
2919        // `blocked` task, not just on `queued`/`held` ones - neither requires
2920        // a release first. A task moved off `Blocked` that way must not still
2921        // carry the dependency it was waiting on: a dependency graph built
2922        // from `blocked_by` would otherwise keep drawing an edge for a task
2923        // that is not blocked on anything any more.
2924        let mut held = task("held straight out of blocked");
2925        held.block(
2926            vec!["20260101-000000-dead".to_owned()],
2927            Some("waiting on the migration script".to_owned()),
2928        );
2929        assert_eq!(held.status, TaskStatus::Blocked);
2930
2931        held.hold_manual(None);
2932        assert_eq!(held.status, TaskStatus::Held);
2933        assert!(
2934            held.blocked_by.is_empty(),
2935            "hold overrides the wait, same as release"
2936        );
2937        assert!(held.block_reason.is_none());
2938
2939        let mut done = task("closed straight out of blocked");
2940        done.block(
2941            vec!["20260101-000000-dead".to_owned()],
2942            Some("waiting on the migration script".to_owned()),
2943        );
2944        done.succeed();
2945        assert_eq!(done.status, TaskStatus::Done);
2946        assert!(
2947            done.blocked_by.is_empty(),
2948            "a done task cannot still be waiting on a dependency"
2949        );
2950        assert!(done.block_reason.is_none());
2951    }
2952
2953    #[test]
2954    fn a_blocked_task_is_never_offered_to_the_loop() {
2955        let mut t = task("blocked");
2956        assert!(t.status.runnable());
2957        t.block(
2958            vec!["dep-id".to_owned()],
2959            Some("waits on dep-id".to_owned()),
2960        );
2961        assert_eq!(t.status, TaskStatus::Blocked);
2962        assert!(!t.status.runnable());
2963        assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
2964    }
2965
2966    #[test]
2967    fn unblocking_the_last_dependency_returns_the_task_to_queued() {
2968        let mut t = task("blocked on two");
2969        t.block(
2970            vec!["a".to_owned(), "b".to_owned()],
2971            Some("waits on a and b".to_owned()),
2972        );
2973
2974        t.unblock("a");
2975        assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
2976        assert_eq!(t.blocked_by, ["b"]);
2977
2978        t.unblock("b");
2979        assert_eq!(t.status, TaskStatus::Queued);
2980        assert!(t.blocked_by.is_empty());
2981        assert!(t.block_reason.is_none());
2982    }
2983
2984    #[test]
2985    fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
2986        let mut t = task("never blocked");
2987        t.unblock("whatever");
2988        assert_eq!(t.status, TaskStatus::Queued);
2989    }
2990
2991    #[test]
2992    fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
2993        // The bug this guards: a task an operator (or `crate::triage`) has
2994        // deliberately held, once `crate::conduct` blocks it on a follow-up
2995        // question, must not silently re-enter the competition queue the
2996        // moment that question is answered - whatever the answer said.
2997        let mut t = task("held, then asked about");
2998        t.hold_machine(Some("out of attempts".to_owned()));
2999        assert_eq!(t.status, TaskStatus::Held);
3000
3001        t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
3002        assert_eq!(t.status, TaskStatus::Blocked);
3003
3004        t.record_answer("what now?".to_owned(), "leave it held".to_owned());
3005        t.unblock("q1");
3006        assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
3007        assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
3008        assert_eq!(t.hold_source, Some(HoldSource::Machine));
3009        assert!(t.blocked_from.is_none(), "consumed once restored");
3010    }
3011
3012    #[test]
3013    fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
3014        let mut t = task("manually held, then asked about");
3015        t.hold_manual(Some("waiting on a dependency".to_owned()));
3016
3017        t.block(vec!["q1".to_owned()], None);
3018        t.unblock("q1");
3019
3020        assert_eq!(t.status, TaskStatus::Held);
3021        assert_eq!(t.hold_source, Some(HoldSource::Manual));
3022    }
3023
3024    #[test]
3025    fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
3026        // A second `Task::block` call - `crate::conduct` adding a question on
3027        // top of an existing block - must not overwrite `blocked_from` with
3028        // `Blocked` itself, or the task would restore into itself.
3029        let mut t = task("held, blocked twice");
3030        t.hold_machine(None);
3031        t.block(vec!["q1".to_owned()], Some("first".to_owned()));
3032        t.block(
3033            vec!["q1".to_owned(), "q2".to_owned()],
3034            Some("second".to_owned()),
3035        );
3036
3037        t.unblock("q1");
3038        assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
3039        t.unblock("q2");
3040        assert_eq!(t.status, TaskStatus::Held);
3041    }
3042
3043    #[test]
3044    fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
3045        // Whatever process was driving the run is gone by the time a
3046        // conductor's question about it gets answered - there is nothing left
3047        // to resume into.
3048        let mut t = task("blocked mid-run");
3049        t.start("run-1".to_owned());
3050        assert_eq!(t.status, TaskStatus::Running);
3051
3052        t.block(vec!["q1".to_owned()], None);
3053        t.unblock("q1");
3054        assert_eq!(t.status, TaskStatus::Queued);
3055    }
3056
3057    #[test]
3058    fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
3059        // `blocked_from` is `None` for a record written before schema 4 (or,
3060        // equivalently, deserialized straight from an on-disk file that never
3061        // had the field). Held evidence surviving on the task - never cleared
3062        // by `block` - is the only way left to tell such a record apart from
3063        // one blocked straight out of `Queued`.
3064        let mut t = task("legacy record, held before it was blocked");
3065        t.hold_source = Some(HoldSource::Machine);
3066        t.hold_reason = Some("legacy hold reason".to_owned());
3067        t.status = TaskStatus::Blocked;
3068        t.blocked_by = vec!["q1".to_owned()];
3069        t.blocked_from = None;
3070
3071        t.unblock("q1");
3072        assert_eq!(t.status, TaskStatus::Held);
3073    }
3074
3075    #[test]
3076    fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
3077        let mut t = task("legacy record, ordinary dependency block");
3078        t.status = TaskStatus::Blocked;
3079        t.blocked_by = vec!["dep".to_owned()];
3080        t.blocked_from = None;
3081
3082        t.unblock("dep");
3083        assert_eq!(t.status, TaskStatus::Queued);
3084    }
3085
3086    #[test]
3087    fn answering_a_question_is_recorded_and_survives_a_release() {
3088        let mut t = task("asked something");
3089        t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
3090        t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
3091        t.unblock("q1");
3092        assert_eq!(t.status, TaskStatus::Queued);
3093        assert_eq!(t.answers.len(), 1);
3094        assert_eq!(t.answers[0].answer, "SQLite");
3095
3096        // A release resets attempts, not evidence - the same rule
3097        // `releasing_a_held_task_gives_it_a_real_second_chance` asserts for
3098        // `runs`.
3099        t.release();
3100        assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
3101    }
3102
3103    #[test]
3104    fn a_refused_handover_keeps_the_review_branch_across_release() {
3105        let mut t = task("refused takeover");
3106        t.start("run-1".to_owned());
3107        t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
3108        assert_eq!(t.status, TaskStatus::Held);
3109        assert_eq!(t.attempts, 1);
3110        t.release();
3111        assert_eq!(t.status, TaskStatus::Queued);
3112        assert_eq!(t.attempts, 0);
3113        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3114
3115        // A fresh competition chosen afterwards is not a review.
3116        t.hold_for_handover(None, "again".to_owned());
3117        t.requeue();
3118        assert!(t.review_branch.is_none());
3119
3120        // A manual hold never keeps one.
3121        let mut m = task("manual");
3122        m.review_branch = Some("magi/x/A".to_owned());
3123        m.hold_manual(None);
3124        m.release();
3125        assert!(m.review_branch.is_none());
3126    }
3127
3128    #[test]
3129    fn requesting_review_requeues_the_task_and_remembers_the_branch() {
3130        let mut t = task("blocked run with a surviving branch");
3131        t.start("run-1".to_owned());
3132        t.fail("blocked with major findings", 5);
3133        assert_eq!(t.status, TaskStatus::Failed);
3134
3135        t.request_review("magi/eba2/A".to_owned());
3136        assert_eq!(t.status, TaskStatus::Queued);
3137        assert_eq!(t.attempts, 0);
3138        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));
3139
3140        // An ordinary release (a human overriding the choice) drops it again.
3141        t.release();
3142        assert!(t.review_branch.is_none());
3143    }
3144
3145    #[test]
3146    fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
3147        let mut t = task("retry");
3148        t.start("run-1".to_owned());
3149        t.requeue();
3150        assert!(t.fresh_start);
3151
3152        t.release();
3153        assert!(!t.fresh_start);
3154    }
3155
3156    #[test]
3157    fn priority_can_be_changed_while_queued_but_not_while_running() {
3158        let mut t = task("reprioritise me");
3159        t.set_priority(5).unwrap();
3160        assert_eq!(t.priority, 5);
3161
3162        t.start("run-1".to_owned());
3163        let err = t.set_priority(9).unwrap_err().to_string();
3164        assert!(err.contains("running"), "{err}");
3165        assert_eq!(t.priority, 5, "the rejected write must not partially apply");
3166    }
3167
3168    #[test]
3169    fn interrupt_can_be_marked_while_queued_but_not_while_running() {
3170        let mut t = task("interrupt me");
3171        assert!(!t.interrupt, "off unless asked, same as any other task");
3172
3173        t.set_interrupt(true).unwrap();
3174        assert!(t.interrupt);
3175
3176        t.start("run-1".to_owned());
3177        assert!(
3178            !t.interrupt,
3179            "the mark is one-shot: dispatching the task fulfils it, \
3180             whatever the run that follows ends up doing"
3181        );
3182        let err = t.set_interrupt(true).unwrap_err().to_string();
3183        assert!(err.contains("running"), "{err}");
3184        // Clearing is always allowed, even on a running task - there is
3185        // nothing left for it to interrupt once it has been claimed.
3186        t.set_interrupt(false).unwrap();
3187        assert!(!t.interrupt);
3188    }
3189
3190    /// R2-1-1: a task whose run fails and requeues must not go on
3191    /// re-triggering `crate::daemon`'s interrupt scheduler on every later
3192    /// boundary, attempt after attempt, until it exhausts its budget.
3193    #[test]
3194    fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
3195        let mut t = task("interrupt me");
3196        t.set_interrupt(true).unwrap();
3197        t.start("run-1".to_owned());
3198        t.fail("mock failure", 5);
3199        assert_eq!(t.status, TaskStatus::Failed);
3200        assert!(
3201            !t.interrupt,
3202            "one attempt already spent the mark; a retry is an ordinary \
3203             requeue, not a fresh interrupt request"
3204        );
3205    }
3206
3207    #[test]
3208    fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
3209        let (_dir, q) = queue();
3210        let mut a = task("first filed");
3211        let mut b = task("second filed");
3212        a.id = "20260101-000001-aaaa".to_owned();
3213        b.id = "20260101-000002-bbbb".to_owned();
3214        q.put(&mut a).unwrap();
3215        q.put(&mut b).unwrap();
3216
3217        assert_eq!(
3218            q.next_runnable().unwrap().id,
3219            a.id,
3220            "with equal priority the older task goes first, so a burst of \
3221             new work cannot starve it"
3222        );
3223        assert_eq!(
3224            q.list()[0].id,
3225            b.id,
3226            "but the list an operator reads is newest first, the same as \
3227             before priority existed - a's turn to run does not make it the \
3228             newest task"
3229        );
3230
3231        let mut a = q.get(&a.id).unwrap();
3232        a.set_priority(10).unwrap();
3233        q.put(&mut a).unwrap();
3234
3235        assert_eq!(
3236            q.next_runnable().unwrap().id,
3237            a.id,
3238            "a raised priority must be reflected the moment it is saved"
3239        );
3240        // `magi task list` and `GET /api/queue` both print `Queue::list()`
3241        // directly, so the raised task has to lead there too - not only in
3242        // what the loop would claim next.
3243        assert_eq!(
3244            q.list()[0].id,
3245            a.id,
3246            "the raised task must sort first in the list an operator reads, \
3247             not only in next_runnable's own ordering"
3248        );
3249    }
3250
3251    #[test]
3252    fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
3253        let mut t = Task::new(
3254            "old title".to_owned(),
3255            "old instruction".to_owned(),
3256            PathBuf::from("/repo"),
3257            Source::Agent {
3258                run: "20260101-000000-beef".to_owned(),
3259                node: "implement".to_owned(),
3260            },
3261        );
3262        let id = t.id.clone();
3263        let created_at = t.created_at;
3264        t.runs.push("20260101-000000-beef".to_owned());
3265
3266        t.edit("new title".to_owned(), "new instruction".to_owned())
3267            .unwrap();
3268
3269        assert_eq!(t.title, "new title");
3270        assert_eq!(t.instruction, "new instruction");
3271        assert_eq!(t.id, id, "editing must not mint a new id");
3272        assert_eq!(t.created_at, created_at);
3273        assert_eq!(
3274            t.source,
3275            Source::Agent {
3276                run: "20260101-000000-beef".to_owned(),
3277                node: "implement".to_owned(),
3278            },
3279            "editing must not turn agent attribution into human"
3280        );
3281        assert_eq!(t.runs, ["20260101-000000-beef"]);
3282    }
3283
3284    #[test]
3285    fn editing_is_refused_once_a_task_is_running_or_finished() {
3286        let mut running = task("in flight");
3287        running.start("run-1".to_owned());
3288        let err = running
3289            .edit("x".to_owned(), "y".to_owned())
3290            .unwrap_err()
3291            .to_string();
3292        assert!(err.contains("running"), "{err}");
3293
3294        let mut done = task("finished");
3295        done.succeed();
3296        let err = done
3297            .edit("x".to_owned(), "y".to_owned())
3298            .unwrap_err()
3299            .to_string();
3300        assert!(err.contains("done"), "{err}");
3301
3302        // Both queued and held are the point of the feature and must work.
3303        let mut queued = task("waiting");
3304        queued.edit("x".to_owned(), "y".to_owned()).unwrap();
3305        let mut held = task("parked");
3306        held.hold_machine(None);
3307        held.edit("x".to_owned(), "y".to_owned()).unwrap();
3308    }
3309
3310    #[test]
3311    fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
3312        let (_dir, q) = queue();
3313        let path = q.path_of("20260101-000000-aaaa");
3314        std::fs::create_dir_all(q.root()).unwrap();
3315        std::fs::write(
3316            &path,
3317            serde_json::json!({
3318                "schema": SCHEMA,
3319                "id": "20260101-000000-aaaa",
3320                "title": "from before hold reasons existed",
3321                "instruction": "from before hold reasons existed",
3322                "repo": ".",
3323                "source": { "kind": "human" },
3324                "status": "held",
3325                "created_at": Timestamp::now().to_string(),
3326                "updated_at": Timestamp::now().to_string(),
3327            })
3328            .to_string(),
3329        )
3330        .unwrap();
3331
3332        let task = q.get("20260101-000000-aaaa").expect("must still read");
3333        assert!(task.hold_reason.is_none());
3334        assert!(task.operator_held());
3335    }
3336
3337    #[test]
3338    fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
3339        let (_dir, q) = queue();
3340        let path = q.path_of("20260101-000000-bbbb");
3341        std::fs::create_dir_all(q.root()).unwrap();
3342        std::fs::write(
3343            &path,
3344            serde_json::json!({
3345                "schema": 2,
3346                "id": "20260101-000000-bbbb",
3347                "title": "old manual recovery",
3348                "instruction": "old manual recovery",
3349                "repo": ".",
3350                "source": { "kind": "human" },
3351                "status": "held",
3352                "hold_reason": "active manual recovery run20260912-224242-daf5",
3353                "created_at": Timestamp::now().to_string(),
3354                "updated_at": Timestamp::now().to_string(),
3355            })
3356            .to_string(),
3357        )
3358        .unwrap();
3359
3360        let task = q.get("20260101-000000-bbbb").expect("must still read");
3361        assert_eq!(task.hold_source, None);
3362        assert!(task.operator_held());
3363    }
3364
3365    #[test]
3366    fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
3367        let (_dir, q) = queue();
3368        let path = q.path_of("20260101-000000-aaaa");
3369        std::fs::create_dir_all(q.root()).unwrap();
3370        std::fs::write(
3371            &path,
3372            serde_json::json!({
3373                "schema": SCHEMA,
3374                "id": "20260101-000000-aaaa",
3375                "title": "from before diagnostics existed",
3376                "instruction": "from before diagnostics existed",
3377                "repo": ".",
3378                "source": { "kind": "human" },
3379                "status": "held",
3380                "created_at": Timestamp::now().to_string(),
3381                "updated_at": Timestamp::now().to_string(),
3382            })
3383            .to_string(),
3384        )
3385        .unwrap();
3386
3387        let task = q.get("20260101-000000-aaaa").expect("must still read");
3388        assert!(task.diagnostic.is_none());
3389    }
3390
3391    #[test]
3392    fn a_schema_1_task_with_no_blocking_fields_still_reads() {
3393        // Written by a build that predates `blocked_by`, `block_reason`,
3394        // `answers` and `review_branch` entirely - literal `"schema": 1`,
3395        // not `SCHEMA`, since the whole point is a build older than this one.
3396        let (_dir, q) = queue();
3397        let path = q.path_of("20260101-000000-aaaa");
3398        std::fs::create_dir_all(q.root()).unwrap();
3399        std::fs::write(
3400            &path,
3401            serde_json::json!({
3402                "schema": 1,
3403                "id": "20260101-000000-aaaa",
3404                "title": "from before blocking existed",
3405                "instruction": "from before blocking existed",
3406                "repo": ".",
3407                "source": { "kind": "human" },
3408                "status": "queued",
3409                "created_at": Timestamp::now().to_string(),
3410                "updated_at": Timestamp::now().to_string(),
3411            })
3412            .to_string(),
3413        )
3414        .unwrap();
3415
3416        let task = q.get("20260101-000000-aaaa").expect("must still read");
3417        assert!(task.blocked_by.is_empty());
3418        assert!(task.block_reason.is_none());
3419        assert!(task.answers.is_empty());
3420        assert!(task.review_branch.is_none());
3421    }
3422
3423    #[test]
3424    fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
3425        // A diagnostic belongs to the run that produced it. Left in place
3426        // across a release, an unrelated later failure - a config error, say -
3427        // would go on showing evidence for a problem that is no longer why the
3428        // task is stuck.
3429        let mut held = task("diagnosed");
3430        held.start("run-1".to_owned());
3431        held.fail("gate red", 1);
3432        held.diagnostic = Some("cargo test failed: ...".to_owned());
3433        assert_eq!(held.status, TaskStatus::Held);
3434
3435        held.release();
3436        assert!(held.diagnostic.is_none());
3437
3438        held.diagnostic = Some("cargo test failed: ...".to_owned());
3439        held.succeed();
3440        assert!(held.diagnostic.is_none());
3441    }
3442
3443    #[test]
3444    fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
3445        let mut t = task("retried");
3446        t.start("run-1".to_owned());
3447        t.diagnostic = Some("stale evidence from a previous hold".to_owned());
3448        t.fail("unrelated config error", 5);
3449        assert_eq!(t.status, TaskStatus::Failed);
3450        assert!(
3451            t.diagnostic.is_none(),
3452            "fail() must not let an old diagnostic outlive the run that produced it"
3453        );
3454    }
3455
3456    #[test]
3457    fn a_claim_is_exclusive_and_releases_on_drop() {
3458        let (_dir, q) = queue();
3459        let mut t = task("contended");
3460        q.put(&mut t).unwrap();
3461
3462        let held = q.claim(&t.id).unwrap();
3463        assert!(
3464            q.claim(&t.id).is_err(),
3465            "two daemons must not drive one task into two runs"
3466        );
3467        drop(held);
3468        assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
3469    }
3470
3471    #[test]
3472    fn a_round_trip_survives_disk() {
3473        let (_dir, q) = queue();
3474        let mut t = Task::new(
3475            "titled".to_owned(),
3476            "body".to_owned(),
3477            PathBuf::from("/repo"),
3478            Source::Agent {
3479                run: "20260101-000000-beef".to_owned(),
3480                node: "implement".to_owned(),
3481            },
3482        );
3483        t.priority = 3;
3484        q.put(&mut t).unwrap();
3485
3486        let back = q.get(&t.id).unwrap();
3487        assert_eq!(back.id, t.id);
3488        assert_eq!(back.priority, 3);
3489        assert_eq!(back.source.label(), "implement@beef");
3490        // A prefix is enough, the way run ids work everywhere else.
3491        assert_eq!(q.get(t.short()).unwrap().id, t.id);
3492    }
3493
3494    #[test]
3495    fn an_unreadable_task_does_not_take_the_queue_down() {
3496        let (_dir, q) = queue();
3497        let mut t = task("fine");
3498        q.put(&mut t).unwrap();
3499        std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();
3500
3501        let listed = q.list();
3502        assert_eq!(listed.len(), 1, "the readable task still lists");
3503        assert_eq!(listed[0].id, t.id);
3504    }
3505
3506    #[test]
3507    fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
3508        let (_dir, q) = queue();
3509        let path = q.path_of("20260101-000000-aaaa");
3510        std::fs::create_dir_all(q.root()).unwrap();
3511        std::fs::write(
3512            &path,
3513            serde_json::json!({
3514                "schema": SCHEMA,
3515                "id": "20260101-000000-aaaa",
3516                "title": "from before solo existed",
3517                "instruction": "from before solo existed",
3518                "repo": ".",
3519                "source": { "kind": "human" },
3520                "status": "queued",
3521                "created_at": Timestamp::now().to_string(),
3522                "updated_at": Timestamp::now().to_string(),
3523            })
3524            .to_string(),
3525        )
3526        .unwrap();
3527
3528        let task = q.get("20260101-000000-aaaa").expect("must still read");
3529        assert!(!task.solo, "a queue file with no `solo` field means false");
3530    }
3531
3532    #[test]
3533    fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
3534        let (_dir, q) = queue();
3535        let path = q.path_of("20260101-000000-bbbb");
3536        std::fs::create_dir_all(q.root()).unwrap();
3537        std::fs::write(
3538            &path,
3539            serde_json::json!({
3540                "schema": SCHEMA,
3541                "id": "20260101-000000-bbbb",
3542                "title": "from before urgent existed",
3543                "instruction": "from before urgent existed",
3544                "repo": ".",
3545                "source": { "kind": "human" },
3546                "status": "queued",
3547                "created_at": Timestamp::now().to_string(),
3548                "updated_at": Timestamp::now().to_string(),
3549            })
3550            .to_string(),
3551        )
3552        .unwrap();
3553
3554        let task = q.get("20260101-000000-bbbb").expect("must still read");
3555        assert!(
3556            !task.urgent,
3557            "a queue file with no `urgent` field means false, same as `solo`"
3558        );
3559    }
3560
3561    #[test]
3562    fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
3563        let (_dir, q) = queue();
3564        let mut t = task("from the future");
3565        q.put(&mut t).unwrap();
3566        let path = q.path_of(&t.id);
3567        let body = std::fs::read_to_string(&path)
3568            .unwrap()
3569            .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
3570        std::fs::write(&path, body).unwrap();
3571
3572        let err = q.get(&t.id).unwrap_err().to_string();
3573        assert!(err.contains("schema 99"), "{err}");
3574    }
3575
3576    #[test]
3577    fn revision_moves_when_the_queue_changes() {
3578        let (_dir, q) = queue();
3579        assert_eq!(q.revision(), 0, "an empty queue has no revision");
3580        let mut t = task("first");
3581        q.put(&mut t).unwrap();
3582        assert!(q.revision() > 0, "a written task moves the revision");
3583    }
3584
3585    #[test]
3586    fn revision_moves_when_deleting_an_older_task() {
3587        let (dir, q) = queue();
3588        let questions = Questions::at(dir.path().join("questions"));
3589        let mut t1 = task("older");
3590        q.put(&mut t1).unwrap();
3591        // Ensure mtime ticks forward.
3592        std::thread::sleep(std::time::Duration::from_millis(10));
3593        let mut t2 = task("newer");
3594        q.put(&mut t2).unwrap();
3595
3596        let rev_before = q.revision();
3597        q.remove(&t1.id, false, &questions).unwrap();
3598        let rev_after = q.revision();
3599
3600        assert_ne!(
3601            rev_before, rev_after,
3602            "deleting an older task must change the revision so other clients see the deletion"
3603        );
3604    }
3605
3606    #[test]
3607    fn removing_a_task_takes_it_out_of_the_listing() {
3608        let (dir, q) = queue();
3609        let questions = Questions::at(dir.path().join("questions"));
3610        let mut t = task("delete me");
3611        q.put(&mut t).unwrap();
3612        let removed = q.remove(t.short(), false, &questions).unwrap();
3613        assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
3614        assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
3615        assert!(q.list().is_empty());
3616        assert!(
3617            q.remove(&t.id, false, &questions).is_err(),
3618            "removing twice is an error"
3619        );
3620    }
3621
3622    #[test]
3623    fn removing_a_task_takes_its_stale_lock_with_it() {
3624        let (dir, q) = queue();
3625        let questions = Questions::at(dir.path().join("questions"));
3626        let mut t = task("interrupted");
3627        q.put(&mut t).unwrap();
3628
3629        // A daemon killed mid-run leaves this behind. Nothing holds it: the
3630        // process that would have dropped the guard is gone.
3631        let claim = q.claim(&t.id).unwrap();
3632        std::mem::forget(claim);
3633        assert!(
3634            q.claim(&t.id).is_err(),
3635            "the orphaned lock is what makes the task look claimed"
3636        );
3637
3638        // A live daemon on this task is refused, whatever the lock says.
3639        let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
3640        assert!(err.contains("live daemon"), "{err}");
3641        assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");
3642
3643        // With no daemon behind it, the lock is stale and goes with the task.
3644        q.remove(&t.id, false, &questions).unwrap();
3645        assert!(q.list().is_empty());
3646        let mut again = task("interrupted");
3647        again.id = t.id.clone();
3648        q.put(&mut again).unwrap();
3649        assert!(
3650            q.claim(&t.id).is_ok(),
3651            "a task that comes back must be claimable, which a left-behind lock would prevent"
3652        );
3653    }
3654
3655    fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
3656        crate::notices::Notices::at(dir.join("notifications")).list()
3657    }
3658
3659    #[test]
3660    fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
3661        let (dir, q) = queue();
3662        let questions = Questions::at(dir.path().join("questions"));
3663        let mut dep = task("dependency");
3664        q.put(&mut dep).unwrap();
3665        let mut blocked = task("waiting");
3666        blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
3667        q.put(&mut blocked).unwrap();
3668
3669        let removed = q.remove(&dep.id, false, &questions).unwrap();
3670        assert_eq!(removed.released, [blocked.id.clone()]);
3671        assert!(removed.still_blocked.is_empty());
3672
3673        let after = q.get(&blocked.id).unwrap();
3674        assert_eq!(after.status, TaskStatus::Queued);
3675        assert!(after.blocked_by.is_empty());
3676        assert!(after.block_reason.is_none());
3677        let notes = notices_of(dir.path());
3678        assert_eq!(notes.len(), 1, "{notes:?}");
3679        assert_eq!(notes[0].severity, crate::notices::Severity::Info);
3680        assert_eq!(
3681            notes[0].link,
3682            Some(crate::notices::Link::Task {
3683                id: blocked.id.clone(),
3684            })
3685        );
3686    }
3687
3688    #[test]
3689    fn removing_one_of_two_dependencies_keeps_the_other() {
3690        let (dir, q) = queue();
3691        let questions = Questions::at(dir.path().join("questions"));
3692        let mut dep = task("dependency");
3693        q.put(&mut dep).unwrap();
3694        let mut other = task("other");
3695        q.put(&mut other).unwrap();
3696        let mut blocked = task("waiting");
3697        blocked.block(
3698            vec![dep.id.clone(), other.id.clone()],
3699            Some("waits on both".to_owned()),
3700        );
3701        q.put(&mut blocked).unwrap();
3702
3703        let removed = q.remove(&dep.id, false, &questions).unwrap();
3704        assert!(removed.released.is_empty());
3705        assert_eq!(removed.still_blocked, [blocked.id.clone()]);
3706
3707        let after = q.get(&blocked.id).unwrap();
3708        assert_eq!(after.status, TaskStatus::Blocked);
3709        assert_eq!(after.blocked_by, [other.id.clone()]);
3710        // The tombstone stays while a task still names the id... only this
3711        // one's dependency is gone, so nothing references it and it may go.
3712        assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
3713    }
3714
3715    #[test]
3716    fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
3717        let (dir, q) = queue();
3718        let questions = Questions::at(dir.path().join("questions"));
3719        let mut dep = task("dependency");
3720        q.put(&mut dep).unwrap();
3721        let mut blocked = task("waiting");
3722        blocked.block(vec![dep.id.clone()], None);
3723        q.put(&mut blocked).unwrap();
3724
3725        // The dependent is claimed elsewhere, so `remove` has to skip it.
3726        let claim = q.claim(&blocked.id).unwrap();
3727        let removed = q.remove(&dep.id, false, &questions).unwrap();
3728        assert!(removed.released.is_empty());
3729        drop(claim);
3730
3731        let mut task = q.get(&blocked.id).unwrap();
3732        assert_eq!(task.status, TaskStatus::Blocked);
3733        assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
3734        assert_eq!(q.apply_deleted_blockers(&mut task), [dep.id.clone()]);
3735        assert_eq!(task.status, TaskStatus::Queued);
3736    }
3737
3738    #[test]
3739    fn a_dependency_without_a_tombstone_is_still_missing() {
3740        let (dir, q) = queue();
3741        let questions = Questions::at(dir.path().join("questions"));
3742        let mut dep = task("dependency");
3743        q.put(&mut dep).unwrap();
3744        std::fs::remove_file(q.path_of(&dep.id)).unwrap();
3745        assert_eq!(
3746            missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
3747            [dep.id.clone()]
3748        );
3749        assert!(deleted_blockers(&q, std::slice::from_ref(&dep.id)).is_empty());
3750    }
3751
3752    #[test]
3753    fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
3754        let mut t = task("held");
3755        t.hold_machine(Some("because".to_owned()));
3756        t.blocked_by = vec!["gone".to_owned()];
3757        assert!(!t.dependency_deleted("gone"));
3758        assert_eq!(t.status, TaskStatus::Held);
3759    }
3760
3761    #[test]
3762    fn a_failed_record_removal_takes_the_tombstone_back() {
3763        let (_dir, q) = queue();
3764        let mut t = task("stays");
3765        q.put(&mut t).unwrap();
3766        q.write_tombstone(&t.id).unwrap();
3767        let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
3768        assert!(err.is_err());
3769        // `remove` is what withdraws the marker; a live record is never
3770        // treated as deleted whatever is on disk.
3771        assert!(!q.was_deleted(&t.id));
3772    }
3773
3774    fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
3775        let p = dir.join(name);
3776        std::fs::write(&p, body).unwrap();
3777        p
3778    }
3779
3780    #[test]
3781    fn an_attachment_copy_survives_deleting_its_source() {
3782        let (dir, q) = queue();
3783        let src = source_file(dir.path(), "shot.png", "pixels");
3784        let mut t = task("with a picture");
3785        let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
3786        q.put(&mut t).unwrap();
3787        std::fs::remove_file(&src).unwrap();
3788        assert_eq!(names, ["shot.png"]);
3789        let loaded = q.get(&t.id).unwrap();
3790        let paths = q.attachment_paths(&loaded);
3791        assert_eq!(paths.len(), 1);
3792        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
3793    }
3794
3795    #[test]
3796    fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
3797        let (dir, q) = queue();
3798        let mut t = task("bad names");
3799        for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
3800            let src = source_file(dir.path(), name, "x");
3801            assert!(
3802                q.attach(&mut t, &[src]).is_err(),
3803                "`{name}` must be refused"
3804            );
3805        }
3806        // A drive-qualified name cannot exist as a file on Windows (joining it
3807        // to a directory yields a drive-relative path to `foo.png` instead),
3808        // so the rule is asserted on the name itself, everywhere.
3809        assert!(!crate::ask::valid_asset_name("C:foo.png"));
3810        #[cfg(not(windows))]
3811        {
3812            let src = source_file(dir.path(), "C:foo.png", "x");
3813            assert!(q.attach(&mut t, &[src]).is_err());
3814        }
3815        let long = format!("{}.png", "a".repeat(70));
3816        let src = source_file(dir.path(), &long, "x");
3817        assert!(q.attach(&mut t, &[src]).is_err());
3818        assert!(t.attachments.is_empty());
3819        assert!(!q.attachments_dir(&t.id).exists());
3820    }
3821
3822    #[test]
3823    fn a_taken_attachment_name_is_numbered_not_overwritten() {
3824        let (dir, q) = queue();
3825        let a = source_file(dir.path(), "shot.png", "one");
3826        let sub = dir.path().join("other");
3827        std::fs::create_dir_all(&sub).unwrap();
3828        let b = source_file(&sub, "shot.png", "two");
3829        let mut t = task("collision");
3830        q.attach(&mut t, &[a]).unwrap();
3831        q.attach(&mut t, &[b]).unwrap();
3832        assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
3833        let paths = q.attachment_paths(&t);
3834        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
3835        assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
3836    }
3837
3838    #[test]
3839    fn a_renumbered_name_stays_inside_the_length_bound() {
3840        let (dir, q) = queue();
3841        let name = format!("{}.png", "a".repeat(60));
3842        assert_eq!(name.len(), 64);
3843        let a = source_file(dir.path(), &name, "one");
3844        let sub = dir.path().join("other");
3845        std::fs::create_dir_all(&sub).unwrap();
3846        let b = source_file(&sub, &name, "two");
3847        let mut t = task("long");
3848        q.attach(&mut t, &[a, b]).unwrap();
3849        assert_eq!(t.attachments.len(), 2);
3850        assert!(
3851            t.attachments
3852                .iter()
3853                .all(|n| crate::ask::valid_asset_name(n))
3854        );
3855        assert!(t.attachments[1].ends_with("-2.png"));
3856    }
3857
3858    #[test]
3859    fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
3860        let (dir, q) = queue();
3861        let good = source_file(dir.path(), "good.png", "ok");
3862        let mut t = task("partial");
3863        q.attach(&mut t, &[good]).unwrap();
3864        let more = source_file(dir.path(), "more.png", "ok");
3865        let missing = dir.path().join("missing.png");
3866        assert!(q.attach(&mut t, &[more, missing]).is_err());
3867        assert_eq!(t.attachments, ["good.png"]);
3868        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
3869            .unwrap()
3870            .flatten()
3871            .collect();
3872        assert_eq!(on_disk.len(), 1);
3873    }
3874
3875    /// Make the next `put` of `t` fail: its temp file's path is a directory.
3876    fn block_put(q: &Queue, t: &Task) -> PathBuf {
3877        let tmp = q.path_of(&t.id).with_extension("json.tmp");
3878        std::fs::create_dir_all(&tmp).unwrap();
3879        tmp
3880    }
3881
3882    #[test]
3883    fn a_failed_put_leaves_no_new_attachment_directory() {
3884        let (dir, q) = queue();
3885        let mut t = task("fresh");
3886        let tmp = block_put(&q, &t);
3887        let src = source_file(dir.path(), "shot.png", "x");
3888        assert!(q.attach_and_put(&mut t, &[src]).is_err());
3889        assert!(t.attachments.is_empty());
3890        assert!(!q.attachments_dir(&t.id).exists());
3891        assert!(!q.path_of(&t.id).exists());
3892        std::fs::remove_dir(tmp).unwrap();
3893    }
3894
3895    #[test]
3896    fn a_failed_put_removes_only_the_copy_it_just_made() {
3897        let (dir, q) = queue();
3898        let mut t = task("edited");
3899        let first = source_file(dir.path(), "first.png", "1");
3900        q.attach_and_put(&mut t, &[first]).unwrap();
3901        block_put(&q, &t);
3902        let second = source_file(dir.path(), "second.png", "2");
3903        assert!(q.attach_and_put(&mut t, &[second]).is_err());
3904        assert_eq!(t.attachments, ["first.png"]);
3905        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
3906            .unwrap()
3907            .flatten()
3908            .map(|e| e.file_name().to_string_lossy().into_owned())
3909            .collect();
3910        assert_eq!(on_disk, ["first.png"]);
3911        assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
3912    }
3913
3914    #[test]
3915    fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
3916        let (dir, q) = queue();
3917        let questions = Questions::at(dir.path().join("questions"));
3918        let gone = task("gone");
3919        let mut other = task("other");
3920        let mut live = task("live");
3921        q.put(&mut other).unwrap();
3922        q.put(&mut live).unwrap();
3923        // `gone` has no record: its removal finished deleting the record but
3924        // not the set-aside attachments.
3925        let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
3926        std::fs::create_dir_all(&orphan).unwrap();
3927        std::fs::write(orphan.join("shot.png"), "x").unwrap();
3928        // `live` still has a record, so its `.removing` belongs to a removal
3929        // in progress and must be left alone.
3930        let busy = q.root.join(format!("{}.attachments.removing", live.id));
3931        std::fs::create_dir_all(&busy).unwrap();
3932
3933        q.remove(&other.id, false, &questions).unwrap();
3934        assert!(!orphan.exists(), "an orphan is swept");
3935        assert!(busy.exists(), "a removal in progress is left alone");
3936    }
3937
3938    #[test]
3939    fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
3940        let (dir, q) = queue();
3941        let questions = Questions::at(dir.path().join("questions"));
3942        let mut t = task("stuck");
3943        let src = source_file(dir.path(), "shot.png", "x");
3944        q.attach_and_put(&mut t, &[src]).unwrap();
3945        // A non-empty directory already at the set-aside name makes the rename
3946        // fail; its record is present, so the sweep leaves it in place.
3947        let aside = q.root.join(format!("{}.attachments.removing", t.id));
3948        std::fs::create_dir_all(&aside).unwrap();
3949        std::fs::write(aside.join("old.png"), "o").unwrap();
3950        assert!(q.remove(&t.id, false, &questions).is_err());
3951        assert!(q.path_of(&t.id).exists());
3952        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
3953    }
3954
3955    #[test]
3956    fn a_failed_record_removal_puts_the_attachments_back() {
3957        let (dir, q) = queue();
3958        let mut t = task("rollback");
3959        let src = source_file(dir.path(), "shot.png", "x");
3960        q.attach_and_put(&mut t, &[src]).unwrap();
3961        let err = q
3962            .remove_record_with_attachments(&t.id, |_| {
3963                Err(std::io::Error::other("injected failure"))
3964            })
3965            .unwrap_err();
3966        assert!(format!("{err:#}").contains("injected failure"));
3967        assert!(q.path_of(&t.id).exists());
3968        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
3969        assert!(
3970            !q.root
3971                .join(format!("{}.attachments.removing", t.id))
3972                .exists()
3973        );
3974    }
3975
3976    #[test]
3977    fn editing_a_task_keeps_its_attachments() {
3978        let (dir, q) = queue();
3979        let src = source_file(dir.path(), "shot.png", "x");
3980        let mut t = task("editable");
3981        q.attach(&mut t, &[src]).unwrap();
3982        t.edit("new".to_owned(), "new text".to_owned()).unwrap();
3983        q.put(&mut t).unwrap();
3984        assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
3985    }
3986
3987    #[test]
3988    fn removing_a_task_deletes_its_attachments() {
3989        let (dir, q) = queue();
3990        let questions = Questions::at(dir.path().join("questions"));
3991        let src = source_file(dir.path(), "shot.png", "x");
3992        let mut t = task("doomed");
3993        q.attach(&mut t, &[src]).unwrap();
3994        q.put(&mut t).unwrap();
3995        assert!(q.attachments_dir(&t.id).is_dir());
3996        q.remove(&t.id, false, &questions).unwrap();
3997        assert!(!q.attachments_dir(&t.id).exists());
3998        assert!(q.list().is_empty());
3999    }
4000
4001    #[test]
4002    fn a_task_written_before_attachments_still_reads() {
4003        let (_dir, q) = queue();
4004        let mut t = task("old");
4005        q.put(&mut t).unwrap();
4006        let path = q.path_of(&t.id);
4007        let mut v: serde_json::Value =
4008            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
4009        v.as_object_mut().unwrap().remove("attachments");
4010        std::fs::write(&path, v.to_string()).unwrap();
4011        assert!(q.get(&t.id).unwrap().attachments.is_empty());
4012    }
4013
4014    #[test]
4015    fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
4016        let q = Queue::at(PathBuf::from("relative-queue"));
4017        let mut t = task("rel");
4018        t.attachments.push("shot.png".to_owned());
4019        let paths = q.attachment_paths(&t);
4020        assert!(paths[0].is_absolute(), "{}", paths[0].display());
4021        assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
4022    }
4023
4024    #[test]
4025    fn link_run_adds_a_run_once_and_touches_nothing_else() {
4026        let dir = tempfile::tempdir().unwrap();
4027        let queue = Queue::at(dir.path().join("queue"));
4028        let mut t = Task::new(
4029            "t".to_owned(),
4030            "do it".to_owned(),
4031            PathBuf::from("."),
4032            Source::Human,
4033        );
4034        queue.put(&mut t).unwrap();
4035        let before = queue.get(&t.id).unwrap();
4036
4037        let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
4038        assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
4039        assert_eq!(linked.status, before.status);
4040        assert_eq!(linked.attempts, before.attempts);
4041        assert_eq!(linked.interrupt, before.interrupt);
4042
4043        let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4044        assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
4045        assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
4046        assert!(queue.link_run("no-such-task", "r").is_err());
4047    }
4048
4049    #[test]
4050    fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
4051        let dir = tempfile::tempdir().unwrap();
4052        let queue = Queue::at(dir.path().join("queue"));
4053        let mut t = Task::new(
4054            "t".to_owned(),
4055            "do it".to_owned(),
4056            PathBuf::from("."),
4057            Source::Human,
4058        );
4059        queue.put(&mut t).unwrap();
4060        // The daemon's copy, taken before the link.
4061        let mut snapshot = queue.get(&t.id).unwrap();
4062        queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
4063
4064        snapshot.start("20260930-000000-aaaa".to_owned());
4065        queue.put(&mut snapshot).unwrap();
4066
4067        let stored = queue.get(&t.id).unwrap();
4068        assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
4069        assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
4070        assert_eq!(stored.attempts, 1);
4071    }
4072
4073    #[test]
4074    fn concurrent_links_and_daemon_saves_lose_nothing() {
4075        let dir = tempfile::tempdir().unwrap();
4076        let queue = Queue::at(dir.path().join("queue"));
4077        let mut t = Task::new(
4078            "t".to_owned(),
4079            "do it".to_owned(),
4080            PathBuf::from("."),
4081            Source::Human,
4082        );
4083        queue.put(&mut t).unwrap();
4084        let id = t.id.clone();
4085
4086        let linkers: Vec<_> = (0..4)
4087            .map(|n| {
4088                let (queue, id) = (queue.clone(), id.clone());
4089                std::thread::spawn(move || {
4090                    for k in 0..10 {
4091                        queue
4092                            .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
4093                            .unwrap();
4094                    }
4095                })
4096            })
4097            .collect();
4098        // The daemon's side: transitions saved from its own, ever staler,
4099        // copy of the task.
4100        let mut mine = queue.get(&id).unwrap();
4101        for k in 0..10 {
4102            mine.start(format!("20260930-000009-d{k:03}"));
4103            queue.put(&mut mine).unwrap();
4104        }
4105        for l in linkers {
4106            l.join().unwrap();
4107        }
4108
4109        let stored = queue.get(&id).unwrap();
4110        assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
4111        assert_eq!(
4112            stored.attempts, 10,
4113            "linking never rewinds the daemon's work"
4114        );
4115    }
4116
4117    fn age_lock(path: &Path) {
4118        let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
4119        f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
4120            .unwrap();
4121    }
4122
4123    #[test]
4124    fn concurrent_stale_takeover_yields_one_holder() {
4125        use std::sync::atomic::{AtomicUsize, Ordering};
4126        use std::sync::{Arc, Barrier};
4127        for _ in 0..5 {
4128            let dir = tempfile::tempdir().unwrap();
4129            let q = Queue::at(dir.path().to_path_buf());
4130            let lock = dir.path().join("t.write-lock");
4131            std::fs::write(&lock, "dead-0000").unwrap();
4132            age_lock(&lock);
4133            let n = 6;
4134            let barrier = Arc::new(Barrier::new(n));
4135            let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
4136            let handles: Vec<_> = (0..n)
4137                .map(|_| {
4138                    let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
4139                    std::thread::spawn(move || {
4140                        b.wait();
4141                        let g = q.lock_task("t").unwrap();
4142                        let held = now.fetch_add(1, Ordering::SeqCst) + 1;
4143                        max.fetch_max(held, Ordering::SeqCst);
4144                        std::thread::sleep(std::time::Duration::from_millis(20));
4145                        now.fetch_sub(1, Ordering::SeqCst);
4146                        drop(g);
4147                    })
4148                })
4149                .collect();
4150            for h in handles {
4151                h.join().unwrap();
4152            }
4153            assert_eq!(max.load(Ordering::SeqCst), 1);
4154            assert!(!lock.exists());
4155        }
4156    }
4157
4158    #[test]
4159    fn dropping_a_stolen_lock_leaves_the_new_holders_lock() {
4160        let dir = tempfile::tempdir().unwrap();
4161        let q = Queue::at(dir.path().to_path_buf());
4162        let lock = dir.path().join("t.write-lock");
4163        let a = q.lock_task("t").unwrap();
4164        age_lock(&lock);
4165        let b = q.lock_task("t").unwrap();
4166        assert_ne!(a.token, b.token);
4167        drop(a);
4168        assert_eq!(std::fs::read_to_string(&lock).unwrap(), b.token);
4169        drop(b);
4170        assert!(!lock.exists());
4171    }
4172
4173    #[test]
4174    fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
4175        let dir = tempfile::tempdir().unwrap();
4176        let lock = dir.path().join("t.write-lock");
4177        std::fs::write(&lock, "new-token").unwrap();
4178        assert!(!break_stale(&lock, "old-token"));
4179        assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
4180        // Same token but fresh: not stale any more, so also left alone.
4181        assert!(!break_stale(&lock, "new-token"));
4182        assert!(lock.exists());
4183        age_lock(&lock);
4184        assert!(break_stale(&lock, "new-token"));
4185        assert!(!lock.exists());
4186    }
4187
4188    #[test]
4189    fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
4190        let dir = tempfile::tempdir().unwrap();
4191        let lock = dir.path().join("t.write-lock");
4192        std::fs::write(&lock, "dead-1").unwrap();
4193        age_lock(&lock);
4194        let marker = break_marker(&lock, "dead-1");
4195        std::fs::write(&marker, "crashed-remover").unwrap();
4196        // Fresh marker: another remover is working, so nothing is broken.
4197        assert!(!break_stale(&lock, "dead-1"));
4198        assert!(lock.exists());
4199        // Stale marker: recovered, then the lock is broken.
4200        age_lock(&marker);
4201        assert!(break_stale(&lock, "dead-1"));
4202        assert!(!lock.exists());
4203        assert!(!marker.exists());
4204    }
4205
4206    fn chat_task() -> Task {
4207        Task::new(
4208            "t".to_owned(),
4209            "do t".to_owned(),
4210            PathBuf::from("."),
4211            Source::Agent {
4212                run: "20260902-000000-beef".to_owned(),
4213                node: CHAT_NODE.to_owned(),
4214            },
4215        )
4216    }
4217
4218    #[test]
4219    fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
4220        let mut t = chat_task();
4221        assert_eq!(t.chat_notice_due(), None, "queued");
4222        t.start("r1".to_owned());
4223        assert_eq!(t.chat_notice_due(), None, "running");
4224        t.fail("boom", 3);
4225        assert_eq!(t.status, TaskStatus::Failed);
4226        assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
4227        t.start("r2".to_owned());
4228        t.park("upgrade");
4229        assert_eq!(t.chat_notice_due(), None, "parked");
4230        t.fail("boom", 1);
4231        assert_eq!(t.status, TaskStatus::Held);
4232        let key = t
4233            .chat_notice_due()
4234            .expect("held on the last attempt is final");
4235        t.chat_notice = Some(ChatNotice {
4236            key,
4237            at: Timestamp::now(),
4238            outcome: ChatNoticeOutcome::Posted,
4239        });
4240        assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
4241        t.release();
4242        assert_eq!(t.chat_notice_due(), None, "requeued after a release");
4243        t.start("r3".to_owned());
4244        t.succeed();
4245        assert!(t.chat_notice_due().is_some(), "done is a new final state");
4246    }
4247
4248    #[test]
4249    fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
4250        for key in ["done", "done#0", "done#1", "done#12"] {
4251            let mut t = chat_task();
4252            t.start("r1".to_owned());
4253            t.succeed();
4254            t.chat_notice = Some(ChatNotice {
4255                key: key.to_owned(),
4256                at: Timestamp::now(),
4257                outcome: ChatNoticeOutcome::Posted,
4258            });
4259            t.requeue();
4260            t.start("r2".to_owned());
4261            t.succeed();
4262            assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
4263            assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
4264        }
4265    }
4266
4267    #[test]
4268    fn a_new_hold_after_a_release_is_a_new_cause() {
4269        let mut t = chat_task();
4270        t.hold_manual(Some("wait".to_owned()));
4271        let first = t.chat_notice_due().expect("held");
4272        t.chat_notice = Some(ChatNotice {
4273            key: first.clone(),
4274            at: Timestamp::now(),
4275            outcome: ChatNoticeOutcome::Posted,
4276        });
4277        t.hold_manual(Some("still waiting".to_owned()));
4278        assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
4279        t.release();
4280        std::thread::sleep(std::time::Duration::from_millis(5));
4281        t.hold_manual(None);
4282        let second = t.chat_notice_due().expect("a fresh hold");
4283        assert_ne!(first, second);
4284    }
4285
4286    #[test]
4287    fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
4288        let mut human = task("h");
4289        human.succeed();
4290        assert_eq!(human.chat_notice_due(), None);
4291        // A follow-up only inherits the chat; it did not file the task.
4292        let mut inherited = task("i");
4293        inherited.origin_chat = Some("20260902-000000-beef".to_owned());
4294        inherited.succeed();
4295        assert_eq!(inherited.chat_notice_due(), None);
4296        let mut old = chat_task();
4297        old.schema = 12;
4298        old.status = TaskStatus::Done;
4299        assert_eq!(old.chat_notice_due(), None, "history is not replayed");
4300        let mut live = chat_task();
4301        live.schema = 12;
4302        live.start("r1".to_owned());
4303        live.succeed();
4304        assert!(live.chat_notice_due().is_some(), "started by this build");
4305        let mut held = chat_task();
4306        held.schema = 12;
4307        held.hold_machine(Some("config".to_owned()));
4308        assert!(held.chat_notice_due().is_some(), "held before any start");
4309    }
4310
4311    #[test]
4312    fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
4313        let tmp = tempfile::tempdir().expect("tempdir");
4314        let q = Queue::at(tmp.path().join("queue"));
4315        let mut t = chat_task();
4316        t.succeed();
4317        q.put(&mut t).expect("put");
4318        let mut stale = q.get(&t.id).expect("get");
4319        let key = t.chat_notice_due().expect("due");
4320        q.modify(&t.id, |x| {
4321            x.chat_notice = Some(ChatNotice {
4322                key: key.clone(),
4323                at: Timestamp::now(),
4324                outcome: ChatNoticeOutcome::Posted,
4325            });
4326            true
4327        })
4328        .expect("modify");
4329        q.put(&mut stale).expect("stale put");
4330        assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
4331    }
4332}