magi/

run.rs

//! Run state: what happened, where it is stored, and how a run is resumed.
//!
//! Every node writes its result into [`RunState`] and the whole struct is
//! flushed to `run.json` before the next node starts. That is what makes a run
//! resumable: a competition can take an hour, and dying in review round four
//! should not throw away three implementations, nine judge reads and a
//! deliberation.
//!
//! Patches and raw agent transcripts are *not* in `run.json` — they live beside
//! it under `artifacts/`, so the state file stays small enough to read by hand.
use std::collections::{BTreeMap, BTreeSet};
use std::path::{Path, PathBuf};

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

use crate::agent::SeatState;
use crate::blind::Leak;
use crate::config::{Config, MergeMode};
use crate::proc::Quiet as _;
use crate::verdict::{Finding, Rejection, ReviewVote, Severity};

/// On-disk format version. Bumped when a field changes meaning, so a resumed
/// run never half-reads a state file written by a different magi.
///
/// 2: added `RunStatus::Stalled`, `RunState::quota` (rate-limit losses), and
/// the quorum fields on `Tally`. `RunState::load` already fails loudly and
/// clearly on a schema mismatch; an old `run.json` from schema 1 now says so
/// instead of silently half-reading.
///
/// 3: added `RunState::judge_skipped`. A solo candidate makes `judge` write
/// only an event, leaving `judgements` empty forever — indistinguishable from
/// "not yet judged" on every later reentry, which is what let `judge` re-run
/// on a finished run and clobber its status back to `Judging`. The flag is
/// the missing record of the fact that judging was skipped on purpose.
///
/// Also 3: a single-viable-candidate tally records `Tally::judges` as `0` and
/// fills `Tally::uncontested`, instead of leaving the full roster size sitting
/// next to a panel that never sat. A schema-2 record keeps reading as "0 of 3
/// judges present" forever, because a tally is computed once and never
/// recomputed on resume; the bump keeps that stale reading from being mixed
/// with the new meaning.
///
/// 4: added `ReviewRound::progressed`. `graph::STAGNANT_LIMIT` counts
/// consecutive rounds with `progressed == false` to decide whether the
/// review loop should give up early, and a schema-3 record's default
/// `false` would misreport a round that, at the time, actually committed a
/// real diff — the field simply did not exist yet to say so. Without the
/// bump, resuming an old multi-round review could spuriously trip the
/// stagnation check on rounds that were never stagnant.
///
/// 5: added `RunStatus::Landing`. A run inside [`crate::land`]'s post-merge
/// loop used to carry whatever status `merge` set before calling it forward
/// unchanged - `Merged`, even while still watching CI or waiting on the
/// owner's approval - which is also the one status [`RunStatus::resumable`]
/// treats as finished. A daemon that gave this run's slot back to poll
/// something else while an approval was outstanding, or one that simply
/// crashed mid-land, had no way to tell "still landing" from "actually
/// merged" and would either restart the whole competition or leave the run
/// stuck reading as done. A schema-4 record has no notion of `Landing` at
/// all, so this is a meaning a resumed old run cannot be guessed into rather
/// than a value it can default to - hence the bump, not a `#[serde(default)]`.
///
/// 6: a deferred e2e is represented by an empty outcome list plus
/// `ReviewRound::e2e_deferred`. Schema 5 treated that same empty list as an
/// unconfigured, successful check, so schema-5 records are migrated with the
/// old (not-deferred) meaning while older binaries reject schema-6 records.
///
/// 7: added `RunState::gate_ran`. An empty `RunState::gate` used to carry two
/// meanings at once — "never attempted, or the last attempt was
/// resource-blocked" (`graph::Runner::gate`'s retry case) and "attempted,
/// zero commands configured, vacuously passed" (a repo with no
/// `verify.gate`) — and nothing told them apart. `graph::Runner::merge`
/// therefore read the second case as the first and refused forever: a
/// review-only run with no gate commands configured reached `Gating` and
/// then could never leave it. A schema-6 record's non-empty `gate` is
/// migrated to `gate_ran = true` (a recorded attempt, real or historical,
/// should not be spent again); an empty one migrates to `gate_ran = false`
/// and is simply re-attempted by the next `gate()` call, which self-heals
/// instantly for the zero-commands case.
///
/// 8: `ReviewRound::verified_head` used to be `None` for the overwhelming
/// majority of rounds — every ordinary round that ran e2e against its own
/// `head` in the main review loop never set it at all, leaving only the
/// rare catch-up-on-a-different-commit case populated. A reader (a review
/// prompt, `magi show`, the web UI) had no field to ask "which commit did
/// this round's `e2e` actually check" and fell back to assuming it was
/// always `head`, which is also what let a stale round's red output get
/// quoted to a later round's reviewers as if it were about their patch, not
/// an earlier one (see `ReviewRound::verification_summary`, which now exists
/// so nowhere else has to guess). `verified_head` is now set whenever `e2e`
/// held a real attempt (`E2eStatus::Passed`/`Failed`), always naming the
/// commit actually checked instead of only the divergent case, and
/// `ReviewRound::verified_at` is new alongside it. A schema-7 round's own
/// unconditional main-loop check was always against `head` whether or not
/// this field said so, so a `None` with a non-empty `e2e` migrates to
/// `Some(head)` — a reconstruction of a fact that was always true, not a
/// guess. `verified_at` has no historical value to reconstruct and stays
/// `None`, which reads through `verification_summary` as "checked at:
/// unknown" — an honest gap, not a fabricated time.
/// 9: added [`RunState::operator_fixes`] — one record per `magi fix`
/// invocation, routing specific, already-recorded findings to a fixer as a
/// targeted, out-of-band fix outside the normal round sequence. Kept in a
/// channel of its own rather than folded into [`ReviewRound`], because a
/// reviewer's own severity and vote (copied verbatim onto
/// [`OperatorFixFinding`]) must never be rewritten to look like the operator
/// manufactured a blocking verdict — see `graph::Runner::fix_selected`. A
/// schema-8 record has no operator-fix history at all, and
/// `#[serde(default)]` reads an empty list as exactly that: "none happened",
/// not an unknown gap. Nothing about an existing field's meaning changes.
///
/// 10: added `RunStatus::VerifiedNoop` and `Candidate::verified_noop`. Before
/// this, an implementer that correctly concluded (with evidence) that a
/// task's request was already satisfied elsewhere had no way to say so: the
/// run ended the same way as one where every candidate simply failed to
/// write anything — `after_implement` bailing with "no candidate produced a
/// change; nothing to judge" and the run settling as a plain `Failed`. That
/// conflated two very different facts (investigation run 391f's audit is
/// what surfaced it: two attempts that had, correctly, found their fix
/// already on `main`). A schema-9 record has no notion of either the new
/// status or field, so a `VerifiedNoop` value is a meaning that cannot be
/// reconstructed from an old record — hence the bump, not a
/// `#[serde(default)]` for the status. `Candidate::verified_noop` alone
/// *does* default-read as `None` on an old record, which is the honest
/// reading: a run written before this schema never made the claim.
///
/// The report task 391f itself was raised from also named `6c5e`, `8df3` and
/// `e9ce` as three more tasks whose implement wave ended the same
/// diff-zero way, and the investigation traced all three — they do not
/// share one cause.
///
/// `6c5e` and `8df3` are the same already-landed pattern as `391f`, not a
/// coincidence: all three were re-queued together by a same-day audit of
/// `done`-but-unlanded magi tasks (queue talk `20260912-115153-7216`,
/// 2026-09-12 02:51–04:24), which found 17 magi tasks marked `done` with no
/// merge to show for it and re-queued 16 of them, `6c5e` (a fix for the
/// owner's `magi ask --thread` back-and-forth) and `8df3` (release
/// automation) included. A second, same-day audit (talk
/// `20260912-222053-07fe`, 13:20–13:36) then found 12 of those re-queued
/// tasks — `391f`, `6c5e` and `8df3` among them — already merged by another
/// route, and the owner had them deleted (`magi task rm`); `391f` alone
/// survived because a daemon still held its run at the moment of deletion,
/// which is the only reason any record of this group still exists to audit.
/// Quoted directly from that second audit's own turn (talk `07fe`, so this
/// reads without needing access to that talk store), naming both by id:
///
/// > 12件がマージ済み(対応不要)、3件が未実装(妥当)、2件が部分実装(要確認)でした。
/// > **マージ済み → hold/rmを推奨:** 6c5e, 1ddc, fcf5, e25b, cea2, 391f, 3202,
/// > b0a1, 5365, af85, 9f26, 8df3
///
/// — followed by the owner answering "削除！" and the agent confirming "11件
/// 削除完了。391f はいま実行中のdaemonが掴んでいて削除できませんでした."
/// `git log` independently confirms both fixes: the ask-back feature `6c5e`
/// wanted landed as `f0df474` ("let the owner ask back on a question...",
/// #93) on 2026-09-06, and the release-bump automation `8df3` wanted landed
/// as `61005dd`/`bedd925` (open a release-bump PR on merge) on 2026-09-07
/// and `116fcdc` (proportional version bump, #108) on 2026-09-08 — all
/// before the 09-12 requeue. No run record survives the deletion for either
/// task, so this schema's evidence is the audit transcript plus the
/// independently re-checked `git log`, not a `run.json`.
///
/// `e9ce` is not that pattern at all, and is the reason the adoption guard
/// below is all-or-nothing rather than "any candidate said so": its task
/// asked an implementer to merge the real repository's `main` and cut a
/// GitHub release — a destructive, out-of-worktree operation `AGENTS.md`
/// names explicitly as not something to hand to an unattended candidate.
/// Both of its runs (`20260912-053352-49ad`, `20260912-062629-bab1`)
/// correctly refused, filed `magi ask` (questions `6196`, `6c9a`), and ended
/// with an empty diff only because no answer arrived before the implement
/// node's timeout — `49ad` looped `magi ask --wait` in the foreground for
/// roughly 50 minutes as instructed before the timeout cut it off; `bab1`
/// ended its turn moments after filing its question without ever actually
/// blocking on the wait, a separate protocol slip this schema change does
/// not attempt to fix. `49ad`'s own `candidates[0].summary` (quoted here
/// because both records predate schema 10 and, separately, predate a
/// still-unrelated struct change that already makes today's `magi show`
/// refuse to parse either of them — `unknown field 'planner'` — so this is
/// read straight from `run.json` on disk, not through that command):
///
/// > タスクの内容（READY 状態の run を実リポジトリの main に `merge --no-ff`
/// > する、GitHub Release を作る）を精査した結果、これは全てこのワーカーの
/// > worktree の外にある実リポジトリと GitHub 上の共有状態に対する不可逆な
/// > 操作であり […] 私自身の運用ルール「Work only inside this worktree.
/// > Nothing outside it is yours.」と正面から矛盾すると判断しました。
///
/// Neither candidate's reply carries the
/// `NO CHANGE NEEDED` marker below, so both runs correctly stay `Failed`
/// under this schema, not `VerifiedNoop`: a run blocked on an unanswered
/// authorization question is not a verified no-op, and reading the two
/// alike is exactly the misclassification the guard's per-candidate and
/// whole-run conditions exist to refuse.
///
/// `RunStatus::Superseded` (added without a bump, still schema 10) is
/// additive in the ordinary sense — an old build reading a run written under
/// a newer one only had `RunStatus` variants to worry about before this, and
/// a record already on disk never contained the new variant to begin with —
/// but it is not additive in the sense every earlier schema bump on this
/// constant was: `daemon::supersede_prior_runs` now rewrites a `Blocked` or
/// `Stalled` run's `status` field *after* it was first written, once a later
/// attempt at the same task lands. Any tooling outside magi that reads
/// `run.json` and assumes a terminal `status` is permanent once set — a
/// monitoring script polling for `blocked`, say — needs to know that
/// `superseded` is where some of those records now go instead of staying put
/// forever; see that function's own doc for exactly when.
///
/// Schema 11 adds [`RunState::released_to`] / [`RunState::released_branches`]:
/// a run whose worktree a later attempt at the same task took over
/// (`crate::handover`) can no longer be resumed from where it was. The fields
/// are additive, but an older build would happily resume such a run into a
/// worktree that no longer exists, which is the one thing the bump is for.
///
/// Schema 12 adds [`RunState::driver_exited`]: the driver records that it
/// stopped walking the graph. Additive, but an older build would keep reading
/// a long-lived daemon's pid as a live driver of a run that ended long ago.
///
/// Schema 13 adds [`RunState::origin`]: who started the run and which task it
/// stood in for. Additive, and an older record is deliberately left with
/// `None` ("origin unknown") rather than guessed at — a run from before this
/// existed must not read as an operator's when nobody knows. The bump exists
/// so a run is never mistaken for one whose origin was known and empty.
///
/// Schema 14 records [`RunState::task_instruction`], authored task criteria
/// separately from generated review context. Older records retain `None`:
/// `instruction` may contain generated context and cannot establish authorship.
///
/// Schema 15 records pending candidate preservation independently of an empty
/// implementation. Older drivers must refuse to resume these records rather
/// than judge without honoring the preservation barrier.
///
/// Schema 16 records empty-attempt retirement. Older executors must reject
/// retired runs rather than ignoring the new execution fence.
pub const SCHEMA: u32 = 16;

/// Why a pass over a run stopped at a node boundary, recorded when it
/// happened. Never inferred afterwards: `upgrade.json` is overwritten by the
/// next upgrade and an event line is prose.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ParkCause {
    /// `magi web` / `magi serve` replaced its own binary. `to` is `None` when
    /// the target release was not known at the time.
    Upgrade {
        /// The version that was running.
        from: String,
        /// The release it was replaced with, when known.
        #[serde(default)]
        to: Option<String>,
    },
    /// The operator asked the loop to park (`/api/loop`).
    Operator,
    /// The interrupt scheduler parked this run so another task runs first.
    Interrupt {
        /// What the scheduler gave as the reason.
        reason: String,
    },
    /// The run waits for the owner's answer (merge approval, withheld text).
    Question,
    /// A park whose requester did not say why.
    Unspecified,
}

impl ParkCause {
    /// One line for an event or the UI.
    #[must_use]
    pub fn label(&self) -> String {
        match self {
            Self::Upgrade { from, to: Some(to) } => format!("upgrade {from} \u{2192} {to}"),
            Self::Upgrade { from, to: None } => format!("upgrade from {from}"),
            Self::Operator => "operator asked the loop to park".to_owned(),
            Self::Interrupt { reason } => reason.clone(),
            Self::Question => "waiting for the owner's answer".to_owned(),
            Self::Unspecified => "park requested, reason not stated".to_owned(),
        }
    }
}

/// How a pass ended, when the process that drove it said so.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "how", rename_all = "snake_case")]
pub enum PassEnd {
    /// Parked at a node boundary on purpose.
    Parked {
        /// Why it parked.
        cause: ParkCause,
    },
    /// The next pass found this one still open: it stopped without saying so
    /// (a crash, a kill). Says nothing about why.
    Unrecorded,
}

/// One pass over a run: the run is the same, the process driving it is not.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunPass {
    /// Position in [`RunState::passes`].
    pub index: usize,
    /// When the process took the run over.
    pub started_at: Timestamp,
    /// The run already had work on disk when this pass began.
    pub resumed: bool,
    /// When the pass ended, if it did.
    #[serde(default)]
    pub ended_at: Option<Timestamp>,
    /// `None` while open (or still driving).
    #[serde(default)]
    pub end: Option<PassEnd>,
    /// The task loop (or `magi run` for a task) counted this pass as a task
    /// attempt. Resumes keep one distinct id in `Task::runs`, while each
    /// counted pass belongs in task history. A resume without a task does not.
    #[serde(default)]
    pub task_entry: bool,
}

/// What an unrecorded origin reads as, wherever one is shown.
pub const ORIGIN_UNKNOWN: &str = "origin unknown (started before origins were recorded)";

/// Who started a run. Legibility only: nothing branches on it, the same rule
/// as `agent::Invocation::run` / `node`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum StartedBy {
    /// The daemon, minting an attempt for a queue task.
    Queue {
        /// The task the attempt belongs to.
        task: String,
    },
    /// An agent inside a talk (chat) conversation, via `magi run` / `magi review`.
    Chat {
        /// Talk id, e.g. `4a7b`.
        talk: String,
    },
    /// An agent seat inside another run, identified by the `MAGI_RUN` /
    /// `MAGI_NODE` it was spawned with.
    Seat {
        /// Run the seat belonged to.
        run: String,
        /// Node it was working in.
        node: String,
    },
    /// A person at a terminal (or anything that carries no attribution).
    Operator,
}

/// Where a run came from, and the task it was meant to finish, if any.
///
/// Kept as two facts because they vary independently: a chat agent may start a
/// run on behalf of a queue task.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Origin {
    /// Who started it.
    pub by: StartedBy,
    /// The queue task this run serves, when it names one.
    #[serde(default)]
    pub task: Option<String>,
}

impl Origin {
    /// A run a person started with no task attached.
    pub fn operator() -> Self {
        Self {
            by: StartedBy::Operator,
            task: None,
        }
    }

    /// An attempt the daemon mints for `task`.
    pub fn queue(task: &str) -> Self {
        Self {
            by: StartedBy::Queue {
                task: task.to_owned(),
            },
            task: Some(task.to_owned()),
        }
    }

    /// Attribute from the `MAGI_RUN` / `MAGI_NODE` pair `agent::invoke`
    /// exports, as `(run, node)`; `None` for no (or blank) `run` is an
    /// operator. A node of `chat` is the talk layer, whose run is the talk id.
    pub fn from_agent_env(env: Option<(String, String)>, task: Option<String>) -> Self {
        let by = match env {
            Some((run, node)) if node == crate::queue::CHAT_NODE => StartedBy::Chat { talk: run },
            Some((run, node)) => StartedBy::Seat { run, node },
            None => StartedBy::Operator,
        };
        Self { by, task }
    }

    /// The same origin, naming `task` as the one it serves.
    pub fn serving(mut self, task: Option<String>) -> Self {
        if task.is_some() {
            self.task = task;
        }
        self
    }

    /// One short label for lists, reports and the web UI.
    pub fn label(&self) -> String {
        let by = match &self.by {
            StartedBy::Queue { task } => format!("task {}", short_of(task)),
            StartedBy::Chat { talk } => format!("chat {}", short_of(talk)),
            StartedBy::Seat { run, node } => format!("{node}@{}", short_of(run)),
            StartedBy::Operator => "operator".to_owned(),
        };
        match (&self.by, &self.task) {
            (StartedBy::Queue { task: a }, Some(b)) if a == b => by,
            (_, Some(t)) => format!("{by}, for task {}", short_of(t)),
            _ => by,
        }
    }
}

/// [`Origin::label`] for a run that may predate origins.
pub fn origin_label(origin: Option<&Origin>) -> String {
    origin.map_or_else(|| ORIGIN_UNKNOWN.to_owned(), Origin::label)
}

/// Where a run got to.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunStatus {
    /// Worktrees being prepared.
    Prep,
    /// Candidates being implemented.
    Implementing,
    /// Judges ranking blind.
    Judging,
    /// Judges deliberating after a split.
    Deliberating,
    /// Final votes being collected privately.
    Voting,
    /// Winner in the review + verification loop.
    Reviewing,
    /// Gate commands running. Transient: `graph::Runner::gate` and
    /// `graph::Runner::merge` always move a run on from here, whether or not
    /// any gate commands are configured — see [`RunState::gate_status`] and
    /// `SCHEMA`'s doc for schema 7, which fixed a repo with an empty
    /// `verify.gate` stranding a review-only run in `Gating` forever.
    Gating,
    /// Inside [`crate::land`]'s post-merge loop: watching CI, running a fix
    /// round, rebasing onto a moved base, or waiting on the owner's merge
    /// approval. A run parked here while an approval is outstanding has
    /// handed its daemon slot back — see [`crate::daemon`] — and resumes
    /// through exactly this status, not a fresh competition.
    Landing,
    /// Winner merged.
    Merged,
    /// Local review or publication finished; no public merge is confirmed.
    Ready,
    /// The judgement did not gather enough judges (e.g. rate limiting took out
    /// seats), so the verdict is not trustworthy. The run stopped and kept its
    /// work so it can be resumed or folded — it must never be confused with a
    /// healthy `Ready`.
    Stalled,
    /// Review rounds exhausted with findings still open, or the gate failed.
    Blocked,
    /// The graph could not complete.
    Failed,
    /// Every candidate wrote nothing, and every one of them said why in a way
    /// that survived [`crate::graph::Runner`]'s adoption guard: a clean CLI
    /// exit, an actually-empty tree, no command left with an unconfirmed
    /// exit status, and non-empty evidence. Distinct from `Failed` on
    /// purpose — see `SCHEMA`'s doc for schema 10 — because the two read
    /// identically to an operator glancing at a card ("nothing happened")
    /// while meaning opposite things: one is an agent that could not do the
    /// work, the other is an agent that checked and the work was already
    /// done. Settles the task through [`crate::queue::Task::handed_off`], not
    /// [`crate::queue::Task::fail`]: a human still has to look — the claim is
    /// unverified by magi itself — and `Held` (not `Failed`-and-requeued)
    /// means nothing retries the task unattended on the same unconfirmed
    /// claim while that look is pending.
    VerifiedNoop,
    /// A later attempt at the same task already finished the job — see
    /// `crate::queue::Task::superseded_attempts` — so this run's own
    /// `Blocked`/`Stalled` no longer means anyone has to look at it. Written
    /// over one of those two statuses once the task itself is
    /// `crate::queue::TaskStatus::Done`, and only onto an attempt that comes
    /// before the one that succeeded. Publication can also retire a stopped,
    /// non-terminal predecessor after an explicit worktree handover to its
    /// successor leaves it nothing to resume. Unlike them, not [`Self::resumable`]:
    /// there is nothing left to resume towards, the task already has its
    /// answer, and a fold is free to clean this run's worktree away without
    /// waiting on a human to confirm that first.
    Superseded,
    /// The winner's whole change is already on the base under other commit
    /// ids (a cherry-pick or squash by another task) — see
    /// [`crate::already`]. Terminal and not a failure: nothing is left to
    /// land, so unlike `Blocked` nobody has to look, and unlike `Merged` this
    /// run did not land anything. Never resumable; `BaseSync::already_in`
    /// names the base commits that carry the change. Added without a schema
    /// bump, like `Superseded`: no record on disk can contain it yet.
    AlreadyInBase,
}

impl RunStatus {
    /// Is this a terminal state?
    pub fn done(self) -> bool {
        matches!(
            self,
            Self::Merged
                | Self::Ready
                | Self::Stalled
                | Self::Blocked
                | Self::Failed
                | Self::VerifiedNoop
                | Self::Superseded
                | Self::AlreadyInBase
        )
    }

    /// The name this status is written and shown under, matching the
    /// `snake_case` serde spelling so a log line, an error message and the
    /// JSON a phone reads all say the same word.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Prep => "prep",
            Self::Implementing => "implementing",
            Self::Judging => "judging",
            Self::Deliberating => "deliberating",
            Self::Voting => "voting",
            Self::Reviewing => "reviewing",
            Self::Gating => "gating",
            Self::Landing => "landing",
            Self::Merged => "merged",
            Self::Ready => "ready",
            Self::Stalled => "stalled",
            Self::Blocked => "blocked",
            Self::Failed => "failed",
            Self::VerifiedNoop => "verified_noop",
            Self::Superseded => "superseded",
            Self::AlreadyInBase => "already_in_base",
        }
    }

    /// Label for a human-facing listing or report — the same word as
    /// [`Self::as_str`] except where the machine spelling would read harsher
    /// than the state actually is. `VerifiedNoop` is the one case: its own
    /// `as_str` exists for logs, JSON and event messages, none of which
    /// should quietly grow a second vocabulary, but a bare "verified_noop" in
    /// a report reads like an error code, not the qualified, evidence-backed
    /// claim it actually is.
    pub fn display_label(self) -> &'static str {
        match self {
            Self::VerifiedNoop => "agent-verified no-op",
            Self::AlreadyInBase => "already in base",
            other => other.as_str(),
        }
    }

    /// Can this run be carried on from where it stopped?
    ///
    /// Everything except a finished run and a failed one. `execute` skips
    /// nodes already recorded, so re-entering is cheap wherever the run
    /// stopped, and the alternative is always a fresh competition against
    /// work that already exists.
    ///
    /// - `Stalled` re-asks only the seats whose absence collapsed the panel,
    ///   keeping the candidates that were already paid for.
    /// - `Blocked` re-enters the review loop against a branch that is built.
    /// - **A non-terminal status** means the run was interrupted: a parked
    ///   run waiting for its upgrade, or one whose daemon was killed. This
    ///   used to be excluded, which left run 4043 stuck at `reviewing` with
    ///   the deck telling the operator it could not be resumed - the one
    ///   state where resuming is the only sensible answer.
    ///
    /// `Failed` does not qualify: the graph could not complete and there is
    /// no established point to continue from. Nor does a finished run, whose
    /// answer is a new competition. Nor does `VerifiedNoop`: every candidate
    /// already agreed nothing belongs in this worktree, and resuming would
    /// only re-ask the same question — the answer is for a human to check
    /// the evidence, not for the graph to run again. Nor does `Superseded`:
    /// a later attempt at the same task already finished it, so there is
    /// nothing left this run's own answer could still contribute.
    ///
    /// Whether anything is *already* driving the run is a separate question,
    /// answered by `daemon::is_working_on` at the callers that need it.
    pub fn resumable(self) -> bool {
        !matches!(
            self,
            Self::Merged
                | Self::Ready
                | Self::Failed
                | Self::VerifiedNoop
                | Self::Superseded
                | Self::AlreadyInBase
        )
    }
}

/// One candidate implementation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Candidate {
    /// Position in the implementer list.
    pub index: usize,
    /// Blind label as presented to judges.
    pub label: char,
    /// Which agent wrote it. Recorded for the stats tables, never shown to a
    /// judge.
    pub agent: String,
    /// Branch, named after the label so judges can inspect it without learning
    /// the author.
    pub branch: String,
    /// Worktree path.
    pub worktree: PathBuf,
    /// Sanitized author summary.
    #[serde(default)]
    pub summary: String,
    /// `git diff --stat`.
    #[serde(default)]
    pub stat: String,
    /// Files touched.
    #[serde(default)]
    pub files: usize,
    /// Commits ahead of base.
    #[serde(default)]
    pub commits: usize,
    /// True when the agent produced no change at all.
    #[serde(default)]
    pub empty: bool,
    /// Why this candidate is not in the running.
    #[serde(default)]
    pub failed: Option<String>,
    /// The evidence this candidate gave for writing no change on purpose —
    /// the `NO CHANGE NEEDED:` marker `prompt::implement`'s reply format
    /// documents, verbatim. `Some` only when [`crate::graph`]'s adoption
    /// guard accepted the claim: the CLI exited cleanly, the tree really is
    /// empty, no command in the reply was left with an unconfirmed exit
    /// status, and the evidence itself is non-empty. A candidate that wrote
    /// nothing and said nothing about why — the ordinary empty loss — always
    /// reads `None` here, same as one written before schema 10 ever existed.
    #[serde(default)]
    pub verified_noop: Option<String>,
    /// Wall-clock time for the implementation.
    #[serde(default)]
    pub duration_ms: u64,
    /// Whether the worktree has been folded away.
    #[serde(default)]
    pub folded: bool,
}

impl Candidate {
    /// Can this candidate be judged?
    pub fn viable(&self) -> bool {
        self.failed.is_none() && !self.empty
    }
}

/// One judge's independent ranking.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Judgement {
    /// Judge seat number, 1-based.
    pub judge: usize,
    /// Seat key.
    pub seat: String,
    /// Agent occupying the seat.
    pub agent: String,
    /// Best-first labels.
    #[serde(default)]
    pub ranking: Vec<char>,
    /// Per-label justification.
    #[serde(default)]
    pub reasons: BTreeMap<String, String>,
    /// Self-reported confidence.
    #[serde(default)]
    pub confidence: Option<u8>,
    /// Per-label strengths this judge attributed to candidates it did not
    /// rank first; the only input of the graft phase.
    #[serde(default)]
    pub strengths: BTreeMap<String, Vec<String>>,
    /// Order the candidates were presented in, as candidate indices.
    #[serde(default)]
    pub order: Vec<usize>,
    /// Why this judge has no ranking.
    #[serde(default)]
    pub failed: Option<String>,
    /// Wall-clock time.
    #[serde(default)]
    pub duration_ms: u64,
}

/// One judge's turn in a deliberation round.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeliberationTurn {
    /// Judge seat number, 1-based.
    pub judge: usize,
    /// Agent occupying the seat.
    pub agent: String,
    /// The argument, as written.
    pub body: String,
    /// Where the judge stood at the end of the turn.
    #[serde(default)]
    pub tentative: Option<char>,
}

/// A deliberation round.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeliberationRound {
    /// 1-based round number.
    pub round: usize,
    /// Turns, in the order they were taken.
    pub turns: Vec<DeliberationTurn>,
}

/// A final vote, collected privately.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VoteRecord {
    /// Judge seat number, 1-based.
    pub judge: usize,
    /// Agent occupying the seat.
    pub agent: String,
    /// The vote.
    #[serde(default)]
    pub vote: Option<char>,
    /// Why.
    #[serde(default)]
    pub reason: String,
    /// Did this judge move from its initial first choice?
    #[serde(default)]
    pub changed: bool,
}

/// A seat that was taken out by a CLI rate limit / quota, recorded so a run
/// whose panel collapsed does not masquerade as a healthy one.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct QuotaLoss {
    /// Seat key, e.g. `judge-1` or `review-2`.
    pub seat: String,
    /// Node that was running, e.g. `judge`, `vote`, `review`.
    pub node: String,
    /// When the CLI reported the limit.
    pub at: Timestamp,
    /// Reset hint if the CLI printed one, free text.
    #[serde(default)]
    pub reset: Option<String>,
}

/// A seat handed from one roster agent to the next after the first failed it
/// (rate limit, timeout or an ordinary failure), recorded so the answer's
/// author is never a mystery and a run that recovered on its second agent
/// still says what it cost.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Handover {
    /// When the seat changed hands.
    pub at: Timestamp,
    /// Node that was running, e.g. `implement`, `judge`, `review`.
    pub node: String,
    /// Seat key, e.g. `impl-A` or `judge-2`.
    pub seat: String,
    /// Agent id that failed the seat.
    pub from: String,
    /// Agent id that took it over.
    pub to: String,
    /// Why: `rate limited (quota)`, `timed out`, or the failure's own message.
    pub reason: String,
}

/// A newly created lockfile of a package manager the directory does not use,
/// which a rescue commit left untracked instead of committing. Recorded so the
/// omission is visible: the file may be one the task really wanted.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Withheld {
    /// Repo-relative path.
    pub path: String,
    /// The manager the file belongs to.
    pub manager: String,
    /// The tracked file (or missing manifest) that made it foreign.
    pub kept_by: String,
    /// Node whose rescue commit withheld it.
    pub node: String,
    /// When.
    pub at: Timestamp,
}

/// The mechanical count.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Tally {
    /// First-choice votes per label.
    pub first_choice: BTreeMap<char, usize>,
    /// Borda points from the initial rankings, used only to break a tie.
    pub borda: BTreeMap<char, usize>,
    /// The winning label.
    pub winner: char,
    /// How many judges produced a usable ranking. A panel of one is not a
    /// consensus and must not be reported as a split.
    #[serde(default)]
    pub rankings: usize,
    /// Did every judge's *initial* first choice agree?
    pub unanimous_initial: bool,
    /// Was deliberation run?
    pub deliberated: bool,
    /// Judges who moved between their initial ranking and their final vote.
    pub changed_votes: usize,
    /// Did the final votes agree?
    pub unanimous_final: bool,
    /// How the tie was broken, when it had to be.
    #[serde(default)]
    pub tie_break: Option<String>,
    /// Configured judge count — the size of the full panel. `0` when no
    /// panel was asked (see `uncontested`), not the roster size a panel that
    /// never sat would have had.
    #[serde(default)]
    pub judges: usize,
    /// Judges who actually contributed to the decision (not taken out by a
    /// rate limit and producing a usable rank or vote).
    #[serde(default)]
    pub present: usize,
    /// How many judges are required for a trustworthy verdict. Chosen as a
    /// strict majority (`judges / 2 + 1`): a verdict backed by a minority must
    /// never be presented as a healthy one, while a bare majority is still
    /// real signal. A one-candidate run needs no quorum.
    #[serde(default)]
    pub quorum: usize,
    /// `present >= quorum`, or no quorum was required.
    #[serde(default)]
    pub met_quorum: bool,
    /// Why no panel was asked, when none was: a single viable candidate, or
    /// a review-only run that never competed. `None` when judges actually
    /// ranked and voted — including when too few of them survived to reach
    /// quorum, which is a collapse and must keep reading as one.
    #[serde(default)]
    pub uncontested: Option<String>,
}

/// One reviewer's report in a round.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReviewRecord {
    /// Reviewer seat number, 1-based.
    pub reviewer: usize,
    /// Agent occupying the seat.
    pub agent: String,
    /// Reviewer prose.
    #[serde(default)]
    pub summary: String,
    /// Findings, with magi-assigned ids.
    #[serde(default)]
    pub findings: Vec<Finding>,
    /// This seat's initial vote. `None` on a record predating votes, exactly
    /// like a round that genuinely had none cast — never a stand-in for a
    /// vote that was lost.
    #[serde(default)]
    pub vote: Option<ReviewVote>,
    /// Why this reviewer produced nothing.
    #[serde(default)]
    pub failed: Option<String>,
    /// Wall-clock time.
    #[serde(default)]
    pub duration_ms: u64,
    /// How many times this seat was asked before it settled — 0 for a first
    /// answer, N after N nudges (`ask_json_wave`'s retry loop). Read this
    /// together with [`Self::failed`], never `failed` alone: `failed: Some(_)`
    /// with `attempts == 0` is a seat that never answered at all, while
    /// `failed: None` with `attempts > 0` is one that only came back after a
    /// nudge — recovered, not silent — and the two must not look the same in
    /// history. A record written before this field existed defaults to `0`,
    /// which under-reports a pre-existing retry rather than inventing one;
    /// see `ask_json_wave`'s own doc for where this is filled in.
    #[serde(default)]
    pub attempts: usize,
}

/// One seat's revote during a round's reconsideration (see
/// [`ReviewRound::reconsideration`]).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReviewRevoteRecord {
    /// Reviewer seat number, 1-based.
    pub reviewer: usize,
    /// Agent occupying the seat.
    pub agent: String,
    /// The revote. `None` when the seat did not answer.
    #[serde(default)]
    pub vote: Option<ReviewVote>,
    /// Why.
    #[serde(default)]
    pub reason: String,
    /// Why this seat produced no revote.
    #[serde(default)]
    pub failed: Option<String>,
}

/// Post-repair e2e evidence. An empty, unsuccessful record is a durable
/// requirement saved before the fixer can change the winner.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GateRepairE2e {
    /// Exact winner SHA requiring or receiving post-repair verification.
    pub head: String,
    /// All configured commands completed successfully on this SHA.
    pub passed: bool,
    /// Last command results; empty before verification starts.
    pub outcomes: Vec<CommandOutcome>,
}

/// Superseded head-specific evidence; repair budgets remain on the run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VerificationEvidence {
    /// Replacement SHA, unknown when archived without a head change.
    #[serde(default)]
    pub superseded_by: Option<String>,
    /// Archive time, unknown for records written before this field existed.
    #[serde(default)]
    pub archived_at: Option<Timestamp>,
    /// Run-wide gate-fix rounds spent at archival, unknown for legacy records.
    #[serde(default)]
    pub fixes_spent: Option<usize>,
    /// SHA of the superseded gate, unknown for legacy records.
    pub gate_head: Option<String>,
    /// Whether the superseded gate completed.
    pub gate_ran: bool,
    /// Original command results, including failures.
    pub gate: Vec<CommandOutcome>,
    /// Superseded verification independent of the review judgment.
    pub e2e: Option<GateRepairE2e>,
}

/// One fix round spent on a failing `verify.gate`: which fixer, what it did to
/// the tree, and the gate output it was shown.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GateFixRecord {
    /// Agent that applied the fix.
    pub agent: String,
    /// The gate output the fixer was handed, tail-truncated.
    pub failed: Vec<CommandOutcome>,
    /// What the fixer said it changed.
    #[serde(default)]
    pub notes: String,
    /// Did the fix produce a commit?
    #[serde(default)]
    pub committed: bool,
    /// Why the fix step produced nothing.
    #[serde(default)]
    pub error: Option<String>,
}

/// One fixer round spent on a rebase that stopped on a conflict: which fixer,
/// what it was shown and whether git says the rebase then finished.
///
/// The records at each sequencer stop count against `graph.review_rounds`,
/// written to disk *before* the fixer runs so a park or crash cannot hand the
/// round back.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RebaseFixRecord {
    /// Agent asked to resolve the conflict.
    pub agent: String,
    /// Paths git reported unmerged when the round started.
    pub paths: Vec<String>,
    /// The branch tip the rebase started from, so a resumed run can tell a
    /// worktree still holding those files from one with edits of its own.
    #[serde(default)]
    pub from: Option<String>,
    /// Did git report the rebase finished after the round?
    #[serde(default)]
    pub finished: bool,
    /// Why the round produced nothing (agent failure, quota).
    #[serde(default)]
    pub error: Option<String>,
    /// The commit the sequencer was replaying (`REBASE_HEAD`) when the round
    /// started: which stop point the round belongs to. A record without it
    /// predates launcher staging.
    #[serde(default)]
    pub stop: Option<String>,
    /// Staged changes outside `paths` when the round started (Git's own
    /// clean applications), so edits made behind them can be told apart. Retries
    /// retain the first baseline for the stop; legacy baselines are rebuilt
    /// from the original three-way merge and verified before being saved.
    #[serde(default)]
    pub index_outside: Option<String>,
    /// How far the launcher got with staging and continuing this stop point.
    #[serde(default)]
    pub launcher: Option<LauncherStep>,
    /// A denied or failed Git metadata write: infrastructure, not a conflict.
    /// The rebase is kept standing and no further round is spent.
    #[serde(default)]
    pub infra: Option<String>,
}

/// Where launcher-owned staging of a resolved conflict stopped.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum LauncherStep {
    /// About to stage the validated paths (or doing it).
    Staging,
    /// Every conflicted path is staged; the sequencer was not continued yet.
    Staged,
    /// `rebase --continue` was started.
    Continuing,
}

/// How the graft phase ended (see `graph::Runner::graft`).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum GraftOutcome {
    /// The fixer's result was kept.
    Applied,
    /// Started but not finished: a record in this state on load means the
    /// process stopped mid-round and the winner is restored on resume.
    Running,
    /// The seat could not start; restore `pre` under coordination, then retry.
    WaitingForRestore,
    /// The winner's pre-graft commit was restored, with the reason.
    RolledBack(String),
}

impl GraftOutcome {
    /// Is the record final, i.e. must a resume leave it alone?
    pub fn settled(&self) -> bool {
        !matches!(
            self,
            GraftOutcome::Running | GraftOutcome::WaitingForRestore
        )
    }
}

/// The graft phase's record: what ran, on which commits, and how it ended.
/// `pre` and a `Running` outcome are saved before the fixer is called.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GraftRecord {
    /// The winner's commit before the graft; restored on a rollback.
    pub pre: String,
    /// The winner's commit after an applied graft.
    #[serde(default)]
    pub post: Option<String>,
    /// Changed lines (added + removed) the graft introduced, when measured.
    #[serde(default)]
    pub lines: Option<usize>,
    /// Historical size limit, retained for old records only. New grafts use `None`.
    #[serde(default)]
    pub limit: Option<usize>,
    /// Agent that answered the round.
    #[serde(default)]
    pub agent: Option<String>,
    /// How it ended.
    pub outcome: GraftOutcome,
    /// The strengths the fixer was asked to port, with the ids the prompt
    /// gave them. Empty on a record written before they were kept.
    #[serde(default)]
    pub requested: Vec<GraftStrength>,
    /// What the fixer reported it did with them; `None` when it gave no usable
    /// report (see `adoption_error`) or the record predates this field.
    #[serde(default)]
    pub adoption: Option<GraftAdoption>,
    /// Why `adoption` is absent although the fixer answered.
    #[serde(default)]
    pub adoption_error: Option<String>,
}

/// One strength a judge attributed to a losing candidate, as the graft
/// prompt listed it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GraftStrength {
    /// Id in the prompt, e.g. `A1`.
    pub id: String,
    /// The candidate it was observed in.
    pub source: char,
    /// The strength, already blind-sanitized.
    pub text: String,
}

/// The fixer's own account of a graft round. It is a claim, not a check: the
/// diff is what actually changed.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct GraftAdoption {
    /// Ids (or texts) the fixer says it ported.
    #[serde(default)]
    pub addressed: Vec<String>,
    /// Ids (or texts) it declined, with the reason.
    #[serde(default)]
    pub rejected: Vec<GraftRejection>,
    /// The report's `notes`.
    #[serde(default)]
    pub notes: String,
    /// The reasoning that preceded the report.
    #[serde(default)]
    pub reasoning: String,
}

/// A strength the fixer declined.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GraftRejection {
    /// Id or text as the fixer wrote it.
    pub id: String,
    /// Its argument.
    #[serde(default)]
    pub why: String,
}

/// The fixer's response to a round.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FixRecord {
    /// Agent that applied the fixes.
    pub agent: String,
    /// Finding ids acted on.
    #[serde(default)]
    pub addressed: Vec<String>,
    /// Findings declined, with reasons.
    #[serde(default)]
    pub rejected: Vec<Rejection>,
    /// What changed.
    #[serde(default)]
    pub notes: String,
    /// Did the fix produce a commit?
    #[serde(default)]
    pub committed: bool,
    /// Why the fix step produced nothing.
    #[serde(default)]
    pub failed: Option<String>,
    /// Wall-clock time.
    #[serde(default)]
    pub duration_ms: u64,
    /// How the fixer's own seat was made to answer when its CLI turn ended
    /// cleanly but without an addressed/rejected report — see
    /// [`graph::Runner::continue_fix_report`]. `None` for a record written
    /// before this existed, which must read as "unknown", not as
    /// [`ContinuationOutcome::NotNeeded`]: an old run really may have hit
    /// this exact gap and simply had no mechanism to say so.
    #[serde(default)]
    pub continuation: Option<ContinuationRecord>,
}

/// How a node recovered — or failed to recover — a structured report after
/// the CLI's own turn ended cleanly (a usable, non-empty, exit-0 reply)
/// without it. A clean CLI turn is not the same fact as the node's own work
/// being done — see the `fix` node's `continue_fix_report`, which is what
/// produces this.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum ContinuationOutcome {
    /// The first reply already carried the report; nothing was resumed.
    NotNeeded,
    /// A follow-up call in the same session recovered the report.
    Resumed,
    /// The continuation budget was spent without ever recovering it.
    Exhausted,
    /// A continuation attempt hit the CLI's rate limit; not retried further
    /// — a quota fails the same way again immediately.
    QuotaLost,
    /// No session was left to resume into, so nothing was attempted.
    NoSession,
}

/// Cost and outcome of one node's attempt to recover a missing report by
/// resuming its own seat.
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct ContinuationRecord {
    /// Follow-up calls made to the same seat. `0` when the outcome is
    /// [`ContinuationOutcome::NotNeeded`] or [`ContinuationOutcome::NoSession`].
    pub attempts: usize,
    /// Wall-clock time spent on those follow-up calls, summed — not counting
    /// the original call whose reply this is recovering from.
    pub cumulative_wait_ms: u64,
    /// What ended the loop.
    pub outcome: ContinuationOutcome,
}

impl ContinuationRecord {
    /// The report was already there on the first try.
    pub fn not_needed() -> Self {
        Self {
            attempts: 0,
            cumulative_wait_ms: 0,
            outcome: ContinuationOutcome::NotNeeded,
        }
    }
}

/// What happened to one operator-selected finding after the fixer ran, as
/// part of an [`OperatorFixRequest`].
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OperatorFixOutcome {
    /// The request has not run yet, or never got far enough to report.
    #[default]
    Pending,
    /// The fixer's adoption report named this finding as addressed.
    Addressed,
    /// The fixer's adoption report declined it, with an argument.
    Rejected {
        /// The fixer's own reason.
        why: String,
    },
    /// The fixer never delivered a usable adoption report at all — a
    /// dropped stream, a quota hit, or a continuation budget spent without
    /// recovering one (see `graph::Runner::continue_fix_report`). Distinct
    /// from `Rejected`, which needs an argument this never produced, and
    /// never written back as "addressed" or silently left `Pending` — a
    /// gap in the report is its own outcome, not evidence either way about
    /// the finding.
    Unreported,
}

/// One finding an operator selected for [`OperatorFixRequest`], with the
/// provenance a reviewer originally gave it, copied here verbatim.
///
/// Severity and vote are snapshots, never recomputed and never treated as
/// blocking just because an operator picked the finding — only
/// [`Severity::blocks`] on the original [`ReviewRecord`] decides that. This
/// type exists so an operator's selection is an auditable *addition* to the
/// record, not a rewrite of what a reviewer actually said.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OperatorFixFinding {
    /// Finding id, e.g. `R2-1-3`.
    pub id: String,
    /// Severity as the reviewer recorded it.
    pub severity: Severity,
    /// The reviewer seat's overall vote for the round this finding came
    /// from, if one was cast.
    #[serde(default)]
    pub reviewer_vote: Option<ReviewVote>,
    /// Review round the finding was raised in.
    pub round: usize,
    /// That round's own head — the commit the finding was actually raised
    /// against, used for the freshness check against the branch's current
    /// head at request time.
    pub round_head: String,
    /// Reviewer seat number, 1-based.
    pub reviewer: usize,
    /// Agent occupying that seat.
    pub agent: String,
    /// File the finding concerns.
    #[serde(default)]
    pub file: Option<String>,
    /// Line the finding concerns.
    #[serde(default)]
    pub line: Option<u32>,
    /// One-line summary.
    pub title: String,
    /// The argument.
    #[serde(default)]
    pub detail: String,
    /// What happened to this finding after the fixer ran.
    #[serde(default)]
    pub outcome: OperatorFixOutcome,
}

/// One `magi fix` invocation: the operator's own record of which
/// already-recorded findings they routed to a fixer, why, and what came
/// back. See [`SCHEMA`]'s doc for schema 9 on why this is a channel of its
/// own rather than a field on [`ReviewRound`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OperatorFixRequest {
    /// When `magi fix` was invoked.
    pub requested_at: Timestamp,
    /// The operator's own reasoning. Required and never empty at the CLI —
    /// the audit trail this feature exists for.
    pub reason: String,
    /// The findings selected, each with its own provenance and outcome.
    pub findings: Vec<OperatorFixFinding>,
    /// The branch's head at the moment this request started executing.
    pub head_at_request: String,
    /// Did the operator pass `--allow-stale`?
    pub allow_stale: bool,
    /// Did any selected finding's own `round_head` differ from
    /// `head_at_request`? Kept distinct from `allow_stale` — flipping that
    /// flag does not retroactively make a request that was actually fresh
    /// read as stale, or the reverse.
    pub stale: bool,
    /// The fixer's own attempt, once dispatched.
    #[serde(default)]
    pub fix: Option<FixRecord>,
    /// Head after the fixer's commit, when it produced one.
    #[serde(default)]
    pub result_head: Option<String>,
    /// The review-only run opened to re-verify the change, when one was
    /// actually committed. `None` when nothing changed, so there was
    /// nothing new to re-review — never left implicit as "not gotten to
    /// yet".
    #[serde(default)]
    pub follow_up_review_run: Option<String>,
}

/// A command a seat's own CLI reported running, kept for `magi show` and for
/// telling "this seat's turn ended" apart from "the process it started is
/// done" — see `agent::CommandEvidence`, which is the only source this is
/// ever built from. Never something magi polled or supervised; a command the
/// CLI never reported finishing (or a CLI this crate has no adapter for at
/// all) simply has no entry here, which must read as "unknown", not as
/// "nothing ran".
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobRecord {
    /// Graph node the seat belongs to, e.g. `"implement"`, `"fix"`.
    pub node: String,
    /// Review round this job belongs to, for a `"review"`/`"fix"` node —
    /// `None` for every other node, where rounds do not apply, and for every
    /// record written before this was tracked. Lets a reader ask "what did
    /// this seat itself actually run this round", distinct from and never
    /// substituted for magi's own recorded `ReviewRound::e2e` — an absent
    /// entry here means unobserved, not that nothing ran (see this type's
    /// own doc).
    #[serde(default)]
    pub round: Option<usize>,
    /// Seat key, e.g. `"impl-A"`.
    pub seat: String,
    /// The CLI's own id for this command.
    pub id: String,
    /// The command itself, as the CLI reported it.
    pub description: String,
    /// When this evidence was captured — the moment this seat's reply
    /// carrying it was read, not the command's own start time, which no
    /// adapter here currently has. A lower bound on staleness only.
    pub checked_at: Timestamp,
    /// What the CLI reported for it.
    pub status: JobStatus,
    /// Exit code the CLI reported.
    pub exit_code: Option<i32>,
    /// Tail of the command's own output, when reported.
    #[serde(default)]
    pub result_summary: String,
    /// Which CLI/event stream this came from, e.g. `"codex"`.
    pub source: String,
}

/// What a [`JobRecord`]'s own CLI reported for it. There is no `Running`
/// variant: nothing here is ever polled live, so "still running" and
/// "finished but never reported" are the same absence of evidence, not a
/// state this type can name — see [`JobRecord`]'s own doc.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum JobStatus {
    /// The command's own reported exit code was `0`.
    Completed,
    /// The command's own reported exit code was non-zero.
    Failed,
    /// The CLI reported this command but not a readable exit code.
    Unknown,
}

/// Outcome of one shell command.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommandOutcome {
    /// The command, as configured.
    pub command: String,
    /// Exit code, `None` on timeout or signal.
    pub code: Option<i32>,
    /// Tail of the combined output, for the report and the fix prompt.
    #[serde(default)]
    pub output_tail: String,
    /// Wall-clock time.
    #[serde(default)]
    pub duration_ms: u64,
    /// Set only by magi itself, never inferred from `output_tail`: the
    /// configured command was never actually run because a resource it
    /// needs — right now, only the shared build cache's lease or the
    /// freshness check that must precede using it — was not available
    /// within budget. Distinct from an ordinary failure or timeout (both of
    /// which *did* run something and are evidence about the patch); this is
    /// evidence about the machine, and must never be read as a verdict on
    /// the tree it named. `#[serde(default)]` so every record written
    /// before this field existed keeps reading as `false` — exactly what it
    /// was.
    #[serde(default)]
    pub resource_blocked: bool,
}

/// Substrings that mark a Cargo/rustc/link failure: the toolchain could not
/// produce a binary to run at all, as opposed to producing one that ran and
/// failed. A Windows link race against a shared `CARGO_TARGET_DIR` (see
/// AGENTS.md, "Running magi on magi") looks exactly like a red command
/// otherwise, and a run has concluded `Blocked` on nothing but that race.
const BUILD_FAILURE_MARKERS: &[&str] = &[
    "error: could not compile",
    "error: linking with",
    "LINK : fatal error",
    "fatal error LNK",
];

impl CommandOutcome {
    /// Did it pass?
    pub fn ok(&self) -> bool {
        self.code == Some(0)
    }

    /// Did this command fail because the code could not be built or linked,
    /// rather than because it ran and produced a wrong result? A failure here
    /// is not a verdict on the patch under review.
    pub fn build_failed(&self) -> bool {
        !self.ok()
            && BUILD_FAILURE_MARKERS
                .iter()
                .any(|m| self.output_tail.contains(m))
    }
}

/// One review + verify + fix round.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReviewRound {
    /// 1-based round number.
    pub round: usize,
    /// Commit the round reviewed.
    pub head: String,
    /// The commit `e2e` was actually attempted against. Set whenever an
    /// attempt was dispatched (`e2e_status()` reads `Passed`, `Failed`, or
    /// `ResourceBlocked`), naming that commit even when it equals `head` —
    /// never left implicit, because an implicit "must have been `head`" is
    /// exactly what let a later round quote an earlier round's result
    /// without saying which commit it came from. A resource-blocked attempt
    /// still targeted a specific commit even though no command finished, and
    /// leaving that unrecorded is exactly what made a *fresh* blocked
    /// attempt read the same as an untracked one from before schema 8.
    /// `None` only when nothing was attempted at all (`NotConfigured`,
    /// `Deferred`). See `SCHEMA`'s doc for schema 8 for why this broadened
    /// from only the catch-up-on-a-different-commit case.
    #[serde(default)]
    pub verified_head: Option<String>,
    /// When the attempt behind `verified_head` actually ran. `None` on
    /// every record written before schema 8, and on a round where nothing
    /// ran —
    /// both read as "unknown", not as "now" or "never asked".
    #[serde(default)]
    pub verified_at: Option<Timestamp>,
    /// Reviewer reports.
    pub reviews: Vec<ReviewRecord>,
    /// E2E command outcomes for this round.
    #[serde(default)]
    pub e2e: Vec<CommandOutcome>,
    /// True when the first verify attempt this round could not build or
    /// link, and `e2e` above holds a second attempt run before concluding.
    /// A run must never be decided on a red it could not tell from an
    /// unrelated build race.
    #[serde(default)]
    pub verify_retried: bool,
    /// True when `e2e` was intentionally left empty this round: the round
    /// already had blocking findings and another round was available, so
    /// `graph::Runner::review_loop` sent the fixer straight at them instead
    /// of spending a full verify run on a head it already knew would need
    /// another fix. Distinct from an `e2e` that is simply empty because
    /// `verify.e2e` has no commands configured — `e2e.is_empty()` alone
    /// cannot tell those apart, and conflating them is exactly how a
    /// deferred check would get painted green. A record written before this
    /// field existed defaults to `false`, which is the truth for it: every
    /// round used to run e2e unconditionally.
    #[serde(default)]
    pub e2e_deferred: bool,
    /// Why `e2e` was deferred, set only when [`Self::e2e_deferred`] is true.
    /// Carried to the fixer's prompt and shown in the report so "deferred"
    /// never reads as silence.
    #[serde(default)]
    pub e2e_defer_reason: Option<String>,
    /// Fixer response, absent when the round was already clean.
    #[serde(default)]
    pub fix: Option<FixRecord>,
    /// Findings that hold the merge.
    #[serde(default)]
    pub blocking: usize,
    /// Reviewer seats that answered (did not time out, crash, or return
    /// something unparsable).
    #[serde(default)]
    pub answered: usize,
    /// Reviewer seats the round expected an answer from — normally
    /// `graph.reviewers`, but recorded per round so a config change between
    /// runs never has to be inferred from history.
    #[serde(default)]
    pub expected: usize,
    /// Round ended with no blocking findings and green verification, judged
    /// against the seats that answered. See [`Self::incomplete`] for whether
    /// that verdict is missing input.
    #[serde(default)]
    pub clean: bool,
    /// Did the tree actually move against `base` this round, comparing the
    /// diff after the fix to the diff the reviewers saw at the start of the
    /// round?
    ///
    /// Never derived from the fixer's own `addressed`/`rejected` count: that
    /// self-report has been caught lying twice on this workload (runs `b455`
    /// and `6218`, both of which committed a real, substantial diff while
    /// reporting `0 addressed`). `git` does not lie about whether the tree
    /// changed, so this is what `graph::Runner::review_loop` counts rounds of
    /// no progress against. Absent on a round with no fix attempt (already
    /// clean, or the round the budget ran out on), where it defaults to
    /// `false` and is not consulted.
    #[serde(default)]
    pub progressed: bool,
    /// Did the seats' initial votes ([`ReviewRecord::vote`]) disagree?
    #[serde(default)]
    pub vote_split: bool,
    /// One round of revoting, run only when `vote_split`: each seat that cast
    /// an initial vote reads every seat's findings and votes, then revotes.
    /// Empty when the initial votes already agreed, the same as a solo
    /// candidate leaving `deliberation` empty.
    #[serde(default)]
    pub reconsideration: Vec<ReviewRevoteRecord>,
    /// The round's verdict: the most cautious vote among the seats that
    /// answered, using each seat's revote where reconsideration ran and its
    /// initial vote otherwise. `None` when no seat produced a usable vote —
    /// including every record written before votes existed, which is the
    /// truth for those rounds, not a gap in this one.
    #[serde(default)]
    pub verdict: Option<ReviewVote>,
}

/// A merge-approval question and the head commit it was asked about.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LandApproval {
    /// Id of the question that was filed for `head`.
    pub question: String,
    /// The commit the question's panel was built from.
    pub head: String,
}

/// Independent public description of a run's whole change from the final diff.
/// Valid only for its (`head`, `base`, `evidence`) snapshot; a changed commit,
/// generation policy, verification result or finding requires regeneration.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChangeSummary {
    /// The final commit the diff was read at.
    pub head: String,
    /// The merge base of the resolved remote base and `head`.
    pub base: String,
    /// Generation specification and verification/finding evidence fingerprint.
    /// Missing on older records, which must be regenerated.
    #[serde(default)]
    pub evidence: String,
    /// Title, sanitized.
    pub title: String,
    /// Body, sanitized, without the run marker.
    pub body: String,
}

/// The owner question filed when the posting gate withheld a pull request
/// title or description, and how it ended. One record per rejected text
/// (`fingerprint`); see `github_text` for the flow.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GithubTextAsk {
    /// Hash of the rejected original title and body.
    pub fingerprint: String,
    /// The title is withheld.
    pub title: bool,
    /// The description is withheld.
    pub body: bool,
    /// Rule categories that fired.
    pub categories: Vec<String>,
    /// The question currently standing, if one was filed.
    #[serde(default)]
    pub question: Option<String>,
    /// Questions filed so far (at most 2: the second follows a bad reply).
    #[serde(default)]
    pub asks: u32,
    /// The decision has been made (saved before the pull request is opened,
    /// so it is applied once and survives a restart).
    #[serde(default)]
    pub resolved: bool,
    /// With `resolved`: the owner's vetted replacement title; `None` means the
    /// neutral text for every withheld field.
    #[serde(default)]
    pub chosen_title: Option<String>,
    /// With `resolved`: which way the question was settled. `None` on a record
    /// written before this field existed (read as the neutral text).
    #[serde(default)]
    pub choice: Option<GithubTextChoice>,
    /// Final rejected rewrite, kept locally for display and an explicit override.
    #[serde(default)]
    pub rewrite: Option<GithubTextRewrite>,
}

/// A usable rewrite that still failed the posting gate.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GithubTextRewrite {
    /// Final title after redaction.
    pub title: String,
    /// Final description after redaction and run marker insertion.
    pub body: String,
    /// Stable rule categories, never sensitive values.
    pub failed: Vec<String>,
}

/// How a withheld pull request text was settled.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum GithubTextChoice {
    /// The changed text: neutral for language, redacted for sensitive data
    /// (also silence, abandonment and an unknown answer).
    Fallback,
    /// The owner's vetted replacement title (`chosen_title`).
    MyTitle,
    /// The original text of every withheld field, past the posting gate.
    Original,
    /// The saved rewrite title and body, unmodified, past the posting gate.
    Rewrite,
}

/// A review hand-off the panel did not agree to: findings that hold the merge
/// are still open and at least one seat's final vote was reject. Recorded on
/// [`RunState::contested_handoff`] so `land` can name them in its question.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ContestedHandoff {
    /// Open findings of severity Major or above, in report order.
    pub findings: Vec<Finding>,
    /// Seats (1-based) whose final vote was reject, with the agent in the seat.
    pub rejecters: Vec<(usize, String)>,
}

pub use crate::review_repair::{Phase as MergeRepairPhase, Request as MergeRepair};

impl ReviewRound {
    /// Each answering seat's final vote: its revote where reconsideration ran
    /// and answered, its initial vote otherwise. The single place this is
    /// decided, shared by the round's verdict and [`Self::contested_handoff`].
    pub fn final_votes(&self) -> Vec<(usize, String, ReviewVote)> {
        self.reviews
            .iter()
            .filter_map(|r| {
                self.reconsideration
                    .iter()
                    .find(|rv| rv.reviewer == r.reviewer)
                    .and_then(|rv| rv.vote)
                    .or(r.vote)
                    .map(|v| (r.reviewer, r.agent.clone(), v))
            })
            .collect()
    }

    /// Is this round's hand-off contested: a Major-or-above finding open and
    /// at least one final vote of reject? `None` otherwise. Kept as one small
    /// function so anything deciding on it reads the same answer.
    pub fn contested_handoff(&self) -> Option<ContestedHandoff> {
        let findings: Vec<Finding> = self
            .reviews
            .iter()
            .flat_map(|r| r.findings.iter())
            .filter(|f| f.severity.blocks())
            .cloned()
            .collect();
        if findings.is_empty() {
            return None;
        }
        let rejecters: Vec<(usize, String)> = self
            .final_votes()
            .into_iter()
            .filter(|(_, _, v)| *v == ReviewVote::Reject)
            .map(|(seat, agent, _)| (seat, agent))
            .collect();
        if rejecters.is_empty() {
            return None;
        }
        Some(ContestedHandoff {
            findings,
            rejecters,
        })
    }

    /// Did at least one reviewer seat fail to answer this round?
    pub fn incomplete(&self) -> bool {
        self.answered < self.expected
    }

    /// The honest state of this round's e2e leg.
    ///
    /// Never derive this from `e2e.is_empty()` alone anywhere else in the
    /// codebase — `NotConfigured` and `Deferred` both leave it empty, and
    /// only this method (backed by [`Self::e2e_deferred`]) tells them apart.
    /// A resource-blocked attempt is checked first and ahead of both: `e2e`
    /// is non-empty for it too, but `CommandOutcome::resource_blocked` says
    /// no command actually ran, and reading that as `Failed` is exactly how
    /// shared build-cache contention gets misreported as a verdict on the
    /// patch (see `CommandOutcome::resource_blocked`'s own doc).
    pub fn e2e_status(&self) -> E2eStatus {
        if self.e2e.iter().any(|o| o.resource_blocked) {
            E2eStatus::ResourceBlocked
        } else if !self.e2e.is_empty() {
            if self.e2e.iter().all(CommandOutcome::ok) {
                E2eStatus::Passed
            } else {
                E2eStatus::Failed
            }
        } else if self.e2e_deferred {
            E2eStatus::Deferred
        } else {
            E2eStatus::NotConfigured
        }
    }

    /// Facts about this round's verification leg, judged against
    /// `current_head` — the commit whoever is asking is actually looking at
    /// right now. `None` when there is nothing worth surfacing: no
    /// `verify.e2e` configured, or the round's own check came back green (a
    /// passing result needs no skepticism attached to it, and an unread
    /// `None` is exactly what keeps a quiet round quiet instead of padding
    /// every prompt with "everything was fine").
    ///
    /// This is the single place that turns `e2e`/`e2e_deferred`/
    /// `verified_head`/`verified_at` into text. Every prompt and report that
    /// shows a round's verification result must build its wording from this,
    /// not re-derive its own summary at the call site — a hand-rolled
    /// version at one more place is exactly how "an old red read as today's
    /// answer" comes back through a different door (see the incident this
    /// type exists to prevent, recorded alongside `SCHEMA`'s doc for schema
    /// 8).
    pub fn verification_summary(&self, current_head: &str) -> Option<VerificationSummary> {
        let status = self.e2e_status();
        if matches!(status, E2eStatus::NotConfigured | E2eStatus::Passed) {
            return None;
        }
        let commit = match &self.verified_head {
            Some(h) if h == current_head => {
                format!("commit {} (this is the head being looked at now)", short(h))
            }
            Some(h) => format!("commit {} (an earlier head, since superseded)", short(h)),
            None => "commit unknown (no command finished checking one)".to_owned(),
        };
        let checked_at = match self.verified_at {
            Some(t) => format!("checked at {t}"),
            None => "checked at: unknown (recorded before this was tracked)".to_owned(),
        };
        let result = match status {
            E2eStatus::NotConfigured | E2eStatus::Passed => unreachable!("checked above"),
            E2eStatus::Failed => "result: FAILED".to_owned(),
            E2eStatus::Deferred => format!(
                "result: not run this round yet — deferred to the fixer{}. Not passed, not \
                 failed.",
                self.e2e_defer_reason
                    .as_deref()
                    .map(|why| format!(" ({why})"))
                    .unwrap_or_default()
            ),
            E2eStatus::ResourceBlocked => "result: could not run — a required resource was \
                                            not available. This is evidence about the machine, \
                                            not about the patch."
                .to_owned(),
        };
        let label = format!("round {}, {commit}, {checked_at}\n{result}", self.round);
        // `Failed` names the command that actually ran and failed;
        // `ResourceBlocked` names the operation magi was waiting on (or the
        // freshness check it could not confirm) — `command` still says what
        // was attempted even though nothing finished, and leaving it out is
        // exactly how a reviewer or fixer lost the one thing this leg *can*
        // still tell them: what was being checked, not whether it passed.
        let tail = matches!(status, E2eStatus::Failed | E2eStatus::ResourceBlocked).then(|| {
            self.e2e
                .iter()
                .filter(|o| !o.ok())
                .map(|o| format!("$ {}\n{}\n", o.command, o.output_tail))
                .collect::<String>()
        });
        Some(VerificationSummary { label, tail })
    }
}

/// The honest state of a round's e2e leg. See [`ReviewRound::e2e_status`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum E2eStatus {
    /// `verify.e2e` has no commands configured.
    NotConfigured,
    /// Skipped this round on purpose: blocking findings already required a
    /// fix, so the round went straight to the fixer instead of spending a
    /// full verify run on a head it already knew would need another pass.
    Deferred,
    /// Ran, and every command exited 0.
    Passed,
    /// Ran, and at least one command did not exit 0.
    Failed,
    /// Magi could not even get a command to run — the shared build cache's
    /// lease or freshness check was not available within budget. Evidence
    /// about the machine, never a verdict on the tree it named; must not be
    /// shown or counted the same as [`Self::Failed`].
    ResourceBlocked,
}

/// [`ReviewRound::verification_summary`]'s output: the facts, pre-worded, for
/// a prompt or report to place under its own heading. Kept as two pieces
/// rather than one pre-joined string so a caller that wants to insert its own
/// note between the label and the raw command tail (see `prompt::review`) can
/// do so without re-parsing text back apart.
#[derive(Debug, Clone)]
pub struct VerificationSummary {
    /// Round, commit, freshness and result — always present.
    pub label: String,
    /// Raw `$ command` / output tail, present for `result: FAILED` and for
    /// a resource-blocked attempt (naming the operation magi was waiting on,
    /// even though nothing finished) — absent for every other result, which
    /// has nothing to add past the label.
    pub tail: Option<String>,
}

/// The honest state of a run's final gate. See [`RunState::gate_status`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GateStatus {
    /// Never attempted, or the last attempt was resource-blocked (the shared
    /// build cache could not be acquired or confirmed fresh in time) and
    /// needs a retry.
    NotRun,
    /// No gate commands were configured. This is not successful verification.
    NotConfigured,
    /// Ran one or more commands, and every one of them exited 0.
    Passed,
    /// Ran one or more commands, and at least one did not exit 0.
    Failed,
}

impl GateStatus {
    /// May a run in this state proceed to merge?
    pub fn ok(self) -> bool {
        matches!(self, Self::Passed)
    }
}

/// What happened to the winning branch.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MergeOutcome {
    /// Requested mode.
    pub mode: MergeMode,
    /// Did the requested operation succeed? Publication and no-op success do not prove a merge.
    pub ok: bool,
    /// Command output, or the command the operator should run.
    #[serde(default)]
    pub detail: String,
    /// The winner had no commits ahead of the base, so no pull request was
    /// attempted. Distinct from a `gh` failure: nothing was wrong with the
    /// tooling, there was simply nothing to land.
    #[serde(default)]
    pub empty: bool,
}

/// A seat currently mid-answer: a prompt was sent and no reply has landed yet.
///
/// This is not the whole story of "is it alive" — a daemon killed mid-wave
/// leaves its last wave's entries here forever, since nothing ran to clear
/// them. A reader must cross-check a live daemon's heartbeat
/// (`daemon::is_working_on`) before trusting one of these as "still running"
/// rather than "abandoned". [`RunState::clear_active`] is what keeps that
/// leftover from surviving into the next attempt at this run: `execute` calls
/// it before doing anything else, so a resumed run never carries a stale
/// entry into its own report before the next wave repopulates it.
///
/// Deliberately carries no agent id: an implementer's agent is no secret, but
/// a judge or reviewer seat is blind (`SeatState::key` is keyed by seat, never
/// agent, for exactly this reason), and this struct has no way to tell which
/// kind of seat it describes. The seat key alone — already in the map this
/// lives under — is what every caller needs to say which seat is running.
///
/// The same map also carries entries for shell-command work that runs
/// outside any seat — `verify.e2e`, `verify.gate` — keyed by the task's own
/// name (`"e2e"`, `"gate"`) rather than a seat key. [`Self::task`] is `Some`
/// only for those; it is how a reader tells the two kinds of entry apart
/// without a second map, a second route, or a second SSE reason to poll for
/// — see [`RunState::seats_active`] / [`RunState::tasks_active`] for the
/// accessors that split them back apart. A task entry is exactly as blind as
/// a seat entry: no agent runs it, so there is nothing to leak, and
/// [`Self::command`] carries only the shell command being run, never
/// anything about who is running it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ActiveSeat {
    /// Node the seat is answering for, e.g. `implement`, `judge`, `review`.
    /// For a task entry, the node the command list runs under (`verify`,
    /// `gate`).
    pub node: String,
    /// When this attempt — or, for a task entry, this one command — was
    /// started. A task entry's timer resets at every command boundary,
    /// because `verify.e2e` / `verify.gate` apply their timeout per command,
    /// not once across the whole list — see [`RunState::task_command`].
    pub started_at: Timestamp,
    /// The wall-clock budget for this attempt (a seat) or this one command
    /// (a task entry).
    pub timeout_secs: u64,
    /// 0 for the first ask, N for the Nth nudge or resume. For a task entry,
    /// 0 for the first pass over the command list, N for the Nth retry (see
    /// `run_e2e_with_retry`'s build/link retry).
    #[serde(default)]
    pub attempt: usize,
    /// `None` for a seat; `Some("e2e")` / `Some("gate")` for a running
    /// command-list task. This is the type tag that lets both kinds of entry
    /// share one map without a task ever being mistaken for a (blind) seat —
    /// see this struct's own doc. Omitted from JSON when absent (the common,
    /// seat case), rather than written out as a literal `null` on every one
    /// of a run's seat entries.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub task: Option<String>,
    /// The command currently running, task entries only. Never set on a
    /// seat entry — a seat has no command, only a prompt, and a prompt is
    /// not safe to show mid-run (see this struct's blindness note).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub command: Option<String>,
    /// 1-based position of [`Self::command`] within the task's command list.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub index: Option<usize>,
    /// Number of commands in the task's list.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub total: Option<usize>,
}

/// What a wave is waiting for before it may spawn its agents.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResourceWait {
    /// Node whose seats wait.
    pub node: String,
    /// What is awaited, e.g. `shared build cache`.
    pub kind: String,
    /// Who holds it, as far as known.
    pub holder: String,
    /// When the wait began.
    pub since: Timestamp,
    /// The wait's own bound, seconds (`[graph] timeout_resource_wait`).
    pub limit_secs: u64,
}

/// Why a resource operation stopped. Older records did not retain this evidence.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ResourceBlockCause {
    /// No structured evidence was saved.
    #[default]
    Unrecorded,
    /// The resource wait actually exhausted its own budget.
    Timeout {
        /// Measured wait duration.
        waited_secs: u64,
        /// Configured resource wait budget.
        limit_secs: u64,
    },
    /// Acquisition was cancelled, possibly by a known park request.
    Cancelled {
        /// Available only when the caller retained the park request.
        park: Option<ParkCause>,
    },
    /// An infrastructure error, rather than an elapsed wait.
    Other,
}

/// A resumable resource stop; earlier nodes or peers may already have run.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ResourceBlock {
    /// Node whose operation was blocked or interrupted.
    pub node: String,
    /// What stayed busy.
    pub kind: String,
    /// Structured evidence from the point where acquisition ended.
    #[serde(default)]
    pub cause: ResourceBlockCause,
    /// Why the operation ended, in words.
    pub detail: String,
    /// When it ended.
    pub at: Timestamp,
}

impl ResourceBlock {
    /// Shared CLI, queue-hold and Web wording. Never infer a timeout from a
    /// resource kind or a legacy record's configured budget.
    #[must_use]
    pub fn description(&self) -> String {
        let reason = match &self.cause {
            ResourceBlockCause::Timeout {
                waited_secs,
                limit_secs,
            } => format!("resource wait timed out after {waited_secs}s (limit {limit_secs}s)"),
            ResourceBlockCause::Cancelled { park: Some(cause) } => {
                format!("resource wait cancelled for parking ({})", cause.label())
            }
            ResourceBlockCause::Cancelled { park: None } => "resource wait cancelled".into(),
            ResourceBlockCause::Other => "resource operation blocked".into(),
            ResourceBlockCause::Unrecorded if self.detail == "the run is parking" => {
                "resource wait cancelled for parking (request source unrecorded)".into()
            }
            ResourceBlockCause::Unrecorded
                if self.detail == "the wait for the build cache was cancelled" =>
            {
                "resource wait cancelled (request source unrecorded)".into()
            }
            ResourceBlockCause::Unrecorded => "resource stop cause unrecorded".into(),
        };
        format!("{}: {}: {reason} ({})", self.node, self.kind, self.detail)
    }
}

impl ActiveSeat {
    /// Seconds since this attempt was sent.
    #[must_use]
    pub fn elapsed_secs(&self, now: Timestamp) -> i64 {
        (now.as_second() - self.started_at.as_second()).max(0)
    }

    /// Seconds left before this attempt's own timeout fires, floored at zero
    /// rather than going negative once the CLI has overrun its budget.
    #[must_use]
    pub fn remaining_secs(&self, now: Timestamp) -> i64 {
        (self.timeout_secs as i64 - self.elapsed_secs(now)).max(0)
    }
}

/// Whether a process is provably still driving a run, provably not, or
/// neither — see [`RunState::liveness`]. Serialized as a lowercase string
/// (`"live"` / `"dead"` / `"unknown"`) rather than a bool: a bool has no room
/// for "could not tell", and folding that case into either `true` or `false`
/// is exactly the wrong call for a display an operator uses to decide
/// whether to wait or to act — see the schema-10 field doc on
/// [`RunState::driver_pid`] for the report it used to produce instead.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Liveness {
    /// Proven: a daemon's heartbeat or a matching held execution lease claims
    /// the run, or (for legacy records) `driver_pid` answers
    /// alive under the same identity (`driver_started_at`) this run recorded
    /// for it.
    Live,
    /// Proven: no daemon claim, and an explicit exit or matching free execution
    /// lease, or `driver_pid` answers dead outright
    /// or it answers alive under a *different* identity than recorded — a
    /// pid the OS has since handed to an unrelated process is exactly as
    /// good as proof the original driver is gone (see
    /// [`RunState::driver_started_at`]'s own doc).
    Dead,
    /// Neither proven — no daemon claim, and either no `driver_pid` to ask,
    /// the platform could not answer for it, or a live pid with nothing (or
    /// nothing queryable) to corroborate its identity against. Never treated
    /// as `Dead`: see [`RunState::liveness_with`].
    Unknown,
}

/// A timestamped note about a node.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Event {
    /// When.
    pub at: Timestamp,
    /// Node name.
    pub node: String,
    /// What happened.
    pub message: String,
}

/// How far the winner's tree trailed the landing base, last time it was
/// checked, and what came of trying to close that gap.
///
/// Set by `graph::Runner::sync_to_base`, which runs before the review loop and
/// again before the gate: verifying against a tree that does not yet contain
/// the base's tip answers "green on the commit this run branched from", not
/// "green on what is about to land", and a merge on that answer can revert
/// whatever landed elsewhere while the run was thinking.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BaseSync {
    /// `<remote>/<base>` tip the tree was last checked against.
    pub tip: String,
    /// Commits `tip` was ahead of the tree at that check, before any rebase
    /// this round tried to close the gap. Zero means the tree already
    /// contained `tip`.
    pub behind: usize,
    /// Cumulative rebase attempts. The ordinary cap is `graph::BASE_SYNC_ROUNDS`;
    /// each explicit base recovery authorization may reserve one more.
    pub attempts: usize,
    /// What git said, if the most recent rebase attempt conflicted or could
    /// not be pushed. `Some` here is what makes a `Blocked` run read as
    /// "stopped on the base, not on review or the gate" - the rebase is not
    /// retried again while this is set; a person has to look.
    #[serde(default)]
    pub conflict: Option<String>,
    /// Set when the winner's whole change turned out to be on the base already
    /// under other commit ids (see [`crate::already`]). Present exactly when the
    /// run ended as [`RunStatus::AlreadyInBase`].
    #[serde(default)]
    pub already_in: Option<crate::already::Evidence>,
}

/// What the land loop saw last time it looked at the pull request.
///
/// Strings for `state` and `checks` on purpose: they are `gh`'s vocabulary, and
/// pinning them into an enum here would mean a new GitHub check conclusion
/// turns a readable status into a deserialisation error on a run someone is
/// trying to look at.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PrRecord {
    /// Pull request url.
    pub url: String,
    /// Pull request number.
    pub number: u64,
    /// `open`, `merged` or `closed`.
    pub state: String,
    /// `pending`, `green`, `red` or `unknown`.
    pub checks: String,
    /// Land round, 1-based, or 0 before the first fix.
    pub round: usize,
    /// Land round budget.
    pub rounds: usize,
    /// Checks that were red when the pull request merged anyway (the forge
    /// said they do not block it). Empty for a green merge, and for any run
    /// recorded before this field existed - so empty is not proof nothing was
    /// red.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub red_at_merge: Vec<String>,
}

/// What the post-merge release-bump step did, and whether it needs a human.
///
/// Recorded next to the `bump` events rather than instead of them: the events
/// are the narrative, this is what `magi show`, the status word and the
/// notification read without parsing prose. `status` stays `Merged` - the
/// merge did happen - so [`RunState::needs_attention`] is the one place that
/// says a merged run is not fully done.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReleaseBump {
    /// The release pull request, when one was opened.
    #[serde(default)]
    pub pr_url: Option<String>,
    /// The version it releases.
    #[serde(default)]
    pub version: Option<String>,
    /// Did enabling automerge on `pr_url` succeed?
    #[serde(default)]
    pub automerge_enabled: bool,
    /// GitHub refused automerge because CI had already finished (the pull
    /// request was in clean status), so magi merged it directly. Kept apart
    /// from `automerge_enabled`, which stays the plain fact it says it is.
    #[serde(default)]
    pub merged_directly: bool,
    /// `[release] mode = "local"`: no automerge was armed; the pull request is
    /// merged on the owner's approval and released by `magi serve`
    /// ([`crate::release_local`]).
    #[serde(default)]
    pub local: bool,
    /// What the local release did (tag, commit, each command's exit code and
    /// output tail), kept here because the watcher's own record is deleted
    /// once the release is done.
    #[serde(default)]
    pub release: Option<crate::release_local::Job>,
    /// What went wrong, verbatim from the tool that said it.
    #[serde(default)]
    pub problem: Option<String>,
    /// What the operator has to do about it. `Some` is what makes a merged run
    /// read as needing attention.
    #[serde(default)]
    pub action_required: Option<String>,
    /// Why no release bump was made. A skipped bump is not a completed
    /// release: the report and the web page say SKIPPED with this reason.
    #[serde(default)]
    pub skipped: Option<String>,
}

/// The workflow runs `land` reran on one head, with when (unix seconds).
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct LandReruns {
    /// The commit the reruns were made on.
    pub head: String,
    /// Workflow run id to the unix second it was rerun.
    pub runs: BTreeMap<String, i64>,
}

/// Successful launcher fetch, persisted independently of moving tracking refs.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FetchedBase {
    /// Remote fetched by the launcher.
    pub remote: String,
    /// Remote branch selected as the run base.
    pub branch: String,
    /// Resolved commit immediately after the successful fetch.
    pub commit: String,
    /// Time the launcher observed the successful fetch.
    pub fetched_at: Timestamp,
}

/// The whole run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunState {
    /// On-disk format version.
    pub schema: u32,
    /// Run id, e.g. `20260830-153012-a1b2`.
    pub id: String,
    /// Repository the run operates on.
    pub repo: PathBuf,
    /// Branch the run started from.
    pub base_branch: String,
    /// Commit the run started from.
    pub base_commit: String,
    /// Absent in older runs; never infer fetch evidence from a tracking ref.
    #[serde(default)]
    pub fetched_base: Option<FetchedBase>,
    /// The task, verbatim.
    pub instruction: String,
    /// Authored task criteria before review scaffolding was generated.
    #[serde(default)]
    pub task_instruction: Option<String>,
    /// When the run was created.
    pub created_at: Timestamp,
    /// Last state flush.
    pub updated_at: Timestamp,
    /// Current status.
    pub status: RunStatus,
    /// Seed for labels and session ids.
    pub seed: u64,
    /// Config snapshot, so a resumed run behaves like the original.
    pub config: Config,
    /// Additive approval policy; retained even if a later head removes it.
    #[serde(default)]
    pub branch_owner_holds: Vec<BranchOwnerHold>,
    /// Did this run take a reference on `extensions.worktreeConfig` being on
    /// (see [`crate::git::acquire_worktree_config`])? If so, cleanup releases
    /// it - which only actually turns the setting back off once every other
    /// run sharing this repository has released its own reference too.
    #[serde(default)]
    pub enabled_worktree_config: bool,
    /// Public pull request head, separate from competition branches.
    #[serde(default)]
    pub pr_head: Option<String>,
    /// Operator-selected source and original public lease, distinct from repaired HEAD.
    #[serde(default)]
    pub pr_recovery: Option<crate::pr_recovery::Selection>,
    /// Actual ordered load inputs (remote-layer paths contain their source commit).
    /// The resolved commands are retained in `config`, without re-rendering provenance.
    #[serde(default)]
    pub recovery_config: Option<crate::pr_recovery::ConfigEvidence>,
    /// Last head successfully pushed by this run; never an observed remote tip.
    #[serde(default)]
    pub pr_pushed_head: Option<String>,
    /// Public PR head observed when the forge confirmed its merge; absent on old records.
    #[serde(default)]
    pub pr_merged_head: Option<String>,
    /// Last title posted by this run, used to preserve human title edits.
    #[serde(default)]
    pub pr_posted_title: Option<String>,
    /// Publication has not completed; resume retries it without competing again.
    #[serde(default)]
    pub pr_publication_pending: bool,
    /// Successor that owns this run's pull request; this record is historical.
    #[serde(default)]
    pub pr_taken_over_by: Option<String>,
    /// Empty attempt retired to a recovery authority; reports and refs remain.
    /// Separate from Superseded: this never settles questions or closes PRs.
    #[serde(default)]
    pub retired_to: Option<String>,
    /// Candidates.
    #[serde(default)]
    pub candidates: Vec<Candidate>,
    /// Initial blind rankings.
    #[serde(default)]
    pub judgements: Vec<Judgement>,
    /// `judge` decided a solo candidate needs no panel and only logged it.
    ///
    /// `judgements` stays empty in that case — nothing to distinguish from
    /// "not yet judged" — so this is the record that makes the skip
    /// idempotent: without it, every reentry re-ran `judge`, re-logged the
    /// same event, and rewrote `status` to `Judging` over whatever a later
    /// node had already concluded.
    #[serde(default)]
    pub judge_skipped: bool,
    /// Deliberation, if it happened.
    #[serde(default)]
    pub deliberation: Vec<DeliberationRound>,
    /// Private final votes.
    #[serde(default)]
    pub votes: Vec<VoteRecord>,
    /// The count.
    #[serde(default)]
    pub tally: Option<Tally>,
    /// Review rounds.
    #[serde(default)]
    pub reviews: Vec<ReviewRound>,
    /// Final gate.
    ///
    /// Never derive whether the gate has run from `gate.is_empty()` alone —
    /// use [`Self::gate_status`] instead. An empty list is ambiguous on its
    /// own: it is what an unattempted gate looks like, what a
    /// resource-blocked attempt leaves behind (see `graph::Runner::gate`'s
    /// own doc), and also what a repo with no `verify.gate` commands
    /// configured produces once it *has* run. [`Self::gate_ran`] is what
    /// tells the third case apart from the first two.
    #[serde(default)]
    pub gate: Vec<CommandOutcome>,
    /// Did `gate()` actually record an attempt — zero commands configured
    /// (not configured), or one or more commands that ran to
    /// completion — as opposed to never having run, or having last hit a
    /// resource-blocked retry?
    ///
    /// `gate.is_empty()` cannot tell those apart by itself: a repo with no
    /// `verify.gate` commands leaves `gate` empty exactly like an
    /// unattempted or resource-blocked one does, and reading that empty list
    /// as "not yet run" is what stranded a review-only run in
    /// `RunStatus::Gating` forever on such a repo — see `SCHEMA`'s doc for
    /// schema 7. A record written before this field existed defaults to
    /// `false` and is migrated in [`migrate_schema`].
    #[serde(default)]
    pub gate_ran: bool,
    /// The winner's commit the recorded `gate` ran against. `None` on a record
    /// from before this existed, which reads as "commit unknown".
    #[serde(default)]
    pub gate_head: Option<String>,
    /// Verification on a changed head, separate from the review judgment.
    #[serde(default)]
    pub head_e2e: Option<GateRepairE2e>,
    /// Append-only evidence replaced when the winner moves to another head.
    #[serde(default)]
    pub verification_history: Vec<VerificationEvidence>,
    /// Resume must reconcile, sync, gate and publish the existing PR head before CI.
    #[serde(default)]
    pub land_sync_pending: bool,
    /// Winner tip before a landing rebase, retained until worktree setup finishes.
    #[serde(default)]
    pub land_rebase_from: Option<String>,
    /// Fix rounds a failing gate has already spent, oldest first. Persisted
    /// right after each fixer call so a resumed run spends only what is left
    /// of `graph.gate_fix_rounds` instead of starting the budget over.
    #[serde(default)]
    pub gate_fixes: Vec<GateFixRecord>,
    /// Publication after gate repair requires successful e2e on the final SHA.
    #[serde(default)]
    pub gate_repair_e2e: Option<GateRepairE2e>,
    /// Fixer rounds spent on a rebase conflict (base sync and land share one
    /// budget, `graph.review_rounds`), oldest first. Persisted right before
    /// each fixer call, so a resumed run spends only what is left.
    #[serde(default)]
    pub rebase_fixes: Vec<RebaseFixRecord>,
    /// The graft phase's record, once it started or was skipped for a
    /// reason worth keeping. `None` for a run that never reached it.
    #[serde(default)]
    pub graft: Option<GraftRecord>,
    /// What this run inherited from the attempt it took over, and why (see
    /// [`crate::recovery`]). `None` for a run that took nothing over.
    /// Additive and `#[serde(default)]`, so `SCHEMA` stays put.
    #[serde(default)]
    pub recovery: Option<crate::recovery::RecoveryRecord>,
    /// Set at the review loop's hand-off when the last round ended with a
    /// blocking finding still open *and* a reviewer's final vote was reject
    /// (see [`ReviewRound::contested_handoff`]). `land` reads it to ask the
    /// owner before merging even with `graph.land_approval` off. A snapshot
    /// taken at hand-off, never recomputed later.
    #[serde(default)]
    pub contested_handoff: Option<ContestedHandoff>,
    /// The head commit `land` has armed (or is about to arm) GitHub auto-merge
    /// on. Saved *before* the arm call, so a crash right after it still tells
    /// a resume that the forge may be holding an armed merge. Cleared the
    /// moment auto-merge is disabled or the pull request is done.
    #[serde(default)]
    pub land_armed_head: Option<String>,
    /// Workflow runs `land` has rerun for the head it is bound to. Saved
    /// *before* each `gh run rerun`, so a restart never reruns twice; replaced
    /// when the head moves.
    #[serde(default)]
    pub land_reruns: Option<LandReruns>,
    /// First unreadable merge-state observation per head, independent of reruns.
    #[serde(default)]
    pub land_unsaid: Option<(String, i64)>,
    /// Structured CI evidence and owner question for a machine hold.
    #[serde(default)]
    pub land_unrelated: Option<crate::land::CiHold>,
    /// The merge-approval question `land` last filed, and the head commit its
    /// panel showed. An answer is only honoured for that exact commit: a fix
    /// or rebase push makes a new head the owner has not seen.
    #[serde(default)]
    pub land_approval: Option<LandApproval>,
    /// Repair history on this same PR and run; no second task owns the branch.
    #[serde(default)]
    pub merge_repairs: Vec<MergeRepair>,
    /// Owner stop/withdraw/replacement checkpoint, independent of PR head movement.
    #[serde(default)]
    pub merge_decision: Option<crate::merge_decision::Decision>,
    /// Previous attempt replaced by this run; its PR is never reused.
    #[serde(default)]
    pub replaces: Option<crate::merge_decision::Replacement>,
    /// The posting-gate question for this run's pull request text, if the gate
    /// withheld any. While `resolved` is `None` the run is parked on it.
    #[serde(default)]
    pub github_text: Option<GithubTextAsk>,
    /// The whole-change description of a review-only run, tied to the final
    /// commit and base it was written from. `None` until one was produced.
    #[serde(default)]
    pub change_summary: Option<ChangeSummary>,
    /// Outcomes of the `verify.pre_gate` commands from the latest time they
    /// ran on the winner. Informational only: a failure here never blocks the
    /// run, the gate stays the single arbiter. Overwritten on each pass.
    #[serde(default)]
    pub pre_gate: Vec<CommandOutcome>,
    /// The latest commit a `pre_gate` pass made, if any pass left changes.
    #[serde(default)]
    pub pre_gate_commit: Option<String>,
    /// Merge outcome.
    #[serde(default)]
    pub merge: Option<MergeOutcome>,
    /// Vendor tokens seen in judged material.
    #[serde(default)]
    pub leaks: Vec<Leak>,
    /// Seats lost to a CLI rate limit / quota, in the order they hit.
    #[serde(default)]
    pub quota: Vec<QuotaLoss>,
    /// Seats handed to the next roster agent after a failure, in order.
    /// Additive, so `SCHEMA` is not bumped.
    #[serde(default)]
    pub handovers: Vec<Handover>,
    /// How many agent-change seats this run has minted; mixed into their
    /// session ids so re-handing a seat to an agent it already had never
    /// reuses that agent's earlier uuid. Additive.
    #[serde(default)]
    pub seat_epoch: u64,
    /// Per reviewer seat, which roster agents already failed it and which one
    /// last answered, carried across review rounds. Additive, so `SCHEMA` is
    /// not bumped. See [`SeatHistory`].
    #[serde(default)]
    pub seat_history: BTreeMap<String, SeatHistory>,
    /// Stray foreign lockfiles a rescue commit left out, one entry per path.
    #[serde(default)]
    pub withheld: Vec<Withheld>,
    /// One entry per time a process took this run over (`Runner::execute`),
    /// oldest first. Append-only: a later resume closes the pass before it
    /// and never rewrites what that pass recorded. A run from before this
    /// existed has none for its earlier passes, which stay "not recorded".
    /// Additive, so `SCHEMA` is not bumped.
    #[serde(default)]
    pub passes: Vec<RunPass>,
    /// Set by whoever is about to execute this run on a task's behalf, read
    /// and cleared by the next [`RunState::begin_pass`]. Never persisted.
    #[serde(skip)]
    pub next_pass_task_entry: bool,
    /// Parked at a node boundary, waiting to be resumed.
    ///
    /// A run that is neither finished nor being worked on is otherwise
    /// indistinguishable from one whose daemon was killed, and the two want
    /// opposite things from an operator: the first is expected to be resumed,
    /// the second is a leftover. Cleared by the resume that carries it on.
    #[serde(default)]
    pub parked: bool,
    /// Seats of `node` are waiting, before any agent is spawned, for a shared
    /// resource (the build cache's lease). While this is `Some` the seats in
    /// [`Self::active`] are *not* running: their countdown has not started.
    /// Cleared the moment the wait ends, whichever way.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resource_wait: Option<ResourceWait>,
    /// The run stopped on a resource operation. This does not establish a timeout.
    /// Pre-spawn acquisition failures do not spend the affected agent's budget,
    /// alter its seat history or cause fallback. Earlier operations may have
    /// completed; resource interruption can also stop a running operation.
    /// Cleared by resume.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub resource_block: Option<ResourceBlock>,
    /// The run that took this run's worktree over, when one did — see
    /// [`crate::handover`]. `Some` means the worktree is gone on purpose and
    /// the run can no longer be resumed from it ([`Self::released`]).
    #[serde(default)]
    pub released_to: Option<String>,
    /// Branches that outlived this run's released worktree and now belong to
    /// the run in [`Self::released_to`] (and to its pull request). A later
    /// fold must leave them alone.
    #[serde(default)]
    pub released_branches: Vec<String>,
    /// Turns whose Git/artifact preservation is not confirmed. A resume retries
    /// persistence before deciding whether an interrupted turn needs execution.
    #[serde(default)]
    pub candidate_preservation: BTreeMap<char, String>,
    /// Interrupted turns whose empty trees must remain eligible for execution.
    /// Kept separately from preservation errors so failed rescue retries retain
    /// the distinction between an unfinished turn and a completed empty reply.
    #[serde(default)]
    pub interrupted_implementations: BTreeSet<char>,
    /// Legacy automatically inferred references, retained only for audit.
    /// These are never explicit authorization and never seed preparation.
    #[serde(default)]
    pub seeds: Vec<crate::refs::Seed>,
    /// Explicit, SHA-pinned implementation imports; separate from instruction text.
    #[serde(default, skip_serializing_if = "crate::refs::Sources::is_empty")]
    pub sources: crate::refs::Sources,
    /// Candidate preparation stopped before agents ran; requires an explicit source correction.
    #[serde(default)]
    pub prep_failure: Option<String>,
    /// None identifies legacy records; new runs persist preparation completion explicitly.
    #[serde(default)]
    pub prep_complete: Option<bool>,
    /// Per-seat conversation state.
    #[serde(default)]
    pub seats: BTreeMap<String, SeatState>,
    /// Seats currently mid-answer, keyed by seat.
    ///
    /// An entry exists from the moment a prompt is sent until a reply (of any
    /// kind — success, failure, quota, drop) comes back, so its keys are
    /// exactly "who hasn't answered yet" for whichever node populated it. See
    /// [`ActiveSeat`] for why a reader still has to check a live daemon
    /// before trusting one of these as "running" rather than "abandoned".
    #[serde(default)]
    pub active: BTreeMap<String, ActiveSeat>,
    /// Process id of whichever `execute()` call last drove this run —
    /// written at the very top of that method, the same place
    /// [`Self::clear_active`] runs, so a fresh reentry always overwrites the
    /// pid a previous, possibly-dead process left behind.
    ///
    /// A daemon-claimed run already has a stronger signal
    /// (`daemon::is_working_on`), but a `magi run` / `magi review` typed
    /// straight into a terminal claims nothing there — before this field
    /// existed, [`report::active_seats`] had no way to tell that run apart
    /// from one a killed process abandoned, and printed the same "no live
    /// daemon claims this run" warning over a run that was, in fact, still
    /// answering. See [`Liveness`] for how this and the daemon claim combine.
    /// Released ownership clears this field together with `driver_started_at`;
    /// legacy records may retain the old pid with `driver_exited = true`.
    #[serde(default)]
    pub driver_pid: Option<u32>,
    /// Whether execution has ever claimed this record. New runs persist false
    /// before exposure, then true atomically with their first pass and PID.
    /// None is legacy/unknown, never evidence that execution did not start.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub driver_claimed: Option<bool>,
    /// The OS-reported moment [`Self::driver_pid`] started, recorded in the
    /// same breath as the pid itself — an opaque marker
    /// (`crate::proc::process_started_at`, epoch seconds as an integer
    /// string), compared only for equality. Older runs hold a locale-dependent
    /// string here; those read as `Unknown`, never `Dead`.
    ///
    /// A pid alone never proves a live process is *this run's* driver: pids
    /// get reused, sometimes within minutes on a busy machine, and a killed
    /// manual `magi run` whose pid a later, wholly unrelated process happens
    /// to receive would otherwise read back as `Liveness::Live` from that
    /// coincidence alone. [`Self::liveness`] re-queries the current holder
    /// of `driver_pid` and requires this marker to still match before
    /// trusting a live answer — a mismatch means a different process now
    /// answers to that number, and no marker to compare (an old run, or a
    /// platform this build could not ask at record time) means neither
    /// extreme can be proven.
    #[serde(default)]
    pub driver_started_at: Option<String>,
    /// The process named by [`Self::driver_pid`] has stopped driving this run:
    /// its graph walk returned, whatever the outcome. Cleared at the start of
    /// every walk, set as the last write of one.
    ///
    /// A long-lived `magi serve` outlives its individual graph walks. The
    /// execution guard sets this and clears the pid/start marker when a walk
    /// returns, is cancelled or unwinds. A kill or failed save instead leaves
    /// a free pass-bound execution lease or a lost process identity for
    /// startup reconciliation. Older exited records may retain a historical pid.
    #[serde(default)]
    pub driver_exited: bool,
    /// Last observation of the winner's pull request, when a land loop ran.
    ///
    /// Persisted rather than derived from the event log because the phone asks
    /// two questions about a run that has opened a PR - how are its checks and
    /// which round is it on - and parsing prose out of events to answer them
    /// would break the first time an event message was reworded.
    #[serde(default)]
    pub pr: Option<PrRecord>,
    /// The post-merge release-bump step's outcome, when it got as far as
    /// having one to report. See [`ReleaseBump`].
    #[serde(default)]
    pub release_bump: Option<ReleaseBump>,
    /// Decision history also survives failures before a release outcome exists.
    #[serde(default)]
    pub bump_decision_attempts: Vec<crate::bump::DecisionAttempt>,
    /// The last look at how far the winner's tree trailed the landing base,
    /// and the rebase(s) tried to close that gap. `None` until the tree has a
    /// winner to check.
    #[serde(default)]
    pub base_sync: Option<BaseSync>,
    /// Stable stop identity and single-use owner authorizations, with prior evidence.
    #[serde(default)]
    pub base_sync_recovery: crate::base_sync_recovery::Recovery,
    /// The design-deliberation stage's output, when `[graph] advise` ran it:
    /// one record per advisor seat, plus the synthesis blended into the
    /// implementer's prompt. `None` when the stage is off, has not run yet,
    /// or could not even resolve its seats - see
    /// [`crate::graph::Runner::advise`].
    #[serde(default)]
    pub advice: Option<crate::advise::Advice>,
    /// Whether the design-deliberation stage has already been attempted this
    /// run, whatever it produced. The idempotency marker `Runner::advise`
    /// checks on reentry, the same role [`Self::judge_skipped`] plays for
    /// `judge` - without it a resumed run whose stage failed (a misconfigured
    /// `[roles] advisors`, every seat quota'd) would re-run it, and re-spend
    /// the agent calls, on every single reentry before `implement`.
    #[serde(default)]
    pub advise_attempted: bool,
    /// Node log.
    #[serde(default)]
    pub events: Vec<Event>,
    /// Commands seats' own CLIs reported running, across every node — see
    /// [`JobRecord`]. Populated in [`crate::graph::wave`] as each seat
    /// answers, so a resumed run keeps what earlier waves already collected
    /// rather than losing it to a reentry. Empty on a record written before
    /// this existed, or wherever no adapter reads structured job events for
    /// the backend a seat used — both read as "no evidence", not "nothing
    /// ran".
    #[serde(default)]
    pub jobs: Vec<JobRecord>,
    /// Operator-triggered targeted fixes — see [`OperatorFixRequest`] and
    /// `SCHEMA`'s doc for schema 9. Empty on every record written before
    /// this existed, which reads correctly as "no operator fix ever
    /// requested".
    #[serde(default)]
    pub operator_fixes: Vec<OperatorFixRequest>,
    /// Absolute paths of the files the task was filed with (see
    /// `crate::queue::Task::attachments`), listed in the implementers' prompt
    /// and made readable to their seats. Refreshed from the task on every
    /// start and resume, so a file attached while the task was held reaches a
    /// resumed run. Additive and `#[serde(default)]`, so `SCHEMA` stays put.
    #[serde(default)]
    pub attachments: Vec<PathBuf>,
    /// Subjects of the commits a review-only run examines, oldest first, as
    /// they stood when the run opened. `Some` marks the run as reviewing
    /// existing work: the pull request title is taken from these rather than
    /// from the review prompt in `instruction`. Additive and
    /// `#[serde(default)]`, so `SCHEMA` stays put.
    #[serde(default)]
    pub reviewed_commits: Option<Vec<String>>,
    /// Who started this run and which task it serves — see [`Origin`] and
    /// `SCHEMA`'s doc for schema 13. `None` is a run recorded before origins
    /// existed, and is reported as unknown, never as an operator's.
    #[serde(default)]
    pub origin: Option<Origin>,
    /// Follow-up tasks filed from this run's open findings after it merged
    /// (`crate::followup`). Additive and `#[serde(default)]`, so `SCHEMA`
    /// stays put.
    #[serde(default)]
    pub followups: Vec<FollowupRecord>,
    /// Follow-up depth of the task this run served, remembered the first
    /// time it was read so deleting that task cannot reset the cap.
    #[serde(default)]
    pub followup_generation: Option<u32>,
    /// Chat the task this run served descends from, fixed at start like
    /// `followup_generation` so deleting the task cannot lose it.
    #[serde(default)]
    pub origin_chat: Option<String>,
    /// The pull-request comment naming the filed follow-ups was posted.
    #[serde(default)]
    pub followup_commented: bool,
    /// Held tasks the merge approval's deputy filed for this run that were
    /// released after the merge (`crate::followup::settle_deputy_tasks`).
    /// Additive and `#[serde(default)]`, so `SCHEMA` stays put; the task
    /// record is the source of truth and this is rebuilt from the queue.
    #[serde(default)]
    pub deputy_followups: Vec<String>,
}

/// One follow-up task filed from a merged run. See [`RunState::followups`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FollowupRecord {
    /// Queue task id.
    pub task: String,
    /// Finding ids it covers.
    pub findings: Vec<String>,
}

/// A monotonic owner hold observed in a reviewed commit.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BranchOwnerHold {
    /// Approval switch enabled by the branch.
    pub control: crate::config::OwnerHold,
    /// Repository-relative configuration file.
    pub file: String,
    /// Pinned commit where the hold was first observed.
    pub head: String,
}

impl RunState {
    /// Effective policy is the OR of trusted config and observed branch holds.
    pub fn owner_hold(&self, control: crate::config::OwnerHold) -> bool {
        control.base_value(&self.config.graph)
            || self.branch_owner_holds.iter().any(|h| h.control == control)
    }

    /// Read only pinned commits of existing-branch reviews, never candidates.
    pub async fn observe_branch_owner_holds(&mut self, head: &str) {
        if !crate::graph::is_review_run(self) {
            return;
        }
        let (holds, diagnostics) = crate::config::branch_owner_holds(&self.repo, head).await;
        for message in diagnostics {
            tracing::warn!("{message}");
            self.event("approval-policy", message);
        }
        for (control, file) in holds {
            if self
                .branch_owner_holds
                .iter()
                .any(|h| h.control == control && h.file == file)
            {
                continue;
            }
            let message = format!(
                "{}=true from the reviewed branch's {file} at {head}",
                control.key()
            );
            tracing::info!("{message}");
            self.event("approval-policy", message);
            self.branch_owner_holds.push(BranchOwnerHold {
                control,
                file,
                head: head.to_owned(),
            });
        }
    }

    /// A fresh, unreserved run. Production callers must call [`Self::reserve`]
    /// before using its id, deriving branches, or saving it. Resumes load instead.
    pub fn new(
        repo: PathBuf,
        base_branch: String,
        base_commit: String,
        instruction: String,
        config: Config,
    ) -> Self {
        let now = Timestamp::now();
        let seed = config.blind.seed.unwrap_or_else(crate::rng::entropy);
        Self {
            schema: SCHEMA,
            id: new_id(),
            repo,
            base_branch,
            base_commit,
            fetched_base: None,
            instruction,
            task_instruction: None,
            created_at: now,
            updated_at: now,
            status: RunStatus::Prep,
            seed,
            config,
            branch_owner_holds: Vec::new(),
            enabled_worktree_config: false,
            pr_head: None,
            pr_recovery: None,
            recovery_config: None,
            pr_pushed_head: None,
            pr_merged_head: None,
            pr_posted_title: None,
            pr_publication_pending: false,
            pr_taken_over_by: None,
            retired_to: None,
            candidates: Vec::new(),
            judgements: Vec::new(),
            judge_skipped: false,
            deliberation: Vec::new(),
            votes: Vec::new(),
            tally: None,
            reviews: Vec::new(),
            gate: Vec::new(),
            gate_ran: false,
            gate_head: None,
            head_e2e: None,
            verification_history: Vec::new(),
            land_sync_pending: false,
            land_rebase_from: None,
            gate_fixes: Vec::new(),
            gate_repair_e2e: None,
            rebase_fixes: Vec::new(),
            graft: None,
            recovery: None,
            contested_handoff: None,
            land_armed_head: None,
            land_reruns: None,
            land_unsaid: None,
            land_unrelated: None,
            land_approval: None,
            merge_repairs: Vec::new(),
            merge_decision: None,
            replaces: None,
            github_text: None,
            change_summary: None,
            pre_gate: Vec::new(),
            pre_gate_commit: None,
            merge: None,
            leaks: Vec::new(),
            quota: Vec::new(),
            handovers: Vec::new(),
            seat_epoch: 0,
            seat_history: BTreeMap::new(),
            withheld: Vec::new(),
            passes: Vec::new(),
            next_pass_task_entry: false,
            parked: false,
            resource_wait: None,
            resource_block: None,
            released_to: None,
            released_branches: Vec::new(),
            candidate_preservation: BTreeMap::new(),
            interrupted_implementations: BTreeSet::new(),
            seeds: Vec::new(),
            sources: crate::refs::Sources::default(),
            prep_failure: None,
            prep_complete: Some(false),
            seats: BTreeMap::new(),
            active: BTreeMap::new(),
            driver_pid: None,
            driver_claimed: Some(false),
            driver_started_at: None,
            driver_exited: false,
            pr: None,
            release_bump: None,
            bump_decision_attempts: Vec::new(),
            base_sync: None,
            base_sync_recovery: Default::default(),
            advice: None,
            advise_attempted: false,
            attachments: Vec::new(),
            reviewed_commits: None,
            origin: None,
            followups: Vec::new(),
            followup_generation: None,
            deputy_followups: Vec::new(),
            origin_chat: None,
            followup_commented: false,
            events: Vec::new(),
            jobs: Vec::new(),
            operator_fixes: Vec::new(),
        }
    }

    /// Directory holding this run's state and artifacts.
    pub fn dir(&self) -> PathBuf {
        run_dir(&self.id)
    }

    /// Short form used in branch names and reports.
    pub fn short(&self) -> &str {
        short_of(&self.id)
    }

    /// Branch name for a label.
    pub fn branch_for(&self, label: char) -> String {
        format!("magi/{}/{}", self.short(), label)
    }

    /// Public head for a queue competition. Review-only and standalone runs
    /// keep their existing branch. Invalid task ids cannot become git refs.
    pub fn task_pr_branch(&self) -> Option<String> {
        if self.reviewed_commits.is_some()
            || self
                .candidates
                .iter()
                .any(|c| c.agent == "(existing branch)")
        {
            // A review-only run continues the task's pull request only when
            // it inherited that head from an earlier attempt (see
            // `Runner::review_taking_over`). A legacy `magi/<run>/<label>`
            // PR or a hand-named branch keeps its own head.
            let inherited = self.pr_head.is_some();
            if !inherited {
                return None;
            }
        }
        let task = self.origin.as_ref()?.task.as_ref()?;
        let branch = if let Some(replaces) = &self.replaces {
            format!("magi/task-{task}-{}", replaces.question)
        } else {
            format!("magi/task-{task}")
        };
        std::process::Command::new("git")
            .quiet()
            .args(["check-ref-format", "--branch", &branch])
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
            .ok()
            .filter(|s| s.success())?;
        if (self.reviewed_commits.is_some()
            || self
                .candidates
                .iter()
                .any(|c| c.agent == "(existing branch)"))
            && self.pr_head.as_deref() != Some(branch.as_str())
        {
            return None;
        }
        Some(branch)
    }

    /// Recorded public head, falling back to the winner for older runs.
    pub fn pr_branch(&self) -> Option<&str> {
        self.pr_head
            .as_deref()
            .or_else(|| self.winner().map(|w| w.branch.as_str()))
    }

    /// Root of this run's worktrees.
    pub fn worktree_root(&self) -> PathBuf {
        self.config
            .graph
            .worktree_root
            .clone()
            .unwrap_or_else(default_worktree_root)
            .join(self.short())
    }

    /// Record lockfiles a rescue commit withheld; a path already recorded by an
    /// earlier round is not repeated.
    pub fn note_withheld(&mut self, node: &str, strays: &[crate::git::Stray]) {
        for s in strays {
            if self.withheld.iter().any(|w| w.path == s.path) {
                continue;
            }
            self.event(
                node,
                format!(
                    "withheld stray lockfile {} ({}; the directory uses {})",
                    s.path, s.manager, s.kept_by
                ),
            );
            self.withheld.push(Withheld {
                path: s.path.clone(),
                manager: s.manager.clone(),
                kept_by: s.kept_by.clone(),
                node: node.to_owned(),
                at: Timestamp::now(),
            });
        }
    }

    /// Open a pass for the process that is about to drive this run. A pass
    /// left open by an earlier process is closed as [`PassEnd::Unrecorded`];
    /// one that recorded its own end is never touched.
    pub fn begin_pass(&mut self) {
        let now = Timestamp::now();
        if let Some(last) = self.passes.last_mut()
            && last.end.is_none()
        {
            last.end = Some(PassEnd::Unrecorded);
            last.ended_at = Some(now);
        }
        let resumed = !self.passes.is_empty() || !self.candidates.is_empty();
        self.passes.push(RunPass {
            index: self.passes.len(),
            started_at: now,
            resumed,
            ended_at: None,
            end: None,
            task_entry: std::mem::take(&mut self.next_pass_task_entry),
        });
    }

    /// Close the open pass as parked for `cause`. A pass that already has an
    /// end keeps it (the first reason wins).
    pub fn end_pass_parked(&mut self, cause: ParkCause) {
        if let Some(last) = self.passes.last_mut()
            && last.end.is_none()
        {
            last.end = Some(PassEnd::Parked { cause });
            last.ended_at = Some(Timestamp::now());
        }
    }

    /// Resolve legacy delivery intent only from an already published PR and approval binding.
    fn recovery_disposition(&self) -> Option<crate::pr_recovery::Disposition> {
        let selected = self.pr_recovery.as_ref()?;
        selected.disposition.or_else(|| {
            (self.config.merge.mode == MergeMode::Pr
                && self.land_approval.is_some()
                && self.pr.as_ref().is_some_and(|pr| pr.url == selected.url))
            .then_some(crate::pr_recovery::Disposition::Land)
        })
    }

    /// Validate a resumed recovery against its own durable configuration and landing records.
    pub fn configure_recovery_resume(&mut self) -> anyhow::Result<()> {
        if let Some(mut selected) = self.pr_recovery.clone() {
            selected.disposition = self.recovery_disposition();
            selected.configure(&mut self.config)?;
        }
        Ok(())
    }

    /// Only a confirmed merge of the selected public PR completes a delivery task.
    pub fn public_delivery_complete(&self) -> bool {
        self.status == RunStatus::Merged
            && self.config.merge.mode == MergeMode::Pr
            && self
                .merge
                .as_ref()
                .is_some_and(|m| m.ok && !m.empty && m.mode == MergeMode::Pr)
            && self.pr.as_ref().is_some_and(|p| {
                p.state.eq_ignore_ascii_case("merged")
                    && self.pr_recovery.as_ref().is_some_and(|s| {
                        s.url == p.url
                            && self.recovery_disposition()
                                == Some(crate::pr_recovery::Disposition::Land)
                    })
            })
            && self.pr_pushed_head.as_deref().is_some_and(|head| {
                self.pr_merged_head.as_deref() == Some(head)
                    && self.check_delivery_evidence(head).is_ok()
                    && self.land_approval.as_ref().is_some_and(|a| {
                        a.head == head
                            && crate::ask::Questions::open()
                                .get(&a.question)
                                .is_ok_and(|q| {
                                    q.run == self.id
                                        && q.node == crate::land::APPROVAL_NODE
                                        && crate::land::approval(q.resolution().as_deref())
                                            == crate::land::Approval::Merge
                                        && q.merge_target.as_ref().is_some_and(|t| {
                                            t.head == head
                                                && self.pr.as_ref().is_some_and(|p| t.pr == p.url)
                                        })
                                })
                    })
            })
    }

    /// Validate evidence for exactly the head about to be published or merged.
    pub fn check_delivery_evidence(&self, head: &str) -> anyhow::Result<()> {
        if self.recovery_disposition() != Some(crate::pr_recovery::Disposition::Land) {
            return Ok(());
        }
        let commands_match = |outcomes: &[CommandOutcome], commands: &[String]| {
            !commands.is_empty()
                && outcomes.len() == commands.len()
                && outcomes
                    .iter()
                    .zip(commands)
                    .all(|(o, c)| o.ok() && &o.command == c)
        };
        anyhow::ensure!(
            self.gate_ran
                && self.gate_head.as_deref() == Some(head)
                && commands_match(&self.gate, &self.config.verify.gate),
            "delivery requires successful configured gate commands on the exact final head; request a fresh recovery review"
        );
        anyhow::ensure!(
            self.reviews.last().is_some_and(|r| r.clean
                && r.expected > 0
                && r.answered == r.expected
                && r.head == head
                && r.verified_head.as_deref() == Some(head)
                && commands_match(&r.e2e, &self.config.verify.e2e)),
            "delivery requires complete clean review and configured e2e commands on the exact final head; request a fresh recovery review"
        );
        Ok(())
    }

    /// Reconcile after Git updates as well as on reentry after an interrupted save.
    /// Unknown legacy heads cannot establish current verification.
    pub fn reconcile_verification_head(&mut self, head: &str) {
        let stale_gate = self.gate_ran && self.gate_head.as_deref() != Some(head);
        let stale_e2e = self.head_e2e.as_ref().is_some_and(|e| e.head != head);
        if stale_gate || stale_e2e {
            self.verification_history.push(VerificationEvidence {
                superseded_by: Some(head.to_owned()),
                archived_at: Some(Timestamp::now()),
                fixes_spent: Some(self.gate_fixes.len()),
                gate_head: self.gate_head.clone(),
                gate_ran: self.gate_ran,
                gate: self.gate.clone(),
                e2e: self.head_e2e.clone(),
            });
            if stale_gate {
                let previous = self
                    .gate_head
                    .as_deref()
                    .map(short)
                    .unwrap_or_else(|| "unknown".into());
                self.event(
                    "gate",
                    format!(
                        "archived gate on {previous}, superseded by {}; {} gate-fix round(s) already spent stay spent",
                        short(head),
                        self.gate_fixes.len()
                    ),
                );
                self.gate.clear();
                self.gate_ran = false;
                self.gate_head = None;
                self.pre_gate.clear();
                self.pre_gate_commit = None;
            }
            if stale_e2e {
                self.head_e2e = None;
            }
        }
    }

    /// Structured current failures take precedence over successful progress events.
    pub fn verification_blocking_cause(&self) -> Option<String> {
        if self.base_sync_blocked() {
            return self
                .base_sync
                .as_ref()
                .and_then(|s| s.conflict.as_ref())
                .map(|detail| format!("Base-sync conflict: {detail}"));
        }
        if let Some(resource) = &self.resource_block {
            return Some(resource.detail.clone());
        }
        if self.status != RunStatus::Blocked {
            return None;
        }
        if crate::graph::review_conclusion(&self.reviews, self.config.graph.review_rounds)
            == Some(RunStatus::Blocked)
        {
            return self.reviews.last().map(|r| format!(
                "review blocked on reviewed head {} ({} blocking finding(s), {}/{} reviewers answered)",
                r.head, r.blocking, r.answered, r.expected
            ));
        }
        if let Some(e) = &self.head_e2e
            && !e.passed
            && e.outcomes.iter().any(|o| !o.ok())
        {
            return Some(format!(
                "verify.e2e failed on head {}; nothing published",
                e.head
            ));
        }
        if let Some(e) = &self.gate_repair_e2e
            && !e.passed
            && e.outcomes.iter().any(|o| !o.ok())
        {
            return Some(format!(
                "verify.e2e failed on head {} after gate repair; nothing published",
                e.head
            ));
        }
        if self.gate_ran && self.gate.iter().any(|o| !o.ok()) {
            return Some(format!(
                "{} on head {} after {} gate-fix round(s); nothing published",
                crate::stop_reason::command_failure(
                    "gate",
                    self.gate.iter().find(|o| !o.ok()).unwrap()
                ),
                self.gate_head.as_deref().unwrap_or("unknown"),
                self.gate_fixes.len()
            ));
        }
        None
    }

    /// Whether the configured gate commands completed successfully, independent of e2e.
    pub(crate) fn gate_commands_allow_progress(&self) -> bool {
        self.gate_status().ok()
            || (self
                .pr_recovery
                .as_ref()
                .is_none_or(|s| s.disposition == Some(crate::pr_recovery::Disposition::Review))
                && self.gate_ran
                && self.gate_status() == GateStatus::NotConfigured)
    }

    /// Empty verification can finish an intentional workflow, without claiming a test pass.
    pub fn gate_allows_progress(&self) -> bool {
        self.gate_commands_allow_progress()
            && self
                .head_e2e
                .as_ref()
                .is_none_or(|e| e.passed && self.gate_head.as_deref() == Some(e.head.as_str()))
    }

    /// Note something in the run log and on the tracing stream.
    pub fn event(&mut self, node: &str, message: impl Into<String>) {
        let message = message.into();
        tracing::info!(node, "{message}");
        self.events.push(Event {
            at: Timestamp::now(),
            node: node.to_owned(),
            message,
        });
    }

    /// The honest state of the final gate.
    ///
    /// Never derive this from `gate.is_empty()` alone anywhere else in the
    /// codebase — `NotRun` and `NotConfigured` both leave `gate`
    /// empty, and only this method (backed by [`Self::gate_ran`]) tells them
    /// apart. An empty attempt is never verification evidence.
    pub fn gate_status(&self) -> GateStatus {
        if !self.gate_ran {
            GateStatus::NotRun
        } else if self.gate.is_empty() {
            GateStatus::NotConfigured
        } else if self.gate.iter().all(CommandOutcome::ok) {
            GateStatus::Passed
        } else {
            GateStatus::Failed
        }
    }

    /// Record that `seat` was just sent a prompt for `node`, with the given
    /// wall-clock budget. `attempt` is 0 for the first ask and N for the Nth
    /// nudge or resume, purely for display — it does not change how the seat
    /// is treated.
    pub fn seat_started(
        &mut self,
        node: &str,
        seat: &str,
        timeout: std::time::Duration,
        attempt: usize,
    ) {
        self.active.insert(
            seat.to_owned(),
            ActiveSeat {
                node: node.to_owned(),
                started_at: Timestamp::now(),
                timeout_secs: timeout.as_secs(),
                attempt,
                task: None,
                command: None,
                index: None,
                total: None,
            },
        );
    }

    /// The wave's pre-spawn wait is over: restart the clock of every seat of
    /// this batch so `started_at` means "the CLI is being spawned now".
    pub fn seats_spawning(&mut self, node: &str, attempt: usize) {
        self.resource_wait = None;
        let now = Timestamp::now();
        for a in self.active.values_mut() {
            if a.task.is_none() && a.node == node && a.attempt == attempt {
                a.started_at = now;
            }
        }
    }

    /// `seat` holds its parallelism permit and its CLI is being spawned now.
    pub fn seat_spawned(&mut self, seat: &str) {
        if let Some(a) = self.active.get_mut(seat) {
            a.started_at = Timestamp::now();
        }
    }

    /// Record that `seat` has answered, whatever the answer was.
    pub fn seat_finished(&mut self, seat: &str) {
        self.active.remove(seat);
    }

    /// Record that `task` (`"e2e"` or `"gate"` — a shell-command list run
    /// outside any seat) has just started `command`, the `index`-th of
    /// `total`. Called at every command boundary, not once for the whole
    /// list: `verify.e2e` / `verify.gate` apply `timeout` per command, so
    /// this is the only way a reader can tell "how long is left" for
    /// whichever command is actually running right now, rather than a stale
    /// budget left over from the first one.
    #[allow(clippy::too_many_arguments)]
    pub fn task_command(
        &mut self,
        task: &str,
        node: &str,
        attempt: usize,
        command: &str,
        index: usize,
        total: usize,
        timeout: std::time::Duration,
    ) {
        self.active.insert(
            task.to_owned(),
            ActiveSeat {
                node: node.to_owned(),
                started_at: Timestamp::now(),
                timeout_secs: timeout.as_secs(),
                attempt,
                task: Some(task.to_owned()),
                command: Some(command.to_owned()),
                index: Some(index),
                total: Some(total),
            },
        );
    }

    /// Record that `task` has finished its whole command list for this
    /// attempt.
    pub fn task_finished(&mut self, task: &str) {
        self.active.remove(task);
    }

    /// The seats — never task entries — currently mid-answer. What
    /// `report::active_seats` and the phone's "who has not answered yet"
    /// note need: a seat's identifier is safe to show ([`ActiveSeat`]'s doc),
    /// so nothing here filters anything out beyond the type tag itself.
    pub fn seats_active(&self) -> impl Iterator<Item = (&String, &ActiveSeat)> {
        self.active.iter().filter(|(_, a)| a.task.is_none())
    }

    /// The command-list tasks — never seat entries — currently running.
    /// Counterpart to [`Self::seats_active`]; see [`ActiveSeat::task`] for
    /// the tag both read.
    pub fn tasks_active(&self) -> impl Iterator<Item = (&String, &ActiveSeat)> {
        self.active.iter().filter(|(_, a)| a.task.is_some())
    }

    /// Drop every seat this state still lists as answering, reporting whether
    /// anything was dropped.
    ///
    /// Called first thing in `execute`, on every entry — fresh, resumed, or
    /// recovering a stall — because an entry here only means something while
    /// the process that wrote it is still asking that seat something. A
    /// process killed mid-wave leaves its last batch of seats here with
    /// nobody left to clear them, and the next process to touch this run must
    /// not let that leftover read as "still going" before it has asked
    /// anyone anything.
    pub fn clear_active(&mut self) -> bool {
        if self.active.is_empty() {
            return false;
        }
        self.active.clear();
        true
    }

    /// Does every seat this run still lists as [`Self::active`] sit past its
    /// own [`ActiveSeat::timeout_secs`]? `false` when nothing is active at
    /// all — an empty map is not evidence of anything overrunning.
    ///
    /// This alone is not proof the run is dead: a seat's own attempt can
    /// legitimately run a little past its budget while the process driving it
    /// is still tearing the attempt down. Every caller pairs this with its own
    /// `!live` reading (`daemon::is_working_on`) before treating the run as
    /// abandoned — this module cannot check that itself without depending on
    /// `crate::daemon`, and callers already have to ask that question anyway.
    #[must_use]
    pub fn active_all_overrun(&self, now: Timestamp) -> bool {
        !self.active.is_empty()
            && self
                .active
                .values()
                .all(|a| a.elapsed_secs(now) > a.timeout_secs as i64)
    }

    /// Whether a process is actually still driving this run, given whether a
    /// daemon's heartbeat claims it and process-liveness/identity queries for
    /// [`Self::driver_pid`].
    ///
    /// A daemon claim wins outright when present — it is the stronger,
    /// independently-heartbeating signal. Absent that (every manual `magi
    /// run` / `magi review`, and every daemon-driven run whose daemon has
    /// since exited cleanly), `driver_pid` is asked directly. A live answer
    /// alone is not enough to trust, though: pids get reused, so `identity`
    /// re-queries whoever currently holds that pid and the result must still
    /// match [`Self::driver_started_at`] — the marker recorded at the same
    /// moment `driver_pid` was — before this reads `Live`. A mismatch means
    /// a *different* process now answers to that number, which is exactly as
    /// good as proof the original driver is gone, so that reads `Dead`. The
    /// marker is an epoch-seconds integer string; one that is not (a run
    /// recorded when it was a locale-dependent `ps` string) reads
    /// [`Liveness::Unknown`], never `Dead`. No
    /// marker to compare against (an old run, or a platform this build could
    /// not ask at record time) or a `None` from either query, and this
    /// cannot tell either way, so it reads [`Liveness::Unknown`] — never
    /// guessed as [`Liveness::Dead`] out of mere silence. A display that
    /// guessed "dead" out of missing information would be exactly the
    /// mtime-and-task-manager guessing this type exists to replace.
    ///
    /// Kept generic over `query` and `identity` so a test can inject answers
    /// without spawning a real process query — production code goes through
    /// [`Self::liveness`], which supplies [`crate::proc::pid_status`] and
    /// [`crate::proc::process_started_at`].
    #[must_use]
    pub fn liveness_with<F, G>(&self, daemon_claims: bool, query: F, identity: G) -> Liveness
    where
        F: FnOnce(u32) -> Option<bool>,
        G: FnOnce(u32) -> Option<String>,
    {
        if daemon_claims {
            return Liveness::Live;
        }
        // The driver said itself that it stopped; its pid may well be alive
        // (a daemon outlives the runs it drives), so it is not asked.
        if self.driver_exited {
            return Liveness::Dead;
        }
        let Some(pid) = self.driver_pid else {
            return Liveness::Unknown;
        };
        match query(pid) {
            Some(false) => Liveness::Dead,
            None => Liveness::Unknown,
            Some(true) => {
                let Some(recorded) = &self.driver_started_at else {
                    return Liveness::Unknown;
                };
                // A marker in the old locale-dependent format can never be
                // compared reliably, so it proves nothing either way.
                if !crate::proc::is_identity_marker(recorded) {
                    return Liveness::Unknown;
                }
                match identity(pid) {
                    Some(current) if !crate::proc::is_identity_marker(&current) => {
                        Liveness::Unknown
                    }
                    Some(current) if *recorded == current => Liveness::Live,
                    Some(_) => Liveness::Dead,
                    None => Liveness::Unknown,
                }
            }
        }
    }

    /// [`Self::liveness_with`], backed by the real process-liveness and
    /// identity queries.
    #[must_use]
    pub fn liveness(&self, daemon_claims: bool) -> Liveness {
        self.liveness_under(&home(), daemon_claims)
    }

    /// The same judge, rooted at the caller's run store.
    #[must_use]
    pub fn liveness_under(&self, home: &Path, daemon_claims: bool) -> Liveness {
        self.liveness_under_with(
            home,
            daemon_claims,
            crate::proc::pid_status,
            crate::proc::process_started_at,
        )
    }

    pub(crate) fn liveness_under_with<F, G>(
        &self,
        home: &Path,
        daemon_claims: bool,
        query: F,
        identity: G,
    ) -> Liveness
    where
        F: FnOnce(u32) -> Option<bool>,
        G: FnOnce(u32) -> Option<String>,
    {
        if daemon_claims {
            return Liveness::Live;
        }
        let process = self.liveness_with(false, query, identity);
        crate::driver::judge(self, home, process)
    }

    /// Clear every seat this run still lists as active and fail it, unless it
    /// had already reached a terminal status some other way.
    ///
    /// Callers must already have proven this run is dead — [`Self::active_all_overrun`]
    /// plus their own `!live` reading — before calling this; it does not
    /// check either itself. Unlike [`Self::clear_active`] (dropping a resumed
    /// run's own stale wave before repopulating it, called unconditionally at
    /// the top of every `execute()`), this is a verdict: a run left this way
    /// has nothing left to repopulate the wave, ever, and must stop reading as
    /// `implementing` (or whichever node) forever.
    pub fn abandon(&mut self, by: &str) {
        let seats: Vec<String> = self.active.keys().cloned().collect();
        self.clear_active();
        if !self.status.done() {
            self.status = RunStatus::Failed;
        }
        self.event(
            by,
            format!(
                "abandoned: seat(s) {} left behind by a killed process, past their own \
                 timeout with no live daemon claiming this run",
                seats.join(", ")
            ),
        );
    }

    /// Reserve a fresh run's identity and persist its initial record.
    /// Call exactly once, before exposing the id; never call on a loaded run.
    pub fn reserve(&mut self) -> Result<()> {
        self.reserve_under(&home())
    }

    /// [`Self::reserve`] with an explicit store home.
    pub fn reserve_under(&mut self, home: &Path) -> Result<()> {
        self.reserve_with(home, &crate::id_reserve::IdSource::new(new_id))
    }

    fn reserve_with(&mut self, home: &Path, ids: &crate::id_reserve::IdSource) -> Result<()> {
        let root = home.join("runs");
        std::fs::create_dir_all(&root).context("create runs root")?;
        let initial = self.id.clone();
        ids.reserve(
            (!ids.is_custom()).then_some(initial.as_str()),
            "run",
            |id| {
                let dir = root.join(id);
                if !crate::id_reserve::create_dir(&dir)? {
                    return Ok(false);
                }
                // Branches and worktrees use only the suffix. Publish our directory
                // before scanning: concurrent contenders either see each other and
                // retry, or the later one sees the earlier reservation and retries.
                for entry in std::fs::read_dir(&root).context("scan reserved run ids")? {
                    let name = entry?.file_name();
                    let Some(other) = name.to_str() else { continue };
                    if other != id && is_run_id(other) && short_of(other) == short_of(id) {
                        // Only remove our own empty directory, never existing data.
                        std::fs::remove_dir(&dir).context("release colliding run suffix")?;
                        return Ok(false);
                    }
                }
                self.id = id.to_owned();
                // A failed first write rolls back only our own directory:
                // `remove_dir` refuses anything with contents, so a partial
                // record or another run's data is never deleted.
                if let Err(e) = self.save_under(home) {
                    let _ = std::fs::remove_dir(&dir);
                    return Err(e);
                }
                Ok(true)
            },
        )?;
        Ok(())
    }

    /// Withdraw a run that was reserved but never exposed (a pre-start check
    /// failed), so its record cannot be mistaken for an owner of anything.
    /// Removes only this run's own freshly reserved directory.
    pub fn discard_reserved(&self) {
        let dir = self.dir();
        if let Err(e) = std::fs::remove_dir_all(&dir) {
            tracing::warn!("discard reserved run {}: {e:#}", self.id);
        }
    }

    /// Flush to `run.json`, atomically, under the process-global [`home`].
    pub fn save(&mut self) -> Result<()> {
        let home = home();
        self.save_under(&home)
    }

    /// [`Self::save`], rooted at an explicit `home` instead of the
    /// process-global one.
    ///
    /// For a caller that was already handed its own `home` explicitly — a
    /// housekeeping pass, mainly, for the same reason `Queue::at` and the
    /// daemon status path are parameters rather than resolved here (see
    /// `daemon::drive`'s own doc) — falling through to the global would write
    /// back through whichever directory some *other* process or test pinned
    /// into that `OnceLock` first, not the one this call was actually handed.
    pub fn save_under(&mut self, home: &Path) -> Result<()> {
        self.save_under_at(home, Timestamp::now())
    }

    pub(crate) fn save_under_at(&mut self, home: &Path, at: Timestamp) -> Result<()> {
        self.updated_at = at;
        let dir = home.join("runs").join(&self.id);
        std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
        let body = serde_json::to_string_pretty(self).context("serialize run state")?;
        let tmp = dir.join("run.json.tmp");
        std::fs::write(&tmp, &body).with_context(|| format!("write {}", tmp.display()))?;
        std::fs::rename(&tmp, dir.join("run.json")).with_context(|| "replace run.json")?;
        Ok(())
    }

    /// Load a run by id or unambiguous id prefix.
    pub fn load(id: &str) -> Result<Self> {
        let resolved = resolve_id(id)?;
        let path = run_dir(&resolved).join("run.json");
        let body =
            std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
        let state: Self =
            serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
        migrate_schema(state)
    }

    /// [`Self::load`], rooted at an explicit `home` instead of the
    /// process-global one, and for a full id rather than a prefix — a caller
    /// with its own `home` already has the exact id (from a `RunState` it
    /// already read, or from `Task::runs`) and has no `runs_root` to search
    /// for a prefix against in the first place. Same reasoning as
    /// [`Self::save_under`]: a caller holding its own `home` explicitly must
    /// not read back through whichever directory some *other* process or
    /// test pinned into the global [`home`] `OnceLock` first.
    pub fn load_under(id: &str, home: &Path) -> Result<Self> {
        let path = home.join("runs").join(id).join("run.json");
        let body =
            std::fs::read_to_string(&path).with_context(|| format!("read {}", path.display()))?;
        let state: Self =
            serde_json::from_str(&body).with_context(|| format!("parse {}", path.display()))?;
        migrate_schema(state)
    }
}

pub(crate) fn migrate_schema(mut state: RunState) -> Result<RunState> {
    // Schema 5 predates deferred e2e. Its empty e2e lists therefore mean
    // "not configured", never "deferred"; serde's field defaults retain
    // exactly that representation while this migration permits resumes.
    if state.schema == 5 {
        state.schema = 6;
    }
    // Schema 6 predates `gate_ran` and could not tell "never attempted or
    // resource-blocked" apart from "ran with zero commands configured" — see
    // `SCHEMA`'s doc for schema 7. A non-empty `gate` is a real recorded
    // attempt either way, so it is trusted as `gate_ran = true` rather than
    // spent again; an empty one is simply handed back to the next `gate()`
    // call, which re-attempts it and, for the zero-commands case, resolves
    // instantly.
    if state.schema == 6 {
        state.gate_ran = !state.gate.is_empty();
        state.schema = 7;
    }
    // Schema 7's main review loop always checked `e2e` against the round's
    // own `head`, it just never wrote that fact into `verified_head` unless
    // a catch-up run had checked a *different* commit — see `SCHEMA`'s doc
    // for schema 8. Reconstructing `Some(head)` for a round whose `e2e` held
    // a real attempt restores a fact that was always true; `verified_at` has
    // no historical value to recover and stays `None`.
    if state.schema == 7 {
        for round in &mut state.reviews {
            if round.verified_head.is_none()
                && matches!(round.e2e_status(), E2eStatus::Passed | E2eStatus::Failed)
            {
                round.verified_head = Some(round.head.clone());
            }
        }
        state.schema = 8;
    }
    // Schema 8 predates `operator_fixes`. There is nothing to reconstruct —
    // an old run simply never had one requested — so `#[serde(default)]`
    // already left it as the correct empty `Vec`; this only advances the
    // version number.
    if state.schema == 8 {
        state.schema = 9;
    }
    // Schema 9 predates `RunStatus::VerifiedNoop` and
    // `Candidate::verified_noop`. Nothing to reconstruct: an old record never
    // made the claim, `#[serde(default)]` already reads `verified_noop` as
    // `None` on every candidate, and a `VerifiedNoop` status cannot appear in
    // a schema-9 record at all — see `SCHEMA`'s doc for schema 10. This only
    // advances the version number.
    if state.schema == 9 {
        state.schema = 10;
    }
    // Schema 10 predates `released_to`: no old run ever had its worktree
    // handed over, and `#[serde(default)]` reads exactly that. Only the
    // version number advances.
    if state.schema == 10 {
        state.schema = 11;
    }
    // Schema 11 predates `driver_exited`. A run that had already ended
    // Blocked / Failed / Stalled, or was parked in Landing (which hands its
    // daemon slot back), stopped walking the graph, but its recorded
    // pid may belong to a daemon still alive doing other work, which would pin
    // it as running for good; those are marked exited. Any other status keeps
    // `false`, and its pid is asked as before. This is an inference from
    // status, sound because a schema-11 record was written by a build that is
    // gone once this one is installed (and a daemon claim, which wins
    // regardless, still protects a run a live daemon is driving).
    if state.schema == 11 {
        if matches!(
            state.status,
            RunStatus::Blocked | RunStatus::Failed | RunStatus::Stalled | RunStatus::Landing
        ) {
            state.driver_exited = true;
        }
        state.schema = 12;
    }
    // Schema 12 predates `origin`. Nothing is reconstructed: `None` already
    // says "unknown", which is the truth for a run nobody recorded. Only the
    // version number advances.
    if state.schema == 12 {
        state.schema = 13;
    }
    // Schema 13 predates `task_instruction`. Its `instruction` may include
    // generated review context, so it cannot supply authored criteria.
    // `#[serde(default)]` leaves the missing field as `None`; only the
    // version number advances, preserving origin and execution state.
    if state.schema == 13 {
        state.schema = 14;
    }
    // Only legacy, unfolded empty candidates lack a confirmed preservation
    // result, even if their CLI failed: failure never proves edits were saved.
    // Native schema-15 records and intentionally folded candidates
    // must not be re-flagged on every resume.
    if state.schema == 14 {
        for c in &state.candidates {
            if !c.folded && c.empty && c.verified_noop.is_none() {
                state
                    .candidate_preservation
                    .entry(c.label)
                    .or_insert_with(|| "legacy empty candidate requires inspection".to_owned());
            }
        }
        state.schema = 15;
    }
    // Native schema-15 preservation outcomes remain unchanged. Retirement
    // adds an execution fence without inferring any new preservation result.
    if state.schema == 15 {
        state.schema = SCHEMA;
    }
    if state.schema != SCHEMA {
        bail!(
            "run {} was written by a different magi (schema {}, this build \
                 speaks {SCHEMA})",
            state.id,
            state.schema
        );
    }
    Ok(state)
}

impl RunState {
    /// Whether a later attempt took this run's worktree over, so a resume has
    /// nothing left to continue from.
    pub fn released(&self) -> bool {
        self.released_to.is_some()
    }

    /// The winning candidate, once the tally has run.
    pub fn winner(&self) -> Option<&Candidate> {
        let label = self.tally.as_ref()?.winner;
        self.candidates.iter().find(|c| c.label == label)
    }

    /// Candidates eligible for judging.
    pub fn viable(&self) -> Vec<&Candidate> {
        self.candidates.iter().filter(|c| c.viable()).collect()
    }

    /// Did every candidate write nothing, and every one of them back it with
    /// evidence [`crate::graph`]'s adoption guard accepted?
    ///
    /// All-or-nothing on purpose: one candidate declaring `NO CHANGE NEEDED`
    /// while another simply failed to produce anything is not agreement, it
    /// is one candidate's unverified claim next to an ordinary loss, and the
    /// run must still read as the `Failed` it is. Only ever meaningful when
    /// [`Self::viable`] is already empty — a run with any real patch to judge
    /// never reaches the caller that asks this.
    pub fn all_candidates_verified_noop(&self) -> bool {
        !self.candidates.is_empty()
            && self
                .candidates
                .iter()
                .all(|c| c.empty && c.verified_noop.is_some())
    }

    /// Findings still open when the review loop stopped trying: the last
    /// round's, exactly when that round was not clean. Empty on a run that
    /// never reviewed, or whose last round was clean.
    ///
    /// This is the last round's findings regardless of what the fixer claims
    /// to have addressed in that same round: a round that stopped the loop
    /// (round budget spent, or no tree progress for
    /// [`crate::graph::STAGNANT_LIMIT`] rounds) never had a *following* round
    /// to confirm the fix actually landed, and the self-reported adoption
    /// count is not trusted for that judgement either — see
    /// [`ReviewRound::progressed`].
    pub fn open_findings(&self) -> Vec<&Finding> {
        match self.reviews.last() {
            Some(r) if !r.clean => r
                .reviews
                .iter()
                .flat_map(|rec| rec.findings.iter())
                .collect(),
            _ => Vec::new(),
        }
    }

    /// Every finding raised in the most recent review round, regardless of
    /// that round's own severity mix — unlike [`Self::open_findings`], not
    /// filtered to a round that was not clean. This is the pool `magi fix`
    /// reports as available to pick from: a round can conclude clean (no
    /// finding blocked merge) while still carrying minor findings nobody
    /// has acted on.
    pub fn last_round_findings(&self) -> Vec<&Finding> {
        self.reviews
            .last()
            .into_iter()
            .flat_map(|r| r.reviews.iter())
            .flat_map(|rec| rec.findings.iter())
            .collect()
    }

    /// Look up a finding by id anywhere in this run's review history,
    /// together with the round and reviewer record that raised it — the
    /// provenance `magi fix` snapshots onto [`OperatorFixFinding`].
    pub fn finding(&self, id: &str) -> Option<(&ReviewRound, &ReviewRecord, &Finding)> {
        self.reviews.iter().find_map(|round| {
            round.reviews.iter().find_map(|rec| {
                rec.findings
                    .iter()
                    .find(|f| f.id == id)
                    .map(|f| (round, rec, f))
            })
        })
    }

    /// Did this run reach a mergeable status (`Ready` or `Merged`) with
    /// review findings still open?
    ///
    /// That combination is the point of the review hand-off: the review
    /// round budget (or an unproductive round, see [`ReviewRound::progressed`])
    /// was spent while gate and e2e stayed green, so the run was handed off
    /// rather than blocked — but the findings did not disappear, and whoever
    /// reads the result should be told they are still there.
    pub fn handed_off_with_open_findings(&self) -> bool {
        matches!(self.status, RunStatus::Ready | RunStatus::Merged)
            && self.reviews.last().is_some_and(|r| !r.clean)
    }

    /// Reached `Ready` because `[merge] mode = "none"` left it there by
    /// design, never to be picked up by the PR-polling merge watcher — as
    /// opposed to a `Ready` that is still a plausible landing candidate (a
    /// PR closed without merging, or a re-entry onto an already-concluded
    /// node). Both leave `status` at `Ready`; only this one leaves the
    /// winning branch permanently unwatched, which is what a caller needs to
    /// know before labelling the run in a listing.
    pub fn unmerged_by_design(&self) -> bool {
        self.status == RunStatus::Ready
            && self
                .merge
                .as_ref()
                .is_some_and(|m| m.mode == MergeMode::None)
    }

    /// Did a step after the merge leave something only a human can finish?
    ///
    /// Derived, not a status: the merge happened, so `status` stays `Merged`
    /// and everything that branches on it keeps working. Display code asks
    /// this so a merged run with an unmerged release PR does not read as
    /// plain green.
    pub fn needs_attention(&self) -> bool {
        self.status == RunStatus::Merged
            && self
                .release_bump
                .as_ref()
                .is_some_and(|b| b.action_required.is_some())
    }

    /// Local-time creation stamp for reports.
    pub fn created_local(&self) -> String {
        self.created_at
            .to_zoned(jiff::tz::TimeZone::system())
            .strftime("%Y-%m-%d %H:%M:%S")
            .to_string()
    }

    /// Assert that this run is safe to delete.
    ///
    /// Refuses a run a live daemon is working on, and refuses any run whose
    /// candidate worktrees and branches have not been folded away with `magi
    /// fold`. The fold requirement is the real protection: it is what makes
    /// "delete" mean "remove a record" rather than "throw away a worktree
    /// somebody may still be editing".
    ///
    /// `in_flight` has to come from the caller, because a run's own status
    /// cannot answer the question. A daemon killed mid-run leaves its status at
    /// `implementing` forever, and a guard that trusted that would make every
    /// interrupted run permanently undeletable - the operator's only recourse
    /// being to edit `run.json` by hand, which is exactly the sort of thing
    /// this command exists to avoid. The queue already treats an orphaned
    /// `.lock` from a `SIGKILL`ed daemon the same way; this is that rule for
    /// runs.
    pub fn ensure_can_delete(&self, in_flight: bool) -> Result<()> {
        if in_flight {
            bail!(
                "run {} is being worked on by a live daemon right now",
                self.short()
            );
        }
        if self.candidates.iter().any(|c| !c.folded) {
            bail!(
                "run {} has unfolded candidates; fold first with `magi fold`",
                self.short()
            );
        }
        Ok(())
    }
}

/// The short form of a commit, for a label a human or an LLM reads.
fn short(commit: &str) -> String {
    commit.chars().take(7).collect()
}

/// The short form of a run id: the trailing block after the last `-`.
///
/// A free function as well as [`RunState::short`], because callers that have
/// only an id - an error message, a daemon status, a route handler - were
/// otherwise reimplementing the split, and two spellings of "short id" is one
/// rename away from branch names that no longer match their run.
pub fn short_of(id: &str) -> &str {
    id.split('-').next_back().unwrap_or(id)
}

/// Where magi keeps its runs.
///
/// `MAGI_HOME` overrides the default, and [`set_home`] overrides both — which
/// is what lets the integration tests drive a whole graph without writing into
/// the operator's real history.
///
/// In a unit test build (`cfg(test)`), falling through to the real
/// `<data_local>/magi` is not a fallback worth having: it is exactly how
/// three broken fixture runs ended up in the operator's actual history and
/// were counted as `unreadable` by the deck. A test that reaches this point
/// forgot to call [`set_home`] (or set `MAGI_HOME`) - that is a bug in the
/// test, not a case to serve, so it panics instead of writing anywhere.
pub fn home() -> PathBuf {
    resolve_home(HOME.get().cloned(), std::env::var_os("MAGI_HOME"))
}

/// The decision `home` makes, taking its two overrides as plain values
/// instead of reading the `OnceLock` and the environment itself.
///
/// Pulled out so the `cfg(test)` panic is asserted directly against a
/// `None, None` input, rather than racing every other unit test in the
/// binary for who touches the process-global `HOME` first.
fn resolve_home(pinned: Option<PathBuf>, magi_home_env: Option<std::ffi::OsString>) -> PathBuf {
    if let Some(dir) = pinned {
        return dir;
    }
    if let Some(dir) = magi_home_env {
        return PathBuf::from(dir);
    }
    #[cfg(test)]
    {
        panic!(
            "run::home() was reached in a test without run::set_home() or \
             MAGI_HOME; this would write into the operator's real \
             <data_local>/magi. Call `run::set_home(temp_dir)` before any \
             code path that touches a RunState."
        );
    }
    #[cfg(not(test))]
    {
        dirs::data_local_dir()
            .unwrap_or_else(|| PathBuf::from("."))
            .join("magi")
    }
}

/// [`home`] without the test panic: `None` in a test that never pinned a
/// home, so best-effort writers (see [`crate::notices::raise`]) skip the write
/// instead of touching the operator's real state or aborting the test.
pub fn try_home() -> Option<PathBuf> {
    if cfg!(test) {
        HOME.get()
            .cloned()
            .or_else(|| std::env::var_os("MAGI_HOME").map(PathBuf::from))
    } else {
        Some(home())
    }
}

/// Pin the run home for this process. The first call wins.
pub fn set_home(dir: PathBuf) {
    let _ = HOME.set(dir);
}

/// A temp directory unique to this test process, created once and kept.
/// `set_home` is a first-wins `OnceLock`, so a fixed `temp_dir().join(..)`
/// would be shared by every concurrent `cargo test` process on the machine.
#[cfg(test)]
pub(crate) fn test_home() -> PathBuf {
    static DIR: std::sync::OnceLock<PathBuf> = std::sync::OnceLock::new();
    DIR.get_or_init(|| {
        tempfile::Builder::new()
            .prefix("magi-unit-home-")
            .tempdir()
            .expect("create the unit-test home")
            .keep()
    })
    .clone()
}

/// [`set_home`] on [`test_home`]; the one way a unit test pins the home.
#[cfg(test)]
pub(crate) fn pin_test_home() {
    set_home(test_home());
}

static HOME: std::sync::OnceLock<PathBuf> = std::sync::OnceLock::new();

/// `<home>/runs`.
pub fn runs_root() -> PathBuf {
    home().join("runs")
}

/// The worktree root a run uses when the config sets none: `~/wt/magi`.
///
/// One definition of the default, so the folder the janitor folds and the
/// folder the health view sizes cannot drift apart: a run with no configured
/// [`crate::config::Graph::worktree_root`] lays its worktrees exactly here.
pub fn default_worktree_root() -> PathBuf {
    dirs::home_dir()
        .unwrap_or_else(|| PathBuf::from("."))
        .join("wt")
        .join("magi")
}

/// Directory for one run id.
pub fn run_dir(id: &str) -> PathBuf {
    runs_root().join(id)
}

/// Every run id on disk, newest first.
///
/// A directory is a run because of its **name**, not because it holds a
/// readable `run.json`. A run whose very first save lost the machine's last
/// free bytes leaves `<id>/run.json.tmp` and nothing else, and filtering on
/// `run.json` made that run invisible everywhere: not in `magi list`, not in
/// `runs_unreadable`, not on the phone, so nothing could report it and no
/// route could clear it. `88c0` sat like that for two days. Unreadable is
/// counted, never hidden - the readers already say why each one cannot be
/// read, and `fold_unreadable` is how a record like this leaves.
pub fn list_ids() -> Vec<String> {
    list_ids_in(&runs_root())
}

/// [`list_ids`] against an explicit runs root, for callers (and tests) that
/// must not depend on the process-global home.
pub fn list_ids_in(root: &Path) -> Vec<String> {
    let mut ids: Vec<String> = std::fs::read_dir(root)
        .into_iter()
        .flatten()
        .flatten()
        .filter(|e| e.path().is_dir())
        .map(|e| e.file_name().to_string_lossy().into_owned())
        .filter(|name| is_run_id(name))
        .collect();
    // Ids start with a sortable timestamp.
    ids.sort_unstable_by(|a, b| b.cmp(a));
    ids
}

/// Does `name` have the shape [`new_id`] mints: `YYYYMMDD-HHMMSS-xxxx`?
///
/// The test for "this directory is a run", so a stray folder under
/// `<home>/runs` is not reported as a broken run.
///
/// The tag is checked for length and for being alphanumeric, not for being
/// hex: real ids are hex, but fixtures across this crate name runs
/// `...-dead` / `...-gone` / `...-once`, and a predicate that disowned those
/// would be asserting the fixtures' spelling rather than the shape.
pub fn is_run_id(name: &str) -> bool {
    let mut parts = name.split('-');
    let (Some(day), Some(time), Some(tag), None) =
        (parts.next(), parts.next(), parts.next(), parts.next())
    else {
        return false;
    };
    day.len() == 8
        && day.bytes().all(|b| b.is_ascii_digit())
        && time.len() == 6
        && time.bytes().all(|b| b.is_ascii_digit())
        && tag.len() == 4
        && tag.bytes().all(|b| b.is_ascii_alphanumeric())
}

/// Expand an id prefix to exactly one run id.
pub fn resolve_id(prefix: &str) -> Result<String> {
    // A whole id names its directory, readable state or not: the run whose
    // `run.json` never landed still has to be reachable by `magi show` and
    // by the fold route, which is the only way its record ever leaves.
    if is_run_id(prefix) && run_dir(prefix).is_dir() {
        return Ok(prefix.to_owned());
    }
    let hits: Vec<String> = list_ids()
        .into_iter()
        .filter(|id| id.starts_with(prefix) || id.ends_with(prefix))
        .collect();
    match hits.len() {
        1 => Ok(hits.into_iter().next().expect("exactly one hit")),
        0 => bail!("no run matches `{prefix}`"),
        _ => bail!(
            "`{prefix}` matches {} runs: {}",
            hits.len(),
            hits.join(", ")
        ),
    }
}

/// The most recent run, if any.
pub fn latest_id() -> Option<String> {
    list_ids().into_iter().next()
}

/// An unreserved `YYYYMMDD-HHMMSS-xxxx` candidate, sortable and branch-safe.
/// Fresh entropy is independent of the blind seed. Entropy alone cannot prevent
/// collisions: production creation must use `RunState::reserve` before the id
/// escapes, reserving both the full id and its branch/worktree suffix on disk.
fn new_id() -> String {
    let stamp = Zoned::now().strftime("%Y%m%d-%H%M%S");
    let entropy = crate::rng::entropy();
    format!("{stamp}-{:04x}", (entropy ^ (entropy >> 32)) & 0xffff)
}

/// Keep the last `max` bytes of `text`, on a line boundary.
pub fn tail(text: &str, max: usize) -> String {
    if text.len() <= max {
        return text.to_owned();
    }
    let mut cut = text.len() - max;
    while cut < text.len() && !text.is_char_boundary(cut) {
        cut += 1;
    }
    let slice = &text[cut..];
    let start = slice.find('\n').map_or(0, |i| i + 1);
    format!(
        "[... {} earlier bytes omitted ...]\n{}",
        cut,
        &slice[start..]
    )
}

/// Bytes of verification output recorded and handed to fixers/reviewers.
pub const VERIFY_SUMMARY: usize = 32_768;

/// A backtrace frame line (`  41: 0x102e80de0 - <path>::{closure#0}` or
/// `  3: std::panicking::begin_panic`), or the `at file:line` line under one.
/// Deliberately narrow: a frame needs a number, a colon and either an address
/// or a path, so ordinary `12: something` output is never dropped.
fn is_frame(line: &str, after_frame: bool) -> bool {
    let t = line.trim_start();
    if after_frame && t.starts_with("at ") {
        return true;
    }
    let digits = t.bytes().take_while(u8::is_ascii_digit).count();
    if digits == 0 || !t[digits..].starts_with(": ") {
        return false;
    }
    let rest = t[digits + 2..].trim_start();
    rest.starts_with("0x") || rest.contains("::")
}

/// Priority (lower first) and following context (assertion text, `left`/`right`).
fn trigger_context(line: &str) -> Option<(u8, usize)> {
    // Match colored nextest markers without changing the stored evidence.
    let mut plain = String::new();
    let mut chars = line.chars();
    while let Some(c) = chars.next() {
        if c == '\x1b' && chars.clone().next() == Some('[') {
            chars.next();
            for c in chars.by_ref() {
                if ('@'..='~').contains(&c) {
                    break;
                }
            }
        } else {
            plain.push(c);
        }
    }
    let t = plain.trim();
    if t.starts_with("[full output:") || t.starts_with("[full output could not be saved:") {
        Some((0, 0))
    } else if t.contains("panicked at") {
        Some((0, 6))
    } else if (t.starts_with("---- ") && t.ends_with(" ----"))
        || t.starts_with("--- STDERR:")
        || t.starts_with("--- STDOUT:")
    {
        Some((1, 3))
    } else if t.starts_with("FAIL ") || t.contains("FAIL [") {
        Some((1, 4))
    } else if t.starts_with("error:")
        || t.starts_with("error[")
        || t.starts_with("failures:")
        || t.contains("test result: FAILED")
    {
        Some((2, 4))
    } else {
        None
    }
}

fn clip_line(line: &str, max: usize) -> &str {
    if line.len() <= max {
        return line;
    }
    let mut cut = max;
    while !line.is_char_boundary(cut) {
        cut -= 1;
    }
    &line[..cut]
}

/// Summarise verification output to roughly `max` bytes, keeping the reason a
/// command failed. A plain tail is useless for a failing test: the panic
/// message sits hundreds of KB before the end, behind a backtrace. So:
///
/// 1. backtrace frame lines are dropped (counted in the header);
/// 2. if that already fits, the rest is returned whole;
/// 3. otherwise keep excerpts, each from a failure marker (`panicked at`,
///    `---- test stdout ----`, nextest `FAIL`/`STDERR`, `error:`, ...) through
///    a few following lines (the assertion and `left`/`right`), share the
///    budget evenly between selected excerpts. Select panics first, then test
///    stdout/stderr and nextest failure blocks, then generic diagnostics: noisy
///    compiler errors must not crowd out the test failure reason. Within each
///    priority keep earlier excerpts, then render in original order and append
///    the final tail (where `test result:` lives).
///
/// Adjacent excerpts merge only within the same priority and while small:
/// a long stdout line must not
/// consume the budget of a following panic anchor. Short output is returned
/// unchanged. The full text belongs in an artifact.
pub fn summarize_verify(text: &str, max: usize) -> String {
    if text.len() <= max {
        return text.to_owned();
    }
    // Tiny callers cannot fit the explanatory header. Respect their exact
    // byte bound, including UTF-8 boundaries, without a larger tail marker.
    if max < 256 {
        return clip_line(text, max).to_owned();
    }
    let mut kept: Vec<&str> = Vec::new();
    let mut dropped = 0usize;
    let mut prev_frame = false;
    for line in text.lines() {
        let frame = is_frame(line, prev_frame);
        if frame {
            dropped += 1;
        } else {
            kept.push(line);
        }
        prev_frame = frame;
    }
    let total: usize = kept.iter().map(|l| l.len() + 1).sum();
    let note = if dropped > 0 {
        format!(" ({dropped} backtrace frame line(s) dropped)")
    } else {
        String::new()
    };
    let overhead = 300;
    if total + overhead <= max {
        return format!("[... output reduced{note} ...]\n{}\n", kept.join("\n"));
    }
    // The tail comes first so that excerpts never repeat it.
    let tail_budget = (max / 4).min(4_096);
    let mut tail_start = kept.len();
    let mut used = 0usize;
    while tail_start > 0 && used + kept[tail_start - 1].len() < tail_budget {
        tail_start -= 1;
        used += kept[tail_start].len() + 1;
    }
    // Excerpts as merged half-open line ranges.
    let mut ranges: Vec<(usize, usize, u8)> = Vec::new();
    for (i, line) in kept[..tail_start].iter().enumerate() {
        if let Some((priority, after)) = trigger_context(line) {
            let end = (i + 1 + after).min(tail_start);
            match ranges.last_mut() {
                Some(last)
                    if priority == last.2
                        && i <= last.1
                        && end - last.0 <= 20
                        && kept[last.0..end].iter().map(|l| l.len() + 1).sum::<usize>() <= 1024 =>
                {
                    last.1 = last.1.max(end);
                }
                _ => ranges.push((i, end, priority)),
            }
        }
    }
    let budget = max.saturating_sub(used + overhead);
    let shown = ranges.len().min((budget / 240).max(1));
    ranges.sort_by_key(|&(start, _, priority)| (priority, start));
    let mut selected = ranges[..shown].to_vec();
    selected.sort_by_key(|&(start, _, _)| start);
    let per = budget / shown.max(1);
    let mut body = String::new();
    for &(a, b, _) in &selected {
        let mut block = String::new();
        for line in &kept[a..b] {
            let remaining = per.saturating_sub(block.len() + 1);
            if remaining == 0 {
                break;
            }
            let line = clip_line(line, 600.min(remaining));
            block.push_str(line);
            block.push('\n');
        }
        body.push_str(&block);
        body.push_str("[...]\n");
    }
    let hidden = ranges.len() - shown;
    let more = if hidden > 0 {
        format!("; {hidden} more excerpt(s) not shown")
    } else {
        String::new()
    };
    let tail_text: String = kept[tail_start..].iter().flat_map(|l| [*l, "\n"]).collect();
    let out = format!(
        "[... output reduced from {} bytes{note}; {shown} failure excerpt(s) kept{more}, then the end ...]\n{body}{tail_text}",
        text.len(),
    );
    if out.len() > max {
        return clip_line(&out, max).to_owned();
    }
    out
}

/// Reserve a fresh verification artifact atomically, so concurrent writes,
/// retries and resumes cannot overwrite earlier evidence.
pub fn write_verification_artifact(run: &RunState, stem: &str, body: &str) -> Result<String> {
    use std::io::Write as _;
    let dir = run.dir().join("artifacts");
    std::fs::create_dir_all(&dir)?;
    for sequence in 1.. {
        let name = format!("{stem}-{sequence}.out");
        match std::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(dir.join(&name))
        {
            Ok(mut file) => {
                file.write_all(body.as_bytes())
                    .with_context(|| format!("write artifacts/{name}"))?;
                return Ok(name);
            }
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(e) => return Err(e).with_context(|| format!("create artifacts/{name}")),
        }
    }
    unreachable!()
}

/// Path of a run artifact.
pub fn artifact_path(run: &RunState, name: &str) -> PathBuf {
    run.dir().join("artifacts").join(name)
}

/// Write an artifact, creating the directory if needed.
pub fn write_artifact(run: &RunState, name: &str, body: &str) -> Result<PathBuf> {
    let path = artifact_path(run, name);
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?;
    }
    std::fs::write(&path, body).with_context(|| format!("write {}", path.display()))?;
    Ok(path)
}

/// Read an artifact back, e.g. a stored patch on resume.
pub fn read_artifact(run: &RunState, name: &str) -> Option<String> {
    std::fs::read_to_string(artifact_path(run, name)).ok()
}

#[cfg(test)]
mod tests {
    #[test]
    fn summarize_verify_preserves_failure_context_and_bounds() {
        let mut text = String::from("[full output: artifacts/gate-1.out]\n");
        for name in ["first", "second"] {
            text.push_str(&format!("---- {name} stdout ----\nmy test output\n"));
            text.push_str(&format!("thread '{name}' panicked at tests/check.rs:42:9:\nassertion `left == right` failed\n  left: 1\n right: 2\n"));
            for frame in 0..4000 {
                text.push_str(&format!(
                    "  {frame}: 0x102e80de0 - backtrace frame with 日本語\n"
                ));
            }
        }
        text.push_str("---- second stderr ----\nsecond test stderr\n");
        text.push_str(&"unrelated output\n".repeat(1000));
        text.push_str(
            "\x1b[31mFAIL [0.01s] suite third\x1b[0m\n--- STDOUT: ---\nthird test stdout\n",
        );
        text.push_str(&"stdout noise\n".repeat(10000));
        text.push_str("--- STDERR: ---\nthird test stderr\n");
        text.push_str("thread 'third' panicked at test.rs:1:\nthird assertion\n");
        text.push_str(&"backtrace noise\n".repeat(10000));
        text.push_str("test result: FAILED. 0 passed; 3 failed\n");
        let summary = summarize_verify(&text, 32 * 1024);
        assert!(summary.len() <= 32 * 1024);
        for expected in [
            "thread 'first' panicked at",
            "thread 'second' panicked at",
            "thread 'third' panicked at",
            "my test output",
            "second test stderr",
            "FAIL [0.01s] suite third",
            "third test stdout",
            "third test stderr",
            "panicked at tests/check.rs:42:9",
            "left: 1",
            "right: 2",
            "third assertion",
            "test result: FAILED",
            "artifacts/gate-1.out",
        ] {
            assert!(summary.contains(expected), "missing {expected}: {summary}");
        }
        assert_eq!(summarize_verify(&summary, 32 * 1024), summary);
        let compact = summarize_verify(&summary, 2000);
        assert!(compact.len() <= 2000);
        assert!(compact.contains("panicked at"));
        assert!(compact.contains("artifacts/gate-1.out"));
        assert_eq!(summarize_verify("short 日本語\n", 1024), "short 日本語\n");
    }

    #[test]
    fn summarize_verify_prioritizes_late_test_failures_over_diagnostics() {
        let mut text = String::new();
        for _ in 0..130 {
            text.push_str("error: diagnostic\n");
            text.push_str(&"ordinary build log\n".repeat(21));
        }
        text.push_str("---- late_test stdout ----\ntest stdout summary\n");
        text.push_str("thread 'late_test' panicked at tests/late.rs:42:9:\nassertion `left == right` failed\n  left: 1\n right: 2\n");
        text.push_str("--- STDERR: ---\ntest stderr summary\n");
        text.push_str(&"ordinary test log\n".repeat(4000));
        text.push_str("test result: FAILED\n");

        let summary = summarize_verify(&text, 32 * 1024);
        assert!(summary.len() <= 32 * 1024);
        for expected in [
            "---- late_test stdout ----",
            "test stdout summary",
            "panicked at tests/late.rs:42:9:",
            "assertion `left == right` failed",
            "left: 1",
            "right: 2",
            "test stderr summary",
            "test result: FAILED",
        ] {
            assert!(summary.contains(expected), "missing {expected}");
        }
    }

    #[test]
    fn summarize_verify_reports_overflow_and_handles_long_utf8_lines() {
        let text = (0..1000)
            .map(|i| {
                format!(
                    "thread '{i}' panicked at test.rs:1:\n{}\n",
                    "あ".repeat(2000)
                )
            })
            .collect::<String>();
        for max in (0..=256).chain([800, 2000, 32 * 1024]) {
            let summary = summarize_verify(&text, max);
            assert!(summary.len() <= max);
            if max >= 256 {
                assert!(summary.contains("more excerpt(s) not shown"));
            }
        }
        let compiler = format!(
            "error[E0063]: missing field\n{}done",
            "noise\n".repeat(20000)
        );
        assert!(summarize_verify(&compiler, 2000).contains("error[E0063]"));
    }

    #[test]
    fn verification_artifacts_reserve_names_atomically() {
        pin_test_home();
        let state = pass_state();
        let names = std::thread::scope(|scope| {
            let handles: Vec<_> = (0..8)
                .map(|i| {
                    let state = &state;
                    scope.spawn(move || {
                        let body = format!("output {i}");
                        let name = write_verification_artifact(state, "gate", &body).unwrap();
                        (name, body)
                    })
                })
                .collect();
            handles
                .into_iter()
                .map(|h| h.join().unwrap())
                .collect::<Vec<_>>()
        });
        let unique: BTreeSet<_> = names.iter().map(|(name, _)| name).collect();
        assert_eq!(unique.len(), names.len());
        for (name, body) in names {
            assert_eq!(read_artifact(&state, &name).unwrap(), body);
        }
    }

    fn pass_state() -> RunState {
        RunState::new(
            PathBuf::from("/repo"),
            "main".to_owned(),
            "0123456789abcdef".to_owned(),
            "Do it".to_owned(),
            Config::default(),
        )
    }

    #[test]
    fn reserving_runs_retries_full_and_short_collisions_without_touching_the_first() {
        let home = tempfile::tempdir().unwrap();
        let first_id = "20260101-000000-abcd";
        let ids = crate::id_reserve::sequence(&[
            first_id,
            first_id,
            "20260102-000000-abcd",
            "20260101-000000-1234",
        ]);
        let mut first = pass_state();
        first.reserve_with(home.path(), &ids).unwrap();
        let first_dir = home.path().join("runs").join(&first.id);
        std::fs::create_dir(first_dir.join("artifacts")).unwrap();
        std::fs::write(first_dir.join("artifacts/keep"), b"original artifact").unwrap();
        let before = std::fs::read(first_dir.join("run.json")).unwrap();
        let mut second = pass_state();
        second.instruction = "independent".into();
        second.reserve_with(home.path(), &ids).unwrap();
        assert_eq!(first.id, first_id);
        assert_eq!(second.id, "20260101-000000-1234");
        assert_ne!(first.branch_for('A'), second.branch_for('A'));
        assert_eq!(std::fs::read(first_dir.join("run.json")).unwrap(), before);
        assert_eq!(
            std::fs::read(first_dir.join("artifacts/keep")).unwrap(),
            b"original artifact"
        );
        assert_eq!(
            RunState::load_under(&second.id, home.path())
                .unwrap()
                .instruction,
            "independent"
        );
        assert!(!home.path().join("runs/20260102-000000-abcd").exists());
        let stuck = crate::id_reserve::IdSource::custom(|| "20260101-000000-abcd".into());
        assert!(
            pass_state()
                .reserve_with(home.path(), &stuck)
                .unwrap_err()
                .to_string()
                .contains("256 attempts")
        );
        assert_eq!(std::fs::read(first_dir.join("run.json")).unwrap(), before);
    }

    #[test]
    fn reserving_a_run_fails_when_the_store_cannot_be_created() {
        let home = tempfile::tempdir().unwrap();
        std::fs::write(home.path().join("runs"), "not a directory").unwrap();
        assert!(pass_state().reserve_under(home.path()).is_err());
    }

    #[test]
    fn passes_append_and_a_resume_never_rewrites_an_earlier_cause() {
        let mut s = pass_state();
        s.begin_pass();
        assert!(!s.passes[0].resumed);
        s.end_pass_parked(ParkCause::Upgrade {
            from: "0.1.0".to_owned(),
            to: Some("0.1.1".to_owned()),
        });
        s.begin_pass();
        s.end_pass_parked(ParkCause::Operator);
        // A second request for the same boundary does not relabel it.
        s.end_pass_parked(ParkCause::Question);
        s.begin_pass();
        assert_eq!(s.passes.len(), 3);
        assert!(s.passes[1].resumed);
        assert_eq!(
            s.passes[0].end,
            Some(PassEnd::Parked {
                cause: ParkCause::Upgrade {
                    from: "0.1.0".to_owned(),
                    to: Some("0.1.1".to_owned()),
                }
            })
        );
        assert_eq!(
            s.passes[1].end,
            Some(PassEnd::Parked {
                cause: ParkCause::Operator
            })
        );
        assert_eq!(s.passes[2].end, None);
    }

    #[test]
    fn a_pass_left_open_is_closed_as_unrecorded_by_the_next() {
        let mut s = pass_state();
        s.begin_pass();
        s.begin_pass();
        assert_eq!(s.passes[0].end, Some(PassEnd::Unrecorded));
        assert!(s.passes[0].ended_at.is_some());
        assert_eq!(s.passes[1].end, None);
    }

    #[test]
    fn a_state_written_before_passes_loads_with_none() {
        let s = pass_state();
        let mut v = serde_json::to_value(&s).unwrap();
        v.as_object_mut().unwrap().remove("passes");
        let back: RunState = serde_json::from_value(v).unwrap();
        assert!(back.passes.is_empty());
    }
    #[test]
    fn test_home_is_private_stable_and_unique_per_process() {
        // Child mode: report this process's home and stop.
        if std::env::var_os("MAGI_HOME_PROBE").is_some() {
            println!("HOME={}", super::test_home().display());
            return;
        }
        let a = super::test_home();
        assert_eq!(a, super::test_home());
        assert!(a.is_dir() && a.starts_with(std::env::temp_dir()));
        let probe = || {
            let out = std::process::Command::new(std::env::current_exe().unwrap())
                .args([
                    "--exact",
                    "run::tests::test_home_is_private_stable_and_unique_per_process",
                    "--nocapture",
                    "--test-threads=1",
                ])
                .env("MAGI_HOME_PROBE", "1")
                .quiet()
                .output()
                .unwrap();
            String::from_utf8_lossy(&out.stdout)
                .lines()
                .find_map(|l| l.split_once("HOME=").map(|(_, h)| h.to_owned()))
                .expect("child reported a home")
        };
        let (b, c) = (probe(), probe());
        assert_ne!(b, c, "two processes must never share a unit-test home");
        assert_ne!(b, a.display().to_string());
    }

    #[test]
    fn no_test_pins_a_fixed_temp_path_as_the_home() {
        let root = std::env::var_os("CARGO_MANIFEST_DIR")
            .map(PathBuf::from)
            .unwrap_or_else(|| PathBuf::from(env!("CARGO_MANIFEST_DIR")));
        let needle = ["set_home(std::env::temp_dir()", ".join(\"magi-"].concat();
        for e in std::fs::read_dir(root.join("src")).unwrap() {
            let p = e.unwrap().path();
            if p.extension().and_then(|x| x.to_str()) != Some("rs") {
                continue;
            }
            let text = std::fs::read_to_string(&p).unwrap();
            assert!(
                !text.contains(&needle),
                "{} pins a fixed temp home; use run::pin_test_home()",
                p.display()
            );
        }
    }

    use super::*;

    fn state() -> RunState {
        RunState::new(
            PathBuf::from("/repo"),
            "main".to_owned(),
            "abc1234def".to_owned(),
            "add retries".to_owned(),
            Config::default(),
        )
    }

    #[test]
    fn task_public_head_contains_only_the_task_identity() {
        let mut run = state();
        run.origin = Some(Origin::queue("abcd-1234"));
        assert_eq!(run.task_pr_branch().as_deref(), Some("magi/task-abcd-1234"));
        assert_ne!(run.task_pr_branch().unwrap(), run.branch_for('A'));
        run.origin = Some(Origin::queue("bad..task"));
        assert!(run.task_pr_branch().is_none());
        run.origin = Some(Origin::queue("abcd-1234"));
        run.reviewed_commits = Some(Vec::new());
        assert!(run.task_pr_branch().is_none());
        run.reviewed_commits = None;
        run.origin = Some(Origin::operator());
        assert!(run.task_pr_branch().is_none());
    }

    #[test]
    fn review_run_keeps_the_task_public_head_only_when_inherited() {
        let mut run = state();
        run.origin = Some(Origin::queue("abcd-1234"));
        run.reviewed_commits = Some(Vec::new());
        let mut cand = |branch: &str| {
            run.candidates = vec![Candidate {
                index: 0,
                label: 'A',
                agent: "(existing branch)".to_owned(),
                branch: branch.to_owned(),
                worktree: PathBuf::from("/w"),
                summary: String::new(),
                stat: String::new(),
                files: 1,
                commits: 1,
                empty: false,
                failed: None,
                verified_noop: None,
                duration_ms: 0,
                folded: false,
            }];
        };
        cand("magi/ab12/A");
        run.tally = Some(Tally {
            first_choice: BTreeMap::from([('A', 0)]),
            borda: BTreeMap::new(),
            winner: 'A',
            rankings: 0,
            unanimous_initial: false,
            deliberated: false,
            changed_votes: 0,
            unanimous_final: false,
            tie_break: None,
            judges: 0,
            present: 0,
            quorum: 0,
            met_quorum: true,
            uncontested: None,
        });
        // A legacy candidate-branch PR: nothing inherited, keep its own head.
        assert!(run.task_pr_branch().is_none());
        run.pr_head = Some("magi/task-abcd-1234".to_owned());
        assert_eq!(run.task_pr_branch().as_deref(), Some("magi/task-abcd-1234"));
        // A head that is not this task's is never followed.
        run.pr_head = Some("magi/task-other".to_owned());
        assert!(run.task_pr_branch().is_none());
    }

    #[test]
    fn superseded_is_terminal_but_not_resumable() {
        assert!(RunStatus::Superseded.done());
        assert!(!RunStatus::Superseded.resumable());
        assert_eq!(RunStatus::Superseded.as_str(), "superseded");
        assert!(RunStatus::AlreadyInBase.done());
        assert!(!RunStatus::AlreadyInBase.resumable());
        assert_eq!(RunStatus::AlreadyInBase.as_str(), "already_in_base");
    }

    #[test]
    fn load_under_reads_back_exactly_what_save_under_wrote_at_an_explicit_home() {
        // Both rooted at an explicit `home` rather than the process-global
        // one - a caller with its own `home` (a test fixture, a housekeeping
        // pass) must round-trip through exactly that directory, never
        // through whichever home some other test in the same binary pinned
        // into the global `OnceLock` first.
        let dir = tempfile::tempdir().unwrap();
        let mut original = state();
        original.id = "20260101-000000-load".to_owned();
        original.status = RunStatus::Blocked;
        original.save_under(dir.path()).unwrap();

        let reloaded = RunState::load_under(&original.id, dir.path()).unwrap();
        assert_eq!(reloaded.id, original.id);
        assert_eq!(reloaded.status, RunStatus::Blocked);

        assert!(
            RunState::load_under("20260101-000000-none", dir.path()).is_err(),
            "an id with nothing saved under this home must not silently read something else"
        );
    }

    #[test]
    fn verification_history_legacy_metadata_remains_unknown() {
        let history: VerificationEvidence = serde_json::from_value(serde_json::json!({
            "gate_head": "old-head",
            "gate_ran": true,
            "gate": [],
            "e2e": null
        }))
        .unwrap();
        assert!(history.superseded_by.is_none());
        assert!(history.archived_at.is_none());
        assert!(history.fixes_spent.is_none());
        assert_eq!(history.gate_head.as_deref(), Some("old-head"));
        assert!(history.gate_ran);
    }

    #[test]
    fn resolve_home_prefers_the_pin_then_the_env_var() {
        let pinned = PathBuf::from("/pinned");
        assert_eq!(
            resolve_home(Some(pinned.clone()), Some("/env".into())),
            pinned,
            "a pin wins even over MAGI_HOME"
        );
        assert_eq!(
            resolve_home(None, Some("/env".into())),
            PathBuf::from("/env")
        );
    }

    #[test]
    #[should_panic(expected = "run::set_home()")]
    fn resolve_home_refuses_to_fall_back_to_the_operators_real_home() {
        // Neither override present is exactly the state a test reaches by
        // forgetting `set_home`/`MAGI_HOME` - the accident that put three
        // broken fixture runs into the operator's real history. Asserted
        // against the pure decision directly, not `home()` itself, because
        // `HOME` is a process-wide `OnceLock` another test may have already
        // set - this must not depend on test execution order.
        resolve_home(None, None);
    }

    #[test]
    fn a_run_is_named_by_shape_so_a_state_less_directory_is_still_a_run() {
        // The shape `new_id` mints. A directory answering to it is a run even
        // with no readable `run.json`: that is how a save that ran out of
        // disk stays visible instead of vanishing from every listing.
        assert!(is_run_id(&new_id()));
        assert!(is_run_id("20260904-014540-88c0"));
        // Not runs: a stray folder, a truncated id, a non-hex tag, and an id
        // with an extra segment (a worktree label, say).
        assert!(!is_run_id("scratch"));
        assert!(!is_run_id("20260904-014540"));
        assert!(!is_run_id("20260904-014540-88c0f"));
        assert!(!is_run_id("2026090x-014540-88c0"));
        assert!(!is_run_id("20260904-014540-88c0-A"));
    }

    #[test]
    fn ids_are_sortable_and_short_suffixed() {
        let s = state();
        let parts: Vec<&str> = s.id.split('-').collect();
        assert_eq!(parts.len(), 3);
        assert_eq!(parts[0].len(), 8);
        assert_eq!(parts[1].len(), 6);
        assert_eq!(parts[2].len(), 4);
        assert_eq!(s.short(), parts[2]);
    }

    #[test]
    fn branch_names_carry_the_label_not_the_author() {
        let s = state();
        let b = s.branch_for('B');
        assert_eq!(b, format!("magi/{}/B", s.short()));
        assert!(!b.contains("claude"));
    }

    /// A pinned seed reproduces the blind decisions. It must **not** reproduce
    /// the run's identity.
    ///
    /// `assert_eq!(a.short(), b.short())` used to stand where the last
    /// assertion is now, and it was pinning the defect: with the id's suffix
    /// derived from the seed, two runs started in the same second were the
    /// same run as far as the filesystem was concerned - one directory, one
    /// `artifacts/`, one set of candidate worktrees. `tests/common` pins a
    /// seed for every integration test, so on Linux, where the suite is fast,
    /// two tests in `graph_dropped_stream` shared a directory and one read the
    /// other's artifact.
    #[test]
    fn a_pinned_seed_is_reproducible_but_never_the_run_id() {
        let mut cfg = Config::default();
        cfg.blind.seed = Some(1234);
        let a = RunState::new(
            PathBuf::from("/r"),
            "main".to_owned(),
            "c".to_owned(),
            "t".to_owned(),
            cfg.clone(),
        );
        let b = RunState::new(
            PathBuf::from("/r"),
            "main".to_owned(),
            "c".to_owned(),
            "t".to_owned(),
            cfg,
        );
        // What the seed is for: the same shuffles, run after run.
        assert_eq!(a.seed, 1234);
        assert_eq!(a.seed, b.seed);
        // What it is not for. Two runs are two runs, in the same second or
        // not, and everything keyed on the id depends on that.
        assert_ne!(
            a.id, b.id,
            "two runs sharing an id share a directory, artifacts and worktrees"
        );
    }

    #[test]
    fn status_terminality() {
        assert!(RunStatus::Merged.done());
        assert!(RunStatus::Blocked.done());
        assert!(!RunStatus::Reviewing.done());
    }

    fn overrun_seat(now: Timestamp, elapsed_secs: i64, timeout_secs: u64) -> ActiveSeat {
        ActiveSeat {
            node: "implement".to_owned(),
            started_at: now - jiff::SignedDuration::new(elapsed_secs, 0),
            timeout_secs,
            attempt: 0,
            task: None,
            command: None,
            index: None,
            total: None,
        }
    }

    #[test]
    fn active_all_overrun_requires_every_seat_past_its_own_timeout() {
        let mut s = state();
        let now = Timestamp::now();
        assert!(
            !s.active_all_overrun(now),
            "nothing active is not evidence of anything"
        );

        s.active
            .insert("impl-A".to_owned(), overrun_seat(now, 21_000, 3_600));
        assert!(
            s.active_all_overrun(now),
            "21000s elapsed against a 3600s budget"
        );

        // A seat still well within its own budget means the run is not
        // provably dead, however far its sibling has overrun.
        s.active
            .insert("impl-B".to_owned(), overrun_seat(now, 0, 3_600));
        assert!(!s.active_all_overrun(now));
    }

    /// A schema-11 run that ended Blocked cannot be pinned by a daemon pid
    /// that is still alive; one still mid-walk keeps asking the pid.
    #[test]
    fn migrating_schema_11_marks_only_ended_runs_as_exited() {
        for (status, exited) in [
            (RunStatus::Blocked, true),
            (RunStatus::Failed, true),
            (RunStatus::Stalled, true),
            (RunStatus::Landing, true),
            (RunStatus::Reviewing, false),
        ] {
            let mut s = state();
            s.schema = 11;
            s.status = status;
            let m = migrate_schema(s).unwrap();
            assert_eq!(m.schema, SCHEMA);
            assert_eq!(m.driver_exited, exited, "{status:?}");
        }
    }

    /// A driver that recorded its own exit reads dead without its pid being
    /// asked: a daemon's pid outlives the runs it drove.
    #[test]
    fn an_exited_driver_is_dead_even_when_its_pid_answers_alive() {
        let mut s = state();
        s.driver_pid = Some(4242);
        s.driver_started_at = Some("1790000000".to_owned());
        s.driver_exited = true;
        let never = |_| -> Option<bool> { panic!("the pid must not be asked") };
        assert_eq!(
            s.liveness_with(false, never, |_| panic!("nor its identity")),
            Liveness::Dead
        );
        assert_eq!(s.liveness_with(true, never, |_| None), Liveness::Live);
    }

    /// A daemon claim wins outright, whatever `driver_pid` or either query
    /// says — the stronger, independently-heartbeating signal. Neither query
    /// closure is even called: a daemon claim short-circuits before either
    /// one, which panicking closures here prove.
    #[test]
    fn liveness_reads_live_from_a_daemon_claim_alone() {
        let mut s = state();
        s.driver_pid = None;
        assert_eq!(
            s.liveness_with(
                true,
                |_| panic!("a daemon claim needs no pid query"),
                |_| panic!("a daemon claim needs no identity query")
            ),
            Liveness::Live,
            "a daemon claim needs no pid to back it up"
        );
    }

    /// The gap `driver_pid` closes: no daemon claim (every manual `magi run`
    /// / `magi review`), but the recorded pid answers alive *and* the
    /// process currently holding it still carries the same start-time
    /// marker this run recorded — proof it is genuinely the same process,
    /// not merely the same number.
    #[test]
    fn liveness_reads_live_from_a_confirmed_pid_with_a_matching_identity() {
        let mut s = state();
        s.driver_pid = Some(4242);
        s.driver_started_at = Some("1790000000".to_owned());
        assert_eq!(
            s.liveness_with(
                false,
                |pid| {
                    assert_eq!(pid, 4242);
                    Some(true)
                },
                |pid| {
                    assert_eq!(pid, 4242);
                    Some("1790000000".to_owned())
                }
            ),
            Liveness::Live
        );
    }

    /// No daemon claim and the recorded pid confirmed gone by the OS itself
    /// — dead outright, and the identity query is never even reached (a
    /// panicking closure proves it), since there is nothing left to
    /// corroborate.
    #[test]
    fn liveness_reads_dead_from_a_confirmed_dead_pid() {
        let mut s = state();
        s.driver_pid = Some(4242);
        assert_eq!(
            s.liveness_with(
                false,
                |_| Some(false),
                |_| panic!("a confirmed-dead pid needs no identity query")
            ),
            Liveness::Dead
        );
    }

    /// The gap this task's review round exists to close: a killed manual
    /// run's pid gets handed to a wholly unrelated later process. `pid_status`
    /// alone would read that as `Live` — the reused pid really is alive —
    /// but the process now holding it started at a different moment than the
    /// one this run recorded, so this must read `Dead`, not `Live`: a
    /// mismatch is exactly as good as proof the original driver is gone.
    #[test]
    fn liveness_reads_dead_when_a_live_pid_no_longer_matches_the_recorded_start_time() {
        let mut s = state();
        s.driver_pid = Some(4242);
        s.driver_started_at = Some("1790000000".to_owned());
        assert_eq!(
            s.liveness_with(false, |_| Some(true), |_| Some("1790005400".to_owned())),
            Liveness::Dead,
            "the pid is alive, but under a different process than the one this run recorded"
        );
    }

    /// Missing information never collapses to `Dead`: an old run with no
    /// `driver_pid` at all, a `driver_pid` this build could not query, a live
    /// pid with no recorded start time to compare (an even older run, before
    /// that field existed), and a live pid whose current identity this build
    /// could not re-query, all read as `Unknown` — never a guess in either
    /// direction.
    #[test]
    fn a_locale_format_marker_on_a_live_pid_reads_unknown_never_dead() {
        let mut s = state();
        s.driver_pid = Some(4242);
        for old in ["日 10/ 4 17:27:03 2026", "Sun Oct  4 17:27:03 2026"] {
            s.driver_started_at = Some(old.to_owned());
            assert_eq!(
                s.liveness_with(
                    false,
                    |_| Some(true),
                    |_| panic!("an old-format marker needs no identity query")
                ),
                Liveness::Unknown,
                "{old}"
            );
        }
    }

    #[test]
    fn integer_markers_compare_live_on_match_and_dead_on_difference() {
        let mut s = state();
        s.driver_pid = Some(4242);
        s.driver_started_at = Some("1790000000".to_owned());
        assert_eq!(
            s.liveness_with(false, |_| Some(true), |_| Some("1790000000".to_owned())),
            Liveness::Live
        );
        assert_eq!(
            s.liveness_with(false, |_| Some(true), |_| Some("1790000001".to_owned())),
            Liveness::Dead
        );
    }

    #[test]
    fn liveness_never_guesses_out_of_missing_information() {
        let mut s = state();
        s.driver_pid = None;
        assert_eq!(
            s.liveness_with(
                false,
                |_| panic!("no pid to query"),
                |_| panic!("no pid to query")
            ),
            Liveness::Unknown,
            "no driver_pid recorded at all — an old run predating this field"
        );

        s.driver_pid = Some(4242);
        assert_eq!(
            s.liveness_with(false, |_| None, |_| panic!("inconclusive already")),
            Liveness::Unknown,
            "a pid to ask, but the platform could not answer for it"
        );

        s.driver_started_at = None;
        assert_eq!(
            s.liveness_with(false, |_| Some(true), |_| Some("anything".to_owned())),
            Liveness::Unknown,
            "a live pid, but no recorded marker to corroborate it against — an old run \
             predating `driver_started_at`"
        );

        s.driver_started_at = Some("1790000000".to_owned());
        assert_eq!(
            s.liveness_with(false, |_| Some(true), |_| None),
            Liveness::Unknown,
            "a live pid and a recorded marker, but the identity re-query itself failed"
        );
    }

    #[test]
    fn abandon_clears_active_and_fails_a_non_terminal_run() {
        let mut s = state();
        s.status = RunStatus::Implementing;
        let now = Timestamp::now();
        s.active
            .insert("impl-A".to_owned(), overrun_seat(now, 21_000, 3_600));

        s.abandon("daemon");

        assert!(s.active.is_empty());
        assert_eq!(s.status, RunStatus::Failed);
        assert!(
            s.events
                .last()
                .expect("an event was logged")
                .message
                .contains("impl-A"),
            "the event names the abandoned seat"
        );
    }

    #[test]
    fn abandon_never_overwrites_a_status_already_terminal() {
        let mut s = state();
        s.status = RunStatus::Ready;
        let now = Timestamp::now();
        s.active
            .insert("impl-A".to_owned(), overrun_seat(now, 21_000, 3_600));

        s.abandon("daemon");

        assert!(s.active.is_empty());
        assert_eq!(
            s.status,
            RunStatus::Ready,
            "a run already done must not be relabelled Failed"
        );
    }

    #[test]
    fn candidate_viability_excludes_empty_and_failed() {
        let mut c = Candidate {
            index: 0,
            label: 'A',
            agent: "a".to_owned(),
            branch: "b".to_owned(),
            worktree: PathBuf::from("/w"),
            summary: String::new(),
            stat: String::new(),
            files: 1,
            commits: 1,
            empty: false,
            failed: None,
            verified_noop: None,
            duration_ms: 0,
            folded: false,
        };
        assert!(c.viable());
        c.empty = true;
        assert!(!c.viable());
        c.empty = false;
        c.failed = Some("timeout".to_owned());
        assert!(!c.viable());
    }

    #[test]
    fn build_failure_is_distinguished_from_a_failing_test() {
        let link_race = CommandOutcome {
            command: "cargo test".to_owned(),
            code: Some(1),
            output_tail: "LINK : fatal error LNK1104: cannot open file \
                          'graph_dirty_tree-71d4dc8e.exe'\n\
                          error: could not compile `magi-cli` (test \"graph_dirty_tree\")"
                .to_owned(),
            duration_ms: 500,
            resource_blocked: false,
        };
        assert!(!link_race.ok());
        assert!(link_race.build_failed());

        let failing_test = CommandOutcome {
            command: "cargo test".to_owned(),
            code: Some(101),
            output_tail: "thread 'it_works' panicked at 'assertion failed'".to_owned(),
            duration_ms: 500,
            resource_blocked: false,
        };
        assert!(!failing_test.ok());
        assert!(
            !failing_test.build_failed(),
            "a real test failure must not be classed as a build failure"
        );

        let passing = CommandOutcome {
            command: "cargo test".to_owned(),
            code: Some(0),
            output_tail: String::new(),
            duration_ms: 500,
            resource_blocked: false,
        };
        assert!(passing.ok());
        assert!(!passing.build_failed());
    }

    #[test]
    fn tail_keeps_the_end_on_a_line_boundary() {
        let text = (0..100).map(|i| format!("line {i}\n")).collect::<String>();
        let t = tail(&text, 40);
        assert!(t.starts_with("[..."));
        assert!(t.ends_with("line 99\n"));
        assert!(t.len() < 120);
        assert_eq!(tail("short", 40), "short");
    }

    #[test]
    fn summarize_verify_keeps_panics_behind_a_huge_backtrace() {
        let mut text =
            String::from("running 1 test\ntest a ... FAILED\n\nfailures:\n\n---- a stdout ----\n");
        text.push_str("thread 'a' panicked at tests/a.rs:9:5:\nassertion `left == right` failed\n  left: 1\n right: 2\nstack backtrace:\n");
        for i in 0..6000 {
            text.push_str(&format!("  {i}: 0x102e80de0 - <web_render::x::{{closure#0}}>\n             at /rustc/abc/library/core/src/ops/function.rs:250:5\n"));
        }
        text.push_str("\nfailures:\n    a\n\ntest result: FAILED. 0 passed; 1 failed\n");
        assert!(text.len() > 400_000);
        let s = summarize_verify(&text, VERIFY_SUMMARY);
        assert!(s.contains("panicked at tests/a.rs:9:5"));
        assert!(s.contains("left: 1") && s.contains("right: 2"));
        assert!(s.contains("---- a stdout ----"));
        assert!(s.contains("test result: FAILED"));
        assert!(!s.contains("0x102e80de0"));
        assert!(s.len() <= VERIFY_SUMMARY);
        // Re-summarising to a smaller budget still keeps the panic.
        assert!(summarize_verify(&s, 1_500).contains("panicked at tests/a.rs:9:5"));
        assert_eq!(summarize_verify("short", 100), "short");
    }

    #[test]
    fn summarize_verify_keeps_panics_when_the_noise_is_not_frames() {
        let mut text = String::from("---- t stdout ----\nthread 't' panicked at x.rs:1:1:\nboom\n");
        text.push_str(&"noise line\n".repeat(50_000));
        text.push_str("test result: FAILED\n");
        let s = summarize_verify(&text, 8_000);
        assert!(s.contains("panicked at x.rs:1:1"));
        assert!(s.contains("boom"));
        assert!(s.contains("test result: FAILED"));
        assert!(s.len() <= 8_000);
    }

    #[test]
    fn tail_survives_multibyte_cuts() {
        let text = "あ".repeat(50);
        let t = tail(&text, 10);
        assert!(t.contains("earlier bytes omitted"));
        assert!(t.ends_with('あ'));
    }

    fn finding(id: &str, severity: crate::verdict::Severity) -> crate::verdict::Finding {
        crate::verdict::Finding {
            id: id.to_owned(),
            severity,
            file: None,
            line: None,
            title: "x".to_owned(),
            detail: String::new(),
        }
    }

    fn round(clean: bool, findings: Vec<crate::verdict::Finding>) -> ReviewRound {
        ReviewRound {
            round: 1,
            head: "h".to_owned(),
            verified_head: None,
            verified_at: None,
            reviews: vec![ReviewRecord {
                attempts: 0,
                reviewer: 1,
                agent: "a".to_owned(),
                summary: String::new(),
                findings,
                vote: None,
                failed: None,
                duration_ms: 0,
            }],
            e2e: Vec::new(),
            verify_retried: false,
            e2e_deferred: false,
            e2e_defer_reason: None,
            fix: None,
            blocking: 0,
            answered: 1,
            expected: 1,
            clean,
            progressed: false,
            vote_split: false,
            reconsideration: Vec::new(),
            verdict: None,
        }
    }

    fn voted(
        findings: Vec<crate::verdict::Finding>,
        votes: &[ReviewVote],
        revotes: &[Option<ReviewVote>],
    ) -> ReviewRound {
        let mut r = round(false, findings);
        r.reviews = votes
            .iter()
            .enumerate()
            .map(|(i, v)| {
                let mut rec = r.reviews[0].clone();
                rec.reviewer = i + 1;
                rec.vote = Some(*v);
                if i > 0 {
                    rec.findings = Vec::new();
                }
                rec
            })
            .collect();
        r.reconsideration = revotes
            .iter()
            .enumerate()
            .map(|(i, v)| ReviewRevoteRecord {
                reviewer: i + 1,
                agent: "a".to_owned(),
                vote: *v,
                reason: String::new(),
                failed: None,
            })
            .collect();
        r
    }

    #[test]
    fn a_hand_off_is_contested_only_by_a_blocking_finding_and_a_reject_vote() {
        use crate::verdict::Severity::{Major, Minor};
        use ReviewVote::{Approve, Reject};
        let major = || vec![finding("R1-1-1", Major)];
        let minor = || vec![finding("R1-1-1", Minor)];

        let c = voted(major(), &[Reject, Approve], &[])
            .contested_handoff()
            .expect("major + reject");
        assert_eq!(c.findings.len(), 1);
        assert_eq!(c.rejecters, vec![(1, "a".to_owned())]);

        assert!(
            voted(minor(), &[Reject, Approve], &[])
                .contested_handoff()
                .is_none()
        );
        assert!(
            voted(major(), &[Approve, Approve], &[])
                .contested_handoff()
                .is_none()
        );
        assert!(
            voted(Vec::new(), &[Reject], &[])
                .contested_handoff()
                .is_none()
        );
    }

    #[test]
    fn a_reject_withdrawn_in_the_revote_does_not_contest_and_an_unanswered_one_stands() {
        use crate::verdict::Severity::Major;
        use ReviewVote::{Approve, Reject};
        let major = || vec![finding("R1-1-1", Major)];

        let withdrawn = voted(major(), &[Reject, Approve], &[Some(Approve), Some(Approve)]);
        assert!(withdrawn.contested_handoff().is_none());

        let unanswered = voted(major(), &[Reject, Approve], &[None, Some(Approve)]);
        assert!(
            unanswered.contested_handoff().is_some(),
            "no revote falls back to the initial reject"
        );

        let raised = voted(major(), &[Approve, Approve], &[Some(Reject), Some(Approve)]);
        assert!(raised.contested_handoff().is_some());
    }

    #[test]
    fn e2e_status_tells_deferred_apart_from_not_configured() {
        let mut r = round(false, Vec::new());
        assert_eq!(r.e2e_status(), E2eStatus::NotConfigured);

        r.e2e_deferred = true;
        assert_eq!(
            r.e2e_status(),
            E2eStatus::Deferred,
            "an empty e2e must not read as unconfigured once it was deferred on purpose"
        );

        r.e2e = vec![CommandOutcome {
            command: "test".to_owned(),
            code: Some(0),
            output_tail: String::new(),
            duration_ms: 0,
            resource_blocked: false,
        }];
        assert_eq!(
            r.e2e_status(),
            E2eStatus::Passed,
            "a round with real outcomes is never read as deferred, even if the flag is still set"
        );
    }

    #[test]
    fn e2e_status_reports_a_real_failure_as_failed_not_deferred() {
        let mut r = round(false, Vec::new());
        r.e2e = vec![CommandOutcome {
            command: "test".to_owned(),
            code: Some(1),
            output_tail: "boom".to_owned(),
            duration_ms: 0,
            resource_blocked: false,
        }];
        assert_eq!(r.e2e_status(), E2eStatus::Failed);
    }

    #[test]
    fn e2e_status_never_reads_a_resource_block_as_a_failure() {
        // The exact shape of contention on the shared build cache: `e2e`
        // holds one outcome, and it is `resource_blocked`, never a command
        // that actually ran and produced a red exit code.
        let mut r = round(false, Vec::new());
        r.e2e = vec![CommandOutcome {
            command: "(waiting for the shared build cache)".to_owned(),
            code: None,
            output_tail: "contended".to_owned(),
            duration_ms: 0,
            resource_blocked: true,
        }];
        assert_eq!(
            r.e2e_status(),
            E2eStatus::ResourceBlocked,
            "magi's own inability to get a command to run must not read as a verdict on the \
             patch"
        );
    }

    #[test]
    fn verification_summary_is_silent_when_there_is_nothing_worth_saying() {
        let mut r = round(true, Vec::new());
        assert!(
            r.verification_summary("h").is_none(),
            "no verify.e2e configured: nothing to surface"
        );
        r.e2e = vec![CommandOutcome {
            command: "test".to_owned(),
            code: Some(0),
            output_tail: String::new(),
            duration_ms: 0,
            resource_blocked: false,
        }];
        assert!(
            r.verification_summary("h").is_none(),
            "a green result needs no skepticism attached to it"
        );
    }

    #[test]
    fn verification_summary_tells_the_current_head_apart_from_an_earlier_one() {
        let mut r = round(false, Vec::new());
        r.head = "h1".to_owned();
        r.e2e = vec![CommandOutcome {
            command: "test".to_owned(),
            code: Some(1),
            output_tail: "boom".to_owned(),
            duration_ms: 0,
            resource_blocked: false,
        }];
        r.verified_head = Some("h1".to_owned());
        r.verified_at = Some(Timestamp::now());

        let fresh = r.verification_summary("h1").expect("a failure is surfaced");
        assert!(
            fresh.label.contains("this is the head being looked at now"),
            "{}",
            fresh.label
        );
        assert_eq!(fresh.tail.as_deref(), Some("$ test\nboom\n"));

        let stale = r.verification_summary("h2").expect("still surfaced");
        assert!(
            stale.label.contains("an earlier head, since superseded"),
            "a result about a different commit than the one being looked at now must say so, \
             not read as current: {}",
            stale.label
        );
    }

    #[test]
    fn verification_summary_marks_a_resource_block_and_a_deferral_distinctly_from_a_failure() {
        let mut r = round(false, Vec::new());
        r.e2e = vec![CommandOutcome {
            command: "(waiting for the shared build cache)".to_owned(),
            code: None,
            output_tail: "contended".to_owned(),
            duration_ms: 0,
            resource_blocked: true,
        }];
        let blocked = r
            .verification_summary("h")
            .expect("a resource block is still surfaced, never silent");
        assert!(blocked.label.contains("could not run"));
        // No command actually ran, but which operation was attempted is
        // still a fact worth showing — never silent past the label either.
        let tail = blocked
            .tail
            .expect("the attempted operation is still named");
        assert!(tail.contains("(waiting for the shared build cache)"));
        assert!(tail.contains("contended"));

        let mut d = round(false, Vec::new());
        d.e2e_deferred = true;
        d.e2e_defer_reason = Some("2 blocking finding(s) already required a fix".to_owned());
        let deferred = d.verification_summary("h").expect("deferred is surfaced");
        assert!(deferred.label.contains("deferred to the fixer"));
        assert!(deferred.label.contains("2 blocking finding(s)"));
        assert!(deferred.tail.is_none());
    }

    #[test]
    fn verification_summary_says_unknown_rather_than_guessing_a_time_or_a_commit() {
        let mut r = round(false, Vec::new());
        r.e2e = vec![CommandOutcome {
            command: "test".to_owned(),
            code: Some(1),
            output_tail: "boom".to_owned(),
            duration_ms: 0,
            resource_blocked: false,
        }];
        // verified_head/verified_at left at their default `None` — exactly
        // the shape a schema-7 round with no reconstructable timestamp has.
        let summary = r.verification_summary("h").expect("a failure is surfaced");
        assert!(summary.label.contains("commit unknown"));
        assert!(summary.label.contains("checked at: unknown"));
    }

    #[test]
    fn gate_status_tells_not_run_apart_from_not_configured() {
        let mut s = state();
        assert_eq!(s.gate_status(), GateStatus::NotRun);

        s.gate_ran = true;
        assert_eq!(
            s.gate_status(),
            GateStatus::NotConfigured,
            "an empty gate is not configured even when gate_ran records an attempt"
        );

        s.gate = vec![CommandOutcome {
            command: "cargo make check".to_owned(),
            code: Some(0),
            output_tail: String::new(),
            duration_ms: 0,
            resource_blocked: false,
        }];
        assert_eq!(s.gate_status(), GateStatus::Passed);

        s.gate[0].code = Some(1);
        assert_eq!(s.gate_status(), GateStatus::Failed);

        s.gate_ran = false;
        assert_eq!(
            s.gate_status(),
            GateStatus::NotRun,
            "gate_ran false must win even over a non-empty gate left from a stale record"
        );
    }

    #[test]
    fn open_findings_is_empty_when_the_last_round_was_clean() {
        let mut s = state();
        s.reviews = vec![round(
            true,
            vec![finding("R1-1-1", crate::verdict::Severity::Minor)],
        )];
        assert!(s.open_findings().is_empty());
    }

    #[test]
    fn open_findings_reads_the_last_non_clean_round() {
        let mut s = state();
        s.reviews = vec![round(
            false,
            vec![finding("R1-1-1", crate::verdict::Severity::Major)],
        )];
        let open = s.open_findings();
        assert_eq!(open.len(), 1);
        assert_eq!(open[0].id, "R1-1-1");
    }

    #[test]
    fn handed_off_with_open_findings_needs_a_mergeable_status_and_an_open_round() {
        let mut s = state();
        s.reviews = vec![round(
            false,
            vec![finding("R1-1-1", crate::verdict::Severity::Major)],
        )];

        s.status = RunStatus::Blocked;
        assert!(
            !s.handed_off_with_open_findings(),
            "a blocked run is not a hand-off"
        );

        s.status = RunStatus::Ready;
        assert!(s.handed_off_with_open_findings());

        s.reviews = vec![round(true, Vec::new())];
        assert!(
            !s.handed_off_with_open_findings(),
            "a clean last round has nothing to hand off"
        );
    }

    #[test]
    fn unmerged_by_design_is_only_ready_reached_via_merge_mode_none() {
        let mut s = state();

        s.status = RunStatus::Ready;
        assert!(
            !s.unmerged_by_design(),
            "no merge outcome recorded at all must not be flagged"
        );

        s.merge = Some(MergeOutcome {
            mode: MergeMode::None,
            ok: true,
            detail: "git merge --no-ff magi/x/A".to_owned(),
            empty: false,
        });
        assert!(
            s.unmerged_by_design(),
            "Ready reached through mode none is the case this exists to flag"
        );

        // A PR closed without merging also leaves `status` at `Ready`, but
        // through `mode = "pr"` — a run that may still have been landable by
        // a person watching the PR, unlike the honest mode-none no-op.
        s.merge = Some(MergeOutcome {
            mode: MergeMode::Pr,
            ok: false,
            detail: "https://example.com/pr/1 was closed without merging".to_owned(),
            empty: false,
        });
        assert!(
            !s.unmerged_by_design(),
            "a closed pull request is a different Ready and must not be relabelled"
        );

        // Same signal must not fire before the run actually got there.
        s.status = RunStatus::Gating;
        s.merge = Some(MergeOutcome {
            mode: MergeMode::None,
            ok: true,
            detail: "git merge --no-ff magi/x/A".to_owned(),
            empty: false,
        });
        assert!(
            !s.unmerged_by_design(),
            "status must actually be Ready, not merely have a stale mode-none merge record"
        );
    }

    #[test]
    fn state_round_trips_through_json() {
        let s = state();
        let body = serde_json::to_string(&s).unwrap();
        let back: RunState = serde_json::from_str(&body).unwrap();
        assert_eq!(back.id, s.id);
        assert_eq!(back.instruction, "add retries");
        assert_eq!(back.status, RunStatus::Prep);
    }

    #[test]
    fn a_round_recorded_before_e2e_deferral_existed_still_loads() {
        // Exactly the shape a pre-existing `run.json` has for a round: no
        // `e2e_deferred`, no `e2e_defer_reason`. Every round used to run e2e
        // unconditionally, so the honest reading of an old record's silence
        // on this is "it was not deferred" — `false`/`None`, not a load
        // failure and not a schema bump (see the `SCHEMA` doc comment: a
        // purely additive field whose absence has one unambiguous meaning
        // does not need one).
        let body = r#"{
            "round": 1,
            "head": "deadbeef",
            "reviews": [],
            "e2e": [],
            "verify_retried": false,
            "fix": null,
            "blocking": 0,
            "answered": 1,
            "expected": 1,
            "clean": true
        }"#;
        let r: ReviewRound = serde_json::from_str(body).expect("an old-shaped round must load");
        assert!(!r.e2e_deferred);
        assert!(r.e2e_defer_reason.is_none());
        assert_eq!(r.e2e_status(), E2eStatus::NotConfigured);
    }

    #[test]
    fn schema_five_state_migrates_old_empty_e2e_and_legacy_verify_budget() {
        let mut value = serde_json::to_value(state()).expect("serialize state");
        let object = value.as_object_mut().expect("state object");
        object.insert("schema".to_owned(), serde_json::json!(5));
        let graph = object["config"]["graph"]
            .as_object_mut()
            .expect("graph object");
        graph.insert("timeout_review".to_owned(), serde_json::json!(3600));
        graph.remove("timeout_verify");
        let review = object["reviews"].as_array_mut().expect("reviews");
        review.push(serde_json::json!({
            "round": 1, "head": "old", "reviews": [], "e2e": [],
            "verify_retried": false, "blocking": 0, "answered": 1,
            "expected": 1, "clean": true
        }));
        let old: RunState = serde_json::from_value(value).expect("schema-5 shape parses");
        let migrated = migrate_schema(old).expect("schema 5 migrates");
        assert_eq!(migrated.schema, SCHEMA);
        assert_eq!(migrated.config.graph.verify_timeout(), 3600);
        assert_eq!(migrated.reviews[0].e2e_status(), E2eStatus::NotConfigured);
    }

    #[test]
    fn schema_six_state_with_a_recorded_gate_migrates_to_gate_ran_true() {
        let mut value = serde_json::to_value(state()).expect("serialize state");
        let object = value.as_object_mut().expect("state object");
        object.insert("schema".to_owned(), serde_json::json!(6));
        object.insert(
            "gate".to_owned(),
            serde_json::json!([{
                "command": "cargo make check",
                "code": 0,
                "output_tail": "",
                "duration_ms": 0,
                "resource_blocked": false
            }]),
        );
        let old: RunState = serde_json::from_value(value).expect("schema-6 shape parses");
        let migrated = migrate_schema(old).expect("schema 6 migrates");
        assert_eq!(migrated.schema, SCHEMA);
        assert!(
            migrated.gate_ran,
            "a non-empty recorded gate is a real attempt, not an unrun one"
        );
        assert_eq!(migrated.gate_status(), GateStatus::Passed);
    }

    #[test]
    fn schema_six_state_with_an_empty_gate_migrates_to_gate_ran_false_and_is_retried() {
        // The exact shape of the stuck `shoka` run this schema bump fixes:
        // `verify.gate` empty, `gate` empty, schema 6. It must come back as
        // "not yet run" so the next `gate()` call re-attempts it — and for a
        // repo with no gate commands configured, that resolves instantly to
        // `NotConfigured` instead of staying stuck forever.
        let mut value = serde_json::to_value(state()).expect("serialize state");
        let object = value.as_object_mut().expect("state object");
        object.insert("schema".to_owned(), serde_json::json!(6));
        object.insert("gate".to_owned(), serde_json::json!([]));
        let old: RunState = serde_json::from_value(value).expect("schema-6 shape parses");
        let migrated = migrate_schema(old).expect("schema 6 migrates");
        assert_eq!(migrated.schema, SCHEMA);
        assert!(
            !migrated.gate_ran,
            "an empty gate on schema 6 is ambiguous and must be treated as unrun"
        );
        assert_eq!(migrated.gate_status(), GateStatus::NotRun);
    }

    #[test]
    fn schema_seven_state_reconstructs_verified_head_for_a_round_that_actually_ran_e2e() {
        // Schema 7's main review loop always checked `e2e` against the
        // round's own `head` — it just never wrote that into `verified_head`
        // unless a catch-up run had checked a *different* commit. Migrating
        // to schema 8 restores that always-true fact instead of leaving a
        // reader to assume it.
        let mut value = serde_json::to_value(state()).expect("serialize state");
        let object = value.as_object_mut().expect("state object");
        object.insert("schema".to_owned(), serde_json::json!(7));
        let reviews = object["reviews"].as_array_mut().expect("reviews");
        reviews.push(serde_json::json!({
            "round": 1, "head": "deadbeef", "reviews": [],
            "e2e": [{
                "command": "cargo test", "code": 0, "output_tail": "",
                "duration_ms": 0, "resource_blocked": false
            }],
            "verify_retried": false, "blocking": 0, "answered": 1,
            "expected": 1, "clean": true
        }));
        let old: RunState = serde_json::from_value(value).expect("schema-7 shape parses");
        let migrated = migrate_schema(old).expect("schema 7 migrates");
        assert_eq!(migrated.schema, SCHEMA);
        assert_eq!(
            migrated.reviews[0].verified_head.as_deref(),
            Some("deadbeef"),
            "a schema-7 round's main-loop e2e was always against its own head, even though the \
             field never said so"
        );
        assert!(
            migrated.reviews[0].verified_at.is_none(),
            "no historical timestamp exists to reconstruct; unknown stays unknown, not a \
             guessed 'now'"
        );
    }

    #[test]
    fn schema_seven_state_leaves_a_deferred_round_with_no_verified_head() {
        let mut value = serde_json::to_value(state()).expect("serialize state");
        let object = value.as_object_mut().expect("state object");
        object.insert("schema".to_owned(), serde_json::json!(7));
        let reviews = object["reviews"].as_array_mut().expect("reviews");
        reviews.push(serde_json::json!({
            "round": 1, "head": "deadbeef", "reviews": [],
            "e2e": [], "e2e_deferred": true,
            "verify_retried": false, "blocking": 1, "answered": 1,
            "expected": 1, "clean": false
        }));
        let old: RunState = serde_json::from_value(value).expect("schema-7 shape parses");
        let migrated = migrate_schema(old).expect("schema 7 migrates");
        assert!(
            migrated.reviews[0].verified_head.is_none(),
            "a deferred round never ran e2e; there is nothing to reconstruct"
        );
    }

    #[test]
    fn origin_round_trips_through_the_run_state() {
        let mut s = state();
        s.origin = Some(Origin::from_agent_env(
            Some(("4a7b".to_owned(), "chat".to_owned())),
            Some("20260930-092817-4f6f".to_owned()),
        ));
        let back: RunState =
            serde_json::from_str(&serde_json::to_string(&s).unwrap()).expect("parse back");
        assert_eq!(back.origin, s.origin);
        assert_eq!(
            back.origin.as_ref().unwrap().by,
            StartedBy::Chat {
                talk: "4a7b".to_owned()
            }
        );
        assert_eq!(
            origin_label(back.origin.as_ref()),
            "chat 4a7b, for task 4f6f"
        );
    }

    #[test]
    fn a_schema_twelve_record_without_an_origin_loads_and_reads_as_unknown() {
        let mut value = serde_json::to_value(state()).expect("serialize state");
        let object = value.as_object_mut().expect("state object");
        object.insert("schema".to_owned(), serde_json::json!(12));
        object.remove("origin");
        let old: RunState = serde_json::from_value(value).expect("schema-12 shape parses");
        assert!(old.origin.is_none());
        let migrated = migrate_schema(old).expect("schema 12 migrates");
        assert_eq!(migrated.schema, SCHEMA);
        assert!(
            migrated.origin.is_none(),
            "an origin nobody recorded is not invented"
        );
        assert_eq!(
            origin_label(migrated.origin.as_ref()),
            "origin unknown (started before origins were recorded)"
        );
    }

    #[test]
    fn schema_supported_versions_all_reach_current() {
        for schema in 5..=SCHEMA {
            let mut value = serde_json::to_value(state()).expect("serialize state");
            value["schema"] = serde_json::json!(schema);
            if schema < 14 {
                value.as_object_mut().unwrap().remove("task_instruction");
            }
            if schema < 13 {
                value.as_object_mut().unwrap().remove("origin");
            }
            let old: RunState = serde_json::from_value(value).expect("legacy shape parses");
            let migrated = migrate_schema(old)
                .unwrap_or_else(|error| panic!("schema {schema} must migrate: {error}"));
            assert_eq!(migrated.schema, SCHEMA, "schema {schema}");
            assert!(migrated.task_instruction.is_none(), "schema {schema}");
            assert!(migrated.origin.is_none(), "schema {schema}");
        }
    }

    #[test]
    fn schema_unsupported_versions_remain_rejected() {
        for schema in [1, 2, 3, 4, SCHEMA + 1] {
            let mut old = state();
            old.schema = schema;
            assert!(migrate_schema(old).is_err(), "schema {schema}");
        }
    }

    #[test]
    fn origin_attributes_agents_by_node_and_never_by_guesswork() {
        let chat = Origin::from_agent_env(Some(("4a7b".to_owned(), "chat".to_owned())), None);
        assert_eq!(
            chat.by,
            StartedBy::Chat {
                talk: "4a7b".into()
            }
        );
        let seat = Origin::from_agent_env(
            Some(("20260930-092817-ec34".to_owned(), "implement".to_owned())),
            None,
        );
        assert_eq!(seat.label(), "implement@ec34");
        assert_eq!(Origin::from_agent_env(None, None), Origin::operator());
        assert_eq!(Origin::operator().label(), "operator");
        assert_eq!(Origin::queue("20260930-000000-4f6f").label(), "task 4f6f");
        assert_eq!(
            Origin::operator()
                .serving(Some("20260930-000000-4f6f".to_owned()))
                .label(),
            "operator, for task 4f6f"
        );
    }

    #[test]
    fn schema_six_serialization_is_rejected_by_a_schema_five_reader() {
        let body = serde_json::to_value(state()).expect("serialize state");
        assert_eq!(body["schema"], serde_json::json!(SCHEMA));
        assert_ne!(body["schema"], serde_json::json!(5));
    }

    #[test]
    fn seat_started_and_finished_track_who_has_not_answered_yet() {
        let mut s = state();
        s.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
        s.seat_started("judge", "judge-2", std::time::Duration::from_secs(60), 0);
        assert_eq!(s.active.len(), 2, "both seats are still out");

        s.seat_finished("judge-1");
        assert_eq!(
            s.active.keys().collect::<Vec<_>>(),
            vec!["judge-2"],
            "only the seat that answered drops out; judge-2 is still waited on"
        );
    }

    /// `seats_active` / `tasks_active` are the accessors report/web read
    /// instead of `active` directly, so neither ever counts the other kind of
    /// entry as a seat — a `verify.e2e` task must never inflate a quorum or
    /// seat count, and a seat must never show up in a task listing.
    #[test]
    fn seats_active_and_tasks_active_never_cross_over() {
        let mut s = state();
        s.seat_started("judge", "judge-1", std::time::Duration::from_secs(60), 0);
        s.task_command(
            "e2e",
            "verify",
            0,
            "cargo test",
            1,
            2,
            std::time::Duration::from_secs(600),
        );

        assert_eq!(
            s.seats_active()
                .map(|(k, _)| k.as_str())
                .collect::<Vec<_>>(),
            vec!["judge-1"]
        );
        assert_eq!(
            s.tasks_active()
                .map(|(k, _)| k.as_str())
                .collect::<Vec<_>>(),
            vec!["e2e"]
        );

        // A command boundary updates the same entry in place — still one
        // task, never a second one accumulating alongside it.
        s.task_command(
            "e2e",
            "verify",
            0,
            "cargo clippy",
            2,
            2,
            std::time::Duration::from_secs(600),
        );
        assert_eq!(s.tasks_active().count(), 1);
        assert_eq!(s.active["e2e"].command.as_deref(), Some("cargo clippy"));

        s.task_finished("e2e");
        assert!(s.tasks_active().next().is_none());
        assert_eq!(
            s.seats_active()
                .map(|(k, _)| k.as_str())
                .collect::<Vec<_>>(),
            vec!["judge-1"],
            "clearing the task must not touch the seat entry"
        );
    }

    #[test]
    fn a_retry_is_recorded_as_a_later_attempt_on_the_same_seat() {
        let mut s = state();
        s.seat_started("review", "review-2", std::time::Duration::from_secs(30), 0);
        s.seat_finished("review-2");
        // A nudge re-asks the same seat; attempt says this is not the first
        // time, which is the only trace a nudge otherwise leaves behind.
        s.seat_started("review", "review-2", std::time::Duration::from_secs(30), 1);
        assert_eq!(s.active["review-2"].attempt, 1);
    }

    #[test]
    fn active_seat_reports_elapsed_and_remaining_time() {
        let now = Timestamp::now();
        let started = now - jiff::SignedDuration::from_secs(30);
        let seat = ActiveSeat {
            node: "judge".to_owned(),
            started_at: started,
            timeout_secs: 100,
            attempt: 0,
            task: None,
            command: None,
            index: None,
            total: None,
        };
        assert_eq!(seat.elapsed_secs(now), 30);
        assert_eq!(seat.remaining_secs(now), 70);
    }

    #[test]
    fn remaining_time_never_goes_negative_past_the_timeout() {
        // `agy`'s own print-timeout occasionally overruns by a hair before the
        // kill lands; a naive subtraction would print a negative "time left".
        let now = Timestamp::now();
        let started = now - jiff::SignedDuration::from_secs(200);
        let seat = ActiveSeat {
            node: "implement".to_owned(),
            started_at: started,
            timeout_secs: 100,
            attempt: 1,
            task: None,
            command: None,
            index: None,
            total: None,
        };
        assert_eq!(seat.remaining_secs(now), 0);
    }

    #[test]
    fn clear_active_drops_stale_seats_and_reports_whether_it_did() {
        let mut s = state();
        assert!(!s.clear_active(), "nothing to clear on a fresh run");
        s.seat_started(
            "implement",
            "impl-B",
            std::time::Duration::from_secs(3600),
            0,
        );
        assert!(s.clear_active(), "a leftover entry is reported as cleared");
        assert!(s.active.is_empty());
    }

    #[test]
    fn active_seat_carries_nothing_that_could_be_read_as_output_bytes() {
        // `agy` prints exactly one JSON object, at the very end (see
        // `agent::dropped_stream`'s doc comment) — a seat can sit at zero
        // captured bytes for its whole timeout while working normally. So
        // `ActiveSeat` records only the wall-clock facts (when it started,
        // its budget, which attempt), never a byte count, which is what
        // keeps a reader from being able to build "0 bytes => dead" out of
        // it even by accident.
        let seat = ActiveSeat {
            node: "implement".to_owned(),
            started_at: Timestamp::now(),
            timeout_secs: 60,
            attempt: 0,
            task: None,
            command: None,
            index: None,
            total: None,
        };
        let value = serde_json::to_value(&seat).unwrap();
        let keys: std::collections::BTreeSet<String> =
            value.as_object().unwrap().keys().cloned().collect();
        assert_eq!(
            keys,
            std::collections::BTreeSet::from([
                "node".to_owned(),
                "started_at".to_owned(),
                "timeout_secs".to_owned(),
                "attempt".to_owned(),
            ]),
            "a byte count here would be a lever to declare a silent-but-healthy seat dead, and \
             the task-only fields must stay absent (not null) on an ordinary seat entry"
        );
    }

    /// The same guarantee as
    /// [`active_seat_carries_nothing_that_could_be_read_as_output_bytes`],
    /// extended to a task entry: `verify.e2e` / `verify.gate` are exactly as
    /// silent as `agy` between commands, so a running command-list task must
    /// never carry anything a reader could mistake for output-byte evidence
    /// either.
    #[test]
    fn task_active_seat_carries_nothing_that_could_be_read_as_output_bytes() {
        let seat = ActiveSeat {
            node: "verify".to_owned(),
            started_at: Timestamp::now(),
            timeout_secs: 600,
            attempt: 0,
            task: Some("e2e".to_owned()),
            command: Some("cargo test".to_owned()),
            index: Some(1),
            total: Some(3),
        };
        let value = serde_json::to_value(&seat).unwrap();
        let keys: std::collections::BTreeSet<String> =
            value.as_object().unwrap().keys().cloned().collect();
        assert_eq!(
            keys,
            std::collections::BTreeSet::from([
                "node".to_owned(),
                "started_at".to_owned(),
                "timeout_secs".to_owned(),
                "attempt".to_owned(),
                "task".to_owned(),
                "command".to_owned(),
                "index".to_owned(),
                "total".to_owned(),
            ]),
        );
    }

    #[test]
    fn an_old_run_json_without_active_seats_still_loads() {
        // Schema did not bump for this field: an already-written run.json
        // simply lacks the key, and `#[serde(default)]` must fill it in
        // rather than fail the whole read.
        let s = state();
        let mut value = serde_json::to_value(&s).unwrap();
        value.as_object_mut().unwrap().remove("active");
        let back: RunState = serde_json::from_value(value).unwrap();
        assert!(back.active.is_empty());
        assert_eq!(back.schema, SCHEMA);
    }

    #[test]
    fn an_old_run_json_without_jobs_still_loads() {
        // No schema bump for this field either, for the same reason: an
        // empty `jobs` list on an old record means exactly what it always
        // meant for that record — no adapter existed yet to report one —
        // and `#[serde(default)]` fills it in rather than failing the read.
        let s = state();
        let mut value = serde_json::to_value(&s).unwrap();
        value.as_object_mut().unwrap().remove("jobs");
        let back: RunState = serde_json::from_value(value).unwrap();
        assert!(back.jobs.is_empty());
        assert_eq!(back.schema, SCHEMA);
    }

    #[test]
    fn ensure_can_delete_guards_live_and_unfolded_runs() {
        let mut s = state();
        // 1. A daemon is working on it right now.
        s.status = RunStatus::Prep;
        let err = s.ensure_can_delete(true).unwrap_err().to_string();
        assert!(err.contains("live daemon"), "{err}");

        // 2. The same unfinished run with no daemon behind it is a leftover
        // from a killed process, and deletable. Without this an interrupted
        // run could never be removed: its status stays `prep` forever.
        assert!(s.ensure_can_delete(false).is_ok());

        // 3. Unfolded candidates are refused either way — that is the guard
        // that stops a delete from discarding a worktree.
        s.status = RunStatus::Merged;
        s.candidates.push(Candidate {
            index: 0,
            label: 'A',
            agent: "a".to_owned(),
            branch: "b".to_owned(),
            worktree: PathBuf::from("/w"),
            summary: String::new(),
            stat: String::new(),
            files: 1,
            commits: 1,
            empty: false,
            failed: None,
            verified_noop: None,
            duration_ms: 0,
            folded: false,
        });
        let err = s.ensure_can_delete(false).unwrap_err().to_string();
        assert!(
            err.contains("magi fold"),
            "error must suggest `magi fold`: {err}"
        );

        // 4. Folded and nobody working on it.
        s.candidates[0].folded = true;
        assert!(s.ensure_can_delete(false).is_ok());
    }
}

impl RunState {
    /// A seed for a seat that changes agent, distinct on every call.
    pub fn next_seat_seed(&mut self) -> u64 {
        self.seat_epoch += 1;
        self.seed ^ self.seat_epoch.wrapping_mul(0x9E37_79B9_7F4A_7C15)
    }
}

/// What kind of failure ended an agent's turn on a seat, for deciding whether
/// the seat is worth handing to the next roster agent.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum FailClass {
    /// A CLI rate limit.
    Quota,
    /// The turn outlived its budget.
    Timeout,
    /// Any other failure, with the message's shape (`failure_signature`,
    /// capped at 120 characters).
    Other(String),
}

/// One reviewer seat's record across review rounds, so a round does not ask
/// an agent that already failed the seat while another one is answering.
///
/// `failed` is a *priority* set, not a bound: the per-round `tried` set in
/// `ask_json_wave` starts empty every round, so an agent that answered is
/// never locked out. A failed id is only asked again once nothing else on the
/// roster is left (once per round, and the rounds are `review_rounds`).
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SeatHistory {
    /// Roster ids that failed this seat and have not answered it since.
    #[serde(default)]
    pub failed: BTreeSet<String>,
    /// The agent that last answered this seat.
    #[serde(default)]
    pub last_ok: Option<String>,
    /// The class of the seat's most recent failure; cleared by an answer.
    #[serde(default)]
    pub last_fail: Option<FailClass>,
}