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 exited normally.
79    Done {
80        exit_code: i32,
81        stderr: Option<String>,
82    },
83    /// Process was cancelled via the cancellation token.
84    Cancelled,
85}
86
87/// Parameters for [`spawn_and_stream`].
88pub(crate) struct SpawnParams<'a> {
89    pub cli_label: &'a str,
90    pub binary: &'a str,
91    pub args: &'a [String],
92    pub extra_env: &'a HashMap<String, String>,
93    /// Keys to remove from the inherited parent env before applying `extra_env`.
94    /// Used to prevent leaks like `ANTHROPIC_API_KEY` overriding subscription auth.
95    pub strip_env: &'a [&'static str],
96    pub cwd: &'a str,
97    pub max_bytes: usize,
98    pub cancel: &'a tokio_util::sync::CancellationToken,
99}
100
101/// Spawn a CLI subprocess and stream its stdout line-by-line.
102///
103/// Handles the boilerplate shared across all adapters: process spawning,
104/// stdout buffering with size limits, stderr collection, and cancellation.
105/// Does **not** clone the parent process environment — `Command` inherits it
106/// automatically; only `extra_env` entries are added.
107pub(crate) async fn spawn_and_stream(
108    params: SpawnParams<'_>,
109    mut on_line: impl FnMut(&str) + Send,
110) -> Result<SpawnOutcome> {
111    let SpawnParams {
112        cli_label,
113        binary,
114        args,
115        extra_env,
116        strip_env,
117        cwd,
118        max_bytes,
119        cancel,
120    } = params;
121    debug!(cli = cli_label, binary = %binary, args = ?args, "spawning CLI");
122
123    let mut cmd = Command::new(binary);
124    cmd.args(args);
125    for key in strip_env {
126        cmd.env_remove(key);
127    }
128    cmd.envs(extra_env)
129        .current_dir(cwd)
130        .stdin(std::process::Stdio::null())
131        .stdout(std::process::Stdio::piped())
132        .stderr(std::process::Stdio::piped())
133        .kill_on_drop(true);
134
135    #[cfg(unix)]
136    {
137        unsafe {
138            cmd.pre_exec(|| {
139                if libc::setpgid(0, 0) != 0 {
140                    return Err(std::io::Error::last_os_error());
141                }
142                Ok(())
143            });
144        }
145    }
146
147    let mut child = cmd
148        .spawn()
149        .map_err(|e| Error::Process(format!("failed to spawn {cli_label}: {e}")))?;
150
151    let child_pid = child.id();
152
153    let stdout = child.stdout.take().expect("stdout piped");
154    let stderr = child.stderr.take().expect("stderr piped");
155
156    let stderr_handle = tokio::spawn(async move {
157        let mut reader = BufReader::new(stderr);
158        let mut buf = String::new();
159        while reader.read_line(&mut buf).await.unwrap_or(0) > 0 {}
160        buf
161    });
162
163    let mut reader = BufReader::new(stdout);
164    let mut line = String::new();
165    let mut total_bytes: usize = 0;
166
167    loop {
168        line.clear();
169        tokio::select! {
170            result = reader.read_line(&mut line) => {
171                match result {
172                    Ok(0) => break,
173                    Ok(n) => {
174                        total_bytes += n;
175                        if total_bytes > max_bytes {
176                            warn!(cli = cli_label, total_bytes, max_bytes, "output exceeded max buffer size");
177                            kill_process_group(&mut child, child_pid).await;
178                            return Err(Error::Process(format!(
179                                "output exceeded max buffer size ({max_bytes} bytes)"
180                            )));
181                        }
182                        on_line(line.trim());
183                    }
184                    Err(e) => {
185                        warn!(cli = cli_label, error = %e, "error reading stdout");
186                        break;
187                    }
188                }
189            }
190            _ = cancel.cancelled() => {
191                kill_process_group(&mut child, child_pid).await;
192                return Ok(SpawnOutcome::Cancelled);
193            }
194        }
195    }
196
197    let status = child.wait().await.map_err(Error::Io)?;
198    let exit_code = status.code().unwrap_or(1);
199    let stderr_text = stderr_handle.await.unwrap_or_default();
200
201    Ok(SpawnOutcome::Done {
202        exit_code,
203        stderr: if stderr_text.is_empty() {
204            None
205        } else {
206            Some(stderr_text)
207        },
208    })
209}
210
211/// Extract a user-friendly error message from CLI stderr.
212/// When an agent fails with no text output, this provides something
213/// meaningful to show the user instead of a blank response.
214pub(crate) fn extract_error_message(stderr: Option<&str>) -> Option<String> {
215    let stderr = stderr?;
216    // Find the most informative error line.
217    let msg = stderr
218        .lines()
219        .filter(|l| !l.is_empty())
220        .find(|l| {
221            let lower = l.to_lowercase();
222            lower.contains("error")
223                || lower.contains("limit")
224                || lower.contains("failed")
225                || lower.contains("denied")
226                || lower.contains("unauthorized")
227        })
228        .or_else(|| stderr.lines().filter(|l| !l.is_empty()).last());
229    msg.map(|s| s.trim().to_string())
230}
231
232async fn kill_process_group(child: &mut tokio::process::Child, pid: Option<u32>) {
233    #[cfg(unix)]
234    {
235        if let Some(pid) = pid {
236            unsafe {
237                libc::killpg(pid as libc::pid_t, libc::SIGKILL);
238            }
239        }
240    }
241    let _ = child.kill().await;
242}
cli_agents/adapters/mod.rs

cli_agents/adapters/
mod.rs
