magi/

agent.rs

//! Driving agent CLIs.
//!
//! Every agent in magi is a subscription CLI (`claude`, `opencode`, `agy`) or
//! an arbitrary command, invoked headless in a working directory. There is no
//! API-key path on purpose: the CLIs carry the operator's own plan, and they
//! are the only interface that exposes an agent's whole tool loop rather than a
//! single completion.
//!
//! # Seats, not agents
//!
//! Conversations are keyed by *seat* ([`SeatState::key`]), never by agent id. A
//! model that implements candidate B and also sits as judge 3 gets two
//! unrelated conversations, so the judge cannot recognise its own work from
//! having written it. Sessions are what make deliberation affordable — a judge
//! remembers its own argument instead of being re-fed the entire candidate set
//! — and seat scoping is what keeps that from destroying blindness.
//!
//! # Session mechanics per CLI
//!
//! | CLI | open | resume |
//! |-----|------|--------|
//! | `claude` | `--session-id <uuid>` (magi mints it) | `--resume <uuid>` |
//! | `opencode` | `--format json` reports `sessionID` | `-s <id>` |
//! | `agy` | `--output-format json` reports `conversation_id` | `--conversation <id>` |
//! | `codex` | `exec --json` reports `thread.started.thread_id` | `exec … resume <id>` |
//! | `omp` | `-p --mode=json` reports `id` on its `"type":"session"` line | `--resume <id>` |
//!
//! Claude is the only one magi can address before the first turn; the others
//! report an id back, so [`SeatState::captured_session`] stays `None` until a
//! turn has completed and [`has_session`] answers honestly instead of
//! optimistically.
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::time::{Duration, Instant};

use anyhow::{Context as _, Result, bail};
use serde::{Deserialize, Serialize};
use std::sync::{Arc, Mutex};

use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
use tokio::process::Command;

use crate::config::{AgentChoice, AgentKind, AgentSpec, Delivery};
use crate::proc::Quiet as _;
use crate::rng::SplitMix64;

/// Conversation state for one seat, persisted with the run so `magi run
/// --resume` continues the same CLI conversations.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SeatState {
    /// Stable seat name, e.g. `impl-A`, `judge-2`, `review-1`, `fix`.
    pub key: String,
    /// Agent id occupying the seat.
    pub agent: String,
    /// Turns already taken in this seat.
    pub turns: usize,
    /// Claude session uuid, minted up front so the first turn and every resume
    /// agree on it without parsing anything back.
    pub claude_session: Option<String>,
    /// Session id reported by a CLI that mints its own (`opencode`, `agy`).
    pub captured_session: Option<String>,
}

impl SeatState {
    /// New seat. `run_seed` scopes the minted Claude uuid to this run.
    pub fn new(key: &str, agent: &str, run_seed: u64) -> Self {
        let mut rng = SplitMix64::new(run_seed ^ crate::rng::fnv1a(key));
        Self {
            key: key.to_owned(),
            agent: agent.to_owned(),
            turns: 0,
            claude_session: Some(rng.uuid_v4()),
            captured_session: None,
        }
    }
}

/// Can a follow-up prompt rely on this seat remembering the conversation?
pub fn has_session(kind: AgentKind, seat: &SeatState, sessions_enabled: bool) -> bool {
    if !sessions_enabled || seat.turns == 0 {
        return false;
    }
    match kind {
        AgentKind::Claude => seat.claude_session.is_some(),
        AgentKind::Opencode | AgentKind::Antigravity | AgentKind::Codex | AgentKind::Omp => {
            seat.captured_session.is_some()
        }
        AgentKind::Command => true,
    }
}

/// Repository access is independent of control-plane write grants.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InvocationScope {
    /// Ordinary repository seat with its configured access.
    Repository,
    /// Question control and supplied evidence without repository scheduling.
    ConsultationControl,
    /// Read-only repository continuation with nonblocking coordination.
    ConsultationInvestigation,
}

/// One agent invocation.
#[derive(Debug, Clone)]
pub struct Invocation<'a> {
    /// Explicit access scope for this turn, independent of writable roots.
    pub scope: InvocationScope,
    /// Rendered repository coordination policy for this invocation.
    pub coordination: Option<&'a crate::config::Coordination>,
    /// Launcher fetch evidence, unchanged on session resume.
    pub fetched_base: Option<&'a crate::run::FetchedBase>,
    /// Working directory. Always a real checkout, so every CLI can read the
    /// repository without per-vendor "extra directory" flags.
    pub cwd: &'a Path,
    /// The full prompt.
    pub prompt: &'a str,
    /// Wall-clock limit; the process tree is killed when it elapses.
    pub timeout: Duration,
    /// May the agent modify files? Judges and reviewers may not.
    pub allow_write: bool,
    /// Drop the CLI's own sandbox and approval gate. Conversational talk
    /// turns only, and only together with `allow_write`: it is the codex
    /// counterpart of claude's `bypassPermissions`. Unattended seats
    /// (implementer, judge, reviewer, deputy, ...) never set it, and only
    /// `talk.rs` may.
    pub unsandboxed: bool,
    /// Continue this seat's conversation when the CLI supports it.
    pub sessions: bool,
    /// Directory for prompt / stdout / stderr artifacts.
    pub artifacts: &'a Path,
    /// Artifact filename stem.
    pub stem: &'a str,
    /// Run this invocation belongs to. Exported as `MAGI_RUN` so an agent that
    /// files a task with `magi task add` is attributed to the run that was
    /// paying for it, rather than looking like a human wandered by.
    pub run: &'a str,
    /// Graph node being executed, e.g. `implement` or `review`. Exported as
    /// `MAGI_NODE` for the same reason: "who asked for this" is the first
    /// question about an autonomously created task.
    pub node: &'a str,
    /// Shared build cache the seat should build into, from the rendered
    /// `CARGO_TARGET_DIR=` in the verify commands. Exported as
    /// `CARGO_TARGET_DIR` so the implementer's compile lands inside the same
    /// directory `verify` reads back from - one cache, one prune, and the
    /// build the agent just paid for is the build the gate reuses.
    pub cache_dir: Option<&'a Path>,
    /// Absolute paths of images the operator attached to this conversation,
    /// outside `cwd` - see `chat`/`talk`'s `attachments_dir`. Empty for every
    /// invocation that is not a chat or talk turn. [`build_command`] uses
    /// this only to decide whether a CLI's sandbox needs widening to read
    /// them; the prompt text naming each path and its mime is built by the
    /// caller, not here.
    pub attachments: &'a [PathBuf],
    /// Directories outside `cwd` the seat may write, when it may write at all.
    /// Explicit, purpose-specific grants (for example a deputy's stores).
    /// Repository editing seats also receive the shared Git directory and
    /// question control-plane directories in [`invoke`].
    pub writable: &'a [PathBuf],
}

fn repository_writer(inv: &Invocation<'_>) -> bool {
    inv.scope == InvocationScope::Repository
        && inv.allow_write
        && (matches!(
            inv.node,
            "implement" | "fix" | "graft" | "gate-fix" | "land" | "rebase"
        ) || inv.node == crate::queue::CHAT_NODE)
}

fn repository_seat(inv: &Invocation<'_>) -> bool {
    // These roles consume supplied state, prompts and conversation history.
    // In particular, they must be able to deliver an answer or make queue
    // decisions while an editor holds its reservation waiting for that answer.
    // Ordinary chat turns bypass repository scheduling, including waiter delivery.
    // Investigation is an explicit continuation with nonblocking acquisition.
    if inv.scope == InvocationScope::ConsultationInvestigation {
        return true;
    }
    !matches!(
        inv.node,
        crate::queue::CHAT_NODE | "deputy" | "conduct" | "dupes"
    ) && inv.scope != InvocationScope::ConsultationControl
}

/// Resolve the additional write scope for a repository editing seat.
///
/// Resolve Git's common directory from the actual seat checkout, never by
/// assuming `.git` is a directory. Canonicalize directories before handing
/// them to a sandbox (notably `/tmp` aliases on macOS). Failure is a startup
/// error: do not start work with an unusable lock/question route.
pub async fn worktree_writable_roots(cwd: &Path, home: &Path) -> Result<Vec<PathBuf>> {
    let mut roots = Vec::new();
    if let Some(common) = crate::repo_lock::common_dir(cwd).await? {
        let cwd = std::fs::canonicalize(cwd).context("canonicalize seat working directory")?;
        if !common.starts_with(&cwd) {
            roots.push(common);
        }
    }
    // Questions::put/update/beat/panels write questions. Its quiet_for hook
    // updates matching notifications. Neither needs queue, runs, talks, nor
    // the entire state directory.
    for name in ["questions", "notifications"] {
        let dir = home.join(name);
        std::fs::create_dir_all(&dir).with_context(|| format!("prepare seat {name} directory"))?;
        roots.push(std::fs::canonicalize(dir)?);
    }
    Ok(roots)
}

/// Current-turn instructions override stale fetch/commit requirements in history.
fn editing_access_note(
    kind: AgentKind,
    unsandboxed: bool,
    base: Option<&crate::run::FetchedBase>,
) -> String {
    let mut note = String::from(
        "Launcher repository access contract (applies to this turn and overrides earlier fetch/commit instructions):\n",
    );
    if let Some(base) = base {
        note.push_str(&format!(
            "The launcher successfully fetched `{}/{}` at {} and pinned the run base to `{}`. Use this exact local commit as the verified fetched base; this is a snapshot, not a claim that the remote has not moved since. Do not fetch again as a prerequisite, move the base, or refresh it on resume. Base updates belong to the launcher's existing base-sync path.\n",
            base.remote, base.branch, base.fetched_at, base.commit,
        ));
    } else {
        note.push_str("No successful launcher fetch evidence is supplied for this invocation. Do not describe local tracking refs as freshly fetched. If fresh remote evidence is essential, report an infrastructure blocker requiring a launcher fetch and a new prepared turn.\n");
    }
    // Only the sandboxed backend has this verified access restriction.
    if kind != AgentKind::Codex || unsandboxed {
        return note;
    }
    note.push_str("Network access is restricted by the seat sandbox. A shared Git directory grant does not guarantee access to protected Git administration paths. Use local status, diff and log to inspect the checkout. If add/commit is denied, leave the implementation files in this assigned worktree: the launcher saves eligible edits after the seat exits. Do not repeatedly retry denied fetches or ask the owner to approve unavailable sandbox permissions. If an essential operation (including rebase continuation or dependency download) is still impossible, stop with an actionable infrastructure report naming the operation, target and error; the operator must repair/deploy the launcher environment before a single new attempt. Do not use bypass flags, change permissions, edit another checkout, or claim a network failure proves host DNS is broken. Specification questions remain valid.\n");
    note
}

/// Evidence that a CLI ran out of its rate limit / quota, distinct from an
/// ordinary failure.
///
/// `reset` is free text: CLIs render the reset time in their own locale, and
/// parsing it exactly would be a bug factory. When it is not readable we say
/// nothing rather than invent a format.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Quota {
    /// Human-readable reset time, when the CLI printed one.
    #[serde(default)]
    pub reset: Option<String>,
}

/// A CLI hung up on its own stream while the agent was working.
///
/// Separate from a failure because the work was done and billed, and separate
/// from a [`Quota`] because it is worth asking again: the answer is in the
/// conversation, not lost to a limit that has to reset first. See
/// [`dropped_stream`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Dropped {
    /// What the CLI said as it hung up, verbatim.
    pub why: String,
    /// Output tokens the CLI reported before it did - the evidence that this
    /// was a delivery failure and not an agent that produced nothing.
    pub output_tokens: u64,
}

/// One command a CLI's own structured event stream reported running, with
/// the result it reported for it.
///
/// This is evidence the CLI chose to report about its own tool loop — never
/// something magi polled, supervised, or inferred from a process list. Only
/// the Codex arm of [`extract`] currently populates it (its `item.completed`
/// / `command_execution` events name `id`, `command`, `exit_code` and
/// `aggregated_output` directly); every other backend's CLI does not expose
/// this in what magi currently captures, so its seats simply never produce
/// any. A command a CLI never reported finishing (still running when the
/// turn ended, or the event stream never named it) has no entry here either
/// — there is no event to build one from, and this type must never be used
/// to *guess* that a command is still in flight.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommandEvidence {
    /// The CLI's own id for this command.
    pub id: String,
    /// The command itself, as the CLI reported it.
    pub description: String,
    /// Exit code the CLI reported for it.
    pub exit_code: Option<i32>,
    /// Tail of the command's own output, when the CLI reported one.
    pub result_summary: String,
    /// Which CLI/event stream this came from, e.g. `"codex"`.
    pub source: String,
}

/// Result of an agent invocation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentOutput {
    /// The agent's final message, extracted from whatever the CLI printed.
    pub text: String,
    /// Exit status code.
    pub exit_code: Option<i32>,
    /// Did the invocation hit its timeout?
    pub timed_out: bool,
    /// Wall-clock duration.
    pub duration_ms: u64,
    /// Artifact file names, relative to the run's `artifacts/` directory.
    pub artifacts: Vec<String>,
    /// Rate-limit / quota exhaustion, when it can be told apart from a normal
    /// failure. `None` for a normal failure, a timeout, or a CLI we cannot
    /// read — the conservative default.
    #[serde(default)]
    pub quota: Option<Quota>,
    /// The CLI hung up on its own stream after the agent had done billed
    /// work. `None` unless that exact shape was recognised — see
    /// [`dropped_stream`].
    #[serde(default)]
    pub dropped: Option<Dropped>,
    /// Commands the CLI's own event stream reported running, see
    /// [`CommandEvidence`]. Always empty for a backend this crate does not
    /// currently read structured job events from.
    #[serde(default)]
    pub commands: Vec<CommandEvidence>,
    /// Input-side tokens the CLI reported for this turn - approximately the
    /// size of the context the model just read. `None` when the CLI reports
    /// no usage (or in a shape this build does not read): unknown, never 0.
    /// See [`context_tokens`].
    #[serde(default)]
    pub context_tokens: Option<u64>,
}

impl AgentOutput {
    /// Did the CLI exit cleanly with something to say?
    pub fn usable(&self) -> bool {
        !self.timed_out && self.exit_code == Some(0) && !self.text.trim().is_empty()
    }

    /// Did this invocation run out of the CLI's rate limit / quota?
    pub fn quota_exhausted(&self) -> bool {
        self.quota.is_some()
    }

    /// Did the agent work and the CLI fail to deliver it?
    ///
    /// Worth re-asking, unlike [`AgentOutput::quota_exhausted`]: the answer is
    /// in a conversation this process can resume.
    pub fn work_undelivered(&self) -> bool {
        self.dropped.is_some()
    }
}

/// Typed context distinguishing an interrupted process from admission that
/// never spawned one. ResourceBlocked remains in the error chain for parking.
#[derive(Debug)]
pub(crate) struct ExecutionInterrupted(pub String);

impl std::fmt::Display for ExecutionInterrupted {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "execution interrupted: {}", self.0)
    }
}

impl std::error::Error for ExecutionInterrupted {}

/// How long to keep reading a pipe after the child is gone.
///
/// Bounded on purpose: a surviving grandchild can hold the write end open
/// forever, and the graph must not hang on a process it has already killed.
const PIPE_GRACE: Duration = Duration::from_secs(3);

/// Bytes a pipe reader has accumulated so far, shared with whoever spawned it.
pub(crate) type Captured = Arc<Mutex<Vec<u8>>>;

/// Read `pipe` to end in its own task, appending into a buffer the caller can
/// inspect at any time.
///
/// The buffer is shared rather than returned because the interesting moment is
/// exactly the one where the reader has *not* finished: a killed agent's pipe
/// may still be held open by a surviving grandchild, and the bytes that did
/// arrive are the only evidence of what it was doing. An earlier version
/// returned the buffer from the task and dropped it on timeout, which is how
/// `<stem>.out` came to be empty on every timeout.
pub(crate) fn drain<R>(pipe: Option<R>) -> (Captured, Option<tokio::task::JoinHandle<()>>)
where
    R: tokio::io::AsyncRead + Unpin + Send + 'static,
{
    let buf: Captured = Arc::new(Mutex::new(Vec::new()));
    let Some(mut pipe) = pipe else {
        return (buf, None);
    };
    let sink = Arc::clone(&buf);
    let handle = tokio::spawn(async move {
        let mut chunk = [0u8; 8192];
        loop {
            match pipe.read(&mut chunk).await {
                Ok(0) | Err(_) => break,
                Ok(n) => {
                    if let Ok(mut guard) = sink.lock() {
                        guard.extend_from_slice(&chunk[..n]);
                    }
                }
            }
        }
    });
    (buf, Some(handle))
}

/// Take whatever a reader has captured, giving it at most `grace` to finish.
///
/// A reader still blocked after that is abandoned, not awaited — but its bytes
/// come back either way, which is the whole point.
pub(crate) async fn collect(
    buf: &Captured,
    handle: Option<tokio::task::JoinHandle<()>>,
    grace: Duration,
) -> String {
    if let Some(handle) = handle {
        if tokio::time::timeout(grace, handle).await.is_err() {
            tracing::debug!("a pipe is still held open after the child exited");
        }
    }
    let bytes = buf.lock().map(|g| g.clone()).unwrap_or_default();
    String::from_utf8_lossy(&bytes).into_owned()
}

/// Invoke `spec` for `seat`, updating the seat's conversation state.
pub async fn invoke(
    spec: &AgentSpec,
    seat: &mut SeatState,
    inv: &Invocation<'_>,
) -> Result<AgentOutput> {
    // Wait for repository scheduling before spawning or starting a seat's
    // budget. Readers hold the real lock; editors exclude local readers while
    // taking their own repository-prescribed lock inside the seat.
    let lease = if inv.scope == InvocationScope::ConsultationInvestigation {
        crate::repo_lock::SeatLease::try_reader(inv.cwd, inv.coordination).await?
    } else if repository_seat(inv) {
        crate::repo_lock::SeatLease::for_checkout(inv.cwd, inv.allow_write, inv.coordination)
            .await?
    } else {
        None
    };
    invoke_leased(spec, seat, inv, lease, || Ok(())).await
}

/// Also used by consultation continuations after nonblocking acquisition and
/// revalidation. Running owns the lease through descendant cleanup on cancel.
pub(crate) async fn invoke_leased(
    spec: &AgentSpec,
    seat: &mut SeatState,
    inv: &Invocation<'_>,
    lease: Option<crate::repo_lock::SeatLease>,
    start_check: impl FnOnce() -> Result<()> + Send,
) -> Result<AgentOutput> {
    let readonly = inv.scope == InvocationScope::ConsultationInvestigation;
    let inv = &Invocation {
        timeout: spec.turn_timeout(inv.timeout),
        allow_write: inv.allow_write && !readonly,
        unsandboxed: inv.unsandboxed && inv.scope == InvocationScope::Repository,
        writable: if readonly { &[] } else { inv.writable },
        ..inv.clone()
    };
    let mut outputs = Vec::new();
    if repository_seat(inv) {
        if let Some(guard) = crate::retention::OutputLease::borrow(inv.cwd).map_err(|e| {
            crate::repo_lock::ResourceBlocked(format!("worktree output ownership: {e:#}"))
        })? {
            outputs.push(guard);
        }
        if let Some(cache) = inv.cache_dir {
            outputs.push(crate::retention::OutputLease::cache(
                &crate::run::home(),
                cache,
                false,
            )?);
        }
    }

    let scope = match inv.scope {
        InvocationScope::ConsultationControl => Some(
            "Current turn scope: consultation control. This replaces all earlier per-turn scope restrictions, including investigation's prohibition on answering. Repository access is suspended. Use supplied context and supported read-only magi control-plane diagnostics; do not inspect repository files/history, fetch, test or edit. If evidence is needed (including an operator request to investigate), return ONLY a JSON object {\"investigate_question\":\"<full open question id>\"}. The launcher will attempt a coordinated read-only continuation in this same conversation. Do not answer or resolve the question to obtain access. Only magi answer settles a question. Additional writes are limited to individually evidenced task add, release, hold, done and priority; use --quote of the entire latest persisted owner send after delivery. No direct merge, discard, withdrawal, posting or other write commands; existing answer-specific effect checks remain in force. These restrictions last this whole invocation even after consultation ends. Consultation task instructions use the literal task-operation JSON protocol explained below. --reply-other records the owner's literal decision and ends consultation only, leaving the question open and waiter held. Owner decisions still require magi answer with exact required evidence.",
        ),
        InvocationScope::ConsultationInvestigation => Some(
            "Current turn scope: consultation investigation. This supersedes earlier control-only restrictions for this invocation. Investigate repository evidence within the standing conversation's read-only permissions. No repository edits, fetch or other Git/network mutations; use existing history and files. Do not run tests that write into the repository. Do not answer, close, withdraw, discard or resume anything, or invoke magi answer. Return evidence and remaining decision points; the next operator turn returns to consultation control while the question remains open.",
        ),
        InvocationScope::Repository if inv.node == "chat" => Some(
            "Current turn scope: ordinary repository conversation. This replaces earlier per-turn consultation restrictions. Use the standing conversation permissions and current filesystem sandbox; owner decisions still require magi answer and exact authorization evidence.",
        ),
        InvocationScope::Repository => None,
    };
    let mut prompt = inv.prompt.to_owned();
    if let Some(scope) = scope {
        prompt.push_str("\n\n");
        prompt.push_str(scope);
        if inv.scope == InvocationScope::ConsultationControl {
            prompt.push_str("\n\n");
            prompt.push_str(crate::prompt::CONSULT_WRITES);
        }
    }
    if let Some(lease) = lease.as_ref().filter(|lease| lease.delegated) {
        prompt.push_str(&format!("\n\nRepository coordination: the launcher holds the exclusive repository lock at {} for this invocation. Do not acquire a second lock or release/delete it. Repository access remains read-only.", lease.path.display()));
    }
    // Apply once for initial, resumed, fallback and waiter invocations alike.
    // Deputies/chat retain their explicit grants; read-only stays read-only.
    let home = repository_writer(inv).then(crate::run::home);
    let mut writable = inv.writable.to_vec();
    if let Some(home) = &home {
        for root in worktree_writable_roots(inv.cwd, home).await? {
            if !writable.contains(&root) {
                writable.push(root);
            }
        }
    }
    if repository_writer(inv) {
        prompt.push_str("\n\n");
        prompt.push_str(&editing_access_note(
            spec.kind,
            inv.unsandboxed,
            inv.fetched_base,
        ));
    }
    let inv = &Invocation {
        writable: &writable,
        prompt: &prompt,
        ..inv.clone()
    };
    tokio::fs::create_dir_all(inv.artifacts)
        .await
        .with_context(|| format!("create {}", inv.artifacts.display()))?;
    let prompt_path = inv.artifacts.join(format!("{}.prompt.md", inv.stem));
    tokio::fs::write(&prompt_path, inv.prompt)
        .await
        .with_context(|| format!("write {}", prompt_path.display()))?;

    let plan = build_command(spec, seat, inv, &prompt_path)?;
    tracing::debug!(seat = %seat.key, agent = %spec.id, argv = ?plan.argv, "spawning agent");
    // Store scope evidence without environment values or arbitrary extra args.
    tokio::fs::write(
        inv.artifacts.join(format!("{}.access.json", inv.stem)),
        serde_json::to_vec_pretty(&serde_json::json!({
            "launcher_version": env!("CARGO_PKG_VERSION"),
            "kind": spec.kind,
            "allow_write": inv.allow_write,
            "unsandboxed": inv.unsandboxed,
            "requested_network_access": if spec.kind == AgentKind::Codex && repository_writer(inv) && !inv.unsandboxed {
                Some(false)
            } else { None },
            "writable_roots": inv.writable,
            "fetched_base": inv.fetched_base,
        }))?,
    )
    .await
    .context("write seat access evidence")?;

    let started = Instant::now();
    // Resolve against the PATH the child will actually see: `spec.env` may
    // override it, and the resolved absolute path bypasses any later lookup.
    let child_path = spec
        .env
        .iter()
        .find(|(k, _)| k.eq_ignore_ascii_case("PATH"))
        .map(|(_, v)| std::ffi::OsString::from(v))
        .or_else(|| std::env::var_os("PATH"))
        .unwrap_or_default();
    let program = crate::config::find_program_on(&plan.argv[0], &child_path).map_or_else(
        || plan.argv[0].clone().into(),
        std::path::PathBuf::into_os_string,
    );
    let mut cmd = Command::new(program);
    cmd.args(&plan.argv[1..])
        .current_dir(inv.cwd)
        .envs(&spec.env)
        .env("MAGI_SEAT", &seat.key)
        .env("MAGI_TURN", seat.turns.to_string())
        .env("MAGI_RUN", inv.run)
        .env("MAGI_NODE", inv.node)
        .env(
            "MAGI_TURN_SCOPE",
            match inv.scope {
                InvocationScope::ConsultationControl => "consultation-control",
                InvocationScope::ConsultationInvestigation => "consultation-investigation",
                InvocationScope::Repository => "repository",
            },
        )
        .env("MAGI_PROMPT_FILE", &prompt_path)
        .env("MAGI_ALLOW_WRITE", if inv.allow_write { "1" } else { "0" })
        .env("GIT_TERMINAL_PROMPT", "0")
        .stdin(if plan.stdin.is_some() {
            Stdio::piped()
        } else {
            Stdio::null()
        })
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true)
        // No console window. `magi web` has no console of its own, so Windows
        // would give each agent a fresh one - and draw it. See `crate::proc`.
        .quiet();
    if let Some(home) = home {
        // Match the control-plane store we granted, even with a relative
        // MAGI_HOME or a per-agent environment override.
        cmd.env("MAGI_HOME", std::fs::canonicalize(home)?);
    }
    if let Some(cache) = inv.cache_dir {
        // Same directory the verify commands build into: one cache to prune,
        // and the compile the seat pays for is the compile the gate reuses.
        cmd.env("CARGO_TARGET_DIR", cache);
    } else {
        // `Command` inherits this process's environment by default, so
        // simply not setting the variable here is not the same as the seat
        // not seeing it: if the magi process itself is running under a
        // shared `CARGO_TARGET_DIR` (the ordinary case), a read-only seat
        // would otherwise inherit that exact path and try to build there
        // anyway - the write refusal this is meant to prevent in the first
        // place. Strip it explicitly.
        cmd.env_remove("CARGO_TARGET_DIR");
    }

    #[cfg(any(unix, windows))]
    let _chat_controls = {
        use crate::chat_control::bridge::{self, Bridge};
        bridge::clear_environment(&mut cmd);
        let bridge = if inv.node == crate::queue::CHAT_NODE
            && inv.scope == InvocationScope::Repository
        {
            crate::run::try_home().and_then(|home| {
                match Bridge::start(&home, inv.run, inv.artifacts) {
                    Ok(bridge) => Some(bridge),
                    Err(error) => {
                        tracing::warn!(%error, "chat controls unavailable; continuing conversation");
                        None
                    }
                }
            })
        } else {
            None
        };
        if let Some(bridge) = &bridge {
            // Read-only chat CLIs also authorize against the launcher's talk
            // store before dispatching through the bridge. Match its home
            // without granting the child filesystem write access.
            cmd.env("MAGI_HOME", std::fs::canonicalize(crate::run::home())?);
            bridge.attach(&mut cmd);
        }
        bridge
    };

    // The final consultation/turn check is synchronous: no await separates
    // it from spawning and transferring the lease to Running.
    start_check()?;
    let mut running = crate::seat_process::Running::spawn(cmd, lease, outputs)
        .with_context(|| format!("spawn `{}` for seat {}", plan.argv[0], seat.key))?;
    // Feed stdin from a task rather than inline: a `command` agent that never
    // reads its stdin, or a prompt larger than the pipe buffer, would
    // otherwise deadlock here before the process is ever waited on.
    if let (Some(body), Some(mut sink)) = (plan.stdin.clone(), running.child().stdin.take()) {
        tokio::spawn(async move {
            sink.write_all(body.as_bytes()).await.ok();
            sink.shutdown().await.ok();
        });
    }

    // Drain the pipes in their own tasks, and wait on the *process*, not on
    // end-of-file. Two failures come out of conflating those:
    //
    // 1. `wait_with_output` returns when both pipes reach EOF, which is not
    //    when the child exits. A CLI that leaves a helper process holding the
    //    inherited stdout handle - normal on Windows, where a `.cmd` shim and
    //    its grandchildren share handles - never closes the pipe, so a seat
    //    that answered in five minutes was billed the full hour and then
    //    recorded as a timeout. The answer was thrown away with it.
    // 2. Cancelling `wait_with_output` at the timeout drops the buffers it
    //    owned, so `<stem>.out` and `<stem>.err` were written empty exactly
    //    when an operator needs them most. "It printed nothing" and "we
    //    discarded what it printed" looked identical on disk.
    //
    // Now the readers own the bytes, so a timeout keeps whatever arrived, and
    // the wait ends at exit even if a stray handle stays open.
    let (out_buf, out_reader) = drain(running.child().stdout.take());
    let (err_buf, err_reader) = drain(running.child().stderr.take());

    let min_free = crate::run::RunState::load_under(inv.run, &crate::run::home())
        .ok()
        .map_or(0, |state| state.config.disk.min_free_bytes);
    let mut paths = vec![inv.cwd.to_owned()];
    if let Some(cache) = inv.cache_dir {
        paths.push(cache.to_owned());
    }
    let mut pressure = None;
    let waited = tokio::select! {
        waited = tokio::time::timeout(inv.timeout, running.child().wait()) => Some(waited),
        reason = crate::retention::emergency(paths, min_free) => {
            pressure = Some(reason);
            None
        }
    };
    let (code, timed_out) = match waited {
        Some(Ok(res)) => {
            let status = res.with_context(|| format!("wait for seat {}", seat.key))?;
            (status.code(), false)
        }
        Some(Err(_)) | None => {
            if let Some(reason) = &pressure {
                tracing::warn!(seat = %seat.key, %reason, "agent execution interrupted");
            } else {
                tracing::warn!(seat = %seat.key, secs = inv.timeout.as_secs(), "agent timed out");
            }
            // Wait for CLI termination before releasing its repository lease.
            if !running.coordinated() {
                running
                    .child()
                    .kill()
                    .await
                    .context("stop timed-out seat")?;
            }
            (None, true)
        }
    };
    running.finish().await;

    // The child is gone either way, so the readers are bounded now. A grace
    // window rather than an unbounded await: a surviving grandchild can still
    // hold the write end open, and losing a few trailing bytes beats hanging
    // the graph on a process we no longer control.
    let stdout = collect(&out_buf, out_reader, PIPE_GRACE).await;
    let stderr = collect(&err_buf, err_reader, PIPE_GRACE).await;

    let out_path = inv.artifacts.join(format!("{}.out", inv.stem));
    let err_path = inv.artifacts.join(format!("{}.err", inv.stem));
    tokio::fs::write(&out_path, &stdout).await.ok();
    tokio::fs::write(&err_path, &stderr).await.ok();
    if let Some(reason) = pressure {
        return Err(
            anyhow::Error::new(crate::repo_lock::ResourceBlocked(reason.clone()))
                .context(ExecutionInterrupted(reason)),
        );
    }

    let mut extracted = extract(spec.kind, &stdout);
    if spec.kind == AgentKind::Antigravity && extracted.quota.is_none() {
        extracted.quota = agy_quota(&stdout, &stderr);
    }
    if let Some(quota) = &extracted.quota {
        // A quota response carries usage too, so it can look like a dropped
        // stream; a rate limit is never worth re-asking, so it wins.
        extracted.dropped = None;
        tracing::warn!(
            seat = %seat.key,
            agent = %spec.id,
            reset = ?quota.reset,
            "agent is out of quota"
        );
    }
    if let Some(session) = extracted.session {
        match spec.kind {
            AgentKind::Claude => seat.claude_session = Some(session),
            AgentKind::Opencode | AgentKind::Antigravity | AgentKind::Codex | AgentKind::Omp => {
                seat.captured_session = Some(session);
            }
            AgentKind::Command => {}
        }
    }
    if let Some(status) = &extracted.status
        && !status.eq_ignore_ascii_case("success")
    {
        tracing::warn!(seat = %seat.key, status = %status, "agent reported a non-success status");
    }
    let text = if extracted.text.trim().is_empty() {
        // A CLI that printed only to stderr still told us something.
        if stdout.trim().is_empty() {
            stderr.trim().to_owned()
        } else {
            stdout.trim().to_owned()
        }
    } else {
        extracted.text
    };
    seat.turns += 1;

    Ok(AgentOutput {
        text,
        exit_code: code,
        timed_out,
        duration_ms: started.elapsed().as_millis() as u64,
        artifacts: vec![
            file_name(&prompt_path),
            file_name(&out_path),
            file_name(&err_path),
        ],
        quota: extracted.quota,
        dropped: extracted.dropped,
        commands: extracted.commands,
        context_tokens: extracted.context_tokens,
    })
}

fn file_name(p: &Path) -> String {
    p.file_name()
        .unwrap_or_default()
        .to_string_lossy()
        .into_owned()
}

/// The argv plus optional stdin body for one invocation.
#[derive(Debug)]
struct Plan {
    argv: Vec<String>,
    stdin: Option<String>,
}

/// How a file-delivered prompt is pointed at, per CLI.
///
/// `agy` has a native file-context syntax, `@<path>`, and it is measurably the
/// better contract: on the same trivial task it finished in 17s against 73s for
/// the prose form, because prose makes the model spend a tool round-trip
/// deciding to read the file. It is also the form yukimemi/rvpm proved out.
///
/// opencode has no equivalent, so it gets the prose. That is not a fallback
/// worth apologising for — it works, and it is what the winning opencode
/// candidates on this repository have been driven by all along.
fn pointer(kind: AgentKind, prompt_path: &Path) -> String {
    if matches!(kind, AgentKind::Antigravity) {
        return format!("@{}", prompt_path.display());
    }
    format!(
        "Read the file at {} and follow every instruction in it exactly. That \
         file is your complete task description; this message contains nothing \
         else.",
        prompt_path.display()
    )
}

fn build_command(
    spec: &AgentSpec,
    seat: &SeatState,
    inv: &Invocation<'_>,
    prompt_path: &Path,
) -> Result<Plan> {
    let mut argv: Vec<String> = Vec::new();
    let mut stdin: Option<String> = None;
    let delivery = spec.delivery();
    let resuming = has_session(spec.kind, seat, inv.sessions);

    match spec.kind {
        AgentKind::Claude => {
            // Claude's own tools have no cwd-confined sandbox - the CLI can
            // already `Read` any absolute path magi hands it, an attachment
            // outside the repository included - so no extra flag is needed
            // here.
            argv.push("claude".to_owned());
            argv.push("-p".to_owned());
            argv.push("--output-format".to_owned());
            argv.push("json".to_owned());
            if let Some(m) = &spec.model {
                argv.push("--model".to_owned());
                argv.push(m.clone());
            }
            if inv.sessions {
                let uuid = seat
                    .claude_session
                    .as_deref()
                    .context("claude seat is missing its session uuid")?;
                argv.push(if resuming { "--resume" } else { "--session-id" }.to_owned());
                argv.push(uuid.to_owned());
            }
            argv.push("--permission-mode".to_owned());
            argv.push("bypassPermissions".to_owned());
            if !inv.allow_write {
                argv.push("--disallowed-tools".to_owned());
                argv.push("Edit,Write,MultiEdit,NotebookEdit".to_owned());
            }
        }
        AgentKind::Opencode => {
            // `--auto` below already bypasses every permission, reads of a
            // path outside `--dir` included, so an attachment elsewhere
            // needs no extra flag.
            argv.push("opencode".to_owned());
            argv.push("run".to_owned());
            argv.push("--format".to_owned());
            argv.push("json".to_owned());
            argv.push("--dir".to_owned());
            argv.push(inv.cwd.to_string_lossy().into_owned());
            // `--auto` gates *every* permission, reads included: without it a
            // non-interactive opencode cannot even open the prompt file, and
            // the seat drops out of the panel with "the user rejected
            // permission to use this specific tool call". opencode has no
            // read-only mode, so read-only seats rely on the prompt plus the
            // fact that judge and reviewer worktrees are disposable — judges'
            // are deleted after the tally, reviewers' are reset to the commit
            // under review every round.
            argv.push("--auto".to_owned());
            if let Some(m) = &spec.model {
                argv.push("-m".to_owned());
                argv.push(m.clone());
            }
            if resuming {
                argv.push("-s".to_owned());
                argv.push(
                    seat.captured_session
                        .clone()
                        .expect("has_session checked the id is present"),
                );
            }
        }
        AgentKind::Antigravity => {
            argv.push("agy".to_owned());
            argv.push("--output-format".to_owned());
            argv.push("json".to_owned());
            // agy's print mode gives up after 5 minutes by default, which is
            // far below an implementation node's budget.
            argv.push("--print-timeout".to_owned());
            argv.push(format!("{}s", inv.timeout.as_secs()));
            argv.push("--mode".to_owned());
            argv.push(
                if inv.allow_write {
                    "accept-edits"
                } else {
                    "plan"
                }
                .to_owned(),
            );
            if inv.allow_write {
                argv.push("--dangerously-skip-permissions".to_owned());
            }
            if let Some(m) = &spec.model {
                argv.push("--model".to_owned());
                argv.push(m.clone());
            }
            if resuming {
                argv.push("--conversation".to_owned());
                argv.push(
                    seat.captured_session
                        .clone()
                        .expect("has_session checked the id is present"),
                );
            }
            // The prompt file lives outside the worktree, so the workspace has
            // to be widened to reach it - and so does an attachment's own
            // directory, which usually lives right beside it under the
            // conversation's `artifacts_dir` (see `chat`/`talk`). "Usually":
            // a chat derived from another one (`chat::derived_background`)
            // can carry attachment paths that live under the *source*
            // conversation's own artifacts dir instead, so each attachment
            // outside `inv.artifacts` gets its own `--add-dir` rather than
            // assuming one directory covers all of `inv.attachments`.
            let mut add_dirs: Vec<String> = Vec::new();
            if delivery == Delivery::File || !inv.attachments.is_empty() {
                add_dirs.push(inv.artifacts.to_string_lossy().into_owned());
            }
            for path in inv.attachments {
                let Some(parent) = path.parent() else {
                    continue;
                };
                if parent.starts_with(inv.artifacts) {
                    continue;
                }
                let dir = parent.to_string_lossy().into_owned();
                if !add_dirs.contains(&dir) {
                    add_dirs.push(dir);
                }
            }
            if inv.allow_write {
                for path in inv.writable {
                    let dir = path.to_string_lossy().into_owned();
                    if !add_dirs.contains(&dir) {
                        add_dirs.push(dir);
                    }
                }
            }
            for dir in add_dirs {
                argv.push("--add-dir".to_owned());
                argv.push(dir);
            }
        }
        AgentKind::Codex => {
            // `--sandbox` below governs writes, not reads (see the module
            // doc: it is what makes codex the one kind whose *read-only*
            // mode is enforced, by refusing edits) - both presets can read
            // anywhere the OS lets the process, so an attachment outside
            // `cwd` is already reachable without an extra flag.
            argv.push("codex".to_owned());
            argv.push("exec".to_owned());
            argv.push("--json".to_owned());
            // The worktrees magi hands out are real checkouts, but a judge's
            // is detached and a fixture's may be no repository at all.
            argv.push("--skip-git-repo-check".to_owned());
            argv.push("-C".to_owned());
            argv.push(inv.cwd.to_string_lossy().into_owned());
            // Codex is the only kind whose read-only-ness is enforced by the
            // CLI rather than by the prompt: a judge or reviewer seat cannot
            // write even if it decides to try. Implementers get the workspace.
            // The one exception is a conversational talk turn that the
            // repository opted into (`unsandboxed` and `allow_write`): it
            // drops sandbox and approvals altogether, as claude's
            // `bypassPermissions` does, so `magi task add`, `git fetch` and
            // the like are not refused. Nothing else ever gets the bypass.
            if inv.allow_write && inv.unsandboxed {
                argv.push("--dangerously-bypass-approvals-and-sandbox".to_owned());
            } else {
                argv.push("--sandbox".to_owned());
                argv.push(
                    if inv.allow_write {
                        "workspace-write"
                    } else {
                        "read-only"
                    }
                    .to_owned(),
                );
                // Nothing is watching to approve anything: an unattended seat
                // that asks blocks until its timeout kills it.
                argv.push("-c".to_owned());
                argv.push("approval_policy=\"never\"".to_owned());
                if repository_writer(inv) {
                    argv.push("-c".to_owned());
                    argv.push("sandbox_workspace_write.network_access=false".to_owned());
                }
                if inv.allow_write && !inv.writable.is_empty() {
                    let roots: Vec<String> = inv
                        .writable
                        .iter()
                        .map(|p| p.to_string_lossy().into_owned())
                        .collect();
                    argv.push("-c".to_owned());
                    argv.push(format!(
                        "sandbox_workspace_write.writable_roots={}",
                        serde_json::to_string(&roots).unwrap_or_else(|_| "[]".to_owned())
                    ));
                }
            }
            if let Some(m) = &spec.model {
                argv.push("-m".to_owned());
                argv.push(m.clone());
            }
            // `resume` is a subcommand of `exec`, and it rejects the flags
            // above when they follow it - so every option is emitted first and
            // the subcommand last. Established by hand against codex-cli
            // 0.153.4: with the order reversed the CLI exits on
            // `unexpected argument '--sandbox'`.
            if resuming {
                argv.push("resume".to_owned());
                argv.push(
                    seat.captured_session
                        .clone()
                        .expect("has_session checked the id is present"),
                );
            }
        }
        AgentKind::Omp => {
            // `omp` reads the prompt from stdin in print mode (see
            // `AgentSpec::delivery`), so the whole instruction arrives without
            // an argv length limit - the same reason codex gets stdin.
            argv.push("omp".to_owned());
            argv.push("-p".to_owned());
            argv.push("--mode=json".to_owned());
            // `--auto-approve` is required, and is the same trade opencode's
            // `--auto` makes: it gates *every* permission, reads included, so
            // without it a non-interactive seat cannot even open the prompt
            // file magi wrote and drops out of the panel on a permission
            // rejection. `omp` has no read-only mode of its own, so a judge or
            // reviewer seat rests on the prompt plus the worktree discipline
            // (judge worktrees are deleted after the tally, reviewer worktrees
            // are reset to the commit under review every round) - never on this
            // flag, and never on a bypass flag.
            argv.push("--auto-approve".to_owned());
            if let Some(m) = &spec.model {
                argv.push("--model".to_owned());
                argv.push(m.clone());
            }
            // Established by hand against omp 18.1.19: `-p --mode=json` reports
            // the session id on its `"type":"session"` line, and
            // `--resume <id>` continues that conversation. `--continue` is
            // deliberately not used - it opens a *new* session rather than the
            // stored one, which silently loses the seat's memory.
            if resuming {
                argv.push("--resume".to_owned());
                argv.push(
                    seat.captured_session
                        .clone()
                        .expect("has_session checked the id is present"),
                );
            }
        }
        AgentKind::Command => {
            // The operator's own command line, not one of the roster CLIs -
            // there is no flag this function could add on its behalf, so an
            // attachment's path has to reach it the same way the prompt
            // does, through the substitutions below.
            if spec.command.is_empty() {
                bail!("agent `{}` has kind = \"command\" but no command", spec.id);
            }
            let vars: BTreeMap<&str, String> = BTreeMap::from([
                ("{prompt_file}", prompt_path.to_string_lossy().into_owned()),
                ("{cwd}", inv.cwd.to_string_lossy().into_owned()),
                ("{label}", seat.key.clone()),
                ("{session}", seat.claude_session.clone().unwrap_or_default()),
            ]);
            for raw in &spec.command {
                let mut arg = raw.clone();
                for (k, v) in &vars {
                    if arg.contains(k) {
                        arg = arg.replace(k, v);
                    }
                }
                argv.push(arg);
            }
        }
    }

    argv.extend(spec.extra_args.iter().cloned());

    // `agy` takes the prompt as the value of `-p`, so the flag has to be
    // emitted right before whatever the delivery mode produces.
    if spec.kind == AgentKind::Antigravity {
        argv.push("-p".to_owned());
    }
    // `codex exec` reads stdin only when its prompt argument is `-`; without
    // it the CLI waits on a prompt it will never be given.
    if spec.kind == AgentKind::Codex && delivery == Delivery::Stdin {
        argv.push("-".to_owned());
    }
    match delivery {
        Delivery::Stdin if spec.kind == AgentKind::Antigravity => {
            // agy has no text stdin path; fall back to the pointer file.
            argv.push(pointer(spec.kind, prompt_path));
        }
        Delivery::Stdin => stdin = Some(inv.prompt.to_owned()),
        Delivery::Argv => argv.push(inv.prompt.to_owned()),
        Delivery::File => argv.push(pointer(spec.kind, prompt_path)),
    }

    Ok(Plan { argv, stdin })
}

/// What a CLI's stdout yielded.
#[derive(Debug, Default)]
struct Extracted {
    text: String,
    session: Option<String>,
    status: Option<String>,
    quota: Option<Quota>,
    dropped: Option<Dropped>,
    commands: Vec<CommandEvidence>,
    context_tokens: Option<u64>,
}

/// Pull the agent's message (and any session id) out of a CLI's stdout.
fn extract(kind: AgentKind, stdout: &str) -> Extracted {
    let mut extracted = extract_answer(kind, stdout);
    extracted.context_tokens = context_tokens(kind, stdout);
    extracted
}

/// `v[key]` as an unsigned integer, or `None` when absent or not a number.
fn uint(v: &serde_json::Value, key: &str) -> Option<u64> {
    v.get(key).and_then(serde_json::Value::as_u64)
}

/// The input side of a usage object: `input` plus whatever the cache served
/// or stored, each key read under the names the CLIs use. A missing cache
/// counter is 0 (the CLI simply had no cache traffic to report) but a missing
/// *input* counter makes the whole thing unknown - `None`, never a made-up 0.
fn input_side(usage: &serde_json::Value, input: &[&str], cache: &[&str]) -> Option<u64> {
    let first = |keys: &[&str]| keys.iter().find_map(|k| uint(usage, k));
    let base = first(input)?;
    let cached: u64 = cache.iter().filter_map(|k| uint(usage, k)).sum();
    Some(base.saturating_add(cached))
}

/// The context size a CLI reported for its last turn: its input-side token
/// count, which approximates how much of the model's window the conversation
/// occupies. A pure function of stdout, so each CLI's shape is unit-testable.
///
/// Rules, all deliberate:
/// - **Last report wins, never a sum.** A multi-event stream (opencode's
///   `step_finish`, codex's `turn.completed`, omp's `message_end`) reports the
///   input of each model call; the final one is the largest context the
///   conversation reached this turn, and adding them would count the same
///   prefix once per tool round-trip.
/// - **Missing or mistyped usage is `None`**, which the UI shows as "unknown".
/// - **A cumulative total is not a context size, so it is `None`.** claude's
///   result `usage` sums every model call of the turn and, on a resumed
///   session, carries the session's earlier usage too, so even `num_turns == 1`
///   does not make it one call's context. codex's `turn.completed.usage` is the
///   session's running input total, and agy's `usage` aggregates the whole
///   print-mode call (the observed sample exceeds any window). None of the
///   three carries a per-call figure, so each is unknown rather than made up.
///   Only the streams that report each model call (opencode `step_finish`, omp
///   message usage) yield a number.
fn context_tokens(kind: AgentKind, stdout: &str) -> Option<u64> {
    use serde_json::Value;
    let lines = || {
        stdout
            .lines()
            .filter_map(|l| serde_json::from_str::<Value>(l.trim()).ok())
    };
    match kind {
        AgentKind::Claude => None,
        AgentKind::Opencode => lines()
            .filter(|v| {
                // The event is `step_finish`; its part is `step-finish`.
                v.get("type").and_then(|t| t.as_str()) == Some("step_finish")
                    || v.get("part")
                        .and_then(|p| p.get("type"))
                        .and_then(|t| t.as_str())
                        == Some("step-finish")
            })
            .filter_map(|v| {
                let tokens = v.get("part")?.get("tokens")?;
                let base = uint(tokens, "input")?;
                let cache = tokens.get("cache");
                let cached = ["read", "write"]
                    .iter()
                    .filter_map(|k| cache.and_then(|c| uint(c, k)))
                    .sum::<u64>();
                Some(base.saturating_add(cached))
            })
            .next_back(),
        AgentKind::Antigravity => None,
        AgentKind::Codex => None,
        AgentKind::Omp => lines()
            .flat_map(|v| {
                // Same walk as the answer: `agent_end` carries the thread,
                // `turn_end` / `message_end` one message each.
                match v.get("type").and_then(|t| t.as_str()) {
                    Some("agent_end") => v
                        .get("messages")
                        .and_then(|m| m.as_array())
                        .cloned()
                        .unwrap_or_default(),
                    Some("turn_end") | Some("message_end") => {
                        v.get("message").cloned().into_iter().collect()
                    }
                    _ => Vec::new(),
                }
            })
            .filter(|m| m.get("role").and_then(|r| r.as_str()) == Some("assistant"))
            .filter_map(|m| {
                input_side(
                    m.get("usage")?,
                    &["input", "input_tokens"],
                    &[
                        "cacheRead",
                        "cacheWrite",
                        "cache_read_tokens",
                        "cache_write_tokens",
                    ],
                )
            })
            .next_back(),
        AgentKind::Command => None,
    }
}

/// The answer text of a stored `<stem>.out`, through the same per-CLI
/// extraction a live call uses. Never the raw event stream.
pub(crate) fn answer_text(kind: AgentKind, stdout: &str) -> String {
    extract_answer(kind, stdout).text
}

fn extract_answer(kind: AgentKind, stdout: &str) -> Extracted {
    match kind {
        AgentKind::Claude => {
            let Ok(v) = serde_json::from_str::<serde_json::Value>(stdout.trim()) else {
                return Extracted {
                    text: stdout.trim().to_owned(),
                    ..Extracted::default()
                };
            };
            Extracted {
                text: v
                    .get("result")
                    .and_then(|r| r.as_str())
                    .unwrap_or_default()
                    .to_owned(),
                session: v
                    .get("session_id")
                    .and_then(|s| s.as_str())
                    .map(str::to_owned),
                status: v.get("is_error").and_then(|e| e.as_bool()).map(|e| {
                    if e {
                        "error".to_owned()
                    } else {
                        "success".to_owned()
                    }
                }),
                quota: claude_quota(&v),
                // Claude reports a truncated stream as an ordinary error; the
                // shape `dropped_stream` keys on is agy's.
                dropped: None,
                commands: Vec::new(),
                context_tokens: None,
            }
        }
        AgentKind::Opencode => {
            // A JSONL event stream: text parts concatenated in arrival order.
            let mut text = String::new();
            let mut session = None;
            for line in stdout.lines() {
                let Ok(v) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
                    continue;
                };
                if session.is_none() {
                    session = v
                        .get("sessionID")
                        .and_then(|s| s.as_str())
                        .map(str::to_owned);
                }
                let part = v.get("part").unwrap_or(&serde_json::Value::Null);
                if part.get("type").and_then(|t| t.as_str()) == Some("text")
                    && let Some(t) = part.get("text").and_then(|t| t.as_str())
                {
                    if !text.is_empty() {
                        text.push('\n');
                    }
                    text.push_str(t);
                }
            }
            Extracted {
                text,
                session,
                status: None,
                quota: None,
                dropped: None,
                commands: Vec::new(),
                context_tokens: None,
            }
        }
        AgentKind::Antigravity => {
            // agy prints warnings before the JSON object, so parse the last
            // line that is one rather than the whole stream.
            let obj = stdout
                .lines()
                .rev()
                .find_map(|l| serde_json::from_str::<serde_json::Value>(l.trim()).ok());
            let Some(v) = obj else {
                return Extracted {
                    text: stdout.trim().to_owned(),
                    ..Extracted::default()
                };
            };
            Extracted {
                text: v
                    .get("response")
                    .and_then(|r| r.as_str())
                    .unwrap_or_default()
                    .trim()
                    .to_owned(),
                session: v
                    .get("conversation_id")
                    .and_then(|s| s.as_str())
                    .map(str::to_owned),
                status: v.get("status").and_then(|s| s.as_str()).map(str::to_owned),
                quota: None,
                dropped: dropped_stream(&v),
                commands: Vec::new(),
                context_tokens: None,
            }
        }
        AgentKind::Codex => {
            // A JSONL event stream, prefixed on a real machine by tracing
            // lines the CLI writes about its own config and skills - so
            // non-JSON lines are skipped rather than treated as the answer.
            //
            // The thread id arrives once, in `thread.started`, and a resumed
            // turn reports the same one. The answer is the last
            // `item.completed` carrying an `agent_message`: earlier ones are
            // the model narrating its way through the tool loop, and taking
            // the first would hand the caller a progress note instead of a
            // verdict.
            //
            // A `command_execution` item is different evidence entirely: the
            // CLI's own record that it ran a command and what that command
            // reported back, kept as `CommandEvidence` — see run
            // 20260912-214939-b3bb's artifacts, where these very fields
            // (`id`/`command`/`exit_code`/`aggregated_output`) were what
            // caught a paired test result magi's own agent prose had missed.
            // Only what `item.completed` actually reports: a command still
            // running when the turn ended emits no such event at all, and is
            // not something this can detect — see [`CommandEvidence`]'s own
            // doc for why that must not be guessed at instead.
            let mut text = String::new();
            let mut session = None;
            let mut status = None;
            let mut commands = Vec::new();
            for line in stdout.lines() {
                let Ok(v) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
                    continue;
                };
                match v.get("type").and_then(|t| t.as_str()) {
                    Some("thread.started") => {
                        session = v
                            .get("thread_id")
                            .and_then(|s| s.as_str())
                            .map(str::to_owned);
                    }
                    Some("item.completed") => {
                        let item = v.get("item").unwrap_or(&serde_json::Value::Null);
                        match item.get("type").and_then(|t| t.as_str()) {
                            Some("agent_message") => {
                                if let Some(t) = item.get("text").and_then(|t| t.as_str()) {
                                    text = t.trim().to_owned();
                                }
                            }
                            Some("command_execution") => {
                                commands.push(command_evidence(item));
                            }
                            _ => {}
                        }
                    }
                    Some("turn.completed") => status = Some("success".to_owned()),
                    Some("turn.failed") => status = Some("error".to_owned()),
                    _ => {}
                }
            }
            Extracted {
                text,
                session,
                status,
                quota: None,
                dropped: None,
                commands,
                context_tokens: None,
            }
        }
        AgentKind::Omp => {
            // A JSONL event stream. The session id arrives once, on the
            // `"type":"session"` line that opens the run.
            //
            // The answer is the *last* non-empty assistant text block anywhere
            // in the stream, and neither of the two obvious shortcuts works:
            //
            // 1. Do not key on `agent_end`. `omp` emits it only for a run that
            //    quiesces on a message turn; a turn that ends on a tool call
            //    (`stopReason: "toolUse"`) ends the run with **no `agent_end`
            //    line at all**, and the answer is in `message_end` / `turn_end`
            //    instead. Reading only `agent_end` silently discards a complete
            //    review - which is exactly what the first hand-written wrapper
            //    did, three times, before this arm existed.
            // 2. Do not take the first assistant text. Earlier ones narrate the
            //    tool loop (sometimes with a single `.`), so the last non-empty
            //    block is the answer and the one before it is a progress note.
            //
            // Every line is parsed independently: a non-JSON line (a CLI
            // warning, a truncated write) is skipped rather than treated as the
            // answer, the same way the codex arm treats its tracing prefix.
            let mut text = String::new();
            let mut session = None;
            for line in stdout.lines() {
                let Ok(v) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
                    continue;
                };
                if v.get("type").and_then(|t| t.as_str()) == Some("session") {
                    session = v.get("id").and_then(|s| s.as_str()).map(str::to_owned);
                    continue;
                }
                // `agent_end` carries the whole thread; `turn_end` and
                // `message_end` each carry one message. Whichever appears, the
                // messages are walked the same way.
                let messages: Vec<&serde_json::Value> = match v.get("type").and_then(|t| t.as_str())
                {
                    Some("agent_end") => v
                        .get("messages")
                        .and_then(|m| m.as_array())
                        .map(|m| m.iter().collect())
                        .unwrap_or_default(),
                    Some("turn_end") | Some("message_end") => {
                        v.get("message").into_iter().collect()
                    }
                    _ => continue,
                };
                for message in messages {
                    if message.get("role").and_then(|r| r.as_str()) != Some("assistant") {
                        continue;
                    }
                    let Some(parts) = message.get("content").and_then(|c| c.as_array()) else {
                        continue;
                    };
                    for part in parts {
                        if part.get("type").and_then(|t| t.as_str()) != Some("text") {
                            continue;
                        }
                        if let Some(t) = part.get("text").and_then(|t| t.as_str())
                            && !t.trim().is_empty()
                        {
                            text = t.trim().to_owned();
                        }
                    }
                }
            }
            Extracted {
                text,
                session,
                status: None,
                quota: None,
                dropped: None,
                commands: Vec::new(),
                context_tokens: None,
            }
        }
        AgentKind::Command => {
            // A `command` agent may wrap a subscription CLI (a fixture, or a
            // thin shim around `claude`). If its output is the claude error
            // shape we recognise the quota the same way, so tests and wrappers
            // do not need their own detection; anything else is just text.
            let parsed = serde_json::from_str::<serde_json::Value>(stdout.trim()).ok();
            let quota = parsed.as_ref().and_then(claude_quota);
            // A `command` fixture may also stand in for a CLI that hangs up on
            // its own stream, which is how that path is tested.
            let dropped = parsed.as_ref().and_then(dropped_stream);
            Extracted {
                text: stdout.trim().to_owned(),
                session: None,
                status: None,
                quota,
                dropped,
                commands: Vec::new(),
                context_tokens: None,
            }
        }
    }
}

/// Build one [`CommandEvidence`] from a Codex `command_execution` item.
///
/// `command` arrives either as a single string or as an argv array,
/// depending on how the CLI shaped the call; both are read rather than
/// assuming one. Missing fields are left at their honest defaults (an empty
/// id/description, `exit_code: None`) rather than guessed at.
fn command_evidence(item: &serde_json::Value) -> CommandEvidence {
    let description = match item.get("command") {
        Some(serde_json::Value::String(s)) => s.clone(),
        Some(serde_json::Value::Array(parts)) => parts
            .iter()
            .filter_map(|p| p.as_str())
            .collect::<Vec<_>>()
            .join(" "),
        _ => String::new(),
    };
    let result_summary = item
        .get("aggregated_output")
        .and_then(|o| o.as_str())
        .map(|s| tail_chars(s.trim(), 400))
        .unwrap_or_default();
    CommandEvidence {
        id: item
            .get("id")
            .and_then(|s| s.as_str())
            .unwrap_or_default()
            .to_owned(),
        description,
        exit_code: item
            .get("exit_code")
            .and_then(serde_json::Value::as_i64)
            .map(|e| e as i32),
        result_summary,
        source: "codex".to_owned(),
    }
}

/// The last `max` characters of `s`, cut on a char boundary.
fn tail_chars(s: &str, max: usize) -> String {
    let count = s.chars().count();
    if count <= max {
        return s.to_owned();
    }
    s.chars().skip(count - max).collect()
}

/// Recognise claude's rate-limit error shape, when it is present.
///
/// Recognises `is_error: true` with a session-limit result or the structured
/// pair `api_error_status: 429` (number or string), `api_error: usage_limit_reached`;
/// every other CLI (and any future shape) returns `None` and is treated as an
/// ordinary failure — the conservative side.
fn claude_quota(v: &serde_json::Value) -> Option<Quota> {
    let is_err = v.get("is_error").and_then(|e| e.as_bool()).unwrap_or(false);
    if !is_err {
        return None;
    }
    let result = v.get("result").and_then(|r| r.as_str()).unwrap_or("");
    let status = v.get("api_error_status");
    let usage_limit = (status.and_then(serde_json::Value::as_u64) == Some(429)
        || status.and_then(serde_json::Value::as_str) == Some("429"))
        && v.get("api_error").and_then(serde_json::Value::as_str) == Some("usage_limit_reached");
    if !usage_limit && !result.to_lowercase().contains("session limit") {
        return None;
    }
    // "…session limit · resets 4:50am (Asia/Tokyo)". The timezone read is not
    // worth parsing exactly; keep the whole phrase after "resets" as free text.
    let reset = result
        .split("resets ")
        .nth(1)
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(str::to_owned);
    Some(Quota { reset })
}

/// Recognise `agy` running out of quota, from either stream.
///
/// Observed shape while out of quota: stdout carries the JSON object
/// `{"status":"ERROR","response":"","error":"Individual quota reached. … Resets
/// in 1h2m49s."}` and stderr carries `AGY_ERROR: {"status":"RESOURCE_EXHAUSTED",
/// "error_code":429,…}`. Either alone is enough (a mangled stdout must not hide
/// a quota that stderr states plainly), but each is keyed on a *pair* of
/// structured fields: `status: ERROR` with the quota text, or
/// `RESOURCE_EXHAUSTED` together with 429. An error status alone, or a 429
/// alone, stays an ordinary failure - the conservative side, as for
/// [`claude_quota`].
///
/// The reset hint is what follows `Resets ` (e.g. `in 1h2m49s`), trailing full
/// stop dropped.
fn agy_quota(stdout: &str, stderr: &str) -> Option<Quota> {
    let from_stdout = stdout
        .lines()
        .rev()
        .find_map(|l| serde_json::from_str::<serde_json::Value>(l.trim()).ok())
        .and_then(|v| {
            let status = v.get("status").and_then(|s| s.as_str()).unwrap_or("");
            let error = v.get("error").and_then(|e| e.as_str()).unwrap_or("");
            (status.eq_ignore_ascii_case("error") && error.to_lowercase().contains("quota reached"))
                .then(|| error.to_owned())
        });
    let from_stderr = || {
        stderr.lines().find_map(|l| {
            let v: serde_json::Value =
                serde_json::from_str(l.trim().strip_prefix("AGY_ERROR:")?.trim()).ok()?;
            let exhausted = v.get("status").and_then(|s| s.as_str()) == Some("RESOURCE_EXHAUSTED");
            let code = v.get("error_code").and_then(serde_json::Value::as_u64) == Some(429);
            (exhausted && code).then(|| {
                v.get("short_error")
                    .and_then(|e| e.as_str())
                    .unwrap_or_default()
                    .to_owned()
            })
        })
    };
    let text = from_stdout.or_else(from_stderr)?;
    let reset = text
        .split_once("Resets ")
        .map(|(_, rest)| rest.trim().trim_end_matches('.').trim())
        .filter(|s| !s.is_empty())
        .map(str::to_owned);
    Some(Quota { reset })
}

/// Recognise a CLI that gave up on its own stream while the agent was working.
///
/// Observed once, verbatim, from `agy` on a candidate that produced nothing:
///
/// ```text
/// {"conversation_id":"36743d06-…","status":"ERROR","response":"",
///  "error":"the connection to the agent was interrupted before the response
///           finished: subscriber fell behind updates, stalled for 5s",
///  "duration_seconds":431.19,"num_turns":1,
///  "usage":{"input_tokens":260113,"output_tokens":14267,
///           "thinking_tokens":9695,"cache_read_tokens":2200925}}
/// ```
///
/// Seven minutes of work and fourteen thousand output tokens, billed, with an
/// empty `response`: the agent did the job and the CLI's own subscriber fell
/// behind and hung up. That is **not** an agent that failed to implement, and
/// counting it as one is how `agy` came to read as 0 wins in 4 entries with
/// five empty candidates - a number that has twice been used to argue the seat
/// out of the roster, and twice been wrong (see `cb6b830`, which reverted the
/// first removal: *"agy does not fail to implement, it fails to report"*).
///
/// The distinction that matters is **billed work with nothing delivered**, so
/// that is what this keys on: an error status, an empty response, and a usage
/// report showing output tokens. Everything else - including an error with no
/// usage at all - returns `None` and stays an ordinary failure, the
/// conservative side, exactly as [`claude_quota`] treats shapes it does not
/// recognise.
///
/// Unlike a quota, this **is** worth re-asking: the work exists in the
/// conversation the CLI just abandoned, and `conversation_id` is right there.
fn dropped_stream(v: &serde_json::Value) -> Option<Dropped> {
    let status = v.get("status").and_then(|s| s.as_str()).unwrap_or("");
    if !status.eq_ignore_ascii_case("error") {
        return None;
    }
    let response = v.get("response").and_then(|r| r.as_str()).unwrap_or("");
    if !response.trim().is_empty() {
        // It answered. Whatever the status says, there is something to read.
        return None;
    }
    let produced = v
        .get("usage")
        .and_then(|u| u.get("output_tokens"))
        .and_then(serde_json::Value::as_u64)
        .unwrap_or(0);
    if produced == 0 {
        // An error with nothing produced is just an error.
        return None;
    }
    Some(Dropped {
        why: v
            .get("error")
            .and_then(|e| e.as_str())
            .unwrap_or("the CLI ended the stream without delivering its answer")
            .trim()
            .to_owned(),
        output_tokens: produced,
    })
}

/// Preflight: which configured agents are not runnable here?
pub fn missing_programs(specs: &[AgentSpec]) -> Vec<String> {
    let mut missing = Vec::new();
    for s in specs {
        let program = match s.kind {
            AgentKind::Command => s.command.first().map(String::as_str),
            other => other.program(),
        };
        if let Some(p) = program
            && !crate::config::which(p)
            && !Path::new(p).is_file()
            && !missing.iter().any(|m: &String| m == p)
        {
            missing.push(p.to_owned());
        }
    }
    missing
}

/// Absolute path of a run's artifact directory.
pub fn artifacts_dir(run_dir: &Path) -> PathBuf {
    run_dir.join("artifacts")
}

/// Can this agent's CLI actually be run on this machine?
pub fn installed(spec: &AgentSpec) -> bool {
    // A `command` agent has no program of its own to look for - its argv is the
    // operator's, and they are the authority on whether it runs.
    spec.kind.program().is_none_or(crate::config::which)
}

/// Choose the agent for a seat that stands alone rather than rotating through
/// the roster: [`crate::talk`]'s standing conversation,
/// or anything else that needs one agent picked once
/// rather than a panel filled in.
///
/// `available` is a parameter rather than a call to [`installed`] so the order
/// below is assertable on a machine with none of these CLIs installed, which is
/// every CI runner.
///
/// The order, and why:
///
/// 1. An explicit id always wins, and is an error rather than a fallback when
///    it is unusable. Naming a seat has a reason, and silently substituting a
///    different model would waste whatever that reason was.
/// 2. Otherwise a [`AgentKind::Claude`] seat, ahead of the roster order: it is
///    the only one of the three CLIs magi can address before the first turn
///    (see this module's own doc on session mechanics), which matters most for
///    a conversation that opens with nothing typed yet.
/// 3. Otherwise the first runnable agent in roster order, because the roster
///    order is the operator's own stated preference and magi has nothing
///    better to go on.
pub fn pick(
    agents: &[AgentSpec],
    want: Option<&str>,
    available: &dyn Fn(&AgentSpec) -> bool,
) -> Result<AgentSpec> {
    if let Some(id) = want {
        let spec = agents
            .iter()
            .find(|a| a.id == id)
            .with_context(|| format!("no agent `{id}` in the roster; it has {}", ids(agents)))?;
        if !available(spec) {
            bail!(
                "agent `{}` needs `{}` on PATH; install it or pass a different \
                 --agent",
                spec.id,
                spec.kind.program().unwrap_or("its command")
            );
        }
        return Ok(spec.clone());
    }

    if agents.is_empty() {
        bail!(
            "the agent roster is empty, so there is nobody to ask: install one \
             of claude, opencode or agy - magi derives a roster from what is on \
             PATH - or add an [[agents]] entry to magi.toml."
        );
    }

    if let Some(spec) = agents
        .iter()
        .find(|a| a.kind == AgentKind::Claude && available(a))
    {
        return Ok(spec.clone());
    }

    agents
        .iter()
        .find(|a| available(a))
        .cloned()
        .with_context(|| {
            let missing = agents
                .iter()
                .filter_map(|a| a.kind.program())
                .collect::<Vec<_>>()
                .join(", ");
            format!(
                "no agent in the roster can be run here: install one of \
                 {missing}, or add an [[agents]] entry to magi.toml for a CLI \
                 you do have"
            )
        })
}

/// Resolve a standalone seat's agent choice into the ordered chain to try.
///
/// Unset (or empty) is [`pick`]'s default order, one entry. Otherwise each
/// named id is kept at its first appearance only - that is what bounds a
/// chain to one attempt per agent - and an id that is not in the roster or
/// not installed is skipped with a warning rather than failing the chain.
/// Errors, naming `role`, only when nothing resolves.
pub fn pick_chain(
    agents: &[AgentSpec],
    choice: Option<&AgentChoice>,
    available: &dyn Fn(&AgentSpec) -> bool,
    role: &str,
) -> Result<Vec<AgentSpec>> {
    let wanted = choice.map(AgentChoice::ids).unwrap_or_default();
    if wanted.is_empty() {
        return pick(agents, None, available)
            .map(|s| vec![s])
            .with_context(|| format!("choose an agent for the {role} role"));
    }
    let mut chain: Vec<AgentSpec> = Vec::new();
    for id in wanted {
        if chain.iter().any(|s| s.id == id) {
            continue;
        }
        match pick(agents, Some(id), available) {
            Ok(spec) => chain.push(spec),
            Err(e) => tracing::warn!("[roles] {role}: skipping `{id}`: {e:#}"),
        }
    }
    if chain.is_empty() {
        bail!(
            "[roles] {role} names no agent that can run here; the roster has {}",
            ids(agents)
        );
    }
    Ok(chain)
}

/// Should a chain move on to its next agent after this call?
///
/// The one place that decides it: an error, a quota hit (judged apart from
/// `usable`, since a CLI can exit 0 with a quota message), or an unusable
/// answer such as a timeout or empty reply.
pub fn chain_advances(outcome: &Result<AgentOutput>) -> bool {
    match outcome {
        Err(e) if e.is::<crate::repo_lock::ResourceBlocked>() => false,
        Err(_) => true,
        Ok(out) => output_advances(out),
    }
}

/// [`chain_advances`] for a call that did return an output: the `Ok` half,
/// shared with callers that classify the call into their own outcome type
/// first (the graph's fixer chain).
pub fn output_advances(out: &AgentOutput) -> bool {
    out.quota_exhausted() || !out.usable()
}

fn ids(agents: &[AgentSpec]) -> String {
    if agents.is_empty() {
        return "no agents at all".to_owned();
    }
    agents
        .iter()
        .map(|a| a.id.clone())
        .collect::<Vec<_>>()
        .join(", ")
}

#[cfg(test)]
mod tests {
    use super::*;

    const COMMAND_HELPER_MODE: &str = "MAGI_TEST_COMMAND_HELPER_MODE";

    /// Test-only command agent implemented by this test binary itself. Unlike
    /// `echo` and `sleep`, it is available wherever the Rust tests run.
    fn command_helper(mode: &str) -> AgentSpec {
        // Invocation's disk monitor reads run state even without a build cache.
        crate::run::pin_test_home();
        AgentSpec {
            id: "helper".to_owned(),
            kind: AgentKind::Command,
            model: None,
            command: vec![
                std::env::current_exe()
                    .expect("locate test helper")
                    .to_string_lossy()
                    .into_owned(),
                "--exact".to_owned(),
                "agent::tests::command_agent_test_helper".to_owned(),
                "--nocapture".to_owned(),
            ],
            extra_args: Vec::new(),
            env: BTreeMap::from([(COMMAND_HELPER_MODE.to_owned(), mode.to_owned())]),
            prompt_delivery: None,
            max_turn_secs: None,
        }
    }

    #[test]
    fn command_agent_test_helper() {
        match std::env::var(COMMAND_HELPER_MODE).as_deref() {
            Ok("reply") => println!("hello {}", std::env::var("MAGI_SEAT").unwrap()),
            Ok("cache") => println!("{}", std::env::var("CARGO_TARGET_DIR").unwrap()),
            Ok("no-cache") => println!(
                "{}",
                std::env::var("CARGO_TARGET_DIR").unwrap_or_else(|_| "ABSENT".to_owned())
            ),
            Ok("ignore-stdin") => println!("done"),
            Ok("chatty-sleep") => {
                println!("i-said-something");
                std::thread::sleep(Duration::from_secs(30));
            }
            Ok("sleep") => std::thread::sleep(Duration::from_secs(30)),
            Ok(other) => panic!("unknown command helper mode {other}"),
            Err(_) => {}
        }
    }

    fn spec(kind: AgentKind, model: Option<&str>) -> AgentSpec {
        AgentSpec {
            id: "a".to_owned(),
            kind,
            model: model.map(str::to_owned),
            command: vec!["echo".to_owned(), "{label}".to_owned()],
            extra_args: Vec::new(),
            env: BTreeMap::new(),
            prompt_delivery: None,
            max_turn_secs: None,
        }
    }

    fn inv<'a>(cwd: &'a Path, art: &'a Path, allow_write: bool) -> Invocation<'a> {
        Invocation {
            scope: crate::agent::InvocationScope::Repository,
            coordination: None,
            fetched_base: None,
            cwd,
            prompt: "do the thing",
            timeout: Duration::from_secs(900),
            allow_write,
            unsandboxed: false,
            sessions: true,
            artifacts: art,
            stem: "t",
            run: "test-run",
            node: "test",
            cache_dir: None,
            attachments: &[],
            writable: &[],
        }
    }

    fn plan_for(kind: AgentKind, seat: &SeatState, allow_write: bool) -> Plan {
        build_command(
            &spec(kind, None),
            seat,
            &inv(Path::new("."), Path::new("/art"), allow_write),
            Path::new("/art/p.md"),
        )
        .unwrap()
    }

    #[test]
    fn editing_access_requires_fetch_evidence_and_keeps_snapshot() {
        let missing = editing_access_note(AgentKind::Codex, false, None);
        assert!(missing.contains("No successful launcher fetch evidence"));
        let base = crate::run::FetchedBase {
            remote: "origin".into(),
            branch: "main".into(),
            commit: "0123456789abcdef".into(),
            fetched_at: "2026-10-18T00:00:00Z".parse().unwrap(),
        };
        let note = editing_access_note(AgentKind::Codex, false, Some(&base));
        for expected in [
            "origin/main",
            "0123456789abcdef",
            "2026-10-18T00:00:00Z",
            "Do not fetch again",
            "launcher saves eligible edits",
            "Specification questions remain valid",
        ] {
            assert!(note.contains(expected), "missing {expected}");
        }
        for kind in [
            AgentKind::Codex,
            AgentKind::Claude,
            AgentKind::Opencode,
            AgentKind::Antigravity,
            AgentKind::Omp,
            AgentKind::Command,
        ] {
            for unsandboxed in [false, true] {
                let scoped = editing_access_note(kind, unsandboxed, Some(&base));
                assert!(scoped.contains(&base.commit));
                assert!(scoped.contains("Do not fetch again"));
                let restricted = kind == AgentKind::Codex && !unsandboxed;
                for claim in [
                    "Network access is restricted",
                    "protected Git administration paths",
                    "launcher saves eligible edits",
                    "unavailable sandbox permissions",
                ] {
                    assert_eq!(scoped.contains(claim), restricted, "{kind:?}/{unsandboxed}");
                    assert_eq!(
                        editing_access_note(kind, unsandboxed, None).contains(claim),
                        restricted,
                        "{kind:?}/{unsandboxed} without fetch evidence",
                    );
                }
            }
        }
        let roundtrip: crate::run::FetchedBase =
            serde_json::from_str(&serde_json::to_string(&base).unwrap()).unwrap();
        assert_eq!(
            editing_access_note(AgentKind::Codex, false, Some(&roundtrip)),
            note
        );
    }

    #[test]
    fn editing_network_policy_is_explicit_on_initial_and_resume() {
        for resumed in [false, true] {
            let mut seat = SeatState::new("impl-A", "a", 7);
            if resumed {
                seat.turns = 1;
                seat.captured_session = Some("fixture-thread".into());
            }
            let mut invocation = inv(Path::new("."), Path::new("/art"), true);
            invocation.node = "implement";
            let plan = build_command(
                &spec(AgentKind::Codex, None),
                &seat,
                &invocation,
                Path::new("/art/p.md"),
            )
            .unwrap();
            let policy = plan
                .argv
                .iter()
                .position(|s| s == "sandbox_workspace_write.network_access=false")
                .unwrap();
            if resumed {
                assert!(policy < plan.argv.iter().position(|s| s == "resume").unwrap());
            }
            assert!(!plan.argv.iter().any(|s| s.contains("bypass")));
            invocation.node = "deputy";
            let deputy = build_command(
                &spec(AgentKind::Codex, None),
                &seat,
                &invocation,
                Path::new("/art/p.md"),
            )
            .unwrap();
            assert!(!deputy.argv.iter().any(|s| s.contains("network_access")));
            assert!(
                !plan_for(AgentKind::Codex, &seat, false)
                    .argv
                    .iter()
                    .any(|s| s.contains("writable_roots"))
            );
        }
    }

    #[test]
    fn claude_mints_then_resumes_the_same_uuid() {
        let mut seat = SeatState::new("judge-1", "a", 7);
        let uuid = seat.claude_session.clone().unwrap();
        let first = plan_for(AgentKind::Claude, &seat, true);
        assert!(first.argv.windows(2).any(|w| w == ["--session-id", &uuid]));
        assert!(!first.argv.iter().any(|a| a == "--resume"));

        seat.turns = 1;
        let second = plan_for(AgentKind::Claude, &seat, true);
        assert!(second.argv.windows(2).any(|w| w == ["--resume", &uuid]));
        assert!(!second.argv.iter().any(|a| a == "--session-id"));
    }

    #[test]
    fn read_only_seats_cannot_edit() {
        let seat = SeatState::new("judge-1", "a", 7);
        let claude = plan_for(AgentKind::Claude, &seat, false);
        assert!(claude.argv.iter().any(|a| a == "--disallowed-tools"));
        assert!(
            !plan_for(AgentKind::Claude, &seat, true)
                .argv
                .iter()
                .any(|a| a == "--disallowed-tools")
        );

        let agy = plan_for(AgentKind::Antigravity, &seat, false);
        assert!(agy.argv.windows(2).any(|w| w == ["--mode", "plan"]));
        assert!(
            !agy.argv
                .iter()
                .any(|a| a == "--dangerously-skip-permissions")
        );
        let agy_rw = plan_for(AgentKind::Antigravity, &seat, true);
        assert!(
            agy_rw
                .argv
                .windows(2)
                .any(|w| w == ["--mode", "accept-edits"])
        );
        assert!(
            agy_rw
                .argv
                .iter()
                .any(|a| a == "--dangerously-skip-permissions")
        );
        // agy is pointed at its prompt with its own `@<path>` syntax, not with
        // prose asking it to read a file. Measured on one trivial task: 17s
        // against 73s, because prose costs a tool round-trip before the model
        // has even seen its instructions. It is also the form rvpm proved.
        let agy_prompt = agy_rw
            .argv
            .iter()
            .position(|a| a == "-p")
            .map(|i| agy_rw.argv[i + 1].clone())
            .expect("agy takes its prompt with -p");
        assert!(
            agy_prompt.starts_with('@'),
            "agy must get a file reference, got {agy_prompt:?}"
        );
        assert!(
            !agy_prompt.contains("Read the file at"),
            "the prose pointer is for CLIs with no file syntax"
        );

        // opencode is the exception: `--auto` also gates reads, so withholding
        // it silently drops the seat out of the panel. Verified against the CLI
        // — a read-only judge failed with "the user rejected permission to use
        // this specific tool call" while trying to open its own prompt.
        for allow_write in [false, true] {
            assert!(
                plan_for(AgentKind::Opencode, &seat, allow_write)
                    .argv
                    .iter()
                    .any(|a| a == "--auto"),
                "opencode needs --auto even to read (allow_write = {allow_write})"
            );
        }
    }

    /// The three things about `codex exec` that were established by hand and
    /// that a rewrite would silently get wrong.
    #[test]
    fn codex_gets_extra_writable_roots_only_when_it_may_write() {
        let seat = SeatState::new("deputy-x", "a", 7);
        let roots = [PathBuf::from("/data/questions")];
        let mk = |allow_write: bool| {
            let mut i = inv(Path::new("."), Path::new("/art"), allow_write);
            i.writable = &roots;
            build_command(
                &spec(AgentKind::Codex, None),
                &seat,
                &i,
                Path::new("/art/p.md"),
            )
            .unwrap()
        };
        let want = "sandbox_workspace_write.writable_roots=[\"/data/questions\"]";
        assert!(mk(true).argv.windows(2).any(|w| w == ["-c", want]));
        assert!(!mk(false).argv.iter().any(|a| a.contains("writable_roots")));
    }

    #[tokio::test]
    async fn investigation_checks_cancellation_after_artifacts_before_spawning() {
        let tmp = tempfile::tempdir().unwrap();
        let artifacts = tmp.path().join("artifacts");
        let mut invocation = inv(tmp.path(), &artifacts, true);
        invocation.scope = InvocationScope::ConsultationInvestigation;
        invocation.unsandboxed = true;
        let mut seat = SeatState::new("chat", "fixture", 1);
        let error = invoke_leased(
            &spec(AgentKind::Command, None),
            &mut seat,
            &invocation,
            None,
            || {
                assert!(artifacts.join("t.prompt.md").exists());
                bail!("question closed during preparation")
            },
        )
        .await
        .unwrap_err();
        assert!(error.to_string().contains("question closed"));
        assert_eq!(seat.turns, 0);
        assert!(!artifacts.join("t.out").exists());
    }

    #[test]
    fn repository_write_scope_covers_editing_nodes_without_changing_other_roles() {
        let mut i = inv(Path::new("."), Path::new("/art"), true);
        for node in [
            "implement",
            "fix",
            "graft",
            "gate-fix",
            "land",
            "rebase",
            "chat",
        ] {
            i.node = node;
            assert!(repository_writer(&i), "{node}");
            i.allow_write = false;
            assert!(!repository_writer(&i), "{node}");
            i.allow_write = true;
        }
        for node in ["deputy", "judge", "review", "bump", "conduct"] {
            i.node = node;
            assert!(!repository_writer(&i), "{node}");
        }
        i.node = "chat";
        let roots = [PathBuf::from("/questions")];
        i.writable = &roots;
        assert!(
            repository_writer(&i),
            "roots alone do not select consultation"
        );
        i.scope = InvocationScope::ConsultationControl;
        assert!(!repository_writer(&i), "consultation keeps explicit grants");
        assert!(!repository_seat(&i));
        i.allow_write = false;
        i.writable = &[];
        assert!(!repository_seat(&i), "control scope needs no write roots");
        i.scope = InvocationScope::ConsultationInvestigation;
        assert!(repository_seat(&i));
        i.allow_write = true;
        assert!(
            !repository_writer(&i),
            "investigation never gains editing grants"
        );
    }

    #[test]
    fn coordination_classification_is_independent_of_write_grants() {
        let roots = [PathBuf::from("/questions")];
        for node in [
            crate::queue::CHAT_NODE,
            "implement",
            "fix",
            "graft",
            "gate-fix",
            "land",
            "rebase",
            "review",
            "judge",
            "advise",
            "deputy",
            "waiter",
            "conduct",
            "dupes",
        ] {
            for allow_write in [false, true] {
                for writable in [&[][..], &roots[..]] {
                    let mut i = inv(Path::new("."), Path::new("/art"), allow_write);
                    i.node = node;
                    i.writable = writable;
                    assert_eq!(
                        repository_seat(&i),
                        !matches!(
                            node,
                            crate::queue::CHAT_NODE | "deputy" | "conduct" | "dupes"
                        ),
                        "{node}, allow_write={allow_write}, writable={writable:?}"
                    );
                    if node == crate::queue::CHAT_NODE {
                        assert_eq!(repository_writer(&i), allow_write);
                    }
                }
            }
        }
    }

    #[test]
    fn writable_directories_reach_initial_and_resumed_adapter_commands() {
        let roots = [
            PathBuf::from("/shared/git"),
            PathBuf::from("/state/questions"),
        ];
        for kind in [AgentKind::Codex, AgentKind::Antigravity] {
            for resumed in [false, true] {
                let mut seat = SeatState::new("impl-A", "a", 7);
                if resumed {
                    seat.turns = 1;
                    seat.captured_session = Some("session-id".into());
                }
                for allow_write in [false, true] {
                    let mut i = inv(Path::new("."), Path::new("/art"), allow_write);
                    i.writable = &roots;
                    let plan = build_command(&spec(kind, None), &seat, &i, Path::new("/art/p.md"))
                        .unwrap();
                    for root in &roots {
                        if kind == AgentKind::Antigravity {
                            assert_eq!(
                                plan.argv.windows(2).any(|w| {
                                    w[0] == "--add-dir" && w[1] == root.to_str().unwrap()
                                }),
                                allow_write,
                                "resumed={resumed}: {:?}",
                                plan.argv
                            );
                        }
                        assert_eq!(
                            plan.argv
                                .iter()
                                .any(|arg| arg.contains(root.to_str().unwrap())),
                            allow_write,
                            "{kind:?}, resumed={resumed}: {:?}",
                            plan.argv
                        );
                    }
                    if kind == AgentKind::Codex && resumed {
                        let at = plan.argv.iter().position(|s| s == "resume").unwrap();
                        assert!(!plan.argv[at..].iter().any(|s| s == "-c"));
                    }
                }
            }
        }
    }

    #[test]
    fn codex_talk_turn_bypasses_sandbox_only_when_opted_in() {
        const BYPASS: &str = "--dangerously-bypass-approvals-and-sandbox";
        let mut seat = SeatState::new("talk", "a", 7);
        let build = |seat: &SeatState, allow_write: bool, unsandboxed: bool| {
            let mut i = inv(Path::new("."), Path::new("/art"), allow_write);
            i.unsandboxed = unsandboxed;
            build_command(
                &spec(AgentKind::Codex, None),
                seat,
                &i,
                Path::new("/art/p.md"),
            )
            .unwrap()
        };

        // (a) both set: bypass, and none of the sandbox flags.
        let first = build(&seat, true, true);
        assert!(first.argv.iter().any(|a| a == BYPASS));
        assert!(!first.argv.iter().any(|a| a == "--sandbox"));
        assert!(!first.argv.iter().any(|a| a.starts_with("approval_policy")));

        // (b) the flag precedes the `resume` subcommand.
        seat.captured_session = Some("thread-1".to_owned());
        seat.turns = 1;
        let resumed = build(&seat, true, true);
        let at = |p: &Plan, x: &str| p.argv.iter().position(|a| a == x).unwrap();
        assert!(at(&resumed, BYPASS) < at(&resumed, "resume"));

        // (c) without write access the read-only sandbox stays.
        let ro = build(&seat, false, true);
        assert!(!ro.argv.iter().any(|a| a == BYPASS));
        assert!(ro.argv.windows(2).any(|w| w == ["--sandbox", "read-only"]));
    }

    #[test]
    fn codex_is_sandboxed_reads_stdin_and_puts_resume_last() {
        let mut seat = SeatState::new("judge-1", "a", 7);

        // 1. Read-only is enforced by the CLI, not by the prompt - the only
        //    roster member for which that is true - and nothing ever asks for
        //    the bypass.
        let ro = plan_for(AgentKind::Codex, &seat, false);
        assert!(ro.argv.windows(2).any(|w| w == ["--sandbox", "read-only"]));
        let rw = plan_for(AgentKind::Codex, &seat, true);
        assert!(
            rw.argv
                .windows(2)
                .any(|w| w == ["--sandbox", "workspace-write"])
        );
        // `unsandboxed` without a talk turn's opt-in never reaches an unattended
        // seat, and on its own it never lifts a read-only seat either.
        let mut ro_flagged = inv(Path::new("."), Path::new("/art"), false);
        ro_flagged.unsandboxed = true;
        let ro_flagged = build_command(
            &spec(AgentKind::Codex, None),
            &seat,
            &ro_flagged,
            Path::new("/art/p.md"),
        )
        .unwrap();
        assert!(
            ro_flagged
                .argv
                .windows(2)
                .any(|w| w == ["--sandbox", "read-only"])
        );
        for p in [&ro, &rw, &ro_flagged] {
            assert!(
                !p.argv
                    .iter()
                    .any(|a| a == "--dangerously-bypass-approvals-and-sandbox"),
                "the bypass defeats the only enforced read-only mode we have"
            );
            // Nobody is watching to approve anything.
            assert!(
                p.argv
                    .windows(2)
                    .any(|w| w == ["-c", "approval_policy=\"never\""]),
                "an unattended seat that asks for approval blocks until timeout"
            );
        }

        // 2. The prompt arrives on stdin, and `-` is what makes codex read it.
        assert_eq!(ro.stdin.as_deref(), Some("do the thing"));
        assert_eq!(
            ro.argv.last().map(String::as_str),
            Some("-"),
            "without the `-` argument codex waits for a prompt it never gets"
        );

        // 3. `resume` is a subcommand and rejects the options above when they
        //    follow it, so it has to be emitted after all of them - and only
        //    once the CLI has reported a thread id.
        seat.turns = 1;
        assert!(!has_session(AgentKind::Codex, &seat, true));
        assert!(
            !plan_for(AgentKind::Codex, &seat, true)
                .argv
                .iter()
                .any(|a| a == "resume")
        );
        seat.captured_session = Some("01a07440-4545-7492-85c1-024e3259a90a".to_owned());
        let resumed = plan_for(AgentKind::Codex, &seat, true);
        let at = resumed
            .argv
            .iter()
            .position(|a| a == "resume")
            .expect("resumes by subcommand");
        assert_eq!(resumed.argv[at + 1], "01a07440-4545-7492-85c1-024e3259a90a");
        assert!(
            resumed.argv[..at].iter().any(|a| a == "--sandbox"),
            "every option precedes the subcommand"
        );
        assert_eq!(resumed.argv.last().map(String::as_str), Some("-"));
    }

    /// The three things about `omp -p --mode=json` that were established by
    /// hand against omp 18.1.19 and that a rewrite would silently get wrong.
    #[test]
    fn omp_reads_stdin_auto_approves_and_resumes_by_id() {
        let mut seat = SeatState::new("review-1", "a", 7);

        // 1. Print mode plus JSON, and the prompt on stdin: a judging prompt
        //    carrying three patches is past the Windows argv cap, so argv
        //    delivery is not an option for every node.
        let first = plan_for(AgentKind::Omp, &seat, false);
        assert!(first.argv.iter().any(|a| a == "-p"));
        assert!(first.argv.iter().any(|a| a == "--mode=json"));
        assert_eq!(first.stdin.as_deref(), Some("do the thing"));
        assert!(
            !first.argv.iter().any(|a| a == "do the thing"),
            "the prompt reached argv, where Windows caps it"
        );

        // 2. `--auto-approve` is required (an unattended seat that stops to ask
        //    blocks until its node timeout kills it), and it is the *only*
        //    permission flag: omp has no read-only mode, so the bypass flag
        //    that would throw away codex's one enforced guarantee must never
        //    appear here either.
        for allow_write in [false, true] {
            let p = plan_for(AgentKind::Omp, &seat, allow_write);
            assert!(
                p.argv.iter().any(|a| a == "--auto-approve"),
                "omp needs --auto-approve even to read (allow_write = {allow_write})"
            );
            assert!(
                !p.argv
                    .iter()
                    .any(|a| a == "--dangerously-bypass-approvals-and-sandbox"),
                "nothing ever asks for the bypass"
            );
        }

        // 3. The id omp reports is the only resume token - magi cannot mint it
        //    up front, so a seat resumes only once a turn has reported one.
        seat.turns = 1;
        assert!(!has_session(AgentKind::Omp, &seat, true));
        assert!(
            !plan_for(AgentKind::Omp, &seat, true)
                .argv
                .iter()
                .any(|a| a == "--resume")
        );
        seat.captured_session = Some("01a09fe9-4e31-7226-85b3-fda6f46689d5".to_owned());
        let resumed = plan_for(AgentKind::Omp, &seat, true);
        assert!(
            resumed
                .argv
                .windows(2)
                .any(|w| w == ["--resume", "01a09fe9-4e31-7226-85b3-fda6f46689d5"]),
            "a captured id is what makes the next turn a resume"
        );
        // `--continue` opens a *new* session instead of the stored one, which
        // would silently drop the seat's memory.
        assert!(!resumed.argv.iter().any(|a| a == "--continue"));
        // stdin still carries the prompt on a resumed turn.
        assert_eq!(resumed.stdin.as_deref(), Some("do the thing"));
    }

    /// The extraction trap that cost three complete reviews when it was done by
    /// hand: a turn that ends on a tool call emits **no** `agent_end` line, so
    /// keying on `agent_end` finds nothing and the seat reads as one that
    /// produced no answer at all.
    #[test]
    fn omp_takes_the_answer_without_an_agent_end_line() {
        let stream = concat!(
            r#"{"type":"session","version":3,"id":"01a09fe9-4e31-7226-85b3-fda6f46689d5","cwd":"C:\\w"}"#,
            "\n",
            r#"{"type":"agent_start"}"#,
            "\n",
            r#"{"type":"turn_start"}"#,
            "\n",
            r#"{"type":"message_update","assistantMessageEvent":{"type":"text_delta","contentIndex":1,"delta":"."}}"#,
            "\n",
            r#"{"type":"message_end","message":{"role":"assistant","content":[{"type":"thinking","thinking":"checking"},{"type":"text","text":"."}]}}"#,
            "\n",
            r#"{"type":"turn_end","message":{"role":"assistant","content":[{"type":"thinking","thinking":"done"},{"type":"text","text":"{\"vote\":\"approve\"}"}]}}"#,
            "\n",
        );
        let out = extract(AgentKind::Omp, stream);
        assert_eq!(
            out.text, "{\"vote\":\"approve\"}",
            "the last assistant text block is the answer even with no agent_end"
        );
        assert_eq!(
            out.session.as_deref(),
            Some("01a09fe9-4e31-7226-85b3-fda6f46689d5")
        );
    }

    /// A stream that *does* carry `agent_end` walks the whole thread, and the
    /// last non-empty assistant text still wins over the tool-loop narration
    /// that came before it.
    #[test]
    fn omp_walks_agent_end_and_ignores_tool_loop_narration() {
        let stream = concat!(
            r#"{"type":"session","version":3,"id":"s1"}"#,
            "\n",
            "{\"type\":\"agent_end\",\"messages\":[{\"role\":\"user\",\"content\":[{\"type\":\"text\",\"text\":\"review this\"}]},{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"Looking at the diff…\"}]},{\"role\":\"assistant\",\"content\":[{\"type\":\"thinking\",\"thinking\":\"…\"},{\"type\":\"text\",\"text\":\"## 判定\\n\\n問題ありません。\"}]}]}",
            "\n",
        );
        let out = extract(AgentKind::Omp, stream);
        assert_eq!(
            out.text, "## 判定\n\n問題ありません。",
            "the narration is not the answer, and non-ASCII survives intact"
        );
        assert_eq!(out.session.as_deref(), Some("s1"));
    }

    /// A line that is not JSON - a CLI warning, a half-written line - is
    /// skipped rather than becoming the answer.
    #[test]
    fn omp_skips_non_json_lines() {
        let stream = concat!(
            "Warning: some omp notice\n",
            r#"{"type":"session","version":3,"id":"s2"}"#,
            "\n",
            r#"{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"the answer"}]}}"#,
            "\n",
            "trailing junk",
            "\n",
        );
        let out = extract(AgentKind::Omp, stream);
        assert_eq!(out.text, "the answer");
        assert_eq!(out.session.as_deref(), Some("s2"));
    }

    /// A real `codex exec --json` stream, tracing prefix included.
    #[test]
    fn codex_takes_the_last_agent_message_and_the_thread_id() {
        let stream = concat!(
            "2026-09-06T01:05:49.394445Z ERROR codex_models_manager: failed to load models cache\n",
            r#"{"type":"thread.started","thread_id":"01a07440-4545-7492-85c1-024e3259a90a"}"#,
            "\n",
            r#"{"type":"turn.started"}"#,
            "\n",
            r#"{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"Looking into it."}}"#,
            "\n",
            r#"{"type":"item.completed","item":{"id":"item_1","type":"command_execution","text":"cargo test"}}"#,
            "\n",
            r#"{"type":"item.completed","item":{"id":"item_2","type":"agent_message","text":"{\"verdict\": \"ok\"}"}}"#,
            "\n",
            r#"{"type":"turn.completed","usage":{"input_tokens":17137}}"#,
            "\n",
        );
        let out = extract(AgentKind::Codex, stream);
        assert_eq!(
            out.text, "{\"verdict\": \"ok\"}",
            "the last agent message is the answer; earlier ones narrate"
        );
        assert_eq!(
            out.session.as_deref(),
            Some("01a07440-4545-7492-85c1-024e3259a90a")
        );
        assert_eq!(out.status.as_deref(), Some("success"));

        let failed = concat!(
            r#"{"type":"thread.started","thread_id":"t1"}"#,
            "\n",
            r#"{"type":"turn.failed","error":{"message":"nope"}}"#,
            "\n",
        );
        assert_eq!(
            extract(AgentKind::Codex, failed).status.as_deref(),
            Some("error")
        );
    }

    #[test]
    fn captured_sessions_resume_only_once_reported() {
        let mut seat = SeatState::new("impl-A", "a", 7);
        seat.turns = 1;
        for kind in [AgentKind::Opencode, AgentKind::Antigravity] {
            assert!(!has_session(kind, &seat, true));
            let p = plan_for(kind, &seat, true);
            assert!(!p.argv.iter().any(|a| a == "-s" || a == "--conversation"));
        }

        seat.captured_session = Some("sid".to_owned());
        assert!(has_session(AgentKind::Opencode, &seat, true));
        assert!(
            plan_for(AgentKind::Opencode, &seat, true)
                .argv
                .windows(2)
                .any(|w| w == ["-s", "sid"])
        );
        assert!(
            plan_for(AgentKind::Antigravity, &seat, true)
                .argv
                .windows(2)
                .any(|w| w == ["--conversation", "sid"])
        );
    }

    #[test]
    fn sessions_disabled_never_resumes() {
        let mut seat = SeatState::new("impl-A", "a", 7);
        seat.turns = 3;
        seat.captured_session = Some("sid".to_owned());
        for kind in [
            AgentKind::Claude,
            AgentKind::Opencode,
            AgentKind::Antigravity,
        ] {
            assert!(!has_session(kind, &seat, false));
        }
    }

    #[test]
    fn long_prompts_never_reach_argv_for_file_delivery_clis() {
        let seat = SeatState::new("judge-1", "a", 7);
        for kind in [AgentKind::Opencode, AgentKind::Antigravity] {
            let p = plan_for(kind, &seat, false);
            assert!(
                p.argv.iter().all(|a| a != "do the thing"),
                "{kind:?} put the prompt on the command line"
            );
            assert!(p.argv.iter().any(|a| a.contains("/art/p.md")));
        }
        // agy has no text stdin, so its `-p` must always carry something.
        let p = plan_for(AgentKind::Antigravity, &seat, false);
        let at = p.argv.iter().position(|a| a == "-p").unwrap();
        assert!(p.argv.get(at + 1).is_some_and(|v| v.contains("p.md")));
        assert!(p.stdin.is_none());
    }

    #[test]
    fn agy_print_timeout_tracks_the_node_budget() {
        let seat = SeatState::new("impl-A", "a", 7);
        let p = build_command(
            &spec(AgentKind::Antigravity, None),
            &seat,
            &Invocation {
                scope: crate::agent::InvocationScope::Repository,
                coordination: None,
                fetched_base: None,
                cwd: Path::new("."),
                prompt: "p",
                timeout: Duration::from_secs(3600),
                allow_write: true,
                unsandboxed: false,
                sessions: true,
                artifacts: Path::new("/art"),
                stem: "t",
                run: "test-run",
                node: "test",
                cache_dir: None,
                attachments: &[],
                writable: &[],
            },
            Path::new("/art/p.md"),
        )
        .unwrap();
        assert!(p.argv.windows(2).any(|w| w == ["--print-timeout", "3600s"]));
    }

    /// `--add-dir` is what lets antigravity open a file outside the
    /// worktree at all. Today that only happens when the delivery mode is
    /// already `File`, but an attachment can arrive on a seat whose delivery
    /// is `Stdin` or `Argv` (an explicit `prompt_delivery` override), and the
    /// image still lives outside `cwd` - so the flag has to widen for that
    /// reason too, independent of how the prompt itself is delivered.
    #[test]
    fn attachments_widen_antigravitys_add_dir_even_off_file_delivery() {
        let mut s = spec(AgentKind::Antigravity, None);
        s.prompt_delivery = Some(Delivery::Argv);
        let seat = SeatState::new("talk", "a", 7);
        let atts = [PathBuf::from("/art/attachments/abc.png")];

        let without = build_command(
            &s,
            &seat,
            &Invocation {
                attachments: &[],
                writable: &[],
                ..inv(Path::new("."), Path::new("/art"), true)
            },
            Path::new("/art/p.md"),
        )
        .unwrap();
        assert!(
            !without.argv.iter().any(|a| a == "--add-dir"),
            "no attachment, no reason to widen the sandbox: {without:?}"
        );

        let with = build_command(
            &s,
            &seat,
            &Invocation {
                attachments: &atts,
                writable: &[],
                ..inv(Path::new("."), Path::new("/art"), true)
            },
            Path::new("/art/p.md"),
        )
        .unwrap();
        assert!(
            with.argv.windows(2).any(|w| w == ["--add-dir", "/art"]),
            "an attachment outside cwd must widen the sandbox even off File delivery: {with:?}"
        );
    }

    /// A chat derived from another one (`chat::derived_background`) can pass
    /// `turn` attachment paths that live under the *source* conversation's
    /// own artifacts dir, not this invocation's `artifacts`. A single
    /// `--add-dir` for `inv.artifacts` alone would leave those unreadable, so
    /// each attachment directory outside it must get its own grant.
    #[test]
    fn an_inherited_attachment_outside_this_conversations_artifacts_dir_gets_its_own_add_dir() {
        let seat = SeatState::new("plan", "a", 7);
        let atts = [
            PathBuf::from("/art/attachments/own.png"),
            PathBuf::from("/other-chat/attachments/inherited.png"),
        ];

        let p = build_command(
            &spec(AgentKind::Antigravity, None),
            &seat,
            &Invocation {
                attachments: &atts,
                writable: &[],
                ..inv(Path::new("."), Path::new("/art"), true)
            },
            Path::new("/art/p.md"),
        )
        .unwrap();

        assert!(
            p.argv.windows(2).any(|w| w == ["--add-dir", "/art"]),
            "this conversation's own artifacts dir must still be granted: {p:?}"
        );
        assert!(
            p.argv
                .windows(2)
                .any(|w| w == ["--add-dir", "/other-chat/attachments"]),
            "the inherited attachment's own directory must be granted too: {p:?}"
        );
    }

    #[test]
    fn command_agents_get_placeholders_substituted() {
        let seat = SeatState::new("impl-A", "a", 7);
        let p = plan_for(AgentKind::Command, &seat, true);
        assert_eq!(p.argv[0], "echo");
        assert_eq!(p.argv[1], "impl-A");
        assert_eq!(p.stdin.as_deref(), Some("do the thing"));
    }

    #[test]
    fn claude_rate_limit_is_detected_and_reset_read_when_present() {
        // The exact shape observed in the wild (run 20260831-031005-ae94).
        let stdout = r#"{"is_error": true, "terminal_reason": "api_error",
                        "result": "You've hit your session limit · resets 4:50am (Asia/Tokyo)",
                        "session_id": "b8e928f1-754e-4bd3-86c5-0567763654e3"}"#;
        let out = extract(AgentKind::Claude, stdout);
        let quota = out.quota.as_ref().expect("rate limit must be detected");
        assert_eq!(
            quota.reset.as_deref(),
            Some("4:50am (Asia/Tokyo)"),
            "reset time read from the body"
        );
    }

    #[test]
    fn claude_rate_limit_without_a_readable_reset_is_still_detected() {
        let out = extract(
            AgentKind::Claude,
            r#"{"is_error":true,"result":"session limit reached"}"#,
        );
        let quota = out.quota.expect("rate limit detected without a reset");
        assert!(quota.reset.is_none(), "unknown reset is kept as unknown");
    }

    #[test]
    fn claude_structured_usage_limit_requires_the_error_and_exact_pair() {
        for kind in [AgentKind::Claude, AgentKind::Command] {
            for reply in [
                r#"{"is_error":true,"api_error_status":429,"api_error":"usage_limit_reached"}"#,
                r#"{"is_error":true,"api_error_status":"429","api_error":"usage_limit_reached"}"#,
            ] {
                assert!(extract(kind, reply).quota.is_some(), "{reply}");
            }
            for reply in [
                r#"{"is_error":false,"api_error_status":429,"api_error":"usage_limit_reached"}"#,
                r#"{"is_error":true,"api_error_status":500,"api_error":"usage_limit_reached"}"#,
                r#"{"is_error":true,"api_error_status":429,"api_error":"other_error"}"#,
                r#"{"is_error":false,"api_error_status":"429","api_error":"usage_limit_reached"}"#,
                r#"{"is_error":true,"api_error_status":"500","api_error":"usage_limit_reached"}"#,
                r#"{"is_error":true,"api_error_status":"429","api_error":"other_error"}"#,
                r#"{"is_error":true,"api_error":"usage_limit_reached"}"#,
            ] {
                assert!(extract(kind, reply).quota.is_none(), "{reply}");
            }
        }
    }

    #[test]
    fn ordinary_failures_are_never_quota() {
        // A normal failed claude call (is_error with a different message).
        let claude_fail = extract(
            AgentKind::Claude,
            r#"{"is_error":true,"result":"account does not exist"}"#,
        );
        assert!(claude_fail.quota.is_none());

        // A command agent that exits 1 with plain text.
        let cmd_fail = extract(AgentKind::Command, "boom");
        assert!(cmd_fail.quota.is_none());

        // A successful call is not quota even if it mentions the phrase.
        let success = extract(
            AgentKind::Command,
            r#"{"is_error":false,"result":"session limit is fine"}"#,
        );
        assert!(success.quota.is_none());
    }

    /// Minimal, anonymised shape of run 20260912-214939-b3bb's
    /// artifacts/impl-A.out: `command_execution` items reporting a paired
    /// test run as `1 passed, 1 failed` twice, while the final
    /// `agent_message` nevertheless claimed the target passed. The point of
    /// `CommandEvidence` is that this claim and the CLI's own structured
    /// record of what actually ran are now two separate things a caller can
    /// compare, rather than the prose being the only account available.
    #[test]
    fn codex_command_execution_events_are_captured_alongside_the_final_message() {
        let stream = concat!(
            r#"{"type":"thread.started","thread_id":"t1"}"#,
            "\n",
            r#"{"type":"item.completed","item":{"id":"item49","type":"command_execution","command":["bash","-lc","cargo test --test graph_cached_gate"],"exit_code":1,"aggregated_output":"test result: 1 passed; 1 failed"}}"#,
            "\n",
            r#"{"type":"item.completed","item":{"id":"item52","type":"command_execution","command":["bash","-lc","cargo test --test graph_cached_gate a_single_test"],"exit_code":0,"aggregated_output":"test result: 1 passed; 0 failed"}}"#,
            "\n",
            r#"{"type":"item.completed","item":{"id":"item99","type":"agent_message","text":"Both tests in the target pass."}}"#,
            "\n",
            r#"{"type":"turn.completed"}"#,
            "\n",
        );
        let out = extract(AgentKind::Codex, stream);
        assert_eq!(out.text, "Both tests in the target pass.");
        assert_eq!(out.commands.len(), 2, "{:?}", out.commands);

        let paired = &out.commands[0];
        assert_eq!(paired.id, "item49");
        assert_eq!(paired.exit_code, Some(1));
        assert!(paired.description.contains("graph_cached_gate"));
        assert!(paired.result_summary.contains("1 failed"));

        let solo = &out.commands[1];
        assert_eq!(solo.exit_code, Some(0));

        // The structured evidence disagrees with the final prose - exactly
        // what a caller must be able to see instead of trusting the message
        // alone: the full target never passed in one command.
        assert!(
            out.commands
                .iter()
                .any(|c| c.exit_code != Some(0) && c.description.contains("graph_cached_gate")),
            "a failed run of the actual target must still be visible: {:?}",
            out.commands
        );
    }

    #[test]
    fn command_agent_can_carry_the_claude_quota_shape() {
        let out = extract(
            AgentKind::Command,
            r#"{"is_error":true,"result":"You've hit your session limit · resets 1:00am (UTC)"}"#,
        );
        assert!(
            out.quota.is_some(),
            "a wrapper emitting the claude shape counts as quota"
        );
    }

    #[test]
    fn claude_json_result_is_extracted() {
        let out = extract(
            AgentKind::Claude,
            r#"{"result":"all done","session_id":"abc","is_error":false}"#,
        );
        assert_eq!(out.text, "all done");
        assert_eq!(out.session.as_deref(), Some("abc"));
        assert_eq!(out.status.as_deref(), Some("success"));
    }

    /// The shape behind fix-2 in run 20260912-114326-d3b8, minimal and
    /// anonymised: a Claude CLI turn that ended `subtype: success`,
    /// `is_error: false`, `terminal_reason: completed`, `stop_reason:
    /// end_turn` — every signal this crate reads as a clean CLI turn — while
    /// `result` is a progress update, not the report the fixer node needed,
    /// and no `FixReport` JSON is anywhere in it.
    ///
    /// `AgentOutput::usable()` (a CLI fact: exit 0, not timed out, non-empty
    /// text) must stay true here — that is the honest reading of what the
    /// CLI reported — while `verdict::extract_json` on the same text must
    /// fail. Conflating the two is exactly the bug this fixture reproduces:
    /// `run.json` recorded `fix.failed = "unparsable fix report: the reply
    /// contained no JSON object"` and moved straight to the next review round
    /// with no report ever recovered from that seat.
    #[test]
    fn a_clean_cli_turn_is_not_the_same_fact_as_the_nodes_own_work_being_done() {
        let stdout = r#"{"type":"result","subtype":"success","is_error":false,"terminal_reason":"completed","stop_reason":"end_turn","result":"I'll pause here until the `cargo make check` background run reports back.","session_id":"11111111-1111-1111-1111-111111111111"}"#;
        let out = extract(AgentKind::Claude, stdout);
        assert_eq!(out.status.as_deref(), Some("success"));
        assert!(out.quota.is_none());
        assert!(!out.text.trim().is_empty());

        let agent_out = AgentOutput {
            text: out.text.clone(),
            exit_code: Some(0),
            timed_out: false,
            duration_ms: 500,
            artifacts: Vec::new(),
            quota: out.quota,
            dropped: out.dropped,
            commands: out.commands,
            context_tokens: None,
        };
        assert!(
            agent_out.usable(),
            "the CLI turn itself ended cleanly and must read as usable"
        );
        assert!(
            crate::verdict::extract_json::<crate::verdict::FixReport>(&agent_out.text).is_err(),
            "a clean CLI turn is not proof the node's own report ever arrived"
        );
    }

    #[test]
    fn opencode_event_stream_is_concatenated() {
        let stream = concat!(
            r#"{"type":"step_start","sessionID":"ses_1","part":{"type":"step-start"}}"#,
            "\n",
            r#"{"type":"text","sessionID":"ses_1","part":{"type":"text","text":"first"}}"#,
            "\n",
            "garbage line\n",
            r#"{"type":"text","sessionID":"ses_1","part":{"type":"text","text":"second"}}"#,
            "\n"
        );
        let out = extract(AgentKind::Opencode, stream);
        assert_eq!(out.text, "first\nsecond");
        assert_eq!(out.session.as_deref(), Some("ses_1"));
    }

    #[test]
    fn agy_json_survives_a_leading_warning_line() {
        let stdout = concat!(
            "warning: --mode plan has no effect while slash commands are disabled.\n",
            r#"{"conversation_id":"eaf2d00a","status":"SUCCESS","response":"persimmon\n"}"#,
            "\n"
        );
        let out = extract(AgentKind::Antigravity, stdout);
        assert_eq!(out.text, "persimmon");
        assert_eq!(out.session.as_deref(), Some("eaf2d00a"));
        assert_eq!(out.status.as_deref(), Some("SUCCESS"));
    }

    /// Run 26c7's candidate B, verbatim from `artifacts/impl-B.out`.
    ///
    /// The seat read as an empty candidate. It was seven minutes of work and
    /// 14,267 output tokens, billed, that the CLI then declined to hand over.
    /// Five such candidates are why `agy` reads as 0 wins in 4 entries, and
    /// that number has twice been used to argue the seat out of the roster.
    const AGY_DROPPED: &str = concat!(
        r#"{"conversation_id":"36743d06-c0b3-4b79-9fa2-23869289d7b6","status":"ERROR","#,
        r#""response":"","error":"the connection to the agent was interrupted before "#,
        r#"the response finished: subscriber fell behind updates, stalled for 5s","#,
        r#""duration_seconds":431.1941803,"num_turns":1,"usage":{"input_tokens":260113,"#,
        r#""output_tokens":14267,"thinking_tokens":9695,"cache_read_tokens":2200925,"#,
        r#""total_tokens":274380}}"#
    );

    /// Run 1798's review seat, from `review-1-1.out` / `.err`.
    const AGY_QUOTA_OUT: &str = concat!(
        r#"{"conversation_id":"323c3b5b-0000","status":"ERROR","response":"","#,
        r#""error":"Individual quota reached. Please upgrade your subscription to "#,
        r#"increase your limits. Resets in 1h2m49s.","duration_seconds":265.9,"#,
        r#""num_turns":2,"usage":{"input_tokens":1000,"output_tokens":50}}"#
    );
    const AGY_QUOTA_ERR: &str = concat!(
        "error: Individual quota reached. Resets in 1h2m49s.\n",
        r#"AGY_ERROR: {"short_error":"RESOURCE_EXHAUSTED (code 429): Individual quota "#,
        r#"reached.","status":"RESOURCE_EXHAUSTED","error_code":429,"code_kind":"http","#,
        r#""retryable":true}"#,
        "\n"
    );

    #[test]
    fn agy_out_of_quota_is_a_quota_with_the_reset_hint() {
        let both = agy_quota(AGY_QUOTA_OUT, AGY_QUOTA_ERR).expect("both streams");
        assert_eq!(both.reset.as_deref(), Some("in 1h2m49s"));
        let stdout_only = agy_quota(AGY_QUOTA_OUT, "").expect("stdout alone");
        assert_eq!(stdout_only.reset.as_deref(), Some("in 1h2m49s"));
        // stderr alone: the hint is not in `short_error`, so none is carried.
        let stderr_only = agy_quota("not json", AGY_QUOTA_ERR).expect("stderr alone");
        assert!(stderr_only.reset.is_none());
    }

    #[test]
    fn ordinary_agy_failures_are_not_a_quota() {
        assert!(agy_quota(AGY_DROPPED, "").is_none());
        assert!(agy_quota(r#"{"status":"ERROR","error":"boom"}"#, "").is_none());
        assert!(
            agy_quota(
                "",
                r#"AGY_ERROR: {"status":"RESOURCE_EXHAUSTED","error_code":500}"#
            )
            .is_none()
        );
        assert!(
            agy_quota(
                "",
                r#"AGY_ERROR: {"status":"UNAVAILABLE","error_code":429}"#
            )
            .is_none()
        );
        assert!(agy_quota(r#"{"status":"SUCCESS","response":"ok"}"#, "").is_none());
    }

    #[test]
    fn a_cli_that_hangs_up_on_billed_work_is_not_an_agent_that_produced_nothing() {
        let out = extract(AgentKind::Antigravity, AGY_DROPPED);
        let dropped = out.dropped.expect("recognised as undelivered work");
        assert_eq!(dropped.output_tokens, 14267);
        assert!(
            dropped.why.contains("subscriber fell behind"),
            "the CLI's own words are kept for the record: {}",
            dropped.why
        );
        // And the conversation is still there to resume, which is the whole
        // reason this is worth re-asking where a quota is not.
        assert_eq!(
            out.session.as_deref(),
            Some("36743d06-c0b3-4b79-9fa2-23869289d7b6")
        );
        assert!(out.quota.is_none(), "a dropped stream is not a rate limit");
    }

    #[test]
    fn an_error_with_nothing_produced_stays_an_ordinary_failure() {
        // No usage at all: the agent never got going, so there is nothing in
        // the conversation to resume and nothing was billed. Treating this as
        // undelivered work would buy a second call for no reason.
        let bare = r#"{"conversation_id":"c1","status":"ERROR","response":"","error":"boom"}"#;
        assert!(extract(AgentKind::Antigravity, bare).dropped.is_none());

        // Produced tokens, but it did answer - so there is something to read
        // and the status is not our business.
        let answered = concat!(
            r#"{"conversation_id":"c2","status":"ERROR","response":"here it is","#,
            r#""usage":{"output_tokens":10}}"#
        );
        assert!(extract(AgentKind::Antigravity, answered).dropped.is_none());

        // A success is a success.
        let ok = concat!(
            r#"{"conversation_id":"c3","status":"SUCCESS","response":"done","#,
            r#""usage":{"output_tokens":10}}"#
        );
        assert!(extract(AgentKind::Antigravity, ok).dropped.is_none());
    }

    #[test]
    fn an_undelivered_output_is_not_usable_but_is_worth_asking_again() {
        let out = AgentOutput {
            text: String::new(),
            exit_code: Some(1),
            timed_out: false,
            duration_ms: 431_194,
            artifacts: Vec::new(),
            quota: None,
            dropped: Some(Dropped {
                why: "subscriber fell behind updates".to_owned(),
                output_tokens: 14267,
            }),
            commands: Vec::new(),
            context_tokens: None,
        };
        assert!(!out.usable());
        assert!(out.work_undelivered());
        // The distinction the retry policy rests on: a quota fails the same way
        // until it resets, an abandoned conversation can be picked up.
        assert!(!out.quota_exhausted());
    }

    #[test]
    fn non_json_stdout_falls_back_to_raw_text() {
        let out = extract(AgentKind::Antigravity, "plain answer\n");
        assert_eq!(out.text, "plain answer");
        assert!(out.session.is_none());
    }

    #[tokio::test]
    async fn command_agent_round_trip_writes_artifacts() {
        let dir = tempfile::tempdir().unwrap();
        let art = dir.path().join("artifacts");
        let mut seat = SeatState::new("impl-A", "a", 7);
        let s = command_helper("reply");
        let out = invoke(
            &s,
            &mut seat,
            &Invocation {
                scope: crate::agent::InvocationScope::Repository,
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_secs(30),
                allow_write: true,
                unsandboxed: false,
                sessions: true,
                artifacts: &art,
                stem: "impl-A",
                run: "test-run",
                node: "test",
                cache_dir: None,
                attachments: &[],
                writable: &[],
            },
        )
        .await
        .unwrap();
        assert!(out.usable(), "{out:?}");
        assert!(out.text.contains("hello impl-A"), "{}", out.text);
        assert_eq!(seat.turns, 1);
        assert!(art.join("impl-A.prompt.md").is_file());
        assert!(art.join("impl-A.out").is_file());
    }

    #[tokio::test]
    async fn the_invocation_cache_dir_reaches_the_seat_as_cargo_target_dir() {
        crate::run::pin_test_home();
        // The whole point of threading the cache path through `Invocation`:
        // the compile the agent pays for lands in the directory `verify` reads
        // back out of its rendered commands, so one cache has one prune.
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("magi-cache");
        let mut seat = SeatState::new("impl-A", "a", 7);
        let s = command_helper("cache");
        let out = invoke(
            &s,
            &mut seat,
            &Invocation {
                scope: crate::agent::InvocationScope::Repository,
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_secs(30),
                allow_write: true,
                unsandboxed: false,
                sessions: true,
                artifacts: &dir.path().join("artifacts"),
                stem: "cache",
                run: "test-run",
                node: "test",
                cache_dir: Some(&cache),
                attachments: &[],
                writable: &[],
            },
        )
        .await
        .unwrap();
        assert!(out.usable(), "{out:?}");
        assert!(
            out.text.contains(cache.to_string_lossy().as_ref()),
            "the seat must see CARGO_TARGET_DIR = the shared cache"
        );
    }

    #[tokio::test]
    async fn cache_dir_none_strips_a_cargo_target_dir_inherited_from_this_process() {
        // `Command` inherits the parent's environment by default, so
        // `cache_dir: None` alone is not the same as a seat never seeing
        // `CARGO_TARGET_DIR` - it also has to be true when *this* process
        // (standing in for the real magi process, which normally does have
        // one set, from its own `[verify]` config) already has the variable
        // set. Simulating that is the only way to exercise the inheritance
        // path at all.
        let previous = std::env::var("CARGO_TARGET_DIR").ok();
        // SAFETY: this crate's tests run single-threaded
        // (`RUST_TEST_THREADS=1`); see
        // `updater::tests::env_kill_switch_semantics` for the same
        // reasoning applied to another process-global env var.
        unsafe {
            std::env::set_var("CARGO_TARGET_DIR", "/should/never/reach/a/read-only/seat");
        }
        let dir = tempfile::tempdir().unwrap();
        let mut seat = SeatState::new("review-1", "a", 7);
        let s = command_helper("no-cache");
        let result = invoke(
            &s,
            &mut seat,
            &Invocation {
                scope: crate::agent::InvocationScope::Repository,
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_secs(30),
                allow_write: false,
                unsandboxed: false,
                sessions: true,
                artifacts: &dir.path().join("artifacts"),
                stem: "no-cache",
                run: "test-run",
                node: "test",
                cache_dir: None,
                attachments: &[],
                writable: &[],
            },
        )
        .await;
        // Restored before any assertion that could panic, so a failure here
        // never leaks a bogus `CARGO_TARGET_DIR` into whichever test runs
        // next in this same process.
        // SAFETY: see above.
        unsafe {
            match &previous {
                Some(v) => std::env::set_var("CARGO_TARGET_DIR", v),
                None => std::env::remove_var("CARGO_TARGET_DIR"),
            }
        }
        let out = result.unwrap();
        assert!(out.usable(), "{out:?}");
        assert!(
            out.text.contains("ABSENT"),
            "a read-only seat must never inherit the process's own CARGO_TARGET_DIR: {}",
            out.text
        );
    }

    #[tokio::test]
    async fn a_prompt_larger_than_the_pipe_buffer_does_not_deadlock() {
        let dir = tempfile::tempdir().unwrap();
        let mut seat = SeatState::new("impl-A", "a", 7);
        // The helper never reads stdin, so an inline write_all would block once
        // the OS pipe buffer filled — long before the process could be waited on.
        let s = command_helper("ignore-stdin");
        let big = "x".repeat(1_000_000);
        let out = invoke(
            &s,
            &mut seat,
            &Invocation {
                scope: crate::agent::InvocationScope::Repository,
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: &big,
                timeout: Duration::from_secs(60),
                allow_write: true,
                unsandboxed: false,
                sessions: true,
                artifacts: &dir.path().join("artifacts"),
                stem: "big",
                run: "test-run",
                node: "test",
                cache_dir: None,
                attachments: &[],
                writable: &[],
            },
        )
        .await
        .unwrap();
        assert!(out.usable(), "{out:?}");
        assert!(out.text.contains("done"), "{}", out.text);
    }

    #[tokio::test]
    async fn agent_turn_cap_limits_a_direct_invocation() {
        let dir = tempfile::tempdir().unwrap();
        let artifacts = dir.path().join("artifacts");
        let mut seat = SeatState::new("impl-A", "helper", 7);
        let mut s = command_helper("sleep");
        s.max_turn_secs = Some(2);
        let out = invoke(&s, &mut seat, &inv(dir.path(), &artifacts, false))
            .await
            .unwrap();
        assert!(out.timed_out, "{out:?}");
        // The caller supplies 900 seconds and the helper sleeps for 30:
        // neither may override the short per-agent cap.
        assert!(out.duration_ms < 20_000, "{out:?}");
    }

    #[tokio::test]
    async fn timeout_is_reported_not_hung() {
        let dir = tempfile::tempdir().unwrap();
        let mut seat = SeatState::new("impl-A", "a", 7);
        let s = command_helper("sleep");
        let out = invoke(
            &s,
            &mut seat,
            &Invocation {
                scope: crate::agent::InvocationScope::Repository,
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_millis(300),
                allow_write: true,
                unsandboxed: false,
                sessions: true,
                artifacts: &dir.path().join("artifacts"),
                stem: "slow",
                run: "test-run",
                node: "test",
                cache_dir: None,
                attachments: &[],
                writable: &[],
            },
        )
        .await
        .unwrap();
        assert!(out.timed_out);
        assert!(!out.usable());
    }

    #[tokio::test]
    async fn a_timeout_keeps_what_the_agent_had_already_printed() {
        // The old implementation cancelled `wait_with_output`, which dropped
        // the buffers it owned, so `<stem>.out` was written empty on every
        // timeout. "It printed nothing" and "we discarded what it printed"
        // looked identical on disk — and one real hour-long stall was
        // diagnosed wrongly twice because of it.
        let dir = tempfile::tempdir().unwrap();
        let artifacts = dir.path().join("artifacts");
        let mut seat = SeatState::new("impl-A", "a", 7);
        let s = command_helper("chatty-sleep");
        let out = invoke(
            &s,
            &mut seat,
            &Invocation {
                scope: crate::agent::InvocationScope::Repository,
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                // Wide enough to cover process-spawn latency inside a loaded
                // parallel test run, not merely the helper's first write. At
                // two seconds this passed alone and failed in the full suite,
                // which is a dice roll rather than a test.
                timeout: Duration::from_secs(10),
                allow_write: true,
                unsandboxed: false,
                sessions: true,
                artifacts: &artifacts,
                stem: "chatty",
                run: "test-run",
                node: "test",
                cache_dir: None,
                attachments: &[],
                writable: &[],
            },
        )
        .await
        .unwrap();

        assert!(out.timed_out, "{out:?}");
        assert!(!out.usable(), "a cut-off answer is still not an answer");
        let recorded = std::fs::read_to_string(artifacts.join("chatty.out")).unwrap();
        assert!(
            recorded.contains("i-said-something"),
            "the artifact must keep what arrived before the kill, got {recorded:?}"
        );
        assert!(
            out.text.contains("i-said-something"),
            "and the graph must be able to see it too, got {:?}",
            out.text
        );
    }

    #[test]
    fn missing_programs_reports_command_binaries() {
        let mut s = spec(AgentKind::Command, None);
        s.command = vec!["definitely-not-a-real-binary-xyz".to_owned()];
        assert_eq!(
            missing_programs(&[s]),
            ["definitely-not-a-real-binary-xyz".to_owned()]
        );
    }

    fn pick_spec(id: &str, kind: AgentKind) -> AgentSpec {
        AgentSpec {
            id: id.to_owned(),
            kind,
            model: None,
            command: Vec::new(),
            extra_args: Vec::new(),
            env: BTreeMap::new(),
            prompt_delivery: None,
            max_turn_secs: None,
        }
    }

    /// Availability stub: an agent is runnable unless its id was listed as
    /// missing. Keeps the selection tests off `PATH` entirely.
    fn without<'a>(missing: &'a [&'a str]) -> impl Fn(&AgentSpec) -> bool + 'a {
        move |a: &AgentSpec| !missing.contains(&a.id.as_str())
    }

    #[test]
    fn pick_prefers_the_claude_seat_even_when_it_is_not_first_in_the_roster() {
        let agents = [
            pick_spec("oc", AgentKind::Opencode),
            pick_spec("opus", AgentKind::Claude),
            pick_spec("agy", AgentKind::Antigravity),
        ];
        let got = pick(&agents, None, &without(&[])).expect("a pick");
        assert_eq!(got.id, "opus");
    }

    #[test]
    fn pick_falls_back_to_the_first_installed_agent_in_roster_order() {
        let agents = [
            pick_spec("opus", AgentKind::Claude),
            pick_spec("oc", AgentKind::Opencode),
            pick_spec("agy", AgentKind::Antigravity),
        ];
        let got = pick(&agents, None, &without(&["opus", "oc"])).expect("a pick");
        assert_eq!(got.id, "agy");
    }

    #[test]
    fn pick_on_an_empty_roster_says_what_to_install() {
        let msg = pick(&[], None, &without(&[]))
            .expect_err("nobody to ask")
            .to_string();
        assert!(msg.contains("roster is empty"), "{msg}");
        assert!(msg.contains("claude"), "{msg}");
        assert!(msg.contains("magi.toml"), "{msg}");
    }

    #[test]
    fn pick_on_a_roster_with_nothing_installed_names_the_programs_that_are_missing() {
        let agents = [
            pick_spec("opus", AgentKind::Claude),
            pick_spec("oc", AgentKind::Opencode),
        ];
        let err = pick(&agents, None, &without(&["opus", "oc"])).expect_err("nothing runnable");
        let msg = format!("{err:#}");
        assert!(msg.contains("claude"), "{msg}");
        assert!(msg.contains("opencode"), "{msg}");
    }

    #[test]
    fn an_explicitly_named_agent_wins_over_the_claude_preference() {
        let agents = [
            pick_spec("opus", AgentKind::Claude),
            pick_spec("oc", AgentKind::Opencode),
        ];
        let got = pick(&agents, Some("oc"), &without(&[])).expect("a pick");
        assert_eq!(got.id, "oc");
    }

    #[test]
    fn an_unknown_agent_id_lists_the_ids_that_do_exist() {
        let agents = [
            pick_spec("opus", AgentKind::Claude),
            pick_spec("oc", AgentKind::Opencode),
        ];
        let msg = pick(&agents, Some("gemini"), &without(&[]))
            .expect_err("no such agent")
            .to_string();
        assert!(msg.contains("gemini"), "{msg}");
        assert!(msg.contains("opus, oc"), "{msg}");
    }

    #[test]
    fn an_explicitly_named_agent_that_is_not_installed_is_an_error_not_a_fallback() {
        let agents = [
            pick_spec("opus", AgentKind::Claude),
            pick_spec("oc", AgentKind::Opencode),
        ];
        let msg = pick(&agents, Some("oc"), &without(&["oc"]))
            .expect_err("must not silently substitute another model")
            .to_string();
        assert!(msg.contains("opencode"), "{msg}");
        assert!(msg.contains("--agent"), "{msg}");
    }

    fn named(id: &str) -> AgentSpec {
        AgentSpec {
            id: id.to_owned(),
            ..spec(AgentKind::Command, None)
        }
    }

    fn output(text: &str, exit: i32, quota: bool) -> AgentOutput {
        AgentOutput {
            text: text.to_owned(),
            exit_code: Some(exit),
            timed_out: false,
            duration_ms: 0,
            artifacts: Vec::new(),
            quota: quota.then_some(Quota { reset: None }),
            dropped: None,
            commands: Vec::new(),
            context_tokens: None,
        }
    }

    #[test]
    fn a_chain_keeps_the_written_order_and_a_string_is_a_chain_of_one() {
        let agents = [named("a"), named("b"), named("c")];
        let all = |_: &AgentSpec| true;
        let chain = AgentChoice::Chain(vec!["c".into(), "a".into()]);
        let got = pick_chain(&agents, Some(&chain), &all, "synthesizer").unwrap();
        assert_eq!(
            got.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(),
            ["c", "a"]
        );

        let one = AgentChoice::from("b");
        let got = pick_chain(&agents, Some(&one), &all, "synthesizer").unwrap();
        assert_eq!(got.len(), 1);
        assert_eq!(got[0].id, "b");

        let got = pick_chain(&agents, None, &all, "synthesizer").unwrap();
        assert_eq!(got.len(), 1, "unset keeps pick's default");
        assert_eq!(got[0].id, "a");
        let empty = AgentChoice::Chain(Vec::new());
        assert_eq!(
            pick_chain(&agents, Some(&empty), &all, "x").unwrap()[0].id,
            "a"
        );
    }

    #[test]
    fn a_chain_skips_unknown_and_uninstalled_ids_and_tries_each_once() {
        let agents = [named("a"), named("b")];
        let not_a = |s: &AgentSpec| s.id != "a";
        let chain = AgentChoice::Chain(
            ["a", "ghost", "b", "b"]
                .iter()
                .map(|s| (*s).to_owned())
                .collect(),
        );
        let got = pick_chain(&agents, Some(&chain), &not_a, "chatter").unwrap();
        assert_eq!(got.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(), ["b"]);

        let dup = AgentChoice::Chain(vec!["b".into(), "a".into(), "b".into()]);
        let got = pick_chain(&agents, Some(&dup), &|_| true, "chatter").unwrap();
        assert_eq!(
            got.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(),
            ["b", "a"]
        );
    }

    #[test]
    fn a_chain_with_nothing_runnable_names_the_role() {
        let agents = [named("a")];
        let chain = AgentChoice::Chain(vec!["a".into(), "ghost".into()]);
        let err = pick_chain(&agents, Some(&chain), &|_| false, "conductor")
            .unwrap_err()
            .to_string();
        assert!(err.contains("conductor"), "{err}");
    }

    #[test]
    fn a_chain_advances_on_error_quota_or_an_unusable_answer_only() {
        assert!(chain_advances(&Err(anyhow::anyhow!("spawn failed"))));
        assert!(chain_advances(&Ok(output("limit", 0, true))));
        assert!(chain_advances(&Ok(output("", 0, false))));
        assert!(chain_advances(&Ok(output("x", 1, false))));
        assert!(!chain_advances(&Ok(output("answer", 0, false))));
    }

    #[test]
    fn claude_context_tokens_are_unknown_because_usage_is_aggregated() {
        for out in [
            r#"{"result":"ok","num_turns":1,"usage":{"input_tokens":10,"cache_read_input_tokens":3000}}"#,
            r#"{"result":"ok","num_turns":3,"usage":{"input_tokens":10}}"#,
            r#"{"result":"ok"}"#,
            "not json",
        ] {
            assert_eq!(context_tokens(AgentKind::Claude, out), None, "{out}");
        }
    }

    #[test]
    fn opencode_context_tokens_take_the_last_step_finish() {
        let out = concat!(
            r#"{"type":"step_finish","sessionID":"s","part":{"type":"step-finish","tokens":{"input":100,"output":5,"cache":{"read":1000,"write":50}}}}"#,
            "\n",
            r#"{"type":"text","part":{"type":"text","text":"hi"}}"#,
            "\n",
            r#"{"type":"step_finish","sessionID":"s","part":{"type":"step-finish","tokens":{"input":120,"output":9,"cache":{"read":1500}}}}"#,
            "\n"
        );
        assert_eq!(context_tokens(AgentKind::Opencode, out), Some(1620));
        let none = r#"{"type":"step_finish","part":{"type":"step-finish"}}"#;
        assert_eq!(context_tokens(AgentKind::Opencode, none), None);
        assert_eq!(context_tokens(AgentKind::Opencode, ""), None);
    }

    #[test]
    fn agy_context_tokens_are_unknown_because_usage_is_aggregated() {
        let out = r#"{"conversation_id":"c","status":"OK","response":"x","usage":{"input_tokens":260113,"cache_read_tokens":2200925}}"#;
        assert_eq!(context_tokens(AgentKind::Antigravity, out), None);
    }

    #[test]
    fn codex_context_tokens_are_unknown_because_usage_is_cumulative() {
        let out = concat!(
            r#"{"type":"turn.completed","usage":{"input_tokens":1000,"cached_input_tokens":900}}"#,
            "\n"
        );
        assert_eq!(context_tokens(AgentKind::Codex, out), None);
    }

    #[test]
    fn omp_context_tokens_come_from_messages_without_an_agent_end() {
        let out = concat!(
            r#"{"type":"session","id":"s"}"#,
            "\n",
            r#"{"type":"message_end","message":{"role":"assistant","content":[],"usage":{"input":50,"cacheRead":400,"cacheWrite":10}}}"#,
            "\n",
            r#"{"type":"turn_end","message":{"role":"assistant","content":[],"usage":{"input":70,"cacheRead":500}}}"#,
            "\n"
        );
        assert_eq!(context_tokens(AgentKind::Omp, out), Some(570));
        let user_only = r#"{"type":"message_end","message":{"role":"user","usage":{"input":9}}}"#;
        assert_eq!(context_tokens(AgentKind::Omp, user_only), None);
        let no_usage = r#"{"type":"message_end","message":{"role":"assistant","content":[]}}"#;
        assert_eq!(context_tokens(AgentKind::Omp, no_usage), None);
    }

    #[test]
    fn command_agents_report_no_context_tokens() {
        let out = r#"{"usage":{"input_tokens":5}}"#;
        assert_eq!(context_tokens(AgentKind::Command, out), None);
    }
}