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