agent_abstraction/run.rs
1//! Spawning an agent and turning its output into events and an outcome.
2//!
3//! Two entry points over the same machinery:
4//! - [`run`] waits and hands back the finished [`Outcome`].
5//! - [`stream`] hands back a [`Run`] that yields [`Event`]s as they arrive, for
6//! a UI that shows work in progress.
7//!
8//! Both read stdout and stderr concurrently. Draining only one would deadlock
9//! the moment the other filled its pipe buffer, which for a chatty agent is a
10//! matter of seconds.
11
12use std::process::Stdio;
13
14use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
15use tokio::process::{Child, Command};
16use tokio::sync::mpsc;
17
18use crate::agent::{Continue, EnvPolicy};
19use crate::error::{Error, Result};
20use crate::event::{Event, MAX_LINE, Parser, Terminal, append_capped};
21use crate::outcome::{Outcome, Stop};
22use crate::proc::{kill_group_by_pid, kill_process_group};
23use crate::request::Request;
24
25/// Read one line, giving up on a line that never ends.
26///
27/// `AsyncBufReadExt::lines` buffers until a newline arrives, so a stream that
28/// emits megabytes without one exhausts memory before any total cap applies.
29/// This reads a bounded amount and, past the limit, returns what it has and
30/// discards the remainder of that line. Returns `None` at end of input.
31async fn read_bounded_line<R>(reader: &mut R, buf: &mut String) -> std::io::Result<Option<bool>>
32where
33 R: tokio::io::AsyncBufRead + Unpin,
34{
35 buf.clear();
36 let mut bytes = Vec::new();
37 let mut truncated = false;
38 loop {
39 let mut byte = [0u8; 1];
40 match reader.read(&mut byte).await? {
41 // End of input: a trailing fragment still counts as a line.
42 0 => {
43 if bytes.is_empty() {
44 return Ok(None);
45 }
46 break;
47 }
48 _ if byte[0] == b'\n' => break,
49 _ => {
50 if bytes.len() < MAX_LINE {
51 bytes.push(byte[0]);
52 } else {
53 // Keep draining to the newline so the pipe does not block,
54 // but stop accumulating.
55 truncated = true;
56 }
57 }
58 }
59 }
60 // Output is not guaranteed to be valid UTF-8, and one bad byte should not
61 // end a run.
62 buf.push_str(&String::from_utf8_lossy(&bytes));
63 Ok(Some(truncated))
64}
65
66/// How many events may queue before the producer waits for the consumer. Deep
67/// enough that a burst of tool events does not stall the agent, shallow enough
68/// that a consumer which stops reading does not grow without bound.
69const EVENT_BUFFER: usize = 256;
70
71/// A run in progress.
72///
73/// Yields events through [`Run::recv`] and settles into an [`Outcome`] through
74/// [`Run::finish`].
75///
76/// **Dropping a `Run` kills the agent.** That is the safe default for the hosts
77/// this crate targets: closing a window or cancelling a request should stop the
78/// work, not leave an agent running invisibly, spending quota and touching
79/// files with nobody watching. Call [`Run::detach`] when background execution is
80/// genuinely what you want.
81///
82/// On Unix, dropping **synchronously signals** the run's process group and then
83/// aborts the driver task. What it cannot do is *wait*: `Drop` cannot await, so
84/// it does not block until the child has exited or its readers have been
85/// joined. Use [`Run::cancel`] when you need to know the tree has actually gone
86/// before continuing, such as before touching the files it was working on. On
87/// Windows only the direct child is signalled.
88#[derive(Debug)]
89pub struct Run {
90 events: mpsc::Receiver<Event>,
91 /// The typed command line, kept so both the plain and redacted views come
92 /// from the same source.
93 typed: Vec<crate::agent::Arg>,
94 /// The child's pid, so `Drop` can tear the group down itself rather than
95 /// depending on an aborted task being polled.
96 pid: Option<u32>,
97 /// Set by the driver once the child has been reaped, so `Drop` never
98 /// signals a pid the OS may since have handed to someone else.
99 reaped: std::sync::Arc<std::sync::atomic::AtomicBool>,
100 /// Dropping or firing this asks the driver to tear down in order. Held as
101 /// an `Option` so `detach` can discard it without signalling.
102 cancel: Option<tokio::sync::oneshot::Sender<()>>,
103 /// `None` only after [`Run::finish`], [`Run::cancel`] or [`Run::detach`]
104 /// has taken ownership, which is what stops `Drop` from aborting a run that
105 /// was already settled deliberately.
106 task: Option<tokio::task::JoinHandle<Result<Outcome>>>,
107 argv: Vec<String>,
108}
109
110impl Run {
111 /// The next event, or `None` once the agent has finished producing them.
112 pub async fn recv(&mut self) -> Option<Event> {
113 self.events.recv().await
114 }
115
116 /// The exact command line that was spawned.
117 ///
118 /// **This contains the prompt and any session id.** Treat it as sensitive:
119 /// logging it verbatim puts user content into your logs. Use
120 /// [`Run::redacted_argv`] for diagnostics.
121 #[must_use]
122 pub fn argv(&self) -> &[String] {
123 &self.argv
124 }
125
126 /// The command line with every non-public value replaced by a placeholder.
127 ///
128 /// Prompts, system prompts, session ids and anything from
129 /// [`crate::Request::unchecked_args`] are removed; flag names are kept so
130 /// the command stays recognisable. Sensitivity is recorded where each
131 /// argument is built rather than inferred from the finished line, so a
132 /// bare positional prompt or an opaque raw argument is covered too.
133 #[must_use]
134 pub fn redacted_argv(&self) -> Vec<String> {
135 redact(&self.typed)
136 }
137
138 /// Wait for the run to finish.
139 ///
140 /// Drains any events still queued, so a caller that only wants the result
141 /// can call this without having consumed the stream.
142 ///
143 /// # Errors
144 /// Whatever the run failed with. See [`Error`].
145 pub async fn finish(mut self) -> Result<Outcome> {
146 // The driver owns teardown from here; `Drop` must not also fire.
147 self.pid = None;
148 while self.events.recv().await.is_some() {}
149 // Taking the handle disarms the `Drop` guard: this run is settling
150 // normally, not being abandoned.
151 let Some(task) = self.task.take() else {
152 unreachable!("the handle is only taken by a consuming method")
153 };
154 match task.await {
155 Ok(result) => result,
156 // The driver task panicked or was cancelled. The process itself
157 // started fine, so this is not a spawn failure and must not claim
158 // to be one.
159 Err(join) => Err(Error::Interrupted {
160 bin: self.argv.first().cloned().unwrap_or_default(),
161 detail: if join.is_panic() {
162 "the driver task panicked".into()
163 } else {
164 "the driver task was cancelled".into()
165 },
166 }),
167 }
168 }
169
170 /// Stop the run and wait until the agent is actually gone.
171 ///
172 /// Cooperative rather than an abort: the driver is asked to stop, signals
173 /// the process group, reaps the child and joins its readers, and only then
174 /// does this return. So when it returns the tree really has exited, which
175 /// matters if the next thing you do touches the files it was working on.
176 ///
177 /// Returns the partial [`Outcome`] if the run happened to finish first,
178 /// otherwise [`Error::Cancelled`].
179 ///
180 /// # Errors
181 /// [`Error::Cancelled`] in the normal case, or whatever the run failed with
182 /// if it failed before the request arrived.
183 pub async fn cancel(mut self) -> Result<Outcome> {
184 // The driver tears down cooperatively and this awaits it, so `Drop`
185 // must not race that with a kill of its own.
186 self.pid = None;
187 // Dropping the sender is itself the signal, so this cannot fail in a
188 // way that leaves the driver waiting.
189 drop(self.cancel.take());
190 let Some(task) = self.task.take() else {
191 unreachable!("the handle is only taken by a consuming method")
192 };
193 match task.await {
194 Ok(result) => result,
195 Err(join) => Err(Error::Interrupted {
196 bin: self.argv.first().cloned().unwrap_or_default(),
197 detail: if join.is_panic() {
198 "the driver task panicked".into()
199 } else {
200 "the driver task was cancelled".into()
201 },
202 }),
203 }
204 }
205
206 /// Let the run continue after this handle goes away.
207 ///
208 /// The opposite of the default. Nothing can observe or stop the agent
209 /// afterwards, so reach for this only when an unsupervised background run
210 /// is genuinely intended.
211 pub fn detach(mut self) {
212 // Disarm `Drop` before it runs, or detaching would immediately kill the
213 // run it exists to keep alive.
214 self.pid = None;
215 // Leak the cancel signal rather than dropping it: a dropped sender is
216 // read by the driver as "stop", which is the opposite of detaching.
217 if let Some(cancel) = self.cancel.take() {
218 std::mem::forget(cancel);
219 }
220 // Dropping the handle without aborting is what detaches a tokio task.
221 drop(self.task.take());
222 }
223}
224
225impl Drop for Run {
226 fn drop(&mut self) {
227 // Abandoned rather than finished, cancelled or detached.
228 //
229 // Kill the group here, directly. Signalling the driver and aborting it
230 // is not enough on its own: that leaves teardown waiting on the runtime
231 // to poll the aborted task so its guard runs, and a dropped `Run` was
232 // observed leaving grandchildren alive and sleeping on Linux while the
233 // same teardown worked from `cancel`. `Drop` cannot await, so it does
234 // the one thing it can do synchronously.
235 if let Some(pid) = self.pid
236 && !self.reaped.load(std::sync::atomic::Ordering::SeqCst)
237 {
238 kill_group_by_pid(pid);
239 }
240 drop(self.cancel.take());
241 if let Some(task) = self.task.take() {
242 task.abort();
243 }
244 }
245}
246
247/// Placeholder substituted for a sensitive argv value.
248const REDACTED: &str = "<redacted>";
249
250/// Render a typed command line for logging, keeping flag names and replacing
251/// every value that is not `Public`.
252///
253/// Derived from the sensitivity recorded where each argument was built, so it
254/// cannot miss a case the way matching on flag names and positions can.
255fn redact(argv: &[crate::agent::Arg]) -> Vec<String> {
256 use crate::agent::Sensitivity;
257
258 argv.iter()
259 .map(|arg| match arg.sensitivity {
260 Sensitivity::Public => arg.value.clone(),
261 _ => REDACTED.to_string(),
262 })
263 .collect()
264}
265
266/// Run `request` to completion, discarding the intermediate events.
267///
268/// # Errors
269/// See [`Error`]; notably [`Error::NotInstalled`], [`Error::Timeout`],
270/// [`Error::RateLimited`] and [`Error::Failed`].
271pub async fn run(request: &Request) -> Result<Outcome> {
272 stream(request)?.finish().await
273}
274
275/// Start `request`, returning a handle that streams its events.
276///
277/// Returns as soon as the child is spawned; the work proceeds on a task.
278///
279/// # Errors
280/// [`Error::NotInstalled`] if the binary is missing, [`Error::Unsupported`] if
281/// the agent cannot honour the request, or [`Error::Spawn`] on an OS failure.
282pub fn stream(request: &Request) -> Result<Run> {
283 // `tokio::spawn` panics outside a runtime. A fallible signature must not
284 // hide that, so the context is checked and reported as an ordinary error.
285 let runtime = tokio::runtime::Handle::try_current().map_err(|_| Error::NoRuntime)?;
286
287 // Written before the argv is built, because the argv has to name it.
288 let schema_file = match (&request.schema, request.agent.caps().schema) {
289 (Some(schema), crate::agent::SchemaSupport::File) => {
290 Some(SchemaFile::write(schema).map_err(|source| Error::Spawn {
291 bin: request.agent.bin().to_string(),
292 source,
293 })?)
294 }
295 _ => None,
296 };
297 let mut request = request.clone();
298 if let Some(file) = &schema_file {
299 request.schema_file = Some(file.0.display().to_string());
300 }
301 let request = &request;
302
303 let plan = request.plan();
304 let typed = request.typed_argv()?;
305 let argv: Vec<String> = typed.iter().map(|a| a.value.clone()).collect();
306
307 let mut command = Command::new(&argv[0]);
308 command
309 .args(&argv[1..])
310 .stdin(if plan.stdin_prompt {
311 Stdio::piped()
312 } else {
313 // Close stdin so an agent that would otherwise wait on it exits
314 // instead of hanging forever with nothing to read.
315 Stdio::null()
316 })
317 .stdout(Stdio::piped())
318 .stderr(Stdio::piped())
319 // Without this a killed run can leave the child alive holding the pipes.
320 .kill_on_drop(true);
321 if let Some(cwd) = &request.cwd {
322 command.current_dir(cwd);
323 }
324 // Narrow the environment first, then apply explicit variables, so an
325 // explicit `env()` always wins over the policy.
326 match &request.env_policy {
327 EnvPolicy::Inherit => {}
328 EnvPolicy::Minimal => {
329 command.env_clear();
330 inherit_named(&mut command, &request.agent.essential_env());
331 }
332 EnvPolicy::Only(names) => {
333 command.env_clear();
334 inherit_named(&mut command, names);
335 }
336 }
337 for (key, value) in &request.env {
338 command.env(key, value);
339 }
340
341 // Put the agent in its own process group so the whole tree can be signalled
342 // together. Killing only the CLI leaves the commands *it* spawned running:
343 // a build, a test run, a server, still holding files and credentials after
344 // the run is supposedly over.
345 // 0 means "make this child its own group leader". `tokio::process::Command`
346 // exposes this directly on unix.
347 #[cfg(unix)]
348 command.process_group(0);
349
350 // Reserve an assigned session id before the child exists. Doing it inside
351 // the driver leaves a window where a spawn that half-succeeds loses the
352 // binding, and this is the id the caller may already be showing in a UI.
353 if let Some(token) = preassigned_token(request) {
354 persist_session(request, &token)?;
355 }
356
357 let child = command.spawn().map_err(|source| {
358 // A missing binary is the common case and deserves an actionable error
359 // with an install hint. Reading it off the spawn avoids resolving PATH
360 // twice, and with it the window where the resolved path is replaced
361 // between the check and the exec.
362 if source.kind() == std::io::ErrorKind::NotFound {
363 Error::NotInstalled {
364 agent: request.agent,
365 bin: plan.bin.clone(),
366 hint: request.agent.install_hint(),
367 }
368 } else {
369 Error::Spawn {
370 bin: plan.bin.clone(),
371 source,
372 }
373 }
374 })?;
375
376 let pid = child.id();
377 let (tx, rx) = mpsc::channel(EVENT_BUFFER);
378 let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel();
379 let reaped = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
380 let reaped_for_task = std::sync::Arc::clone(&reaped);
381 let request = request.clone();
382 let task = runtime.spawn(async move {
383 // Moved in so the file outlives the run and is removed with it.
384 let _schema_file = schema_file;
385 drive(child, request, tx, cancel_rx, reaped_for_task).await
386 });
387 Ok(Run {
388 events: rx,
389 typed,
390 pid,
391 reaped,
392 cancel: Some(cancel_tx),
393 task: Some(task),
394 argv,
395 })
396}
397
398/// Copy the named variables from this process into `command`, skipping any that
399/// are unset so nothing is invented.
400fn inherit_named<S: AsRef<str>>(command: &mut Command, names: &[S]) {
401 for name in names {
402 if let Some(value) = std::env::var_os(name.as_ref()) {
403 command.env(name.as_ref(), value);
404 }
405 }
406}
407
408/// A schema file written for one run, removed when the run ends.
409///
410/// Codex reads its schema from disk, so the file has to outlive the spawn and
411/// not outlive the process. Tying it to a guard means every exit path removes
412/// it, including a cancel or a timeout, without each one remembering.
413struct SchemaFile(std::path::PathBuf);
414
415impl SchemaFile {
416 /// Write `schema` somewhere the agent can read it.
417 fn write(schema: &str) -> std::io::Result<SchemaFile> {
418 use std::io::Write as _;
419 use std::sync::atomic::{AtomicU64, Ordering};
420 static COUNTER: AtomicU64 = AtomicU64::new(0);
421
422 let path = std::env::temp_dir().join(format!(
423 "agent-abstraction-schema-{}-{}.json",
424 std::process::id(),
425 COUNTER.fetch_add(1, Ordering::Relaxed)
426 ));
427 let mut options = std::fs::OpenOptions::new();
428 options.write(true).create_new(true);
429 // A schema can encode what a caller is looking for, so it is no more
430 // public than the prompt.
431 #[cfg(unix)]
432 {
433 use std::os::unix::fs::OpenOptionsExt as _;
434 options.mode(0o600);
435 }
436 options.open(&path)?.write_all(schema.as_bytes())?;
437 Ok(SchemaFile(path))
438 }
439}
440
441impl Drop for SchemaFile {
442 fn drop(&mut self) {
443 let _ = std::fs::remove_file(&self.0);
444 }
445}
446
447/// Owns the child and tears down its whole process group when dropped.
448///
449/// `kill_on_drop` alone is not enough: it kills the CLI, leaving the commands
450/// *it* spawned running. Since aborting the driver task drops this guard, the
451/// same teardown covers cancellation, a dropped [`Run`] and a timeout, without
452/// each path having to remember to do it.
453struct ChildGuard {
454 child: Child,
455 /// Cleared once the child has been reaped, so a pid the OS may since have
456 /// recycled is never signalled.
457 armed: bool,
458}
459
460impl Drop for ChildGuard {
461 fn drop(&mut self) {
462 if self.armed {
463 kill_process_group(&self.child);
464 }
465 }
466}
467
468/// Feed the child, read both its pipes, and assemble the outcome.
469#[allow(
470 clippy::too_many_lines,
471 reason = "one linear lifecycle: feed, read, wait, classify. Splitting it \
472 would thread the child, parser, buffers and cancellation state \
473 through helpers and obscure the ordering that matters, such as \
474 killing the group before reaping."
475)]
476async fn drive(
477 child: Child,
478 request: Request,
479 events: mpsc::Sender<Event>,
480 cancel: tokio::sync::oneshot::Receiver<()>,
481 reaped: std::sync::Arc<std::sync::atomic::AtomicBool>,
482) -> Result<Outcome> {
483 // From here on the child is owned by a guard, so every exit path from this
484 // task, including an abort, takes the process group with it.
485 let mut child = ChildGuard { child, armed: true };
486 let plan = request.plan();
487 let bin = plan.bin.clone();
488
489 // Deliver a piped prompt and close the pipe, or the agent waits on EOF.
490 if plan.stdin_prompt {
491 if let Some(mut stdin) = child.child.stdin.take() {
492 let prompt = request.agent.effective_prompt(&plan);
493 stdin
494 .write_all(prompt.as_bytes())
495 .await
496 .map_err(|source| Error::Spawn {
497 bin: bin.clone(),
498 source,
499 })?;
500 drop(stdin);
501 }
502 }
503
504 // Drain stderr on its own task: a full stderr pipe blocks the child even
505 // while stdout still has room.
506 let stderr = child.child.stderr.take();
507 let stderr_task = tokio::spawn(async move {
508 let mut buf = String::new();
509 if let Some(handle) = stderr {
510 let mut reader = BufReader::new(handle);
511 let mut line = String::new();
512 // Keep draining after the cap is hit: an undrained pipe blocks the
513 // child even though we no longer want the bytes.
514 while let Ok(Some(_)) = read_bounded_line(&mut reader, &mut line).await {
515 append_capped(&mut buf, &line);
516 }
517 }
518 buf
519 });
520
521 let stdout = child.child.stdout.take();
522 let mut parser = Parser::new(request.agent, plan.format);
523 // Raw stdout is retained only as a fallback answer for a run that exited
524 // cleanly without producing a structured one, and as evidence when
525 // classifying a failure. It is capped for the same reason as everything
526 // else here: an agent can stream for hours.
527 let mut raw = String::new();
528 // Tracks the first `Started`, so the binding is written once, and carries a
529 // store failure back out instead of discarding it.
530 let mut bound = false;
531 let mut persist_result: Result<()> = Ok(());
532
533 let read_stdout = async {
534 if let Some(handle) = stdout {
535 let mut reader = BufReader::new(handle);
536 let mut line = String::new();
537 while read_bounded_line(&mut reader, &mut line).await?.is_some() {
538 append_capped(&mut raw, &line);
539 for event in parser.push(&line) {
540 // Bind a printed id the moment it appears rather than at the
541 // end. Codex announces its thread before answering, so a
542 // turn killed mid-answer stays resumable.
543 if let Event::Started { session, .. } = &event
544 && !bound
545 {
546 bound = true;
547 persist_result = persist_session(&request, session);
548 }
549 // A receiver that went away is not a failure: the run should
550 // still finish and produce its outcome.
551 if events.send(event).await.is_err() {
552 break;
553 }
554 }
555 }
556 }
557 Ok::<_, std::io::Error>(())
558 };
559
560 // Race three outcomes: the run finishing, the deadline, and a cancellation
561 // request. Reading and waiting are one future so a child that produces
562 // output forever is still bounded by the timeout.
563 let work = async {
564 read_stdout.await?;
565 child.child.wait().await
566 };
567 // A timeout is optional; `pending()` makes the un-timed case the same shape
568 // rather than duplicating the whole select.
569 let deadline = async {
570 match request.timeout {
571 Some(limit) => tokio::time::sleep(limit).await,
572 None => std::future::pending().await,
573 }
574 };
575
576 let status = tokio::select! {
577 // Biased so a finished run is reported as finished even if a deadline
578 // or cancellation lands in the same tick.
579 biased;
580 result = work => result,
581 () = deadline => {
582 // Order matters: signal the group *before* reaping. Reaping clears
583 // the child's pid, and the group kill needs that pid to target the
584 // group, so the other order silently leaves grandchildren running.
585 let partial = shut_down(&mut child, stderr_task).await;
586 reaped.store(true, std::sync::atomic::Ordering::SeqCst);
587 return Err(Error::Timeout {
588 bin,
589 timeout: request.timeout.unwrap_or_default(),
590 partial: parser.finish().text,
591 })
592 .inspect_err(|_| drop(partial));
593 }
594 _ = cancel => {
595 // Cooperative teardown: the caller is waiting on this, so the tree
596 // is signalled, reaped and joined before returning.
597 shut_down(&mut child, stderr_task).await;
598 reaped.store(true, std::sync::atomic::Ordering::SeqCst);
599 return Err(Error::Cancelled { bin });
600 }
601 }
602 .map_err(|source| Error::Spawn {
603 bin: bin.clone(),
604 source,
605 })?;
606
607 // The child has been reaped, so its pid must not be signalled again, by the
608 // guard here or by `Run::drop` racing this.
609 child.armed = false;
610 reaped.store(true, std::sync::atomic::Ordering::SeqCst);
611
612 drop(events);
613 let stderr = stderr_task.await.unwrap_or_default();
614 let saw_structured = parser.saw_structured_record();
615 let saw_terminal = parser.saw_terminal_record();
616 let terminal = parser.finish();
617 let exit_code = status.code().unwrap_or(-1);
618
619 // Under a structured format, silently handing back raw stdout would turn a
620 // protocol failure into a plausible-looking answer. A run that recognized
621 // nothing, or never reached its terminal record, did not produce a result
622 // this crate can vouch for, so it is reported rather than papered over.
623 let structured = plan.format != crate::Format::Text;
624 if structured && exit_code == 0 {
625 if !saw_structured {
626 return Err(Error::Parse {
627 agent: request.agent,
628 detail: format!(
629 "no recognizable {} records in {} lines of output; the CLI's output shape has probably changed",
630 request.agent,
631 raw.lines().count()
632 ),
633 });
634 }
635 if !saw_terminal {
636 return Err(Error::Parse {
637 agent: request.agent,
638 detail: "the stream ended without its terminal record, so the turn did not complete"
639 .into(),
640 });
641 }
642 }
643
644 // Plain text has no structure to validate: the stream is the answer.
645 let mut terminal = terminal;
646 if terminal.text.is_empty() && !structured {
647 terminal.text = raw.trim().to_string();
648 }
649
650 // A provider refusal is not always an exit code. Claude can report a
651 // blocking `rate_limit_event` and still exit 0, and the crate promises that
652 // quota refusals surface as `Error::RateLimited`, so the terminal state is
653 // checked regardless of how the process exited.
654 let quota_blocked = terminal
655 .rate_limit
656 .as_ref()
657 .is_some_and(crate::outcome::RateLimit::is_blocking);
658 // An unauthenticated Claude run exits 0 and reports the problem in its
659 // result text, so checking only the exit code would hand back a successful
660 // Outcome whose answer is "Please run /login".
661 //
662 // Read from stderr and the agent's own prose rather than the raw stream, for
663 // the reason `classify` does the same with quota wording: a phrase hunted
664 // through structured output matches ids and field names, not statements.
665 let unauthenticated = looks_unauthenticated(&terminal.text) || looks_unauthenticated(&stderr);
666 // The agent saying its turn failed is as much a failure as a non-zero exit,
667 // and Claude reports an unknown model exactly this way: exit 0, `is_error`
668 // true, and the explanation where the answer would be.
669 let turn_failed = terminal.stop == Stop::Error;
670 if exit_code != 0 || quota_blocked || unauthenticated || turn_failed {
671 return Err(classify_run(
672 request.agent,
673 &bin,
674 exit_code,
675 &stderr,
676 &raw,
677 &terminal,
678 ));
679 }
680
681 // A fork lands on a *new* id the agent only reveals at the end, so the name
682 // has to be repointed once the run settles. Everything else was bound above.
683 persist_result?;
684 // Resolved before the terminal is consumed by the Outcome below.
685 let structured = terminal.structured.clone().or_else(|| {
686 request
687 .schema
688 .as_ref()
689 .and_then(|_| serde_json::from_str(&terminal.text).ok())
690 });
691 if let Some(token) = &terminal.session
692 && !bound
693 {
694 persist_session(&request, token)?;
695 }
696 Ok(Outcome {
697 agent: request.agent,
698 session: terminal.session,
699 text: terminal.text,
700 usage: terminal.usage,
701 stop: terminal.stop,
702 rate_limit: terminal.rate_limit,
703 exit_code,
704 stderr,
705 unparsed: terminal.unparsed,
706 first_unparsed: terminal.first_unparsed,
707 // Claude reports the conforming value separately; Codex returns it as
708 // the answer text, so that is parsed only when a schema was asked for.
709 // Prose is never reinterpreted as data.
710 structured,
711 })
712}
713
714/// Kill the process group, reap the child, and join the stderr reader.
715///
716/// The orderly teardown both cancellation and timeout share. Returns whatever
717/// stderr had been captured, so a caller can still report why a run was stopped.
718async fn shut_down(child: &mut ChildGuard, stderr_task: tokio::task::JoinHandle<String>) -> String {
719 kill_process_group(&child.child);
720 // Reap, so the caller is not left with a zombie once this returns.
721 let _ = child.child.kill().await;
722 child.armed = false;
723 // The pipes are closed now that the child is gone, so this finishes
724 // promptly rather than hanging the cancellation.
725 stderr_task.await.unwrap_or_default()
726}
727
728/// Turn a failure into the most specific error available, agent included so an
729/// auth failure can carry the right login command.
730fn classify_run(
731 agent: crate::Agent,
732 bin: &str,
733 code: i32,
734 stderr: &str,
735 stdout: &str,
736 terminal: &Terminal,
737) -> Error {
738 // Checked before quota and before a plain failure: a login problem is the
739 // most specific reading of the output, and the only one a user can act on
740 // directly.
741 for source in [terminal.text.as_str(), stderr, stdout] {
742 if looks_unauthenticated(source) {
743 return Error::NotAuthenticated {
744 agent,
745 bin: bin.to_string(),
746 message: first_meaningful_line(source).unwrap_or_default(),
747 hint: agent.login_hint(),
748 };
749 }
750 }
751 classify(agent, bin, code, stderr, stdout, terminal)
752}
753
754/// Whether text is an agent saying it has no usable credentials.
755///
756/// Narrow on purpose. Mislabelling an ordinary failure as an auth problem sends
757/// someone to re-login over something unrelated, so these are phrases the CLIs
758/// actually emit rather than every string containing "auth".
759fn looks_unauthenticated(text: &str) -> bool {
760 const PHRASES: &[&str] = &[
761 // Claude, verified: an unauthenticated run answers exactly this.
762 "not logged in",
763 "please run /login",
764 // Copilot, verified: it exits 1 with plain text, and none of the other
765 // phrases here appear in it. Its wording shares no vocabulary with the
766 // other two, which is why this had to be observed rather than guessed.
767 "no authentication information",
768 "invalid api key",
769 "authentication_error",
770 "unauthorized",
771 "not authenticated",
772 "no credentials",
773 "credentials not found",
774 "please log in",
775 ];
776 let lower = text.to_ascii_lowercase();
777 PHRASES.iter().any(|needle| lower.contains(needle)) || mentions_status(&lower, "401")
778}
779
780/// Whether `code` appears as a standalone token rather than inside a longer run
781/// of characters.
782///
783/// `401` was previously matched as a bare substring, which made any Copilot
784/// failure an auth failure whenever one of the UUIDs it prints happened to
785/// contain those three digits: `"id":"1b0b1401-cb86-..."` was enough. That is
786/// not rare, since a run emits several ids, so the misdiagnosis was
787/// intermittent and told someone to re-login over an unrelated failure.
788///
789/// A status code is a word. Requiring non-alphanumeric neighbours keeps
790/// `HTTP 401` and `(status 401)` while rejecting every hex blob, and a UUID
791/// cannot produce a standalone `401` at all because its groups are four, eight
792/// or twelve characters long.
793fn mentions_status(haystack: &str, code: &str) -> bool {
794 haystack.match_indices(code).any(|(at, _)| {
795 let before = haystack[..at].chars().next_back();
796 let after = haystack[at + code.len()..].chars().next();
797 let free = |c: Option<char>| c.is_none_or(|c| !c.is_alphanumeric());
798 free(before) && free(after)
799 })
800}
801
802/// Turn a non-zero exit into the most specific error available.
803fn classify(
804 agent: crate::Agent,
805 bin: &str,
806 code: i32,
807 stderr: &str,
808 stdout: &str,
809 terminal: &Terminal,
810) -> Error {
811 let quota_signalled = terminal
812 .rate_limit
813 .as_ref()
814 .is_some_and(crate::outcome::RateLimit::is_blocking);
815 // Scanning the *raw* stream for quota wording is a false-positive machine:
816 // under `stream-json` Claude prints a `rate_limit_event` record on every
817 // run, including one whose status is `allowed`, so the substring
818 // `rate_limit` is present in perfectly healthy output. Where the stream
819 // parsed, the parsed signal and the agent's own prose decide; the raw scan
820 // is only the fallback for output that produced neither.
821 let prose = match (&terminal.error_message, terminal.text.as_str()) {
822 (Some(message), text) => format!("{message}\n{text}"),
823 (None, text) if !text.is_empty() => text.to_string(),
824 _ => stdout.to_string(),
825 };
826 if quota_signalled || looks_rate_limited(stderr) || looks_rate_limited(&prose) {
827 return Error::RateLimited {
828 bin: bin.to_string(),
829 message: first_meaningful_line(stderr)
830 .or_else(|| first_meaningful_line(&prose))
831 .unwrap_or_else(|| "usage limit reached".to_string()),
832 };
833 }
834 // A rejected flag is not a failed request, it is this crate and the CLI
835 // disagreeing about what the CLI accepts. Naming that is the difference
836 // between "the run failed" and "your codex is a different version".
837 if let Some(detail) = rejected_flag(stderr).or_else(|| rejected_flag(stdout)) {
838 return Error::FlagRejected {
839 bin: bin.to_string(),
840 detail,
841 };
842 }
843 // Checked before the generic failure but after quota and a rejected flag,
844 // which are more specific readings of the same output.
845 if terminal.stop == Stop::Error {
846 return Error::AgentError {
847 agent,
848 bin: bin.to_string(),
849 status: terminal.error_status,
850 // Codex reports the reason apart from the answer; Claude puts it
851 // where the answer would be.
852 message: terminal
853 .error_message
854 .clone()
855 .or_else(|| first_meaningful_line(&terminal.text))
856 .or_else(|| first_meaningful_line(stderr))
857 .unwrap_or_else(|| "the agent reported a failure without explaining it".into()),
858 };
859 }
860
861 Error::Failed {
862 bin: bin.to_string(),
863 code,
864 // Fall back to stdout when stderr explains nothing. Codex reports a
865 // rejected schema as an `{"type":"error"}` event on *stdout* while
866 // stderr carries only "Reading additional input from stdin...", so
867 // reporting stderr alone describes the failure as a status message.
868 stderr: first_meaningful_line(stderr)
869 .filter(|line| looks_explanatory(line))
870 .or_else(|| first_meaningful_line(stdout))
871 .or_else(|| first_meaningful_line(stderr))
872 .unwrap_or_default(),
873 }
874}
875
876/// Whether a line plausibly explains a failure rather than narrating progress.
877fn looks_explanatory(line: &str) -> bool {
878 const NOISE: &[&str] = &[
879 "reading additional input",
880 "reading prompt",
881 "waiting",
882 "connecting",
883 "loading",
884 ];
885 let lower = line.to_ascii_lowercase();
886 !NOISE.iter().any(|noise| lower.contains(noise))
887}
888
889/// The CLI's complaint, if it refused an argument.
890///
891/// The phrasings are clap's and commander's, which is what all three CLIs are
892/// built on. Matched narrowly: a false positive would relabel a genuine failure
893/// as a version problem and send someone chasing the wrong thing.
894fn rejected_flag(text: &str) -> Option<String> {
895 const REJECTIONS: &[&str] = &[
896 "unexpected argument",
897 "unknown option",
898 "unrecognized option",
899 "unknown flag",
900 "invalid option",
901 "unexpected option",
902 ];
903 let lower = text.to_ascii_lowercase();
904 REJECTIONS
905 .iter()
906 .any(|needle| lower.contains(needle))
907 .then(|| first_meaningful_line(text).unwrap_or_default())
908}
909
910/// Whether text carries a provider quota refusal.
911///
912/// Deliberately a small set of unambiguous phrases: a false positive here would
913/// relabel an ordinary failure as a quota problem and send a caller into a
914/// pointless backoff.
915fn looks_rate_limited(text: &str) -> bool {
916 let lower = text.to_ascii_lowercase();
917 [
918 "rate limit",
919 "rate_limit",
920 "usage limit",
921 "quota exceeded",
922 "too many requests",
923 "429",
924 ]
925 .iter()
926 .any(|needle| lower.contains(needle))
927}
928
929/// The most useful line of a CLI's output for an error message.
930///
931/// Not simply the first non-blank one. CLIs open with progress and status
932/// chatter, so the first line is often "Reading additional input from stdin..."
933/// while the actual cause is further down. That turns a report into a
934/// misdirection: it looks like an explanation and is not one.
935///
936/// So a line that looks like an error wins, and the first non-blank line is the
937/// fallback when nothing does.
938fn first_meaningful_line(text: &str) -> Option<String> {
939 const ERROR_MARKERS: &[&str] = &[
940 "error",
941 "failed",
942 "fatal",
943 "panic",
944 "denied",
945 "invalid",
946 "unexpected",
947 "cannot",
948 "unable",
949 ];
950 let lines: Vec<&str> = text
951 .lines()
952 .map(str::trim)
953 .filter(|line| !line.is_empty())
954 .collect();
955
956 lines
957 .iter()
958 .find(|line| {
959 let lower = line.to_ascii_lowercase();
960 ERROR_MARKERS.iter().any(|marker| lower.contains(marker))
961 })
962 .or_else(|| lines.first())
963 .map(|line| (*line).to_string())
964}
965
966/// Write the session binding back, reporting any store failure.
967///
968/// Called as soon as an id is known rather than only on a clean exit. Waiting
969/// for success would lose the binding for exactly the runs where continuity
970/// matters most: a timeout, a crash, or a cancelled turn.
971fn persist_session(request: &Request, token: &str) -> Result<()> {
972 let Some(binding) = &request.binding else {
973 return Ok(());
974 };
975 binding
976 .store
977 .bind(request.agent, &binding.project, &binding.name, token)
978 .map(|_| ())
979}
980
981/// The id this run is already known by before it starts, if any.
982///
983/// Only a caller-assigned id qualifies: a printed id does not exist yet. This
984/// is what makes an assigned session survive a run that never finishes.
985fn preassigned_token(request: &Request) -> Option<String> {
986 match &request.plan().cont {
987 Continue::NewWith(id) => Some(id.clone()),
988 _ => None,
989 }
990}
991
992/// Reported by an agent that exited cleanly but said nothing useful.
993impl Outcome {
994 /// Whether the agent produced any answer at all.
995 #[must_use]
996 pub fn is_empty(&self) -> bool {
997 self.text.trim().is_empty() && self.stop == Stop::Completed
998 }
999}
1000
1001#[cfg(test)]
1002mod tests {
1003 use super::*;
1004 use crate::agent::Agent;
1005
1006 #[test]
1007 fn quota_phrases_are_recognized_and_ordinary_errors_are_not() {
1008 assert!(looks_rate_limited("Error: rate limit exceeded"));
1009 assert!(looks_rate_limited("HTTP 429 Too Many Requests"));
1010 assert!(looks_rate_limited("You have hit your usage limit"));
1011 // A plain failure must not be mistaken for a quota problem.
1012 assert!(!looks_rate_limited("error: no such file or directory"));
1013 assert!(!looks_rate_limited("model not found"));
1014 }
1015
1016 #[test]
1017 fn a_blocking_rate_limit_event_classifies_as_rate_limited() {
1018 let terminal = Terminal {
1019 rate_limit: Some(crate::outcome::RateLimit {
1020 status: "rejected".into(),
1021 window: Some("five_hour".into()),
1022 resets_at: None,
1023 }),
1024 ..Terminal::default()
1025 };
1026 assert!(matches!(
1027 classify(Agent::Claude, "claude", 1, "", "", &terminal),
1028 Error::RateLimited { .. }
1029 ));
1030 }
1031
1032 #[test]
1033 fn an_allowed_rate_limit_event_is_not_a_failure_cause() {
1034 let terminal = Terminal {
1035 rate_limit: Some(crate::outcome::RateLimit {
1036 status: "allowed".into(),
1037 window: None,
1038 resets_at: None,
1039 }),
1040 ..Terminal::default()
1041 };
1042 assert!(matches!(
1043 classify(Agent::Claude, "claude", 1, "boom", "", &terminal),
1044 Error::Failed { .. }
1045 ));
1046 }
1047
1048 /// The exact shape that made a Copilot run look unauthenticated: a UUID
1049 /// carrying the digits 401. Copilot prints several ids per run, so this
1050 /// misfired intermittently and told the user to re-login over a failure
1051 /// that had nothing to do with credentials.
1052 #[test]
1053 fn an_id_containing_401_is_not_an_auth_failure() {
1054 let line = r#"{"type":"session.mcp_server_status_changed","id":"1b0b1401-cb86-4276-9874-e84b94c96499"}"#;
1055 assert!(
1056 !looks_unauthenticated(line),
1057 "a hex blob is not a status code"
1058 );
1059 }
1060
1061 /// The needle still has to work where it was meant to. A status code is a
1062 /// word, and these are the forms an agent actually prints.
1063 #[test]
1064 fn a_real_401_is_still_recognized() {
1065 for text in [
1066 "HTTP 401",
1067 "request failed (status 401)",
1068 "401: unauthorized",
1069 "got a 401 from the API",
1070 ] {
1071 assert!(looks_unauthenticated(text), "should match: {text}");
1072 }
1073 }
1074
1075 /// Neighbouring digits mean it is part of some longer number, not a status.
1076 #[test]
1077 fn digits_around_401_keep_it_from_matching() {
1078 for text in ["error 4010", "code 1401", "seq 24019"] {
1079 assert!(!looks_unauthenticated(text), "should not match: {text}");
1080 }
1081 }
1082
1083 /// Verbatim from a healthy claude 2.1.205 run. Every `stream-json` run
1084 /// carries this record, and its status is `allowed`: nothing is refused.
1085 /// Scanning the raw stream for `rate_limit` matched it anyway, so any
1086 /// Claude failure was reported as a quota refusal, sending a caller to back
1087 /// off when the real cause was something they could fix.
1088 #[test]
1089 fn a_healthy_rate_limit_heartbeat_is_not_a_refusal() {
1090 let stdout = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"allowed","resetsAt":1785331800,"rateLimitType":"five_hour","overageStatus":"rejected","isUsingOverage":false}}"#;
1091 let terminal = Terminal {
1092 stop: Stop::Error,
1093 error_status: Some(404),
1094 text: "There's an issue with the selected model (bogus-model-xyz).".into(),
1095 rate_limit: Some(crate::outcome::RateLimit {
1096 status: "allowed".into(),
1097 window: Some("five_hour".into()),
1098 resets_at: Some(1_785_331_800),
1099 }),
1100 ..Terminal::default()
1101 };
1102 let err = classify_run(Agent::Claude, "claude", 0, "", stdout, &terminal);
1103 assert!(
1104 matches!(err, Error::AgentError { .. }),
1105 "the heartbeat must not mask the real cause: {err:?}"
1106 );
1107 }
1108
1109 /// The counterpart: a refusal the parser did read must still be one, even
1110 /// though it arrives with the same zero exit code.
1111 #[test]
1112 fn a_rejected_quota_signal_is_still_a_refusal() {
1113 let terminal = Terminal {
1114 rate_limit: Some(crate::outcome::RateLimit {
1115 status: "rejected".into(),
1116 window: Some("five_hour".into()),
1117 resets_at: None,
1118 }),
1119 ..Terminal::default()
1120 };
1121 assert!(matches!(
1122 classify_run(Agent::Claude, "claude", 0, "", "", &terminal),
1123 Error::RateLimited { .. }
1124 ));
1125 }
1126
1127 /// Verbatim from a real run with an unknown model. Claude exits **0** with
1128 /// `subtype: "success"` while `is_error` is true and the explanation sits
1129 /// where the answer would be, so a caller checking only `Result::is_ok`
1130 /// renders "There's an issue with the selected model" as the answer.
1131 #[test]
1132 fn a_failed_turn_is_an_error_even_though_the_process_exited_cleanly() {
1133 let terminal = Terminal {
1134 stop: Stop::Error,
1135 error_status: Some(404),
1136 text: "There's an issue with the selected model (bogus-model-xyz). \
1137 It may not exist or you may not have access to it."
1138 .into(),
1139 ..Terminal::default()
1140 };
1141 let err = classify_run(Agent::Claude, "claude", 0, "", "", &terminal);
1142 let Error::AgentError {
1143 agent,
1144 status,
1145 message,
1146 ..
1147 } = &err
1148 else {
1149 panic!("expected AgentError, got {err:?}")
1150 };
1151 assert_eq!(*agent, Agent::Claude);
1152 assert_eq!(*status, Some(404), "the provider status must survive");
1153 assert!(message.contains("selected model"), "{message}");
1154 }
1155
1156 /// A quota refusal and a missing login are more specific readings of the
1157 /// same shape, so they must not be swallowed by the general case.
1158 #[test]
1159 fn a_failed_turn_does_not_mask_a_more_specific_cause() {
1160 let auth = Terminal {
1161 stop: Stop::Error,
1162 text: "Not logged in · Please run /login".into(),
1163 ..Terminal::default()
1164 };
1165 assert!(
1166 classify_run(Agent::Claude, "claude", 0, "", "", &auth).is_auth_failure(),
1167 "an unauthenticated failed turn must stay an auth failure"
1168 );
1169
1170 let quota = Terminal {
1171 stop: Stop::Error,
1172 rate_limit: Some(crate::outcome::RateLimit {
1173 status: "rejected".into(),
1174 window: None,
1175 resets_at: None,
1176 }),
1177 ..Terminal::default()
1178 };
1179 assert!(
1180 matches!(
1181 classify_run(Agent::Claude, "claude", 0, "", "", "a),
1182 Error::RateLimited { .. }
1183 ),
1184 "a quota-blocked failed turn must stay a rate limit"
1185 );
1186 }
1187
1188 /// Verified against the real CLI: with `USER` withheld, claude answers
1189 /// "Not logged in · Please run /login" and exits **0**. Checking only the
1190 /// exit code hands back a successful Outcome whose answer is a login
1191 /// prompt.
1192 #[test]
1193 fn an_unauthenticated_run_is_named_even_though_it_exits_zero() {
1194 let terminal = Terminal {
1195 text: "Not logged in · Please run /login".into(),
1196 ..Terminal::default()
1197 };
1198 let err = classify_run(Agent::Claude, "claude", 0, "", "", &terminal);
1199 let Error::NotAuthenticated { agent, hint, .. } = &err else {
1200 panic!("expected NotAuthenticated, got {err:?}")
1201 };
1202 assert_eq!(*agent, Agent::Claude);
1203 assert!(hint.contains("/login"), "{hint}");
1204 assert!(err.is_auth_failure());
1205 }
1206
1207 /// Verbatim from an unauthenticated Copilot run, captured by pointing it at
1208 /// an empty HOME. Its wording shares no phrase with Claude's or Codex's, so
1209 /// before this was observed the phrase list did not match it at all and a
1210 /// missing Copilot login was reported as a generic failure.
1211 #[test]
1212 fn copilots_own_unauthenticated_wording_is_recognized() {
1213 let stderr = "Error: No authentication information found.\n\n\
1214 Copilot can be authenticated with GitHub using an OAuth Token or a \
1215 Fine-Grained Personal Access Token.\n\n\
1216 To authenticate, you can use any of the following methods:\n\
1217 \u{2022} Start 'copilot' and run the '/login' command\n\
1218 \u{2022} Set the COPILOT_GITHUB_TOKEN, GH_TOKEN, or GITHUB_TOKEN \
1219 environment variable";
1220 let err = classify_run(
1221 Agent::Copilot,
1222 "copilot",
1223 1,
1224 stderr,
1225 "",
1226 &Terminal::default(),
1227 );
1228 let Error::NotAuthenticated { agent, hint, .. } = &err else {
1229 panic!("expected NotAuthenticated, got {err:?}")
1230 };
1231 assert_eq!(*agent, Agent::Copilot);
1232 assert!(hint.contains("copilot login"), "{hint}");
1233 }
1234
1235 /// Each agent's hint has to name its own login route, since they differ:
1236 /// Codex and Copilot have `login` subcommands, Claude does not.
1237 #[test]
1238 fn every_agent_offers_its_own_login_route() {
1239 for (agent, expected) in [
1240 (Agent::Claude, "setup-token"),
1241 (Agent::Codex, "codex login"),
1242 (Agent::Copilot, "copilot login"),
1243 ] {
1244 let err = classify_run(
1245 agent,
1246 agent.bin(),
1247 1,
1248 "error: unauthorized",
1249 "",
1250 &Terminal::default(),
1251 );
1252 let Error::NotAuthenticated { hint, .. } = &err else {
1253 panic!("{agent}: expected NotAuthenticated, got {err:?}")
1254 };
1255 assert!(hint.contains(expected), "{agent}: {hint}");
1256 }
1257 }
1258
1259 /// Auth is the most specific reading, so it wins over a generic failure,
1260 /// but must not swallow unrelated errors.
1261 #[test]
1262 fn ordinary_failures_are_not_mistaken_for_auth_problems() {
1263 for stderr in [
1264 "error: no such file or directory",
1265 "model not found",
1266 "rate limit exceeded",
1267 "error: unexpected argument '--sandbox' found",
1268 ] {
1269 let err = classify_run(Agent::Codex, "codex", 1, stderr, "", &Terminal::default());
1270 assert!(
1271 !err.is_auth_failure(),
1272 "{stderr:?} was misread as an auth failure: {err:?}"
1273 );
1274 }
1275 }
1276
1277 /// The exact failure that cost a round of debugging: `codex exec resume`
1278 /// rejects `--sandbox`, which `Error::Failed` reported as a generic
1279 /// non-zero exit naming a flag rather than a version mismatch.
1280 #[test]
1281 fn a_rejected_flag_is_named_as_a_version_mismatch() {
1282 let err = classify(
1283 Agent::Codex,
1284 "codex",
1285 2,
1286 "error: unexpected argument '--sandbox' found",
1287 "",
1288 &Terminal::default(),
1289 );
1290 let Error::FlagRejected { bin, detail } = err else {
1291 panic!("expected FlagRejected, got {err:?}")
1292 };
1293 assert_eq!(bin, "codex");
1294 assert!(detail.contains("--sandbox"), "{detail}");
1295 }
1296
1297 #[test]
1298 fn ordinary_failures_are_not_mistaken_for_version_drift() {
1299 for stderr in [
1300 "error: no such file or directory",
1301 "model not found",
1302 "permission denied",
1303 ] {
1304 assert!(
1305 matches!(
1306 classify(Agent::Codex, "codex", 1, stderr, "", &Terminal::default()),
1307 Error::Failed { .. }
1308 ),
1309 "{stderr:?} should stay a plain failure"
1310 );
1311 }
1312 }
1313
1314 /// Real output from a failing codex run: the first line is status, the
1315 /// cause is below it. Reporting the first line looks like an explanation
1316 /// while pointing at the wrong thing.
1317 #[test]
1318 fn a_status_line_does_not_masquerade_as_the_cause() {
1319 let stderr = "Reading additional input from stdin...\n\
1320 error: invalid value 'nope' for '--sandbox <SANDBOX_MODE>'";
1321 let err = classify_run(Agent::Codex, "codex", 1, stderr, "", &Terminal::default());
1322 let Error::Failed {
1323 stderr: reported, ..
1324 } = err
1325 else {
1326 panic!("expected Failed, got {err:?}")
1327 };
1328 assert!(reported.contains("invalid value"), "reported {reported:?}");
1329 }
1330
1331 /// Codex reports a rejected schema as a JSON error event on **stdout**
1332 /// while stderr carries only a status line. Reporting stderr alone
1333 /// described the failure as "Reading additional input from stdin...",
1334 /// which is not what went wrong.
1335 #[test]
1336 fn a_cause_on_stdout_is_reported_when_stderr_only_narrates() {
1337 let stdout = r#"{"type":"error","message":"invalid_json_schema: 'additionalProperties' is required to be supplied and to be false."}"#;
1338 let err = classify_run(
1339 Agent::Codex,
1340 "codex",
1341 1,
1342 "Reading additional input from stdin...",
1343 stdout,
1344 &Terminal::default(),
1345 );
1346 let Error::Failed {
1347 stderr: reported, ..
1348 } = err
1349 else {
1350 panic!("expected Failed, got {err:?}")
1351 };
1352 assert!(
1353 reported.contains("additionalProperties"),
1354 "reported {reported:?}, which explains nothing"
1355 );
1356 }
1357
1358 #[test]
1359 fn failures_report_the_first_useful_line() {
1360 let err = classify(
1361 Agent::Claude,
1362 "claude",
1363 2,
1364 "\n\n real problem \nstack",
1365 "",
1366 &Terminal::default(),
1367 );
1368 let Error::Failed { code, stderr, .. } = err else {
1369 panic!("expected a plain failure")
1370 };
1371 assert_eq!(code, 2);
1372 assert_eq!(stderr, "real problem");
1373 }
1374
1375 /// Prompts and session ids ride the argv, and `Run::argv` invites logging
1376 /// it. The redacted form must keep the shape while dropping the content.
1377 #[test]
1378 fn redaction_removes_prompts_and_session_ids_but_keeps_flags() {
1379 let request = crate::Request::new(Agent::Claude, "my secret prompt")
1380 .system("secret system")
1381 .session_id("11111111-2222-3333-4444-555555555555");
1382 let safe = redact(&request.typed_argv().unwrap());
1383
1384 for secret in [
1385 "my secret prompt",
1386 "secret system",
1387 "11111111-2222-3333-4444-555555555555",
1388 ] {
1389 assert!(
1390 !safe.iter().any(|a| a.contains(secret)),
1391 "{secret:?} survived redaction: {safe:?}"
1392 );
1393 }
1394 // Still recognisable as the same command.
1395 assert_eq!(safe[0], "claude");
1396 assert!(safe.contains(&"--permission-mode".to_string()));
1397 assert!(safe.contains(&"--session-id".to_string()));
1398 }
1399
1400 #[test]
1401 fn codex_trailing_prompt_is_redacted_even_without_a_flag() {
1402 let request = crate::Request::new(Agent::Codex, "my secret prompt");
1403 let safe = redact(&request.typed_argv().unwrap());
1404 assert_eq!(safe.last().unwrap(), REDACTED);
1405 assert_eq!(safe[1], "exec", "the subcommand must survive");
1406 }
1407
1408 /// Redaction must cover the two shapes positional guesswork misses: Codex's
1409 /// bare trailing prompt, and raw arguments whose contents are unknowable.
1410 #[test]
1411 fn redaction_covers_positional_prompts_and_unchecked_arguments() {
1412 let request = crate::Request::new(Agent::Codex, "my secret prompt")
1413 .unchecked_args(["-c", "api_key=hunter2"]);
1414 let safe = redact(&request.typed_argv().unwrap());
1415 assert!(!safe.iter().any(|a| a.contains("my secret prompt")));
1416 assert!(
1417 !safe.iter().any(|a| a.contains("hunter2")),
1418 "unchecked arguments may hold secrets: {safe:?}"
1419 );
1420 assert_eq!(safe[1], "exec", "the subcommand must survive");
1421 }
1422
1423 /// A resume id is a capability: it continues someone's conversation.
1424 #[test]
1425 fn redaction_covers_the_codex_positional_resume_id() {
1426 let request = crate::Request::new(Agent::Codex, "hi").resume("thread-secret-9");
1427 let safe = redact(&request.typed_argv().unwrap());
1428 assert!(
1429 !safe.iter().any(|a| a.contains("thread-secret-9")),
1430 "{safe:?}"
1431 );
1432 assert!(safe.contains(&"resume".to_string()));
1433 }
1434
1435 /// `stream` is synchronous but spawns a task. Outside a runtime that would
1436 /// panic, which a `Result`-returning function must not do.
1437 #[test]
1438 fn stream_outside_a_runtime_errors_instead_of_panicking() {
1439 let err = stream(&crate::Request::new(Agent::Claude, "hi")).unwrap_err();
1440 assert!(matches!(err, Error::NoRuntime), "got {err:?}");
1441 }
1442
1443 #[tokio::test]
1444 async fn a_missing_binary_names_the_install_command() {
1445 let request = Request::new(Agent::Claude, "hi").bin("definitely-not-a-real-binary-xyz");
1446 let err = run(&request).await.unwrap_err();
1447 let Error::NotInstalled { hint, agent, .. } = err else {
1448 panic!("expected NotInstalled, got {err:?}")
1449 };
1450 assert_eq!(agent, Agent::Claude);
1451 assert!(hint.contains("claude-code"));
1452 }
1453
1454 #[test]
1455 fn transient_errors_are_distinguished_from_permanent_ones() {
1456 assert!(
1457 Error::RateLimited {
1458 bin: "claude".into(),
1459 message: String::new()
1460 }
1461 .is_transient()
1462 );
1463 assert!(
1464 !Error::NotInstalled {
1465 agent: Agent::Claude,
1466 bin: "claude".into(),
1467 hint: ""
1468 }
1469 .is_transient()
1470 );
1471 }
1472}