cli-agents 0.2.13

mod claude;
mod codex;
mod gemini;

pub use claude::ClaudeAdapter;
pub use codex::CodexAdapter;
pub use gemini::GeminiAdapter;

use crate::error::{Error, Result};
use crate::events::StreamEvent;
use crate::types::{CliName, RunOptions, RunResult};
use std::collections::HashMap;
use process_wrap::tokio::TokioChildWrapper;
use process_wrap::tokio::TokioCommandWrap;
#[cfg(unix)]
use process_wrap::tokio::ProcessGroup;
#[cfg(windows)]
use process_wrap::tokio::JobObject;
use tokio::io::{AsyncBufReadExt, BufReader};
use tracing::{debug, warn};

/// Trait implemented by each CLI adapter.
pub trait CliAdapter: Send + Sync {
    fn name(&self) -> CliName;

    fn run(
        &self,
        opts: &RunOptions,
        emit: &(dyn Fn(StreamEvent) + Send + Sync),
        cancel: tokio_util::sync::CancellationToken,
    ) -> impl std::future::Future<Output = crate::error::Result<RunResult>> + Send;
}

/// Get the adapter for a given CLI.
pub(crate) fn get_adapter(cli: CliName) -> Box<dyn CliAdapterBoxed> {
    match cli {
        CliName::Claude => Box::new(ClaudeAdapter),
        CliName::Codex => Box::new(CodexAdapter),
        CliName::Gemini => Box::new(GeminiAdapter),
    }
}

/// Object-safe version of [`CliAdapter`] for dynamic dispatch.
///
/// Needed because `CliAdapter::run` uses RPITIT (`impl Future`), which makes
/// the trait non-object-safe. This wrapper boxes the future for `dyn` dispatch.
/// The blanket impl below bridges the two automatically.
#[allow(dead_code)]
pub(crate) trait CliAdapterBoxed: Send + Sync {
    fn name(&self) -> CliName;

    fn run_boxed<'a>(
        &'a self,
        opts: &'a RunOptions,
        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
        cancel: tokio_util::sync::CancellationToken,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
    >;
}

impl<T: CliAdapter> CliAdapterBoxed for T {
    fn name(&self) -> CliName {
        CliAdapter::name(self)
    }

    fn run_boxed<'a>(
        &'a self,
        opts: &'a RunOptions,
        emit: &'a (dyn Fn(StreamEvent) + Send + Sync),
        cancel: tokio_util::sync::CancellationToken,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = crate::error::Result<RunResult>> + Send + 'a>,
    > {
        Box::pin(self.run(opts, emit, cancel))
    }
}

// ── Shared subprocess infrastructure ──

/// Outcome of a spawned CLI process.
pub(crate) enum SpawnOutcome {
    /// Process reached a terminal state on its own (exited, or was signalled).
    Done {
        /// The process's exit status, or `None` when a SIGNAL ended it instead.
        ///
        /// A signalled process genuinely has no exit code. Substituting one
        /// makes an out-of-memory kill indistinguishable from an agent that
        /// cleanly decided it had failed, and [`RunResult::exit_code`] is an
        /// `Option` precisely so a caller can tell the two apart.
        exit_code: Option<i32>,
        /// The signal that terminated the process, when one did. Unix only;
        /// always `None` elsewhere.
        signal: Option<i32>,
        stderr: Option<String>,
    },
    /// Process was cancelled via the cancellation token.
    Cancelled,
}

/// Parameters for [`spawn_and_stream`].
pub(crate) struct SpawnParams<'a> {
    pub cli_label: &'a str,
    pub binary: &'a str,
    pub args: &'a [String],
    pub extra_env: &'a HashMap<String, String>,
    /// Keys to remove from the inherited parent env before applying `extra_env`.
    /// Used to prevent leaks like `ANTHROPIC_API_KEY` overriding subscription auth.
    pub strip_env: &'a [&'static str],
    pub cwd: &'a str,
    pub max_bytes: usize,
    pub cancel: &'a tokio_util::sync::CancellationToken,
}

/// Spawn a CLI subprocess and stream its stdout line-by-line.
///
/// Handles the boilerplate shared across all adapters: process spawning,
/// stdout buffering with size limits, stderr collection, and cancellation.
/// Does **not** clone the parent process environment — `Command` inherits it
/// automatically; only `extra_env` entries are added.
pub(crate) async fn spawn_and_stream(
    params: SpawnParams<'_>,
    mut on_line: impl FnMut(&str) + Send,
) -> Result<SpawnOutcome> {
    let SpawnParams {
        cli_label,
        binary,
        args,
        extra_env,
        strip_env,
        cwd,
        max_bytes,
        cancel,
    } = params;
    debug!(cli = cli_label, binary = %binary, args = ?args, "spawning CLI");

    // ── The child owns a KILL GROUP, on every platform ──
    //
    // Cancelling a run has to take the whole tree, not just the process we
    // spawned: `claude` is a launcher, and the work happens in node processes
    // below it. Killing only the parent orphans those — they keep running, keep
    // holding the model session, and keep writing to a pipe nobody reads.
    //
    // This used to be `pre_exec(setpgid)` plus `libc::killpg(SIGKILL)`, which is
    // correct on unix and does not exist on Windows — where the equivalent is a
    // Job Object, a completely different mechanism with the same purpose.
    // `process-wrap` is that difference, already written and tested: the unix
    // arm is the same process-group call, and the Windows arm assigns the child
    // to a job that dies with it.
    let mut wrap = TokioCommandWrap::with_new(binary, |cmd| {
        cmd.args(args);
        for key in strip_env {
            cmd.env_remove(key);
        }
        cmd.envs(extra_env)
            .current_dir(cwd)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .kill_on_drop(true);
    });
    #[cfg(unix)]
    wrap.wrap(ProcessGroup::leader());
    #[cfg(windows)]
    wrap.wrap(JobObject);

    let mut child = wrap
        .spawn()
        .map_err(|e| Error::Process(format!("failed to spawn {cli_label}: {e}")))?;

    let stdout = child.stdout().take().expect("stdout piped");
    let stderr = child.stderr().take().expect("stderr piped");

    let stderr_handle = tokio::spawn(async move {
        let mut reader = BufReader::new(stderr);
        let mut buf = String::new();
        while reader.read_line(&mut buf).await.unwrap_or(0) > 0 {}
        buf
    });

    let mut reader = BufReader::new(stdout);
    let mut line = String::new();
    let mut total_bytes: usize = 0;

    loop {
        line.clear();
        tokio::select! {
            result = reader.read_line(&mut line) => {
                match result {
                    Ok(0) => break,
                    Ok(n) => {
                        total_bytes += n;
                        if total_bytes > max_bytes {
                            warn!(cli = cli_label, total_bytes, max_bytes, "output exceeded max buffer size");
                            kill_process_group(&mut child).await;
                            return Err(Error::Process(format!(
                                "output exceeded max buffer size ({max_bytes} bytes)"
                            )));
                        }
                        on_line(line.trim());
                    }
                    Err(e) => {
                        warn!(cli = cli_label, error = %e, "error reading stdout");
                        break;
                    }
                }
            }
            _ = cancel.cancelled() => {
                kill_process_group(&mut child).await;
                return Ok(SpawnOutcome::Cancelled);
            }
        }
    }

    let status = Box::into_pin(child.wait()).await.map_err(Error::Io)?;
    // `code()` is `None` for a signalled process. This used to be
    // `.unwrap_or(1)`, which reported a SIGKILL as a clean `exit 1` and left
    // callers with no way to recover the difference — the exact ambiguity that
    // sent a downstream app hunting for an error message a killed process never
    // wrote. Report what actually happened and let the caller decide.
    let exit_code = status.code();
    #[cfg(unix)]
    let signal = std::os::unix::process::ExitStatusExt::signal(&status);
    #[cfg(not(unix))]
    let signal: Option<i32> = None;
    let stderr_text = stderr_handle.await.unwrap_or_default();

    Ok(SpawnOutcome::Done {
        exit_code,
        signal,
        stderr: if stderr_text.is_empty() {
            None
        } else {
            Some(stderr_text)
        },
    })
}

/// A sentence for a process that died without writing one.
///
/// A signalled CLI usually produces NO stderr and no result event — there was
/// no chance to. Without this the only fact reaching the user is a number, and
/// the most common case by far (the OS reclaiming memory) reads as an
/// unexplained failure.
pub(crate) fn describe_signal(signal: Option<i32>) -> Option<String> {
    let sig = signal?;
    Some(match sig {
        2 => "The agent was interrupted (SIGINT).".to_string(),
        6 => "The agent aborted (SIGABRT).".to_string(),
        9 => "The agent was killed (SIGKILL), most often by the system reclaiming memory."
            .to_string(),
        11 => "The agent crashed (SIGSEGV).".to_string(),
        15 => "The agent was terminated (SIGTERM).".to_string(),
        other => format!("The agent was terminated by signal {other}."),
    })
}

/// Extract a user-friendly error message from CLI stderr.
/// When an agent fails with no text output, this provides something
/// meaningful to show the user instead of a blank response.
pub(crate) fn extract_error_message(stderr: Option<&str>) -> Option<String> {
    let stderr = stderr?;
    // Find the most informative error line.
    let msg = stderr
        .lines()
        .filter(|l| !l.is_empty())
        .find(|l| {
            let lower = l.to_lowercase();
            lower.contains("error")
                || lower.contains("limit")
                || lower.contains("failed")
                || lower.contains("denied")
                || lower.contains("unauthorized")
        })
        .or_else(|| stderr.lines().rfind(|l| !l.is_empty()));
    msg.map(|s| s.trim().to_string())
}

/// Kill the child AND everything it spawned.
///
/// `TokioChildWrapper::kill` dispatches to whichever group mechanism was wrapped
/// on at spawn — the process group on unix, the Job Object on Windows — so the
/// `#[cfg]` that used to live here is gone. It returns a boxed future, hence the
/// pin.
async fn kill_process_group(child: &mut Box<dyn TokioChildWrapper>) {
    let _ = Box::into_pin(child.kill()).await;
}

#[cfg(test)]
mod tests {
    use super::*;

    /// CANCELLING TAKES THE WHOLE TREE, not just the process we spawned.
    ///
    /// This is the contract `setpgid`/`killpg` existed to provide, and it had no
    /// test — so the swap to `process-wrap` would have been unverifiable, and so
    /// would any future change to it. It matters because `claude` is a
    /// launcher: the work runs in node processes underneath. Killing only the
    /// parent leaves those alive, holding a model session, writing to a pipe
    /// nobody is reading.
    ///
    /// HOW IT PROVES IT WITHOUT TIMING GAMES: the shell writes a marker file,
    /// spawns a grandchild that would DELETE that file after a delay, then
    /// sleeps. Cancel immediately. If the group died, the grandchild never runs
    /// and the marker survives. If only the parent died, the orphan wakes up and
    /// removes it. The assertion is on a filesystem fact, not on a pid still
    /// being enumerable, which is what makes it honest on both platforms.
    ///
    /// Unix-only for now: it needs a shell that can background a process, and
    /// the Windows equivalent (`cmd /c start`) has different semantics worth
    /// writing deliberately rather than transliterating. The Job Object path is
    /// exercised by CI compiling this file for Windows; that it KILLS the tree
    /// there is not yet proved. Marked plainly rather than assumed.
    #[cfg(unix)]
    #[tokio::test]
    async fn cancelling_kills_the_grandchild_not_just_the_child() {
        let dir = tempfile::tempdir().unwrap();
        let marker = dir.path().join("survivor");
        std::fs::write(&marker, "alive").unwrap();

        // Grandchild removes the marker after 3s; parent then sleeps 10s.
        let script = format!("(sleep 3; rm -f '{}') & sleep 10", marker.display());
        let args = vec!["-c".to_string(), script];
        let cancel = tokio_util::sync::CancellationToken::new();

        let token = cancel.clone();
        tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(300)).await;
            token.cancel();
        });

        let outcome = spawn_and_stream(
            SpawnParams {
                cli_label: "test",
                binary: "sh",
                args: &args,
                extra_env: &HashMap::new(),
                strip_env: &[],
                cwd: dir.path().to_str().unwrap(),
                max_bytes: 1024,
                cancel: &cancel,
            },
            |_: &str| {},
        )
        .await
        .expect("spawn");

        assert!(matches!(outcome, SpawnOutcome::Cancelled), "run was cancelled");

        // Past when the grandchild would have deleted it, had it survived.
        tokio::time::sleep(std::time::Duration::from_secs(5)).await;
        assert!(
            marker.exists(),
            "the grandchild outlived cancellation and deleted the marker — the kill did not reach the process group"
        );
    }

    /// A process that ends by SIGNAL has no exit code, and says so.
    ///
    /// THE REGRESSION THIS PINS. `spawn_and_stream` used to finish with
    /// `status.code().unwrap_or(1)`, so a killed process was reported as a
    /// clean `exit 1`. Downstream that is unrecoverable: an out-of-memory kill
    /// and an agent that decided it had failed become the same fact, and a
    /// consumer looking for the reason searches a stderr the process never got
    /// to write. `sh -c 'kill -9 $$'` reproduces it without timing games — the
    /// shell signals itself, so the outcome is deterministic.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_signalled_process_reports_the_signal_not_a_fabricated_exit_code() {
        let args = vec!["-c".to_string(), "kill -9 $$".to_string()];
        let cancel = tokio_util::sync::CancellationToken::new();
        let outcome = spawn_and_stream(
            SpawnParams {
                cli_label: "test",
                binary: "sh",
                args: &args,
                extra_env: &HashMap::new(),
                strip_env: &[],
                cwd: ".",
                max_bytes: 1024,
                cancel: &cancel,
            },
            |_| {},
        )
        .await
        .expect("spawn should succeed");

        match outcome {
            SpawnOutcome::Done {
                exit_code, signal, ..
            } => {
                assert_eq!(exit_code, None, "a signalled process has no exit code");
                assert_eq!(signal, Some(9), "SIGKILL should be reported as itself");
            }
            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
        }
    }

    /// An ordinary non-zero exit still reports its code — the change above must
    /// not turn every failure into `None`.
    #[tokio::test]
    async fn a_normal_exit_still_reports_its_code() {
        let args = vec!["-c".to_string(), "exit 3".to_string()];
        let cancel = tokio_util::sync::CancellationToken::new();
        let outcome = spawn_and_stream(
            SpawnParams {
                cli_label: "test",
                binary: "sh",
                args: &args,
                extra_env: &HashMap::new(),
                strip_env: &[],
                cwd: ".",
                max_bytes: 1024,
                cancel: &cancel,
            },
            |_| {},
        )
        .await
        .expect("spawn should succeed");

        match outcome {
            SpawnOutcome::Done {
                exit_code, signal, ..
            } => {
                assert_eq!(exit_code, Some(3));
                assert_eq!(signal, None, "an ordinary exit was not signalled");
            }
            SpawnOutcome::Cancelled => panic!("nothing cancelled this run"),
        }
    }

    /// The user-facing sentence for the case that writes no stderr at all.
    #[test]
    fn describe_signal_names_the_common_kills() {
        assert!(describe_signal(Some(9)).unwrap().contains("SIGKILL"));
        assert!(describe_signal(Some(9)).unwrap().contains("memory"));
        assert!(describe_signal(Some(15)).unwrap().contains("SIGTERM"));
        assert!(describe_signal(Some(42)).unwrap().contains("42"));
        assert_eq!(describe_signal(None), None);
    }
}