magi/

queue.rs

//! The task queue: what magi should do next, and who asked for it.
//!
//! The queue is what lets magi run unattended. `magi serve` takes the next
//! task, runs the graph on it, records the outcome, and takes the next one.
//!
//! It is also the reason an agent can ask for work. `magi task add` is the
//! whole interface, and it is the same command whether a human types it at a
//! prompt, a phone posts it through the web UI, or an implementer inside a run
//! shells out to it because it noticed something worth doing but out of scope.
//! magi's CLI is the operating surface for both kinds of user; the queue is
//! where their intentions meet.
//!
//! One task is one JSON file under [`Queue`]'s root. Files rather than a
//! database because the operator has to be able to read, edit, and delete the
//! backlog with the tools already on the machine, and because a crashed daemon
//! must leave a queue the next one can pick up without recovery ceremony.
//!
//! # Shape
//!
//! [`Task`] is data plus *pure* state transitions - [`Task::fail`] decides
//! whether an attempt was the last one, and touches no disk. [`Queue`] owns all
//! I/O and is constructed with its root, so a test drives a real queue in a
//! temp directory without setting a process-global home. Splitting them this
//! way is why the retry policy below can be asserted directly.
//!
//! # Bounded by construction
//!
//! An autonomous loop that retries forever is a way to spend money on a task
//! that cannot succeed. Every claim increments [`Task::attempts`]; a task that
//! has burned its attempts becomes [`TaskStatus::Held`] and waits for a human
//! rather than for another agent.

use std::collections::HashMap;
use std::num::NonZeroUsize;
use std::path::{Path, PathBuf};

use anyhow::{Context, Result, bail};
use jiff::Timestamp;
use serde::{Deserialize, Serialize};

use crate::ask::Questions;

/// On-disk format for a queued task. Bumped when a field's meaning changes.
///
/// 18: retained review criteria snapshots survive operator edits.
///
/// 17: durable explicit operator resumption, independent of run startup.
///
/// 7: added [`Task::attachments`], names of files copied under
/// `<id>.attachments/` beside the task file (a field only; `#[serde(default)]`,
/// so an older record reads as empty and [`read_path`] still accepts it).
///
/// 6: added [`Task::resume_override`] (a field only; `#[serde(default)]`, so
/// an older record reads as `None` and [`read_path`] still accepts it).
///
/// 5: added [`Task::triage_applied`], the ids of triage questions whose
/// answer has already been applied to this task. A "resume" answer used to
/// leave no trace ([`Task::release`] clears [`Task::hold_reason`], which was
/// the only place the applied marker lived), so when the released task failed
/// its attempts and went back to `held`, the next idle pass found the same
/// answered question "not applied" and released it again with `attempts` reset
/// to 0 - the `max_attempts` bound never held. `#[serde(default)]` so an older
/// record reads as empty. A task already looping when this build arrives has
/// no record, so it is released once more, recorded, and then stays held.
///
/// 8: added [`Task::followup`], the origin of a task `crate::followup` filed
/// from a merged run's leftover findings. Field-only bump, `#[serde(default)]`.
///
/// 4: added [`Task::blocked_from`], the status a task had the moment it
/// became [`TaskStatus::Blocked`], so [`Task::unblock`] restores it instead
/// of always landing on [`TaskStatus::Queued`]. Without it, a task a human
/// or `crate::triage` had deliberately left [`TaskStatus::Held`] — machine
/// or manual — would lose that the instant `crate::conduct` blocked it on a
/// follow-up question, and come back `Queued` the moment the question was
/// answered, regardless of what the answer said: exactly the loop where a
/// task the operator told to stay held instead re-enters the competition
/// queue every time someone answers a question about it. `#[serde(default)]`
/// so an older record reads as `None`; [`Task::unblock`] then falls back to
/// inferring `Held` from surviving hold evidence ([`Task::hold_reason`] /
/// [`Task::hold_source`], never cleared by [`Task::block`]) rather than
/// guessing `Queued` outright — see [`Task::unblock`]'s own doc.
///
/// 9: added [`Task::held_at`], the start of the current hold, so a notice can
/// tell a question filed for this hold from one about an older cause.
/// Field-only, `#[serde(default)]`.
///
/// 10: added [`Task::overrides`] and [`Task::review_of`], so a run started by
/// hand (`magi run` / `magi review`) is carried by a task without losing its
/// command-line choices. Field-only.
///
/// 11: added [`TaskStatus::Parked`] and [`Task::park_reason`], so a run that
/// stopped at a node boundary (an upgrade, an operator's park) is no longer
/// recorded as a failed task. Variant plus `#[serde(default)]` field; an older
/// build must not half-read the new status.
///
/// 12: added [`Task::origin_chat`], the talk a task descends from, so a
/// follow-up still knows its chat after an ancestor task is deleted.
/// Field-only, `#[serde(default)]`.
///
/// 13: added [`Task::chat_notice`], the record that a chat-filed task's final
/// state was reported to its conversation. Field-only, `#[serde(default)]`;
/// a record below this schema is never notified (an upgrade must not replay
/// the whole history into chats).
///
/// 3: added [`HoldSource`] so conductor recovery cannot release a hold an
/// operator deliberately placed. Old records default to `None` and are
/// protected as operator-held until an explicit release; the safe direction
/// when their author was never recorded.
///
/// 2: added [`TaskStatus::Blocked`], [`Task::blocked_by`] and
/// [`Task::block_reason`] (`crate::conduct`'s decisions) and
/// [`Task::answers`] (operator answers carried forward to the next
/// conductor prompt and the next run's instruction). All three are
/// `#[serde(default)]`, so [`read_path`] accepts anything up to and
/// including this schema rather than only an exact match — a task written
/// by a build that only knew about schema 1 has nothing to say about
/// blocking or answers, and defaulting those fields is exactly as good a
/// reading as a value that build never had a chance to write.
///
/// 15: added [`Task::passes`], what each settled pass over a run did to the
/// attempt budget and why it stopped. Field-only, `#[serde(default)]`; a task
/// from before this has no entry for its earlier passes, which stay unknown.
///
/// 14: record whether a task count edit's overrides inherit daemon options.
/// Old overrides default to isolated command-line settings, including empty ones.
///
/// 16: added [`Task::attempt_limit`], an optional task-specific retry budget.
/// Field-only, `#[serde(default)]`; old records inherit the daemon limit.
///
/// 19: persist settled pass start times for upgrade-resume scheduling.
/// Old passes default to unknown and use settlement order as an approximation.
pub const SCHEMA: u32 = 19;

/// Candidate order only: all claims, lanes and eligibility gates still apply.
/// Upgrade-parked runs retain their pass-start order ahead of ordinary work,
/// regardless of ordinary priority. Old records without a start time use
/// settlement time and task id as an approximation. Other parks are ordinary.
/// Urgent slots and explicit interrupts are handled separately by the daemon.
pub(crate) fn dispatch_order(a: &Task, b: &Task) -> std::cmp::Ordering {
    match (a.upgrade_resume_key(), b.upgrade_resume_key()) {
        (Some(a), Some(b)) => a.cmp(&b),
        (Some(_), None) => std::cmp::Ordering::Less,
        (None, Some(_)) => std::cmp::Ordering::Greater,
        (None, None) => b.priority.cmp(&a.priority).then(a.id.cmp(&b.id)),
    }
}

/// Who placed the current hold.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum HoldSource {
    /// An operator used the CLI or web UI.
    Manual,
    /// The daemon or conductor placed the hold as part of its own recovery.
    Machine,
}

impl HoldSource {
    /// Short human-facing label for reports and the CLI.
    pub fn label(self) -> &'static str {
        match self {
            Self::Manual => "manual",
            Self::Machine => "machine",
        }
    }
}

/// Where a task came from. Recorded because "who asked for this" is the first
/// question about an autonomous run, and the answer is not recoverable later.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
pub enum Source {
    /// A person, at a terminal or through the web UI.
    Human,
    /// An agent inside a run, via `magi task add`. Both ids are recorded so a
    /// task can be traced back to the exact seat that asked for it.
    Agent {
        /// Run the asking agent belonged to.
        run: String,
        /// Node it was working in, e.g. `implement` or `review`.
        node: String,
    },
    /// A GitHub issue, imported by number.
    Issue {
        /// Issue number.
        number: u64,
        /// `owner/repo`, as `gh` reports it.
        repo: String,
    },
}

/// The `node` a [`Source::Agent`] carries when the asker is a chat
/// conversation (its `run` is then the conversation's id).
pub const CHAT_NODE: &str = "chat";

impl Source {
    /// Short human-facing label, for lists and the web UI.
    pub fn label(&self) -> String {
        match self {
            Self::Human => "human".to_owned(),
            Self::Agent { run, node } => format!("{node}@{}", short(run)),
            Self::Issue { number, .. } => format!("issue #{number}"),
        }
    }
}

/// Where a task is in its life.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum TaskStatus {
    /// Waiting to be claimed.
    Queued,
    /// Claimed by a daemon; a run is in flight.
    Running,
    /// A run finished and its gate passed.
    Done,
    /// A run finished without passing, and attempts remain.
    Failed,
    /// Out of attempts, or held by hand. The loop will not pick it up.
    Held,
    /// Waiting on another task or an unanswered question. See
    /// [`Task::blocked_by`]. Set and cleared by `crate::conduct` and
    /// `crate::daemon`'s deterministic resolver, never by hand.
    Blocked,
    /// The run stopped at a node boundary because the process was asked to
    /// give itself back. Nothing failed: the run's work is intact and the next
    /// loop resumes it. Runnable like [`TaskStatus::Queued`]; the reason is in
    /// [`Task::park_reason`], never [`Task::last_error`].
    Parked,
}

impl TaskStatus {
    /// Is this task eligible for a daemon to claim?
    pub fn runnable(self) -> bool {
        matches!(self, Self::Queued | Self::Failed | Self::Parked)
    }

    /// Lowercase name, as it appears on disk and in the API.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Queued => "queued",
            Self::Running => "running",
            Self::Done => "done",
            Self::Failed => "failed",
            Self::Held => "held",
            Self::Blocked => "blocked",
            Self::Parked => "parked",
        }
    }
}

/// How many tasks sit in each [`TaskStatus`], for a dashboard tile — never a
/// per-task view, so it carries no ids.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TaskCounts {
    /// Waiting to be claimed.
    pub queued: usize,
    /// Claimed; a run is in flight.
    pub running: usize,
    /// A run finished and its gate passed.
    pub done: usize,
    /// A run finished without passing, and attempts remain.
    pub failed: usize,
    /// Out of attempts, or held by hand.
    pub held: usize,
    /// Waiting on another task or an unanswered question.
    pub blocked: usize,
    /// Run parked at a node boundary; resumable, not a failure.
    pub parked: usize,
}

impl TaskCounts {
    /// Tally `tasks` by status. A pure function over whatever [`Queue::list`]
    /// already read, so it needs no root of its own and stays trivially
    /// testable against a hand-built slice.
    pub fn of(tasks: &[Task]) -> Self {
        Self::of_statuses(tasks.iter().map(|task| task.status))
    }

    /// Count current statuses from a reconciled display projection.
    pub(crate) fn of_statuses(statuses: impl IntoIterator<Item = TaskStatus>) -> Self {
        let mut counts = Self::default();
        for status in statuses {
            match status {
                TaskStatus::Queued => counts.queued += 1,
                TaskStatus::Running => counts.running += 1,
                TaskStatus::Done => counts.done += 1,
                TaskStatus::Failed => counts.failed += 1,
                TaskStatus::Held => counts.held += 1,
                TaskStatus::Blocked => counts.blocked += 1,
                TaskStatus::Parked => counts.parked += 1,
            }
        }
        counts
    }
}

/// One unit of work.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Task {
    /// On-disk format version.
    pub schema: u32,
    /// Task id, e.g. `20260902-140501-a1b2`.
    pub id: String,
    /// One line, for lists and notifications.
    pub title: String,
    /// The task itself, handed to the graph verbatim.
    pub instruction: String,
    /// Explicit, SHA-pinned implementation imports; separate from instruction text.
    #[serde(default, skip_serializing_if = "crate::refs::Sources::is_empty")]
    pub sources: crate::refs::Sources,
    /// Repository to work in.
    pub repo: PathBuf,
    /// Who asked.
    pub source: Source,
    /// Higher runs first; ties break oldest-first so nothing starves.
    #[serde(default)]
    pub priority: i32,
    /// Run this task alone: one implementer, no panel of judges to convince.
    ///
    /// `#[serde(default)]` so a queue file written before this field existed
    /// still reads, as `false` - the ordinary multi-candidate competition,
    /// unchanged. A task set to `solo` still runs the whole graph; only the
    /// candidate count the daemon builds it with changes, and
    /// [`crate::graph::Runner`] already collapses a single-candidate run to
    /// implement → review → gate → merge on its own (see
    /// [`crate::graph::Runner::review`]'s doc), so nothing about judging,
    /// deliberation or voting had to change to support this.
    #[serde(default)]
    pub solo: bool,
    /// Current state.
    pub status: TaskStatus,
    /// How many times this task has been claimed.
    #[serde(default)]
    pub attempts: usize,
    /// Total attempt budget; `None` inherits the daemon limit.
    #[serde(default)]
    pub attempt_limit: Option<NonZeroUsize>,
    /// Distinct runs, ordered by their most recent claim. Older records may
    /// contain repeated ids; execution history lives in the run pass records.
    #[serde(default)]
    pub runs: Vec<String>,
    /// Why the last attempt did not land.
    #[serde(default)]
    pub last_error: Option<String>,
    /// What a human hold is waiting on.
    ///
    /// `None` covers both the ordinary cases: a hold the loop makes itself
    /// (out of attempts, or the disk gate closed) explains itself through
    /// [`Task::last_error`] instead, and a human hold nobody bothered to
    /// explain is still a valid hold. The queue has no way to express a
    /// dependency between two tasks, so on the occasions a hold really is
    /// "wait for that other task first", this is the only place that reason
    /// survives - see [`Task::hold_manual`] and [`Task::release`].
    ///
    /// `#[serde(default)]` so a queue file written before this field existed
    /// still reads, with no reason recorded rather than a parse error.
    #[serde(default)]
    pub hold_reason: Option<String>,
    /// Who placed [`Task::hold_reason`].  `None` is a compatible old record;
    /// see [`Task::operator_held`] for its deliberately conservative meaning.
    #[serde(default)]
    pub hold_source: Option<HoldSource>,
    /// Diagnostic detail excerpted from the run that led to a hold - what a
    /// human would have found opening `artifacts/` by hand, not the one-line
    /// reason in [`Task::last_error`]. Set only when a run's own attempts are
    /// exhausted and the task becomes [`TaskStatus::Held`]; `daemon` computes
    /// it from the run's own record, since this module has no notion of a
    /// run's internals. Bounded in length by the writer - see
    /// `daemon::diagnostic` - so a verbose run cannot make this file grow
    /// without limit.
    ///
    /// `#[serde(default)]` so a queue file written before this field existed
    /// still reads, with no diagnostic recorded rather than a parse error.
    #[serde(default)]
    pub diagnostic: Option<String>,
    /// What this task is waiting on: other task ids, unanswered
    /// `crate::ask::Question` ids, or both. Non-empty exactly when
    /// [`TaskStatus::Blocked`]; emptying it — see [`Task::unblock`] — is what
    /// puts the task back at [`TaskStatus::Queued`].
    ///
    /// Set by `crate::conduct`'s decisions and cleared deterministically by
    /// `crate::daemon` as each dependency resolves, never by a person. Never
    /// `#[serde(default)]` is skipped: a queue file from before this field
    /// existed has nothing to report here, and an empty list is exactly that.
    #[serde(default)]
    pub blocked_by: Vec<String>,
    /// One line explaining the current [`Task::blocked_by`], written by
    /// `crate::conduct`. Cleared whenever `blocked_by` empties.
    #[serde(default)]
    pub block_reason: Option<String>,
    /// The status this task had the moment [`Task::block`] most recently
    /// moved it to [`TaskStatus::Blocked`] — what [`Task::unblock`] restores
    /// once nothing is left in `blocked_by`, instead of always landing on
    /// [`TaskStatus::Queued`]. See [`SCHEMA`]'s doc for schema 4 on why this
    /// exists: an answer to a question `crate::conduct` filed about a
    /// [`TaskStatus::Held`] task must not itself be what puts the task back
    /// in the competition queue.
    ///
    /// `#[serde(default)]` so a queue file written before this field existed
    /// reads as `None`; [`Task::unblock`] treats that the same as a task
    /// blocked straight from `Queued`, unless surviving hold evidence says
    /// otherwise.
    #[serde(default)]
    pub blocked_from: Option<TaskStatus>,
    /// Last change of blocking targets, independent of explanation or delivery writes.
    #[serde(default)]
    pub blocked_at: Option<Timestamp>,
    /// Questions `crate::conduct` asked about this task that the operator has
    /// since answered, oldest first — what was asked, and what they said.
    ///
    /// A blocking question's id leaves [`Task::blocked_by`] the moment
    /// [`crate::ask::QuestionStatus::Answered`] is observed, but the id alone
    /// tells nobody what was decided. This is what carries the answer's
    /// *content* forward: into the next conductor prompt for this task, and
    /// into the instruction handed to the next run — see `crate::daemon`'s
    /// deterministic blocker resolution. Kept for the task's whole life, the
    /// same as [`Task::runs`]: a release resets attempts, not evidence.
    #[serde(default)]
    pub answers: Vec<AnsweredQuestion>,
    /// Ids of the `crate::triage` questions whose answer has been applied to
    /// this task. Unlike [`Task::hold_reason`], [`Task::release`] and every
    /// hold transition leave it alone, so an answer is applied at most once
    /// however many times the task is held again. See [`SCHEMA`]'s doc for
    /// schema 5. `#[serde(default)]` so an older record reads as empty.
    #[serde(default)]
    pub triage_applied: Vec<String>,
    /// Ids of the `magi ask` questions whose [`crate::ask::ChoiceAction`] has
    /// been applied to this task. Separate from [`Task::triage_applied`]
    /// (a different producer) and, like it, untouched by [`Task::release`],
    /// so one answer acts at most once however often the task is held again.
    /// `#[serde(default)]` so an older record reads as empty.
    #[serde(default)]
    pub actions_applied: Vec<String>,
    /// The operator's "resume" answer to a triage question, kept until the
    /// task actually runs (or is done) so `crate::conduct` cannot silently
    /// undo it and `crate::triage` can tell that a hold it sees now came
    /// *after* the answer. See [`OperatorResume`]. `#[serde(default)]`.
    #[serde(default)]
    pub resume_override: Option<OperatorResume>,
    /// Latest explicit operator release. Historical answers remain audit evidence;
    /// this approval takes precedence over their earlier waiting premises.
    #[serde(default)]
    pub resumption: Option<Resumption>,
    /// Earlier explicit approvals retained for audit and instruction suffix matching.
    #[serde(default)]
    pub resumption_history: Vec<Resumption>,
    /// Latest explicit release or manual hold, used to reject delayed answers.
    #[serde(default)]
    pub operator_decided_at: Option<Timestamp>,
    /// Set by `crate::conduct` when it chooses `Review` recovery for a task
    /// whose branch survived a blocked run: the branch to reopen with
    /// `crate::graph::Runner::review` instead of competing from scratch.
    ///
    /// Requeues the task the same way [`Task::release`] does, so it is
    /// picked up by the ordinary loop; `crate::daemon` reads this field once,
    /// when it actually starts the run, and clears it either way — consumed
    /// on success, dropped if the branch no longer exists by then. Never set
    /// from the conductor's own words: `crate::daemon` derives the branch
    /// name itself from the task's last run, so a hallucinated branch can
    /// never reach here.
    #[serde(default)]
    pub review_branch: Option<String>,
    /// Authored criteria when the retained recovery review was selected.
    #[serde(default)]
    pub review_instruction: Option<String>,
    /// Explicit public implementation selected by the operator; never inferred from prose.
    #[serde(default)]
    pub pr_recovery: Option<crate::pr_recovery::Selection>,
    /// A release deliberately starts a new competition instead of resuming
    /// the prior run. History remains as evidence in `runs`.
    #[serde(default)]
    pub fresh_start: bool,
    /// Explicit replacement selection; persists through dispatch and resume.
    #[serde(default)]
    pub replacement: Option<crate::merge_decision::Replacement>,
    /// Marked by an operator (`magi task interrupt`) to ask `magi serve` to
    /// run this one ahead of whatever it already has in flight, once
    /// `[daemon] pause_for_interrupts` is on - see
    /// `crate::daemon::advance_interrupt`. Never set by the loop itself, and
    /// deliberately a different operation from [`Task::set_priority`]: a
    /// priority only reorders the queue a claim has not reached yet, while
    /// this asks a run already in flight to park at its next safe boundary
    /// and step aside. `#[serde(default)]` so a queue file written before
    /// this field existed still reads, as `false` - no task interrupts
    /// anything unless asked to, exactly as before.
    #[serde(default)]
    pub interrupt: bool,
    /// Marked by `magi task add --urgent` or `magi task urgent`.
    /// `crate::daemon::poll` dispatches this task through its own one-slot `urgent_sem` the moment it is
    /// runnable, in addition to whatever is already running under the
    /// ordinary `[daemon] max_concurrent_runs` pool - never instead of it,
    /// and never by pausing or otherwise touching that run. This is the
    /// opposite direction from [`Task::interrupt`]: that one asks a run
    /// already in flight to step aside; this one never asks anything to
    /// step aside, it only spends one additional, temporary concurrency
    /// slot. The two are independent and may both be set on the same task,
    /// but this exemption stops at `[daemon] pause_for_interrupts`'s own
    /// park/resume handoff (75dd): while an interrupt sequence is actively
    /// parking, running, or resuming - its own, or an unrelated task's -
    /// `crate::daemon::interrupt_gate` withholds an urgent candidate exactly
    /// like an ordinary one, never exempted. 75dd's "at most one run, ever,
    /// at once" guarantee takes precedence, because the alternative is a run
    /// still genuinely in flight (only *asked* to park, not yet gone) ending
    /// up alongside a second one this feature let through - the very thing
    /// that guarantee exists to rule out.
    ///
    /// `#[serde(default)]` so a queue file written before this field existed
    /// still reads, as `false` - no task claims the urgent slot unless asked
    /// to, exactly as before.
    #[serde(default)]
    pub urgent: bool,
    /// Files (typically screenshots) copied into `<id>.attachments/` beside
    /// this task's file by [`Queue::attach`], as bare validated names. The
    /// copy is what makes them reach the implementer: the original may be
    /// cleaned up long before the task runs. Untouched by [`Task::edit`].
    /// `#[serde(default)]` so an older record reads as empty.
    #[serde(default)]
    pub attachments: Vec<String>,
    /// Set on a task `crate::followup` filed from the findings a merged run
    /// left open: which run and findings it carries, and how many follow-ups
    /// deep it is. `None` for every ordinary task. `#[serde(default)]`.
    #[serde(default)]
    pub followup: Option<FollowUp>,
    /// Id of the chat this task descends from: set by [`Task::new`] when a
    /// chat files it, inherited by the follow-ups `crate::followup` files, so
    /// the provenance survives deleting an ancestor. `None` for a task written
    /// before the field existed (`crate::consult` then walks the ancestry).
    /// `#[serde(default)]`.
    #[serde(default)]
    pub origin_chat: Option<String>,
    /// The command-line choices (`--merge`, `-c`, `--seed`, ...) of the
    /// `magi run` / `magi review` that filed this task, applied on top of the
    /// repository's config by whoever executes it - a daemon included - so
    /// handing the task over does not change what it does. A count edit on an
    /// ordinary queue task explicitly inherits daemon options instead.
    /// `#[serde(default)]`.
    #[serde(default)]
    pub overrides: Option<RunOverrides>,
    /// The branch an explicit `magi review <branch>` asked to review. Unlike
    /// [`Task::review_branch`] (a recovery choice the daemon consumes and
    /// drops back to a competition when the branch is gone), this is the whole
    /// request: it is never consumed, and a missing branch fails the attempt
    /// rather than paying for an implementation nobody asked for.
    /// `#[serde(default)]`.
    #[serde(default)]
    pub review_of: Option<String>,
    /// When the current hold began. Set by the transitions into
    /// [`TaskStatus::Held`] (a re-hold keeps it), cleared by
    /// [`Task::release`]. `crate::notices` uses it as the identity of the
    /// cause: a question filed before it is about something older.
    #[serde(default)]
    pub held_at: Option<Timestamp>,
    /// Why the run parked, while the task is [`TaskStatus::Parked`]. Kept apart
    /// from [`Task::last_error`] because nothing failed. Cleared by every
    /// transition out of the parked state. `#[serde(default)]`.
    #[serde(default)]
    pub park_reason: Option<String>,
    /// The last final state reported to the originating chat. See
    /// [`Task::chat_notice_due`]. Never cleared by a release or requeue.
    #[serde(default)]
    pub chat_notice: Option<ChatNotice>,
    /// Delivery ledger, unioned under the task lock even on stale writes.
    #[serde(default)]
    pub chat_notices: Vec<String>,
    /// Structured producer identity of the unresolved hold. Requeues and prose
    /// updates retain it; a settled run or a distinct startup stage replaces it.
    #[serde(default)]
    pub hold_notice_cause: Option<NoticeCause>,
    /// Append-only: one entry per pass over a run that the loop settled. A
    /// resumed run is one run with several passes, and `attempts` counts
    /// claims net of refunds, so the two are not the same number; this says
    /// which pass cost what. Kept across a release.
    #[serde(default)]
    pub passes: Vec<TaskPass>,
    /// When the loop claimed this task for its latest dispatch. Stamped under
    /// the claim, before any slow start-up work, so it is the real dispatch
    /// order; `record_pass_at` prefers it over the graph's own pass start.
    #[serde(default)]
    pub dispatched_at: Option<Timestamp>,
    /// Durable failed-input budget, unaffected by automatic release/requeue.
    #[serde(default)]
    pub recovery_budget: Option<RecoveryBudget>,
    /// Inputs resolved before starting the current automatic attempt. The
    /// settlement uses this same identity even for review-only runs.
    #[serde(default)]
    pub recovery_admission: Option<RecoveryAdmission>,
    /// Previously seen input generations. Switching inputs away and back does
    /// not refund an exhausted generation without explicit operator authority.
    #[serde(default)]
    pub recovery_generations: Vec<RecoveryBudget>,
    /// When the task was filed.
    pub created_at: Timestamp,
    /// Last change to this file.
    pub updated_at: Timestamp,
}

/// Admission identity shared by start and settlement.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecoveryAdmission {
    /// Fetched upstream commit before any run-local rebase.
    pub base: String,
    /// Resolved instruction references and explicit review source.
    pub inputs: String,
}

/// Failures against one recovery input generation. An owner release explicitly
/// starts a new generation; a machine requeue never replenishes it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecoveryBudget {
    /// Authored instructions, not generated review scaffolding.
    pub instruction: String,
    /// Upstream commit used to prepare this generation.
    pub base: String,
    /// Referenced input commits, including their resolved identities.
    pub inputs: String,
    /// Stable processing stage of the last failure.
    pub stage: String,
    /// Failed passes, independent of the resettable claim counter.
    pub failures: usize,
    /// Fingerprint of actual candidate patches; equivalent rebased commits do
    /// not count as progress. Unknown inputs never replenish the budget.
    #[serde(default)]
    pub progress: Option<String>,
}

fn recovery_progress(state: &crate::run::RunState) -> Option<String> {
    use crate::proc::Quiet as _;
    use std::hash::{Hash, Hasher};
    let mut patches = Vec::new();
    for candidate in state.candidates.iter().filter(|c| c.commits > 0) {
        let output = std::process::Command::new("git")
            .current_dir(&state.repo)
            .args(["diff", "--binary", &state.base_commit, &candidate.branch])
            .quiet()
            .output()
            .ok()?;
        if !output.status.success() {
            return None;
        }
        patches.push(output.stdout);
    }
    patches.sort();
    patches.dedup();
    let mut hash = std::collections::hash_map::DefaultHasher::new();
    patches.hash(&mut hash);
    Some(format!("{:x}", hash.finish()))
}

/// The implementation inputs a recovery budget is keyed on: the explicit,
/// SHA-pinned sources only, never references parsed from instruction prose.
/// No sources is `[]`, the key an input-free task always had.
pub(crate) fn recovery_inputs(sources: &crate::refs::Sources) -> String {
    serde_json::to_string(
        &sources
            .start_from
            .iter()
            .map(|i| ("start", &i.token, &i.sha))
            .chain(
                sources
                    .pick_commits
                    .iter()
                    .map(|i| ("pick", &i.token, &i.sha)),
            )
            .collect::<Vec<_>>(),
    )
    .unwrap_or_default()
}

/// Facts a producer can independently attest about a held outcome.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct NoticeCause {
    /// Processing stage or verified external evidence.
    pub stage: String,
    /// Producer-verified blocker object, independent of the displayed error.
    /// Omitted when absent to preserve existing persisted notice keys.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub target: Option<String>,
    /// An actual run that reached settlement; absent for preflight failures.
    pub run: Option<String>,
    /// The settled pass, so a resumed run can report a new failure.
    pub pass: Option<usize>,
}

/// Current and legacy completion keys; legacy run counts contain only digits.
pub(crate) fn is_done_notice_key(key: &str) -> bool {
    key == "done"
        || key
            .strip_prefix("done#")
            .is_some_and(|count| !count.is_empty() && count.bytes().all(|b| b.is_ascii_digit()))
}

/// What a settled pass did to the attempt budget.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PassCost {
    /// The attempt stays counted.
    Spent,
    /// The attempt was handed back.
    Refunded,
}

/// One settled pass over a run. See [`Task::passes`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TaskPass {
    /// The run the pass belongs to.
    pub run: String,
    /// [`crate::run::RunPass::index`] in that run.
    pub pass_index: usize,
    /// What it did to the budget.
    pub cost: PassCost,
    /// `attempts` once this pass was settled.
    pub attempts_after: usize,
    /// When this pass began, retained across daemon upgrades.
    #[serde(default)]
    pub started_at: Option<Timestamp>,
    /// When the loop settled it.
    pub settled_at: Timestamp,
    /// Why the pass parked, when it did.
    #[serde(default)]
    pub parked: Option<crate::run::ParkCause>,
}

/// What became of a completion notice. See [`Task::chat_notice`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ChatNoticeOutcome {
    /// Queued as a draft of the originating talk.
    Posted,
    /// The talk is gone or closed; nothing was queued, and nothing will be.
    Skipped,
}

/// The final state of a chat-filed task that was already reported (or skipped).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChatNotice {
    /// Identity of the cause, see [`Task::chat_notice_key`].
    pub key: String,
    /// When it was recorded.
    pub at: Timestamp,
    /// Posted or skipped.
    pub outcome: ChatNoticeOutcome,
}

/// Command-line choices carried by a task. See [`Task::overrides`].
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunOverrides {
    /// Count edits on ordinary queued tasks inherit the loop's config and merge.
    /// False for direct-run overrides and old records, even when otherwise empty.
    #[serde(default)]
    pub inherit_daemon_options: bool,
    /// `--merge`: `none`, `local` or `pr`.
    #[serde(default)]
    pub merge: Option<String>,
    /// `-c` / `--implementers`: parallel implementations (persisted as `candidates`).
    #[serde(default)]
    pub candidates: Option<usize>,
    /// `-j`: independent judges.
    #[serde(default)]
    pub judges: Option<usize>,
    /// `--reviewers`: reviewers per round.
    #[serde(default)]
    pub reviewers: Option<usize>,
    /// `--review-rounds`.
    #[serde(default)]
    pub review_rounds: Option<usize>,
    /// `--seed`.
    #[serde(default)]
    pub seed: Option<u64>,
    /// `--config`, absolute: the daemon's own working directory differs.
    #[serde(default)]
    pub config: Option<PathBuf>,
}

impl RunOverrides {
    /// Lay `other` over these: every choice `other` made wins, the rest stay.
    pub fn merge_over(&mut self, other: &RunOverrides) {
        macro_rules! take {
            ($($f:ident),*) => {$(
                if other.$f.is_some() {
                    self.$f.clone_from(&other.$f);
                }
            )*};
        }
        take!(
            merge,
            candidates,
            judges,
            reviewers,
            review_rounds,
            seed,
            config
        );
    }

    /// Apply these on top of `config`.
    pub fn apply(&self, config: &mut crate::config::Config) {
        if let Some(n) = self.candidates {
            config.graph.implementers = n;
        }
        if let Some(n) = self.judges {
            config.graph.judges = n;
        }
        if let Some(n) = self.reviewers {
            config.graph.reviewers = n;
        }
        if let Some(n) = self.review_rounds {
            config.graph.review_rounds = n;
        }
        if let Some(m) = &self.merge
            && let Ok(mode) = crate::daemon::merge_mode(m)
        {
            config.merge.mode = mode;
        }
        if let Some(s) = self.seed {
            config.blind.seed = Some(s);
        }
    }
}

/// What a follow-up task came from. See [`Task::followup`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FollowUp {
    /// The merged run whose open findings this task carries.
    pub run: String,
    /// The queue task that run served, when it had one.
    #[serde(default)]
    pub origin_task: Option<String>,
    /// The merged pull request.
    pub pr: String,
    /// Ids of the findings (e.g. `R3-1-1`) this task covers.
    pub findings: Vec<String>,
    /// Follow-up depth: 1 for a follow-up of an ordinary task's run, 2 for a
    /// follow-up of that, and so on. An ordinary task is generation 0.
    pub generation: u32,
}

/// Durable scope of an explicit operator resumption, surviving run startup.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Resumption {
    /// Monotonic count of explicit resumptions.
    pub generation: u64,
    /// Mutation path that applied the approval.
    pub source: String,
    /// Answered question, if the approval came from one.
    pub question_id: Option<String>,
    /// When this approval was applied.
    pub at: Timestamp,
    /// Authored instruction approved at release.
    pub instruction: String,
    /// Number of historical answers superseded as waiting premises.
    pub answers_through: usize,
    /// Legacy run-list boundary retained for record compatibility. Evidence
    /// uses approval time and run passes because resumes retain the same id.
    pub runs_through: usize,
    /// Previous hold or dependency reason cleared by this release.
    pub cleared_reason: Option<String>,
    /// Verified evidence versions already used to block after this approval.
    #[serde(default)]
    pub blocking_evidence: Vec<String>,
}

/// A triage "resume" answer and what became of it. See
/// [`Task::resume_override`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OperatorResume {
    /// 回答された Question の ID。Question を伴わない再開では空文字列。
    pub question_id: String,
    /// When the answer was applied.
    pub at: Timestamp,
    /// The reason `crate::conduct` gave for holding the task again after the
    /// answer, if it did. The conductor may do this once.
    #[serde(default)]
    pub conductor_rehold: Option<String>,
    /// The operator answered "resume" a second time, to the question about
    /// that contradiction: the conductor may no longer hold this task.
    #[serde(default)]
    pub forced: bool,
    /// Set when the answer was a structured `resume` action on a `magi ask`
    /// choice ([`crate::ask::ChoiceAction::Resume`]): the run the operator
    /// chose to continue. Its presence also stops `crate::conduct` from
    /// requeuing the task into a fresh competition, which would discard
    /// exactly the run that was named. Cleared with the rest of the record
    /// once the task runs. `#[serde(default)]` so older records read as `None`.
    #[serde(default)]
    pub pinned_run: Option<String>,
}

/// One question `crate::conduct` asked about a task, and what the operator
/// said back. See [`Task::answers`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AnsweredQuestion {
    /// The question as asked, e.g. [`crate::ask::Question::summary`].
    pub question: String,
    /// What the operator answered.
    pub answer: String,
}

impl Task {
    /// File a new task. Persist it with [`Queue::put`].
    pub fn new(title: String, instruction: String, repo: PathBuf, source: Source) -> Self {
        let now = Timestamp::now();
        let origin_chat = match &source {
            Source::Agent { run, node } if node == CHAT_NODE => Some(run.clone()),
            _ => None,
        };
        Self {
            schema: SCHEMA,
            id: new_id(),
            title,
            instruction,
            sources: crate::refs::Sources::default(),
            repo,
            source,
            priority: 0,
            solo: false,
            status: TaskStatus::Queued,
            attempts: 0,
            attempt_limit: None,
            runs: Vec::new(),
            last_error: None,
            hold_reason: None,
            hold_source: None,
            diagnostic: None,
            blocked_by: Vec::new(),
            block_reason: None,
            blocked_from: None,
            blocked_at: None,
            answers: Vec::new(),
            triage_applied: Vec::new(),
            actions_applied: Vec::new(),
            resume_override: None,
            resumption: None,
            resumption_history: Vec::new(),
            operator_decided_at: None,
            review_branch: None,
            review_instruction: None,
            pr_recovery: None,
            fresh_start: false,
            replacement: None,
            interrupt: false,
            urgent: false,
            attachments: Vec::new(),
            followup: None,
            origin_chat,
            overrides: None,
            review_of: None,
            held_at: None,
            park_reason: None,
            chat_notice: None,
            chat_notices: Vec::new(),
            hold_notice_cause: None,
            passes: Vec::new(),
            dispatched_at: None,
            recovery_budget: None,
            recovery_admission: None,
            recovery_generations: Vec::new(),
            created_at: now,
            updated_at: now,
        }
    }

    /// The chat this task itself says it came from: the recorded
    /// [`Task::origin_chat`], else its own source when a chat filed it.
    pub fn chat_talk(&self) -> Option<&str> {
        if let Some(id) = &self.origin_chat {
            return Some(id);
        }
        match &self.source {
            Source::Agent { run, node } if node == CHAT_NODE => Some(run),
            _ => None,
        }
    }

    /// The talk that filed this task *itself* (`Source::Agent` with
    /// [`CHAT_NODE`]); unlike [`Task::chat_talk`] a follow-up's inherited
    /// `origin_chat` does not count.
    pub fn filed_by_chat(&self) -> Option<&str> {
        match &self.source {
            Source::Agent { run, node } if node == CHAT_NODE => Some(run),
            _ => None,
        }
    }

    /// Identity of a material outcome. Prose and wall-clock times are never
    /// evidence. Dependencies are a set; held causes survive intermediate states.
    pub fn chat_notice_key(&self) -> Option<String> {
        match self.status {
            TaskStatus::Done => Some("done".to_owned()),
            TaskStatus::Held => {
                let cause = self
                    .hold_notice_cause
                    .clone()
                    .unwrap_or_else(|| self.unresolved_hold_cause());
                Some(format!(
                    "cause:{}",
                    serde_json::to_string(&cause).expect("cause")
                ))
            }
            TaskStatus::Blocked => {
                let ids: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
                // Only new verified blocking evidence versions the dependency
                // set. Freezing an unrelated hold on release must not do so.
                let evidence = self
                    .hold_notice_cause
                    .as_ref()
                    .filter(|c| c.stage.starts_with("conductor-evidence:"));
                Some(format!(
                    "dependencies:{}",
                    serde_json::to_string(&(ids, evidence)).expect("ids")
                ))
            }
            _ => None,
        }
    }

    fn unresolved_hold_cause(&self) -> NoticeCause {
        NoticeCause {
            stage: "hold".into(),
            target: None,
            run: self.runs.last().cloned(),
            pass: None,
        }
    }

    /// Exact old delivery identity, only for reconciling an existing record.
    pub(crate) fn legacy_chat_notice_key(&self) -> Option<String> {
        let key = match self.status {
            TaskStatus::Held if self.hold_notice_cause.is_none() => Some(format!(
                "held#{}",
                self.held_at.map(|t| t.to_string()).unwrap_or_default()
            )),
            TaskStatus::Blocked => Some(format!(
                "blocked#{}#{}",
                self.blocked_by.join(","),
                self.block_reason.as_deref().unwrap_or_default()
            )),
            _ => None,
        };
        key.filter(|key| self.legacy_notice_matches(key))
    }

    pub(crate) fn legacy_notice_matches(&self, key: &str) -> bool {
        match self.status {
            TaskStatus::Held if self.hold_notice_cause.is_none() => key
                .strip_prefix("held#")
                .is_some_and(|time| time.is_empty() || time.parse::<Timestamp>().is_ok()),
            TaskStatus::Blocked
                if self
                    .hold_notice_cause
                    .as_ref()
                    .is_none_or(|c| !c.stage.starts_with("conductor-evidence:")) =>
            {
                key.strip_prefix("blocked#").is_some_and(|tail| {
                    let Some((ids, _)) = tail.split_once('#') else {
                        return false;
                    };
                    let old: std::collections::BTreeSet<_> =
                        ids.split(',').filter(|id| !id.is_empty()).collect();
                    let current: std::collections::BTreeSet<_> =
                        self.blocked_by.iter().map(String::as_str).collect();
                    old == current
                })
            }
            _ => false,
        }
    }

    fn import_chat_notice(&mut self) {
        if let Some(n) = &self.chat_notice {
            let key = if is_done_notice_key(&n.key) {
                "done".to_owned()
            } else if self.legacy_notice_matches(&n.key) {
                self.chat_notice_key().unwrap_or_else(|| n.key.clone())
            } else {
                n.key.clone()
            };
            if !self.chat_notices.contains(&key) {
                self.chat_notices.push(key);
            }
        }
    }

    /// A producer records the processing stage and actual run/pass, never the
    /// error's rendered prose. No run means startup made no implementation progress.
    pub(crate) fn notice_cause(&mut self, stage: &str, run: Option<&str>, pass: Option<usize>) {
        self.import_chat_notice();
        self.hold_notice_cause = Some(NoticeCause {
            stage: stage.to_owned(),
            target: None,
            run: run.map(str::to_owned),
            pass,
        });
    }

    /// Startup retries cannot replace an unresolved run outcome without doing
    /// work. Distinct startup stages can still explain a newly encountered blocker.
    pub(crate) fn startup_notice_cause(&mut self, stage: &str) {
        let recovery_only = matches!(
            stage,
            "startup-source" | "startup-candidates" | "startup-public-head"
        );
        match self.hold_notice_cause.as_ref() {
            Some(cause) if recovery_only && !cause.stage.starts_with("startup-") => return,
            None if recovery_only
                && self
                    .chat_notice
                    .as_ref()
                    .is_some_and(|n| !is_done_notice_key(&n.key)) =>
            {
                return;
            }
            _ => {}
        }
        self.notice_cause(stage, None, None);
    }

    /// Apply a preflight failure without allowing Running's generic failure
    /// identity to overwrite the producer's stage or its unresolved recovery anchor.
    pub(crate) fn fail_startup(
        &mut self,
        stage: &str,
        why: impl Into<String>,
        max_attempts: usize,
    ) {
        let unresolved = self.hold_notice_cause.clone();
        self.fail(why, max_attempts);
        self.hold_notice_cause = unresolved;
        self.startup_notice_cause(stage);
    }

    /// A startup failure tied to a concrete object, such as a missing attachment.
    pub(crate) fn fail_startup_target(
        &mut self,
        stage: &str,
        target: String,
        why: impl Into<String>,
        max_attempts: usize,
    ) {
        self.fail_startup(stage, why, max_attempts);
        if let Some(cause) = self.hold_notice_cause.as_mut() {
            cause.target = Some(target);
        }
    }

    /// A chat-filed task owes each material final outcome at most one notice.
    pub fn chat_notice_due(&self) -> Option<String> {
        self.filed_by_chat()?;
        if self.schema < 13 {
            return None;
        }
        let key = self.chat_notice_key()?;
        if self.chat_notices.contains(&key) {
            return None;
        }
        match &self.chat_notice {
            Some(n)
                if n.key == key
                    || (key == "done" && is_done_notice_key(&n.key))
                    || self.legacy_notice_matches(&n.key) =>
            {
                None
            }
            _ => Some(key),
        }
    }

    /// Short form used in reports, matching a run's short id.
    pub fn short(&self) -> &str {
        short(&self.id)
    }

    /// Record that triage question `question_id`'s answer has been applied.
    pub fn mark_triage_applied(&mut self, question_id: &str) {
        if !self.triage_applied(question_id) {
            self.triage_applied.push(question_id.to_owned());
        }
    }

    /// Has the action of `magi ask` question `question_id` already been
    /// applied, or is it being applied now? See [`Task::actions_applied`].
    pub fn action_applied(&self, question_id: &str) -> bool {
        self.actions_applied.iter().any(|id| id == question_id)
    }

    /// Record that `magi ask` question `question_id`'s action was applied.
    pub fn mark_action_applied(&mut self, question_id: &str) {
        if !self.action_applied(question_id) {
            self.actions_applied.push(question_id.to_owned());
        }
    }

    /// Has triage question `question_id`'s answer already been applied?
    pub fn triage_applied(&self, question_id: &str) -> bool {
        self.triage_applied.iter().any(|id| id == question_id)
    }

    /// Record that a run has started for this task.
    /// Move an already linked run to the end so it remains the current attempt,
    /// with one history entry per run and separate pass records for each resume.
    ///
    /// Clears [`Task::interrupt`]: a mark to run ahead of whatever else is
    /// in flight is fulfilled the moment this task actually gets its turn,
    /// dispatched same as any other. Without this, a task whose run fails
    /// and requeues - still `runnable`, still carrying the mark from its
    /// first attempt - would keep re-triggering `crate::daemon`'s interrupt
    /// scheduler and re-parking whatever it interrupted on every later
    /// boundary, for as long as its attempts hold out, instead of the
    /// one-shot "let this go next" the mark is meant to be.
    pub fn start(&mut self, run: String) {
        self.stamp_schema();
        self.status = TaskStatus::Running;
        self.attempts += 1;
        self.runs.retain(|id| id != &run);
        self.runs.push(run);
        self.last_error = None;
        self.park_reason = None;
        self.fresh_start = false;
        self.interrupt = false;
        // The answer has been honoured: the task got its turn.
        self.resume_override = None;
    }

    /// Note that `run` serves this task without otherwise changing its life:
    /// no status, attempt or interrupt change, unlike [`Task::start`]. For a
    /// run somebody else started on the task's behalf (`magi run --task`);
    /// idempotent. Returns whether anything was added.
    pub fn link_run(&mut self, run: &str) -> bool {
        if self.runs.iter().any(|r| r == run) {
            return false;
        }
        self.runs.push(run.to_owned());
        true
    }

    /// Record a successful run.
    ///
    /// Both `magi task done` and `POST /api/queue/{id}/done` can close a held
    /// *or blocked* task directly, with no release in between, so this clears
    /// `hold_reason` and `blocked_by`/`block_reason` the same way
    /// [`Task::release`] does. Otherwise a task held for "waiting on 3ed9", or
    /// blocked on a dependency that never actually finished, and then closed
    /// as done without ever being released would still read as waiting on
    /// something in `magi task show` and on its card, after it no longer is.
    pub fn succeed(&mut self) {
        self.stamp_schema();
        self.status = TaskStatus::Done;
        self.resume_override = None;
        self.last_error = None;
        self.park_reason = None;
        self.hold_reason = None;
        self.hold_source = None;
        self.diagnostic = None;
        self.blocked_by.clear();
        self.block_reason = None;
        self.blocked_from = None;
    }

    /// Finish the task because the run's whole change turned out to be on the
    /// base already ([`crate::run::RunStatus::AlreadyInBase`]): `Done`, with
    /// `note` kept in [`Task::last_error`] so `magi task show` says why, and
    /// the attempt refunded. Nothing went wrong and nothing was spent
    /// misbehaving, so this is neither [`Task::fail`] nor
    /// [`Task::handed_off`]; and unlike [`Task::succeed`] it does not pretend
    /// the run landed anything.
    pub fn already_landed(&mut self, note: impl Into<String>) {
        self.succeed();
        self.attempts = self.attempts.saturating_sub(1);
        self.last_error = Some(note.into());
    }

    /// Earlier attempts at this task that a later one has since made moot —
    /// empty unless the task is [`TaskStatus::Done`] *and* `last_run_succeeded`
    /// says `runs.last()` is actually why.
    ///
    /// `runs` is oldest first, and [`Task::start`] is the only thing that
    /// pushes to it, always right before the attempt it names either succeeds
    /// or fails; `succeed` itself never touches `runs`. So whenever `status`
    /// is `Done` *because* the loop itself saw that last attempt land
    /// (`Merged`/`Ready`), everything before it in the same list is a retry
    /// this task no longer needs. But `succeed` is also reachable directly —
    /// `magi task done`, the web UI's equivalent, and the conductor's
    /// `Recovery::Done` all call it on a task in *any* status, including one
    /// whose last recorded attempt never landed at all (closed by hand after
    /// a merge magi's own loop never saw). `runs.last()` alone cannot tell
    /// those two cases apart — that requires the caller to have actually
    /// looked at that run's own `status`, which this module has no way to
    /// do — so `last_run_succeeded` is the caller's answer to exactly that
    /// question, not something this function can derive from `Task` alone.
    ///
    /// Deliberately narrower than [`Queue::superseded`]'s "every earlier
    /// attempt has a later one" walk, which fires the moment a retry starts
    /// even though the retry itself might still fail: that reading is right
    /// for the web UI's "a newer attempt exists, go look at that one
    /// instead" note, but wrong for deciding a run no longer needs a human's
    /// attention, which is only true once the task's story has actually
    /// ended well. Two Blocked runs sitting side by side while a third
    /// attempt is still in flight must not be touched by this — see
    /// `daemon::supersede_prior_runs`, the caller that turns this list into
    /// rewritten `run.json` files.
    pub fn superseded_attempts(&self, last_run_succeeded: bool) -> &[String] {
        if self.status != TaskStatus::Done || !last_run_succeeded || self.runs.len() < 2 {
            return &[];
        }
        &self.runs[..self.runs.len() - 1]
    }

    /// The attempt right after the last occurrence of `run`, if there is one.
    /// Resuming records the same run again; it cannot supersede itself.
    ///
    /// The one place "a later attempt replaced this run" is decided:
    /// [`Queue::superseded`] and [`Queue::superseded_by`] (the web UI's
    /// note) read it, and so does [`Task::earlier_attempts`], so what is shown
    /// and what `crate::handover` acts on cannot disagree.
    pub fn successor_of(&self, run: &str) -> Option<&String> {
        let pos = self.runs.iter().rposition(|r| r == run)?;
        self.runs.get(pos + 1)
    }

    /// Every run already recorded for this task, i.e. every run that an
    /// attempt being minted *right now* supersedes.
    ///
    /// The new run is not in [`Task::runs`] yet when it decides what to take
    /// over (`Task::start` pushes it afterwards), so "has a later attempt" is
    /// true of everything listed here by the time the new run exists.
    pub fn earlier_attempts(&self) -> &[String] {
        &self.runs
    }

    /// Stamp [`Task::held_at`] on the way into [`TaskStatus::Held`]; a task
    /// that is already held keeps the time its hold began.
    /// A transition this build makes (start, hold, block, done) makes the task
    /// eligible for a chat notice even when it was filed under an older
    /// schema; only records no build has touched since stay silent
    /// ([`Task::chat_notice_due`]).
    fn stamp_schema(&mut self) {
        self.schema = self.schema.max(SCHEMA);
    }

    fn note_held(&mut self) {
        self.stamp_schema();
        let continuing = self.status == TaskStatus::Held
            || (self.status == TaskStatus::Blocked
                && (self.blocked_from == Some(TaskStatus::Held)
                    || (self.blocked_from.is_none()
                        && (self.hold_reason.is_some() || self.hold_source.is_some()))));
        if !continuing || self.held_at.is_none() {
            self.held_at = Some(Timestamp::now());
        }
    }

    /// Record a failed attempt. Out of attempts means held for a human, rather
    /// than retried until the money runs out.
    ///
    /// Clears [`Task::diagnostic`] unconditionally: it belongs to whatever run
    /// produced it, and a caller that has one for *this* attempt sets it
    /// itself right after calling this, once it knows the task actually ended
    /// up [`TaskStatus::Held`] - see `daemon::diagnostic`. Without the clear, a
    /// task released after a diagnosed hold and then failed again for an
    /// unrelated, undiagnosed reason (a config error, say) would go on
    /// showing the previous run's diagnostic as if it explained the new one.
    ///
    /// Also records `why` into [`Task::hold_reason`] when this ends up
    /// [`TaskStatus::Held`], so the notification and `magi task show` say the
    /// same thing as [`Task::last_error`] instead of leaving the hold's own
    /// reason blank.
    pub fn fail(&mut self, why: impl Into<String>, max_attempts: usize) {
        let why = why.into();
        if self.status == TaskStatus::Running {
            let run = self.runs.last().cloned();
            self.notice_cause("run-failed", run.as_deref(), Some(self.attempts));
        }
        self.diagnostic = None;
        self.park_reason = None;
        self.status = if self.attempts >= self.effective_attempt_limit(max_attempts) {
            self.note_held();
            self.hold_source = Some(HoldSource::Machine);
            self.hold_reason = Some(why.clone());
            TaskStatus::Held
        } else {
            TaskStatus::Failed
        };
        self.last_error = Some(why);
    }

    /// Record an attempt that failed for a reason the task is not responsible
    /// for - the agent CLIs ran out of quota and the judging panel collapsed.
    ///
    /// This refunds the attempt on purpose. A quota window closing at 4am must
    /// not spend the backlog's retry budget: the operator would come back to a
    /// queue of held tasks that were never actually judged, and would have to
    /// release every one by hand to find out which had a real problem. The task
    /// goes back to `Failed`, which the loop retries, so a reset quota picks the
    /// work up where it stopped.
    pub fn stall(&mut self, why: impl Into<String>) {
        self.last_error = Some(why.into());
        self.diagnostic = None;
        self.attempts = self.attempts.saturating_sub(1);
        self.status = TaskStatus::Failed;
    }

    /// Record a run that parked at a node boundary. Nothing failed: the attempt
    /// is refunded, [`Task::last_error`] is left as it was, the run history and
    /// `fresh_start` are kept so the next loop resumes the same run, and the
    /// reason goes to [`Task::park_reason`]. Runnable like a requeued task and
    /// never counted against `max_attempts`.
    pub fn park(&mut self, why: impl Into<String>) {
        self.diagnostic = None;
        self.attempts = self.attempts.saturating_sub(1);
        self.status = TaskStatus::Parked;
        self.park_reason = Some(why.into());
        // The variant and field are schema 11; stamp it so an older build
        // refuses the record instead of half-reading it.
        self.schema = self.schema.max(SCHEMA);
    }

    /// Record a settled pass whose start time is unknown (see [`Self::record_pass_at`]).
    pub fn record_pass(
        &mut self,
        run: &str,
        pass_index: usize,
        attempts_before: usize,
        parked: Option<crate::run::ParkCause>,
    ) {
        self.record_pass_at(run, pass_index, None, attempts_before, parked);
    }

    /// Record a settled pass. `attempts_before` is the count just before the
    /// settle, so a refund is read off what actually changed, not inferred.
    pub fn record_pass_at(
        &mut self,
        run: &str,
        pass_index: usize,
        started_at: Option<Timestamp>,
        attempts_before: usize,
        parked: Option<crate::run::ParkCause>,
    ) {
        if self
            .passes
            .iter()
            .any(|p| p.run == run && p.pass_index == pass_index)
        {
            return;
        }
        let cost = if self.attempts < attempts_before {
            PassCost::Refunded
        } else {
            PassCost::Spent
        };
        self.passes.push(TaskPass {
            run: run.to_owned(),
            pass_index,
            started_at: self.dispatched_at.or(started_at),
            cost,
            attempts_after: self.attempts,
            settled_at: Timestamp::now(),
            parked,
        });
        self.schema = self.schema.max(SCHEMA);
    }

    /// Only the latest pass of the current, resumable upgrade-parked run
    /// retains its place. Explicit fresh starts and other park causes do not.
    pub(crate) fn upgrade_resume_key(&self) -> Option<(Timestamp, &str)> {
        if self.status != TaskStatus::Parked || self.fresh_start {
            return None;
        }
        let pass = self.passes.last()?;
        if self.runs.last() != Some(&pass.run)
            || !matches!(pass.parked, Some(crate::run::ParkCause::Upgrade { .. }))
        {
            return None;
        }
        // Before schema 19 the dispatch start was not copied to the task.
        // Settlement time is only an approximation for those old records.
        Some((pass.started_at.unwrap_or(pass.settled_at), &self.id))
    }

    /// Whether this held task may only be released by an operator.
    ///
    /// Old files did not record a source. Preserve every such hold rather
    /// than guessing that it was automatic and risking duplicate work. New
    /// automatic holds record [`HoldSource::Machine`] and remain recoverable.
    pub fn operator_held(&self) -> bool {
        self.status == TaskStatus::Held && !matches!(self.hold_source, Some(HoldSource::Machine))
    }

    /// Take this task out of the loop's reach by an operator action.
    ///
    /// Clears `blocked_by`/`block_reason` unconditionally, the same as
    /// [`Task::release`] and for the same reason its own comment already
    /// gives: a human choosing to hold a *blocked* task overrides its wait
    /// outright, the same as it overrides an ordinary hold. Without this, a
    /// task held straight out of [`TaskStatus::Blocked`] - the web UI's "Hold"
    /// button is reachable on a blocked task, same as "Mark done" - kept
    /// reading as still waiting on a dependency it no longer had any claim on.
    pub fn hold_manual(&mut self, reason: Option<String>) {
        // A later explicit hold supersedes outstanding triage answers.
        self.operator_decided_at = Some(Timestamp::now());
        self.hold_unrequested_retry(reason);
    }

    /// Keep a hand-started run from retrying automatically, without recording
    /// an operator decision that could invalidate its outstanding questions.
    pub(crate) fn hold_unrequested_retry(&mut self, reason: Option<String>) {
        self.park_reason = None;
        self.note_held();
        self.status = TaskStatus::Held;
        if reason.is_some() {
            self.hold_reason = reason;
        }
        self.hold_source = Some(HoldSource::Manual);
        self.blocked_by.clear();
        self.block_reason = None;
        self.blocked_from = None;
    }

    /// Take this task out of the loop's reach during automatic recovery.
    ///
    /// Clears `blocked_by`/`block_reason` for the same reason
    /// [`Task::hold_manual`] does.
    pub fn hold_machine(&mut self, reason: Option<String>) {
        self.park_reason = None;
        self.note_held();
        self.status = TaskStatus::Held;
        if reason.is_some() {
            self.hold_reason = reason;
        }
        self.hold_source = Some(HoldSource::Machine);
        self.blocked_by.clear();
        self.block_reason = None;
        self.blocked_from = None;
    }

    /// Block this task on other task ids and/or open question ids, chosen by
    /// `crate::conduct`. Pure: the caller still owns writing it back with
    /// [`Queue::put`].
    ///
    /// Records [`Task::blocked_from`] the first time this moves the task into
    /// [`TaskStatus::Blocked`], and leaves it alone on a later call that adds
    /// or replaces `blocked_by` while the task is already `Blocked` - a
    /// second question about an already-blocked task must not overwrite the
    /// status it should eventually return to with `Blocked` itself.
    pub fn block(&mut self, blocked_by: Vec<String>, reason: Option<String>) {
        self.stamp_schema();
        let before: std::collections::BTreeSet<_> = self.blocked_by.iter().collect();
        let after: std::collections::BTreeSet<_> = blocked_by.iter().collect();
        if self.status != TaskStatus::Blocked || before != after {
            self.blocked_at = Some(Timestamp::now());
        }
        if self.status != TaskStatus::Blocked {
            self.blocked_from = Some(self.status);
        }
        self.status = TaskStatus::Blocked;
        self.blocked_by = blocked_by;
        self.block_reason = reason;
    }

    /// Remove one resolved dependency (a task id that became [`TaskStatus::Done`],
    /// or a question id that became [`crate::ask::QuestionStatus::Answered`]).
    /// Once nothing is left in [`Task::blocked_by`], the task returns to
    /// whatever [`Task::blocked_from`] recorded - deciding *why* a task was
    /// blocked was `crate::conduct`'s job, but noticing a dependency resolved
    /// needs no model at all, and restoring the status it interrupted needs
    /// nothing more than what `block` already wrote down.
    ///
    /// A task blocked while `Running` restores to [`TaskStatus::Queued`]
    /// instead: whatever process was running it is gone by the time this
    /// runs, so there is nothing left to resume. A task with no recorded
    /// `blocked_from` - a pre-schema-4 record, or one blocked before this
    /// field existed - falls back to [`TaskStatus::Held`] when it still
    /// carries hold evidence ([`Task::hold_reason`] or [`Task::hold_source`],
    /// neither ever cleared by `block`), and to `Queued` otherwise: the same
    /// choice `block` itself would have recorded, reconstructed from what
    /// survived.
    ///
    /// A no-op, on purpose, for a task that is not [`TaskStatus::Blocked`]:
    /// `crate::daemon`'s deterministic resolver runs over every task on every
    /// poll, and a task that moved on for some other reason must not be
    /// dragged back by a stale id it still happens to carry.
    pub fn unblock(&mut self, resolved_id: &str) {
        if self.status != TaskStatus::Blocked {
            return;
        }
        self.blocked_by.retain(|id| id != resolved_id);
        self.restore_if_unblocked();
    }

    /// A dependency this task waited on was deleted on purpose
    /// ([`Queue::remove`]): drop just that id from [`Task::blocked_by`] and,
    /// once nothing is left, return to the status the block interrupted, the
    /// same way [`Task::unblock`] does. Returns whether anything changed.
    ///
    /// Never a hold: the operator chose the deletion, so the dependent is not
    /// in trouble. Touches nothing but a `Blocked` task that actually names
    /// `deleted_id`, so a `Held` task (or a manual hold) is left exactly as it
    /// was. A dependency that vanished *without* going through a deletion is
    /// not this method's case - that stays a machine hold.
    pub fn dependency_deleted(&mut self, deleted_id: &str) -> bool {
        if self.status != TaskStatus::Blocked || !self.blocked_by.iter().any(|b| b == deleted_id) {
            return false;
        }
        self.blocked_by.retain(|id| id != deleted_id);
        self.restore_if_unblocked();
        true
    }

    /// The shared tail of [`Task::unblock`] and [`Task::dependency_deleted`]:
    /// with nothing left in `blocked_by`, restore the interrupted status and
    /// clear the now-stale block bookkeeping.
    fn restore_if_unblocked(&mut self) {
        if self.blocked_by.is_empty() {
            self.status = match self.blocked_from {
                Some(TaskStatus::Running) => TaskStatus::Queued,
                Some(other) => other,
                None if self.hold_reason.is_some() || self.hold_source.is_some() => {
                    TaskStatus::Held
                }
                None => TaskStatus::Queued,
            };
            self.block_reason = None;
            self.blocked_from = None;
        }
    }

    /// Record that a question `crate::conduct` asked about this task has been
    /// answered, so the answer's content — not just the fact that the
    /// question is gone — reaches the next conductor prompt and the next
    /// run's instruction. See [`Task::answers`].
    pub fn record_answer(&mut self, question: String, answer: String) {
        self.answers.push(AnsweredQuestion { question, answer });
    }

    /// Requeue this task to reopen its last run as a review-only pass against
    /// `branch` (`crate::graph::Runner::review`) rather than competing from
    /// scratch. See [`Task::review_branch`].
    pub fn request_review(&mut self, branch: String) {
        self.release();
        self.review_branch = Some(branch);
        self.review_instruction = Some(self.instruction.clone());
    }

    /// Requeue after a conductor chose a new competition. Unlike an ordinary
    /// operator release, this deliberately does not resume the old run.
    pub fn requeue(&mut self) {
        self.release();
        // A fresh competition is the opposite of a review of the old branch.
        self.review_branch = None;
        self.review_instruction = None;
        self.fresh_start = true;
    }

    /// Hold this task because a takeover of `branch` was refused, keeping
    /// [`Task::review_branch`] so that once the operator has cleaned up and
    /// released it, the retry is a review of the same branch rather than a
    /// resume or a fresh competition. Spends no attempt.
    ///
    /// This is the only way a machine hold keeps a `review_branch`: the
    /// daemon takes it at the start of every attempt, so [`Task::release`]
    /// relies on that to tell this hold from every other.
    pub fn hold_for_handover(&mut self, branch: Option<String>, reason: String) {
        if branch.is_some() {
            self.review_branch = branch;
            self.review_instruction = Some(self.instruction.clone());
        }
        self.hold_machine(Some(reason));
    }

    /// Change the urgent lane setting without releasing a held task.
    pub fn set_urgent(&mut self, urgent: bool) -> Result<()> {
        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
            bail!(
                "task {} is {}; urgent can only change while queued or held; \
                 wait for the active attempt to finish and requeue or hold the task first",
                self.short(),
                self.status.as_str()
            );
        }
        self.urgent = urgent;
        Ok(())
    }

    /// Change how urgently this task should run next.
    ///
    /// Refused once the task is `running`: priority only feeds the sort
    /// [`Queue::next_runnable`] does over tasks waiting to be claimed, and a
    /// running task has already left that pool. Accepting the write anyway
    /// would look like it worked while changing nothing until - and unless -
    /// this attempt fails and the task becomes runnable again, which is a
    /// surprise the phone should not hand back as a success.
    pub fn set_priority(&mut self, priority: i32) -> Result<()> {
        if self.status == TaskStatus::Running {
            bail!(
                "task {} is running; its priority cannot be changed until \
                 this attempt finishes",
                self.short()
            );
        }
        self.priority = priority;
        Ok(())
    }

    /// Mark (or unmark) this task to interrupt whatever `magi serve` already
    /// has in flight, once `[daemon] pause_for_interrupts` is on. See
    /// [`Task::interrupt`].
    ///
    /// Setting it is restricted to a task the loop could pick up on its own
    /// right now - [`TaskStatus::runnable`] - for the same reason as
    /// [`Task::set_priority`]: a task already `running` has been claimed, and
    /// a task that is `done`, `held`, or `blocked` is not going to compete
    /// for the daemon's attention regardless of this flag. Unlike priority,
    /// this is never silently inert while `running` - it is refused outright,
    /// because the entire feature this flag drives (`crate::daemon`'s
    /// interrupt scheduler) is scoped to tasks still waiting to be claimed.
    /// Clearing it back to `false` carries no such risk and is always
    /// allowed, including on a task that moved on since it was set.
    pub fn set_interrupt(&mut self, interrupt: bool) -> Result<()> {
        if interrupt && !self.status.runnable() {
            bail!(
                "task {} is {}; only a queued or failed task can be marked \
                 to interrupt",
                self.short(),
                self.status.as_str()
            );
        }
        self.interrupt = interrupt;
        Ok(())
    }

    /// Resolve the task override against the running daemon's limit.
    pub fn effective_attempt_limit(&self, daemon_limit: usize) -> usize {
        self.attempt_limit.map_or(daemon_limit, NonZeroUsize::get)
    }

    /// Change the budget without resetting attempts or releasing a hold.
    /// Callers must hold the queue claim and reload the task before calling.
    pub fn change_attempt_limit(&mut self, limit: Option<NonZeroUsize>) -> Result<()> {
        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
            bail!("only queued or held tasks can change the attempt limit");
        }
        self.stamp_schema();
        self.attempt_limit = limit;
        Ok(())
    }

    /// Explicit implementer choice, with Solo taking precedence for old records.
    pub fn implementers(&self) -> Option<usize> {
        if self.solo {
            Some(1)
        } else {
            self.overrides.as_ref().and_then(|o| o.candidates)
        }
    }

    /// Validate against this task's configuration and change only its seat count.
    /// Callers must hold the queue claim and reload the task before calling.
    pub fn change_implementers(&mut self, count: u8) -> Result<()> {
        self.check_implementers_editable()?;
        let explicit = self.overrides.as_ref().and_then(|o| o.config.as_deref());
        let (mut config, _) = crate::config::Config::discover(&self.repo, explicit)?;
        if let Some(o) = &self.overrides {
            o.apply(&mut config);
        }
        self.set_implementers(count, &config)
    }

    fn check_implementers_editable(&self) -> Result<()> {
        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
            bail!(
                "task {} is {}; only queued or held tasks can change implementers",
                self.short(),
                self.status.as_str()
            );
        }
        Ok(())
    }

    /// Normalize Solo and candidate overrides without altering task history.
    pub fn set_implementers(&mut self, count: u8, config: &crate::config::Config) -> Result<()> {
        self.check_implementers_editable()?;
        crate::talk::check_implementers(count, config).map_err(anyhow::Error::msg)?;
        self.stamp_schema();
        self.solo = count == 1;
        let mut overrides = self.overrides.clone().unwrap_or_else(|| RunOverrides {
            inherit_daemon_options: true,
            ..Default::default()
        });
        overrides.candidates = (count > 1).then_some(usize::from(count));
        let mut settings = overrides.clone();
        settings.inherit_daemon_options = false;
        // Some(empty) is a direct-run setting too: retain its isolation on Solo.
        self.overrides = if overrides.inherit_daemon_options && settings == RunOverrides::default()
        {
            None
        } else {
            Some(overrides)
        };
        Ok(())
    }

    /// Replace this task's title and instruction wholesale.
    ///
    /// Restricted to `queued` and `held`. A `running` task's instruction has
    /// already been handed to the graph, so a run in flight and the file on
    /// disk must not be allowed to disagree about what was asked; a `done` or
    /// `failed` task is a record of what actually happened and editing it
    /// after the fact would falsify that record. `id`, `created_at`,
    /// `source`, and `runs` are left untouched on purpose - an edit stands in
    /// for "delete and refile", and keeping the id, the timestamp, the
    /// attribution, and the run history is the entire reason it exists
    /// instead.
    pub fn edit(&mut self, title: String, instruction: String) -> Result<()> {
        if !matches!(self.status, TaskStatus::Queued | TaskStatus::Held) {
            bail!(
                "task {} is {}; only a queued or held task's instruction can \
                 be edited",
                self.short(),
                self.status.as_str()
            );
        }
        // Older retained reviews have no criteria snapshot yet. Capture the
        // old instruction before this supported edit can change the target.
        if self.review_branch.is_some() && self.review_instruction.is_none() {
            self.review_instruction = Some(self.instruction.clone());
        }
        if self.instruction != instruction {
            // A run selected for earlier criteria is no longer authorized by
            // this revised instruction. Title-only edits preserve the pin.
            self.resume_override = None;
        }
        self.title = title;
        self.instruction = instruction;
        Ok(())
    }

    /// Record a run that produced a pull request without merging it.
    ///
    /// The task is held rather than retried, and it costs no further attempt
    /// either way. The work the task asked for exists: it is sitting on a
    /// branch, in a pull request, waiting for CI or for a person. Retrying
    /// would spend the whole competition budget a second time and then race a
    /// second branch against the pull request the first one opened - which is
    /// exactly what happened to run 01c2, whose finished and green pull request
    /// was re-competed from scratch four seconds after it opened.
    ///
    /// A pull request nobody merged is a request for a person, not a failure.
    ///
    /// Records `why` into [`Task::hold_reason`] as well as
    /// [`Task::last_error`], so the notification centre and `magi task show`
    /// say why the task is held rather than "no reason recorded". This is
    /// also the path a verified no-op with an outstanding `magi ask` question
    /// settles through - see `daemon::settle` and `daemon::settle_and_diagnose`,
    /// which append the question id to `hold_reason` when one is still open
    /// for the run.
    pub fn handed_off(&mut self, why: impl Into<String>) {
        let why = why.into();
        self.diagnostic = None;
        self.note_held();
        self.status = TaskStatus::Held;
        self.hold_source = Some(HoldSource::Machine);
        self.hold_reason = Some(why.clone());
        self.last_error = Some(why);
    }

    /// Release on explicit operator authority, rather than automatic recovery.
    pub fn release_operator(&mut self, source: &str, question_id: Option<&str>) {
        self.recovery_budget = None;
        self.recovery_generations.clear();
        self.recovery_admission = None;
        if self
            .resumption
            .as_ref()
            .is_some_and(|r| r.instruction != self.instruction)
        {
            // Also cover persisted revisions whose earlier approval still
            // carries a transient pin when this release is applied.
            self.resume_override = None;
        }
        let approval = Resumption {
            generation: self.resumption.as_ref().map_or(1, |r| r.generation + 1),
            source: source.to_owned(),
            question_id: question_id.map(str::to_owned),
            at: Timestamp::now(),
            instruction: self.instruction.clone(),
            answers_through: self.answers.len(),
            runs_through: self.runs.len(),
            blocking_evidence: Vec::new(),
            cleared_reason: self
                .hold_reason
                .clone()
                .or_else(|| self.block_reason.clone()),
        };
        self.release();
        self.operator_decided_at = Some(approval.at);
        if let Some(previous) = self.resumption.replace(approval) {
            self.resumption_history.push(previous);
        }
    }

    /// Record a failed graph pass without allowing automatic requeues to refund it.
    pub(crate) fn recovery_failed(&mut self, state: &crate::run::RunState) {
        let (base, inputs) = self.recovery_admission.as_ref().map_or_else(
            || (state.base_commit.clone(), recovery_inputs(&state.sources)),
            |a| (a.base.clone(), a.inputs.clone()),
        );
        let progress = recovery_progress(state);
        let stage = if state.prep_failure.is_some() {
            "preparation"
        } else {
            "execution"
        };
        let matches = |b: &RecoveryBudget| {
            b.instruction == self.instruction
                && b.base == base
                && b.inputs == inputs
                && (progress.is_none() || b.progress == progress)
        };
        if !self.recovery_budget.as_ref().is_some_and(matches) {
            let previous = self
                .recovery_generations
                .iter()
                .position(matches)
                .map(|index| self.recovery_generations.remove(index));
            if let Some(old) = self.recovery_budget.take() {
                self.recovery_generations.push(old);
            }
            self.recovery_budget = previous;
        }
        if let Some(budget) = &mut self.recovery_budget {
            budget.failures = budget.failures.saturating_add(1);
            budget.stage = stage.to_owned();
        } else {
            self.recovery_budget = Some(RecoveryBudget {
                instruction: self.instruction.clone(),
                base,
                inputs,
                stage: stage.to_owned(),
                failures: 1,
                progress,
            });
        }
    }

    pub(crate) fn recovery_exhausted(
        &self,
        base: &str,
        inputs: &str,
        limit: usize,
    ) -> Option<String> {
        let same_input = |b: &&RecoveryBudget| {
            b.instruction == self.instruction && b.base == base && b.inputs == inputs
        };
        let budget = self
            .recovery_budget
            .as_ref()
            .filter(same_input)
            .or_else(|| self.recovery_generations.iter().rev().find(same_input))?;
        (budget.instruction == self.instruction && budget.base == base && budget.inputs == inputs
            && budget.failures >= limit.max(1)).then(|| format!(
                "[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.",
                budget.failures, budget.stage))
    }

    /// Put a held or finished task back in line, with its attempt count reset
    /// so a release is a real second chance rather than an instant re-hold.
    /// The run history is kept: attempts reset, evidence does not.
    pub fn release(&mut self) {
        // Freeze the unresolved cause before a recovery can mint a run without
        // doing work. Approval and timestamps alone are not new outcomes.
        self.import_chat_notice();
        if self.hold_notice_cause.is_none()
            && matches!(self.status, TaskStatus::Held | TaskStatus::Blocked)
        {
            self.hold_notice_cause = Some(self.unresolved_hold_cause());
        }
        let refused_handover = self.status == TaskStatus::Held
            && self.hold_source == Some(HoldSource::Machine)
            && self.review_branch.is_some()
            && self
                .review_instruction
                .as_ref()
                .is_none_or(|i| *i == self.instruction);
        self.status = TaskStatus::Queued;
        self.held_at = None;
        self.attempts = 0;
        self.last_error = None;
        self.park_reason = None;
        // Otherwise the next person who holds this task reads a reason that
        // belonged to whatever it was waiting on last time.
        self.hold_reason = None;
        self.hold_source = None;
        self.diagnostic = None;
        // A release also un-blocks: the dependency or question `blocked_by`
        // named may still be unresolved, but a human (or `crate::conduct`)
        // choosing to release the task overrides that wait outright, the same
        // as it overrides an ordinary hold.
        self.blocked_by.clear();
        self.block_reason = None;
        self.blocked_from = None;
        // A machine hold that still names a review branch is a refused
        // handover (see [`Task::hold_for_handover`]): the retry must review
        // that same branch. Any other release drops it.
        if !refused_handover {
            self.review_branch = None;
            self.review_instruction = None;
        }
        self.fresh_start = false;
    }
}

/// Copy `src` into `dir` as `name`, or `stem-2.ext`, `stem-3.ext`, ... when
/// that is taken. `create_new` makes the collision check atomic and also
/// catches a case-insensitive filesystem. Returns the stored name and path.
fn copy_new(dir: &Path, src: &Path, name: &str) -> Result<(String, PathBuf)> {
    use std::io::ErrorKind;
    let (stem, ext) = match name.rfind('.') {
        Some(i) if i > 0 => (&name[..i], &name[i..]),
        _ => (name, ""),
    };
    for n in 1u32.. {
        let candidate = if n == 1 {
            name.to_owned()
        } else {
            let suffix = format!("-{n}");
            let room = 64usize.saturating_sub(suffix.len() + ext.len());
            let stem: String = stem.chars().take(room).collect();
            format!("{stem}{suffix}{ext}")
        };
        if !crate::ask::valid_asset_name(&candidate) {
            bail!("no valid attachment name is left for `{name}`");
        }
        let path = dir.join(&candidate);
        match std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&path)
        {
            Ok(mut out) => {
                let copied = std::fs::File::open(src)
                    .and_then(|mut input| std::io::copy(&mut input, &mut out));
                if let Err(e) = copied {
                    drop(out);
                    let _ = std::fs::remove_file(&path);
                    return Err(e).with_context(|| format!("copy {}", src.display()));
                }
                return Ok((candidate, path));
            }
            Err(e) if e.kind() == ErrorKind::AlreadyExists => continue,
            Err(e) => return Err(e).with_context(|| format!("create {}", path.display())),
        }
    }
    unreachable!("the counter never runs out")
}

/// Minimum age before a dead writer's lock may be recovered, and maximum wait
/// for a contending writer. Age alone never revokes a live writer's ownership.
const TASK_LOCK_STALE: std::time::Duration = std::time::Duration::from_secs(10);

/// Guard for [`Queue::lock_task`]; removes the lock file on drop, but only
/// while the file still carries this guard's owner token.
pub(crate) struct TaskLock {
    path: PathBuf,
    token: String,
}

/// A token no other holder shares: pid, then a mix of entropy and a process
/// counter so threads born in the same nanosecond still differ.
fn owner_token() -> String {
    static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
    let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let mut r = crate::rng::SplitMix64::new(crate::rng::entropy() ^ n.rotate_left(32));
    format!("{}-{:016x}", std::process::id(), r.next_u64())
}

/// The marker that serializes everyone removing the lock that carries `token`.
/// Hashed (FNV-1a) so an empty or foreign token is still a valid file name.
fn break_marker(path: &Path, token: &str) -> PathBuf {
    let mut h: u64 = 0xcbf29ce484222325;
    for b in token.bytes() {
        h ^= u64::from(b);
        h = h.wrapping_mul(0x100000001b3);
    }
    let mut name = path.as_os_str().to_owned();
    name.push(format!(".break-{h:016x}"));
    PathBuf::from(name)
}

fn older_than_stale(path: &Path) -> bool {
    std::fs::metadata(path)
        .and_then(|m| m.modified())
        .ok()
        .and_then(|t| t.elapsed().ok())
        .is_some_and(|age| age > TASK_LOCK_STALE)
}

fn writer_may_be_gone(token: &str) -> bool {
    let Some(pid) = token
        .split_once('-')
        .and_then(|(pid, _)| pid.parse::<u32>().ok())
    else {
        // Legacy/incomplete tokens have no valid writer identity. They remain
        // subject to the age and compare/remove checks.
        return true;
    };
    if pid == std::process::id() {
        return false;
    }
    // Unknown process-table access must not steal a long-running retirement.
    crate::proc::pid_status(pid) == Some(false)
}

/// How many levels of "a stale marker guarding a stale marker" are recovered
/// before giving up. Each level needs its own crashed remover, so the cap is
/// never reached without several independent crashes.
const MAX_MARKER_DEPTH: u8 = 3;

/// Take the break marker for `token`, or `None` if another remover holds it.
/// The marker carries its own owner token. One older than [`TASK_LOCK_STALE`]
/// was left by a remover that died; it is cleared by the same compare-and-
/// remove discipline as a lock, guarded by a marker of its own, so a recoverer
/// that judged it stale cannot delete a fresh marker that replaced it.
fn take_marker(
    path: &Path,
    token: &str,
    depth: u8,
    retry: crate::lock_file::Retry,
) -> Option<(PathBuf, String)> {
    use std::io::Write;
    let marker = break_marker(path, token);
    for _ in 0..2 {
        match retry.create(&marker) {
            Ok(mut f) => {
                let mine = owner_token();
                if f.write_all(mine.as_bytes()).is_err() {
                    drop(f);
                    let _ = std::fs::remove_file(&marker);
                    return None;
                }
                return Some((marker, mine));
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                let judged = std::fs::read_to_string(&marker).ok();
                match judged.filter(|_| older_than_stale(&marker)) {
                    Some(judged) if depth < MAX_MARKER_DEPTH => {
                        remove_lock_if(&marker, &judged, true, depth + 1, retry)?;
                    }
                    _ => return None,
                }
            }
            Err(_) => return None,
        }
    }
    None
}

/// Remove the lock at `path` if it still carries `token` (and, when
/// `require_stale`, is still stale). Removal is serialized per token by the
/// break marker, so a remover that judged an old lock cannot delete the fresh
/// one that replaced it. `None` means the marker was busy: try again.
fn remove_lock_if(
    path: &Path,
    token: &str,
    require_stale: bool,
    depth: u8,
    retry: crate::lock_file::Retry,
) -> Option<bool> {
    let (marker, mine) = take_marker(path, token, depth, retry)?;
    let still = std::fs::read_to_string(path).is_ok_and(|c| c == token)
        && (!require_stale || older_than_stale(path) && writer_may_be_gone(token));
    if still {
        let _ = std::fs::remove_file(path);
    }
    // Release the marker only if it is still ours.
    if depth >= MAX_MARKER_DEPTH {
        if std::fs::read_to_string(&marker).is_ok_and(|c| c == mine) {
            let _ = std::fs::remove_file(&marker);
        }
    } else {
        let _ = remove_lock_if(&marker, &mine, false, depth + 1, retry);
    }
    Some(still)
}

/// Break the lock judged stale while it held `judged`. Returns whether it was
/// removed; a lock that has since been replaced is left alone.
#[cfg(test)]
fn break_stale(path: &Path, judged: &str) -> bool {
    break_stale_with(path, judged, crate::lock_file::Retry::new(None))
}

fn break_stale_with(path: &Path, judged: &str, retry: crate::lock_file::Retry) -> bool {
    remove_lock_if(path, judged, true, 0, retry).unwrap_or(false)
}

impl Drop for TaskLock {
    fn drop(&mut self) {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
        let retry = crate::lock_file::Retry::new(Some(deadline));
        loop {
            match remove_lock_if(&self.path, &self.token, false, 0, retry) {
                Some(_) => return,
                None if std::time::Instant::now() > deadline => return,
                None => std::thread::sleep(std::time::Duration::from_millis(5)),
            }
        }
    }
}

/// A queue on disk.
#[derive(Debug, Clone)]
pub struct Queue {
    root: PathBuf,
}

impl Queue {
    /// The operator's queue, `<home>/queue`.
    pub fn open() -> Self {
        Self::at(crate::run::home().join("queue"))
    }

    /// A queue at an explicit root. Tests use this; so could an operator who
    /// wants a queue per project.
    pub fn at(root: PathBuf) -> Self {
        Self { root }
    }

    /// Directory holding the task files.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Path for one task id.
    pub fn path_of(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.json"))
    }

    /// Directory holding a task's attachments. A sibling of the task file,
    /// not a `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
    pub fn attachments_dir(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.attachments"))
    }

    /// Copy each of `sources` into `task`'s attachment directory and record
    /// the stored names on `task` (persist with [`Queue::put`]). Returns the
    /// names, in order.
    ///
    /// Every source is checked (a readable file, a name passing
    /// [`crate::ask::valid_asset_name`]) before anything is copied, and a
    /// copy that fails part-way removes only what this call created. A name
    /// already taken is never overwritten: it becomes `stem-2.ext`,
    /// `stem-3.ext`, ... A name that does not validate is refused rather than
    /// sanitised - the operator renames the file.
    pub fn attach(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
        let mut wanted = Vec::new();
        for src in sources {
            let name = src
                .file_name()
                .and_then(|n| n.to_str())
                .with_context(|| format!("`{}` has no usable file name", src.display()))?;
            if !crate::ask::valid_asset_name(name) {
                bail!(
                    "attachment name `{name}` must match ^[A-Za-z0-9][A-Za-z0-9._-]{{0,63}}$ \
                     with no `..`; rename the file and try again"
                );
            }
            if !src.is_file() {
                bail!("attachment `{}` is not a file", src.display());
            }
            wanted.push((src, name));
        }
        let dir = self.attachments_dir(&task.id);
        let existed = dir.is_dir();
        std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
        let mut created: Vec<PathBuf> = Vec::new();
        let mut names = Vec::new();
        let mut copy_all = || -> Result<()> {
            for (src, name) in &wanted {
                let (stored, path) = copy_new(&dir, src, name)?;
                created.push(path);
                names.push(stored);
            }
            Ok(())
        };
        if let Err(e) = copy_all() {
            for path in &created {
                let _ = std::fs::remove_file(path);
            }
            if !existed {
                let _ = std::fs::remove_dir(&dir);
            }
            return Err(e);
        }
        task.attachments.extend(names.iter().cloned());
        Ok(names)
    }

    /// [`Queue::attach`] then [`Queue::put`], undoing the copies when the
    /// record cannot be written. Without the undo the files sit in
    /// `<id>.attachments/` with no task pointing at them.
    ///
    /// Only what this call copied is removed (the names `attach` returned,
    /// and the directory itself only when it did not exist before and is
    /// empty), never an attachment an earlier save recorded. `task` is rolled
    /// back too, and the error the caller sees is `put`'s own. This relies on
    /// a failed `put` leaving the stored record untouched, which its
    /// write-then-rename guarantees.
    pub fn attach_and_put(&self, task: &mut Task, sources: &[PathBuf]) -> Result<Vec<String>> {
        let dir = self.attachments_dir(&task.id);
        let existed = dir.is_dir();
        let before = task.attachments.len();
        let names = self.attach(task, sources)?;
        if let Err(e) = self.put(task) {
            for name in &names {
                let _ = std::fs::remove_file(dir.join(name));
            }
            if !existed {
                let _ = std::fs::remove_dir(&dir);
            }
            task.attachments.truncate(before);
            return Err(e);
        }
        Ok(names)
    }

    /// Absolute paths of `task`'s attachments, whatever shape this queue's
    /// root has. Absolute because the prompt hands them to an agent whose
    /// working directory is somewhere else entirely.
    pub fn attachment_paths(&self, task: &Task) -> Vec<PathBuf> {
        let dir = self.attachments_dir(&task.id);
        task.attachments
            .iter()
            .map(|n| {
                let p = dir.join(n);
                std::path::absolute(&p).unwrap_or(p)
            })
            .collect()
    }

    /// Write a task, atomically, so a daemon killed mid-write leaves the
    /// previous state readable rather than a truncated file.
    ///
    /// Runs already recorded on disk that `task` does not carry are kept, not
    /// dropped: `magi run --task` links a run into a task from another
    /// process ([`Queue::link_run`]) while a daemon may hold a snapshot taken
    /// before it, and writing that snapshot back whole would erase the link.
    /// Nothing ever removes a run from a task, so the union loses nothing.
    pub fn put(&self, task: &mut Task) -> Result<()> {
        let _lock = self.lock_task(&task.id)?;
        self.put_unlocked(task)
    }

    /// Write `task` only if no task with its id exists yet; `Ok(false)` when
    /// one does. Unlike [`Queue::put`] it never replaces a record, so a
    /// producer with a deterministic id cannot rewind a task that has since
    /// started running or finished.
    pub fn create_new(&self, task: &mut Task) -> Result<bool> {
        let _lock = self.lock_task(&task.id)?;
        if self.path_of(&task.id).exists() {
            return Ok(false);
        }
        self.put_unlocked(task)?;
        Ok(true)
    }

    /// [`Queue::put`] for a caller already holding [`Queue::lock_task`].
    fn put_unlocked(&self, task: &mut Task) -> Result<()> {
        if let Ok(mut stored) = read_path(&self.path_of(&task.id)) {
            stored.import_chat_notice();
            for key in stored.chat_notices {
                if !task.chat_notices.contains(&key) {
                    task.chat_notices.push(key);
                }
            }
            for run in stored.runs {
                if !task.runs.contains(&run) {
                    task.runs.push(run);
                }
            }
            for p in stored.passes {
                if !task
                    .passes
                    .iter()
                    .any(|q| q.run == p.run && q.pass_index == p.pass_index)
                {
                    task.passes.push(p);
                }
            }
            // A completion notice already recorded is never rewound by an
            // older snapshot: that would report the same state twice.
            if let Some(n) = stored.chat_notice
                && task.chat_notice.as_ref().is_none_or(|mine| mine.at < n.at)
            {
                task.chat_notice = Some(n);
            }
        }
        task.import_chat_notice();
        task.updated_at = Timestamp::now();
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        let body = serde_json::to_string_pretty(task).context("serialize task")?;
        let path = self.path_of(&task.id);
        let tmp = path.with_extension("json.tmp");
        std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
        std::fs::rename(&tmp, &path).with_context(|| format!("replace {}", path.display()))?;
        // A machine hold is news for the notification centre, filed beside
        // this queue (`<home>/queue` -> `<home>/notifications`) rather than
        // through a process-global, so a queue in a temp directory notifies
        // into that directory.
        if let (Some(notice), Some(home)) = (
            crate::notices::task_held(task),
            self.root.parent().filter(|p| !p.as_os_str().is_empty()),
        ) {
            crate::notices::raise_in(home, notice);
        }
        Ok(())
    }

    /// Change the task `id` in place: read it inside its write lock, let `f`
    /// edit it, and write it back only when `f` returns `true`. The check a
    /// caller makes in `f` therefore sees the record as stored now, not a
    /// snapshot taken before another writer got in. Returns whether it wrote.
    pub fn modify(&self, id: &str, f: impl FnOnce(&mut Task) -> bool) -> Result<bool> {
        let id = self.resolve_id(id)?;
        let _lock = self.lock_task(&id)?;
        let mut task = self.get(&id)?;
        if !f(&mut task) {
            return Ok(false);
        }
        self.put_unlocked(&mut task)?;
        Ok(true)
    }

    /// Append `run` to the runs of the task `id` (or an unambiguous prefix),
    /// leaving everything else about it alone — see [`Task::link_run`]. Read
    /// and written back in one breath, because [`Queue::put`] replaces the
    /// whole record. Returns the task as stored.
    pub fn link_run(&self, id: &str, run: &str) -> Result<Task> {
        let id = self.resolve_id(id)?;
        // Read inside the lock, so the record written back is the current one
        // and a daemon's write cannot slip between the read and the write.
        let _lock = self.lock_task(&id)?;
        let mut task = self.get(&id)?;
        if task.link_run(run) {
            self.put_unlocked(&mut task)?;
        }
        Ok(task)
    }

    /// Exclusive right to rewrite task `id`'s record, held until the guard
    /// drops. Every writer ([`Queue::put`], [`Queue::link_run`]) takes it, so
    /// a process linking a run and the daemon saving a transition are
    /// serialized instead of last-writer-wins. An old lock may be broken only
    /// after its writer is confirmed dead (legacy tokens without a PID retain
    /// age-based recovery). Not the claim lock (`<id>.lock`), held for a
    /// whole run.
    pub(crate) fn lock_task(&self, id: &str) -> Result<TaskLock> {
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        let path = self.root.join(format!("{id}.write-lock"));
        let started = std::time::Instant::now();
        let retry = crate::lock_file::Retry::task_write(started + TASK_LOCK_STALE);
        let marker_retry = crate::lock_file::Retry::new(Some(started + TASK_LOCK_STALE));
        loop {
            match retry.create(&path) {
                Ok(mut file) => {
                    use std::io::Write;
                    let token = owner_token();
                    if let Err(e) = file.write_all(token.as_bytes()) {
                        drop(file);
                        let _ = std::fs::remove_file(&path);
                        return Err(e).with_context(|| format!("lock {}", path.display()));
                    }
                    return Ok(TaskLock { path, token });
                }
                Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                    // Content first, then age: a lock replaced in between
                    // reads fresh, never stale.
                    let judged = std::fs::read_to_string(&path).ok();
                    let broken = judged
                        .filter(|_| older_than_stale(&path))
                        .is_some_and(|judged| break_stale_with(&path, &judged, marker_retry));
                    // A break that does not happen (marker busy, recovery
                    // exhausted) is a wait like any other, so it meets the
                    // same deadline instead of spinning forever.
                    if !broken {
                        if started.elapsed() > TASK_LOCK_STALE {
                            bail!("could not lock task {id}");
                        }
                        std::thread::sleep(std::time::Duration::from_millis(15));
                    }
                }
                Err(e) => return Err(e).with_context(|| format!("lock {}", path.display())),
            }
        }
    }

    /// Load a task by id or unambiguous id prefix.
    pub fn get(&self, id: &str) -> Result<Task> {
        let resolved = self.resolve_id(id)?;
        read_path(&self.path_of(&resolved))
    }

    /// Remove a task, and the claim lock that belongs to it.
    ///
    /// `in_flight` comes from the caller — a live daemon's heartbeat naming
    /// this task — because the task's own `running` status cannot answer the
    /// question. A daemon killed mid-competition leaves the status at
    /// `running` and an orphaned `.lock` behind, and a guard that trusted
    /// either would make the task undeletable for good: the phone showed
    /// exactly that, refusing a task whose daemon had been gone for an hour.
    ///
    /// So the lock is removed with the task rather than respected. Any lock
    /// still there once no live daemon claims the task is by definition stale,
    /// and leaving it would make a deleted task look claimed to
    /// [`Queue::claim`] and to whoever reads the directory.
    ///
    /// A deletion is deliberate, so whatever was `blocked` on the id is
    /// resolved, not held: a `<id>.removed` tombstone is written first (it
    /// is what tells [`missing_blockers`] and the daemon's resolver that this
    /// id was deleted on purpose, so a pass that races this call cannot
    /// mistake it for a vanished dependency), the record is deleted (the
    /// tombstone is taken back if that fails), and then each dependent is
    /// rewritten under its claim - see [`Queue::release_dependents_of`].
    /// Best-effort for the dependents: one claimed by something else right
    /// now, or whose write fails, is released silently by
    /// `crate::daemon::resolve_blockers` (or `crate::triage::run_once`) on
    /// its next pass, because the tombstone is still there. It does not fail
    /// this removal.
    ///
    /// `questions` is the store [`missing_blockers`] checks a `blocked_by` id
    /// against - the same store the caller already resolves `id`'s own home
    /// from, passed in rather than reopened here so a test queue at an
    /// explicit root never reads the operator's real questions directory.
    pub fn remove(&self, id: &str, in_flight: bool, questions: &Questions) -> Result<Removal> {
        let resolved = self.resolve_id(id)?;
        if in_flight {
            bail!("task {resolved} is being run by a live daemon right now");
        }
        self.write_tombstone(&resolved)?;
        if let Err(e) = self.remove_record_with_attachments(&resolved, |p| std::fs::remove_file(p))
        {
            // A record that is still there was not deleted: take the marker
            // back. One already gone keeps it, so the deletion still reads
            // as deliberate.
            if self.path_of(&resolved).exists() {
                let _ = std::fs::remove_file(self.tombstone_path(&resolved));
            }
            return Err(e);
        }
        let (released, still_blocked) = self.release_dependents_of(&resolved, questions);
        Ok(Removal {
            id: resolved,
            released,
            still_blocked,
        })
    }

    /// Path of the marker saying task `id` was deleted on purpose. Not a
    /// `*.json`, so [`Queue::list`] and [`Queue::revision`] never see it.
    fn tombstone_path(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.removed"))
    }

    fn write_tombstone(&self, id: &str) -> Result<()> {
        let path = self.tombstone_path(id);
        let tmp = self.root.join(format!("{id}.removed.tmp"));
        std::fs::write(&tmp, b"")
            .and_then(|()| std::fs::rename(&tmp, &path))
            .with_context(|| format!("write {}", path.display()))
    }

    /// Was `id` deleted through [`Queue::remove`] (and is its record gone)?
    /// A live record always wins: a tombstone never makes an existing task
    /// look deleted.
    fn was_deleted(&self, id: &str) -> bool {
        self.tombstone_path(id).is_file() && !self.path_of(id).exists()
    }

    /// Apply [`Task::dependency_deleted`] to `task` for every `blocked_by` id
    /// that [`Queue::remove`] deleted. Returns those ids. The caller holds
    /// the task's claim and saves it.
    pub fn apply_deleted_blockers(&self, task: &mut Task, questions: &Questions) -> Vec<String> {
        let deleted = deleted_blockers(self, questions, &task.blocked_by);
        deleted
            .into_iter()
            .filter(|id| task.dependency_deleted(id))
            .collect()
    }

    /// Record, as a fixed-wording info notice, that `dependent` stopped
    /// waiting on the deleted task `deleted`. Best-effort and beside the
    /// queue, like the hold notice [`Queue::put`] files.
    pub fn note_dependency_deleted(&self, dependent: &Task, deleted: &str) {
        let Some(home) = self.root.parent().filter(|p| !p.as_os_str().is_empty()) else {
            return;
        };
        let ja = crate::lang::is_japanese(&crate::lang::of_repo(&dependent.repo));
        let waiting = dependent.status == TaskStatus::Blocked;
        let message = match (ja, waiting) {
            (true, false) => format!(
                "タスク {} は、待っていた {} が削除されたため待機を解除し、元の状態に戻しました",
                dependent.short(),
                short(deleted)
            ),
            (true, true) => format!(
                "タスク {} は、待っていた {} が削除されたため、残りの依存を待っています",
                dependent.short(),
                short(deleted)
            ),
            (false, false) => format!(
                "Task {} stopped waiting on {} because it was deleted, and returned to its previous state",
                dependent.short(),
                short(deleted)
            ),
            (false, true) => format!(
                "Task {} stopped waiting on {} because it was deleted, and is still waiting on its other dependencies",
                dependent.short(),
                short(deleted)
            ),
        };
        crate::notices::raise_in(
            home,
            crate::notices::Notice::info(
                &format!("unblocked:{}:{}", dependent.id, deleted),
                message,
            )
            .link(crate::notices::Link::Task {
                id: dependent.id.clone(),
            }),
        );
    }

    /// Resolve every `blocked` task naming `dependency` (just deleted) with
    /// [`Task::dependency_deleted`]. Returns the ids released to another
    /// status and those that remain `blocked` on something else. Each
    /// dependent is re-read inside its claim, never written from a stale copy.
    fn release_dependents_of(
        &self,
        dependency: &str,
        questions: &Questions,
    ) -> (Vec<String>, Vec<String>) {
        let (mut released, mut still_blocked) = (Vec::new(), Vec::new());
        for listed in self.list() {
            if listed.status != TaskStatus::Blocked || !listed.blocked_by.iter().any(|b| {
                matches!(crate::queue::dependency(self, questions, b), Dependency::Deleted(id) if id == dependency)
            }) {
                continue;
            }
            let Ok(_claim) = self.claim(&listed.id) else {
                continue;
            };
            let Ok(mut task) = self.get(&listed.id) else {
                continue;
            };
            let references: Vec<String> = task.blocked_by.iter().filter(|id| {
                matches!(crate::queue::dependency(self, questions, id), Dependency::Deleted(full) if full == dependency)
            }).cloned().collect();
            if references.is_empty() {
                continue;
            }
            for reference in references {
                task.dependency_deleted(&reference);
            }
            if self.put(&mut task).is_err() {
                continue;
            }
            self.note_dependency_deleted(&task, dependency);
            if task.status == TaskStatus::Blocked {
                still_blocked.push(task.id.clone());
            } else {
                released.push(task.id.clone());
            }
        }
        (released, still_blocked)
    }

    /// The attachment-and-record half of [`Queue::remove`]. `remove_record` is
    /// how the record file is deleted; production passes `remove_file`, and a
    /// test passes a failing one to prove the attachments come back.
    fn remove_record_with_attachments(
        &self,
        resolved: &str,
        remove_record: impl FnOnce(&Path) -> std::io::Result<()>,
    ) -> Result<()> {
        // Neither step may strand the other. The attachments are set aside by a
        // rename (cheap, and undone if the record cannot be removed), then the
        // record goes, and only then is the set-aside directory deleted for
        // real. A failure before the record is gone loses nothing; a failure
        // after it leaves at most an orphan `.removing` directory, which the
        // next removal sweeps.
        self.sweep_removed_attachments();
        let attachments = self.attachments_dir(resolved);
        let aside = self.root.join(format!("{resolved}.attachments.removing"));
        let moved = match std::fs::rename(&attachments, &aside) {
            Ok(()) => true,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => false,
            Err(e) => {
                return Err(e).with_context(|| format!("remove {}", attachments.display()));
            }
        };
        let path = self.path_of(resolved);
        if let Err(e) = remove_record(&path) {
            if moved {
                let _ = std::fs::rename(&aside, &attachments);
            }
            return Err(e).with_context(|| format!("remove {}", path.display()));
        }
        if moved {
            if let Err(e) = std::fs::remove_dir_all(&aside) {
                tracing::warn!("leftover attachments {}: {e}", aside.display());
            }
        }
        let lock = self.lock_path(resolved);
        if let Err(e) = std::fs::remove_file(&lock) {
            if e.kind() != std::io::ErrorKind::NotFound {
                return Err(e).with_context(|| format!("remove {}", lock.display()));
            }
        }
        Ok(())
    }

    /// Delete `*.attachments.removing` directories a previous [`Queue::remove`]
    /// could not finish deleting. The task record is already gone by then, so
    /// the id no longer resolves and the removal cannot be retried by name;
    /// every later removal sweeps them instead. A directory whose record still
    /// exists belongs to a removal in progress and is left alone. Best-effort.
    fn sweep_removed_attachments(&self) {
        let Ok(entries) = std::fs::read_dir(&self.root) else {
            return;
        };
        for entry in entries.flatten() {
            let name = entry.file_name();
            let name = name.to_string_lossy();
            let Some(id) = name.strip_suffix(".attachments.removing") else {
                continue;
            };
            // A record still on disk means a removal is mid-flight (or about
            // to roll back): the directory is its to delete or restore.
            if !self.path_of(id).exists() {
                if let Err(e) = std::fs::remove_dir_all(entry.path()) {
                    tracing::warn!("leftover attachments {}: {e}", entry.path().display());
                }
            }
        }
        self.sweep_tombstones();
    }

    /// Delete `*.removed` markers that are no longer needed: the record is
    /// gone, no task's `blocked_by` names the id, and the marker is older than
    /// [`TOMBSTONE_GRACE`]. The age is what makes this safe against a
    /// concurrent writer: the task list read here can be stale, so a dependent
    /// that was just blocked on an id (a conductor applying a decision while a
    /// removal is under way) would otherwise lose the marker and be held as a
    /// vanished dependency. A marker whose record still exists belongs to a
    /// removal in progress and is never touched.
    fn sweep_tombstones(&self) {
        let Ok(entries) = std::fs::read_dir(&self.root) else {
            return;
        };
        let tasks = self.list();
        for entry in entries.flatten() {
            let name = entry.file_name();
            let name = name.to_string_lossy();
            let Some(id) = name.strip_suffix(".removed") else {
                continue;
            };
            if self.path_of(id).exists()
                || tasks.iter().any(|t| {
                    t.blocked_by
                        .iter()
                        .any(|b| id.starts_with(b) || id.ends_with(b))
                })
            {
                continue;
            }
            let old = entry
                .metadata()
                .and_then(|m| m.modified())
                .ok()
                .and_then(|m| m.elapsed().ok())
                .is_some_and(|age| age >= TOMBSTONE_GRACE);
            if old {
                let _ = std::fs::remove_file(entry.path());
            }
        }
    }

    /// Path of the claim lock for a task. One definition, so `claim` and
    /// `remove` cannot end up naming different files.
    fn lock_path(&self, id: &str) -> PathBuf {
        self.root.join(format!("{id}.lock"))
    }

    /// Every task on disk, highest priority first and newest first within a
    /// priority. This is what `magi task list` and `GET /api/queue` print, so
    /// a raised priority has to move a task here the moment it is saved, not
    /// only in [`Queue::next_runnable`]'s own ordering - the operator reading
    /// the backlog and the loop about to drain it must agree on what "first"
    /// means. Every existing task defaults to priority 0, so this is a no-op
    /// change from the old newest-first order for a queue nobody has
    /// reprioritised.
    ///
    /// Unreadable files are skipped rather than fatal: one corrupt task must
    /// not take the queue - or the web UI, or an unattended daemon - down
    /// with it.
    pub fn list(&self) -> Vec<Task> {
        let mut tasks: Vec<Task> = std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|x| x == "json"))
            .filter_map(|p| read_path(&p).ok())
            .collect();
        tasks.sort_unstable_by(|a, b| b.priority.cmp(&a.priority).then_with(|| b.id.cmp(&a.id)));
        tasks
    }

    /// Task files [`Queue::list`] skips because they cannot be parsed (a
    /// newer schema, a torn write). A guard that looks a task up by scanning
    /// the list must treat any of these as "the task might be there".
    pub fn unreadable(&self) -> Vec<PathBuf> {
        std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .map(|e| e.path())
            .filter(|p| p.extension().is_some_and(|x| x == "json"))
            .filter(|p| read_path(p).is_err())
            .collect()
    }

    /// Runs that a later attempt at the same task replaced, mapped to the id
    /// of the attempt that replaced them.
    ///
    /// A task keeps its attempts in order, and the deck showed them as two
    /// cards with the same title and no hint which was which: yukimemi asked
    /// why `stalled` and `blocked` appeared twice for one task, and the
    /// answer - "those are two tries, and the second one exists because of a
    /// bug since fixed" - was not on the screen anywhere.
    ///
    /// Read from the queue rather than stored on the run, because the
    /// ordering is the queue's fact: a `RunState` has no idea another attempt
    /// happened after it.
    pub fn superseded(&self) -> HashMap<String, String> {
        let mut by = HashMap::new();
        for task in self.list() {
            for earlier in &task.runs {
                if let Some(later) = task.successor_of(earlier) {
                    by.insert(earlier.clone(), later.clone());
                }
            }
        }
        by
    }

    /// Whether `run` is an earlier attempt a later one replaced, and if so
    /// the id of that later attempt.
    ///
    /// Same walk as [`Queue::superseded`], narrowed to one run: a run detail
    /// page asks about exactly one run at a time, and this keeps that call
    /// site from building (and discarding) the whole map's `HashMap` just to
    /// read one entry out of it.
    pub fn superseded_by(&self, run: &str) -> Option<String> {
        for task in self.list() {
            if task.runs.iter().any(|r| r == run) {
                return task.successor_of(run).cloned();
            }
        }
        None
    }

    /// The task's own most recent attempt, when `run` belongs to that task
    /// but is not already that attempt.
    ///
    /// Distinct from [`Queue::superseded_by`], which names only the very
    /// next attempt: a chain of retries (A superseded by B superseded by C)
    /// leaves an older run pointing at an intermediate one that may itself
    /// be unresolved, and a run's own detail page needs to know where the
    /// task's story currently stands - the chain's current head, C - not an
    /// attempt in the middle of it that a client would otherwise have to
    /// walk to by hand.
    pub fn latest_attempt(&self, run: &str) -> Option<String> {
        for task in self.list() {
            if task.runs.iter().any(|r| r == run) {
                return task.runs.last().filter(|last| **last != run).cloned();
            }
        }
        None
    }

    /// The task a daemon should run next, or `None` when the queue is idle.
    ///
    /// Upgrade resumes first, then highest priority and oldest id first.
    /// See [`dispatch_order`] for the durable ordering rule.
    pub fn next_runnable(&self) -> Option<Task> {
        let mut runnable: Vec<Task> = self
            .list()
            .into_iter()
            .filter(|t| t.status.runnable())
            .collect();
        runnable.sort_unstable_by(dispatch_order);
        runnable.into_iter().next()
    }

    /// Take exclusive ownership of a task.
    ///
    /// The lock is a `create_new` file next to the task, which is atomic on
    /// every platform magi targets. It exists so two daemons - or a daemon and
    /// a human running `magi run` - cannot drive one task into two competing
    /// runs. The returned guard releases on drop, including on panic.
    pub fn claim(&self, id: &str) -> Result<Claim> {
        std::fs::create_dir_all(&self.root)
            .with_context(|| format!("create {}", self.root.display()))?;
        let path = self.lock_path(id);
        let retry = crate::lock_file::Retry::new(None);
        match retry.create(&path) {
            Ok(mut f) => {
                use std::io::Write as _;
                // Best effort: the pid is for the human looking at a stale lock.
                let _ = writeln!(f, "{}", std::process::id());
                Ok(Claim { path })
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                bail!("task {id} is already claimed ({} exists)", path.display())
            }
            Err(e) => Err(e).with_context(|| format!("lock {}", path.display())),
        }
    }

    /// Expand an id prefix to exactly one task id.
    pub fn resolve_id(&self, prefix: &str) -> Result<String> {
        if !prefix.is_empty()
            && prefix != "."
            && prefix != ".."
            && !prefix.contains(['/', '\\'])
            && self.path_of(prefix).is_file()
        {
            return Ok(prefix.to_owned());
        }
        let ids = record_ids(&self.root, "json")?;
        match resolve_reference(prefix, ids.iter().map(String::as_str)) {
            Reference::Found(id) => Ok(id),
            Reference::Missing => bail!("no task matches `{prefix}`"),
            Reference::Ambiguous => bail!("`{prefix}` matches multiple tasks"),
        }
    }

    /// Change detection token for the queue.
    ///
    /// Combines file names and modification times of all task files in the
    /// queue, so adding, modifying, or deleting any task — even an older one —
    /// moves the revision and notifies connected clients via the change stream.
    /// Returns 0 when the queue is completely empty.
    pub fn revision(&self) -> u64 {
        self.revision_excluding(&std::collections::BTreeSet::new())
    }

    /// [`Queue::revision`] ignoring the files of the named task ids, for a
    /// caller that tracks those tasks' content some other way.
    pub fn revision_excluding(&self, skip: &std::collections::BTreeSet<String>) -> u64 {
        use std::hash::{Hash as _, Hasher as _};

        let mut entries: Vec<(String, u64)> = std::fs::read_dir(&self.root)
            .into_iter()
            .flatten()
            .flatten()
            .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
            .filter(|e| {
                let path = e.path();
                !path
                    .file_stem()
                    .is_some_and(|stem| skip.contains(stem.to_string_lossy().as_ref()))
            })
            .filter_map(|e| {
                let name = e.file_name().to_string_lossy().into_owned();
                let mtime = e
                    .metadata()
                    .ok()?
                    .modified()
                    .ok()?
                    .duration_since(std::time::UNIX_EPOCH)
                    .ok()?
                    .as_millis() as u64;
                Some((name, mtime))
            })
            .collect();

        if entries.is_empty() {
            return 0;
        }

        entries.sort_unstable();
        let mut hasher = std::hash::DefaultHasher::new();
        for (name, mtime) in &entries {
            name.hash(&mut hasher);
            mtime.hash(&mut hasher);
        }
        let h = hasher.finish();
        if h == 0 { 1 } else { h }
    }
}

/// How long a deletion marker is kept after its record is gone, even with
/// nothing referencing it. See [`Queue::sweep_tombstones`].
const TOMBSTONE_GRACE: std::time::Duration = std::time::Duration::from_secs(3600);

/// What [`Queue::remove`] did, beyond deleting the named task's own file.
#[derive(Debug, Clone)]
pub struct Removal {
    /// The id actually removed - `id` expanded from a prefix, if it was one.
    pub id: String,
    /// Every task that was `blocked` on [`Removal::id`] and left the block
    /// (back to the status it interrupted) because that was its only
    /// remaining dependency.
    pub released: Vec<String>,
    /// Every task that named [`Removal::id`] and is still `blocked` on
    /// another dependency.
    pub still_blocked: Vec<String>,
}

/// Exclusive ownership of a task, released on drop.
#[derive(Debug)]
pub struct Claim {
    path: PathBuf,
}

impl Drop for Claim {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.path);
    }
}

/// The first line of a task, trimmed to a title. Used when the caller gives a
/// body but no title, which is the normal case for an agent piping a file in.
pub fn title_from(instruction: &str, max: usize) -> String {
    let Some(line) = first_line(instruction) else {
        return "(empty task)".to_owned();
    };
    if line.chars().count() <= max {
        return line.to_owned();
    }
    let head: String = line.chars().take(max.saturating_sub(1)).collect();
    format!("{head}…")
}

/// The first non-blank line, whatever it is. A markdown heading is the
/// task's own summary - agents pipe in `# Rework the config loader` and mean
/// exactly that - so it is preferred over the prose beneath it rather than
/// skipped as decoration. Leading list and heading markers are stripped
/// because they are syntax, not words. `None` when nothing is left.
pub(crate) fn first_line(instruction: &str) -> Option<&str> {
    let line = instruction
        .lines()
        .map(str::trim)
        .find(|l| !l.is_empty())?
        .trim_start_matches(['#', '-', '*', '>', ' '])
        .trim();
    (!line.is_empty()).then_some(line)
}

fn read_path(path: &Path) -> Result<Task> {
    let body = std::fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
    let task: Task =
        serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
    // Greater-than, not not-equal: every field added since schema 1 carries
    // `#[serde(default)]`, so an older task has nothing to say about it and
    // defaulting is exactly as good a reading as a value that build never had
    // a chance to write. Only a schema *ahead* of this build - a meaning it
    // cannot possibly know - is refused rather than guessed at.
    if task.schema > SCHEMA {
        bail!(
            "task {} was written by a different magi (schema {}, this build \
             speaks {SCHEMA})",
            task.id,
            task.schema
        );
    }
    Ok(task)
}

/// Resolution shared by queue lookups and dependency inventories. Exact ids
/// win; otherwise both the CLI prefix and displayed suffix are supported.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum Reference {
    Found(String),
    Missing,
    Ambiguous,
}

pub(crate) fn resolve_reference<'a>(id: &str, ids: impl Iterator<Item = &'a str>) -> Reference {
    let mut hits = Vec::new();
    for candidate in ids {
        if candidate == id {
            return Reference::Found(candidate.to_owned());
        }
        if candidate.starts_with(id) || candidate.ends_with(id) {
            hits.push(candidate);
        }
    }
    match hits.as_slice() {
        [] => Reference::Missing,
        [one] => Reference::Found((*one).to_owned()),
        _ => Reference::Ambiguous,
    }
}

/// Read names, not deserialized records: list() intentionally skips corrupt
/// and future-schema records, which still exist and must keep their waiters.
pub(crate) fn record_ids(root: &Path, extension: &str) -> Result<Vec<String>> {
    let entries = match std::fs::read_dir(root) {
        Ok(entries) => entries,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(e) => return Err(e.into()),
    };
    let mut ids = Vec::new();
    for entry in entries {
        let path = entry?.path();
        if path.extension().is_some_and(|e| e == extension)
            && let Some(id) = path.file_stem().and_then(|s| s.to_str())
        {
            ids.push(id.to_owned());
        }
    }
    Ok(ids)
}

/// Unknown and ambiguous dependencies stay waiting; neither is evidence that
/// the operator deleted a prerequisite or needs to supply its replacement.
#[derive(Debug, PartialEq, Eq)]
pub enum Dependency {
    /// Existing task record, with its canonical id; its contents may be unreadable.
    Task(String),
    /// Existing question record, with its canonical id.
    Question(String),
    /// Explicitly removed task, identified by its tombstone.
    Deleted(String),
    /// No matching record or tombstone after a successful inventory read.
    Missing,
    /// More than one record matches the reference.
    Ambiguous,
    /// The inventory cannot currently be read, so existence is unknown.
    Unknown,
}

/// Resolve against authoritative task, question and deletion record names.
pub fn dependency(queue: &Queue, questions: &Questions, id: &str) -> Dependency {
    let (Ok(tasks), Ok(qs), Ok(removed)) = (
        record_ids(queue.root(), "json"),
        record_ids(questions.root(), "json"),
        record_ids(queue.root(), "removed"),
    ) else {
        return Dependency::Unknown;
    };
    // Include tombstones in ambiguity checks: a reused suffix must never
    // release a waiter on a different, still-live task.
    let mut ids = tasks
        .iter()
        .chain(qs.iter())
        .chain(removed.iter())
        .cloned()
        .collect::<Vec<_>>();
    ids.sort();
    ids.dedup();
    match resolve_reference(id, ids.iter().map(String::as_str)) {
        Reference::Found(id) if tasks.contains(&id) => Dependency::Task(id),
        Reference::Found(id) if qs.contains(&id) => Dependency::Question(id),
        Reference::Found(id) if queue.was_deleted(&id) => Dependency::Deleted(id),
        Reference::Found(_) => Dependency::Unknown,
        Reference::Missing => Dependency::Missing,
        Reference::Ambiguous => Dependency::Ambiguous,
    }
}

/// Canonicalize only proven dependencies, preserving unknown legacy references.
pub fn normalize_blockers(queue: &Queue, questions: &Questions, ids: &mut [String]) {
    for id in ids {
        if let Dependency::Task(full) | Dependency::Question(full) | Dependency::Deleted(full) =
            dependency(queue, questions, id)
        {
            *id = full;
        }
    }
}

/// Live queue facts for decision and deputy prompts. Historical hold reasons
/// and question text are claims, never an inventory of existing prerequisites.
pub fn dependency_facts(queue: &Queue, questions: &Questions) -> String {
    let inventory = crate::blockers::Inventory::load(queue, questions);
    let mut text = String::from(
        "Current authoritative queue (historical missing-task claims may be stale; unreadable or ambiguous references must keep waiting):\n",
    );
    match record_ids(queue.root(), "json") {
        Ok(ids) => {
            for id in ids {
                match queue.get(&id) {
                    Ok(task) => text.push_str(&format!(
                        "{}: {}; blocked_by={:?}; chain={:?}\n",
                        task.id,
                        task.status.as_str(),
                        task.blocked_by,
                        inventory.waits_on(&task)
                    )),
                    Err(_) => text.push_str(&format!("{id}: exists, state unreadable\n")),
                }
            }
        }
        Err(_) => text.push_str("Queue inventory unreadable: existence is unknown.\n"),
    }
    text
}

/// Only a successful directory scan with no matching record proves absence.
pub fn missing_blockers(
    queue: &Queue,
    questions: &Questions,
    blocked_by: &[String],
) -> Vec<String> {
    blocked_by
        .iter()
        .filter(|id| dependency(queue, questions, id) == Dependency::Missing)
        .cloned()
        .collect()
}

/// Return the original references, so legacy short ids are removed by the
/// same Task transition as full ids.
pub fn deleted_blockers(
    queue: &Queue,
    questions: &Questions,
    blocked_by: &[String],
) -> Vec<String> {
    blocked_by
        .iter()
        .filter(|id| matches!(dependency(queue, questions, id), Dependency::Deleted(_)))
        .cloned()
        .collect()
}

/// The `hold_reason` text for a task quarantined because one or more of its
/// `blocked_by` ids no longer exist. Shared by `crate::daemon::resolve_blockers`,
/// `crate::triage::run_once`, and [`Queue::remove`]'s own dependent
/// quarantine, so the three call sites read as the same event to an operator
/// looking at `magi task show` rather than three different wordings for it.
///
/// Names the full original `blocked_by` list, not just `missing` - a task
/// quarantined here can also have named a dependency that was still
/// perfectly valid, and [`Task::hold_machine`] clears `blocked_by` on the way
/// in, so this text is the only place that information survives for an
/// operator deciding whether to release the task outright.
pub fn missing_blocker_hold_reason(blocked_by: &[String], missing: &[String]) -> String {
    missing_blocker_hold_reason_in(blocked_by, missing, "en")
}

/// [`missing_blocker_hold_reason`] in `language` (Japanese, else English).
pub fn missing_blocker_hold_reason_in(
    blocked_by: &[String],
    missing: &[String],
    language: &str,
) -> String {
    if crate::lang::is_japanese(language) {
        format!(
            "{} を待っていましたが、{} はディスク上に存在しません - `magi task triage` を参照",
            blocked_by.join(", "),
            missing.join(", "),
        )
    } else {
        format!(
            "blocked on {} but {} no longer exist(s) on disk - see `magi task triage`",
            blocked_by.join(", "),
            missing.join(", "),
        )
    }
}

/// The last dash-separated part of an id, the form people say aloud.
pub fn short(id: &str) -> &str {
    id.split('-').next_back().unwrap_or(id)
}

fn new_id() -> String {
    let stamp = jiff::Zoned::now().strftime("%Y%m%d-%H%M%S");
    let seed = crate::rng::entropy();
    format!("{stamp}-{:04x}", (seed ^ (seed >> 32)) & 0xffff)
}

#[cfg(test)]
mod tests {
    #[test]
    fn explicit_resumption_is_not_invented_for_legacy_or_automatic_release() {
        let t = Task::new(
            "retry".into(),
            "do it".into(),
            PathBuf::from("."),
            Source::Human,
        );
        let mut value = serde_json::to_value(&t).unwrap();
        for field in ["resumption", "resumption_history", "operator_decided_at"] {
            value.as_object_mut().unwrap().remove(field);
        }
        value["schema"] = serde_json::json!(15);
        let mut legacy: Task = serde_json::from_value(value).unwrap();
        legacy.hold_machine(Some("automatic retry".into()));
        legacy.release();
        assert!(legacy.resumption.is_none());
        assert!(legacy.resumption_history.is_empty());
        assert!(legacy.operator_decided_at.is_none());
        legacy.release_operator("cli", None);
        legacy.start("run".into());
        let reloaded: Task =
            serde_json::from_str(&serde_json::to_string(&legacy).unwrap()).unwrap();
        assert!(reloaded.resume_override.is_none());
        assert_eq!(reloaded.resumption.as_ref().unwrap().generation, 1);
    }

    #[test]
    fn merge_over_lets_the_later_choice_win_and_keeps_the_rest() {
        let mut base = RunOverrides {
            merge: Some("pr".to_owned()),
            candidates: Some(3),
            ..RunOverrides::default()
        };
        base.merge_over(&RunOverrides {
            merge: Some("none".to_owned()),
            seed: Some(7),
            ..RunOverrides::default()
        });
        assert_eq!(base.merge.as_deref(), Some("none"));
        assert_eq!(base.candidates, Some(3));
        assert_eq!(base.seed, Some(7));
    }

    #[test]
    fn an_already_landed_task_is_done_with_its_attempt_refunded() {
        let mut t = task("relanded");
        t.attempts = 1;
        t.status = TaskStatus::Running;
        t.already_landed("already in main as 0e368de");
        assert_eq!(t.status, TaskStatus::Done);
        assert_eq!(t.attempts, 0);
        assert!(t.hold_reason.is_none());
        assert_eq!(t.last_error.as_deref(), Some("already in main as 0e368de"));
    }

    #[test]
    fn missing_blocker_reason_follows_the_language() {
        let b = vec!["a".to_owned()];
        let en = missing_blocker_hold_reason_in(&b, &b, "en");
        assert_eq!(en, missing_blocker_hold_reason(&b, &b));
        assert!(en.starts_with("blocked on a"));
        assert!(missing_blocker_hold_reason_in(&b, &b, "ja").contains("存在しません"));
        assert_eq!(missing_blocker_hold_reason_in(&b, &b, "de"), en);
    }

    use super::*;

    #[test]
    fn triage_applied_survives_release_and_old_records_read_as_empty() {
        let mut t = Task::new(
            "t".to_owned(),
            "i".to_owned(),
            PathBuf::from("r"),
            Source::Human,
        );
        t.mark_triage_applied("q1");
        t.mark_triage_applied("q1");
        t.hold_machine(Some("x".to_owned()));
        t.release();
        assert_eq!(t.triage_applied, ["q1"]);
        assert!(t.triage_applied("q1") && !t.triage_applied("q2"));

        let mut v = serde_json::to_value(&t).unwrap();
        v.as_object_mut().unwrap().remove("triage_applied");
        let old: Task = serde_json::from_value(v).unwrap();
        assert!(old.triage_applied.is_empty());
    }

    #[test]
    fn task_counts_of_empty_is_all_zero() {
        assert_eq!(TaskCounts::of(&[]), TaskCounts::default());
    }

    #[test]
    fn task_counts_of_tallies_every_status() {
        let mut queued = Task::new(
            "q".to_owned(),
            "i".to_owned(),
            PathBuf::from("."),
            Source::Human,
        );
        queued.status = TaskStatus::Queued;
        let mut running = queued.clone();
        running.status = TaskStatus::Running;
        let mut done = queued.clone();
        done.status = TaskStatus::Done;
        let mut failed = queued.clone();
        failed.status = TaskStatus::Failed;
        let mut held = queued.clone();
        held.status = TaskStatus::Held;
        let mut blocked = queued.clone();
        blocked.status = TaskStatus::Blocked;

        let mut parked = queued.clone();
        parked.status = TaskStatus::Parked;

        let counts = TaskCounts::of(&[
            queued,
            running,
            done.clone(),
            done,
            failed,
            held,
            blocked,
            parked,
        ]);
        assert_eq!(
            counts,
            TaskCounts {
                queued: 1,
                running: 1,
                done: 2,
                failed: 1,
                held: 1,
                blocked: 1,
                parked: 1,
            }
        );
    }

    #[test]
    fn task_implementers_persist_normalized_without_losing_history() {
        let (_dir, q) = queue();
        let mut t = task("keep me");
        t.hold_manual(Some("waiting".into()));
        t.runs.push("earlier-run".into());
        t.attachments.push("reference.txt".into());
        let cfg: crate::config::Config =
            toml::from_str("[[agents]]\nid = 'mock'\nkind = 'command'\ncommand = ['true']\n")
                .unwrap();
        q.put(&mut t).unwrap();
        let original = serde_json::to_value(&t).unwrap();
        for n in [2, 1, 2] {
            t.set_implementers(n, &cfg).unwrap();
            q.put(&mut t).unwrap();
            t = q.get(&t.id).unwrap();
            assert_eq!(t.implementers(), Some(usize::from(n)));
            assert_eq!(t.solo, n == 1);
            if n == 1 {
                assert!(t.overrides.is_none());
            }
            let mut actual = serde_json::to_value(&t).unwrap();
            for field in ["solo", "overrides", "updated_at"] {
                actual[field] = original[field].clone();
            }
            assert_eq!(actual, original);
        }
        t.overrides.as_mut().unwrap().judges = Some(2);
        t.set_implementers(1, &cfg).unwrap();
        assert_eq!(t.overrides.as_ref().unwrap().judges, Some(2));
        assert_eq!(t.overrides.as_ref().unwrap().candidates, None);
    }

    #[test]
    fn task_implementers_old_records_keep_inheritance_until_changed() {
        let mut value = serde_json::to_value(task("old task")).unwrap();
        value.as_object_mut().unwrap().remove("solo");
        value.as_object_mut().unwrap().remove("overrides");
        let mut old: Task = serde_json::from_value(value).unwrap();
        assert_eq!(old.implementers(), None);
        old.set_implementers(1, &crate::config::Config::default())
            .unwrap();
        assert!(old.solo);
        assert!(old.overrides.is_none());
    }

    #[test]
    fn task_implementers_reject_invalid_counts_and_states() {
        let cfg = crate::config::Config::default();
        let mut t = task("unchanged");
        for n in [0, 4, 255] {
            assert!(t.set_implementers(n, &cfg).is_err());
        }
        for status in [
            TaskStatus::Running,
            TaskStatus::Done,
            TaskStatus::Failed,
            TaskStatus::Blocked,
            TaskStatus::Parked,
        ] {
            t.status = status;
            let before = serde_json::to_value(&t).unwrap();
            assert!(t.set_implementers(1, &cfg).is_err());
            assert_eq!(serde_json::to_value(&t).unwrap(), before);
        }
    }

    /// A queue of its own, with no process-global state - which is the point of
    /// `Queue::at`, and why these can run in parallel.
    fn queue() -> (tempfile::TempDir, Queue) {
        let dir = tempfile::tempdir().unwrap();
        let q = Queue::at(dir.path().join("queue"));
        (dir, q)
    }

    #[test]
    fn putting_a_machine_held_task_files_a_notification_beside_the_queue() {
        let dir = tempfile::tempdir().unwrap();
        let q = Queue::at(dir.path().join("queue"));
        let mut t = task("held");
        q.put(&mut t).unwrap();
        assert_eq!(
            crate::notices::Notices::at(dir.path().join("notifications"))
                .list()
                .len(),
            0
        );
        t.hold_machine(Some("out of attempts".to_owned()));
        q.put(&mut t).unwrap();
        let listed = crate::notices::Notices::at(dir.path().join("notifications")).list();
        assert_eq!(listed.len(), 1);
        assert!(listed[0].message.contains("out of attempts"));
    }

    fn task(title: &str) -> Task {
        Task::new(
            title.to_owned(),
            format!("do {title}"),
            PathBuf::from("."),
            Source::Human,
        )
    }

    #[test]
    fn attempt_limit_inherits_old_records_and_controls_failure() {
        let mut t = task("budget");
        let mut old = serde_json::to_value(&t).unwrap();
        old.as_object_mut().unwrap().remove("attempt_limit");
        old["schema"] = serde_json::json!(15);
        let legacy: Task = serde_json::from_value(old).unwrap();
        assert_eq!(legacy.attempt_limit, None);
        assert_eq!(legacy.effective_attempt_limit(2), 2);
        t.attempts = 2;
        t.change_attempt_limit(NonZeroUsize::new(3)).unwrap();
        t.fail("retry", 1);
        assert_eq!(t.status, TaskStatus::Failed);
        t.attempts = 3;
        t.fail("stop", 10);
        assert_eq!(t.status, TaskStatus::Held);
        t.change_attempt_limit(None).unwrap();
        assert_eq!(t.status, TaskStatus::Held);
        assert_eq!(t.attempts, 3);
        t.fail("inherit", 4);
        assert_eq!(t.status, TaskStatus::Failed);
        for status in [
            TaskStatus::Running,
            TaskStatus::Done,
            TaskStatus::Failed,
            TaskStatus::Blocked,
            TaskStatus::Parked,
        ] {
            t.status = status;
            assert!(t.change_attempt_limit(NonZeroUsize::new(1)).is_err());
            assert_eq!(t.attempt_limit, None);
        }
    }

    #[test]
    fn earlier_attempts_is_every_recorded_run_and_agrees_with_the_display() {
        let mut t = task("retried");
        assert!(
            t.earlier_attempts().is_empty(),
            "a first attempt takes nothing over"
        );
        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
        assert_eq!(t.earlier_attempts(), ["aaaa", "bbbb"]);
        // What the web note shows and what a takeover acts on are one rule.
        assert_eq!(t.successor_of("aaaa"), Some(&"bbbb".to_owned()));
        assert_eq!(t.successor_of("bbbb"), None);
        assert_eq!(t.successor_of("zzzz"), None);
    }

    #[test]
    fn resumed_runs_are_superseded_only_after_their_last_occurrence() {
        let (_dir, q) = queue();
        let mut t = task("resumed");
        for (runs, successor) in [
            (vec!["aaaa", "aaaa"], None),
            (vec!["aaaa", "bbbb", "aaaa"], None),
            (vec!["aaaa", "aaaa", "bbbb"], Some("bbbb")),
            (vec!["aaaa", "bbbb", "aaaa", "cccc"], Some("cccc")),
        ] {
            t.runs = runs.into_iter().map(str::to_owned).collect();
            q.put(&mut t).unwrap();

            assert_eq!(t.successor_of("aaaa").map(String::as_str), successor);
            assert_eq!(q.superseded_by("aaaa").as_deref(), successor);
            assert_eq!(q.superseded().get("aaaa").map(String::as_str), successor);
        }
    }

    #[test]
    fn superseded_by_names_the_next_attempt_and_none_for_the_last() {
        let (_dir, q) = queue();
        let mut t = task("retried");
        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
        q.put(&mut t).unwrap();

        assert_eq!(q.superseded_by("aaaa"), Some("bbbb".to_owned()));
        assert_eq!(q.superseded_by("bbbb"), Some("cccc".to_owned()));
        assert_eq!(
            q.superseded_by("cccc"),
            None,
            "the latest attempt replaces nothing"
        );
        assert_eq!(
            q.superseded_by("never-heard-of-it"),
            None,
            "a run belonging to no task on this queue is not superseded"
        );

        let mut by = HashMap::new();
        by.insert("aaaa".to_owned(), "bbbb".to_owned());
        by.insert("bbbb".to_owned(), "cccc".to_owned());
        assert_eq!(
            q.superseded(),
            by,
            "the whole-map and single-run forms must agree"
        );
    }

    #[test]
    fn superseded_attempts_is_empty_until_the_task_is_done() {
        let mut t = task("retried");
        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
        t.status = TaskStatus::Failed;
        assert_eq!(
            t.superseded_attempts(true),
            &[] as &[String],
            "a task still retrying has no attempt yet that a later one made moot"
        );

        t.status = TaskStatus::Running;
        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
    }

    #[test]
    fn superseded_attempts_names_every_run_before_the_one_that_succeeded() {
        let mut t = task("retried");
        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
        t.status = TaskStatus::Done;
        assert_eq!(
            t.superseded_attempts(true),
            &["aaaa".to_owned(), "bbbb".to_owned()],
            "cccc is the attempt whose success made the task done, and stays out"
        );
    }

    #[test]
    fn superseded_attempts_is_empty_for_a_done_task_with_only_one_attempt() {
        let mut t = task("first try landed");
        t.runs = vec!["aaaa".to_owned()];
        t.status = TaskStatus::Done;
        assert_eq!(t.superseded_attempts(true), &[] as &[String]);
    }

    #[test]
    fn superseded_attempts_is_empty_when_the_last_run_never_actually_succeeded() {
        // `succeed` is reachable directly - `magi task done`, its web
        // equivalent, and the conductor's `Recovery::Done` - on a task in
        // any status, including one whose last recorded attempt is itself
        // `Blocked`/`Failed`/anything but `Merged`/`Ready`. `runs.last()`
        // alone cannot tell that apart from the loop's own settle path, so
        // the caller's own read of that run's status is what decides this.
        let mut t = task("closed by hand after a manual merge");
        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned()];
        t.status = TaskStatus::Done;
        assert_eq!(
            t.superseded_attempts(false),
            &[] as &[String],
            "nothing here is provably why the task is done, so nothing is superseded"
        );
    }

    #[test]
    fn latest_attempt_names_the_chain_s_current_head_not_just_the_next_one() {
        let (_dir, q) = queue();
        let mut t = task("retried twice");
        t.runs = vec!["aaaa".to_owned(), "bbbb".to_owned(), "cccc".to_owned()];
        q.put(&mut t).unwrap();

        assert_eq!(
            q.latest_attempt("aaaa"),
            Some("cccc".to_owned()),
            "an old attempt points straight at the chain's current head, not the \
             next attempt in the middle of it"
        );
        assert_eq!(q.latest_attempt("bbbb"), Some("cccc".to_owned()));
        assert_eq!(
            q.latest_attempt("cccc"),
            None,
            "the latest attempt is not superseded by anything"
        );
        assert_eq!(
            q.latest_attempt("never-heard-of-it"),
            None,
            "a run belonging to no task on this queue is not superseded"
        );
    }

    #[test]
    fn a_markdown_heading_is_the_title_not_decoration() {
        // A task file's heading is the summary its author already wrote, so it
        // beats the prose underneath. Getting this backwards was visible in the
        // first smoke test: a task titled "# Rework the config loader" listed
        // as "It re-reads the file on every lookup".
        assert_eq!(
            title_from("# Rework the config loader\n\nIt re-reads it.\n", 40),
            "Rework the config loader"
        );
        assert_eq!(title_from("- fix the thing", 40), "fix the thing");
        assert_eq!(title_from("> quoted task", 40), "quoted task");
        // Nothing usable at all still has to produce something printable.
        assert_eq!(title_from("   \n\n", 40), "(empty task)");
        assert_eq!(title_from("###\n", 40), "(empty task)");
    }

    #[test]
    fn a_long_title_is_elided_by_characters_not_bytes() {
        // Byte truncation would split a multi-byte character and panic.
        let long = "課題".repeat(30);
        let title = title_from(&long, 10);
        assert_eq!(title.chars().count(), 10);
        assert!(title.ends_with('…'));
    }

    #[test]
    fn priority_wins_and_ties_break_oldest_first() {
        let (_dir, q) = queue();
        let mut a = task("first");
        let mut b = task("second");
        let mut c = task("urgent");
        // Ids carry a timestamp, so force a known order.
        a.id = "20260101-000001-aaaa".to_owned();
        b.id = "20260101-000002-bbbb".to_owned();
        c.id = "20260101-000003-cccc".to_owned();
        c.priority = 5;
        for t in [&mut a, &mut b, &mut c] {
            q.put(t).unwrap();
        }

        // Priority first...
        assert_eq!(q.next_runnable().unwrap().id, c.id);
        c.hold_machine(None);
        q.put(&mut c).unwrap();
        // ...then oldest, so a burst of new work cannot starve older work.
        assert_eq!(q.next_runnable().unwrap().id, a.id);
        assert_eq!(q.list().len(), 3, "b is still waiting its turn");
    }

    #[test]
    fn a_blocked_task_never_starves_another_runnable_one() {
        let (_dir, q) = queue();
        let mut blocked = task("blocked");
        blocked.block(vec!["something".to_owned()], None);
        q.put(&mut blocked).unwrap();

        let mut runnable = task("free to go");
        q.put(&mut runnable).unwrap();

        let next = q.next_runnable().expect("a runnable task is still offered");
        assert_eq!(next.id, runnable.id);
    }

    #[test]
    fn a_held_task_is_never_offered_to_the_loop() {
        let (_dir, q) = queue();
        let mut t = task("held");
        q.put(&mut t).unwrap();
        assert!(q.next_runnable().is_some());

        t.hold_machine(None);
        q.put(&mut t).unwrap();
        assert!(
            q.next_runnable().is_none(),
            "a held task must wait for a human"
        );

        // A failed task, by contrast, is exactly what the loop should retry.
        t.status = TaskStatus::Failed;
        q.put(&mut t).unwrap();
        assert!(q.next_runnable().is_some());
    }

    #[test]
    fn attempts_are_capped_and_then_the_task_is_held() {
        let mut t = task("doomed");

        t.start("run-1".to_owned());
        t.fail("gate red", 2);
        assert_eq!(t.status, TaskStatus::Failed, "one attempt of two: retry");

        t.start("run-2".to_owned());
        t.fail("gate red", 2);
        assert_eq!(
            t.status,
            TaskStatus::Held,
            "out of attempts: stop spending money on it"
        );
        assert_eq!(t.runs, ["run-1", "run-2"]);
        assert_eq!(t.last_error.as_deref(), Some("gate red"));
        assert_eq!(
            t.hold_reason.as_deref(),
            Some("gate red"),
            "the hold must say why, not leave hold_reason null next to a \
             populated last_error"
        );
    }

    #[test]
    fn handing_off_a_task_records_a_hold_reason_too() {
        let mut t = task("left a pull request");
        t.start("run-1".to_owned());
        t.handed_off("run ended with a pull request open [run run-1]");
        assert_eq!(t.status, TaskStatus::Held);
        assert_eq!(t.hold_source, Some(HoldSource::Machine));
        assert_eq!(
            t.hold_reason.as_deref(),
            Some("run ended with a pull request open [run run-1]")
        );
        assert_eq!(t.hold_reason, t.last_error);
    }

    #[test]
    fn a_quota_stall_is_refunded_so_the_backlog_survives_the_night() {
        let mut t = task("stalled by quota");

        t.start("run-1".to_owned());
        assert_eq!(t.attempts, 1);
        t.stall("judge-1, judge-2 out of quota");
        assert_eq!(
            t.attempts, 0,
            "a closed quota window must not spend the task's retry budget"
        );
        assert_eq!(t.status, TaskStatus::Failed, "the loop should retry it");
        assert_eq!(
            t.last_error.as_deref(),
            Some("judge-1, judge-2 out of quota")
        );

        // A task can therefore stall all night and still get its real attempts
        // once the quota resets - which is the whole point.
        for _ in 0..20 {
            t.start("run-n".to_owned());
            t.stall("still out of quota");
        }
        t.start("run-real".to_owned());
        t.fail("gate red", 2);
        assert_eq!(
            t.status,
            TaskStatus::Failed,
            "the first attempt that was really judged is attempt one"
        );
    }

    #[test]
    fn releasing_a_held_task_gives_it_a_real_second_chance() {
        let mut t = task("retry me");
        t.start("run-1".to_owned());
        t.fail("gate red", 1);
        assert_eq!(t.status, TaskStatus::Held);

        t.release();
        assert_eq!(t.status, TaskStatus::Queued);
        // Without resetting attempts the next failure would re-hold at once,
        // and a release would be a no-op the operator cannot see.
        assert_eq!(t.attempts, 0);
        assert!(t.last_error.is_none());
        assert_eq!(
            t.runs.len(),
            1,
            "history is kept: attempts reset, evidence does not"
        );
    }

    #[test]
    fn a_hold_reason_survives_and_a_release_clears_it() {
        let mut t = task("waiting on something else");
        t.hold_manual(Some(
            "waiting for 20260101-000000-aaaa to land first".to_owned(),
        ));
        assert_eq!(t.status, TaskStatus::Held);
        assert_eq!(
            t.hold_reason.as_deref(),
            Some("waiting for 20260101-000000-aaaa to land first")
        );

        // Holding again with no reason must not erase the one already there.
        t.hold_manual(None);
        assert_eq!(
            t.hold_reason.as_deref(),
            Some("waiting for 20260101-000000-aaaa to land first"),
            "a bare re-hold keeps whatever a human already wrote down"
        );

        // A hold with no reason at all is still an ordinary, allowed hold.
        let mut plain = task("no reason given");
        plain.hold_manual(None);
        assert_eq!(plain.status, TaskStatus::Held);
        assert!(plain.hold_reason.is_none());

        t.release();
        assert_eq!(t.status, TaskStatus::Queued);
        assert!(
            t.hold_reason.is_none(),
            "a stale reason must not greet the next person who holds this task"
        );
    }

    #[test]
    fn closing_a_held_task_as_done_clears_its_hold_reason_too() {
        // `done` can close a held task directly - neither `magi task done`
        // nor `POST /api/queue/{id}/done` requires a release first - so a
        // task held for "waiting on 3ed9" and then closed without ever being
        // released must not still read as waiting on it afterwards.
        let mut t = task("landed by hand while held");
        t.hold_manual(Some("waiting on 3ed9".to_owned()));
        assert_eq!(t.hold_reason.as_deref(), Some("waiting on 3ed9"));

        t.succeed();
        assert_eq!(t.status, TaskStatus::Done);
        assert!(
            t.hold_reason.is_none(),
            "a done task cannot still be waiting on something"
        );
    }

    #[test]
    fn holding_or_closing_a_blocked_task_clears_its_dependency_too() {
        // The web UI's "Hold" and "Mark done" buttons are both reachable on a
        // `blocked` task, not just on `queued`/`held` ones - neither requires
        // a release first. A task moved off `Blocked` that way must not still
        // carry the dependency it was waiting on: a dependency graph built
        // from `blocked_by` would otherwise keep drawing an edge for a task
        // that is not blocked on anything any more.
        let mut held = task("held straight out of blocked");
        held.block(
            vec!["20260101-000000-dead".to_owned()],
            Some("waiting on the migration script".to_owned()),
        );
        assert_eq!(held.status, TaskStatus::Blocked);

        held.hold_manual(None);
        assert_eq!(held.status, TaskStatus::Held);
        assert!(
            held.blocked_by.is_empty(),
            "hold overrides the wait, same as release"
        );
        assert!(held.block_reason.is_none());

        let mut done = task("closed straight out of blocked");
        done.block(
            vec!["20260101-000000-dead".to_owned()],
            Some("waiting on the migration script".to_owned()),
        );
        done.succeed();
        assert_eq!(done.status, TaskStatus::Done);
        assert!(
            done.blocked_by.is_empty(),
            "a done task cannot still be waiting on a dependency"
        );
        assert!(done.block_reason.is_none());
    }

    #[test]
    fn a_blocked_task_is_never_offered_to_the_loop() {
        let mut t = task("blocked");
        assert!(t.status.runnable());
        t.block(
            vec!["dep-id".to_owned()],
            Some("waits on dep-id".to_owned()),
        );
        assert_eq!(t.status, TaskStatus::Blocked);
        assert!(!t.status.runnable());
        assert_eq!(TaskStatus::Blocked.as_str(), "blocked");
    }

    #[test]
    fn unblocking_the_last_dependency_returns_the_task_to_queued() {
        let mut t = task("blocked on two");
        t.block(
            vec!["a".to_owned(), "b".to_owned()],
            Some("waits on a and b".to_owned()),
        );

        t.unblock("a");
        assert_eq!(t.status, TaskStatus::Blocked, "b is still outstanding");
        assert_eq!(t.blocked_by, ["b"]);

        t.unblock("b");
        assert_eq!(t.status, TaskStatus::Queued);
        assert!(t.blocked_by.is_empty());
        assert!(t.block_reason.is_none());
    }

    #[test]
    fn unblocking_an_id_on_a_task_that_is_not_blocked_is_a_no_op() {
        let mut t = task("never blocked");
        t.unblock("whatever");
        assert_eq!(t.status, TaskStatus::Queued);
    }

    #[test]
    fn a_held_task_blocked_on_a_question_returns_to_held_not_queued() {
        // The bug this guards: a task an operator (or `crate::triage`) has
        // deliberately held, once `crate::conduct` blocks it on a follow-up
        // question, must not silently re-enter the competition queue the
        // moment that question is answered - whatever the answer said.
        let mut t = task("held, then asked about");
        t.hold_machine(Some("out of attempts".to_owned()));
        assert_eq!(t.status, TaskStatus::Held);

        t.block(vec!["q1".to_owned()], Some("what now?".to_owned()));
        assert_eq!(t.status, TaskStatus::Blocked);

        t.record_answer("what now?".to_owned(), "leave it held".to_owned());
        t.unblock("q1");
        assert_eq!(t.status, TaskStatus::Held, "must restore, not requeue");
        assert_eq!(t.hold_reason.as_deref(), Some("out of attempts"));
        assert_eq!(t.hold_source, Some(HoldSource::Machine));
        assert!(t.blocked_from.is_none(), "consumed once restored");
    }

    #[test]
    fn a_manually_held_task_blocked_on_a_question_returns_to_held() {
        let mut t = task("manually held, then asked about");
        t.hold_manual(Some("waiting on a dependency".to_owned()));

        t.block(vec!["q1".to_owned()], None);
        t.unblock("q1");

        assert_eq!(t.status, TaskStatus::Held);
        assert_eq!(t.hold_source, Some(HoldSource::Manual));
    }

    #[test]
    fn re_blocking_an_already_blocked_task_keeps_the_original_blocked_from() {
        // A second `Task::block` call - `crate::conduct` adding a question on
        // top of an existing block - must not overwrite `blocked_from` with
        // `Blocked` itself, or the task would restore into itself.
        let mut t = task("held, blocked twice");
        t.hold_machine(None);
        t.block(vec!["q1".to_owned()], Some("first".to_owned()));
        t.block(
            vec!["q1".to_owned(), "q2".to_owned()],
            Some("second".to_owned()),
        );

        t.unblock("q1");
        assert_eq!(t.status, TaskStatus::Blocked, "q2 still outstanding");
        t.unblock("q2");
        assert_eq!(t.status, TaskStatus::Held);
    }

    #[test]
    fn unblocking_a_task_blocked_while_running_lands_on_queued_not_running() {
        // Whatever process was driving the run is gone by the time a
        // conductor's question about it gets answered - there is nothing left
        // to resume into.
        let mut t = task("blocked mid-run");
        t.start("run-1".to_owned());
        assert_eq!(t.status, TaskStatus::Running);

        t.block(vec!["q1".to_owned()], None);
        t.unblock("q1");
        assert_eq!(t.status, TaskStatus::Queued);
    }

    #[test]
    fn a_pre_schema_4_blocked_record_with_hold_evidence_restores_to_held() {
        // `blocked_from` is `None` for a record written before schema 4 (or,
        // equivalently, deserialized straight from an on-disk file that never
        // had the field). Held evidence surviving on the task - never cleared
        // by `block` - is the only way left to tell such a record apart from
        // one blocked straight out of `Queued`.
        let mut t = task("legacy record, held before it was blocked");
        t.hold_source = Some(HoldSource::Machine);
        t.hold_reason = Some("legacy hold reason".to_owned());
        t.status = TaskStatus::Blocked;
        t.blocked_by = vec!["q1".to_owned()];
        t.blocked_from = None;

        t.unblock("q1");
        assert_eq!(t.status, TaskStatus::Held);
    }

    #[test]
    fn a_pre_schema_4_blocked_record_with_no_hold_evidence_restores_to_queued() {
        let mut t = task("legacy record, ordinary dependency block");
        t.status = TaskStatus::Blocked;
        t.blocked_by = vec!["dep".to_owned()];
        t.blocked_from = None;

        t.unblock("dep");
        assert_eq!(t.status, TaskStatus::Queued);
    }

    #[test]
    fn answering_a_question_is_recorded_and_survives_a_release() {
        let mut t = task("asked something");
        t.block(vec!["q1".to_owned()], Some("which backend?".to_owned()));
        t.record_answer("Which backend?".to_owned(), "SQLite".to_owned());
        t.unblock("q1");
        assert_eq!(t.status, TaskStatus::Queued);
        assert_eq!(t.answers.len(), 1);
        assert_eq!(t.answers[0].answer, "SQLite");

        // A release resets attempts, not evidence - the same rule
        // `releasing_a_held_task_gives_it_a_real_second_chance` asserts for
        // `runs`.
        t.release();
        assert_eq!(t.answers.len(), 1, "the answer is not lost on release");
    }

    #[test]
    fn resumption_review_branch_requires_unchanged_authored_criteria() {
        for source in ["cli", "web"] {
            let mut t = task("review old candidate");
            t.request_review("candidate-A".into());
            t.hold_for_handover(Some("candidate-A".into()), "handover refused".into());
            let mut unchanged = t.clone();
            unchanged.release_operator(source, None);
            assert_eq!(unchanged.review_branch.as_deref(), Some("candidate-A"));
            t.edit(
                "review approved head".into(),
                "Reuse measured PR head B; fix Clippy; review".into(),
            )
            .unwrap();
            t.release_operator(source, None);
            assert!(t.review_branch.is_none());
            assert!(t.review_instruction.is_none());
            // Legacy records capture their old criteria at the supported edit.
            let mut legacy = unchanged;
            legacy.review_instruction = None;
            legacy.hold_for_handover(None, "still checked out".into());
            legacy
                .edit("new plan".into(), "Reuse measured PR head B".into())
                .unwrap();
            legacy.release_operator(source, None);
            assert!(legacy.review_branch.is_none());
        }
    }

    #[test]
    fn a_refused_handover_keeps_the_review_branch_across_release() {
        let mut t = task("refused takeover");
        t.start("run-1".to_owned());
        t.hold_for_handover(Some("magi/eba2/A".to_owned()), "checked out".to_owned());
        assert_eq!(t.status, TaskStatus::Held);
        assert_eq!(t.attempts, 1);
        t.release();
        assert_eq!(t.status, TaskStatus::Queued);
        assert_eq!(t.attempts, 0);
        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));

        // A fresh competition chosen afterwards is not a review.
        t.hold_for_handover(None, "again".to_owned());
        t.requeue();
        assert!(t.review_branch.is_none());

        // A manual hold never keeps one.
        let mut m = task("manual");
        m.review_branch = Some("magi/x/A".to_owned());
        m.hold_manual(None);
        m.release();
        assert!(m.review_branch.is_none());
    }

    #[test]
    fn requesting_review_requeues_the_task_and_remembers_the_branch() {
        let mut t = task("blocked run with a surviving branch");
        t.start("run-1".to_owned());
        t.fail("blocked with major findings", 5);
        assert_eq!(t.status, TaskStatus::Failed);

        t.request_review("magi/eba2/A".to_owned());
        assert_eq!(t.status, TaskStatus::Queued);
        assert_eq!(t.attempts, 0);
        assert_eq!(t.review_branch.as_deref(), Some("magi/eba2/A"));

        // An ordinary release (a human overriding the choice) drops it again.
        t.release();
        assert!(t.review_branch.is_none());
    }

    #[test]
    fn conductor_requeue_but_not_an_ordinary_release_forces_a_fresh_start() {
        let mut t = task("retry");
        t.start("run-1".to_owned());
        t.requeue();
        assert!(t.fresh_start);

        t.release();
        assert!(!t.fresh_start);
    }

    #[test]
    fn urgent_only_changes_queued_or_held_and_preserves_settings() {
        for status in [
            TaskStatus::Queued,
            TaskStatus::Held,
            TaskStatus::Running,
            TaskStatus::Done,
            TaskStatus::Failed,
            TaskStatus::Blocked,
            TaskStatus::Parked,
        ] {
            let mut t = task("urgent settings");
            t.status = status;
            t.attempts = 2;
            t.hold_reason = Some("waiting".into());
            t.priority = 7;
            t.interrupt = true;
            let before = serde_json::to_value(&t).unwrap();
            for urgent in [true, true, false, false] {
                let result = t.set_urgent(urgent);
                let mut expected = before.clone();
                if matches!(status, TaskStatus::Queued | TaskStatus::Held) {
                    result.unwrap();
                    expected["urgent"] = urgent.into();
                } else {
                    let error = result.unwrap_err().to_string();
                    assert!(error.contains(status.as_str()), "{error}");
                    assert!(error.contains("queued or held"), "{error}");
                }
                assert_eq!(serde_json::to_value(&t).unwrap(), expected);
            }
        }
    }

    #[test]
    fn priority_can_be_changed_while_queued_but_not_while_running() {
        let mut t = task("reprioritise me");
        t.set_priority(5).unwrap();
        assert_eq!(t.priority, 5);

        t.start("run-1".to_owned());
        let err = t.set_priority(9).unwrap_err().to_string();
        assert!(err.contains("running"), "{err}");
        assert_eq!(t.priority, 5, "the rejected write must not partially apply");
    }

    #[test]
    fn interrupt_can_be_marked_while_queued_but_not_while_running() {
        let mut t = task("interrupt me");
        assert!(!t.interrupt, "off unless asked, same as any other task");

        t.set_interrupt(true).unwrap();
        assert!(t.interrupt);

        t.start("run-1".to_owned());
        assert!(
            !t.interrupt,
            "the mark is one-shot: dispatching the task fulfils it, \
             whatever the run that follows ends up doing"
        );
        let err = t.set_interrupt(true).unwrap_err().to_string();
        assert!(err.contains("running"), "{err}");
        // Clearing is always allowed, even on a running task - there is
        // nothing left for it to interrupt once it has been claimed.
        t.set_interrupt(false).unwrap();
        assert!(!t.interrupt);
    }

    /// R2-1-1: a task whose run fails and requeues must not go on
    /// re-triggering `crate::daemon`'s interrupt scheduler on every later
    /// boundary, attempt after attempt, until it exhausts its budget.
    #[test]
    fn a_failed_run_does_not_leave_the_task_still_marked_to_interrupt() {
        let mut t = task("interrupt me");
        t.set_interrupt(true).unwrap();
        t.start("run-1".to_owned());
        t.fail("mock failure", 5);
        assert_eq!(t.status, TaskStatus::Failed);
        assert!(
            !t.interrupt,
            "one attempt already spent the mark; a retry is an ordinary \
             requeue, not a fresh interrupt request"
        );
    }

    #[test]
    fn changing_priority_moves_a_task_ahead_in_the_real_queue_order() {
        let (_dir, q) = queue();
        let mut a = task("first filed");
        let mut b = task("second filed");
        a.id = "20260101-000001-aaaa".to_owned();
        b.id = "20260101-000002-bbbb".to_owned();
        q.put(&mut a).unwrap();
        q.put(&mut b).unwrap();

        assert_eq!(
            q.next_runnable().unwrap().id,
            a.id,
            "with equal priority the older task goes first, so a burst of \
             new work cannot starve it"
        );
        assert_eq!(
            q.list()[0].id,
            b.id,
            "but the list an operator reads is newest first, the same as \
             before priority existed - a's turn to run does not make it the \
             newest task"
        );

        let mut a = q.get(&a.id).unwrap();
        a.set_priority(10).unwrap();
        q.put(&mut a).unwrap();

        assert_eq!(
            q.next_runnable().unwrap().id,
            a.id,
            "a raised priority must be reflected the moment it is saved"
        );
        // `magi task list` and `GET /api/queue` both print `Queue::list()`
        // directly, so the raised task has to lead there too - not only in
        // what the loop would claim next.
        assert_eq!(
            q.list()[0].id,
            a.id,
            "the raised task must sort first in the list an operator reads, \
             not only in next_runnable's own ordering"
        );
    }

    #[test]
    fn editing_replaces_title_and_instruction_but_keeps_identity_and_history() {
        let mut t = Task::new(
            "old title".to_owned(),
            "old instruction".to_owned(),
            PathBuf::from("/repo"),
            Source::Agent {
                run: "20260101-000000-beef".to_owned(),
                node: "implement".to_owned(),
            },
        );
        let id = t.id.clone();
        let created_at = t.created_at;
        t.runs.push("20260101-000000-beef".to_owned());

        t.edit("new title".to_owned(), "new instruction".to_owned())
            .unwrap();

        assert_eq!(t.title, "new title");
        assert_eq!(t.instruction, "new instruction");
        assert_eq!(t.id, id, "editing must not mint a new id");
        assert_eq!(t.created_at, created_at);
        assert_eq!(
            t.source,
            Source::Agent {
                run: "20260101-000000-beef".to_owned(),
                node: "implement".to_owned(),
            },
            "editing must not turn agent attribution into human"
        );
        assert_eq!(t.runs, ["20260101-000000-beef"]);
    }

    #[test]
    fn editing_is_refused_once_a_task_is_running_or_finished() {
        let mut running = task("in flight");
        running.start("run-1".to_owned());
        let err = running
            .edit("x".to_owned(), "y".to_owned())
            .unwrap_err()
            .to_string();
        assert!(err.contains("running"), "{err}");

        let mut done = task("finished");
        done.succeed();
        let err = done
            .edit("x".to_owned(), "y".to_owned())
            .unwrap_err()
            .to_string();
        assert!(err.contains("done"), "{err}");

        // Both queued and held are the point of the feature and must work.
        let mut queued = task("waiting");
        queued.edit("x".to_owned(), "y".to_owned()).unwrap();
        let mut held = task("parked");
        held.hold_machine(None);
        held.edit("x".to_owned(), "y".to_owned()).unwrap();
    }

    #[test]
    fn a_task_recorded_without_a_hold_reason_still_reads_as_none() {
        let (_dir, q) = queue();
        let path = q.path_of("20260101-000000-aaaa");
        std::fs::create_dir_all(q.root()).unwrap();
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": SCHEMA,
                "id": "20260101-000000-aaaa",
                "title": "from before hold reasons existed",
                "instruction": "from before hold reasons existed",
                "repo": ".",
                "source": { "kind": "human" },
                "status": "held",
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
            })
            .to_string(),
        )
        .unwrap();

        let task = q.get("20260101-000000-aaaa").expect("must still read");
        assert!(task.hold_reason.is_none());
        assert!(task.operator_held());
    }

    #[test]
    fn a_legacy_reasoned_hold_defaults_to_operator_protection() {
        let (_dir, q) = queue();
        let path = q.path_of("20260101-000000-bbbb");
        std::fs::create_dir_all(q.root()).unwrap();
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": 2,
                "id": "20260101-000000-bbbb",
                "title": "old manual recovery",
                "instruction": "old manual recovery",
                "repo": ".",
                "source": { "kind": "human" },
                "status": "held",
                "hold_reason": "active manual recovery run20260912-224242-daf5",
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
            })
            .to_string(),
        )
        .unwrap();

        let task = q.get("20260101-000000-bbbb").expect("must still read");
        assert_eq!(task.hold_source, None);
        assert!(task.operator_held());
    }

    #[test]
    fn a_task_recorded_without_a_diagnostic_still_reads_as_none() {
        let (_dir, q) = queue();
        let path = q.path_of("20260101-000000-aaaa");
        std::fs::create_dir_all(q.root()).unwrap();
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": SCHEMA,
                "id": "20260101-000000-aaaa",
                "title": "from before diagnostics existed",
                "instruction": "from before diagnostics existed",
                "repo": ".",
                "source": { "kind": "human" },
                "status": "held",
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
            })
            .to_string(),
        )
        .unwrap();

        let task = q.get("20260101-000000-aaaa").expect("must still read");
        assert!(task.diagnostic.is_none());
    }

    #[test]
    fn a_schema_1_task_with_no_blocking_fields_still_reads() {
        // Written by a build that predates `blocked_by`, `block_reason`,
        // `answers` and `review_branch` entirely - literal `"schema": 1`,
        // not `SCHEMA`, since the whole point is a build older than this one.
        let (_dir, q) = queue();
        let path = q.path_of("20260101-000000-aaaa");
        std::fs::create_dir_all(q.root()).unwrap();
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": 1,
                "id": "20260101-000000-aaaa",
                "title": "from before blocking existed",
                "instruction": "from before blocking existed",
                "repo": ".",
                "source": { "kind": "human" },
                "status": "queued",
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
            })
            .to_string(),
        )
        .unwrap();

        let task = q.get("20260101-000000-aaaa").expect("must still read");
        assert!(task.blocked_by.is_empty());
        assert!(task.block_reason.is_none());
        assert!(task.answers.is_empty());
        assert!(task.review_branch.is_none());
    }

    #[test]
    fn releasing_or_finishing_a_task_clears_its_stale_diagnostic() {
        // A diagnostic belongs to the run that produced it. Left in place
        // across a release, an unrelated later failure - a config error, say -
        // would go on showing evidence for a problem that is no longer why the
        // task is stuck.
        let mut held = task("diagnosed");
        held.start("run-1".to_owned());
        held.fail("gate red", 1);
        held.diagnostic = Some("cargo test failed: ...".to_owned());
        assert_eq!(held.status, TaskStatus::Held);

        held.release();
        assert!(held.diagnostic.is_none());

        held.diagnostic = Some("cargo test failed: ...".to_owned());
        held.succeed();
        assert!(held.diagnostic.is_none());
    }

    #[test]
    fn failing_a_task_always_clears_whatever_diagnostic_it_carried() {
        let mut t = task("retried");
        t.start("run-1".to_owned());
        t.diagnostic = Some("stale evidence from a previous hold".to_owned());
        t.fail("unrelated config error", 5);
        assert_eq!(t.status, TaskStatus::Failed);
        assert!(
            t.diagnostic.is_none(),
            "fail() must not let an old diagnostic outlive the run that produced it"
        );
    }

    #[test]
    fn a_claim_is_exclusive_and_releases_on_drop() {
        let (_dir, q) = queue();
        let mut t = task("contended");
        q.put(&mut t).unwrap();

        let held = q.claim(&t.id).unwrap();
        assert!(
            q.claim(&t.id).is_err(),
            "two daemons must not drive one task into two runs"
        );
        drop(held);
        assert!(q.claim(&t.id).is_ok(), "a released claim is reclaimable");
    }

    #[test]
    fn a_round_trip_survives_disk() {
        let (_dir, q) = queue();
        let mut t = Task::new(
            "titled".to_owned(),
            "body".to_owned(),
            PathBuf::from("/repo"),
            Source::Agent {
                run: "20260101-000000-beef".to_owned(),
                node: "implement".to_owned(),
            },
        );
        t.priority = 3;
        q.put(&mut t).unwrap();

        let back = q.get(&t.id).unwrap();
        assert_eq!(back.id, t.id);
        assert_eq!(back.priority, 3);
        assert_eq!(back.source.label(), "implement@beef");
        // A prefix is enough, the way run ids work everywhere else.
        assert_eq!(q.get(t.short()).unwrap().id, t.id);
    }

    #[test]
    fn an_unreadable_task_does_not_take_the_queue_down() {
        let (_dir, q) = queue();
        let mut t = task("fine");
        q.put(&mut t).unwrap();
        std::fs::write(q.root().join("broken.json"), "{ not json").unwrap();

        let listed = q.list();
        assert_eq!(listed.len(), 1, "the readable task still lists");
        assert_eq!(listed[0].id, t.id);
    }

    #[test]
    fn a_task_recorded_without_a_solo_field_still_reads_as_not_solo() {
        let (_dir, q) = queue();
        let path = q.path_of("20260101-000000-aaaa");
        std::fs::create_dir_all(q.root()).unwrap();
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": SCHEMA,
                "id": "20260101-000000-aaaa",
                "title": "from before solo existed",
                "instruction": "from before solo existed",
                "repo": ".",
                "source": { "kind": "human" },
                "status": "queued",
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
            })
            .to_string(),
        )
        .unwrap();

        let task = q.get("20260101-000000-aaaa").expect("must still read");
        assert!(!task.solo, "a queue file with no `solo` field means false");
    }

    #[test]
    fn a_task_recorded_without_an_urgent_field_still_reads_as_not_urgent() {
        let (_dir, q) = queue();
        let path = q.path_of("20260101-000000-bbbb");
        std::fs::create_dir_all(q.root()).unwrap();
        std::fs::write(
            &path,
            serde_json::json!({
                "schema": SCHEMA,
                "id": "20260101-000000-bbbb",
                "title": "from before urgent existed",
                "instruction": "from before urgent existed",
                "repo": ".",
                "source": { "kind": "human" },
                "status": "queued",
                "created_at": Timestamp::now().to_string(),
                "updated_at": Timestamp::now().to_string(),
            })
            .to_string(),
        )
        .unwrap();

        let task = q.get("20260101-000000-bbbb").expect("must still read");
        assert!(
            !task.urgent,
            "a queue file with no `urgent` field means false, same as `solo`"
        );
    }

    #[test]
    fn a_task_from_a_future_schema_is_refused_rather_than_guessed_at() {
        let (_dir, q) = queue();
        let mut t = task("from the future");
        q.put(&mut t).unwrap();
        let path = q.path_of(&t.id);
        let body = std::fs::read_to_string(&path)
            .unwrap()
            .replace(&format!("\"schema\": {SCHEMA}"), "\"schema\": 99");
        std::fs::write(&path, body).unwrap();

        let err = q.get(&t.id).unwrap_err().to_string();
        assert!(err.contains("schema 99"), "{err}");
    }

    #[test]
    fn revision_moves_when_the_queue_changes() {
        let (_dir, q) = queue();
        assert_eq!(q.revision(), 0, "an empty queue has no revision");
        let mut t = task("first");
        q.put(&mut t).unwrap();
        assert!(q.revision() > 0, "a written task moves the revision");
    }

    #[test]
    fn revision_moves_when_deleting_an_older_task() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut t1 = task("older");
        q.put(&mut t1).unwrap();
        // Ensure mtime ticks forward.
        std::thread::sleep(std::time::Duration::from_millis(10));
        let mut t2 = task("newer");
        q.put(&mut t2).unwrap();

        let rev_before = q.revision();
        q.remove(&t1.id, false, &questions).unwrap();
        let rev_after = q.revision();

        assert_ne!(
            rev_before, rev_after,
            "deleting an older task must change the revision so other clients see the deletion"
        );
    }

    #[test]
    fn removing_a_task_takes_it_out_of_the_listing() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("delete me");
        q.put(&mut t).unwrap();
        let removed = q.remove(t.short(), false, &questions).unwrap();
        assert_eq!(removed.id, t.id, "a prefix resolves before deleting");
        assert!(removed.released.is_empty() && removed.still_blocked.is_empty());
        assert!(q.list().is_empty());
        assert!(
            q.remove(&t.id, false, &questions).is_err(),
            "removing twice is an error"
        );
    }

    #[test]
    fn removing_a_task_takes_its_stale_lock_with_it() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("interrupted");
        q.put(&mut t).unwrap();

        // A daemon killed mid-run leaves this behind. Nothing holds it: the
        // process that would have dropped the guard is gone.
        let claim = q.claim(&t.id).unwrap();
        std::mem::forget(claim);
        assert!(
            q.claim(&t.id).is_err(),
            "the orphaned lock is what makes the task look claimed"
        );

        // A live daemon on this task is refused, whatever the lock says.
        let err = q.remove(&t.id, true, &questions).unwrap_err().to_string();
        assert!(err.contains("live daemon"), "{err}");
        assert!(q.get(&t.id).is_ok(), "a refused delete keeps the task");

        // With no daemon behind it, the lock is stale and goes with the task.
        q.remove(&t.id, false, &questions).unwrap();
        assert!(q.list().is_empty());
        let mut again = task("interrupted");
        again.id = t.id.clone();
        q.put(&mut again).unwrap();
        assert!(
            q.claim(&t.id).is_ok(),
            "a task that comes back must be claimable, which a left-behind lock would prevent"
        );
    }

    fn notices_of(dir: &Path) -> Vec<crate::notices::Notice> {
        crate::notices::Notices::at(dir.join("notifications")).list()
    }

    #[test]
    fn removing_a_sole_dependency_releases_the_dependent_without_a_hold() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut dep = task("dependency");
        q.put(&mut dep).unwrap();
        let mut blocked = task("waiting");
        blocked.block(vec![dep.id.clone()], Some("waits".to_owned()));
        q.put(&mut blocked).unwrap();

        let removed = q.remove(&dep.id, false, &questions).unwrap();
        assert_eq!(removed.released, [blocked.id.clone()]);
        assert!(removed.still_blocked.is_empty());

        let after = q.get(&blocked.id).unwrap();
        assert_eq!(after.status, TaskStatus::Queued);
        assert!(after.blocked_by.is_empty());
        assert!(after.block_reason.is_none());
        let notes = notices_of(dir.path());
        assert_eq!(notes.len(), 1, "{notes:?}");
        assert_eq!(notes[0].severity, crate::notices::Severity::Info);
        assert_eq!(
            notes[0].link,
            Some(crate::notices::Link::Task {
                id: blocked.id.clone(),
            })
        );
    }

    #[test]
    fn removing_one_of_two_dependencies_keeps_the_other() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut dep = task("dependency");
        q.put(&mut dep).unwrap();
        let mut other = task("other");
        q.put(&mut other).unwrap();
        let mut blocked = task("waiting");
        blocked.block(
            vec![dep.id.clone(), other.id.clone()],
            Some("waits on both".to_owned()),
        );
        q.put(&mut blocked).unwrap();

        let removed = q.remove(&dep.id, false, &questions).unwrap();
        assert!(removed.released.is_empty());
        assert_eq!(removed.still_blocked, [blocked.id.clone()]);

        let after = q.get(&blocked.id).unwrap();
        assert_eq!(after.status, TaskStatus::Blocked);
        assert_eq!(after.blocked_by, [other.id.clone()]);
        // The tombstone stays while a task still names the id... only this
        // one's dependency is gone, so nothing references it and it may go.
        assert!(missing_blockers(&q, &questions, &after.blocked_by).is_empty());
    }

    #[test]
    fn a_dependency_deleted_before_its_dependents_were_rewritten_is_released_later() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut dep = task("dependency");
        q.put(&mut dep).unwrap();
        let mut blocked = task("waiting");
        blocked.block(vec![dep.id.clone()], None);
        q.put(&mut blocked).unwrap();

        // The dependent is claimed elsewhere, so `remove` has to skip it.
        let claim = q.claim(&blocked.id).unwrap();
        let removed = q.remove(&dep.id, false, &questions).unwrap();
        assert!(removed.released.is_empty());
        drop(claim);

        let mut task = q.get(&blocked.id).unwrap();
        assert_eq!(task.status, TaskStatus::Blocked);
        assert!(missing_blockers(&q, &questions, &task.blocked_by).is_empty());
        assert_eq!(
            q.apply_deleted_blockers(&mut task, &questions),
            [dep.id.clone()]
        );
        assert_eq!(task.status, TaskStatus::Queued);
    }

    #[test]
    fn dependency_resolution_short_deleted_reference_releases_only_its_waiters() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut dep = task("dependency");
        dep.id = "20261013-193611-ca5b".into();
        q.put(&mut dep).unwrap();
        let mut waiting = task("waiting");
        waiting.block(vec!["ca5b".into()], None);
        q.put(&mut waiting).unwrap();
        q.remove(&dep.id, false, &questions).unwrap();
        assert_eq!(q.get(&waiting.id).unwrap().status, TaskStatus::Queued);
        assert_eq!(
            dependency(&q, &questions, "ca5b"),
            Dependency::Deleted(dep.id)
        );
    }

    #[test]
    fn dependency_resolution_preserves_cross_store_ambiguity_and_exact_ids() {
        let (dir, queue) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut dep = task("dependency");
        dep.id = "20261013-193611-ca5b".into();
        queue.put(&mut dep).unwrap();
        std::fs::create_dir_all(questions.root()).unwrap();
        let question_id = "20261014-193611-ca5b";
        std::fs::write(questions.path_of(question_id), "unreadable question").unwrap();
        assert_eq!(questions.resolve_id("ca5b").unwrap(), question_id);
        assert_eq!(
            dependency(&queue, &questions, "ca5b"),
            Dependency::Ambiguous
        );
        assert_eq!(
            dependency(&queue, &questions, &dep.id),
            Dependency::Task(dep.id)
        );
        assert_eq!(
            dependency(&queue, &questions, question_id),
            Dependency::Question(question_id.into())
        );
        assert_eq!(
            dependency(&queue, &questions, "missing"),
            Dependency::Missing
        );
    }

    #[test]
    fn dependency_resolution_directory_errors_are_unknown_not_missing() {
        let dir = tempfile::tempdir().unwrap();
        let q = Queue::at(dir.path().join("not-a-directory"));
        std::fs::write(q.root(), "file").unwrap();
        let questions = Questions::at(dir.path().join("questions"));
        assert_eq!(dependency(&q, &questions, "ca5b"), Dependency::Unknown);
        assert!(missing_blockers(&q, &questions, &["ca5b".into()]).is_empty());
    }

    #[test]
    fn a_dependency_without_a_tombstone_is_still_missing() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut dep = task("dependency");
        q.put(&mut dep).unwrap();
        std::fs::remove_file(q.path_of(&dep.id)).unwrap();
        assert_eq!(
            missing_blockers(&q, &questions, std::slice::from_ref(&dep.id)),
            [dep.id.clone()]
        );
        assert!(deleted_blockers(&q, &questions, std::slice::from_ref(&dep.id)).is_empty());
    }

    #[test]
    fn a_held_dependent_is_left_alone_by_a_dependency_deletion() {
        let mut t = task("held");
        t.hold_machine(Some("because".to_owned()));
        t.blocked_by = vec!["gone".to_owned()];
        assert!(!t.dependency_deleted("gone"));
        assert_eq!(t.status, TaskStatus::Held);
    }

    #[test]
    fn a_failed_record_removal_takes_the_tombstone_back() {
        let (_dir, q) = queue();
        let mut t = task("stays");
        q.put(&mut t).unwrap();
        q.write_tombstone(&t.id).unwrap();
        let err = q.remove_record_with_attachments(&t.id, |_| Err(std::io::Error::other("nope")));
        assert!(err.is_err());
        // `remove` is what withdraws the marker; a live record is never
        // treated as deleted whatever is on disk.
        assert!(!q.was_deleted(&t.id));
    }

    fn source_file(dir: &Path, name: &str, body: &str) -> PathBuf {
        let p = dir.join(name);
        std::fs::write(&p, body).unwrap();
        p
    }

    #[test]
    fn an_attachment_copy_survives_deleting_its_source() {
        let (dir, q) = queue();
        let src = source_file(dir.path(), "shot.png", "pixels");
        let mut t = task("with a picture");
        let names = q.attach(&mut t, std::slice::from_ref(&src)).unwrap();
        q.put(&mut t).unwrap();
        std::fs::remove_file(&src).unwrap();
        assert_eq!(names, ["shot.png"]);
        let loaded = q.get(&t.id).unwrap();
        let paths = q.attachment_paths(&loaded);
        assert_eq!(paths.len(), 1);
        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "pixels");
    }

    #[test]
    fn attachment_names_that_could_traverse_or_are_odd_are_refused() {
        let (dir, q) = queue();
        let mut t = task("bad names");
        for name in ["a..b.png", ".hidden", "with space.png", "-x.png"] {
            let src = source_file(dir.path(), name, "x");
            assert!(
                q.attach(&mut t, &[src]).is_err(),
                "`{name}` must be refused"
            );
        }
        // A drive-qualified name cannot exist as a file on Windows (joining it
        // to a directory yields a drive-relative path to `foo.png` instead),
        // so the rule is asserted on the name itself, everywhere.
        assert!(!crate::ask::valid_asset_name("C:foo.png"));
        #[cfg(not(windows))]
        {
            let src = source_file(dir.path(), "C:foo.png", "x");
            assert!(q.attach(&mut t, &[src]).is_err());
        }
        let long = format!("{}.png", "a".repeat(70));
        let src = source_file(dir.path(), &long, "x");
        assert!(q.attach(&mut t, &[src]).is_err());
        assert!(t.attachments.is_empty());
        assert!(!q.attachments_dir(&t.id).exists());
    }

    #[test]
    fn a_taken_attachment_name_is_numbered_not_overwritten() {
        let (dir, q) = queue();
        let a = source_file(dir.path(), "shot.png", "one");
        let sub = dir.path().join("other");
        std::fs::create_dir_all(&sub).unwrap();
        let b = source_file(&sub, "shot.png", "two");
        let mut t = task("collision");
        q.attach(&mut t, &[a]).unwrap();
        q.attach(&mut t, &[b]).unwrap();
        assert_eq!(t.attachments, ["shot.png", "shot-2.png"]);
        let paths = q.attachment_paths(&t);
        assert_eq!(std::fs::read_to_string(&paths[0]).unwrap(), "one");
        assert_eq!(std::fs::read_to_string(&paths[1]).unwrap(), "two");
    }

    #[test]
    fn a_renumbered_name_stays_inside_the_length_bound() {
        let (dir, q) = queue();
        let name = format!("{}.png", "a".repeat(60));
        assert_eq!(name.len(), 64);
        let a = source_file(dir.path(), &name, "one");
        let sub = dir.path().join("other");
        std::fs::create_dir_all(&sub).unwrap();
        let b = source_file(&sub, &name, "two");
        let mut t = task("long");
        q.attach(&mut t, &[a, b]).unwrap();
        assert_eq!(t.attachments.len(), 2);
        assert!(
            t.attachments
                .iter()
                .all(|n| crate::ask::valid_asset_name(n))
        );
        assert!(t.attachments[1].ends_with("-2.png"));
    }

    #[test]
    fn a_failed_attach_keeps_existing_attachments_and_leaves_no_partial_copy() {
        let (dir, q) = queue();
        let good = source_file(dir.path(), "good.png", "ok");
        let mut t = task("partial");
        q.attach(&mut t, &[good]).unwrap();
        let more = source_file(dir.path(), "more.png", "ok");
        let missing = dir.path().join("missing.png");
        assert!(q.attach(&mut t, &[more, missing]).is_err());
        assert_eq!(t.attachments, ["good.png"]);
        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
            .unwrap()
            .flatten()
            .collect();
        assert_eq!(on_disk.len(), 1);
    }

    /// Make the next `put` of `t` fail: its temp file's path is a directory.
    fn block_put(q: &Queue, t: &Task) -> PathBuf {
        let tmp = q.path_of(&t.id).with_extension("json.tmp");
        std::fs::create_dir_all(&tmp).unwrap();
        tmp
    }

    #[test]
    fn a_failed_put_leaves_no_new_attachment_directory() {
        let (dir, q) = queue();
        let mut t = task("fresh");
        let tmp = block_put(&q, &t);
        let src = source_file(dir.path(), "shot.png", "x");
        assert!(q.attach_and_put(&mut t, &[src]).is_err());
        assert!(t.attachments.is_empty());
        assert!(!q.attachments_dir(&t.id).exists());
        assert!(!q.path_of(&t.id).exists());
        std::fs::remove_dir(tmp).unwrap();
    }

    #[test]
    fn a_failed_put_removes_only_the_copy_it_just_made() {
        let (dir, q) = queue();
        let mut t = task("edited");
        let first = source_file(dir.path(), "first.png", "1");
        q.attach_and_put(&mut t, &[first]).unwrap();
        block_put(&q, &t);
        let second = source_file(dir.path(), "second.png", "2");
        assert!(q.attach_and_put(&mut t, &[second]).is_err());
        assert_eq!(t.attachments, ["first.png"]);
        let on_disk: Vec<_> = std::fs::read_dir(q.attachments_dir(&t.id))
            .unwrap()
            .flatten()
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .collect();
        assert_eq!(on_disk, ["first.png"]);
        assert_eq!(q.get(&t.id).unwrap().attachments, ["first.png"]);
    }

    #[test]
    fn a_leftover_removing_directory_is_swept_by_the_next_removal() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let gone = task("gone");
        let mut other = task("other");
        let mut live = task("live");
        q.put(&mut other).unwrap();
        q.put(&mut live).unwrap();
        // `gone` has no record: its removal finished deleting the record but
        // not the set-aside attachments.
        let orphan = q.root.join(format!("{}.attachments.removing", gone.id));
        std::fs::create_dir_all(&orphan).unwrap();
        std::fs::write(orphan.join("shot.png"), "x").unwrap();
        // `live` still has a record, so its `.removing` belongs to a removal
        // in progress and must be left alone.
        let busy = q.root.join(format!("{}.attachments.removing", live.id));
        std::fs::create_dir_all(&busy).unwrap();

        q.remove(&other.id, false, &questions).unwrap();
        assert!(!orphan.exists(), "an orphan is swept");
        assert!(busy.exists(), "a removal in progress is left alone");
    }

    #[test]
    fn a_blocked_aside_rename_fails_the_removal_and_loses_nothing() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let mut t = task("stuck");
        let src = source_file(dir.path(), "shot.png", "x");
        q.attach_and_put(&mut t, &[src]).unwrap();
        // A non-empty directory already at the set-aside name makes the rename
        // fail; its record is present, so the sweep leaves it in place.
        let aside = q.root.join(format!("{}.attachments.removing", t.id));
        std::fs::create_dir_all(&aside).unwrap();
        std::fs::write(aside.join("old.png"), "o").unwrap();
        assert!(q.remove(&t.id, false, &questions).is_err());
        assert!(q.path_of(&t.id).exists());
        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
    }

    #[test]
    fn a_failed_record_removal_puts_the_attachments_back() {
        let (dir, q) = queue();
        let mut t = task("rollback");
        let src = source_file(dir.path(), "shot.png", "x");
        q.attach_and_put(&mut t, &[src]).unwrap();
        let err = q
            .remove_record_with_attachments(&t.id, |_| {
                Err(std::io::Error::other("injected failure"))
            })
            .unwrap_err();
        assert!(format!("{err:#}").contains("injected failure"));
        assert!(q.path_of(&t.id).exists());
        assert!(q.attachments_dir(&t.id).join("shot.png").is_file());
        assert!(
            !q.root
                .join(format!("{}.attachments.removing", t.id))
                .exists()
        );
    }

    #[test]
    fn editing_a_task_keeps_its_attachments() {
        let (dir, q) = queue();
        let src = source_file(dir.path(), "shot.png", "x");
        let mut t = task("editable");
        q.attach(&mut t, &[src]).unwrap();
        t.edit("new".to_owned(), "new text".to_owned()).unwrap();
        q.put(&mut t).unwrap();
        assert_eq!(q.get(&t.id).unwrap().attachments, ["shot.png"]);
    }

    #[test]
    fn removing_a_task_deletes_its_attachments() {
        let (dir, q) = queue();
        let questions = Questions::at(dir.path().join("questions"));
        let src = source_file(dir.path(), "shot.png", "x");
        let mut t = task("doomed");
        q.attach(&mut t, &[src]).unwrap();
        q.put(&mut t).unwrap();
        assert!(q.attachments_dir(&t.id).is_dir());
        q.remove(&t.id, false, &questions).unwrap();
        assert!(!q.attachments_dir(&t.id).exists());
        assert!(q.list().is_empty());
    }

    #[test]
    fn a_task_written_before_attachments_still_reads() {
        let (_dir, q) = queue();
        let mut t = task("old");
        q.put(&mut t).unwrap();
        let path = q.path_of(&t.id);
        let mut v: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        v.as_object_mut().unwrap().remove("attachments");
        std::fs::write(&path, v.to_string()).unwrap();
        assert!(q.get(&t.id).unwrap().attachments.is_empty());
    }

    #[test]
    fn attachment_paths_are_absolute_even_when_the_root_is_relative() {
        let q = Queue::at(PathBuf::from("relative-queue"));
        let mut t = task("rel");
        t.attachments.push("shot.png".to_owned());
        let paths = q.attachment_paths(&t);
        assert!(paths[0].is_absolute(), "{}", paths[0].display());
        assert!(paths[0].ends_with(format!("{}.attachments/shot.png", t.id)));
    }

    fn upgrade_parked(id: &str, started: &str) -> Task {
        let mut t = task(id);
        t.id = id.into();
        t.start(format!("run-{id}"));
        t.park("upgrade");
        t.record_pass_at(
            &format!("run-{id}"),
            0,
            Some(started.parse().unwrap()),
            1,
            Some(crate::run::ParkCause::Upgrade {
                from: "old".into(),
                to: None,
            }),
        );
        t
    }

    #[test]
    fn upgrade_resume_order_follows_dispatch_claim_not_graph_start() {
        // A was claimed first but its graph pass started later (slow start-up).
        let stamp = |id: &str, dispatched: &str, started: &str| {
            let mut t = task(id);
            t.id = id.into();
            t.start(format!("run-{id}"));
            t.park("upgrade");
            t.dispatched_at = Some(dispatched.parse().unwrap());
            t.record_pass_at(
                &format!("run-{id}"),
                0,
                Some(started.parse().unwrap()),
                1,
                Some(crate::run::ParkCause::Upgrade {
                    from: "old".into(),
                    to: None,
                }),
            );
            t
        };
        let a = stamp("a", "2026-10-22T00:00:01Z", "2026-10-22T00:00:09Z");
        let b = stamp("b", "2026-10-22T00:00:02Z", "2026-10-22T00:00:03Z");
        assert!(dispatch_order(&a, &b).is_lt());
    }

    #[test]
    fn upgrade_resume_order_survives_queue_reload_and_reversed_settlement() {
        let (_dir, q) = queue();
        let mut later = upgrade_parked("later", "2026-10-22T00:00:02Z");
        let mut earlier = upgrade_parked("earlier", "2026-10-22T00:00:01Z");
        earlier.priority = -10;
        let mut fresh = task("fresh");
        fresh.id = "newer".into();
        fresh.priority = 100;
        for t in [&mut fresh, &mut later, &mut earlier] {
            q.put(t).unwrap();
        }
        let reloaded = Queue::at(q.root().to_owned());
        let mut tasks = reloaded.list();
        tasks.sort_unstable_by(dispatch_order);
        assert_eq!(
            tasks.iter().map(|t| t.id.as_str()).collect::<Vec<_>>(),
            vec!["earlier", "later", "newer"]
        );
        assert_eq!(reloaded.next_runnable().unwrap().id, "earlier");
        assert_eq!(reloaded.get("earlier").unwrap().attempts, 0);
    }

    #[test]
    fn upgrade_resume_legacy_order_uses_settlement_and_id() {
        let mut a = upgrade_parked("a", "2026-10-22T00:00:02Z");
        let mut b = upgrade_parked("b", "2026-10-22T00:00:01Z");
        for t in [&mut a, &mut b] {
            let mut value = serde_json::to_value(&*t).unwrap();
            value["passes"][0]
                .as_object_mut()
                .unwrap()
                .remove("started_at");
            *t = serde_json::from_value(value).unwrap();
        }
        assert!(dispatch_order(&a, &b).is_lt());
        b.passes[0].settled_at = a.passes[0].settled_at;
        assert!(dispatch_order(&a, &b).is_lt());
    }

    #[test]
    fn upgrade_resume_key_excludes_other_parks_and_explicit_task_changes() {
        let original = upgrade_parked("a", "2026-10-22T00:00:01Z");
        assert!(original.upgrade_resume_key().is_some());
        for cause in [
            crate::run::ParkCause::Operator,
            crate::run::ParkCause::Question,
            crate::run::ParkCause::Interrupt {
                reason: "interrupt".into(),
            },
            crate::run::ParkCause::Unspecified,
        ] {
            let mut t = original.clone();
            t.passes[0].parked = Some(cause);
            assert!(t.upgrade_resume_key().is_none());
            let mut fresh = task("fresh");
            fresh.priority = 1;
            assert!(dispatch_order(&fresh, &t).is_lt());
        }
        let mut t = original.clone();
        t.release();
        assert!(t.upgrade_resume_key().is_none());
        t.edit("edited".into(), "new instruction".into()).unwrap();
        assert!(t.upgrade_resume_key().is_none());
        let mut t = original.clone();
        t.requeue();
        t.status = TaskStatus::Parked;
        assert!(t.upgrade_resume_key().is_none());
        let mut t = original.clone();
        t.start("different-run".into());
        t.park("no upgrade");
        assert!(t.upgrade_resume_key().is_none());
    }

    #[test]
    fn explicitly_resumed_older_run_becomes_current_without_duplicate_history() {
        let dir = tempfile::tempdir().unwrap();
        let q = Queue::at(dir.path().join("queue"));
        let mut t = task("resume");
        t.start("A".into());
        t.record_pass("A", 0, 1, None);
        t.start("B".into());
        q.put(&mut t).unwrap();
        assert_eq!(t.runs, vec!["A", "B"]);
        t.start("A".into());
        t.record_pass("A", 1, 3, None);
        q.put(&mut t).unwrap();
        let saved = q.get(&t.id).unwrap();
        assert_eq!(saved.runs, vec!["B", "A"]);
        assert_eq!(saved.runs.last().map(String::as_str), Some("A"));
        assert_eq!(saved.attempts, 3);
        assert_eq!(saved.passes.len(), 2);
        assert_eq!(saved.passes[0].run, "A");
        assert_eq!(saved.passes[1].run, "A");
    }

    #[test]
    fn repeated_start_records_one_run_and_separate_passes() {
        let mut t = task("resume");
        t.start("same-run".into());
        t.record_pass("same-run", 0, 1, None);
        t.start("same-run".into());
        t.record_pass("same-run", 1, 2, None);
        assert_eq!(t.runs, vec!["same-run"]);
        assert_eq!(t.attempts, 2);
        assert_eq!(t.passes.len(), 2);
    }

    #[test]
    fn link_run_adds_a_run_once_and_touches_nothing_else() {
        let dir = tempfile::tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let mut t = Task::new(
            "t".to_owned(),
            "do it".to_owned(),
            PathBuf::from("."),
            Source::Human,
        );
        queue.put(&mut t).unwrap();
        let before = queue.get(&t.id).unwrap();

        let linked = queue.link_run(&t.id[..4], "20260930-092817-ec34").unwrap();
        assert_eq!(linked.runs, vec!["20260930-092817-ec34".to_owned()]);
        assert_eq!(linked.status, before.status);
        assert_eq!(linked.attempts, before.attempts);
        assert_eq!(linked.interrupt, before.interrupt);

        let again = queue.link_run(&t.id, "20260930-092817-ec34").unwrap();
        assert_eq!(again.runs.len(), 1, "linking twice must not duplicate");
        assert_eq!(queue.get(&t.id).unwrap().runs.len(), 1);
        assert!(queue.link_run("no-such-task", "r").is_err());
    }

    #[test]
    fn put_keeps_a_run_linked_after_the_writer_took_its_snapshot() {
        let dir = tempfile::tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let mut t = Task::new(
            "t".to_owned(),
            "do it".to_owned(),
            PathBuf::from("."),
            Source::Human,
        );
        queue.put(&mut t).unwrap();
        // The daemon's copy, taken before the link.
        let mut snapshot = queue.get(&t.id).unwrap();
        queue.link_run(&t.id, "20260930-092817-ec34").unwrap();

        snapshot.start("20260930-000000-aaaa".to_owned());
        queue.put(&mut snapshot).unwrap();

        let stored = queue.get(&t.id).unwrap();
        assert!(stored.runs.contains(&"20260930-092817-ec34".to_owned()));
        assert!(stored.runs.contains(&"20260930-000000-aaaa".to_owned()));
        assert_eq!(stored.attempts, 1);
    }

    #[test]
    fn concurrent_links_and_daemon_saves_lose_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let queue = Queue::at(dir.path().join("queue"));
        let mut t = Task::new(
            "t".to_owned(),
            "do it".to_owned(),
            PathBuf::from("."),
            Source::Human,
        );
        queue.put(&mut t).unwrap();
        let id = t.id.clone();

        let linkers: Vec<_> = (0..4)
            .map(|n| {
                let (queue, id) = (queue.clone(), id.clone());
                std::thread::spawn(move || {
                    for k in 0..10 {
                        queue
                            .link_run(&id, &format!("20260930-00000{n}-l{k:03}"))
                            .unwrap();
                    }
                })
            })
            .collect();
        // The daemon's side: transitions saved from its own, ever staler,
        // copy of the task.
        let mut mine = queue.get(&id).unwrap();
        for k in 0..10 {
            mine.start(format!("20260930-000009-d{k:03}"));
            queue.put(&mut mine).unwrap();
        }
        for l in linkers {
            l.join().unwrap();
        }

        let stored = queue.get(&id).unwrap();
        assert_eq!(stored.runs.len(), 50, "{:?}", stored.runs);
        assert_eq!(
            stored.attempts, 10,
            "linking never rewinds the daemon's work"
        );
    }

    fn age_lock(path: &Path) {
        let f = std::fs::OpenOptions::new().write(true).open(path).unwrap();
        f.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(60))
            .unwrap();
    }

    #[test]
    fn concurrent_stale_takeover_yields_one_holder() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::{Arc, Barrier};
        for _ in 0..5 {
            let dir = tempfile::tempdir().unwrap();
            let q = Queue::at(dir.path().to_path_buf());
            let lock = dir.path().join("t.write-lock");
            std::fs::write(&lock, "dead-0000").unwrap();
            age_lock(&lock);
            let n = 6;
            let barrier = Arc::new(Barrier::new(n));
            let (now, max) = (Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)));
            let handles: Vec<_> = (0..n)
                .map(|_| {
                    let (q, b, now, max) = (q.clone(), barrier.clone(), now.clone(), max.clone());
                    std::thread::spawn(move || {
                        b.wait();
                        let g = q.lock_task("t").unwrap();
                        let held = now.fetch_add(1, Ordering::SeqCst) + 1;
                        max.fetch_max(held, Ordering::SeqCst);
                        std::thread::sleep(std::time::Duration::from_millis(20));
                        now.fetch_sub(1, Ordering::SeqCst);
                        drop(g);
                    })
                })
                .collect();
            for h in handles {
                h.join().unwrap();
            }
            assert_eq!(max.load(Ordering::SeqCst), 1);
            assert!(!lock.exists());
        }
    }

    #[test]
    fn dropping_a_replaced_lock_leaves_the_new_holders_token() {
        let dir = tempfile::tempdir().unwrap();
        let q = Queue::at(dir.path().to_path_buf());
        let lock = dir.path().join("t.write-lock");
        let a = q.lock_task("t").unwrap();
        // Simulate a foreign/legacy writer replacing the file. A modern
        // writer cannot acquire it while this guard's process is alive.
        let replacement = owner_token();
        std::fs::write(&lock, &replacement).unwrap();
        drop(a);
        assert_eq!(std::fs::read_to_string(&lock).unwrap(), replacement);
    }

    #[test]
    fn aged_live_task_lock_excludes_updates_until_retirement_releases_it() {
        let dir = tempfile::tempdir().unwrap();
        let q = Queue::at(dir.path().to_path_buf());
        let held = q.lock_task("t").unwrap();
        let path = dir.path().join("t.write-lock");
        age_lock(&path);
        assert!(!break_stale(&path, &held.token));
        let (started_tx, started_rx) = std::sync::mpsc::channel();
        let (acquired_tx, acquired_rx) = std::sync::mpsc::channel();
        let updater = std::thread::spawn(move || {
            started_tx.send(()).unwrap();
            let _write = q.lock_task("t").unwrap();
            acquired_tx.send(()).unwrap();
        });
        started_rx.recv().unwrap();
        assert!(
            acquired_rx
                .recv_timeout(std::time::Duration::from_millis(100))
                .is_err()
        );
        assert_eq!(std::fs::read_to_string(&path).unwrap(), held.token);
        drop(held);
        acquired_rx
            .recv_timeout(std::time::Duration::from_secs(2))
            .unwrap();
        updater.join().unwrap();
        assert!(!path.exists());
    }

    #[test]
    fn break_stale_leaves_a_lock_that_replaced_the_one_judged() {
        let dir = tempfile::tempdir().unwrap();
        let lock = dir.path().join("t.write-lock");
        std::fs::write(&lock, "new-token").unwrap();
        assert!(!break_stale(&lock, "old-token"));
        assert_eq!(std::fs::read_to_string(&lock).unwrap(), "new-token");
        // Same token but fresh: not stale any more, so also left alone.
        assert!(!break_stale(&lock, "new-token"));
        assert!(lock.exists());
        age_lock(&lock);
        assert!(break_stale(&lock, "new-token"));
        assert!(!lock.exists());
    }

    #[test]
    fn a_stale_break_marker_is_recovered_and_a_fresh_one_is_respected() {
        let dir = tempfile::tempdir().unwrap();
        let lock = dir.path().join("t.write-lock");
        std::fs::write(&lock, "dead-1").unwrap();
        age_lock(&lock);
        let marker = break_marker(&lock, "dead-1");
        std::fs::write(&marker, "crashed-remover").unwrap();
        // Fresh marker: another remover is working, so nothing is broken.
        assert!(!break_stale(&lock, "dead-1"));
        assert!(lock.exists());
        // Stale marker: recovered, then the lock is broken.
        age_lock(&marker);
        assert!(break_stale(&lock, "dead-1"));
        assert!(!lock.exists());
        assert!(!marker.exists());
    }

    fn chat_task() -> Task {
        Task::new(
            "t".to_owned(),
            "do t".to_owned(),
            PathBuf::from("."),
            Source::Agent {
                run: "20260902-000000-beef".to_owned(),
                node: CHAT_NODE.to_owned(),
            },
        )
    }

    #[test]
    fn a_blocked_hold_keeps_its_anchor_but_a_new_run_starts_a_new_hold() {
        let mut t = chat_task();
        t.hold_machine(Some("waiting".into()));
        let old: Timestamp = "2020-01-01T00:00:00Z".parse().unwrap();
        t.held_at = Some(old);
        t.block(vec!["decision".into()], None);
        t.hold_machine(Some("same waiting decision".into()));
        assert_eq!(t.held_at, Some(old));
        t.start("real-new-run".into());
        t.fail("new failure", 1);
        assert_ne!(
            t.held_at,
            Some(old),
            "old answers cannot approve a new hold"
        );
    }

    #[test]
    fn legacy_delivery_survives_upgrade_and_a_no_progress_run_link() {
        let (_dir, queue) = queue();
        let mut t = chat_task();
        t.start("old-run".into());
        t.hold_machine(Some("operator is considering recovery".into()));
        t.chat_notice = Some(ChatNotice {
            key: format!("held#{}", t.held_at.unwrap()),
            at: Timestamp::now(),
            outcome: ChatNoticeOutcome::Posted,
        });
        let mut raw = serde_json::to_value(&t).unwrap();
        raw.as_object_mut().unwrap().remove("chat_notices");
        raw.as_object_mut().unwrap().remove("hold_notice_cause");
        std::fs::create_dir_all(queue.path_of(&t.id).parent().unwrap()).unwrap();
        std::fs::write(queue.path_of(&t.id), serde_json::to_string(&raw).unwrap()).unwrap();
        assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
        queue
            .modify(&t.id, |task| {
                task.requeue();
                task.link_run("unprepared-run");
                task.startup_notice_cause("startup-source");
                task.fail("same unresolved preflight", 0);
                true
            })
            .unwrap();
        assert_eq!(queue.get(&t.id).unwrap().chat_notice_due(), None);
        queue
            .modify(&t.id, |task| {
                task.succeed();
                true
            })
            .unwrap();
        assert_eq!(
            queue.get(&t.id).unwrap().chat_notice_due().as_deref(),
            Some("done")
        );
    }

    #[test]
    fn a_dependency_wait_does_not_change_when_released_and_blocked_again() {
        let mut t = chat_task();
        t.block(vec!["a".into(), "b".into()], Some("waiting".into()));
        let key = t.chat_notice_key().unwrap();
        t.chat_notices.push(key);
        t.release();
        t.block(
            vec!["b".into(), "a".into(), "a".into()],
            Some("still waiting on prerequisites".into()),
        );
        assert_eq!(t.chat_notice_due(), None);
        t.block(
            vec!["new-question".into()],
            Some("different decision".into()),
        );
        assert!(t.chat_notice_due().is_some());
    }

    #[test]
    fn running_startup_failures_preserve_stage_and_report_distinct_blockers() {
        let mut t = chat_task();
        t.start("existing-run".into());
        t.fail("implementation failed", 1);
        let failed = t.chat_notice_key().unwrap();
        t.chat_notices.push(failed.clone());
        t.release_operator("test", None);
        t.start("existing-run".into());
        t.fail_startup("startup-source", "same unresolved recovery source", 1);
        assert_eq!(
            t.chat_notice_due(),
            None,
            "no-progress recovery stays quiet"
        );
        t.release_operator("test", None);
        t.start("existing-run".into());
        t.fail_startup("startup-attachments", "attachment missing", 1);
        let attachments = t.chat_notice_due().expect("new attachment blocker");
        assert_ne!(attachments, failed);
        assert_eq!(
            t.hold_notice_cause.as_ref().unwrap().stage,
            "startup-attachments"
        );
        t.chat_notices.push(attachments);
        t.release_operator("test", None);
        t.start("existing-run".into());
        t.fail_startup(
            "startup-attachments",
            "same missing attachment, reworded",
            1,
        );
        assert_eq!(
            t.chat_notice_due(),
            None,
            "same stage after attempt counter reset"
        );
        t.release_operator("test", None);
        t.start("existing-run".into());
        t.fail_startup("startup-config", "different preflight stage", 1);
        assert!(t.chat_notice_due().is_some());
        assert_eq!(
            t.hold_notice_cause.as_ref().unwrap().stage,
            "startup-config"
        );
    }

    #[test]
    fn structured_causes_preserve_no_progress_retries_and_report_new_passes() {
        let mut t = chat_task();
        t.hold_machine(Some("startup source missing".into()));
        t.startup_notice_cause("startup-source");
        let first = t.chat_notice_key().unwrap();
        t.chat_notices.push(first.clone());
        t.release_operator("test", None);
        t.startup_notice_cause("startup-source");
        t.hold_machine(Some("same source cannot be read".into()));
        assert_eq!(t.chat_notice_due(), None);
        t.startup_notice_cause("startup-config");
        assert!(t.chat_notice_due().is_some(), "different startup stage");
        t.notice_cause("failed", Some("existing-run"), Some(0));
        let run_failure = t.chat_notice_key().unwrap();
        t.chat_notices.push(run_failure.clone());
        t.release_operator("test", None);
        t.startup_notice_cause("startup-source");
        t.hold_machine(Some("no new work happened".into()));
        assert_eq!(t.chat_notice_due(), None);
        t.notice_cause("failed", Some("existing-run"), Some(1));
        assert!(t.chat_notice_due().is_some(), "a real resumed pass failed");
        assert_ne!(t.chat_notice_key().unwrap(), run_failure);
    }

    #[test]
    fn legacy_hold_baseline_does_not_replay_or_hide_new_failures_and_completion() {
        let mut t = chat_task();
        t.hold_machine(Some("still waiting".into()));
        t.chat_notice = Some(ChatNotice {
            key: "held#2026-10-14T00:00:00Z".into(),
            at: Timestamp::now(),
            outcome: ChatNoticeOutcome::Posted,
        });
        assert_eq!(t.chat_notice_due(), None, "old hold baseline");
        t.import_chat_notice();
        t.notice_cause("failed", Some("new-run"), Some(0));
        assert!(t.chat_notice_due().is_some(), "new producer evidence");
        t.succeed();
        assert_eq!(t.chat_notice_due().as_deref(), Some("done"));
    }

    #[test]
    fn a_chat_notice_is_due_only_for_a_final_state_and_only_once() {
        let mut t = chat_task();
        assert_eq!(t.chat_notice_due(), None, "queued");
        t.start("r1".to_owned());
        assert_eq!(t.chat_notice_due(), None, "running");
        t.fail("boom", 3);
        assert_eq!(t.status, TaskStatus::Failed);
        assert_eq!(t.chat_notice_due(), None, "a retried failure is not final");
        t.start("r2".to_owned());
        t.park("upgrade");
        assert_eq!(t.chat_notice_due(), None, "parked");
        t.fail("boom", 1);
        assert_eq!(t.status, TaskStatus::Held);
        let key = t
            .chat_notice_due()
            .expect("held on the last attempt is final");
        t.chat_notice = Some(ChatNotice {
            key,
            at: Timestamp::now(),
            outcome: ChatNoticeOutcome::Posted,
        });
        assert_eq!(t.chat_notice_due(), None, "same hold, no second notice");
        t.release();
        assert_eq!(t.chat_notice_due(), None, "requeued after a release");
        t.start("r3".to_owned());
        t.succeed();
        assert!(t.chat_notice_due().is_some(), "done is a new final state");
    }

    #[test]
    fn completed_chat_notices_keep_current_and_legacy_keys_after_requeue() {
        for key in ["done", "done#0", "done#1", "done#12"] {
            let mut t = chat_task();
            t.start("r1".to_owned());
            t.succeed();
            t.chat_notice = Some(ChatNotice {
                key: key.to_owned(),
                at: Timestamp::now(),
                outcome: ChatNoticeOutcome::Posted,
            });
            t.requeue();
            t.start("r2".to_owned());
            t.succeed();
            assert_eq!(t.chat_notice_key().as_deref(), Some("done"));
            assert_eq!(t.chat_notice_due(), None, "recorded key: {key}");
        }
    }

    #[test]
    fn a_release_without_progress_keeps_the_unresolved_notice_cause() {
        let mut t = chat_task();
        t.hold_manual(Some("wait".to_owned()));
        let first = t.chat_notice_due().expect("held");
        t.chat_notice = Some(ChatNotice {
            key: first.clone(),
            at: Timestamp::now(),
            outcome: ChatNoticeOutcome::Posted,
        });
        t.hold_manual(Some("still waiting".to_owned()));
        assert_eq!(t.chat_notice_due(), None, "a re-hold keeps its start");
        t.release();
        std::thread::sleep(std::time::Duration::from_millis(5));
        t.hold_manual(None);
        assert_eq!(t.chat_notice_due(), None, "no progress, same cause");
    }

    #[test]
    fn a_chat_notice_needs_a_chat_source_and_a_current_schema() {
        let mut human = task("h");
        human.succeed();
        assert_eq!(human.chat_notice_due(), None);
        // A follow-up only inherits the chat; it did not file the task.
        let mut inherited = task("i");
        inherited.origin_chat = Some("20260902-000000-beef".to_owned());
        inherited.succeed();
        assert_eq!(inherited.chat_notice_due(), None);
        let mut old = chat_task();
        old.schema = 12;
        old.status = TaskStatus::Done;
        assert_eq!(old.chat_notice_due(), None, "history is not replayed");
        let mut live = chat_task();
        live.schema = 12;
        live.start("r1".to_owned());
        live.succeed();
        assert!(live.chat_notice_due().is_some(), "started by this build");
        let mut held = chat_task();
        held.schema = 12;
        held.hold_machine(Some("config".to_owned()));
        assert!(held.chat_notice_due().is_some(), "held before any start");
    }

    #[test]
    fn a_stale_snapshot_does_not_rewind_a_recorded_chat_notice() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let q = Queue::at(tmp.path().join("queue"));
        let mut t = chat_task();
        t.succeed();
        q.put(&mut t).expect("put");
        let mut stale = q.get(&t.id).expect("get");
        let key = t.chat_notice_due().expect("due");
        q.modify(&t.id, |x| {
            x.chat_notice = Some(ChatNotice {
                key: key.clone(),
                at: Timestamp::now(),
                outcome: ChatNoticeOutcome::Posted,
            });
            true
        })
        .expect("modify");
        q.put(&mut stale).expect("stale put");
        assert_eq!(q.get(&t.id).expect("get").chat_notice_due(), None);
    }

    #[test]
    fn empty_sources_are_omitted_so_older_builds_still_read_the_task() {
        // `Task` rejects unknown fields, so a build without `sources` cannot
        // read a record that carries the key, even an empty one.
        let mut t = task("plain");
        let value = serde_json::to_value(&t).unwrap();
        assert!(value.get("sources").is_none(), "{value}");
        t.sources.pick_commits.push(crate::refs::Input {
            token: "abc1234".into(),
            sha: "a".repeat(40),
            provenance: crate::refs::Provenance::ExplicitCli,
        });
        let value = serde_json::to_value(&t).unwrap();
        assert_eq!(value["sources"]["pick_commits"][0]["sha"], "a".repeat(40));
        let restored: Task = serde_json::from_value(value).unwrap();
        assert_eq!(restored.sources, t.sources);
    }
}