car-inference 0.55.0

//! Subscription-backed text generation through `codex exec`.
//!
//! This is deliberately separate from `car-external-agents`' `external:codex`
//! adapter. That adapter is an autonomous coding-agent surface: it grants a
//! workspace, observes tool calls, and may run multiple turns. This backend is
//! the narrower inference seam. It accepts one text prompt, disables every
//! Codex tool-bearing feature we rely on, runs in a fresh read-only directory,
//! and returns only the final model message plus Codex's own token counts.

use crate::{InferenceError, TokenUsage};
use serde_json::Value;
use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::time::Duration;
use tokio::io::AsyncWriteExt;
use tokio::process::Command;

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(600);
const MAX_DIAGNOSTIC_BYTES: usize = 4096;
const CODEX_BIN_ENV: &str = "CAR_CODEX_BIN";

/// Successful output from one tool-free Codex turn.
#[derive(Debug, Clone)]
pub struct CodexCliOutput {
    pub text: String,
    pub usage: TokenUsage,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct ModelSpec {
    model: String,
    reasoning_effort: Option<String>,
}

/// Resolve the executable without inspecting Codex's credential store.
fn configured_binary() -> PathBuf {
    let path = std::env::var_os(CODEX_BIN_ENV)
        .filter(|value| !value.is_empty())
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("codex"));
    if path.is_relative() && path.components().count() > 1 {
        // `Command::current_dir` makes relative executable paths ambiguous
        // across platforms. Resolve an explicit `./path` before moving the
        // child into its empty scratch directory.
        std::fs::canonicalize(&path).unwrap_or(path)
    } else {
        path
    }
}

/// Cheap readiness check for catalog routing. Authentication remains wholly
/// owned by Codex and is checked by the child when a request actually runs.
pub fn is_available() -> bool {
    let binary = configured_binary();
    if binary.components().count() > 1 || binary.is_absolute() {
        return binary.is_file();
    }
    std::env::var_os("PATH")
        .map(|paths| {
            std::env::split_paths(&paths).any(|dir| {
                executable_candidates(&dir, &binary).any(|candidate| candidate.is_file())
            })
        })
        .unwrap_or(false)
}

fn executable_candidates<'a>(
    dir: &'a Path,
    binary: &'a Path,
) -> impl Iterator<Item = PathBuf> + 'a {
    let plain = dir.join(binary);
    #[allow(unused_mut)]
    let mut candidates = vec![plain];
    #[cfg(target_os = "windows")]
    {
        if binary.extension().is_none() {
            candidates.push(dir.join(format!("{}.exe", binary.to_string_lossy())));
            candidates.push(dir.join(format!("{}.cmd", binary.to_string_lossy())));
            candidates.push(dir.join(format!("{}.bat", binary.to_string_lossy())));
        }
    }
    candidates.into_iter()
}

fn parse_model_spec(spec: &str) -> Result<ModelSpec, InferenceError> {
    const EFFORTS: &[&str] = &["minimal", "low", "medium", "high", "xhigh", "max"];
    let spec = spec.trim();
    if spec.is_empty() {
        return Err(InferenceError::InferenceFailed(
            "Codex CLI model source has an empty model".to_string(),
        ));
    }
    if let Some((model, suffix)) = spec.rsplit_once(':') {
        if EFFORTS.contains(&suffix) {
            if model.trim().is_empty() {
                return Err(InferenceError::InferenceFailed(
                    "Codex CLI model source has an empty model before its effort suffix"
                        .to_string(),
                ));
            }
            return Ok(ModelSpec {
                model: model.to_string(),
                reasoning_effort: Some(suffix.to_string()),
            });
        }
    }
    Ok(ModelSpec {
        model: spec.to_string(),
        reasoning_effort: None,
    })
}

/// Build the exact `codex exec` invocation. The caller supplies the prompt on
/// stdin so it never appears in the process list.
fn build_args(spec: &ModelSpec, max_output_tokens: usize, scratch: &Path) -> Vec<OsString> {
    let mut args: Vec<OsString> = vec![
        "exec".into(),
        "--json".into(),
        "--ephemeral".into(),
        "--skip-git-repo-check".into(),
        "--ignore-user-config".into(),
        "--ignore-rules".into(),
        "--strict-config".into(),
        "--sandbox".into(),
        "read-only".into(),
        "--cd".into(),
        scratch.as_os_str().to_owned(),
        "--model".into(),
        spec.model.clone().into(),
        // These stable Codex features are the possible side-effect carriers.
        // Disable them explicitly even though ignore-user-config removes any
        // user MCP/plugin configuration and the scratch cwd contains no project
        // configuration.
        "--disable".into(),
        "shell_tool".into(),
        "--disable".into(),
        "multi_agent".into(),
        "--disable".into(),
        "apps".into(),
        "--disable".into(),
        "plugins".into(),
        "--disable".into(),
        "code_mode_host".into(),
        "--disable".into(),
        "standalone_web_search".into(),
        "--config".into(),
        "web_search=\"disabled\"".into(),
        "--config".into(),
        format!(
            "developer_instructions={}",
            toml_string(&format!(
                "You are serving one CAR text-generation request. Return one final answer. Do not use tools, delegate, browse, inspect files, or run commands. Keep the final answer within approximately {max_output_tokens} tokens."
            ))
        )
        .into(),
    ];
    if let Some(effort) = &spec.reasoning_effort {
        args.push("--config".into());
        args.push(format!("model_reasoning_effort={}", toml_string(effort)).into());
    }
    args.push("-".into());
    args
}

fn toml_string(value: &str) -> String {
    serde_json::to_string(value).expect("JSON strings are valid TOML basic strings")
}

/// Run one text-only Codex generation. `OPENAI_API_KEY` is removed from the
/// child environment so this route can only use credentials Codex itself owns
/// (normally a ChatGPT subscription login).
pub async fn generate(
    model: &str,
    prompt: &str,
    context: Option<&str>,
    max_output_tokens: usize,
    context_window: usize,
) -> Result<CodexCliOutput, InferenceError> {
    generate_with_program(
        &configured_binary(),
        model,
        prompt,
        context,
        max_output_tokens,
        context_window,
        DEFAULT_TIMEOUT,
    )
    .await
}

async fn generate_with_program(
    program: &Path,
    model: &str,
    prompt: &str,
    context: Option<&str>,
    max_output_tokens: usize,
    context_window: usize,
    timeout: Duration,
) -> Result<CodexCliOutput, InferenceError> {
    let spec = parse_model_spec(model)?;
    let scratch = tempfile::tempdir().map_err(|error| {
        InferenceError::InferenceFailed(format!("create Codex CLI scratch directory: {error}"))
    })?;
    let args = build_args(&spec, max_output_tokens, scratch.path());
    let rendered_prompt = match context.filter(|value| !value.trim().is_empty()) {
        Some(context) => format!("Context:\n{context}\n\nRequest:\n{prompt}"),
        None => prompt.to_string(),
    };

    let mut command = Command::new(program);
    command
        .args(&args)
        .current_dir(scratch.path())
        .env_remove("OPENAI_API_KEY")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .kill_on_drop(true);
    let mut child = command.spawn().map_err(|error| {
        InferenceError::InferenceFailed(format!(
            "spawn Codex CLI `{}`: {error}; install Codex and run `codex login` with ChatGPT",
            program.display()
        ))
    })?;
    let mut stdin = child.stdin.take().ok_or_else(|| {
        InferenceError::InferenceFailed("Codex CLI stdin was not available".to_string())
    })?;
    // Drain output while writing stdin. Codex normally waits for EOF before it
    // emits events, but a diagnostic burst must not fill stderr and deadlock a
    // large prompt before the timeout has even started.
    let write_prompt = async move {
        stdin.write_all(rendered_prompt.as_bytes()).await?;
        stdin.shutdown().await
    };
    let communicate = async {
        let (_, output) = tokio::try_join!(write_prompt, child.wait_with_output())?;
        Ok::<_, std::io::Error>(output)
    };
    let output = tokio::time::timeout(timeout, communicate)
        .await
        .map_err(|_| {
            InferenceError::InferenceFailed(format!(
                "Codex CLI did not finish within {} seconds",
                timeout.as_secs()
            ))
        })?
        .map_err(|error| {
            InferenceError::InferenceFailed(format!("communicate with Codex CLI: {error}"))
        })?;

    if !output.status.success() {
        let stderr = bounded_diagnostic(&output.stderr);
        return Err(InferenceError::InferenceFailed(format!(
            "Codex CLI exited with status {}: {stderr}",
            output.status
        )));
    }
    parse_output(&output.stdout, context_window)
}

fn bounded_diagnostic(bytes: &[u8]) -> String {
    let start = bytes.len().saturating_sub(MAX_DIAGNOSTIC_BYTES);
    String::from_utf8_lossy(&bytes[start..]).trim().to_string()
}

fn parse_output(stdout: &[u8], context_window: usize) -> Result<CodexCliOutput, InferenceError> {
    let text = String::from_utf8_lossy(stdout);
    let mut answer_parts = Vec::new();
    let mut turns = 0usize;
    let mut completed_turns = 0usize;
    let mut usage = None;
    let mut forbidden_items = Vec::new();
    let mut provider_error = None;

    for line in text.lines().filter(|line| !line.trim().is_empty()) {
        let Ok(event) = serde_json::from_str::<Value>(line) else {
            continue;
        };
        match event.get("type").and_then(Value::as_str).unwrap_or("") {
            "turn.started" => turns += 1,
            "item.started" | "item.updated" | "item.completed" => {
                let Some(item) = event.get("item") else {
                    continue;
                };
                match item.get("type").and_then(Value::as_str).unwrap_or("") {
                    "agent_message" => {
                        if event.get("type").and_then(Value::as_str) == Some("item.completed") {
                            if let Some(part) = item.get("text").and_then(Value::as_str) {
                                answer_parts.push(part.to_string());
                            }
                        }
                    }
                    "reasoning" => {}
                    other
                        if !other.is_empty()
                            && forbidden_items.len() < 8
                            && !forbidden_items.iter().any(|seen| seen == other) =>
                    {
                        forbidden_items.push(other.to_string());
                    }
                    _ => {}
                }
            }
            "turn.completed" => {
                completed_turns += 1;
                if let Some(raw) = event.get("usage") {
                    if let (Some(input_total), Some(output)) = (
                        raw.get("input_tokens").and_then(Value::as_u64),
                        raw.get("output_tokens").and_then(Value::as_u64),
                    ) {
                        let cached = raw
                            .get("cached_input_tokens")
                            .and_then(Value::as_u64)
                            .unwrap_or(0)
                            .min(input_total);
                        let cache_write = raw
                            .get("cache_write_input_tokens")
                            .and_then(Value::as_u64)
                            .unwrap_or(0)
                            .min(input_total.saturating_sub(cached));
                        usage = Some(TokenUsage {
                            prompt_tokens: input_total - cached - cache_write,
                            completion_tokens: output,
                            total_tokens: input_total + output,
                            context_window: context_window as u64,
                            cache_read_input_tokens: cached,
                            cache_creation_input_tokens: cache_write,
                        });
                    }
                }
            }
            "turn.failed" | "error" => {
                provider_error = event
                    .get("error")
                    .and_then(|error| {
                        error
                            .get("message")
                            .and_then(Value::as_str)
                            .or_else(|| error.as_str())
                    })
                    .or_else(|| event.get("message").and_then(Value::as_str))
                    .map(str::to_string);
            }
            _ => {}
        }
    }

    if let Some(error) = provider_error {
        return Err(InferenceError::InferenceFailed(format!(
            "Codex CLI turn failed: {}",
            bounded_diagnostic(error.as_bytes())
        )));
    }
    if turns != 1 {
        return Err(InferenceError::InferenceFailed(format!(
            "Codex CLI inference requires exactly one turn; observed {turns}"
        )));
    }
    if completed_turns != 1 {
        return Err(InferenceError::InferenceFailed(format!(
            "Codex CLI inference requires exactly one completed turn; observed {completed_turns}"
        )));
    }
    if !forbidden_items.is_empty() {
        return Err(InferenceError::InferenceFailed(format!(
            "Codex CLI inference emitted forbidden non-generation items: {}",
            forbidden_items.join(", ")
        )));
    }
    let text = answer_parts.join("");
    if text.trim().is_empty() {
        return Err(InferenceError::InferenceFailed(
            "Codex CLI produced no final agent message".to_string(),
        ));
    }
    let usage = usage.ok_or_else(|| {
        InferenceError::InferenceFailed(
            "Codex CLI completed without token usage; refusing to fabricate accounting".to_string(),
        )
    })?;
    Ok(CodexCliOutput { text, usage })
}

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

    #[test]
    fn model_effort_suffix_maps_to_codex_config() {
        let spec = parse_model_spec("gpt-5.6-sol:high").unwrap();
        assert_eq!(spec.model, "gpt-5.6-sol");
        assert_eq!(spec.reasoning_effort.as_deref(), Some("high"));
        let args = build_args(&spec, 1056, Path::new("/tmp/codex-test"));
        let rendered: Vec<String> = args
            .iter()
            .map(|arg| arg.to_string_lossy().into_owned())
            .collect();
        assert!(rendered
            .windows(2)
            .any(|pair| pair == ["--model", "gpt-5.6-sol"]));
        assert!(rendered
            .windows(2)
            .any(|pair| pair == ["--config", "model_reasoning_effort=\"high\""]));
    }

    #[test]
    fn args_disable_agentic_and_tool_surfaces() {
        let spec = parse_model_spec("gpt-5.6-sol:high").unwrap();
        let args = build_args(&spec, 1056, Path::new("/tmp/codex-test"));
        let rendered: Vec<String> = args
            .iter()
            .map(|arg| arg.to_string_lossy().into_owned())
            .collect();
        for feature in [
            "shell_tool",
            "multi_agent",
            "apps",
            "plugins",
            "code_mode_host",
            "standalone_web_search",
        ] {
            assert!(rendered
                .windows(2)
                .any(|pair| pair == ["--disable", feature]));
        }
        assert!(rendered.contains(&"--ignore-user-config".to_string()));
        assert!(rendered.contains(&"--ignore-rules".to_string()));
        assert!(rendered
            .windows(2)
            .any(|pair| pair == ["--sandbox", "read-only"]));
    }

    #[test]
    fn parses_answer_and_real_usage() {
        let stdout = br#"{"type":"thread.started","thread_id":"t"}
{"type":"turn.started"}
{"type":"item.completed","item":{"type":"reasoning","text":"hidden"}}
{"type":"item.completed","item":{"type":"agent_message","text":"newsroom answer"}}
{"type":"turn.completed","usage":{"input_tokens":1200,"cached_input_tokens":800,"cache_write_input_tokens":100,"output_tokens":92}}
"#;
        let output = parse_output(stdout, 272_000).unwrap();
        assert_eq!(output.text, "newsroom answer");
        assert_eq!(output.usage.prompt_tokens, 300);
        assert_eq!(output.usage.cache_read_input_tokens, 800);
        assert_eq!(output.usage.cache_creation_input_tokens, 100);
        assert_eq!(output.usage.completion_tokens, 92);
        assert_eq!(output.usage.total_tokens, 1292);
        assert_eq!(output.usage.context_window, 272_000);
    }

    #[test]
    fn rejects_tool_or_multi_turn_output() {
        let tool = br#"{"type":"turn.started"}
{"type":"item.started","item":{"type":"command_execution","command":"pwd"}}
{"type":"item.completed","item":{"type":"agent_message","text":"answer"}}
{"type":"turn.completed","usage":{"input_tokens":1,"output_tokens":1}}
"#;
        assert!(parse_output(tool, 1)
            .unwrap_err()
            .to_string()
            .contains("forbidden non-generation items"));

        let multi = br#"{"type":"turn.started"}
{"type":"turn.started"}
{"type":"item.completed","item":{"type":"agent_message","text":"answer"}}
{"type":"turn.completed","usage":{"input_tokens":1,"output_tokens":1}}
"#;
        assert!(parse_output(multi, 1)
            .unwrap_err()
            .to_string()
            .contains("exactly one turn"));
    }

    #[test]
    fn refuses_incomplete_usage_instead_of_inventing_zeroes() {
        let stdout = br#"{"type":"turn.started"}
{"type":"item.completed","item":{"type":"agent_message","text":"answer"}}
{"type":"turn.completed","usage":{"input_tokens":4}}
"#;
        assert!(parse_output(stdout, 1)
            .unwrap_err()
            .to_string()
            .contains("refusing to fabricate accounting"));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn child_never_receives_openai_api_key() {
        use std::os::unix::fs::PermissionsExt;

        let temp = tempfile::tempdir().unwrap();
        let fixture = temp.path().join("codex-fixture.sh");
        std::fs::write(
            &fixture,
            r#"#!/bin/sh
if [ -n "${OPENAI_API_KEY-}" ]; then
  echo 'OPENAI_API_KEY leaked' >&2
  exit 91
fi
cat >/dev/null
printf '%s\n' '{"type":"turn.started"}'
printf '%s\n' '{"type":"item.completed","item":{"type":"agent_message","text":"fixture answer"}}'
printf '%s\n' '{"type":"turn.completed","usage":{"input_tokens":7,"output_tokens":3}}'
"#,
        )
        .unwrap();
        std::fs::set_permissions(&fixture, std::fs::Permissions::from_mode(0o700)).unwrap();

        let original_api_key = std::env::var_os("OPENAI_API_KEY");
        // SAFETY: process-environment mutation is confined to this child-spawn
        // regression and restored below. Other tests in this crate already use
        // the same scoped pattern for credential routing.
        unsafe { std::env::set_var("OPENAI_API_KEY", "must-not-reach-child") };
        let result = generate_with_program(
            &fixture,
            "gpt-5.6-sol:high",
            "write a brief",
            None,
            1056,
            272_000,
            Duration::from_secs(5),
        )
        .await;
        match original_api_key {
            Some(value) => unsafe { std::env::set_var("OPENAI_API_KEY", value) },
            None => unsafe { std::env::remove_var("OPENAI_API_KEY") },
        }

        let output = result.unwrap();
        assert_eq!(output.text, "fixture answer");
        assert_eq!(output.usage.total_tokens, 10);
    }
}