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,
    }
}

/// One agent invocation.
#[derive(Debug, Clone)]
pub struct Invocation<'a> {
    /// 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. Together
    /// with an editing node (see [`repository_writer`]) it also drops a codex
    /// seat's OS sandbox and approval gate, as claude's `bypassPermissions`
    /// does for every claude seat.
    pub allow_write: 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`]; adapters that widen
    /// a workspace per directory (agy's `--add-dir`) need them, while a codex
    /// editing seat runs without a sandbox and ignores them.
    pub writable: &'a [PathBuf],
}

/// A repository editing seat: implementer, fixer, grafter, gate fixer, land
/// or rebase resolver, or a chat turn the repository opted into writing.
///
/// For codex this is also the whole sandbox policy: these seats run with
/// `--dangerously-bypass-approvals-and-sandbox` (owner decision 2026-10-27),
/// matching claude's `bypassPermissions`, because codex's seatbelt sandbox
/// refuses `$TMPDIR` and aborts Chrome, so repository tests cannot run inside
/// it. Every other seat keeps its sandbox.
fn repository_writer(inv: &Invocation<'_>) -> bool {
    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.
    // Chat turns and their transcript-only title seats bypass repository
    // scheduling, including waiter delivery.
    !matches!(
        inv.node,
        crate::queue::CHAT_NODE | "talk-title" | "deputy" | "conduct" | "dupes"
    )
}

/// 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 requirements in history.
fn editing_access_note(base: Option<&crate::run::FetchedBase>) -> String {
    let mut note = String::from(
        "Launcher repository access contract (applies to this turn and overrides earlier fetch 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");
    }
    note
}

/// Evidence of a CLI usage limit or authentication outage, distinct from an
/// ordinary task 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 {
    /// Machine-side cause; old records describe a usage limit.
    #[serde(default)]
    pub cause: QuotaCause,
    /// Agent id, filled by the invocation boundary.
    #[serde(default)]
    pub agent: String,
    /// Human-readable reset time, when the CLI printed one.
    #[serde(default)]
    pub reset: Option<String>,
}

/// Why an invocation needs machine-side recovery rather than a task retry.
#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum QuotaCause {
    /// Subscription usage must reset or be replenished.
    #[default]
    UsageLimit,
    /// The operator must restore the CLI's authenticated session.
    Authentication,
}

/// 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>,
    /// Usage limit or authentication outage, when distinguishable 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.quota_exhausted()
            && !self.timed_out
            && self.exit_code == Some(0)
            && !self.text.trim().is_empty()
    }

    /// Does this invocation need machine-side quota or authentication recovery?
    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. Running
    // owns the lease through descendant cleanup on cancel.
    let lease = if repository_seat(inv) {
        crate::repo_lock::SeatLease::for_checkout(inv.cwd, inv.allow_write, inv.coordination)
            .await?
    } else {
        None
    };
    let inv = &Invocation {
        timeout: spec.turn_timeout(inv.timeout),
        ..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 mut prompt = inv.prompt.to_owned();
    if inv.node == crate::queue::CHAT_NODE {
        prompt.push_str("\n\n");
        prompt.push_str(crate::prompt::CHAT_FOLLOW_THROUGH);
    }
    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(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.logged_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": repository_writer(inv),
            "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_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 {
            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
    };

    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) = &mut extracted.quota {
        quota.agent.clone_from(&spec.id);
        if quota.cause == QuotaCause::Authentication {
            let _ = crate::notices::Notices::open().raise(crate::notices::Notice::warn(
                &format!("agent-authentication:{}", spec.id),
                format!("Agent `{}` cannot authenticate. Re-authenticate this agent, then resume the held task.", spec.id),
            ));
        }
        // 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,
            cause = ?quota.cause,
            "agent needs machine-side recovery"
        );
    }
    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>,
}

impl Plan {
    /// Settings may have been loaded from a file containing credentials.
    /// Log the option, never its value, including repeated and equals forms.
    fn logged_argv(&self) -> Vec<&str> {
        let mut settings_value = false;
        self.argv
            .iter()
            .map(|arg| {
                if settings_value {
                    settings_value = false;
                    "[redacted]"
                } else if arg == "--settings" {
                    settings_value = true;
                    arg.as_str()
                } else if arg.starts_with("--settings=") {
                    "--settings=[redacted]"
                } else {
                    arg.as_str()
                }
            })
            .collect()
    }
}

/// 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. Repository editing seats
            // (implementer, fixer, grafter, gate fixer, land, rebase and an
            // opted-in chat turn - see `repository_writer`) drop sandbox and
            // approvals altogether, as claude's `bypassPermissions` does: the
            // seatbelt sandbox refuses `$TMPDIR` and aborts Chrome, so the
            // repository's own tests could not run (owner decision
            // 2026-10-27). Every other seat keeps its sandbox.
            if repository_writer(inv) {
                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 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: codex_quota(stdout),
                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()
}

/// Require both the observed top-level limit event and a failed turn. Text
/// inside an assistant message is never evidence of an outage.
fn codex_quota(stdout: &str) -> Option<Quota> {
    let mut message = None;
    let mut failed = false;
    for line in stdout.lines() {
        let Ok(v) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        match v.get("type").and_then(serde_json::Value::as_str) {
            Some("error") => {
                if let Some(m) = v.get("message").and_then(serde_json::Value::as_str)
                    && m.starts_with("You've hit your usage limit. Visit https://chatgpt.com/settings/usage to purchase more credits")
                {
                    message = Some(m.to_owned());
                }
            }
            Some("turn.failed") => failed = true,
            _ => {}
        }
    }
    if !failed {
        return None;
    }
    let message = message?;
    Some(Quota {
        reset: message
            .split_once("try again at ")
            .map(|(_, hint)| hint.trim().trim_end_matches('.').trim().to_owned())
            .filter(|hint| !hint.is_empty()),
        cause: QuotaCause::UsageLimit,
        agent: String::new(),
    })
}

/// Recognise claude's usage-limit and expired OAuth error shapes, 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`;
/// also accepts the observed api_error result for an expired OAuth session.
/// Unknown error shapes remain ordinary failures.
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("");
    if v.get("terminal_reason").and_then(serde_json::Value::as_str) == Some("api_error")
        && result.starts_with("Failed to authenticate:")
        && result.contains("OAuth session expired and could not be refreshed")
    {
        return Some(Quota {
            reset: None,
            cause: QuotaCause::Authentication,
            agent: String::new(),
        });
    }
    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,
        cause: QuotaCause::UsageLimit,
        agent: String::new(),
    })
}

/// 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,
        cause: QuotaCause::UsageLimit,
        agent: String::new(),
    })
}

/// 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"
            )
        })
}

/// `spec` for a seat whose answer is posted to GitHub (a pull request title or
/// description, a release reason), so the answer comes back in
/// `github_language` whatever language the operator set for their own CLI.
///
/// claude applies the user-level `language` setting (`"language": "Japanese"`
/// in `~/.claude/settings.json`) over a sentence in the prompt: the change
/// summary prompt said "Write ... in English" and the description came back in
/// Japanese, three times out of three when replayed, and in English every time
/// with this override. A per-invocation `--settings` outranks the user file, so
/// it is the CLI's own mechanism rather than more prompt wording. The other
/// CLIs have no operator-level answer-language preference magi knows of and
/// are returned unchanged.
pub fn for_github_text(spec: &AgentSpec, github_language: &str, cwd: &Path) -> Result<AgentSpec> {
    let mut spec = spec.clone();
    if spec.kind == AgentKind::Claude {
        // The CLI uses the last --settings value. Replace that value in
        // place, preserving its settings and the precedence of earlier flags.
        let mut existing = None;
        let mut args = spec.extra_args.iter().enumerate();
        while let Some((index, arg)) = args.next() {
            if arg == "--settings" {
                let (value_index, value) = args.next().context("--settings requires a value")?;
                existing = Some((value_index, value.as_str(), false));
            } else if let Some(value) = arg.strip_prefix("--settings=") {
                existing = Some((index, value, true));
            }
        }
        let mut settings = if let Some((_, value, _)) = existing {
            let json = if value.trim_start().starts_with('{') {
                value.to_owned()
            } else {
                std::fs::read_to_string(cwd.join(value))
                    .context("cannot read existing seat --settings file")?
            };
            serde_json::from_str::<serde_json::Value>(&json)
                .context("cannot parse existing seat --settings JSON")?
        } else {
            serde_json::json!({})
        };
        settings
            .as_object_mut()
            .context("seat --settings must be a JSON object")?
            .insert(
                "language".to_owned(),
                serde_json::json!(crate::lang::name(github_language)),
            );
        let settings = settings.to_string();
        if let Some((index, _, equals)) = existing {
            spec.extra_args[index] = if equals {
                format!("--settings={settings}")
            } else {
                settings
            };
        } else {
            spec.extra_args.extend(["--settings".to_owned(), settings]);
        }
    }
    Ok(spec)
}

/// 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 {
            coordination: None,
            fetched_base: None,
            cwd,
            prompt: "do the thing",
            timeout: Duration::from_secs(900),
            allow_write,
            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(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(Some(&base));
        for expected in [
            "origin/main",
            "0123456789abcdef",
            "2026-10-18T00:00:00Z",
            "Do not fetch again",
        ] {
            assert!(note.contains(expected), "missing {expected}");
        }
        // Editing seats run without an OS sandbox, so nothing tells them
        // their network, Git or temporary-directory access is restricted.
        for text in [&missing, &note] {
            assert!(!text.to_lowercase().contains("sandbox"), "{text}");
            assert!(!text.contains("Network access is restricted"), "{text}");
        }
        let roundtrip: crate::run::FetchedBase =
            serde_json::from_str(&serde_json::to_string(&base).unwrap()).unwrap();
        assert_eq!(editing_access_note(Some(&roundtrip)), note);
    }

    #[test]
    fn codex_editing_seats_bypass_the_sandbox_on_initial_and_resume() {
        const BYPASS: &str = "--dangerously-bypass-approvals-and-sandbox";
        let roots = [PathBuf::from("/shared/git"), PathBuf::from("/questions")];
        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 build = |invocation: &Invocation<'_>| {
                build_command(
                    &spec(AgentKind::Codex, None),
                    &seat,
                    invocation,
                    Path::new("/art/p.md"),
                )
                .unwrap()
            };
            for node in [
                "implement",
                "fix",
                "graft",
                "gate-fix",
                "land",
                "rebase",
                crate::queue::CHAT_NODE,
            ] {
                let mut invocation = inv(Path::new("."), Path::new("/art"), true);
                invocation.node = node;
                invocation.writable = &roots;
                let plan = build(&invocation);
                let at = plan.argv.iter().position(|s| s == BYPASS);
                assert!(at.is_some(), "{node}: {:?}", plan.argv);
                if resumed {
                    let resume = plan.argv.iter().position(|s| s == "resume").unwrap();
                    assert!(at.unwrap() < resume, "{node}: {:?}", plan.argv);
                }
                // No sandbox, approval or sandbox-scoped override survives.
                assert!(!plan.argv.iter().any(|s| s == "--sandbox"), "{node}");
                assert!(!plan.argv.iter().any(|s| s == "-c"), "{node}");
            }
            for (node, write) in [
                ("review", false),
                ("judge", false),
                ("advise", false),
                ("implement", false),
                ("fix", false),
                ("dupes", false),
                ("conduct", false),
                ("waiter", false),
                ("talk-title", false),
                (crate::queue::CHAT_NODE, false),
                ("deputy", true),
            ] {
                let mut invocation = inv(Path::new("."), Path::new("/art"), write);
                invocation.node = node;
                let plan = build(&invocation);
                assert!(!plan.argv.iter().any(|s| s == BYPASS), "{node}");
                let mode = if write {
                    "workspace-write"
                } else {
                    "read-only"
                };
                assert!(
                    plan.argv.windows(2).any(|w| w == ["--sandbox", mode]),
                    "{node}: {:?}",
                    plan.argv
                );
                assert!(
                    plan.argv
                        .windows(2)
                        .any(|w| w == ["-c", "approval_policy=\"never\""]),
                    "{node}"
                );
            }
        }
    }

    #[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 github_bound_claude_seats_override_the_operators_answer_language() {
        let seat = SeatState::new("change-summary", "a", 7);
        let art = Path::new("/art");
        for (language, name) in [("en", "English"), ("", "English"), ("ja", "Japanese")] {
            let claude =
                for_github_text(&spec(AgentKind::Claude, None), language, Path::new(".")).unwrap();
            let plan = build_command(
                &claude,
                &seat,
                &inv(Path::new("."), art, false),
                Path::new("/art/p.md"),
            )
            .unwrap();
            let settings = format!(r#"{{"language":"{name}"}}"#);
            assert!(
                plan.argv
                    .windows(2)
                    .any(|w| w == ["--settings", settings.as_str()]),
                "{:?}",
                plan.argv
            );
            // The prompt still arrives on stdin, not as an argument a
            // variadic flag could swallow.
            assert_eq!(plan.stdin.as_deref(), Some("do the thing"));
        }
        for kind in [
            AgentKind::Codex,
            AgentKind::Opencode,
            AgentKind::Antigravity,
            AgentKind::Omp,
        ] {
            assert!(
                for_github_text(&spec(kind, None), "en", Path::new("."))
                    .unwrap()
                    .extra_args
                    .is_empty()
            );
        }
    }

    #[test]
    fn github_text_preserves_existing_settings() {
        let dir = tempfile::tempdir().unwrap();
        let original = serde_json::json!({
            "language": "Japanese",
            "env": { "SEAT_MODE": "configured" },
            "apiKeyHelper": "credential-helper",
            "hooks": { "Stop": [] }
        });
        std::fs::write(dir.path().join("seat-settings.json"), original.to_string()).unwrap();
        for value in [original.to_string(), "./seat-settings.json".to_owned()] {
            for equals in [false, true] {
                let mut spec = spec(AgentKind::Claude, None);
                // Earlier settings stay earlier; the CLI's last value wins.
                spec.extra_args = vec!["--settings={\"language\":\"French\"}".to_owned()];
                if equals {
                    spec.extra_args.push(format!("--settings={value}"));
                } else {
                    spec.extra_args
                        .extend(["--settings".to_owned(), value.clone()]);
                }
                spec.extra_args.push("--verbose".to_owned());
                let updated = for_github_text(&spec, "en", dir.path()).unwrap();
                assert_eq!(updated.extra_args.len(), spec.extra_args.len());
                assert_eq!(updated.extra_args[0], spec.extra_args[0]);
                assert_eq!(updated.extra_args.last(), spec.extra_args.last());
                let value = &updated.extra_args[updated.extra_args.len() - 2];
                let actual: serde_json::Value =
                    serde_json::from_str(value.strip_prefix("--settings=").unwrap_or(value))
                        .unwrap();
                let mut expected = original.clone();
                expected["language"] = serde_json::json!("English");
                assert_eq!(actual, expected);
            }
        }
        let on_disk: serde_json::Value = serde_json::from_str(
            &std::fs::read_to_string(dir.path().join("seat-settings.json")).unwrap(),
        )
        .unwrap();
        assert_eq!(on_disk, original);
    }

    #[test]
    fn github_text_settings_are_redacted_from_command_logging() {
        let dir = tempfile::tempdir().unwrap();
        let settings = serde_json::json!({ "env": { "AUTH_TOKEN": "test-only-secret" } });
        std::fs::write(dir.path().join("seat-settings.json"), settings.to_string()).unwrap();
        for equals in [false, true] {
            let mut spec = spec(AgentKind::Claude, None);
            spec.extra_args =
                vec!["--settings={\"env\":{\"AUTH_TOKEN\":\"earlier-secret\"}}".to_owned()];
            if equals {
                spec.extra_args
                    .push("--settings=./seat-settings.json".to_owned());
            } else {
                spec.extra_args
                    .extend(["--settings".to_owned(), "./seat-settings.json".to_owned()]);
            }
            let spec = for_github_text(&spec, "en", dir.path()).unwrap();
            let plan = build_command(
                &spec,
                &SeatState::new("change-summary", "a", 7),
                &inv(dir.path(), dir.path(), false),
                &dir.path().join("prompt.md"),
            )
            .unwrap();
            // Execution still receives both the configured token and language.
            let actual = plan.argv.last().unwrap();
            let actual: serde_json::Value =
                serde_json::from_str(actual.strip_prefix("--settings=").unwrap_or(actual)).unwrap();
            assert_eq!(actual["env"]["AUTH_TOKEN"], "test-only-secret");
            assert_eq!(actual["language"], "English");
            let logged = format!("{:?}", plan.logged_argv());
            assert!(!logged.contains("secret"));
            assert!(!logged.contains("AUTH_TOKEN"));
            assert!(logged.contains("[redacted]"));
            assert_eq!(plan.logged_argv()[0], "claude");
            assert!(plan.logged_argv().contains(&"--output-format"));
        }
    }

    #[test]
    fn github_text_rejects_unreadable_or_invalid_settings() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("bad.json"), "invalid json").unwrap();
        std::fs::write(dir.path().join("array.json"), "[]").unwrap();
        for args in [
            vec!["--settings"],
            vec!["--settings", "missing.json"],
            vec!["--settings=bad.json"],
            vec!["--settings", "array.json"],
        ] {
            let mut spec = spec(AgentKind::Claude, None);
            spec.extra_args = args.into_iter().map(str::to_owned).collect();
            assert!(for_github_text(&spec, "en", dir.path()).is_err());
        }
    }

    #[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")));
    }

    #[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", "talk-title"] {
            i.node = node;
            assert!(!repository_writer(&i), "{node}");
        }
        i.node = "chat";
        let roots = [PathBuf::from("/questions")];
        i.writable = &roots;
        assert!(
            repository_writer(&i),
            "explicit roots do not change the role"
        );
        assert!(
            !repository_seat(&i),
            "chat turns bypass repository scheduling"
        );
    }

    #[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",
            "talk-title",
        ] {
            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 | "talk-title" | "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_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 only repository
        //    editing seats (see `codex_editing_seats_bypass_the_sandbox_on_initial_and_resume`)
        //    ever get the bypass. A seat that may write without editing the
        //    repository (a deputy's stores) keeps the workspace sandbox.
        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"])
        );
        for p in [&ro, &rw] {
            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 {
                coordination: None,
                fetched_base: None,
                cwd: Path::new("."),
                prompt: "p",
                timeout: Duration::from_secs(3600),
                allow_write: true,
                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"));
    }

    const EXPIRED_AUTH: &str = r#"{"is_error":true,"terminal_reason":"api_error","result":"Failed to authenticate: OAuth session expired and could not be refreshed"}"#;
    const CODEX_LIMIT: &str = r#"{"type":"error","message":"You've hit your usage limit. Visit https://chatgpt.com/settings/usage to purchase more credits or try again at Oct 27th, 2026 6:18 PM."}
{"type":"turn.failed","error":{"message":"You've hit your usage limit."}}"#;

    #[test]
    fn observed_machine_failures_are_detected_conservatively() {
        let auth = extract(AgentKind::Claude, EXPIRED_AUTH).quota.unwrap();
        assert_eq!(auth.cause, QuotaCause::Authentication);
        assert_eq!(auth.reset, None);
        let limit = extract(AgentKind::Codex, CODEX_LIMIT).quota.unwrap();
        assert_eq!(limit.cause, QuotaCause::UsageLimit);
        assert_eq!(limit.reset.as_deref(), Some("Oct 27th, 2026 6:18 PM"));
        // Duplicate events still yield one classification.
        assert!(
            extract(AgentKind::Codex, &format!("{CODEX_LIMIT}\n{CODEX_LIMIT}"))
                .quota
                .is_some()
        );
        let without_reset = CODEX_LIMIT.replace(" or try again at Oct 27th, 2026 6:18 PM", "");
        assert!(
            extract(AgentKind::Codex, &without_reset)
                .quota
                .unwrap()
                .reset
                .is_none()
        );
        for text in [
            EXPIRED_AUTH.replace("true", "false"),
            EXPIRED_AUTH.replace("api_error", "unknown"),
            EXPIRED_AUTH.replace(
                "OAuth session expired and could not be refreshed",
                "unknown error",
            ),
        ] {
            assert!(extract(AgentKind::Claude, &text).quota.is_none());
        }
        assert!(
            extract(AgentKind::Codex, CODEX_LIMIT.lines().next().unwrap())
                .quota
                .is_none()
        );
        assert!(
            extract(
                AgentKind::Codex,
                &CODEX_LIMIT.replace("You've hit your usage limit.", "Something failed.")
            )
            .quota
            .is_none()
        );
        let quoted = serde_json::json!({"type":"item.completed","item":{"type":"agent_message","text":CODEX_LIMIT}}).to_string();
        assert!(extract(AgentKind::Codex, &quoted).quota.is_none());
        assert!(extract(AgentKind::Codex, EXPIRED_AUTH).quota.is_none());
        assert!(
            extract(AgentKind::Antigravity, EXPIRED_AUTH)
                .quota
                .is_none()
        );
        let legacy: Quota = serde_json::from_str(r#"{"reset":null}"#).unwrap();
        assert_eq!(legacy.cause, QuotaCause::UsageLimit);
    }

    #[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 {
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_secs(30),
                allow_write: true,
                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 {
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_secs(30),
                allow_write: true,
                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 {
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_secs(30),
                allow_write: 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 {
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: &big,
                timeout: Duration::from_secs(60),
                allow_write: true,
                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 {
                coordination: None,
                fetched_base: None,
                cwd: dir.path(),
                prompt: "unused",
                timeout: Duration::from_millis(300),
                allow_write: true,
                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 {
                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,
                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,
                cause: QuotaCause::UsageLimit,
                agent: String::new(),
            }),
            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);
    }
}