cli_agents/adapters/
mod.rs

1mod claude;
2mod codex;
3mod gemini;
4
5pub use claude::ClaudeAdapter;
6pub use codex::CodexAdapter;
7pub use gemini::GeminiAdapter;
8
9use crate::error::{Error, Result};
10use crate::events::StreamEvent;
11use crate::types::{CliName, RunOptions, RunResult};
12use std::collections::HashMap;
13use tokio::io::{AsyncBufReadExt, BufReader};
14use tokio::process::Command;
15use tracing::{debug, warn};
16
17/// Trait implemented by each CLI adapter.
18pub trait CliAdapter: Send + Sync {
19    fn name(&self) -> CliName;
20
21    fn run(
22        &self,
23        opts: &RunOptions,
24        emit: &(dyn Fn(StreamEvent) + Send + Sync),
25        cancel: tokio_util::sync::CancellationToken,
26    ) -> impl std::future::Future<Output = crate::error::Result<RunResult>> + Send;
27}
28
29/// Get the adapter for a given CLI.
30pub(crate) fn get_adapter(cli: CliName) -> Box<dyn CliAdapterBoxed> {
31    match cli {
32        CliName::Claude => Box::new(ClaudeAdapter),
33        CliName::Codex => Box::new(CodexAdapter),
34        CliName::Gemini => Box::new(GeminiAdapter),
35    }
36}
37
38/// Object-safe version of [`CliAdapter`] for dynamic dispatch.
39///
40/// Needed because `CliAdapter::run` uses RPITIT (`impl Future`), which makes
41/// the trait non-object-safe. This wrapper boxes the future for `dyn` dispatch.
42/// The blanket impl below bridges the two automatically.
43#[allow(dead_code)]
44pub(crate) trait CliAdapterBoxed: Send + Sync {
45    fn name(&self) -> CliName;
46
47    fn run_boxed<'a>(
48        &'a self,
49        opts: &'a RunOptions,
50        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
51        cancel: tokio_util::sync::CancellationToken,
52    ) -> std::pin::Pin<
53        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
54    >;
55}
56
57impl<T: CliAdapter> CliAdapterBoxed for T {
58    fn name(&self) -> CliName {
59        CliAdapter::name(self)
60    }
61
62    fn run_boxed<'a>(
63        &'a self,
64        opts: &'a RunOptions,
65        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
66        cancel: tokio_util::sync::CancellationToken,
67    ) -> std::pin::Pin<
68        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
69    > {
70        Box::pin(self.run(opts, emit, cancel))
71    }
72}
73
74// ── Shared subprocess infrastructure ──
75
76/// Outcome of a spawned CLI process.
77pub(crate) enum SpawnOutcome {
78    /// Process reached a terminal state on its own (exited, or was signalled).
79    Done {
80        /// The process's exit status, or `None` when a SIGNAL ended it instead.
81        ///
82        /// A signalled process genuinely has no exit code. Substituting one
83        /// makes an out-of-memory kill indistinguishable from an agent that
84        /// cleanly decided it had failed, and [`RunResult::exit_code`] is an
85        /// `Option` precisely so a caller can tell the two apart.
86        exit_code: Option<i32>,
87        /// The signal that terminated the process, when one did. Unix only;
88        /// always `None` elsewhere.
89        signal: Option<i32>,
90        stderr: Option<String>,
91    },
92    /// Process was cancelled via the cancellation token.
93    Cancelled,
94}
95
96/// Parameters for [`spawn_and_stream`].
97pub(crate) struct SpawnParams<'a> {
98    pub cli_label: &'a str,
99    pub binary: &'a str,
100    pub args: &'a [String],
101    pub extra_env: &'a HashMap<String, String>,
102    /// Keys to remove from the inherited parent env before applying `extra_env`.
103    /// Used to prevent leaks like `ANTHROPIC_API_KEY` overriding subscription auth.
104    pub strip_env: &'a [&'static str],
105    pub cwd: &'a str,
106    pub max_bytes: usize,
107    pub cancel: &'a tokio_util::sync::CancellationToken,
108}
109
110/// Spawn a CLI subprocess and stream its stdout line-by-line.
111///
112/// Handles the boilerplate shared across all adapters: process spawning,
113/// stdout buffering with size limits, stderr collection, and cancellation.
114/// Does **not** clone the parent process environment — `Command` inherits it
115/// automatically; only `extra_env` entries are added.
116pub(crate) async fn spawn_and_stream(
117    params: SpawnParams<'_>,
118    mut on_line: impl FnMut(&str) + Send,
119) -> Result<SpawnOutcome> {
120    let SpawnParams {
121        cli_label,
122        binary,
123        args,
124        extra_env,
125        strip_env,
126        cwd,
127        max_bytes,
128        cancel,
129    } = params;
130    debug!(cli = cli_label, binary = %binary, args = ?args, "spawning CLI");
131
132    let mut cmd = Command::new(binary);
133    cmd.args(args);
134    for key in strip_env {
135        cmd.env_remove(key);
136    }
137    cmd.envs(extra_env)
138        .current_dir(cwd)
139        .stdin(std::process::Stdio::null())
140        .stdout(std::process::Stdio::piped())
141        .stderr(std::process::Stdio::piped())
142        .kill_on_drop(true);
143
144    #[cfg(unix)]
145    {
146        unsafe {
147            cmd.pre_exec(|| {
148                if libc::setpgid(0, 0) != 0 {
149                    return Err(std::io::Error::last_os_error());
150                }
151                Ok(())
152            });
153        }
154    }
155
156    let mut child = cmd
157        .spawn()
158        .map_err(|e| Error::Process(format!("failed to spawn {cli_label}: {e}")))?;
159
160    let child_pid = child.id();
161
162    let stdout = child.stdout.take().expect("stdout piped");
163    let stderr = child.stderr.take().expect("stderr piped");
164
165    let stderr_handle = tokio::spawn(async move {
166        let mut reader = BufReader::new(stderr);
167        let mut buf = String::new();
168        while reader.read_line(&mut buf).await.unwrap_or(0) > 0 {}
169        buf
170    });
171
172    let mut reader = BufReader::new(stdout);
173    let mut line = String::new();
174    let mut total_bytes: usize = 0;
175
176    loop {
177        line.clear();
178        tokio::select! {
179            result = reader.read_line(&mut line) => {
180                match result {
181                    Ok(0) => break,
182                    Ok(n) => {
183                        total_bytes += n;
184                        if total_bytes > max_bytes {
185                            warn!(cli = cli_label, total_bytes, max_bytes, "output exceeded max buffer size");
186                            kill_process_group(&mut child, child_pid).await;
187                            return Err(Error::Process(format!(
188                                "output exceeded max buffer size ({max_bytes} bytes)"
189                            )));
190                        }
191                        on_line(line.trim());
192                    }
193                    Err(e) => {
194                        warn!(cli = cli_label, error = %e, "error reading stdout");
195                        break;
196                    }
197                }
198            }
199            _ = cancel.cancelled() => {
200                kill_process_group(&mut child, child_pid).await;
201                return Ok(SpawnOutcome::Cancelled);
202            }
203        }
204    }
205
206    let status = child.wait().await.map_err(Error::Io)?;
207    // `code()` is `None` for a signalled process. This used to be
208    // `.unwrap_or(1)`, which reported a SIGKILL as a clean `exit 1` and left
209    // callers with no way to recover the difference — the exact ambiguity that
210    // sent a downstream app hunting for an error message a killed process never
211    // wrote. Report what actually happened and let the caller decide.
212    let exit_code = status.code();
213    #[cfg(unix)]
214    let signal = std::os::unix::process::ExitStatusExt::signal(&status);
215    #[cfg(not(unix))]
216    let signal: Option<i32> = None;
217    let stderr_text = stderr_handle.await.unwrap_or_default();
218
219    Ok(SpawnOutcome::Done {
220        exit_code,
221        signal,
222        stderr: if stderr_text.is_empty() {
223            None
224        } else {
225            Some(stderr_text)
226        },
227    })
228}
229
230/// A sentence for a process that died without writing one.
231///
232/// A signalled CLI usually produces NO stderr and no result event — there was
233/// no chance to. Without this the only fact reaching the user is a number, and
234/// the most common case by far (the OS reclaiming memory) reads as an
235/// unexplained failure.
236pub(crate) fn describe_signal(signal: Option<i32>) -> Option<String> {
237    let sig = signal?;
238    Some(match sig {
239        2 => "The agent was interrupted (SIGINT).".to_string(),
240        6 => "The agent aborted (SIGABRT).".to_string(),
241        9 => "The agent was killed (SIGKILL), most often by the system reclaiming memory."
242            .to_string(),
243        11 => "The agent crashed (SIGSEGV).".to_string(),
244        15 => "The agent was terminated (SIGTERM).".to_string(),
245        other => format!("The agent was terminated by signal {other}."),
246    })
247}
248
249/// Extract a user-friendly error message from CLI stderr.
250/// When an agent fails with no text output, this provides something
251/// meaningful to show the user instead of a blank response.
252pub(crate) fn extract_error_message(stderr: Option<&str>) -> Option<String> {
253    let stderr = stderr?;
254    // Find the most informative error line.
255    let msg = stderr
256        .lines()
257        .filter(|l| !l.is_empty())
258        .find(|l| {
259            let lower = l.to_lowercase();
260            lower.contains("error")
261                || lower.contains("limit")
262                || lower.contains("failed")
263                || lower.contains("denied")
264                || lower.contains("unauthorized")
265        })
266        .or_else(|| stderr.lines().rfind(|l| !l.is_empty()));
267    msg.map(|s| s.trim().to_string())
268}
269
270async fn kill_process_group(child: &mut tokio::process::Child, pid: Option<u32>) {
271    #[cfg(unix)]
272    {
273        if let Some(pid) = pid {
274            unsafe {
275                libc::killpg(pid as libc::pid_t, libc::SIGKILL);
276            }
277        }
278    }
279    let _ = child.kill().await;
280}
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285
286    /// A process that ends by SIGNAL has no exit code, and says so.
287    ///
288    /// THE REGRESSION THIS PINS. `spawn_and_stream` used to finish with
289    /// `status.code().unwrap_or(1)`, so a killed process was reported as a
290    /// clean `exit 1`. Downstream that is unrecoverable: an out-of-memory kill
291    /// and an agent that decided it had failed become the same fact, and a
292    /// consumer looking for the reason searches a stderr the process never got
293    /// to write. `sh -c 'kill -9 $$'` reproduces it without timing games — the
294    /// shell signals itself, so the outcome is deterministic.
295    #[cfg(unix)]
296    #[tokio::test]
297    async fn a_signalled_process_reports_the_signal_not_a_fabricated_exit_code() {
298        let args = vec!["-c".to_string(), "kill -9 $$".to_string()];
299        let cancel = tokio_util::sync::CancellationToken::new();
300        let outcome = spawn_and_stream(
301            SpawnParams {
302                cli_label: "test",
303                binary: "sh",
304                args: &args,
305                extra_env: &HashMap::new(),
306                strip_env: &[],
307                cwd: ".",
308                max_bytes: 1024,
309                cancel: &cancel,
310            },
311            |_| {},
312        )
313        .await
314        .expect("spawn should succeed");
315
316        match outcome {
317            SpawnOutcome::Done {
318                exit_code, signal, ..
319            } => {
320                assert_eq!(exit_code, None, "a signalled process has no exit code");
321                assert_eq!(signal, Some(9), "SIGKILL should be reported as itself");
322            }
323            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
324        }
325    }
326
327    /// An ordinary non-zero exit still reports its code — the change above must
328    /// not turn every failure into `None`.
329    #[tokio::test]
330    async fn a_normal_exit_still_reports_its_code() {
331        let args = vec!["-c".to_string(), "exit 3".to_string()];
332        let cancel = tokio_util::sync::CancellationToken::new();
333        let outcome = spawn_and_stream(
334            SpawnParams {
335                cli_label: "test",
336                binary: "sh",
337                args: &args,
338                extra_env: &HashMap::new(),
339                strip_env: &[],
340                cwd: ".",
341                max_bytes: 1024,
342                cancel: &cancel,
343            },
344            |_| {},
345        )
346        .await
347        .expect("spawn should succeed");
348
349        match outcome {
350            SpawnOutcome::Done {
351                exit_code, signal, ..
352            } => {
353                assert_eq!(exit_code, Some(3));
354                assert_eq!(signal, None, "an ordinary exit was not signalled");
355            }
356            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
357        }
358    }
359
360    /// The user-facing sentence for the case that writes no stderr at all.
361    #[test]
362    fn describe_signal_names_the_common_kills() {
363        assert!(describe_signal(Some(9)).unwrap().contains("SIGKILL"));
364        assert!(describe_signal(Some(9)).unwrap().contains("memory"));
365        assert!(describe_signal(Some(15)).unwrap().contains("SIGTERM"));
366        assert!(describe_signal(Some(42)).unwrap().contains("42"));
367        assert_eq!(describe_signal(None), None);
368    }
369}
cli_agents/adapters/mod.rs

cli_agents/adapters/
mod.rs
