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