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