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 process_wrap::tokio::TokioChildWrapper;
14use process_wrap::tokio::TokioCommandWrap;
15#[cfg(unix)]
16use process_wrap::tokio::ProcessGroup;
17#[cfg(windows)]
18use process_wrap::tokio::JobObject;
19use tokio::io::{AsyncBufReadExt, BufReader};
20use tracing::{debug, warn};
21
22/// Trait implemented by each CLI adapter.
23pub trait CliAdapter: Send + Sync {
24    fn name(&self) -> CliName;
25
26    fn run(
27        &self,
28        opts: &RunOptions,
29        emit: &(dyn Fn(StreamEvent) + Send + Sync),
30        cancel: tokio_util::sync::CancellationToken,
31    ) -> impl std::future::Future<Output = crate::error::Result<RunResult>> + Send;
32}
33
34/// Get the adapter for a given CLI.
35pub(crate) fn get_adapter(cli: CliName) -> Box<dyn CliAdapterBoxed> {
36    match cli {
37        CliName::Claude => Box::new(ClaudeAdapter),
38        CliName::Codex => Box::new(CodexAdapter),
39        CliName::Gemini => Box::new(GeminiAdapter),
40    }
41}
42
43/// Object-safe version of [`CliAdapter`] for dynamic dispatch.
44///
45/// Needed because `CliAdapter::run` uses RPITIT (`impl Future`), which makes
46/// the trait non-object-safe. This wrapper boxes the future for `dyn` dispatch.
47/// The blanket impl below bridges the two automatically.
48#[allow(dead_code)]
49pub(crate) trait CliAdapterBoxed: Send + Sync {
50    fn name(&self) -> CliName;
51
52    fn run_boxed<'a>(
53        &'a self,
54        opts: &'a RunOptions,
55        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
56        cancel: tokio_util::sync::CancellationToken,
57    ) -> std::pin::Pin<
58        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
59    >;
60}
61
62impl<T: CliAdapter> CliAdapterBoxed for T {
63    fn name(&self) -> CliName {
64        CliAdapter::name(self)
65    }
66
67    fn run_boxed<'a>(
68        &'a self,
69        opts: &'a RunOptions,
70        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
71        cancel: tokio_util::sync::CancellationToken,
72    ) -> std::pin::Pin<
73        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
74    > {
75        Box::pin(self.run(opts, emit, cancel))
76    }
77}
78
79// ── Shared subprocess infrastructure ──
80
81/// Outcome of a spawned CLI process.
82pub(crate) enum SpawnOutcome {
83    /// Process reached a terminal state on its own (exited, or was signalled).
84    Done {
85        /// The process's exit status, or `None` when a SIGNAL ended it instead.
86        ///
87        /// A signalled process genuinely has no exit code. Substituting one
88        /// makes an out-of-memory kill indistinguishable from an agent that
89        /// cleanly decided it had failed, and [`RunResult::exit_code`] is an
90        /// `Option` precisely so a caller can tell the two apart.
91        exit_code: Option<i32>,
92        /// The signal that terminated the process, when one did. Unix only;
93        /// always `None` elsewhere.
94        signal: Option<i32>,
95        stderr: Option<String>,
96    },
97    /// Process was cancelled via the cancellation token.
98    Cancelled,
99}
100
101/// Parameters for [`spawn_and_stream`].
102pub(crate) struct SpawnParams<'a> {
103    pub cli_label: &'a str,
104    pub binary: &'a str,
105    pub args: &'a [String],
106    pub extra_env: &'a HashMap<String, String>,
107    /// Keys to remove from the inherited parent env before applying `extra_env`.
108    /// Used to prevent leaks like `ANTHROPIC_API_KEY` overriding subscription auth.
109    pub strip_env: &'a [&'static str],
110    pub cwd: &'a str,
111    pub max_bytes: usize,
112    pub cancel: &'a tokio_util::sync::CancellationToken,
113}
114
115/// Spawn a CLI subprocess and stream its stdout line-by-line.
116///
117/// Handles the boilerplate shared across all adapters: process spawning,
118/// stdout buffering with size limits, stderr collection, and cancellation.
119/// Does **not** clone the parent process environment — `Command` inherits it
120/// automatically; only `extra_env` entries are added.
121pub(crate) async fn spawn_and_stream(
122    params: SpawnParams<'_>,
123    mut on_line: impl FnMut(&str) + Send,
124) -> Result<SpawnOutcome> {
125    let SpawnParams {
126        cli_label,
127        binary,
128        args,
129        extra_env,
130        strip_env,
131        cwd,
132        max_bytes,
133        cancel,
134    } = params;
135    debug!(cli = cli_label, binary = %binary, args = ?args, "spawning CLI");
136
137    // ── The child owns a KILL GROUP, on every platform ──
138    //
139    // Cancelling a run has to take the whole tree, not just the process we
140    // spawned: `claude` is a launcher, and the work happens in node processes
141    // below it. Killing only the parent orphans those — they keep running, keep
142    // holding the model session, and keep writing to a pipe nobody reads.
143    //
144    // This used to be `pre_exec(setpgid)` plus `libc::killpg(SIGKILL)`, which is
145    // correct on unix and does not exist on Windows — where the equivalent is a
146    // Job Object, a completely different mechanism with the same purpose.
147    // `process-wrap` is that difference, already written and tested: the unix
148    // arm is the same process-group call, and the Windows arm assigns the child
149    // to a job that dies with it.
150    let mut wrap = TokioCommandWrap::with_new(binary, |cmd| {
151        cmd.args(args);
152        for key in strip_env {
153            cmd.env_remove(key);
154        }
155        cmd.envs(extra_env)
156            .current_dir(cwd)
157            .stdin(std::process::Stdio::null())
158            .stdout(std::process::Stdio::piped())
159            .stderr(std::process::Stdio::piped())
160            .kill_on_drop(true);
161    });
162    #[cfg(unix)]
163    wrap.wrap(ProcessGroup::leader());
164    #[cfg(windows)]
165    wrap.wrap(JobObject);
166
167    let mut child = wrap
168        .spawn()
169        .map_err(|e| Error::Process(format!("failed to spawn {cli_label}: {e}")))?;
170
171    let stdout = child.stdout().take().expect("stdout piped");
172    let stderr = child.stderr().take().expect("stderr piped");
173
174    let stderr_handle = tokio::spawn(async move {
175        let mut reader = BufReader::new(stderr);
176        let mut buf = String::new();
177        while reader.read_line(&mut buf).await.unwrap_or(0) > 0 {}
178        buf
179    });
180
181    let mut reader = BufReader::new(stdout);
182    let mut line = String::new();
183    let mut total_bytes: usize = 0;
184
185    loop {
186        line.clear();
187        tokio::select! {
188            result = reader.read_line(&mut line) => {
189                match result {
190                    Ok(0) => break,
191                    Ok(n) => {
192                        total_bytes += n;
193                        if total_bytes > max_bytes {
194                            warn!(cli = cli_label, total_bytes, max_bytes, "output exceeded max buffer size");
195                            kill_process_group(&mut child).await;
196                            return Err(Error::Process(format!(
197                                "output exceeded max buffer size ({max_bytes} bytes)"
198                            )));
199                        }
200                        on_line(line.trim());
201                    }
202                    Err(e) => {
203                        warn!(cli = cli_label, error = %e, "error reading stdout");
204                        break;
205                    }
206                }
207            }
208            _ = cancel.cancelled() => {
209                kill_process_group(&mut child).await;
210                return Ok(SpawnOutcome::Cancelled);
211            }
212        }
213    }
214
215    let status = Box::into_pin(child.wait()).await.map_err(Error::Io)?;
216    // `code()` is `None` for a signalled process. This used to be
217    // `.unwrap_or(1)`, which reported a SIGKILL as a clean `exit 1` and left
218    // callers with no way to recover the difference — the exact ambiguity that
219    // sent a downstream app hunting for an error message a killed process never
220    // wrote. Report what actually happened and let the caller decide.
221    let exit_code = status.code();
222    #[cfg(unix)]
223    let signal = std::os::unix::process::ExitStatusExt::signal(&status);
224    #[cfg(not(unix))]
225    let signal: Option<i32> = None;
226    let stderr_text = stderr_handle.await.unwrap_or_default();
227
228    Ok(SpawnOutcome::Done {
229        exit_code,
230        signal,
231        stderr: if stderr_text.is_empty() {
232            None
233        } else {
234            Some(stderr_text)
235        },
236    })
237}
238
239/// A sentence for a process that died without writing one.
240///
241/// A signalled CLI usually produces NO stderr and no result event — there was
242/// no chance to. Without this the only fact reaching the user is a number, and
243/// the most common case by far (the OS reclaiming memory) reads as an
244/// unexplained failure.
245pub(crate) fn describe_signal(signal: Option<i32>) -> Option<String> {
246    let sig = signal?;
247    Some(match sig {
248        2 => "The agent was interrupted (SIGINT).".to_string(),
249        6 => "The agent aborted (SIGABRT).".to_string(),
250        9 => "The agent was killed (SIGKILL), most often by the system reclaiming memory."
251            .to_string(),
252        11 => "The agent crashed (SIGSEGV).".to_string(),
253        15 => "The agent was terminated (SIGTERM).".to_string(),
254        other => format!("The agent was terminated by signal {other}."),
255    })
256}
257
258/// Extract a user-friendly error message from CLI stderr.
259/// When an agent fails with no text output, this provides something
260/// meaningful to show the user instead of a blank response.
261pub(crate) fn extract_error_message(stderr: Option<&str>) -> Option<String> {
262    let stderr = stderr?;
263    // Find the most informative error line.
264    let msg = stderr
265        .lines()
266        .filter(|l| !l.is_empty())
267        .find(|l| {
268            let lower = l.to_lowercase();
269            lower.contains("error")
270                || lower.contains("limit")
271                || lower.contains("failed")
272                || lower.contains("denied")
273                || lower.contains("unauthorized")
274        })
275        .or_else(|| stderr.lines().rfind(|l| !l.is_empty()));
276    msg.map(|s| s.trim().to_string())
277}
278
279/// Kill the child AND everything it spawned.
280///
281/// `TokioChildWrapper::kill` dispatches to whichever group mechanism was wrapped
282/// on at spawn — the process group on unix, the Job Object on Windows — so the
283/// `#[cfg]` that used to live here is gone. It returns a boxed future, hence the
284/// pin.
285async fn kill_process_group(child: &mut Box<dyn TokioChildWrapper>) {
286    let _ = Box::into_pin(child.kill()).await;
287}
288
289#[cfg(test)]
290mod tests {
291    use super::*;
292
293    /// CANCELLING TAKES THE WHOLE TREE, not just the process we spawned.
294    ///
295    /// This is the contract `setpgid`/`killpg` existed to provide, and it had no
296    /// test — so the swap to `process-wrap` would have been unverifiable, and so
297    /// would any future change to it. It matters because `claude` is a
298    /// launcher: the work runs in node processes underneath. Killing only the
299    /// parent leaves those alive, holding a model session, writing to a pipe
300    /// nobody is reading.
301    ///
302    /// HOW IT PROVES IT WITHOUT TIMING GAMES: the shell writes a marker file,
303    /// spawns a grandchild that would DELETE that file after a delay, then
304    /// sleeps. Cancel immediately. If the group died, the grandchild never runs
305    /// and the marker survives. If only the parent died, the orphan wakes up and
306    /// removes it. The assertion is on a filesystem fact, not on a pid still
307    /// being enumerable, which is what makes it honest on both platforms.
308    ///
309    /// Unix-only for now: it needs a shell that can background a process, and
310    /// the Windows equivalent (`cmd /c start`) has different semantics worth
311    /// writing deliberately rather than transliterating. The Job Object path is
312    /// exercised by CI compiling this file for Windows; that it KILLS the tree
313    /// there is not yet proved. Marked plainly rather than assumed.
314    #[cfg(unix)]
315    #[tokio::test]
316    async fn cancelling_kills_the_grandchild_not_just_the_child() {
317        let dir = tempfile::tempdir().unwrap();
318        let marker = dir.path().join("survivor");
319        std::fs::write(&marker, "alive").unwrap();
320
321        // Grandchild removes the marker after 3s; parent then sleeps 10s.
322        let script = format!("(sleep 3; rm -f '{}') & sleep 10", marker.display());
323        let args = vec!["-c".to_string(), script];
324        let cancel = tokio_util::sync::CancellationToken::new();
325
326        let token = cancel.clone();
327        tokio::spawn(async move {
328            tokio::time::sleep(std::time::Duration::from_millis(300)).await;
329            token.cancel();
330        });
331
332        let outcome = spawn_and_stream(
333            SpawnParams {
334                cli_label: "test",
335                binary: "sh",
336                args: &args,
337                extra_env: &HashMap::new(),
338                strip_env: &[],
339                cwd: dir.path().to_str().unwrap(),
340                max_bytes: 1024,
341                cancel: &cancel,
342            },
343            |_: &str| {},
344        )
345        .await
346        .expect("spawn");
347
348        assert!(matches!(outcome, SpawnOutcome::Cancelled), "run was cancelled");
349
350        // Past when the grandchild would have deleted it, had it survived.
351        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
352        assert!(
353            marker.exists(),
354            "the grandchild outlived cancellation and deleted the marker — the kill did not reach the process group"
355        );
356    }
357
358    /// A process that ends by SIGNAL has no exit code, and says so.
359    ///
360    /// THE REGRESSION THIS PINS. `spawn_and_stream` used to finish with
361    /// `status.code().unwrap_or(1)`, so a killed process was reported as a
362    /// clean `exit 1`. Downstream that is unrecoverable: an out-of-memory kill
363    /// and an agent that decided it had failed become the same fact, and a
364    /// consumer looking for the reason searches a stderr the process never got
365    /// to write. `sh -c 'kill -9 $$'` reproduces it without timing games — the
366    /// shell signals itself, so the outcome is deterministic.
367    #[cfg(unix)]
368    #[tokio::test]
369    async fn a_signalled_process_reports_the_signal_not_a_fabricated_exit_code() {
370        let args = vec!["-c".to_string(), "kill -9 $$".to_string()];
371        let cancel = tokio_util::sync::CancellationToken::new();
372        let outcome = spawn_and_stream(
373            SpawnParams {
374                cli_label: "test",
375                binary: "sh",
376                args: &args,
377                extra_env: &HashMap::new(),
378                strip_env: &[],
379                cwd: ".",
380                max_bytes: 1024,
381                cancel: &cancel,
382            },
383            |_| {},
384        )
385        .await
386        .expect("spawn should succeed");
387
388        match outcome {
389            SpawnOutcome::Done {
390                exit_code, signal, ..
391            } => {
392                assert_eq!(exit_code, None, "a signalled process has no exit code");
393                assert_eq!(signal, Some(9), "SIGKILL should be reported as itself");
394            }
395            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
396        }
397    }
398
399    /// An ordinary non-zero exit still reports its code — the change above must
400    /// not turn every failure into `None`.
401    #[tokio::test]
402    async fn a_normal_exit_still_reports_its_code() {
403        let args = vec!["-c".to_string(), "exit 3".to_string()];
404        let cancel = tokio_util::sync::CancellationToken::new();
405        let outcome = spawn_and_stream(
406            SpawnParams {
407                cli_label: "test",
408                binary: "sh",
409                args: &args,
410                extra_env: &HashMap::new(),
411                strip_env: &[],
412                cwd: ".",
413                max_bytes: 1024,
414                cancel: &cancel,
415            },
416            |_| {},
417        )
418        .await
419        .expect("spawn should succeed");
420
421        match outcome {
422            SpawnOutcome::Done {
423                exit_code, signal, ..
424            } => {
425                assert_eq!(exit_code, Some(3));
426                assert_eq!(signal, None, "an ordinary exit was not signalled");
427            }
428            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
429        }
430    }
431
432    /// The user-facing sentence for the case that writes no stderr at all.
433    #[test]
434    fn describe_signal_names_the_common_kills() {
435        assert!(describe_signal(Some(9)).unwrap().contains("SIGKILL"));
436        assert!(describe_signal(Some(9)).unwrap().contains("memory"));
437        assert!(describe_signal(Some(15)).unwrap().contains("SIGTERM"));
438        assert!(describe_signal(Some(42)).unwrap().contains("42"));
439        assert_eq!(describe_signal(None), None);
440    }
441}
cli_agents/adapters/mod.rs

cli_agents/adapters/
mod.rs
