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