Skip to main content

magi/
queue.rs

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