//! 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>,
/// Working directory. Always a real checkout, so every CLI can read the
/// repository without per-vendor "extra directory" flags.
pub cwd: &'a Path,
/// The full prompt.
pub prompt: &'a str,
/// Wall-clock limit; the process tree is killed when it elapses.
pub timeout: Duration,
/// May the agent modify files? Judges and reviewers may not.
pub allow_write: bool,
/// Drop the CLI's own sandbox and approval gate. Conversational talk
/// turns only, and only together with `allow_write`: it is the codex
/// counterpart of claude's `bypassPermissions`. Unattended seats
/// (implementer, judge, reviewer, deputy, ...) never set it, and only
/// `talk.rs` may.
pub unsandboxed: bool,
/// Continue this seat's conversation when the CLI supports it.
pub sessions: bool,
/// Directory for prompt / stdout / stderr artifacts.
pub artifacts: &'a Path,
/// Artifact filename stem.
pub stem: &'a str,
/// Run this invocation belongs to. Exported as `MAGI_RUN` so an agent that
/// files a task with `magi task add` is attributed to the run that was
/// paying for it, rather than looking like a human wandered by.
pub run: &'a str,
/// Graph node being executed, e.g. `implement` or `review`. Exported as
/// `MAGI_NODE` for the same reason: "who asked for this" is the first
/// question about an autonomously created task.
pub node: &'a str,
/// Shared build cache the seat should build into, from the rendered
/// `CARGO_TARGET_DIR=` in the verify commands. Exported as
/// `CARGO_TARGET_DIR` so the implementer's compile lands inside the same
/// directory `verify` reads back from - one cache, one prune, and the
/// build the agent just paid for is the build the gate reuses.
pub cache_dir: Option<&'a Path>,
/// Absolute paths of images the operator attached to this conversation,
/// outside `cwd` - see `chat`/`talk`'s `attachments_dir`. Empty for every
/// invocation that is not a chat or talk turn. [`build_command`] uses
/// this only to decide whether a CLI's sandbox needs widening to read
/// them; the prompt text naming each path and its mime is built by the
/// caller, not here.
pub attachments: &'a [PathBuf],
/// Directories outside `cwd` the seat may write, when it may write at all.
/// Explicit, purpose-specific grants (for example a deputy's stores).
/// Repository editing seats also receive the shared Git directory and
/// question control-plane directories in [`invoke`].
/// For chat, nonempty explicit roots with write permission designate a
/// control-only consultation turn; repository investigation is suspended.
pub writable: &'a [PathBuf],
}
fn repository_writer(inv: &Invocation<'_>) -> bool {
inv.allow_write
&& (matches!(
inv.node,
"implement" | "fix" | "graft" | "gate-fix" | "land" | "rebase"
) || (inv.node == "chat" && repository_seat(inv)))
}
fn consulted_chat(inv: &Invocation<'_>) -> bool {
// talk::turn supplies explicit question/answer-store roots only while
// delivering a consultation or an authorized hold.
inv.node == "chat" && inv.allow_write && !inv.writable.is_empty()
}
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.
!matches!(inv.node, "deputy" | "conduct" | "dupes") && !consulted_chat(inv)
}
/// 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)
}
/// Evidence that a CLI ran out of its rate limit / quota, distinct from an
/// ordinary failure.
///
/// `reset` is free text: CLIs render the reset time in their own locale, and
/// parsing it exactly would be a bug factory. When it is not readable we say
/// nothing rather than invent a format.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Quota {
/// Human-readable reset time, when the CLI printed one.
#[serde(default)]
pub reset: Option<String>,
}
/// A CLI hung up on its own stream while the agent was working.
///
/// Separate from a failure because the work was done and billed, and separate
/// from a [`Quota`] because it is worth asking again: the answer is in the
/// conversation, not lost to a limit that has to reset first. See
/// [`dropped_stream`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Dropped {
/// What the CLI said as it hung up, verbatim.
pub why: String,
/// Output tokens the CLI reported before it did - the evidence that this
/// was a delivery failure and not an agent that produced nothing.
pub output_tokens: u64,
}
/// One command a CLI's own structured event stream reported running, with
/// the result it reported for it.
///
/// This is evidence the CLI chose to report about its own tool loop — never
/// something magi polled, supervised, or inferred from a process list. Only
/// the Codex arm of [`extract`] currently populates it (its `item.completed`
/// / `command_execution` events name `id`, `command`, `exit_code` and
/// `aggregated_output` directly); every other backend's CLI does not expose
/// this in what magi currently captures, so its seats simply never produce
/// any. A command a CLI never reported finishing (still running when the
/// turn ended, or the event stream never named it) has no entry here either
/// — there is no event to build one from, and this type must never be used
/// to *guess* that a command is still in flight.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommandEvidence {
/// The CLI's own id for this command.
pub id: String,
/// The command itself, as the CLI reported it.
pub description: String,
/// Exit code the CLI reported for it.
pub exit_code: Option<i32>,
/// Tail of the command's own output, when the CLI reported one.
pub result_summary: String,
/// Which CLI/event stream this came from, e.g. `"codex"`.
pub source: String,
}
/// Result of an agent invocation.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentOutput {
/// The agent's final message, extracted from whatever the CLI printed.
pub text: String,
/// Exit status code.
pub exit_code: Option<i32>,
/// Did the invocation hit its timeout?
pub timed_out: bool,
/// Wall-clock duration.
pub duration_ms: u64,
/// Artifact file names, relative to the run's `artifacts/` directory.
pub artifacts: Vec<String>,
/// Rate-limit / quota exhaustion, when it can be told apart from a normal
/// failure. `None` for a normal failure, a timeout, or a CLI we cannot
/// read — the conservative default.
#[serde(default)]
pub quota: Option<Quota>,
/// The CLI hung up on its own stream after the agent had done billed
/// work. `None` unless that exact shape was recognised — see
/// [`dropped_stream`].
#[serde(default)]
pub dropped: Option<Dropped>,
/// Commands the CLI's own event stream reported running, see
/// [`CommandEvidence`]. Always empty for a backend this crate does not
/// currently read structured job events from.
#[serde(default)]
pub commands: Vec<CommandEvidence>,
/// Input-side tokens the CLI reported for this turn - approximately the
/// size of the context the model just read. `None` when the CLI reports
/// no usage (or in a shape this build does not read): unknown, never 0.
/// See [`context_tokens`].
#[serde(default)]
pub context_tokens: Option<u64>,
}
impl AgentOutput {
/// Did the CLI exit cleanly with something to say?
pub fn usable(&self) -> bool {
!self.timed_out && self.exit_code == Some(0) && !self.text.trim().is_empty()
}
/// Did this invocation run out of the CLI's rate limit / quota?
pub fn quota_exhausted(&self) -> bool {
self.quota.is_some()
}
/// Did the agent work and the CLI fail to deliver it?
///
/// Worth re-asking, unlike [`AgentOutput::quota_exhausted`]: the answer is
/// in a conversation this process can resume.
pub fn work_undelivered(&self) -> bool {
self.dropped.is_some()
}
}
/// 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.
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.
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.
async fn collect(
buf: &Captured,
handle: Option<tokio::task::JoinHandle<()>>,
grace: Duration,
) -> String {
if let Some(handle) = handle {
if tokio::time::timeout(grace, handle).await.is_err() {
tracing::debug!("a pipe is still held open after the child exited");
}
}
let bytes = buf.lock().map(|g| g.clone()).unwrap_or_default();
String::from_utf8_lossy(&bytes).into_owned()
}
/// Invoke `spec` for `seat`, updating the seat's conversation state.
pub async fn invoke(
spec: &AgentSpec,
seat: &mut SeatState,
inv: &Invocation<'_>,
) -> Result<AgentOutput> {
// Wait for repository scheduling before spawning or starting a seat's
// budget. Readers hold the real lock; editors exclude local readers while
// taking their own repository-prescribed lock inside the seat.
let lease = if repository_seat(inv) {
crate::repo_lock::SeatLease::for_checkout(inv.cwd, inv.allow_write, inv.coordination)
.await?
} else {
None
};
let prompt = if consulted_chat(inv) {
Some(format!(
"{}\n\nCurrent turn scope: answer the supplied consultation or authorized hold using the conversation and supplied question context and supported magi control-plane commands only. This overrides repository investigation permissions in the standing briefing or earlier turns. No repository lease is held: do not read repository files or history, fetch, run tests, or edit repository files. If repository investigation is needed, report that it must wait for a normal repository-coordinated turn after the consultation is resolved.",
inv.prompt
))
} else {
lease.as_ref().filter(|lease| lease.delegated).map(|lease| format!(
"Repository coordination: the launcher holds the exclusive repository lock at {} on behalf of this read-only seat for this entire invocation. Do not acquire a second lock or release/delete that file. Your filesystem access remains read-only.\n\n{}",
lease.path.display(), inv.prompt
))
};
// 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);
}
}
}
let inv = &Invocation {
writable: &writable,
prompt: prompt.as_deref().unwrap_or(inv.prompt),
..inv.clone()
};
tokio::fs::create_dir_all(inv.artifacts)
.await
.with_context(|| format!("create {}", inv.artifacts.display()))?;
let prompt_path = inv.artifacts.join(format!("{}.prompt.md", inv.stem));
tokio::fs::write(&prompt_path, inv.prompt)
.await
.with_context(|| format!("write {}", prompt_path.display()))?;
let plan = build_command(spec, seat, inv, &prompt_path)?;
tracing::debug!(seat = %seat.key, agent = %spec.id, argv = ?plan.argv, "spawning agent");
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(unix)]
if lease.is_some() {
cmd.process_group(0);
}
let child = cmd
.spawn()
.with_context(|| format!("spawn `{}` for seat {}", plan.argv[0], seat.key))?;
// No await occurs between spawn and transfer to the cancellation owner.
let mut running = crate::seat_process::Running::new(child, lease);
// 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 (code, timed_out) = match tokio::time::timeout(inv.timeout, running.child().wait()).await {
Ok(res) => {
let status = res.with_context(|| format!("wait for seat {}", seat.key))?;
(status.code(), false)
}
Err(_) => {
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();
let mut extracted = extract(spec.kind, &stdout);
if spec.kind == AgentKind::Antigravity && extracted.quota.is_none() {
extracted.quota = agy_quota(&stdout, &stderr);
}
if let Some(quota) = &extracted.quota {
// A quota response carries usage too, so it can look like a dropped
// stream; a rate limit is never worth re-asking, so it wins.
extracted.dropped = None;
tracing::warn!(
seat = %seat.key,
agent = %spec.id,
reset = ?quota.reset,
"agent is out of quota"
);
}
if let Some(session) = extracted.session {
match spec.kind {
AgentKind::Claude => seat.claude_session = Some(session),
AgentKind::Opencode | AgentKind::Antigravity | AgentKind::Codex | AgentKind::Omp => {
seat.captured_session = Some(session);
}
AgentKind::Command => {}
}
}
if let Some(status) = &extracted.status
&& !status.eq_ignore_ascii_case("success")
{
tracing::warn!(seat = %seat.key, status = %status, "agent reported a non-success status");
}
let text = if extracted.text.trim().is_empty() {
// A CLI that printed only to stderr still told us something.
if stdout.trim().is_empty() {
stderr.trim().to_owned()
} else {
stdout.trim().to_owned()
}
} else {
extracted.text
};
seat.turns += 1;
Ok(AgentOutput {
text,
exit_code: code,
timed_out,
duration_ms: started.elapsed().as_millis() as u64,
artifacts: vec![
file_name(&prompt_path),
file_name(&out_path),
file_name(&err_path),
],
quota: extracted.quota,
dropped: extracted.dropped,
commands: extracted.commands,
context_tokens: extracted.context_tokens,
})
}
fn file_name(p: &Path) -> String {
p.file_name()
.unwrap_or_default()
.to_string_lossy()
.into_owned()
}
/// The argv plus optional stdin body for one invocation.
#[derive(Debug)]
struct Plan {
argv: Vec<String>,
stdin: Option<String>,
}
/// How a file-delivered prompt is pointed at, per CLI.
///
/// `agy` has a native file-context syntax, `@<path>`, and it is measurably the
/// better contract: on the same trivial task it finished in 17s against 73s for
/// the prose form, because prose makes the model spend a tool round-trip
/// deciding to read the file. It is also the form yukimemi/rvpm proved out.
///
/// opencode has no equivalent, so it gets the prose. That is not a fallback
/// worth apologising for — it works, and it is what the winning opencode
/// candidates on this repository have been driven by all along.
fn pointer(kind: AgentKind, prompt_path: &Path) -> String {
if matches!(kind, AgentKind::Antigravity) {
return format!("@{}", prompt_path.display());
}
format!(
"Read the file at {} and follow every instruction in it exactly. That \
file is your complete task description; this message contains nothing \
else.",
prompt_path.display()
)
}
fn build_command(
spec: &AgentSpec,
seat: &SeatState,
inv: &Invocation<'_>,
prompt_path: &Path,
) -> Result<Plan> {
let mut argv: Vec<String> = Vec::new();
let mut stdin: Option<String> = None;
let delivery = spec.delivery();
let resuming = has_session(spec.kind, seat, inv.sessions);
match spec.kind {
AgentKind::Claude => {
// Claude's own tools have no cwd-confined sandbox - the CLI can
// already `Read` any absolute path magi hands it, an attachment
// outside the repository included - so no extra flag is needed
// here.
argv.push("claude".to_owned());
argv.push("-p".to_owned());
argv.push("--output-format".to_owned());
argv.push("json".to_owned());
if let Some(m) = &spec.model {
argv.push("--model".to_owned());
argv.push(m.clone());
}
if inv.sessions {
let uuid = seat
.claude_session
.as_deref()
.context("claude seat is missing its session uuid")?;
argv.push(if resuming { "--resume" } else { "--session-id" }.to_owned());
argv.push(uuid.to_owned());
}
argv.push("--permission-mode".to_owned());
argv.push("bypassPermissions".to_owned());
if !inv.allow_write {
argv.push("--disallowed-tools".to_owned());
argv.push("Edit,Write,MultiEdit,NotebookEdit".to_owned());
}
}
AgentKind::Opencode => {
// `--auto` below already bypasses every permission, reads of a
// path outside `--dir` included, so an attachment elsewhere
// needs no extra flag.
argv.push("opencode".to_owned());
argv.push("run".to_owned());
argv.push("--format".to_owned());
argv.push("json".to_owned());
argv.push("--dir".to_owned());
argv.push(inv.cwd.to_string_lossy().into_owned());
// `--auto` gates *every* permission, reads included: without it a
// non-interactive opencode cannot even open the prompt file, and
// the seat drops out of the panel with "the user rejected
// permission to use this specific tool call". opencode has no
// read-only mode, so read-only seats rely on the prompt plus the
// fact that judge and reviewer worktrees are disposable — judges'
// are deleted after the tally, reviewers' are reset to the commit
// under review every round.
argv.push("--auto".to_owned());
if let Some(m) = &spec.model {
argv.push("-m".to_owned());
argv.push(m.clone());
}
if resuming {
argv.push("-s".to_owned());
argv.push(
seat.captured_session
.clone()
.expect("has_session checked the id is present"),
);
}
}
AgentKind::Antigravity => {
argv.push("agy".to_owned());
argv.push("--output-format".to_owned());
argv.push("json".to_owned());
// agy's print mode gives up after 5 minutes by default, which is
// far below an implementation node's budget.
argv.push("--print-timeout".to_owned());
argv.push(format!("{}s", inv.timeout.as_secs()));
argv.push("--mode".to_owned());
argv.push(
if inv.allow_write {
"accept-edits"
} else {
"plan"
}
.to_owned(),
);
if inv.allow_write {
argv.push("--dangerously-skip-permissions".to_owned());
}
if let Some(m) = &spec.model {
argv.push("--model".to_owned());
argv.push(m.clone());
}
if resuming {
argv.push("--conversation".to_owned());
argv.push(
seat.captured_session
.clone()
.expect("has_session checked the id is present"),
);
}
// The prompt file lives outside the worktree, so the workspace has
// to be widened to reach it - and so does an attachment's own
// directory, which usually lives right beside it under the
// conversation's `artifacts_dir` (see `chat`/`talk`). "Usually":
// a chat derived from another one (`chat::derived_background`)
// can carry attachment paths that live under the *source*
// conversation's own artifacts dir instead, so each attachment
// outside `inv.artifacts` gets its own `--add-dir` rather than
// assuming one directory covers all of `inv.attachments`.
let mut add_dirs: Vec<String> = Vec::new();
if delivery == Delivery::File || !inv.attachments.is_empty() {
add_dirs.push(inv.artifacts.to_string_lossy().into_owned());
}
for path in inv.attachments {
let Some(parent) = path.parent() else {
continue;
};
if parent.starts_with(inv.artifacts) {
continue;
}
let dir = parent.to_string_lossy().into_owned();
if !add_dirs.contains(&dir) {
add_dirs.push(dir);
}
}
if inv.allow_write {
for path in inv.writable {
let dir = path.to_string_lossy().into_owned();
if !add_dirs.contains(&dir) {
add_dirs.push(dir);
}
}
}
for dir in add_dirs {
argv.push("--add-dir".to_owned());
argv.push(dir);
}
}
AgentKind::Codex => {
// `--sandbox` below governs writes, not reads (see the module
// doc: it is what makes codex the one kind whose *read-only*
// mode is enforced, by refusing edits) - both presets can read
// anywhere the OS lets the process, so an attachment outside
// `cwd` is already reachable without an extra flag.
argv.push("codex".to_owned());
argv.push("exec".to_owned());
argv.push("--json".to_owned());
// The worktrees magi hands out are real checkouts, but a judge's
// is detached and a fixture's may be no repository at all.
argv.push("--skip-git-repo-check".to_owned());
argv.push("-C".to_owned());
argv.push(inv.cwd.to_string_lossy().into_owned());
// Codex is the only kind whose read-only-ness is enforced by the
// CLI rather than by the prompt: a judge or reviewer seat cannot
// write even if it decides to try. Implementers get the workspace.
// The one exception is a conversational talk turn that the
// repository opted into (`unsandboxed` and `allow_write`): it
// drops sandbox and approvals altogether, as claude's
// `bypassPermissions` does, so `magi task add`, `git fetch` and
// the like are not refused. Nothing else ever gets the bypass.
if inv.allow_write && inv.unsandboxed {
argv.push("--dangerously-bypass-approvals-and-sandbox".to_owned());
} else {
argv.push("--sandbox".to_owned());
argv.push(
if inv.allow_write {
"workspace-write"
} else {
"read-only"
}
.to_owned(),
);
// Nothing is watching to approve anything: an unattended seat
// that asks blocks until its timeout kills it.
argv.push("-c".to_owned());
argv.push("approval_policy=\"never\"".to_owned());
if inv.allow_write && !inv.writable.is_empty() {
let roots: Vec<String> = inv
.writable
.iter()
.map(|p| p.to_string_lossy().into_owned())
.collect();
argv.push("-c".to_owned());
argv.push(format!(
"sandbox_workspace_write.writable_roots={}",
serde_json::to_string(&roots).unwrap_or_else(|_| "[]".to_owned())
));
}
}
if let Some(m) = &spec.model {
argv.push("-m".to_owned());
argv.push(m.clone());
}
// `resume` is a subcommand of `exec`, and it rejects the flags
// above when they follow it - so every option is emitted first and
// the subcommand last. Established by hand against codex-cli
// 0.153.4: with the order reversed the CLI exits on
// `unexpected argument '--sandbox'`.
if resuming {
argv.push("resume".to_owned());
argv.push(
seat.captured_session
.clone()
.expect("has_session checked the id is present"),
);
}
}
AgentKind::Omp => {
// `omp` reads the prompt from stdin in print mode (see
// `AgentSpec::delivery`), so the whole instruction arrives without
// an argv length limit - the same reason codex gets stdin.
argv.push("omp".to_owned());
argv.push("-p".to_owned());
argv.push("--mode=json".to_owned());
// `--auto-approve` is required, and is the same trade opencode's
// `--auto` makes: it gates *every* permission, reads included, so
// without it a non-interactive seat cannot even open the prompt
// file magi wrote and drops out of the panel on a permission
// rejection. `omp` has no read-only mode of its own, so a judge or
// reviewer seat rests on the prompt plus the worktree discipline
// (judge worktrees are deleted after the tally, reviewer worktrees
// are reset to the commit under review every round) - never on this
// flag, and never on a bypass flag.
argv.push("--auto-approve".to_owned());
if let Some(m) = &spec.model {
argv.push("--model".to_owned());
argv.push(m.clone());
}
// Established by hand against omp 18.1.19: `-p --mode=json` reports
// the session id on its `"type":"session"` line, and
// `--resume <id>` continues that conversation. `--continue` is
// deliberately not used - it opens a *new* session rather than the
// stored one, which silently loses the seat's memory.
if resuming {
argv.push("--resume".to_owned());
argv.push(
seat.captured_session
.clone()
.expect("has_session checked the id is present"),
);
}
}
AgentKind::Command => {
// The operator's own command line, not one of the roster CLIs -
// there is no flag this function could add on its behalf, so an
// attachment's path has to reach it the same way the prompt
// does, through the substitutions below.
if spec.command.is_empty() {
bail!("agent `{}` has kind = \"command\" but no command", spec.id);
}
let vars: BTreeMap<&str, String> = BTreeMap::from([
("{prompt_file}", prompt_path.to_string_lossy().into_owned()),
("{cwd}", inv.cwd.to_string_lossy().into_owned()),
("{label}", seat.key.clone()),
("{session}", seat.claude_session.clone().unwrap_or_default()),
]);
for raw in &spec.command {
let mut arg = raw.clone();
for (k, v) in &vars {
if arg.contains(k) {
arg = arg.replace(k, v);
}
}
argv.push(arg);
}
}
}
argv.extend(spec.extra_args.iter().cloned());
// `agy` takes the prompt as the value of `-p`, so the flag has to be
// emitted right before whatever the delivery mode produces.
if spec.kind == AgentKind::Antigravity {
argv.push("-p".to_owned());
}
// `codex exec` reads stdin only when its prompt argument is `-`; without
// it the CLI waits on a prompt it will never be given.
if spec.kind == AgentKind::Codex && delivery == Delivery::Stdin {
argv.push("-".to_owned());
}
match delivery {
Delivery::Stdin if spec.kind == AgentKind::Antigravity => {
// agy has no text stdin path; fall back to the pointer file.
argv.push(pointer(spec.kind, prompt_path));
}
Delivery::Stdin => stdin = Some(inv.prompt.to_owned()),
Delivery::Argv => argv.push(inv.prompt.to_owned()),
Delivery::File => argv.push(pointer(spec.kind, prompt_path)),
}
Ok(Plan { argv, stdin })
}
/// What a CLI's stdout yielded.
#[derive(Debug, Default)]
struct Extracted {
text: String,
session: Option<String>,
status: Option<String>,
quota: Option<Quota>,
dropped: Option<Dropped>,
commands: Vec<CommandEvidence>,
context_tokens: Option<u64>,
}
/// Pull the agent's message (and any session id) out of a CLI's stdout.
fn extract(kind: AgentKind, stdout: &str) -> Extracted {
let mut extracted = extract_answer(kind, stdout);
extracted.context_tokens = context_tokens(kind, stdout);
extracted
}
/// `v[key]` as an unsigned integer, or `None` when absent or not a number.
fn uint(v: &serde_json::Value, key: &str) -> Option<u64> {
v.get(key).and_then(serde_json::Value::as_u64)
}
/// The input side of a usage object: `input` plus whatever the cache served
/// or stored, each key read under the names the CLIs use. A missing cache
/// counter is 0 (the CLI simply had no cache traffic to report) but a missing
/// *input* counter makes the whole thing unknown - `None`, never a made-up 0.
fn input_side(usage: &serde_json::Value, input: &[&str], cache: &[&str]) -> Option<u64> {
let first = |keys: &[&str]| keys.iter().find_map(|k| uint(usage, k));
let base = first(input)?;
let cached: u64 = cache.iter().filter_map(|k| uint(usage, k)).sum();
Some(base.saturating_add(cached))
}
/// The context size a CLI reported for its last turn: its input-side token
/// count, which approximates how much of the model's window the conversation
/// occupies. A pure function of stdout, so each CLI's shape is unit-testable.
///
/// Rules, all deliberate:
/// - **Last report wins, never a sum.** A multi-event stream (opencode's
/// `step_finish`, codex's `turn.completed`, omp's `message_end`) reports the
/// input of each model call; the final one is the largest context the
/// conversation reached this turn, and adding them would count the same
/// prefix once per tool round-trip.
/// - **Missing or mistyped usage is `None`**, which the UI shows as "unknown".
/// - **A cumulative total is not a context size, so it is `None`.** claude's
/// result `usage` sums every model call of the turn and, on a resumed
/// session, carries the session's earlier usage too, so even `num_turns == 1`
/// does not make it one call's context. codex's `turn.completed.usage` is the
/// session's running input total, and agy's `usage` aggregates the whole
/// print-mode call (the observed sample exceeds any window). None of the
/// three carries a per-call figure, so each is unknown rather than made up.
/// Only the streams that report each model call (opencode `step_finish`, omp
/// message usage) yield a number.
fn context_tokens(kind: AgentKind, stdout: &str) -> Option<u64> {
use serde_json::Value;
let lines = || {
stdout
.lines()
.filter_map(|l| serde_json::from_str::<Value>(l.trim()).ok())
};
match kind {
AgentKind::Claude => None,
AgentKind::Opencode => lines()
.filter(|v| {
// The event is `step_finish`; its part is `step-finish`.
v.get("type").and_then(|t| t.as_str()) == Some("step_finish")
|| v.get("part")
.and_then(|p| p.get("type"))
.and_then(|t| t.as_str())
== Some("step-finish")
})
.filter_map(|v| {
let tokens = v.get("part")?.get("tokens")?;
let base = uint(tokens, "input")?;
let cache = tokens.get("cache");
let cached = ["read", "write"]
.iter()
.filter_map(|k| cache.and_then(|c| uint(c, k)))
.sum::<u64>();
Some(base.saturating_add(cached))
})
.next_back(),
AgentKind::Antigravity => None,
AgentKind::Codex => None,
AgentKind::Omp => lines()
.flat_map(|v| {
// Same walk as the answer: `agent_end` carries the thread,
// `turn_end` / `message_end` one message each.
match v.get("type").and_then(|t| t.as_str()) {
Some("agent_end") => v
.get("messages")
.and_then(|m| m.as_array())
.cloned()
.unwrap_or_default(),
Some("turn_end") | Some("message_end") => {
v.get("message").cloned().into_iter().collect()
}
_ => Vec::new(),
}
})
.filter(|m| m.get("role").and_then(|r| r.as_str()) == Some("assistant"))
.filter_map(|m| {
input_side(
m.get("usage")?,
&["input", "input_tokens"],
&[
"cacheRead",
"cacheWrite",
"cache_read_tokens",
"cache_write_tokens",
],
)
})
.next_back(),
AgentKind::Command => None,
}
}
/// The answer text of a stored `<stem>.out`, through the same per-CLI
/// extraction a live call uses. Never the raw event stream.
pub(crate) fn answer_text(kind: AgentKind, stdout: &str) -> String {
extract_answer(kind, stdout).text
}
fn extract_answer(kind: AgentKind, stdout: &str) -> Extracted {
match kind {
AgentKind::Claude => {
let Ok(v) = serde_json::from_str::<serde_json::Value>(stdout.trim()) else {
return Extracted {
text: stdout.trim().to_owned(),
..Extracted::default()
};
};
Extracted {
text: v
.get("result")
.and_then(|r| r.as_str())
.unwrap_or_default()
.to_owned(),
session: v
.get("session_id")
.and_then(|s| s.as_str())
.map(str::to_owned),
status: v.get("is_error").and_then(|e| e.as_bool()).map(|e| {
if e {
"error".to_owned()
} else {
"success".to_owned()
}
}),
quota: claude_quota(&v),
// Claude reports a truncated stream as an ordinary error; the
// shape `dropped_stream` keys on is agy's.
dropped: None,
commands: Vec::new(),
context_tokens: None,
}
}
AgentKind::Opencode => {
// A JSONL event stream: text parts concatenated in arrival order.
let mut text = String::new();
let mut session = None;
for line in stdout.lines() {
let Ok(v) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
continue;
};
if session.is_none() {
session = v
.get("sessionID")
.and_then(|s| s.as_str())
.map(str::to_owned);
}
let part = v.get("part").unwrap_or(&serde_json::Value::Null);
if part.get("type").and_then(|t| t.as_str()) == Some("text")
&& let Some(t) = part.get("text").and_then(|t| t.as_str())
{
if !text.is_empty() {
text.push('\n');
}
text.push_str(t);
}
}
Extracted {
text,
session,
status: None,
quota: None,
dropped: None,
commands: Vec::new(),
context_tokens: None,
}
}
AgentKind::Antigravity => {
// agy prints warnings before the JSON object, so parse the last
// line that is one rather than the whole stream.
let obj = stdout
.lines()
.rev()
.find_map(|l| serde_json::from_str::<serde_json::Value>(l.trim()).ok());
let Some(v) = obj else {
return Extracted {
text: stdout.trim().to_owned(),
..Extracted::default()
};
};
Extracted {
text: v
.get("response")
.and_then(|r| r.as_str())
.unwrap_or_default()
.trim()
.to_owned(),
session: v
.get("conversation_id")
.and_then(|s| s.as_str())
.map(str::to_owned),
status: v.get("status").and_then(|s| s.as_str()).map(str::to_owned),
quota: None,
dropped: dropped_stream(&v),
commands: Vec::new(),
context_tokens: None,
}
}
AgentKind::Codex => {
// A JSONL event stream, prefixed on a real machine by tracing
// lines the CLI writes about its own config and skills - so
// non-JSON lines are skipped rather than treated as the answer.
//
// The thread id arrives once, in `thread.started`, and a resumed
// turn reports the same one. The answer is the last
// `item.completed` carrying an `agent_message`: earlier ones are
// the model narrating its way through the tool loop, and taking
// the first would hand the caller a progress note instead of a
// verdict.
//
// A `command_execution` item is different evidence entirely: the
// CLI's own record that it ran a command and what that command
// reported back, kept as `CommandEvidence` — see run
// 20260912-214939-b3bb's artifacts, where these very fields
// (`id`/`command`/`exit_code`/`aggregated_output`) were what
// caught a paired test result magi's own agent prose had missed.
// Only what `item.completed` actually reports: a command still
// running when the turn ended emits no such event at all, and is
// not something this can detect — see [`CommandEvidence`]'s own
// doc for why that must not be guessed at instead.
let mut text = String::new();
let mut session = None;
let mut status = None;
let mut commands = Vec::new();
for line in stdout.lines() {
let Ok(v) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
continue;
};
match v.get("type").and_then(|t| t.as_str()) {
Some("thread.started") => {
session = v
.get("thread_id")
.and_then(|s| s.as_str())
.map(str::to_owned);
}
Some("item.completed") => {
let item = v.get("item").unwrap_or(&serde_json::Value::Null);
match item.get("type").and_then(|t| t.as_str()) {
Some("agent_message") => {
if let Some(t) = item.get("text").and_then(|t| t.as_str()) {
text = t.trim().to_owned();
}
}
Some("command_execution") => {
commands.push(command_evidence(item));
}
_ => {}
}
}
Some("turn.completed") => status = Some("success".to_owned()),
Some("turn.failed") => status = Some("error".to_owned()),
_ => {}
}
}
Extracted {
text,
session,
status,
quota: None,
dropped: None,
commands,
context_tokens: None,
}
}
AgentKind::Omp => {
// A JSONL event stream. The session id arrives once, on the
// `"type":"session"` line that opens the run.
//
// The answer is the *last* non-empty assistant text block anywhere
// in the stream, and neither of the two obvious shortcuts works:
//
// 1. Do not key on `agent_end`. `omp` emits it only for a run that
// quiesces on a message turn; a turn that ends on a tool call
// (`stopReason: "toolUse"`) ends the run with **no `agent_end`
// line at all**, and the answer is in `message_end` / `turn_end`
// instead. Reading only `agent_end` silently discards a complete
// review - which is exactly what the first hand-written wrapper
// did, three times, before this arm existed.
// 2. Do not take the first assistant text. Earlier ones narrate the
// tool loop (sometimes with a single `.`), so the last non-empty
// block is the answer and the one before it is a progress note.
//
// Every line is parsed independently: a non-JSON line (a CLI
// warning, a truncated write) is skipped rather than treated as the
// answer, the same way the codex arm treats its tracing prefix.
let mut text = String::new();
let mut session = None;
for line in stdout.lines() {
let Ok(v) = serde_json::from_str::<serde_json::Value>(line.trim()) else {
continue;
};
if v.get("type").and_then(|t| t.as_str()) == Some("session") {
session = v.get("id").and_then(|s| s.as_str()).map(str::to_owned);
continue;
}
// `agent_end` carries the whole thread; `turn_end` and
// `message_end` each carry one message. Whichever appears, the
// messages are walked the same way.
let messages: Vec<&serde_json::Value> = match v.get("type").and_then(|t| t.as_str())
{
Some("agent_end") => v
.get("messages")
.and_then(|m| m.as_array())
.map(|m| m.iter().collect())
.unwrap_or_default(),
Some("turn_end") | Some("message_end") => {
v.get("message").into_iter().collect()
}
_ => continue,
};
for message in messages {
if message.get("role").and_then(|r| r.as_str()) != Some("assistant") {
continue;
}
let Some(parts) = message.get("content").and_then(|c| c.as_array()) else {
continue;
};
for part in parts {
if part.get("type").and_then(|t| t.as_str()) != Some("text") {
continue;
}
if let Some(t) = part.get("text").and_then(|t| t.as_str())
&& !t.trim().is_empty()
{
text = t.trim().to_owned();
}
}
}
}
Extracted {
text,
session,
status: None,
quota: None,
dropped: None,
commands: Vec::new(),
context_tokens: None,
}
}
AgentKind::Command => {
// A `command` agent may wrap a subscription CLI (a fixture, or a
// thin shim around `claude`). If its output is the claude error
// shape we recognise the quota the same way, so tests and wrappers
// do not need their own detection; anything else is just text.
let parsed = serde_json::from_str::<serde_json::Value>(stdout.trim()).ok();
let quota = parsed.as_ref().and_then(claude_quota);
// A `command` fixture may also stand in for a CLI that hangs up on
// its own stream, which is how that path is tested.
let dropped = parsed.as_ref().and_then(dropped_stream);
Extracted {
text: stdout.trim().to_owned(),
session: None,
status: None,
quota,
dropped,
commands: Vec::new(),
context_tokens: None,
}
}
}
}
/// Build one [`CommandEvidence`] from a Codex `command_execution` item.
///
/// `command` arrives either as a single string or as an argv array,
/// depending on how the CLI shaped the call; both are read rather than
/// assuming one. Missing fields are left at their honest defaults (an empty
/// id/description, `exit_code: None`) rather than guessed at.
fn command_evidence(item: &serde_json::Value) -> CommandEvidence {
let description = match item.get("command") {
Some(serde_json::Value::String(s)) => s.clone(),
Some(serde_json::Value::Array(parts)) => parts
.iter()
.filter_map(|p| p.as_str())
.collect::<Vec<_>>()
.join(" "),
_ => String::new(),
};
let result_summary = item
.get("aggregated_output")
.and_then(|o| o.as_str())
.map(|s| tail_chars(s.trim(), 400))
.unwrap_or_default();
CommandEvidence {
id: item
.get("id")
.and_then(|s| s.as_str())
.unwrap_or_default()
.to_owned(),
description,
exit_code: item
.get("exit_code")
.and_then(serde_json::Value::as_i64)
.map(|e| e as i32),
result_summary,
source: "codex".to_owned(),
}
}
/// The last `max` characters of `s`, cut on a char boundary.
fn tail_chars(s: &str, max: usize) -> String {
let count = s.chars().count();
if count <= max {
return s.to_owned();
}
s.chars().skip(count - max).collect()
}
/// Recognise claude's rate-limit error shape, when it is present.
///
/// Recognises `is_error: true` with a session-limit result or the structured
/// pair `api_error_status: 429` (number or string), `api_error: usage_limit_reached`;
/// every other CLI (and any future shape) returns `None` and is treated as an
/// ordinary failure — the conservative side.
fn claude_quota(v: &serde_json::Value) -> Option<Quota> {
let is_err = v.get("is_error").and_then(|e| e.as_bool()).unwrap_or(false);
if !is_err {
return None;
}
let result = v.get("result").and_then(|r| r.as_str()).unwrap_or("");
let status = v.get("api_error_status");
let usage_limit = (status.and_then(serde_json::Value::as_u64) == Some(429)
|| status.and_then(serde_json::Value::as_str) == Some("429"))
&& v.get("api_error").and_then(serde_json::Value::as_str) == Some("usage_limit_reached");
if !usage_limit && !result.to_lowercase().contains("session limit") {
return None;
}
// "…session limit · resets 4:50am (Asia/Tokyo)". The timezone read is not
// worth parsing exactly; keep the whole phrase after "resets" as free text.
let reset = result
.split("resets ")
.nth(1)
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_owned);
Some(Quota { reset })
}
/// Recognise `agy` running out of quota, from either stream.
///
/// Observed shape while out of quota: stdout carries the JSON object
/// `{"status":"ERROR","response":"","error":"Individual quota reached. … Resets
/// in 1h2m49s."}` and stderr carries `AGY_ERROR: {"status":"RESOURCE_EXHAUSTED",
/// "error_code":429,…}`. Either alone is enough (a mangled stdout must not hide
/// a quota that stderr states plainly), but each is keyed on a *pair* of
/// structured fields: `status: ERROR` with the quota text, or
/// `RESOURCE_EXHAUSTED` together with 429. An error status alone, or a 429
/// alone, stays an ordinary failure - the conservative side, as for
/// [`claude_quota`].
///
/// The reset hint is what follows `Resets ` (e.g. `in 1h2m49s`), trailing full
/// stop dropped.
fn agy_quota(stdout: &str, stderr: &str) -> Option<Quota> {
let from_stdout = stdout
.lines()
.rev()
.find_map(|l| serde_json::from_str::<serde_json::Value>(l.trim()).ok())
.and_then(|v| {
let status = v.get("status").and_then(|s| s.as_str()).unwrap_or("");
let error = v.get("error").and_then(|e| e.as_str()).unwrap_or("");
(status.eq_ignore_ascii_case("error") && error.to_lowercase().contains("quota reached"))
.then(|| error.to_owned())
});
let from_stderr = || {
stderr.lines().find_map(|l| {
let v: serde_json::Value =
serde_json::from_str(l.trim().strip_prefix("AGY_ERROR:")?.trim()).ok()?;
let exhausted = v.get("status").and_then(|s| s.as_str()) == Some("RESOURCE_EXHAUSTED");
let code = v.get("error_code").and_then(serde_json::Value::as_u64) == Some(429);
(exhausted && code).then(|| {
v.get("short_error")
.and_then(|e| e.as_str())
.unwrap_or_default()
.to_owned()
})
})
};
let text = from_stdout.or_else(from_stderr)?;
let reset = text
.split_once("Resets ")
.map(|(_, rest)| rest.trim().trim_end_matches('.').trim())
.filter(|s| !s.is_empty())
.map(str::to_owned);
Some(Quota { reset })
}
/// Recognise a CLI that gave up on its own stream while the agent was working.
///
/// Observed once, verbatim, from `agy` on a candidate that produced nothing:
///
/// ```text
/// {"conversation_id":"36743d06-…","status":"ERROR","response":"",
/// "error":"the connection to the agent was interrupted before the response
/// finished: subscriber fell behind updates, stalled for 5s",
/// "duration_seconds":431.19,"num_turns":1,
/// "usage":{"input_tokens":260113,"output_tokens":14267,
/// "thinking_tokens":9695,"cache_read_tokens":2200925}}
/// ```
///
/// Seven minutes of work and fourteen thousand output tokens, billed, with an
/// empty `response`: the agent did the job and the CLI's own subscriber fell
/// behind and hung up. That is **not** an agent that failed to implement, and
/// counting it as one is how `agy` came to read as 0 wins in 4 entries with
/// five empty candidates - a number that has twice been used to argue the seat
/// out of the roster, and twice been wrong (see `cb6b830`, which reverted the
/// first removal: *"agy does not fail to implement, it fails to report"*).
///
/// The distinction that matters is **billed work with nothing delivered**, so
/// that is what this keys on: an error status, an empty response, and a usage
/// report showing output tokens. Everything else - including an error with no
/// usage at all - returns `None` and stays an ordinary failure, the
/// conservative side, exactly as [`claude_quota`] treats shapes it does not
/// recognise.
///
/// Unlike a quota, this **is** worth re-asking: the work exists in the
/// conversation the CLI just abandoned, and `conversation_id` is right there.
fn dropped_stream(v: &serde_json::Value) -> Option<Dropped> {
let status = v.get("status").and_then(|s| s.as_str()).unwrap_or("");
if !status.eq_ignore_ascii_case("error") {
return None;
}
let response = v.get("response").and_then(|r| r.as_str()).unwrap_or("");
if !response.trim().is_empty() {
// It answered. Whatever the status says, there is something to read.
return None;
}
let produced = v
.get("usage")
.and_then(|u| u.get("output_tokens"))
.and_then(serde_json::Value::as_u64)
.unwrap_or(0);
if produced == 0 {
// An error with nothing produced is just an error.
return None;
}
Some(Dropped {
why: v
.get("error")
.and_then(|e| e.as_str())
.unwrap_or("the CLI ended the stream without delivering its answer")
.trim()
.to_owned(),
output_tokens: produced,
})
}
/// Preflight: which configured agents are not runnable here?
pub fn missing_programs(specs: &[AgentSpec]) -> Vec<String> {
let mut missing = Vec::new();
for s in specs {
let program = match s.kind {
AgentKind::Command => s.command.first().map(String::as_str),
other => other.program(),
};
if let Some(p) = program
&& !crate::config::which(p)
&& !Path::new(p).is_file()
&& !missing.iter().any(|m: &String| m == p)
{
missing.push(p.to_owned());
}
}
missing
}
/// Absolute path of a run's artifact directory.
pub fn artifacts_dir(run_dir: &Path) -> PathBuf {
run_dir.join("artifacts")
}
/// Can this agent's CLI actually be run on this machine?
pub fn installed(spec: &AgentSpec) -> bool {
// A `command` agent has no program of its own to look for - its argv is the
// operator's, and they are the authority on whether it runs.
spec.kind.program().is_none_or(crate::config::which)
}
/// Choose the agent for a seat that stands alone rather than rotating through
/// the roster: [`crate::talk`]'s standing conversation,
/// or anything else that needs one agent picked once
/// rather than a panel filled in.
///
/// `available` is a parameter rather than a call to [`installed`] so the order
/// below is assertable on a machine with none of these CLIs installed, which is
/// every CI runner.
///
/// The order, and why:
///
/// 1. An explicit id always wins, and is an error rather than a fallback when
/// it is unusable. Naming a seat has a reason, and silently substituting a
/// different model would waste whatever that reason was.
/// 2. Otherwise a [`AgentKind::Claude`] seat, ahead of the roster order: it is
/// the only one of the three CLIs magi can address before the first turn
/// (see this module's own doc on session mechanics), which matters most for
/// a conversation that opens with nothing typed yet.
/// 3. Otherwise the first runnable agent in roster order, because the roster
/// order is the operator's own stated preference and magi has nothing
/// better to go on.
pub fn pick(
agents: &[AgentSpec],
want: Option<&str>,
available: &dyn Fn(&AgentSpec) -> bool,
) -> Result<AgentSpec> {
if let Some(id) = want {
let spec = agents
.iter()
.find(|a| a.id == id)
.with_context(|| format!("no agent `{id}` in the roster; it has {}", ids(agents)))?;
if !available(spec) {
bail!(
"agent `{}` needs `{}` on PATH; install it or pass a different \
--agent",
spec.id,
spec.kind.program().unwrap_or("its command")
);
}
return Ok(spec.clone());
}
if agents.is_empty() {
bail!(
"the agent roster is empty, so there is nobody to ask: install one \
of claude, opencode or agy - magi derives a roster from what is on \
PATH - or add an [[agents]] entry to magi.toml."
);
}
if let Some(spec) = agents
.iter()
.find(|a| a.kind == AgentKind::Claude && available(a))
{
return Ok(spec.clone());
}
agents
.iter()
.find(|a| available(a))
.cloned()
.with_context(|| {
let missing = agents
.iter()
.filter_map(|a| a.kind.program())
.collect::<Vec<_>>()
.join(", ");
format!(
"no agent in the roster can be run here: install one of \
{missing}, or add an [[agents]] entry to magi.toml for a CLI \
you do have"
)
})
}
/// Resolve a standalone seat's agent choice into the ordered chain to try.
///
/// Unset (or empty) is [`pick`]'s default order, one entry. Otherwise each
/// named id is kept at its first appearance only - that is what bounds a
/// chain to one attempt per agent - and an id that is not in the roster or
/// not installed is skipped with a warning rather than failing the chain.
/// Errors, naming `role`, only when nothing resolves.
pub fn pick_chain(
agents: &[AgentSpec],
choice: Option<&AgentChoice>,
available: &dyn Fn(&AgentSpec) -> bool,
role: &str,
) -> Result<Vec<AgentSpec>> {
let wanted = choice.map(AgentChoice::ids).unwrap_or_default();
if wanted.is_empty() {
return pick(agents, None, available)
.map(|s| vec![s])
.with_context(|| format!("choose an agent for the {role} role"));
}
let mut chain: Vec<AgentSpec> = Vec::new();
for id in wanted {
if chain.iter().any(|s| s.id == id) {
continue;
}
match pick(agents, Some(id), available) {
Ok(spec) => chain.push(spec),
Err(e) => tracing::warn!("[roles] {role}: skipping `{id}`: {e:#}"),
}
}
if chain.is_empty() {
bail!(
"[roles] {role} names no agent that can run here; the roster has {}",
ids(agents)
);
}
Ok(chain)
}
/// Should a chain move on to its next agent after this call?
///
/// The one place that decides it: an error, a quota hit (judged apart from
/// `usable`, since a CLI can exit 0 with a quota message), or an unusable
/// answer such as a timeout or empty reply.
pub fn chain_advances(outcome: &Result<AgentOutput>) -> bool {
match outcome {
Err(e) if e.is::<crate::repo_lock::ResourceBlocked>() => false,
Err(_) => true,
Ok(out) => output_advances(out),
}
}
/// [`chain_advances`] for a call that did return an output: the `Ok` half,
/// shared with callers that classify the call into their own outcome type
/// first (the graph's fixer chain).
pub fn output_advances(out: &AgentOutput) -> bool {
out.quota_exhausted() || !out.usable()
}
fn ids(agents: &[AgentSpec]) -> String {
if agents.is_empty() {
return "no agents at all".to_owned();
}
agents
.iter()
.map(|a| a.id.clone())
.collect::<Vec<_>>()
.join(", ")
}
#[cfg(test)]
mod tests {
use super::*;
const COMMAND_HELPER_MODE: &str = "MAGI_TEST_COMMAND_HELPER_MODE";
/// Test-only command agent implemented by this test binary itself. Unlike
/// `echo` and `sleep`, it is available wherever the Rust tests run.
fn command_helper(mode: &str) -> AgentSpec {
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,
}
}
#[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,
}
}
fn inv<'a>(cwd: &'a Path, art: &'a Path, allow_write: bool) -> Invocation<'a> {
Invocation {
coordination: None,
cwd,
prompt: "do the thing",
timeout: Duration::from_secs(900),
allow_write,
unsandboxed: false,
sessions: true,
artifacts: art,
stem: "t",
run: "test-run",
node: "test",
cache_dir: None,
attachments: &[],
writable: &[],
}
}
fn plan_for(kind: AgentKind, seat: &SeatState, allow_write: bool) -> Plan {
build_command(
&spec(kind, None),
seat,
&inv(Path::new("."), Path::new("/art"), allow_write),
Path::new("/art/p.md"),
)
.unwrap()
}
#[test]
fn claude_mints_then_resumes_the_same_uuid() {
let mut seat = SeatState::new("judge-1", "a", 7);
let uuid = seat.claude_session.clone().unwrap();
let first = plan_for(AgentKind::Claude, &seat, true);
assert!(first.argv.windows(2).any(|w| w == ["--session-id", &uuid]));
assert!(!first.argv.iter().any(|a| a == "--resume"));
seat.turns = 1;
let second = plan_for(AgentKind::Claude, &seat, true);
assert!(second.argv.windows(2).any(|w| w == ["--resume", &uuid]));
assert!(!second.argv.iter().any(|a| a == "--session-id"));
}
#[test]
fn read_only_seats_cannot_edit() {
let seat = SeatState::new("judge-1", "a", 7);
let claude = plan_for(AgentKind::Claude, &seat, false);
assert!(claude.argv.iter().any(|a| a == "--disallowed-tools"));
assert!(
!plan_for(AgentKind::Claude, &seat, true)
.argv
.iter()
.any(|a| a == "--disallowed-tools")
);
let agy = plan_for(AgentKind::Antigravity, &seat, false);
assert!(agy.argv.windows(2).any(|w| w == ["--mode", "plan"]));
assert!(
!agy.argv
.iter()
.any(|a| a == "--dangerously-skip-permissions")
);
let agy_rw = plan_for(AgentKind::Antigravity, &seat, true);
assert!(
agy_rw
.argv
.windows(2)
.any(|w| w == ["--mode", "accept-edits"])
);
assert!(
agy_rw
.argv
.iter()
.any(|a| a == "--dangerously-skip-permissions")
);
// agy is pointed at its prompt with its own `@<path>` syntax, not with
// prose asking it to read a file. Measured on one trivial task: 17s
// against 73s, because prose costs a tool round-trip before the model
// has even seen its instructions. It is also the form rvpm proved.
let agy_prompt = agy_rw
.argv
.iter()
.position(|a| a == "-p")
.map(|i| agy_rw.argv[i + 1].clone())
.expect("agy takes its prompt with -p");
assert!(
agy_prompt.starts_with('@'),
"agy must get a file reference, got {agy_prompt:?}"
);
assert!(
!agy_prompt.contains("Read the file at"),
"the prose pointer is for CLIs with no file syntax"
);
// opencode is the exception: `--auto` also gates reads, so withholding
// it silently drops the seat out of the panel. Verified against the CLI
// — a read-only judge failed with "the user rejected permission to use
// this specific tool call" while trying to open its own prompt.
for allow_write in [false, true] {
assert!(
plan_for(AgentKind::Opencode, &seat, allow_write)
.argv
.iter()
.any(|a| a == "--auto"),
"opencode needs --auto even to read (allow_write = {allow_write})"
);
}
}
/// The three things about `codex exec` that were established by hand and
/// that a rewrite would silently get wrong.
#[test]
fn codex_gets_extra_writable_roots_only_when_it_may_write() {
let seat = SeatState::new("deputy-x", "a", 7);
let roots = [PathBuf::from("/data/questions")];
let mk = |allow_write: bool| {
let mut i = inv(Path::new("."), Path::new("/art"), allow_write);
i.writable = &roots;
build_command(
&spec(AgentKind::Codex, None),
&seat,
&i,
Path::new("/art/p.md"),
)
.unwrap()
};
let want = "sandbox_workspace_write.writable_roots=[\"/data/questions\"]";
assert!(mk(true).argv.windows(2).any(|w| w == ["-c", want]));
assert!(!mk(false).argv.iter().any(|a| a.contains("writable_roots")));
}
#[test]
fn repository_write_scope_covers_editing_nodes_without_changing_other_roles() {
let mut i = inv(Path::new("."), Path::new("/art"), true);
for node in [
"implement",
"fix",
"graft",
"gate-fix",
"land",
"rebase",
"chat",
] {
i.node = node;
assert!(repository_writer(&i), "{node}");
i.allow_write = false;
assert!(!repository_writer(&i), "{node}");
i.allow_write = true;
}
for node in ["deputy", "judge", "review", "bump", "conduct"] {
i.node = node;
assert!(!repository_writer(&i), "{node}");
}
i.node = "chat";
let roots = [PathBuf::from("/questions")];
i.writable = &roots;
assert!(!repository_writer(&i), "consultation keeps explicit grants");
}
#[test]
fn writable_directories_reach_initial_and_resumed_adapter_commands() {
let roots = [
PathBuf::from("/shared/git"),
PathBuf::from("/state/questions"),
];
for kind in [AgentKind::Codex, AgentKind::Antigravity] {
for resumed in [false, true] {
let mut seat = SeatState::new("impl-A", "a", 7);
if resumed {
seat.turns = 1;
seat.captured_session = Some("session-id".into());
}
for allow_write in [false, true] {
let mut i = inv(Path::new("."), Path::new("/art"), allow_write);
i.writable = &roots;
let plan = build_command(&spec(kind, None), &seat, &i, Path::new("/art/p.md"))
.unwrap();
for root in &roots {
if kind == AgentKind::Antigravity {
assert_eq!(
plan.argv.windows(2).any(|w| {
w[0] == "--add-dir" && w[1] == root.to_str().unwrap()
}),
allow_write,
"resumed={resumed}: {:?}",
plan.argv
);
}
assert_eq!(
plan.argv
.iter()
.any(|arg| arg.contains(root.to_str().unwrap())),
allow_write,
"{kind:?}, resumed={resumed}: {:?}",
plan.argv
);
}
if kind == AgentKind::Codex && resumed {
let at = plan.argv.iter().position(|s| s == "resume").unwrap();
assert!(!plan.argv[at..].iter().any(|s| s == "-c"));
}
}
}
}
}
#[test]
fn codex_talk_turn_bypasses_sandbox_only_when_opted_in() {
const BYPASS: &str = "--dangerously-bypass-approvals-and-sandbox";
let mut seat = SeatState::new("talk", "a", 7);
let build = |seat: &SeatState, allow_write: bool, unsandboxed: bool| {
let mut i = inv(Path::new("."), Path::new("/art"), allow_write);
i.unsandboxed = unsandboxed;
build_command(
&spec(AgentKind::Codex, None),
seat,
&i,
Path::new("/art/p.md"),
)
.unwrap()
};
// (a) both set: bypass, and none of the sandbox flags.
let first = build(&seat, true, true);
assert!(first.argv.iter().any(|a| a == BYPASS));
assert!(!first.argv.iter().any(|a| a == "--sandbox"));
assert!(!first.argv.iter().any(|a| a.starts_with("approval_policy")));
// (b) the flag precedes the `resume` subcommand.
seat.captured_session = Some("thread-1".to_owned());
seat.turns = 1;
let resumed = build(&seat, true, true);
let at = |p: &Plan, x: &str| p.argv.iter().position(|a| a == x).unwrap();
assert!(at(&resumed, BYPASS) < at(&resumed, "resume"));
// (c) without write access the read-only sandbox stays.
let ro = build(&seat, false, true);
assert!(!ro.argv.iter().any(|a| a == BYPASS));
assert!(ro.argv.windows(2).any(|w| w == ["--sandbox", "read-only"]));
}
#[test]
fn codex_is_sandboxed_reads_stdin_and_puts_resume_last() {
let mut seat = SeatState::new("judge-1", "a", 7);
// 1. Read-only is enforced by the CLI, not by the prompt - the only
// roster member for which that is true - and nothing ever asks for
// the bypass.
let ro = plan_for(AgentKind::Codex, &seat, false);
assert!(ro.argv.windows(2).any(|w| w == ["--sandbox", "read-only"]));
let rw = plan_for(AgentKind::Codex, &seat, true);
assert!(
rw.argv
.windows(2)
.any(|w| w == ["--sandbox", "workspace-write"])
);
// `unsandboxed` without a talk turn's opt-in never reaches an unattended
// seat, and on its own it never lifts a read-only seat either.
let mut ro_flagged = inv(Path::new("."), Path::new("/art"), false);
ro_flagged.unsandboxed = true;
let ro_flagged = build_command(
&spec(AgentKind::Codex, None),
&seat,
&ro_flagged,
Path::new("/art/p.md"),
)
.unwrap();
assert!(
ro_flagged
.argv
.windows(2)
.any(|w| w == ["--sandbox", "read-only"])
);
for p in [&ro, &rw, &ro_flagged] {
assert!(
!p.argv
.iter()
.any(|a| a == "--dangerously-bypass-approvals-and-sandbox"),
"the bypass defeats the only enforced read-only mode we have"
);
// Nobody is watching to approve anything.
assert!(
p.argv
.windows(2)
.any(|w| w == ["-c", "approval_policy=\"never\""]),
"an unattended seat that asks for approval blocks until timeout"
);
}
// 2. The prompt arrives on stdin, and `-` is what makes codex read it.
assert_eq!(ro.stdin.as_deref(), Some("do the thing"));
assert_eq!(
ro.argv.last().map(String::as_str),
Some("-"),
"without the `-` argument codex waits for a prompt it never gets"
);
// 3. `resume` is a subcommand and rejects the options above when they
// follow it, so it has to be emitted after all of them - and only
// once the CLI has reported a thread id.
seat.turns = 1;
assert!(!has_session(AgentKind::Codex, &seat, true));
assert!(
!plan_for(AgentKind::Codex, &seat, true)
.argv
.iter()
.any(|a| a == "resume")
);
seat.captured_session = Some("01a07440-4545-7492-85c1-024e3259a90a".to_owned());
let resumed = plan_for(AgentKind::Codex, &seat, true);
let at = resumed
.argv
.iter()
.position(|a| a == "resume")
.expect("resumes by subcommand");
assert_eq!(resumed.argv[at + 1], "01a07440-4545-7492-85c1-024e3259a90a");
assert!(
resumed.argv[..at].iter().any(|a| a == "--sandbox"),
"every option precedes the subcommand"
);
assert_eq!(resumed.argv.last().map(String::as_str), Some("-"));
}
/// The three things about `omp -p --mode=json` that were established by
/// hand against omp 18.1.19 and that a rewrite would silently get wrong.
#[test]
fn omp_reads_stdin_auto_approves_and_resumes_by_id() {
let mut seat = SeatState::new("review-1", "a", 7);
// 1. Print mode plus JSON, and the prompt on stdin: a judging prompt
// carrying three patches is past the Windows argv cap, so argv
// delivery is not an option for every node.
let first = plan_for(AgentKind::Omp, &seat, false);
assert!(first.argv.iter().any(|a| a == "-p"));
assert!(first.argv.iter().any(|a| a == "--mode=json"));
assert_eq!(first.stdin.as_deref(), Some("do the thing"));
assert!(
!first.argv.iter().any(|a| a == "do the thing"),
"the prompt reached argv, where Windows caps it"
);
// 2. `--auto-approve` is required (an unattended seat that stops to ask
// blocks until its node timeout kills it), and it is the *only*
// permission flag: omp has no read-only mode, so the bypass flag
// that would throw away codex's one enforced guarantee must never
// appear here either.
for allow_write in [false, true] {
let p = plan_for(AgentKind::Omp, &seat, allow_write);
assert!(
p.argv.iter().any(|a| a == "--auto-approve"),
"omp needs --auto-approve even to read (allow_write = {allow_write})"
);
assert!(
!p.argv
.iter()
.any(|a| a == "--dangerously-bypass-approvals-and-sandbox"),
"nothing ever asks for the bypass"
);
}
// 3. The id omp reports is the only resume token - magi cannot mint it
// up front, so a seat resumes only once a turn has reported one.
seat.turns = 1;
assert!(!has_session(AgentKind::Omp, &seat, true));
assert!(
!plan_for(AgentKind::Omp, &seat, true)
.argv
.iter()
.any(|a| a == "--resume")
);
seat.captured_session = Some("01a09fe9-4e31-7226-85b3-fda6f46689d5".to_owned());
let resumed = plan_for(AgentKind::Omp, &seat, true);
assert!(
resumed
.argv
.windows(2)
.any(|w| w == ["--resume", "01a09fe9-4e31-7226-85b3-fda6f46689d5"]),
"a captured id is what makes the next turn a resume"
);
// `--continue` opens a *new* session instead of the stored one, which
// would silently drop the seat's memory.
assert!(!resumed.argv.iter().any(|a| a == "--continue"));
// stdin still carries the prompt on a resumed turn.
assert_eq!(resumed.stdin.as_deref(), Some("do the thing"));
}
/// The extraction trap that cost three complete reviews when it was done by
/// hand: a turn that ends on a tool call emits **no** `agent_end` line, so
/// keying on `agent_end` finds nothing and the seat reads as one that
/// produced no answer at all.
#[test]
fn omp_takes_the_answer_without_an_agent_end_line() {
let stream = concat!(
r#"{"type":"session","version":3,"id":"01a09fe9-4e31-7226-85b3-fda6f46689d5","cwd":"C:\\w"}"#,
"\n",
r#"{"type":"agent_start"}"#,
"\n",
r#"{"type":"turn_start"}"#,
"\n",
r#"{"type":"message_update","assistantMessageEvent":{"type":"text_delta","contentIndex":1,"delta":"."}}"#,
"\n",
r#"{"type":"message_end","message":{"role":"assistant","content":[{"type":"thinking","thinking":"checking"},{"type":"text","text":"."}]}}"#,
"\n",
r#"{"type":"turn_end","message":{"role":"assistant","content":[{"type":"thinking","thinking":"done"},{"type":"text","text":"{\"vote\":\"approve\"}"}]}}"#,
"\n",
);
let out = extract(AgentKind::Omp, stream);
assert_eq!(
out.text, "{\"vote\":\"approve\"}",
"the last assistant text block is the answer even with no agent_end"
);
assert_eq!(
out.session.as_deref(),
Some("01a09fe9-4e31-7226-85b3-fda6f46689d5")
);
}
/// A stream that *does* carry `agent_end` walks the whole thread, and the
/// last non-empty assistant text still wins over the tool-loop narration
/// that came before it.
#[test]
fn omp_walks_agent_end_and_ignores_tool_loop_narration() {
let stream = concat!(
r#"{"type":"session","version":3,"id":"s1"}"#,
"\n",
"{\"type\":\"agent_end\",\"messages\":[{\"role\":\"user\",\"content\":[{\"type\":\"text\",\"text\":\"review this\"}]},{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"Looking at the diff…\"}]},{\"role\":\"assistant\",\"content\":[{\"type\":\"thinking\",\"thinking\":\"…\"},{\"type\":\"text\",\"text\":\"## 判定\\n\\n問題ありません。\"}]}]}",
"\n",
);
let out = extract(AgentKind::Omp, stream);
assert_eq!(
out.text, "## 判定\n\n問題ありません。",
"the narration is not the answer, and non-ASCII survives intact"
);
assert_eq!(out.session.as_deref(), Some("s1"));
}
/// A line that is not JSON - a CLI warning, a half-written line - is
/// skipped rather than becoming the answer.
#[test]
fn omp_skips_non_json_lines() {
let stream = concat!(
"Warning: some omp notice\n",
r#"{"type":"session","version":3,"id":"s2"}"#,
"\n",
r#"{"type":"message_end","message":{"role":"assistant","content":[{"type":"text","text":"the answer"}]}}"#,
"\n",
"trailing junk",
"\n",
);
let out = extract(AgentKind::Omp, stream);
assert_eq!(out.text, "the answer");
assert_eq!(out.session.as_deref(), Some("s2"));
}
/// A real `codex exec --json` stream, tracing prefix included.
#[test]
fn codex_takes_the_last_agent_message_and_the_thread_id() {
let stream = concat!(
"2026-09-06T01:05:49.394445Z ERROR codex_models_manager: failed to load models cache\n",
r#"{"type":"thread.started","thread_id":"01a07440-4545-7492-85c1-024e3259a90a"}"#,
"\n",
r#"{"type":"turn.started"}"#,
"\n",
r#"{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"Looking into it."}}"#,
"\n",
r#"{"type":"item.completed","item":{"id":"item_1","type":"command_execution","text":"cargo test"}}"#,
"\n",
r#"{"type":"item.completed","item":{"id":"item_2","type":"agent_message","text":"{\"verdict\": \"ok\"}"}}"#,
"\n",
r#"{"type":"turn.completed","usage":{"input_tokens":17137}}"#,
"\n",
);
let out = extract(AgentKind::Codex, stream);
assert_eq!(
out.text, "{\"verdict\": \"ok\"}",
"the last agent message is the answer; earlier ones narrate"
);
assert_eq!(
out.session.as_deref(),
Some("01a07440-4545-7492-85c1-024e3259a90a")
);
assert_eq!(out.status.as_deref(), Some("success"));
let failed = concat!(
r#"{"type":"thread.started","thread_id":"t1"}"#,
"\n",
r#"{"type":"turn.failed","error":{"message":"nope"}}"#,
"\n",
);
assert_eq!(
extract(AgentKind::Codex, failed).status.as_deref(),
Some("error")
);
}
#[test]
fn captured_sessions_resume_only_once_reported() {
let mut seat = SeatState::new("impl-A", "a", 7);
seat.turns = 1;
for kind in [AgentKind::Opencode, AgentKind::Antigravity] {
assert!(!has_session(kind, &seat, true));
let p = plan_for(kind, &seat, true);
assert!(!p.argv.iter().any(|a| a == "-s" || a == "--conversation"));
}
seat.captured_session = Some("sid".to_owned());
assert!(has_session(AgentKind::Opencode, &seat, true));
assert!(
plan_for(AgentKind::Opencode, &seat, true)
.argv
.windows(2)
.any(|w| w == ["-s", "sid"])
);
assert!(
plan_for(AgentKind::Antigravity, &seat, true)
.argv
.windows(2)
.any(|w| w == ["--conversation", "sid"])
);
}
#[test]
fn sessions_disabled_never_resumes() {
let mut seat = SeatState::new("impl-A", "a", 7);
seat.turns = 3;
seat.captured_session = Some("sid".to_owned());
for kind in [
AgentKind::Claude,
AgentKind::Opencode,
AgentKind::Antigravity,
] {
assert!(!has_session(kind, &seat, false));
}
}
#[test]
fn long_prompts_never_reach_argv_for_file_delivery_clis() {
let seat = SeatState::new("judge-1", "a", 7);
for kind in [AgentKind::Opencode, AgentKind::Antigravity] {
let p = plan_for(kind, &seat, false);
assert!(
p.argv.iter().all(|a| a != "do the thing"),
"{kind:?} put the prompt on the command line"
);
assert!(p.argv.iter().any(|a| a.contains("/art/p.md")));
}
// agy has no text stdin, so its `-p` must always carry something.
let p = plan_for(AgentKind::Antigravity, &seat, false);
let at = p.argv.iter().position(|a| a == "-p").unwrap();
assert!(p.argv.get(at + 1).is_some_and(|v| v.contains("p.md")));
assert!(p.stdin.is_none());
}
#[test]
fn agy_print_timeout_tracks_the_node_budget() {
let seat = SeatState::new("impl-A", "a", 7);
let p = build_command(
&spec(AgentKind::Antigravity, None),
&seat,
&Invocation {
coordination: None,
cwd: Path::new("."),
prompt: "p",
timeout: Duration::from_secs(3600),
allow_write: true,
unsandboxed: false,
sessions: true,
artifacts: Path::new("/art"),
stem: "t",
run: "test-run",
node: "test",
cache_dir: None,
attachments: &[],
writable: &[],
},
Path::new("/art/p.md"),
)
.unwrap();
assert!(p.argv.windows(2).any(|w| w == ["--print-timeout", "3600s"]));
}
/// `--add-dir` is what lets antigravity open a file outside the
/// worktree at all. Today that only happens when the delivery mode is
/// already `File`, but an attachment can arrive on a seat whose delivery
/// is `Stdin` or `Argv` (an explicit `prompt_delivery` override), and the
/// image still lives outside `cwd` - so the flag has to widen for that
/// reason too, independent of how the prompt itself is delivered.
#[test]
fn attachments_widen_antigravitys_add_dir_even_off_file_delivery() {
let mut s = spec(AgentKind::Antigravity, None);
s.prompt_delivery = Some(Delivery::Argv);
let seat = SeatState::new("talk", "a", 7);
let atts = [PathBuf::from("/art/attachments/abc.png")];
let without = build_command(
&s,
&seat,
&Invocation {
attachments: &[],
writable: &[],
..inv(Path::new("."), Path::new("/art"), true)
},
Path::new("/art/p.md"),
)
.unwrap();
assert!(
!without.argv.iter().any(|a| a == "--add-dir"),
"no attachment, no reason to widen the sandbox: {without:?}"
);
let with = build_command(
&s,
&seat,
&Invocation {
attachments: &atts,
writable: &[],
..inv(Path::new("."), Path::new("/art"), true)
},
Path::new("/art/p.md"),
)
.unwrap();
assert!(
with.argv.windows(2).any(|w| w == ["--add-dir", "/art"]),
"an attachment outside cwd must widen the sandbox even off File delivery: {with:?}"
);
}
/// A chat derived from another one (`chat::derived_background`) can pass
/// `turn` attachment paths that live under the *source* conversation's
/// own artifacts dir, not this invocation's `artifacts`. A single
/// `--add-dir` for `inv.artifacts` alone would leave those unreadable, so
/// each attachment directory outside it must get its own grant.
#[test]
fn an_inherited_attachment_outside_this_conversations_artifacts_dir_gets_its_own_add_dir() {
let seat = SeatState::new("plan", "a", 7);
let atts = [
PathBuf::from("/art/attachments/own.png"),
PathBuf::from("/other-chat/attachments/inherited.png"),
];
let p = build_command(
&spec(AgentKind::Antigravity, None),
&seat,
&Invocation {
attachments: &atts,
writable: &[],
..inv(Path::new("."), Path::new("/art"), true)
},
Path::new("/art/p.md"),
)
.unwrap();
assert!(
p.argv.windows(2).any(|w| w == ["--add-dir", "/art"]),
"this conversation's own artifacts dir must still be granted: {p:?}"
);
assert!(
p.argv
.windows(2)
.any(|w| w == ["--add-dir", "/other-chat/attachments"]),
"the inherited attachment's own directory must be granted too: {p:?}"
);
}
#[test]
fn command_agents_get_placeholders_substituted() {
let seat = SeatState::new("impl-A", "a", 7);
let p = plan_for(AgentKind::Command, &seat, true);
assert_eq!(p.argv[0], "echo");
assert_eq!(p.argv[1], "impl-A");
assert_eq!(p.stdin.as_deref(), Some("do the thing"));
}
#[test]
fn claude_rate_limit_is_detected_and_reset_read_when_present() {
// The exact shape observed in the wild (run 20260831-031005-ae94).
let stdout = r#"{"is_error": true, "terminal_reason": "api_error",
"result": "You've hit your session limit · resets 4:50am (Asia/Tokyo)",
"session_id": "b8e928f1-754e-4bd3-86c5-0567763654e3"}"#;
let out = extract(AgentKind::Claude, stdout);
let quota = out.quota.as_ref().expect("rate limit must be detected");
assert_eq!(
quota.reset.as_deref(),
Some("4:50am (Asia/Tokyo)"),
"reset time read from the body"
);
}
#[test]
fn claude_rate_limit_without_a_readable_reset_is_still_detected() {
let out = extract(
AgentKind::Claude,
r#"{"is_error":true,"result":"session limit reached"}"#,
);
let quota = out.quota.expect("rate limit detected without a reset");
assert!(quota.reset.is_none(), "unknown reset is kept as unknown");
}
#[test]
fn claude_structured_usage_limit_requires_the_error_and_exact_pair() {
for kind in [AgentKind::Claude, AgentKind::Command] {
for reply in [
r#"{"is_error":true,"api_error_status":429,"api_error":"usage_limit_reached"}"#,
r#"{"is_error":true,"api_error_status":"429","api_error":"usage_limit_reached"}"#,
] {
assert!(extract(kind, reply).quota.is_some(), "{reply}");
}
for reply in [
r#"{"is_error":false,"api_error_status":429,"api_error":"usage_limit_reached"}"#,
r#"{"is_error":true,"api_error_status":500,"api_error":"usage_limit_reached"}"#,
r#"{"is_error":true,"api_error_status":429,"api_error":"other_error"}"#,
r#"{"is_error":false,"api_error_status":"429","api_error":"usage_limit_reached"}"#,
r#"{"is_error":true,"api_error_status":"500","api_error":"usage_limit_reached"}"#,
r#"{"is_error":true,"api_error_status":"429","api_error":"other_error"}"#,
r#"{"is_error":true,"api_error":"usage_limit_reached"}"#,
] {
assert!(extract(kind, reply).quota.is_none(), "{reply}");
}
}
}
#[test]
fn ordinary_failures_are_never_quota() {
// A normal failed claude call (is_error with a different message).
let claude_fail = extract(
AgentKind::Claude,
r#"{"is_error":true,"result":"account does not exist"}"#,
);
assert!(claude_fail.quota.is_none());
// A command agent that exits 1 with plain text.
let cmd_fail = extract(AgentKind::Command, "boom");
assert!(cmd_fail.quota.is_none());
// A successful call is not quota even if it mentions the phrase.
let success = extract(
AgentKind::Command,
r#"{"is_error":false,"result":"session limit is fine"}"#,
);
assert!(success.quota.is_none());
}
/// Minimal, anonymised shape of run 20260912-214939-b3bb's
/// artifacts/impl-A.out: `command_execution` items reporting a paired
/// test run as `1 passed, 1 failed` twice, while the final
/// `agent_message` nevertheless claimed the target passed. The point of
/// `CommandEvidence` is that this claim and the CLI's own structured
/// record of what actually ran are now two separate things a caller can
/// compare, rather than the prose being the only account available.
#[test]
fn codex_command_execution_events_are_captured_alongside_the_final_message() {
let stream = concat!(
r#"{"type":"thread.started","thread_id":"t1"}"#,
"\n",
r#"{"type":"item.completed","item":{"id":"item49","type":"command_execution","command":["bash","-lc","cargo test --test graph_cached_gate"],"exit_code":1,"aggregated_output":"test result: 1 passed; 1 failed"}}"#,
"\n",
r#"{"type":"item.completed","item":{"id":"item52","type":"command_execution","command":["bash","-lc","cargo test --test graph_cached_gate a_single_test"],"exit_code":0,"aggregated_output":"test result: 1 passed; 0 failed"}}"#,
"\n",
r#"{"type":"item.completed","item":{"id":"item99","type":"agent_message","text":"Both tests in the target pass."}}"#,
"\n",
r#"{"type":"turn.completed"}"#,
"\n",
);
let out = extract(AgentKind::Codex, stream);
assert_eq!(out.text, "Both tests in the target pass.");
assert_eq!(out.commands.len(), 2, "{:?}", out.commands);
let paired = &out.commands[0];
assert_eq!(paired.id, "item49");
assert_eq!(paired.exit_code, Some(1));
assert!(paired.description.contains("graph_cached_gate"));
assert!(paired.result_summary.contains("1 failed"));
let solo = &out.commands[1];
assert_eq!(solo.exit_code, Some(0));
// The structured evidence disagrees with the final prose - exactly
// what a caller must be able to see instead of trusting the message
// alone: the full target never passed in one command.
assert!(
out.commands
.iter()
.any(|c| c.exit_code != Some(0) && c.description.contains("graph_cached_gate")),
"a failed run of the actual target must still be visible: {:?}",
out.commands
);
}
#[test]
fn command_agent_can_carry_the_claude_quota_shape() {
let out = extract(
AgentKind::Command,
r#"{"is_error":true,"result":"You've hit your session limit · resets 1:00am (UTC)"}"#,
);
assert!(
out.quota.is_some(),
"a wrapper emitting the claude shape counts as quota"
);
}
#[test]
fn claude_json_result_is_extracted() {
let out = extract(
AgentKind::Claude,
r#"{"result":"all done","session_id":"abc","is_error":false}"#,
);
assert_eq!(out.text, "all done");
assert_eq!(out.session.as_deref(), Some("abc"));
assert_eq!(out.status.as_deref(), Some("success"));
}
/// The shape behind fix-2 in run 20260912-114326-d3b8, minimal and
/// anonymised: a Claude CLI turn that ended `subtype: success`,
/// `is_error: false`, `terminal_reason: completed`, `stop_reason:
/// end_turn` — every signal this crate reads as a clean CLI turn — while
/// `result` is a progress update, not the report the fixer node needed,
/// and no `FixReport` JSON is anywhere in it.
///
/// `AgentOutput::usable()` (a CLI fact: exit 0, not timed out, non-empty
/// text) must stay true here — that is the honest reading of what the
/// CLI reported — while `verdict::extract_json` on the same text must
/// fail. Conflating the two is exactly the bug this fixture reproduces:
/// `run.json` recorded `fix.failed = "unparsable fix report: the reply
/// contained no JSON object"` and moved straight to the next review round
/// with no report ever recovered from that seat.
#[test]
fn a_clean_cli_turn_is_not_the_same_fact_as_the_nodes_own_work_being_done() {
let stdout = r#"{"type":"result","subtype":"success","is_error":false,"terminal_reason":"completed","stop_reason":"end_turn","result":"I'll pause here until the `cargo make check` background run reports back.","session_id":"11111111-1111-1111-1111-111111111111"}"#;
let out = extract(AgentKind::Claude, stdout);
assert_eq!(out.status.as_deref(), Some("success"));
assert!(out.quota.is_none());
assert!(!out.text.trim().is_empty());
let agent_out = AgentOutput {
text: out.text.clone(),
exit_code: Some(0),
timed_out: false,
duration_ms: 500,
artifacts: Vec::new(),
quota: out.quota,
dropped: out.dropped,
commands: out.commands,
context_tokens: None,
};
assert!(
agent_out.usable(),
"the CLI turn itself ended cleanly and must read as usable"
);
assert!(
crate::verdict::extract_json::<crate::verdict::FixReport>(&agent_out.text).is_err(),
"a clean CLI turn is not proof the node's own report ever arrived"
);
}
#[test]
fn opencode_event_stream_is_concatenated() {
let stream = concat!(
r#"{"type":"step_start","sessionID":"ses_1","part":{"type":"step-start"}}"#,
"\n",
r#"{"type":"text","sessionID":"ses_1","part":{"type":"text","text":"first"}}"#,
"\n",
"garbage line\n",
r#"{"type":"text","sessionID":"ses_1","part":{"type":"text","text":"second"}}"#,
"\n"
);
let out = extract(AgentKind::Opencode, stream);
assert_eq!(out.text, "first\nsecond");
assert_eq!(out.session.as_deref(), Some("ses_1"));
}
#[test]
fn agy_json_survives_a_leading_warning_line() {
let stdout = concat!(
"warning: --mode plan has no effect while slash commands are disabled.\n",
r#"{"conversation_id":"eaf2d00a","status":"SUCCESS","response":"persimmon\n"}"#,
"\n"
);
let out = extract(AgentKind::Antigravity, stdout);
assert_eq!(out.text, "persimmon");
assert_eq!(out.session.as_deref(), Some("eaf2d00a"));
assert_eq!(out.status.as_deref(), Some("SUCCESS"));
}
/// Run 26c7's candidate B, verbatim from `artifacts/impl-B.out`.
///
/// The seat read as an empty candidate. It was seven minutes of work and
/// 14,267 output tokens, billed, that the CLI then declined to hand over.
/// Five such candidates are why `agy` reads as 0 wins in 4 entries, and
/// that number has twice been used to argue the seat out of the roster.
const AGY_DROPPED: &str = concat!(
r#"{"conversation_id":"36743d06-c0b3-4b79-9fa2-23869289d7b6","status":"ERROR","#,
r#""response":"","error":"the connection to the agent was interrupted before "#,
r#"the response finished: subscriber fell behind updates, stalled for 5s","#,
r#""duration_seconds":431.1941803,"num_turns":1,"usage":{"input_tokens":260113,"#,
r#""output_tokens":14267,"thinking_tokens":9695,"cache_read_tokens":2200925,"#,
r#""total_tokens":274380}}"#
);
/// Run 1798's review seat, from `review-1-1.out` / `.err`.
const AGY_QUOTA_OUT: &str = concat!(
r#"{"conversation_id":"323c3b5b-0000","status":"ERROR","response":"","#,
r#""error":"Individual quota reached. Please upgrade your subscription to "#,
r#"increase your limits. Resets in 1h2m49s.","duration_seconds":265.9,"#,
r#""num_turns":2,"usage":{"input_tokens":1000,"output_tokens":50}}"#
);
const AGY_QUOTA_ERR: &str = concat!(
"error: Individual quota reached. Resets in 1h2m49s.\n",
r#"AGY_ERROR: {"short_error":"RESOURCE_EXHAUSTED (code 429): Individual quota "#,
r#"reached.","status":"RESOURCE_EXHAUSTED","error_code":429,"code_kind":"http","#,
r#""retryable":true}"#,
"\n"
);
#[test]
fn agy_out_of_quota_is_a_quota_with_the_reset_hint() {
let both = agy_quota(AGY_QUOTA_OUT, AGY_QUOTA_ERR).expect("both streams");
assert_eq!(both.reset.as_deref(), Some("in 1h2m49s"));
let stdout_only = agy_quota(AGY_QUOTA_OUT, "").expect("stdout alone");
assert_eq!(stdout_only.reset.as_deref(), Some("in 1h2m49s"));
// stderr alone: the hint is not in `short_error`, so none is carried.
let stderr_only = agy_quota("not json", AGY_QUOTA_ERR).expect("stderr alone");
assert!(stderr_only.reset.is_none());
}
#[test]
fn ordinary_agy_failures_are_not_a_quota() {
assert!(agy_quota(AGY_DROPPED, "").is_none());
assert!(agy_quota(r#"{"status":"ERROR","error":"boom"}"#, "").is_none());
assert!(
agy_quota(
"",
r#"AGY_ERROR: {"status":"RESOURCE_EXHAUSTED","error_code":500}"#
)
.is_none()
);
assert!(
agy_quota(
"",
r#"AGY_ERROR: {"status":"UNAVAILABLE","error_code":429}"#
)
.is_none()
);
assert!(agy_quota(r#"{"status":"SUCCESS","response":"ok"}"#, "").is_none());
}
#[test]
fn a_cli_that_hangs_up_on_billed_work_is_not_an_agent_that_produced_nothing() {
let out = extract(AgentKind::Antigravity, AGY_DROPPED);
let dropped = out.dropped.expect("recognised as undelivered work");
assert_eq!(dropped.output_tokens, 14267);
assert!(
dropped.why.contains("subscriber fell behind"),
"the CLI's own words are kept for the record: {}",
dropped.why
);
// And the conversation is still there to resume, which is the whole
// reason this is worth re-asking where a quota is not.
assert_eq!(
out.session.as_deref(),
Some("36743d06-c0b3-4b79-9fa2-23869289d7b6")
);
assert!(out.quota.is_none(), "a dropped stream is not a rate limit");
}
#[test]
fn an_error_with_nothing_produced_stays_an_ordinary_failure() {
// No usage at all: the agent never got going, so there is nothing in
// the conversation to resume and nothing was billed. Treating this as
// undelivered work would buy a second call for no reason.
let bare = r#"{"conversation_id":"c1","status":"ERROR","response":"","error":"boom"}"#;
assert!(extract(AgentKind::Antigravity, bare).dropped.is_none());
// Produced tokens, but it did answer - so there is something to read
// and the status is not our business.
let answered = concat!(
r#"{"conversation_id":"c2","status":"ERROR","response":"here it is","#,
r#""usage":{"output_tokens":10}}"#
);
assert!(extract(AgentKind::Antigravity, answered).dropped.is_none());
// A success is a success.
let ok = concat!(
r#"{"conversation_id":"c3","status":"SUCCESS","response":"done","#,
r#""usage":{"output_tokens":10}}"#
);
assert!(extract(AgentKind::Antigravity, ok).dropped.is_none());
}
#[test]
fn an_undelivered_output_is_not_usable_but_is_worth_asking_again() {
let out = AgentOutput {
text: String::new(),
exit_code: Some(1),
timed_out: false,
duration_ms: 431_194,
artifacts: Vec::new(),
quota: None,
dropped: Some(Dropped {
why: "subscriber fell behind updates".to_owned(),
output_tokens: 14267,
}),
commands: Vec::new(),
context_tokens: None,
};
assert!(!out.usable());
assert!(out.work_undelivered());
// The distinction the retry policy rests on: a quota fails the same way
// until it resets, an abandoned conversation can be picked up.
assert!(!out.quota_exhausted());
}
#[test]
fn non_json_stdout_falls_back_to_raw_text() {
let out = extract(AgentKind::Antigravity, "plain answer\n");
assert_eq!(out.text, "plain answer");
assert!(out.session.is_none());
}
#[tokio::test]
async fn command_agent_round_trip_writes_artifacts() {
let dir = tempfile::tempdir().unwrap();
let art = dir.path().join("artifacts");
let mut seat = SeatState::new("impl-A", "a", 7);
let s = command_helper("reply");
let out = invoke(
&s,
&mut seat,
&Invocation {
coordination: None,
cwd: dir.path(),
prompt: "unused",
timeout: Duration::from_secs(30),
allow_write: true,
unsandboxed: false,
sessions: true,
artifacts: &art,
stem: "impl-A",
run: "test-run",
node: "test",
cache_dir: None,
attachments: &[],
writable: &[],
},
)
.await
.unwrap();
assert!(out.usable(), "{out:?}");
assert!(out.text.contains("hello impl-A"), "{}", out.text);
assert_eq!(seat.turns, 1);
assert!(art.join("impl-A.prompt.md").is_file());
assert!(art.join("impl-A.out").is_file());
}
#[tokio::test]
async fn the_invocation_cache_dir_reaches_the_seat_as_cargo_target_dir() {
// 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,
cwd: dir.path(),
prompt: "unused",
timeout: Duration::from_secs(30),
allow_write: true,
unsandboxed: false,
sessions: true,
artifacts: &dir.path().join("artifacts"),
stem: "cache",
run: "test-run",
node: "test",
cache_dir: Some(&cache),
attachments: &[],
writable: &[],
},
)
.await
.unwrap();
assert!(out.usable(), "{out:?}");
assert!(
out.text.contains(cache.to_string_lossy().as_ref()),
"the seat must see CARGO_TARGET_DIR = the shared cache"
);
}
#[tokio::test]
async fn cache_dir_none_strips_a_cargo_target_dir_inherited_from_this_process() {
// `Command` inherits the parent's environment by default, so
// `cache_dir: None` alone is not the same as a seat never seeing
// `CARGO_TARGET_DIR` - it also has to be true when *this* process
// (standing in for the real magi process, which normally does have
// one set, from its own `[verify]` config) already has the variable
// set. Simulating that is the only way to exercise the inheritance
// path at all.
let previous = std::env::var("CARGO_TARGET_DIR").ok();
// SAFETY: this crate's tests run single-threaded
// (`RUST_TEST_THREADS=1`); see
// `updater::tests::env_kill_switch_semantics` for the same
// reasoning applied to another process-global env var.
unsafe {
std::env::set_var("CARGO_TARGET_DIR", "/should/never/reach/a/read-only/seat");
}
let dir = tempfile::tempdir().unwrap();
let mut seat = SeatState::new("review-1", "a", 7);
let s = command_helper("no-cache");
let result = invoke(
&s,
&mut seat,
&Invocation {
coordination: None,
cwd: dir.path(),
prompt: "unused",
timeout: Duration::from_secs(30),
allow_write: false,
unsandboxed: false,
sessions: true,
artifacts: &dir.path().join("artifacts"),
stem: "no-cache",
run: "test-run",
node: "test",
cache_dir: None,
attachments: &[],
writable: &[],
},
)
.await;
// Restored before any assertion that could panic, so a failure here
// never leaks a bogus `CARGO_TARGET_DIR` into whichever test runs
// next in this same process.
// SAFETY: see above.
unsafe {
match &previous {
Some(v) => std::env::set_var("CARGO_TARGET_DIR", v),
None => std::env::remove_var("CARGO_TARGET_DIR"),
}
}
let out = result.unwrap();
assert!(out.usable(), "{out:?}");
assert!(
out.text.contains("ABSENT"),
"a read-only seat must never inherit the process's own CARGO_TARGET_DIR: {}",
out.text
);
}
#[tokio::test]
async fn a_prompt_larger_than_the_pipe_buffer_does_not_deadlock() {
let dir = tempfile::tempdir().unwrap();
let mut seat = SeatState::new("impl-A", "a", 7);
// The helper never reads stdin, so an inline write_all would block once
// the OS pipe buffer filled — long before the process could be waited on.
let s = command_helper("ignore-stdin");
let big = "x".repeat(1_000_000);
let out = invoke(
&s,
&mut seat,
&Invocation {
coordination: None,
cwd: dir.path(),
prompt: &big,
timeout: Duration::from_secs(60),
allow_write: true,
unsandboxed: false,
sessions: true,
artifacts: &dir.path().join("artifacts"),
stem: "big",
run: "test-run",
node: "test",
cache_dir: None,
attachments: &[],
writable: &[],
},
)
.await
.unwrap();
assert!(out.usable(), "{out:?}");
assert!(out.text.contains("done"), "{}", out.text);
}
#[tokio::test]
async fn 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,
cwd: dir.path(),
prompt: "unused",
timeout: Duration::from_millis(300),
allow_write: true,
unsandboxed: false,
sessions: true,
artifacts: &dir.path().join("artifacts"),
stem: "slow",
run: "test-run",
node: "test",
cache_dir: None,
attachments: &[],
writable: &[],
},
)
.await
.unwrap();
assert!(out.timed_out);
assert!(!out.usable());
}
#[tokio::test]
async fn a_timeout_keeps_what_the_agent_had_already_printed() {
// The old implementation cancelled `wait_with_output`, which dropped
// the buffers it owned, so `<stem>.out` was written empty on every
// timeout. "It printed nothing" and "we discarded what it printed"
// looked identical on disk — and one real hour-long stall was
// diagnosed wrongly twice because of it.
let dir = tempfile::tempdir().unwrap();
let artifacts = dir.path().join("artifacts");
let mut seat = SeatState::new("impl-A", "a", 7);
let s = command_helper("chatty-sleep");
let out = invoke(
&s,
&mut seat,
&Invocation {
coordination: None,
cwd: dir.path(),
prompt: "unused",
// Wide enough to cover process-spawn latency inside a loaded
// parallel test run, not merely the helper's first write. At
// two seconds this passed alone and failed in the full suite,
// which is a dice roll rather than a test.
timeout: Duration::from_secs(10),
allow_write: true,
unsandboxed: false,
sessions: true,
artifacts: &artifacts,
stem: "chatty",
run: "test-run",
node: "test",
cache_dir: None,
attachments: &[],
writable: &[],
},
)
.await
.unwrap();
assert!(out.timed_out, "{out:?}");
assert!(!out.usable(), "a cut-off answer is still not an answer");
let recorded = std::fs::read_to_string(artifacts.join("chatty.out")).unwrap();
assert!(
recorded.contains("i-said-something"),
"the artifact must keep what arrived before the kill, got {recorded:?}"
);
assert!(
out.text.contains("i-said-something"),
"and the graph must be able to see it too, got {:?}",
out.text
);
}
#[test]
fn missing_programs_reports_command_binaries() {
let mut s = spec(AgentKind::Command, None);
s.command = vec!["definitely-not-a-real-binary-xyz".to_owned()];
assert_eq!(
missing_programs(&[s]),
["definitely-not-a-real-binary-xyz".to_owned()]
);
}
fn pick_spec(id: &str, kind: AgentKind) -> AgentSpec {
AgentSpec {
id: id.to_owned(),
kind,
model: None,
command: Vec::new(),
extra_args: Vec::new(),
env: BTreeMap::new(),
prompt_delivery: None,
}
}
/// Availability stub: an agent is runnable unless its id was listed as
/// missing. Keeps the selection tests off `PATH` entirely.
fn without<'a>(missing: &'a [&'a str]) -> impl Fn(&AgentSpec) -> bool + 'a {
move |a: &AgentSpec| !missing.contains(&a.id.as_str())
}
#[test]
fn pick_prefers_the_claude_seat_even_when_it_is_not_first_in_the_roster() {
let agents = [
pick_spec("oc", AgentKind::Opencode),
pick_spec("opus", AgentKind::Claude),
pick_spec("agy", AgentKind::Antigravity),
];
let got = pick(&agents, None, &without(&[])).expect("a pick");
assert_eq!(got.id, "opus");
}
#[test]
fn pick_falls_back_to_the_first_installed_agent_in_roster_order() {
let agents = [
pick_spec("opus", AgentKind::Claude),
pick_spec("oc", AgentKind::Opencode),
pick_spec("agy", AgentKind::Antigravity),
];
let got = pick(&agents, None, &without(&["opus", "oc"])).expect("a pick");
assert_eq!(got.id, "agy");
}
#[test]
fn pick_on_an_empty_roster_says_what_to_install() {
let msg = pick(&[], None, &without(&[]))
.expect_err("nobody to ask")
.to_string();
assert!(msg.contains("roster is empty"), "{msg}");
assert!(msg.contains("claude"), "{msg}");
assert!(msg.contains("magi.toml"), "{msg}");
}
#[test]
fn pick_on_a_roster_with_nothing_installed_names_the_programs_that_are_missing() {
let agents = [
pick_spec("opus", AgentKind::Claude),
pick_spec("oc", AgentKind::Opencode),
];
let err = pick(&agents, None, &without(&["opus", "oc"])).expect_err("nothing runnable");
let msg = format!("{err:#}");
assert!(msg.contains("claude"), "{msg}");
assert!(msg.contains("opencode"), "{msg}");
}
#[test]
fn an_explicitly_named_agent_wins_over_the_claude_preference() {
let agents = [
pick_spec("opus", AgentKind::Claude),
pick_spec("oc", AgentKind::Opencode),
];
let got = pick(&agents, Some("oc"), &without(&[])).expect("a pick");
assert_eq!(got.id, "oc");
}
#[test]
fn an_unknown_agent_id_lists_the_ids_that_do_exist() {
let agents = [
pick_spec("opus", AgentKind::Claude),
pick_spec("oc", AgentKind::Opencode),
];
let msg = pick(&agents, Some("gemini"), &without(&[]))
.expect_err("no such agent")
.to_string();
assert!(msg.contains("gemini"), "{msg}");
assert!(msg.contains("opus, oc"), "{msg}");
}
#[test]
fn an_explicitly_named_agent_that_is_not_installed_is_an_error_not_a_fallback() {
let agents = [
pick_spec("opus", AgentKind::Claude),
pick_spec("oc", AgentKind::Opencode),
];
let msg = pick(&agents, Some("oc"), &without(&["oc"]))
.expect_err("must not silently substitute another model")
.to_string();
assert!(msg.contains("opencode"), "{msg}");
assert!(msg.contains("--agent"), "{msg}");
}
fn named(id: &str) -> AgentSpec {
AgentSpec {
id: id.to_owned(),
..spec(AgentKind::Command, None)
}
}
fn output(text: &str, exit: i32, quota: bool) -> AgentOutput {
AgentOutput {
text: text.to_owned(),
exit_code: Some(exit),
timed_out: false,
duration_ms: 0,
artifacts: Vec::new(),
quota: quota.then_some(Quota { reset: None }),
dropped: None,
commands: Vec::new(),
context_tokens: None,
}
}
#[test]
fn a_chain_keeps_the_written_order_and_a_string_is_a_chain_of_one() {
let agents = [named("a"), named("b"), named("c")];
let all = |_: &AgentSpec| true;
let chain = AgentChoice::Chain(vec!["c".into(), "a".into()]);
let got = pick_chain(&agents, Some(&chain), &all, "synthesizer").unwrap();
assert_eq!(
got.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(),
["c", "a"]
);
let one = AgentChoice::from("b");
let got = pick_chain(&agents, Some(&one), &all, "synthesizer").unwrap();
assert_eq!(got.len(), 1);
assert_eq!(got[0].id, "b");
let got = pick_chain(&agents, None, &all, "synthesizer").unwrap();
assert_eq!(got.len(), 1, "unset keeps pick's default");
assert_eq!(got[0].id, "a");
let empty = AgentChoice::Chain(Vec::new());
assert_eq!(
pick_chain(&agents, Some(&empty), &all, "x").unwrap()[0].id,
"a"
);
}
#[test]
fn a_chain_skips_unknown_and_uninstalled_ids_and_tries_each_once() {
let agents = [named("a"), named("b")];
let not_a = |s: &AgentSpec| s.id != "a";
let chain = AgentChoice::Chain(
["a", "ghost", "b", "b"]
.iter()
.map(|s| (*s).to_owned())
.collect(),
);
let got = pick_chain(&agents, Some(&chain), ¬_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);
}
}