car-external-agents 0.26.0

//! Per-task invocation of external CLI agents.
//!
//! Phase 2 stage 3: spawn `claude -p --output-format stream-json
//! --input-format stream-json`, write the user's task to stdin,
//! consume the event stream, and return an aggregate
//! [`InvokeResult`].
//!
//! ## Cost discipline
//!
//! Every live invocation burns the user's subscription quota
//! (Claude Code uses ~30K cache-creation tokens per cold call).
//! Tests in this module use mock event streams via
//! [`process_stream`] — never spawn the real binary in CI.
//! Integration tests against a real CLI live behind an env-gated
//! `--ignored` flag so contributors opt in deliberately.
//!
//! ## Tool execution
//!
//! Phase 2 stage 3 (this file): observe-only. The agent uses its
//! own built-in tool set; we capture every `tool_use` block in
//! the assistant's content stream and report the count, but we
//! don't yet round-trip them through CAR's policy + eventlog.
//! That governance layer lands in stage 4.

use crate::protocol::{
    parse_line, AssistantEventData, StreamEvent, SystemEventData, ToolUseRequest,
};

/// Callback fired for each [`StreamEvent`] as it's parsed off the
/// child's stdout. The default invocation path (FFI callers, the
/// one-shot daemon path) passes `None` — `process_stream` runs
/// silently and the [`InvokeResult`] is the only output. The
/// streaming JSON-RPC path (`agents.invoke_external { stream: true }`)
/// passes a closure that fans each event out to the host's WS as
/// an `agents.chat.event` notification — same channel supervised
/// agents use, so host UIs render a single streaming surface for
/// both kinds of agent.
///
/// `Sync` so the closure can be held across `.await` in
/// `process_stream`; `'static` so it can be cloned into the
/// runner task without lifetime threading.
pub type StreamEventEmitter = Arc<dyn Fn(StreamEvent) + Send + Sync + 'static>;

use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWriteExt};
use tokio::process::Command;

/// One image attached to an external-agent invocation. Carried as a
/// path to a file on the daemon's filesystem (the host writes the
/// image to a sidecar first), plus its media type. External CLIs are
/// local subprocesses, so a path is the natural hand-off — no base64
/// over the wire. The runner adapts per adapter: Claude Code reads the
/// bytes and inlines a base64 image block on stdin; Codex passes the
/// path via `--image`; Gemini references it with `@path` in the prompt.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ImageAttachment {
    /// Absolute path to the image file on the daemon's filesystem.
    pub path: PathBuf,
    /// IANA media type, e.g. `image/png`. Optional — when absent the
    /// runner infers it from the file extension (Claude Code needs it
    /// for the base64 block; the path-passing adapters ignore it).
    #[serde(default)]
    pub media_type: Option<String>,
}

/// Caller-supplied options for a single invocation.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct InvokeOptions {
    /// Working directory the agent runs in. Inherits the daemon's
    /// cwd when `None`.
    #[serde(default)]
    pub cwd: Option<PathBuf>,
    /// Allowed tool names. When provided, passed through as
    /// `--allowed-tools "<list>"` so the binary refuses anything
    /// outside the list. `None` = use the binary's default policy.
    /// `Some(vec![])` = deny every tool (text-only response).
    #[serde(default)]
    pub allowed_tools: Option<Vec<String>>,
    /// Cap on model turns. Maps to `--max-turns`. Cost-control
    /// guard for runaway loops.
    #[serde(default)]
    pub max_turns: Option<u32>,
    /// Hard deadline. Defaults to 5 minutes when `None`. Anything
    /// above an hour is clamped down to one hour to prevent
    /// stuck-process resource exhaustion.
    #[serde(default)]
    pub timeout_secs: Option<u64>,
    /// MCP server URL to inject via `--mcp-config`. When set, the
    /// runner writes a temporary `mcp-config.json` pointing the
    /// agent at this endpoint and adds the flag to the spawn args
    /// — typically the daemon's own `/mcp` endpoint, so every tool
    /// call the agent issues against the CAR namespace flows
    /// through the daemon's policy + memgine. The temp file lives
    /// in the OS temp dir for the duration of the spawn and is
    /// removed on drop.
    ///
    /// Phase 2 stage 4b first cut. When set with the daemon's MCP
    /// URL, claude has access to `memory_add_fact`,
    /// `memory_query`, `verify`, and the `skill_*` tools — all
    /// gated by car-server's runtime, all logged to the same
    /// memgine WS clients see.
    #[serde(default)]
    pub mcp_endpoint: Option<String>,
    /// Images to attach to the initial prompt. Empty/omitted for a
    /// text-only turn. Each adapter wires these to its CLI's native
    /// image input (see `ImageAttachment`). Adapters whose CLI has no
    /// image support ignore them.
    #[serde(default)]
    pub attachments: Vec<ImageAttachment>,
}

const DEFAULT_TIMEOUT_SECS: u64 = 300;
const MAX_TIMEOUT_SECS: u64 = 3600;

/// Aggregate outcome of one external-agent invocation.
///
/// Self-contained shape so this crate doesn't pull `car-multi`
/// just for `AgentOutput`. The WS layer in `car-server-core`
/// re-shapes this when callers want the canonical AgentOutput.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct InvokeResult {
    /// Final answer from the agent's `result` event. Empty when
    /// the run errored before producing one.
    pub answer: String,
    /// Stable session id from the `system.init` event, when present.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Number of model turns the agent took.
    #[serde(default)]
    pub turns: u32,
    /// Count of `tool_use` blocks the assistant emitted across all
    /// turns. Useful for cost / behavior auditing.
    #[serde(default)]
    pub tool_calls: u32,
    /// Each `tool_use` block the assistant emitted, in stream order.
    /// Phase 2 stage 4a: observation only — the agent executes its
    /// own built-in tools internally (Read/Edit/Bash/etc.), and we
    /// record what it did. Stage 4b will add full policy gating
    /// via a CAR-managed MCP server route, which requires a
    /// different invocation mode.
    #[serde(default)]
    pub tool_uses: Vec<ToolUseRequest>,
    /// Wall-clock duration the agent reported in its result event.
    /// `0` when the agent didn't reach a result event.
    #[serde(default)]
    pub duration_ms: u64,
    /// Would-be API cost in USD (subscription users don't pay this;
    /// reported for transparency).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub total_cost_usd: Option<f64>,
    /// Number of attached images the runner dropped before sending —
    /// because the path was unreadable, over the size cap, or (on the
    /// adapters that read the bytes) not a recognized image. `0` for a
    /// clean run. Surfaced so a host can warn the user instead of
    /// letting a vision turn silently answer with no image.
    #[serde(default, skip_serializing_if = "is_zero_u32")]
    pub dropped_attachments: u32,
    /// `true` when the agent reported `is_error` or we hit a
    /// transport error before a result.
    #[serde(default)]
    pub is_error: bool,
    /// Human-readable error context — populated only when
    /// `is_error` is `true` or no result event arrived.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

fn is_zero_u32(n: &u32) -> bool {
    *n == 0
}

#[derive(Debug, Error)]
pub enum InvokeError {
    #[error("subprocess spawn failed: {0}")]
    Spawn(String),
    #[error("subprocess I/O failed: {0}")]
    Io(String),
    #[error("subprocess timed out after {0}s")]
    Timeout(u64),
    #[error("stream produced malformed JSON: {0}")]
    BadJson(String),
}

/// Build the argv for `claude` based on the invocation options.
/// Pulled out so tests can assert the flag combinations without
/// spawning a process. When `mcp_config_path` is set, threads
/// `--mcp-config <path>` through so the agent loads the supplied
/// MCP server file at startup.
pub(crate) fn build_claude_args(
    opts: &InvokeOptions,
    mcp_config_path: Option<&std::path::Path>,
) -> Vec<String> {
    let mut args = vec![
        "-p".to_string(),
        "--output-format".to_string(),
        "stream-json".to_string(),
        "--input-format".to_string(),
        "stream-json".to_string(),
        "--verbose".to_string(),
    ];
    if let Some(allowed) = &opts.allowed_tools {
        args.push("--allowed-tools".to_string());
        // Empty allowlist denies every tool — pass an empty arg
        // explicitly so the binary doesn't think the flag was unset.
        args.push(allowed.join(" "));
    }
    if let Some(turns) = opts.max_turns {
        args.push("--max-turns".to_string());
        args.push(turns.to_string());
    }
    if let Some(path) = mcp_config_path {
        args.push("--mcp-config".to_string());
        args.push(path.to_string_lossy().into_owned());
    }
    args
}

/// Render the MCP config JSON Claude Code expects when
/// `--mcp-config` points at an HTTP-streamable endpoint. The server
/// name `"car"` is what shows up in claude's MCP listing — keep it
/// stable so users learn one name across the project.
pub(crate) fn build_mcp_config_json(endpoint: &str) -> String {
    let cfg = serde_json::json!({
        "mcpServers": {
            "car": {
                "type": "http",
                "url": endpoint,
            }
        }
    });
    cfg.to_string()
}

/// Hard cap on the bytes the runner will read from an attachment path.
/// A WS client supplies the path, so without this a pointer at a
/// multi-GB file would let it exhaust daemon memory (read + base64 +
/// JSON-serialize). Far above any real screenshot, far below dangerous.
const MAX_IMAGE_BYTES: u64 = 32 * 1024 * 1024;

/// Identify a supported image from its leading magic bytes. Returns the
/// canonical media type, or `None` when the bytes aren't a PNG/JPEG/
/// GIF/WebP. This is the load-bearing guard: the runner never
/// base64-inlines a file it can't positively identify as an image, so a
/// client can't point `path` at a secret (`~/.ssh/id_rsa`, an auth
/// token) and have the daemon read + exfiltrate it into the model
/// stream. Content — not the caller-supplied `media_type` or the file
/// extension — is authoritative.
fn sniff_image_media_type(bytes: &[u8]) -> Option<&'static str> {
    if bytes.starts_with(&[0x89, 0x50, 0x4E, 0x47]) {
        Some("image/png")
    } else if bytes.starts_with(&[0xFF, 0xD8, 0xFF]) {
        Some("image/jpeg")
    } else if bytes.starts_with(b"GIF87a") || bytes.starts_with(b"GIF89a") {
        Some("image/gif")
    } else if bytes.len() >= 12 && &bytes[0..4] == b"RIFF" && &bytes[8..12] == b"WEBP" {
        Some("image/webp")
    } else {
        None
    }
}

/// `true` when `path` is a regular file within the size cap. Cheap
/// (stat only, no read) — used by the path-passing adapters (Codex)
/// that hand the file to the CLI without the daemon reading it.
fn image_within_size_cap(path: &Path) -> bool {
    match std::fs::metadata(path) {
        Ok(m) if m.is_file() && m.len() <= MAX_IMAGE_BYTES => true,
        Ok(m) if m.len() > MAX_IMAGE_BYTES => {
            tracing::warn!(path = %path.display(), bytes = m.len(), "image attachment exceeds size cap; skipping");
            false
        }
        _ => false,
    }
}

/// Read an attachment's bytes with the size cap, then verify it's
/// actually an image by magic bytes. Returns `(bytes, media_type)` from
/// the sniffed type, or `None` (logged) when missing / oversized /
/// not-an-image. Used by the adapters that need the bytes (Claude
/// base64) or copy the file themselves (Gemini staging).
fn read_validated_image(path: &Path) -> Option<(Vec<u8>, &'static str)> {
    if !image_within_size_cap(path) {
        return None;
    }
    let bytes = match std::fs::read(path) {
        Ok(b) => b,
        Err(e) => {
            tracing::warn!(path = %path.display(), error = %e, "skipping unreadable image attachment");
            return None;
        }
    };
    match sniff_image_media_type(&bytes) {
        Some(mt) => Some((bytes, mt)),
        None => {
            tracing::warn!(path = %path.display(), "attachment is not a recognized image (png/jpeg/gif/webp); skipping");
            None
        }
    }
}

/// Build an Anthropic Messages base64 image block for one attachment.
/// Returns `None` (and logs) when the file can't be read, is too large,
/// or isn't a recognized image — so one bad attachment doesn't sink the
/// whole turn, and a non-image file is never inlined. The `media_type`
/// is taken from the sniffed content, not the caller's claim.
fn claude_image_block(att: &ImageAttachment) -> Option<serde_json::Value> {
    use base64::Engine as _;
    let (bytes, media_type) = read_validated_image(&att.path)?;
    let data = base64::engine::general_purpose::STANDARD.encode(&bytes);
    Some(serde_json::json!({
        "type": "image",
        "source": { "type": "base64", "media_type": media_type, "data": data },
    }))
}

/// Build the single-line user message JSON to write on stdin, plus the
/// count of images actually inlined.
///
/// Text-only turns keep the legacy string `content`. When images are
/// attached, `content` becomes the Anthropic Messages content array —
/// a leading text block (when the prompt is non-empty) followed by one
/// base64 image block per readable, recognized image. The returned
/// count lets the caller report how many attachments were dropped.
pub(crate) fn build_user_message(task: &str, attachments: &[ImageAttachment]) -> (String, usize) {
    let mut accepted = 0usize;
    let content = if attachments.is_empty() {
        serde_json::json!(task)
    } else {
        let mut blocks: Vec<serde_json::Value> = Vec::new();
        if !task.is_empty() {
            blocks.push(serde_json::json!({ "type": "text", "text": task }));
        }
        for att in attachments {
            if let Some(block) = claude_image_block(att) {
                blocks.push(block);
                accepted += 1;
            }
        }
        serde_json::json!(blocks)
    };
    let payload = serde_json::json!({
        "type": "user",
        "message": {
            "role": "user",
            "content": content,
        },
    });
    (payload.to_string(), accepted)
}

/// Top-level dispatch by adapter id. Looks up the detected
/// binary path, then dispatches to the per-tool runner.
///
/// Phase 2 stage 3 supports `"claude-code"` only. Calls with other
/// adapter ids return an `InvokeResult` with `is_error = true` and
/// a structured `error` so the WS layer can surface the gap to
/// hosts without a separate error code.
pub async fn invoke(
    id: &str,
    task: &str,
    opts: InvokeOptions,
) -> Result<InvokeResult, InvokeError> {
    invoke_with_emitter(id, task, opts, None).await
}

/// `invoke` plus an optional [`StreamEventEmitter`]. All three
/// adapters fire events through it (car#213):
/// - `claude-code` emits native stream-json events as they arrive.
/// - `codex` synthesizes Anthropic-Messages-shaped `Assistant`
///   events from its JSONL `item.completed` lines so the host's
///   chat renderer fills the bubble incrementally.
/// - `gemini` is text-only in 0.1.x (no event stream), so it emits
///   a single `Assistant` text event with the full reply once the
///   process exits — a single-pop reply rather than a token stream,
///   but the host sees the answer instead of an empty bubble.
pub async fn invoke_with_emitter(
    id: &str,
    task: &str,
    opts: InvokeOptions,
    emitter: Option<StreamEventEmitter>,
) -> Result<InvokeResult, InvokeError> {
    let detected = crate::detect().await;
    let spec = detected
        .iter()
        .find(|s| s.id == id)
        .ok_or_else(|| InvokeError::Spawn(format!("no detected external agent with id `{id}`")))?;
    match id {
        "claude-code" => invoke_claude_code(&spec.binary_path, task, opts, emitter).await,
        "codex" => invoke_codex(&spec.binary_path, task, opts, emitter).await,
        "gemini" => invoke_gemini(&spec.binary_path, task, opts, emitter).await,
        _ => Err(InvokeError::Spawn(format!("unknown adapter id: {id}"))),
    }
}

/// Spawn `binary_path` with `claude -p stream-json` flags, write
/// `task` as a user message, and consume the event stream until
/// the binary exits or the timeout fires.
///
/// `binary_path` is treated as already-validated — caller is
/// expected to use the path the detection module resolved from
/// `$PATH`, which has already passed the scratch-dir denylist.
pub async fn invoke_claude_code(
    binary_path: &Path,
    task: &str,
    opts: InvokeOptions,
    emitter: Option<StreamEventEmitter>,
) -> Result<InvokeResult, InvokeError> {
    tracing::info!(
        adapter = "claude-code",
        binary = %binary_path.display(),
        task_len = task.len(),
        "external agent invocation started"
    );
    let timeout_secs = opts
        .timeout_secs
        .unwrap_or(DEFAULT_TIMEOUT_SECS)
        .min(MAX_TIMEOUT_SECS)
        .max(1);
    let timeout = Duration::from_secs(timeout_secs);

    // Write the MCP config to a temp file when an endpoint was
    // supplied. Held in a `_mcp_config` binding for the lifetime of
    // the invocation so its `Drop` runs *after* claude exits — same
    // RAII shape as `tempfile::NamedTempFile`. An explicit empty
    // string is treated as "opt out of MCP" (the daemon caller uses
    // it to override the auto-fill).
    let mcp_config: Option<tempfile::NamedTempFile> = match opts.mcp_endpoint.as_deref() {
        Some(endpoint) if !endpoint.is_empty() => {
            let json = build_mcp_config_json(endpoint);
            let mut tmp = tempfile::Builder::new()
                .prefix("car-mcp-config-")
                .suffix(".json")
                .tempfile()
                .map_err(|e| InvokeError::Io(format!("mcp config tempfile: {e}")))?;
            std::io::Write::write_all(&mut tmp, json.as_bytes())
                .map_err(|e| InvokeError::Io(format!("mcp config write: {e}")))?;
            Some(tmp)
        }
        _ => None,
    };
    let mcp_config_path = mcp_config.as_ref().map(|t| t.path());

    let args = build_claude_args(&opts, mcp_config_path);
    let mut cmd = Command::new(binary_path);
    cmd.args(&args);
    if let Some(cwd) = &opts.cwd {
        cmd.current_dir(cwd);
    }
    cmd.stdin(std::process::Stdio::piped());
    cmd.stdout(std::process::Stdio::piped());
    cmd.stderr(std::process::Stdio::piped());
    cmd.kill_on_drop(true);

    let mut child = cmd.spawn().map_err(|e| InvokeError::Spawn(e.to_string()))?;

    // Build the stdin message up front so we know how many images were
    // actually inlined (vs. dropped as unreadable / oversized / not an
    // image). Reported on the result so the host can warn the user.
    let (user_message, accepted_images) = build_user_message(task, &opts.attachments);
    let dropped_attachments = (opts.attachments.len().saturating_sub(accepted_images)) as u32;

    // Write the task on stdin and close so the agent sees EOF.
    {
        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| InvokeError::Io("stdin closed unexpectedly".to_string()))?;
        let mut stdin = stdin;
        stdin
            .write_all(user_message.as_bytes())
            .await
            .map_err(|e| InvokeError::Io(format!("stdin write: {e}")))?;
        stdin
            .write_all(b"\n")
            .await
            .map_err(|e| InvokeError::Io(format!("stdin newline: {e}")))?;
        // Drop closes stdin; agent now sees EOF and starts processing.
    }

    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| InvokeError::Io("stdout missing".to_string()))?;
    let stderr = child
        .stderr
        .take()
        .ok_or_else(|| InvokeError::Io("stderr missing".to_string()))?;
    let reader = tokio::io::BufReader::new(stdout);

    let process_fut = async {
        let mut result = process_stream(reader, emitter).await?;
        result.dropped_attachments = dropped_attachments;
        // Wait for the child so we don't orphan it.
        let exit = child
            .wait()
            .await
            .map_err(|e| InvokeError::Io(format!("wait: {e}")))?;
        if !exit.success() && !result.is_error {
            // Capture stderr for context — don't let a non-zero
            // exit pass silently when the stream itself didn't
            // surface an error.
            let mut stderr_buf = Vec::new();
            let mut stderr_reader = tokio::io::BufReader::new(stderr);
            let _ = tokio::io::AsyncReadExt::read_to_end(&mut stderr_reader, &mut stderr_buf).await;
            let stderr_text = String::from_utf8_lossy(&stderr_buf).to_string();
            result.is_error = true;
            result.error = Some(format!(
                "exit code {}: {}",
                exit.code().unwrap_or(-1),
                stderr_text.trim()
            ));
        }
        Ok::<_, InvokeError>(result)
    };

    match tokio::time::timeout(timeout, process_fut).await {
        Ok(Ok(res)) => Ok(res),
        Ok(Err(e)) => Err(e),
        Err(_) => Err(InvokeError::Timeout(timeout_secs)),
    }
}

/// Consume the `claude` stream-json output line-by-line and
/// aggregate it into an [`InvokeResult`]. Pulled out from
/// [`invoke_claude_code`] so it can be unit-tested with mock
/// readers (avoids burning subscription quota on every test run).
pub async fn process_stream<R>(
    reader: R,
    emitter: Option<StreamEventEmitter>,
) -> Result<InvokeResult, InvokeError>
where
    R: AsyncBufRead + Unpin,
{
    let mut result = InvokeResult::default();
    let mut lines = reader.lines();
    loop {
        let line = match lines.next_line().await {
            Ok(Some(line)) => line,
            Ok(None) => break,
            Err(e) => return Err(InvokeError::Io(format!("stdout read: {e}"))),
        };
        let event = match parse_line(&line) {
            Ok(Some(e)) => e,
            Ok(None) => continue,
            Err(e) => {
                // One bad line shouldn't kill the whole stream —
                // log it via the error field if it's the only thing
                // we got, but keep reading so a benign trailing
                // newline doesn't cost us the result.
                if result.answer.is_empty() && result.error.is_none() {
                    result.error = Some(format!("malformed JSON: {e}"));
                }
                continue;
            }
        };
        // Fire the emitter (if any) BEFORE apply_event consumes the
        // value into the running result. Clone so apply_event still
        // owns its copy — StreamEvent is small enough that the clone
        // cost is irrelevant next to the WS round-trip.
        if let Some(e) = &emitter {
            e(event.clone());
        }
        apply_event(&mut result, event);
    }
    Ok(result)
}

fn apply_event(result: &mut InvokeResult, event: StreamEvent) {
    match event {
        StreamEvent::System(s) => {
            // Capture session id on the first system event we see;
            // subsequent system events keep the original.
            if result.session_id.is_none() {
                result.session_id = Some(s.session_id);
            }
        }
        StreamEvent::Assistant(a) => {
            result.turns = result.turns.saturating_add(1);
            // Count + record tool_use blocks. tool_uses is the
            // post-run audit trail; tool_calls is the cheap
            // count for cost-control summaries.
            if let Some(content) = a.message.get("content").and_then(|v| v.as_array()) {
                for block in content {
                    if block.get("type").and_then(|v| v.as_str()) == Some("tool_use") {
                        result.tool_calls = result.tool_calls.saturating_add(1);
                        let id = block
                            .get("id")
                            .and_then(|v| v.as_str())
                            .unwrap_or("")
                            .to_string();
                        let name = block
                            .get("name")
                            .and_then(|v| v.as_str())
                            .unwrap_or("")
                            .to_string();
                        let input = block
                            .get("input")
                            .cloned()
                            .unwrap_or(serde_json::Value::Null);
                        tracing::info!(
                            adapter = "claude-code",
                            tool_name = %name,
                            "external agent emitted tool_use (observation-only — \
                             policy gating via MCP route lands in Phase 2 stage 4b)"
                        );
                        result.tool_uses.push(ToolUseRequest { id, name, input });
                    }
                }
            }
        }
        StreamEvent::Result(r) => {
            result.answer = r.result.unwrap_or_default();
            result.duration_ms = r.duration_ms.unwrap_or(0);
            result.total_cost_usd = r.total_cost_usd;
            result.is_error = r.is_error;
            // Prefer the result event's view of turns when it
            // disagrees with our running count (the agent may
            // count internal turns we don't see).
            if let Some(t) = r.num_turns {
                result.turns = t;
            }
            if r.is_error {
                result.error = Some(format!(
                    "agent reported error (subtype={}, terminal={:?})",
                    r.subtype, r.terminal_reason
                ));
            }
        }
        StreamEvent::User(_) | StreamEvent::RateLimitEvent(_) | StreamEvent::Other => {
            // Not load-bearing for the aggregate result.
        }
    }
}

// --- Codex (OpenAI codex-cli) -----------------------------------------------

/// Build the argv for `codex exec --json`.
///
/// `codex exec --json --skip-git-repo-check --ephemeral` runs
/// non-interactively, emits NDJSON events on stdout, and doesn't
/// persist a session file (matching the per-task semantics this
/// runner expects). When `mcp_config_path` is set we inject the
/// CAR namespace via `-c` overrides — Codex stores MCP servers in
/// TOML config under `mcp_servers.<name>` keys, and the inline
/// override syntax accepts the same dotted path.
pub(crate) fn build_codex_args(opts: &InvokeOptions, mcp_endpoint: Option<&str>) -> Vec<String> {
    let mut args = vec![
        "exec".to_string(),
        "--json".to_string(),
        "--skip-git-repo-check".to_string(),
        "--ephemeral".to_string(),
    ];
    if let Some(cwd) = &opts.cwd {
        args.push("--cd".to_string());
        args.push(cwd.to_string_lossy().into_owned());
    }
    if let Some(endpoint) = mcp_endpoint.filter(|s| !s.is_empty()) {
        // Codex's `-c` flag takes `key=value` where the value is
        // parsed as TOML; an inline-table is the standard shape for
        // `mcp_servers.<name>`. The `type="http"` + URL form mirrors
        // Codex's own MCP config docs.
        let value = format!(r#"{{type="http",url="{}"}}"#, endpoint);
        args.push("-c".to_string());
        args.push(format!("mcp_servers.car={}", value));
    }
    // Attach images via `--image <FILE>` (codex `exec -i/--image`,
    // repeatable). Paths only — codex reads + decodes the files itself,
    // so we don't sniff content here; just skip missing/oversized paths.
    for att in &opts.attachments {
        if image_within_size_cap(&att.path) {
            args.push("--image".to_string());
            args.push(att.path.to_string_lossy().into_owned());
        }
    }
    // The prompt itself goes on stdin (we use `-` as the prompt
    // arg so codex reads stdin even when stdin is a pipe rather
    // than a TTY). Chosen over passing the prompt as a CLI arg so
    // long prompts don't trip the OS argv limit.
    args.push("-".to_string());
    args
}

/// Run codex with `task` piped on stdin, parse the JSONL event
/// stream, return an aggregate [`InvokeResult`].
///
/// Codex events observed (2026-05, codex-cli 0.128.0):
/// - `thread.started` — session id
/// - `turn.started` / `turn.completed` — turn boundaries + usage
/// - `item.completed` with `item.type = "agent_message"` — answer
///   text. Multiple items may arrive; we concatenate.
/// - Other `item.completed` shapes (tool calls, etc.) get counted
///   in `tool_calls` and recorded under `tool_uses` so the audit
///   trail matches the Claude Code adapter's shape.
pub async fn invoke_codex(
    binary_path: &Path,
    task: &str,
    opts: InvokeOptions,
    emitter: Option<StreamEventEmitter>,
) -> Result<InvokeResult, InvokeError> {
    tracing::info!(
        adapter = "codex",
        binary = %binary_path.display(),
        task_len = task.len(),
        "external agent invocation started"
    );
    let timeout_secs = opts
        .timeout_secs
        .unwrap_or(DEFAULT_TIMEOUT_SECS)
        .min(MAX_TIMEOUT_SECS)
        .max(1);
    let timeout = Duration::from_secs(timeout_secs);

    let args = build_codex_args(&opts, opts.mcp_endpoint.as_deref());
    // How many images survived the size/existence filter in
    // build_codex_args (each accepted one adds a `--image` arg).
    let accepted_images = args.iter().filter(|a| a.as_str() == "--image").count();
    let dropped_attachments = (opts.attachments.len().saturating_sub(accepted_images)) as u32;
    let mut cmd = Command::new(binary_path);
    cmd.args(&args);
    if let Some(cwd) = &opts.cwd {
        cmd.current_dir(cwd);
    }
    cmd.stdin(std::process::Stdio::piped());
    cmd.stdout(std::process::Stdio::piped());
    cmd.stderr(std::process::Stdio::piped());
    cmd.kill_on_drop(true);

    let mut child = cmd.spawn().map_err(|e| InvokeError::Spawn(e.to_string()))?;

    {
        let mut stdin = child
            .stdin
            .take()
            .ok_or_else(|| InvokeError::Io("stdin closed unexpectedly".to_string()))?;
        stdin
            .write_all(task.as_bytes())
            .await
            .map_err(|e| InvokeError::Io(format!("stdin write: {e}")))?;
    }

    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| InvokeError::Io("stdout missing".to_string()))?;
    let stderr = child
        .stderr
        .take()
        .ok_or_else(|| InvokeError::Io("stderr missing".to_string()))?;
    let reader = tokio::io::BufReader::new(stdout);

    let process_fut = async {
        let mut result = process_codex_stream(reader, emitter).await?;
        result.dropped_attachments = dropped_attachments;
        let exit = child
            .wait()
            .await
            .map_err(|e| InvokeError::Io(format!("wait: {e}")))?;
        if !exit.success() && !result.is_error {
            let mut stderr_buf = Vec::new();
            let mut stderr_reader = tokio::io::BufReader::new(stderr);
            let _ = tokio::io::AsyncReadExt::read_to_end(&mut stderr_reader, &mut stderr_buf).await;
            let stderr_text = String::from_utf8_lossy(&stderr_buf).to_string();
            result.is_error = true;
            result.error = Some(format!(
                "exit code {}: {}",
                exit.code().unwrap_or(-1),
                stderr_text.trim()
            ));
        }
        Ok::<_, InvokeError>(result)
    };

    match tokio::time::timeout(timeout, process_fut).await {
        Ok(Ok(res)) => Ok(res),
        Ok(Err(e)) => Err(e),
        Err(_) => Err(InvokeError::Timeout(timeout_secs)),
    }
}

/// Build a synthetic `assistant` [`StreamEvent`] wrapping a single
/// content block. Codex and Gemini don't emit Anthropic-Messages-
/// shaped frames natively, but the host's chat renderer keys off
/// `Assistant.message.content[]` (`text` → `kind:"token"`,
/// `tool_use` → `kind:"tool_call"`). Synthesizing that shape keeps
/// the streaming surface uniform across all three external CLIs so
/// hosts render every adapter with one decoder (car#213). `uuid` is
/// left empty — no emitter consumer reads it for synthesized events,
/// and these lines have no upstream uuid to carry.
fn synth_assistant_event(session_id: &str, content: serde_json::Value) -> StreamEvent {
    StreamEvent::Assistant(AssistantEventData {
        message: serde_json::json!({ "role": "assistant", "content": [content] }),
        session_id: session_id.to_string(),
        uuid: String::new(),
        parent_tool_use_id: None,
    })
}

/// Codex stream processor, split out for mock-stream testing.
///
/// `emitter`, when present, fires per parsed JSONL line *in addition
/// to* the running aggregate — same emit-before-aggregate ordering
/// `process_stream` uses for claude-code. Pre-car#213 the emitter
/// was dropped here and the host only ever saw the synthesized
/// terminal `done` (empty bubble).
pub async fn process_codex_stream<R>(
    reader: R,
    emitter: Option<StreamEventEmitter>,
) -> Result<InvokeResult, InvokeError>
where
    R: tokio::io::AsyncBufRead + Unpin,
{
    use serde_json::Value;

    let started = std::time::Instant::now();
    let mut result = InvokeResult::default();
    let mut answer_parts: Vec<String> = Vec::new();
    let mut lines = reader.lines();

    loop {
        let line = match lines.next_line().await {
            Ok(Some(line)) => line,
            Ok(None) => break,
            Err(e) => return Err(InvokeError::Io(format!("stdout read: {e}"))),
        };
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        let value: Value = match serde_json::from_str(trimmed) {
            Ok(v) => v,
            Err(_) => {
                // codex emits an "Reading prompt from stdin..."
                // banner before the JSON stream — skip non-JSON
                // lines silently rather than treating them as
                // errors. Same defensive shape as the claude
                // adapter's malformed-line handling.
                continue;
            }
        };
        let kind = value.get("type").and_then(Value::as_str).unwrap_or("");
        match kind {
            "thread.started" => {
                if let Some(id) = value.get("thread_id").and_then(Value::as_str) {
                    if result.session_id.is_none() {
                        result.session_id = Some(id.to_string());
                    }
                }
                if let Some(e) = &emitter {
                    e(StreamEvent::System(SystemEventData {
                        subtype: "init".to_string(),
                        session_id: result.session_id.clone().unwrap_or_default(),
                        model: None,
                        cwd: None,
                        tools: Vec::new(),
                        permission_mode: None,
                        claude_code_version: None,
                        extra: serde_json::Map::new(),
                    }));
                }
            }
            "turn.started" => {
                result.turns = result.turns.saturating_add(1);
            }
            "item.completed" => {
                let Some(item) = value.get("item") else {
                    continue;
                };
                let item_type = item.get("type").and_then(Value::as_str).unwrap_or("");
                match item_type {
                    "agent_message" => {
                        if let Some(text) = item.get("text").and_then(Value::as_str) {
                            if let Some(e) = &emitter {
                                let sid = result.session_id.clone().unwrap_or_default();
                                e(synth_assistant_event(
                                    &sid,
                                    serde_json::json!({ "type": "text", "text": text }),
                                ));
                            }
                            answer_parts.push(text.to_string());
                        }
                    }
                    other if other.contains("tool") || other.contains("call") => {
                        result.tool_calls = result.tool_calls.saturating_add(1);
                        let id = item
                            .get("id")
                            .and_then(Value::as_str)
                            .unwrap_or("")
                            .to_string();
                        // Codex tool-call items typically carry a
                        // `name` and structured `arguments`; fall
                        // back to the item type when name is
                        // missing so the audit trail isn't blank.
                        let name = item
                            .get("name")
                            .and_then(Value::as_str)
                            .unwrap_or(other)
                            .to_string();
                        let input = item
                            .get("arguments")
                            .or_else(|| item.get("input"))
                            .cloned()
                            .unwrap_or(Value::Null);
                        tracing::info!(
                            adapter = "codex",
                            tool_name = %name,
                            "external agent emitted tool_use (observation-only)"
                        );
                        if let Some(e) = &emitter {
                            let sid = result.session_id.clone().unwrap_or_default();
                            e(synth_assistant_event(
                                &sid,
                                serde_json::json!({
                                    "type": "tool_use",
                                    "id": id,
                                    "name": name,
                                    "input": input,
                                }),
                            ));
                        }
                        result.tool_uses.push(ToolUseRequest { id, name, input });
                    }
                    _ => {}
                }
            }
            "turn.completed" => {
                // No fields needed for the aggregate today; cost
                // accounting could be threaded here if codex
                // exposes USD cost in a future release.
            }
            _ => {}
        }
    }

    result.answer = answer_parts.join("");
    result.duration_ms = started.elapsed().as_millis() as u64;
    if result.answer.is_empty() && result.error.is_none() {
        // No agent_message arrived but the stream completed —
        // mark as error so callers don't silently treat empty
        // strings as success.
        result.is_error = true;
        result.error = Some("codex produced no agent_message".to_string());
    }
    Ok(result)
}

// --- Gemini CLI -------------------------------------------------------------

/// Build the argv for Gemini CLI.
///
/// Gemini in 0.1.x doesn't expose a JSON-stream output mode, so the
/// adapter is text-only: `gemini -p "<task>" --yolo` runs
/// non-interactively, auto-accepts actions, and prints the final
/// response to stdout. `--yolo` is required for unattended runs;
/// without it the binary blocks waiting for interactive approvals.
///
/// MCP injection is **not yet supported on Gemini** (the CLI in
/// this version doesn't expose an `--mcp-config` equivalent). When
/// `mcp_endpoint` is set, the runner emits a tracing warning and
/// proceeds without MCP — Gemini operates with its own built-in
/// tools, no CAR-namespace tools available.
pub(crate) fn build_gemini_args(opts: &InvokeOptions, task: &str, image_refs: &[String]) -> Vec<String> {
    // Gemini includes files via `@path` references inside the prompt
    // (its multimodal-context mechanism — there's no `--image` flag).
    // The refs are pre-staged into a space-free temp dir by the caller
    // (`stage_gemini_images`), because `@` is whitespace-delimited and
    // CarHost's sidecar dir lives under a path with spaces.
    let mut prompt = task.to_string();
    for r in image_refs {
        if !prompt.is_empty() {
            prompt.push(' ');
        }
        prompt.push_str(&format!("@{r}"));
    }
    let args = vec!["-p".to_string(), prompt, "--yolo".to_string()];
    if let Some(cwd) = &opts.cwd {
        // Gemini doesn't have a `--cd` flag. We honor cwd via
        // tokio::process::Command::current_dir below; no argv
        // entry needed. Pulling cwd into a no-op path here just
        // documents the gap.
        let _ = cwd;
    }
    args
}

/// Copy image attachments into a fresh temp dir with safe, space-free
/// names so Gemini's whitespace-delimited `@path` references resolve.
/// Returns the temp dir (which must be kept alive for the spawn — its
/// `Drop` removes the files) and the absolute staged paths. Unreadable
/// attachments are skipped.
fn stage_gemini_images(
    attachments: &[ImageAttachment],
) -> (Option<tempfile::TempDir>, Vec<String>) {
    if attachments.is_empty() {
        return (None, Vec::new());
    }
    let dir = match tempfile::tempdir() {
        Ok(d) => d,
        Err(e) => {
            tracing::warn!(error = %e, "could not create temp dir for gemini images; sending text only");
            return (None, Vec::new());
        }
    };
    let mut refs = Vec::new();
    for (i, att) in attachments.iter().enumerate() {
        // Validate (size cap + magic-byte sniff) before staging so we
        // don't copy a huge or non-image file the client pointed us at.
        let Some((bytes, media_type)) = read_validated_image(&att.path) else {
            continue;
        };
        let ext = match media_type {
            "image/jpeg" => "jpg",
            "image/gif" => "gif",
            "image/webp" => "webp",
            _ => "png",
        };
        let dest = dir.path().join(format!("img{i}.{ext}"));
        match std::fs::write(&dest, &bytes) {
            Ok(_) => refs.push(dest.to_string_lossy().into_owned()),
            Err(e) => {
                tracing::warn!(path = %dest.display(), error = %e, "could not stage image for gemini; skipping");
            }
        }
    }
    (Some(dir), refs)
}

/// Run gemini with `-p task --yolo`, capture stdout, return an
/// aggregate [`InvokeResult`]. Gemini's text-only output means
/// `tool_calls`, `tool_uses`, and `total_cost_usd` are always
/// empty — the agent's actions aren't visible at the adapter
/// boundary in this version.
pub async fn invoke_gemini(
    binary_path: &Path,
    task: &str,
    opts: InvokeOptions,
    emitter: Option<StreamEventEmitter>,
) -> Result<InvokeResult, InvokeError> {
    tracing::info!(
        adapter = "gemini",
        binary = %binary_path.display(),
        task_len = task.len(),
        "external agent invocation started"
    );
    if opts
        .mcp_endpoint
        .as_deref()
        .filter(|s| !s.is_empty())
        .is_some()
    {
        tracing::warn!(
            adapter = "gemini",
            "mcp_endpoint supplied but Gemini CLI v0.1.x doesn't support \
             --mcp-config; agent will run without CAR's MCP namespace"
        );
    }

    let timeout_secs = opts
        .timeout_secs
        .unwrap_or(DEFAULT_TIMEOUT_SECS)
        .min(MAX_TIMEOUT_SECS)
        .max(1);
    let timeout = Duration::from_secs(timeout_secs);

    let started = std::time::Instant::now();
    // Stage images into a space-free temp dir and reference them with
    // `@path`. `_image_stage` holds the TempDir alive until the spawn
    // finishes (its Drop deletes the staged copies).
    let (_image_stage, image_refs) = stage_gemini_images(&opts.attachments);
    let dropped_attachments = (opts.attachments.len().saturating_sub(image_refs.len())) as u32;
    let args = build_gemini_args(&opts, task, &image_refs);
    let mut cmd = Command::new(binary_path);
    cmd.args(&args);
    if let Some(cwd) = &opts.cwd {
        cmd.current_dir(cwd);
    }
    cmd.stdin(std::process::Stdio::null());
    cmd.stdout(std::process::Stdio::piped());
    cmd.stderr(std::process::Stdio::piped());
    cmd.kill_on_drop(true);

    let child = cmd.spawn().map_err(|e| InvokeError::Spawn(e.to_string()))?;

    let process_fut = async {
        let output = child
            .wait_with_output()
            .await
            .map_err(|e| InvokeError::Io(format!("wait: {e}")))?;
        let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
        let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
        let mut result = InvokeResult {
            answer: stdout.trim().to_string(),
            duration_ms: started.elapsed().as_millis() as u64,
            dropped_attachments,
            ..Default::default()
        };
        if !output.status.success() {
            result.is_error = true;
            result.error = Some(format!(
                "exit code {}: {}",
                output.status.code().unwrap_or(-1),
                stderr.trim()
            ));
        } else if result.answer.is_empty() {
            result.is_error = true;
            result.error = Some("gemini produced no stdout output".to_string());
        } else if let Some(e) = &emitter {
            // Gemini 0.1.x has no event-stream output mode (see
            // build_gemini_args), so there are no incremental tokens
            // to forward. Surface the whole reply as one synthetic
            // Assistant text event so the host bubble fills with the
            // answer (single-pop, not a token stream) instead of
            // staying empty. car#213.
            e(synth_assistant_event(
                "",
                serde_json::json!({ "type": "text", "text": result.answer.clone() }),
            ));
        }
        Ok::<_, InvokeError>(result)
    };

    match tokio::time::timeout(timeout, process_fut).await {
        Ok(Ok(res)) => Ok(res),
        Ok(Err(e)) => Err(e),
        Err(_) => Err(InvokeError::Timeout(timeout_secs)),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Cursor;
    use tokio::io::BufReader;

    fn mock_reader(lines: &[&str]) -> BufReader<Cursor<Vec<u8>>> {
        let joined = lines.join("\n");
        BufReader::new(Cursor::new(joined.into_bytes()))
    }

    #[tokio::test]
    async fn aggregates_simple_text_response() {
        let lines = [
            r#"{"type":"system","subtype":"init","session_id":"s1","model":"opus","tools":[]}"#,
            r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"ok"}],"usage":{}},"session_id":"s1","uuid":"u1"}"#,
            r#"{"type":"result","subtype":"success","is_error":false,"duration_ms":1500,"num_turns":1,"result":"ok","session_id":"s1","total_cost_usd":0.05,"usage":{},"modelUsage":{},"uuid":"r1"}"#,
        ];
        let result = process_stream(mock_reader(&lines), None).await.unwrap();
        assert_eq!(result.answer, "ok");
        assert_eq!(result.session_id.as_deref(), Some("s1"));
        assert_eq!(result.turns, 1);
        assert_eq!(result.tool_calls, 0);
        assert_eq!(result.duration_ms, 1500);
        assert_eq!(result.total_cost_usd, Some(0.05));
        assert!(!result.is_error);
        assert!(result.error.is_none());
    }

    #[tokio::test]
    async fn counts_tool_use_blocks_across_turns() {
        let lines = [
            r#"{"type":"system","subtype":"init","session_id":"s2","model":"opus","tools":[]}"#,
            r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"t1","name":"Read","input":{"file_path":"/x"}}],"usage":{}},"session_id":"s2","uuid":"u1"}"#,
            r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"t2","name":"Bash","input":{"command":"ls"}},{"type":"text","text":"done"}],"usage":{}},"session_id":"s2","uuid":"u2"}"#,
            r#"{"type":"result","subtype":"success","is_error":false,"duration_ms":3000,"num_turns":2,"result":"done","session_id":"s2","total_cost_usd":0.10,"usage":{},"modelUsage":{},"uuid":"r1"}"#,
        ];
        let result = process_stream(mock_reader(&lines), None).await.unwrap();
        assert_eq!(result.tool_calls, 2);
        assert_eq!(result.turns, 2);
        assert_eq!(result.answer, "done");
        // tool_uses captures the full audit trail in stream order.
        assert_eq!(result.tool_uses.len(), 2);
        assert_eq!(result.tool_uses[0].id, "t1");
        assert_eq!(result.tool_uses[0].name, "Read");
        assert_eq!(
            result.tool_uses[0]
                .input
                .get("file_path")
                .and_then(|v| v.as_str()),
            Some("/x")
        );
        assert_eq!(result.tool_uses[1].id, "t2");
        assert_eq!(result.tool_uses[1].name, "Bash");
    }

    #[tokio::test]
    async fn surfaces_agent_reported_error() {
        let lines = [
            r#"{"type":"system","subtype":"init","session_id":"s3","model":"opus","tools":[]}"#,
            r#"{"type":"result","subtype":"error","is_error":true,"duration_ms":500,"session_id":"s3","total_cost_usd":0.0,"usage":{},"modelUsage":{},"terminal_reason":"timeout","uuid":"r1"}"#,
        ];
        let result = process_stream(mock_reader(&lines), None).await.unwrap();
        assert!(result.is_error);
        assert!(result.error.as_deref().unwrap().contains("error"));
    }

    #[tokio::test]
    async fn empty_stream_produces_empty_result_with_no_panic() {
        let result = process_stream(mock_reader(&[]), None).await.unwrap();
        assert_eq!(result.answer, "");
        assert_eq!(result.turns, 0);
        // No result event → not technically an error in the agent's
        // view, but caller can detect via empty answer + missing
        // session id.
        assert!(result.session_id.is_none());
    }

    #[tokio::test]
    async fn malformed_line_logged_but_stream_continues() {
        let lines = [
            r#"{not valid"#,
            r#"{"type":"system","subtype":"init","session_id":"s4","model":"opus","tools":[]}"#,
            r#"{"type":"result","subtype":"success","is_error":false,"duration_ms":100,"num_turns":0,"result":"recovered","session_id":"s4","total_cost_usd":0.0,"usage":{},"modelUsage":{},"uuid":"r1"}"#,
        ];
        let result = process_stream(mock_reader(&lines), None).await.unwrap();
        assert_eq!(result.answer, "recovered");
        assert_eq!(result.session_id.as_deref(), Some("s4"));
    }

    #[test]
    fn build_claude_args_includes_required_format_flags() {
        let args = build_claude_args(&InvokeOptions::default(), None);
        assert!(args.contains(&"-p".to_string()));
        assert!(args.iter().any(|a| a == "stream-json"));
        // Default: no --allowed-tools, no --max-turns, no --mcp-config.
        assert!(!args.iter().any(|a| a == "--allowed-tools"));
        assert!(!args.iter().any(|a| a == "--max-turns"));
        assert!(!args.iter().any(|a| a == "--mcp-config"));
    }

    #[test]
    fn build_claude_args_passes_allowed_tools() {
        let opts = InvokeOptions {
            allowed_tools: Some(vec!["Read".to_string(), "Bash".to_string()]),
            max_turns: Some(5),
            ..Default::default()
        };
        let args = build_claude_args(&opts, None);
        let pos = args.iter().position(|a| a == "--allowed-tools").unwrap();
        assert_eq!(args[pos + 1], "Read Bash");
        let pos = args.iter().position(|a| a == "--max-turns").unwrap();
        assert_eq!(args[pos + 1], "5");
    }

    #[test]
    fn build_claude_args_empty_allowed_tools_denies_everything() {
        let opts = InvokeOptions {
            allowed_tools: Some(vec![]),
            ..Default::default()
        };
        let args = build_claude_args(&opts, None);
        let pos = args.iter().position(|a| a == "--allowed-tools").unwrap();
        // Empty list still produces a flag with empty value so the
        // binary knows the policy is "deny all", not "use default".
        assert_eq!(args[pos + 1], "");
    }

    #[test]
    fn build_claude_args_threads_mcp_config_path() {
        let path = std::path::Path::new("/tmp/mcp-config.json");
        let args = build_claude_args(&InvokeOptions::default(), Some(path));
        let pos = args.iter().position(|a| a == "--mcp-config").unwrap();
        assert_eq!(args[pos + 1], "/tmp/mcp-config.json");
    }

    #[test]
    fn build_mcp_config_renders_http_server_entry() {
        let json = build_mcp_config_json("http://127.0.0.1:9102/mcp");
        let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
        assert_eq!(parsed["mcpServers"]["car"]["type"], "http");
        assert_eq!(
            parsed["mcpServers"]["car"]["url"],
            "http://127.0.0.1:9102/mcp"
        );
    }

    #[test]
    fn build_user_message_is_valid_json() {
        let (msg, _accepted) = build_user_message("hello world", &[]);
        let parsed: serde_json::Value = serde_json::from_str(&msg).unwrap();
        assert_eq!(parsed["type"], "user");
        assert_eq!(parsed["message"]["role"], "user");
        assert_eq!(parsed["message"]["content"], "hello world");
    }

    // --- Codex adapter ---------------------------------------------------

    #[test]
    fn codex_args_use_exec_subcommand_and_json_output() {
        let args = build_codex_args(&InvokeOptions::default(), None);
        assert_eq!(args[0], "exec");
        assert!(args.contains(&"--json".to_string()));
        assert!(args.contains(&"--skip-git-repo-check".to_string()));
        assert!(args.contains(&"--ephemeral".to_string()));
        // Last arg is `-` so codex reads prompt from stdin.
        assert_eq!(args.last().map(String::as_str), Some("-"));
    }

    #[test]
    fn codex_args_inject_mcp_via_inline_config_override() {
        let args = build_codex_args(&InvokeOptions::default(), Some("http://127.0.0.1:9102/mcp"));
        let mcp_pos = args.iter().position(|a| a == "-c").unwrap();
        assert!(args[mcp_pos + 1].starts_with("mcp_servers.car="));
        assert!(args[mcp_pos + 1].contains("http://127.0.0.1:9102/mcp"));
        assert!(args[mcp_pos + 1].contains("type=\"http\""));
    }

    #[test]
    fn codex_args_threads_cd_flag() {
        let opts = InvokeOptions {
            cwd: Some("/tmp/work".into()),
            ..Default::default()
        };
        let args = build_codex_args(&opts, None);
        let pos = args.iter().position(|a| a == "--cd").unwrap();
        assert_eq!(args[pos + 1], "/tmp/work");
    }

    #[tokio::test]
    async fn codex_stream_aggregates_agent_message() {
        let lines = [
            r#"{"type":"thread.started","thread_id":"thread-abc"}"#,
            r#"{"type":"turn.started"}"#,
            r#"{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"hello world"}}"#,
            r#"{"type":"turn.completed","usage":{"input_tokens":5,"output_tokens":2}}"#,
        ];
        let result = process_codex_stream(mock_reader(&lines), None)
            .await
            .unwrap();
        assert_eq!(result.answer, "hello world");
        assert_eq!(result.session_id.as_deref(), Some("thread-abc"));
        assert_eq!(result.turns, 1);
        assert!(!result.is_error);
    }

    #[tokio::test]
    async fn codex_stream_concatenates_multiple_agent_messages() {
        let lines = [
            r#"{"type":"thread.started","thread_id":"t"}"#,
            r#"{"type":"turn.started"}"#,
            r#"{"type":"item.completed","item":{"type":"agent_message","text":"part 1 "}}"#,
            r#"{"type":"item.completed","item":{"type":"agent_message","text":"part 2"}}"#,
            r#"{"type":"turn.completed"}"#,
        ];
        let result = process_codex_stream(mock_reader(&lines), None)
            .await
            .unwrap();
        assert_eq!(result.answer, "part 1 part 2");
    }

    #[tokio::test]
    async fn codex_stream_records_tool_calls() {
        let lines = [
            r#"{"type":"thread.started","thread_id":"t"}"#,
            r#"{"type":"turn.started"}"#,
            r#"{"type":"item.completed","item":{"id":"call_1","type":"tool_call","name":"shell","arguments":{"cmd":"ls"}}}"#,
            r#"{"type":"item.completed","item":{"type":"agent_message","text":"done"}}"#,
            r#"{"type":"turn.completed"}"#,
        ];
        let result = process_codex_stream(mock_reader(&lines), None)
            .await
            .unwrap();
        assert_eq!(result.tool_calls, 1);
        assert_eq!(result.tool_uses.len(), 1);
        assert_eq!(result.tool_uses[0].name, "shell");
        assert_eq!(result.answer, "done");
    }

    #[tokio::test]
    async fn codex_stream_skips_non_json_banner_lines() {
        let lines = [
            "Reading prompt from stdin...",
            r#"{"type":"thread.started","thread_id":"t"}"#,
            r#"{"type":"turn.started"}"#,
            r#"{"type":"item.completed","item":{"type":"agent_message","text":"ok"}}"#,
            r#"{"type":"turn.completed"}"#,
        ];
        let result = process_codex_stream(mock_reader(&lines), None)
            .await
            .unwrap();
        assert_eq!(result.answer, "ok");
    }

    #[tokio::test]
    async fn codex_stream_no_agent_message_marks_error() {
        let lines = [r#"{"type":"thread.started","thread_id":"t"}"#];
        let result = process_codex_stream(mock_reader(&lines), None)
            .await
            .unwrap();
        assert!(result.is_error);
        assert!(result.answer.is_empty());
    }

    #[tokio::test]
    async fn codex_stream_fires_assistant_events_to_emitter() {
        // car#213 regression: pre-fix the emitter was dropped in the
        // dispatcher, so a streaming codex invocation produced zero
        // host-visible events (empty bubble). The emitter must now
        // fire an Anthropic-Messages-shaped Assistant text event the
        // host's chat renderer can turn into a `kind:"token"` frame.
        use std::sync::Mutex;
        let captured: Arc<Mutex<Vec<StreamEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let sink = captured.clone();
        let emitter: StreamEventEmitter = Arc::new(move |ev| sink.lock().unwrap().push(ev));
        let lines = [
            r#"{"type":"thread.started","thread_id":"t"}"#,
            r#"{"type":"turn.started"}"#,
            r#"{"type":"item.completed","item":{"type":"agent_message","text":"hi there"}}"#,
            r#"{"type":"turn.completed"}"#,
        ];
        let result = process_codex_stream(mock_reader(&lines), Some(emitter))
            .await
            .unwrap();
        assert_eq!(result.answer, "hi there");

        let events = captured.lock().unwrap();
        let texts: Vec<String> = events
            .iter()
            .filter_map(|e| match e {
                StreamEvent::Assistant(a) => a
                    .message
                    .get("content")
                    .and_then(|c| c.as_array())
                    .and_then(|arr| arr.first())
                    .filter(|b| b.get("type").and_then(|t| t.as_str()) == Some("text"))
                    .and_then(|b| b.get("text"))
                    .and_then(|t| t.as_str())
                    .map(str::to_string),
                _ => None,
            })
            .collect();
        assert!(
            texts.iter().any(|t| t == "hi there"),
            "codex emitter must fire an Assistant text event (car#213); got {events:?}"
        );
    }

    // --- Gemini adapter --------------------------------------------------

    #[test]
    fn gemini_args_use_prompt_and_yolo() {
        let args = build_gemini_args(&InvokeOptions::default(), "say hi", &[]);
        assert_eq!(args[0], "-p");
        assert_eq!(args[1], "say hi");
        assert!(args.contains(&"--yolo".to_string()));
    }

    // --- Image attachments -----------------------------------------------

    /// Minimal byte sequences carrying each format's magic signature,
    /// so they pass `sniff_image_media_type`.
    const PNG_BYTES: &[u8] = b"\x89PNG\r\n\x1a\n\x00\x00\x00rest";
    const JPEG_BYTES: &[u8] = b"\xFF\xD8\xFF\xE0rest";

    /// Write a throwaway file and return an ImageAttachment pointing at
    /// it.
    fn temp_attachment(dir: &std::path::Path, name: &str, bytes: &[u8]) -> ImageAttachment {
        let path = dir.join(name);
        std::fs::write(&path, bytes).unwrap();
        ImageAttachment { path, media_type: None }
    }

    #[test]
    fn sniff_recognizes_supported_formats() {
        assert_eq!(sniff_image_media_type(PNG_BYTES), Some("image/png"));
        assert_eq!(sniff_image_media_type(JPEG_BYTES), Some("image/jpeg"));
        assert_eq!(sniff_image_media_type(b"GIF89a..."), Some("image/gif"));
        assert_eq!(sniff_image_media_type(b"RIFF\0\0\0\0WEBPrest"), Some("image/webp"));
        // Non-images (incl. a secret a client might point us at) sniff None.
        assert_eq!(sniff_image_media_type(b"-----BEGIN OPENSSH PRIVATE KEY-----"), None);
        assert_eq!(sniff_image_media_type(b"x"), None);
        assert_eq!(sniff_image_media_type(b""), None);
    }

    #[test]
    fn user_message_with_image_builds_content_array() {
        let dir = tempfile::tempdir().unwrap();
        let att = temp_attachment(dir.path(), "shot.png", PNG_BYTES);
        let (msg, _accepted) = build_user_message("describe this", std::slice::from_ref(&att));
        let parsed: serde_json::Value = serde_json::from_str(&msg).unwrap();
        let content = parsed["message"]["content"].as_array().unwrap();
        assert_eq!(content.len(), 2);
        assert_eq!(content[0]["type"], "text");
        assert_eq!(content[0]["text"], "describe this");
        assert_eq!(content[1]["type"], "image");
        assert_eq!(content[1]["source"]["type"], "base64");
        // media_type comes from the sniffed content, not the extension.
        assert_eq!(content[1]["source"]["media_type"], "image/png");
        use base64::Engine as _;
        let data = content[1]["source"]["data"].as_str().unwrap();
        let decoded = base64::engine::general_purpose::STANDARD.decode(data).unwrap();
        assert_eq!(decoded, PNG_BYTES);
    }

    #[test]
    fn user_message_media_type_from_content_not_extension() {
        let dir = tempfile::tempdir().unwrap();
        // JPEG bytes behind a `.png` name — content wins.
        let att = temp_attachment(dir.path(), "mislabeled.png", JPEG_BYTES);
        let (msg, _accepted) = build_user_message("x", std::slice::from_ref(&att));
        let parsed: serde_json::Value = serde_json::from_str(&msg).unwrap();
        let content = parsed["message"]["content"].as_array().unwrap();
        assert_eq!(content[1]["source"]["media_type"], "image/jpeg");
    }

    #[test]
    fn user_message_image_only_omits_text_block() {
        let dir = tempfile::tempdir().unwrap();
        let att = temp_attachment(dir.path(), "shot.png", PNG_BYTES);
        let (msg, _accepted) = build_user_message("", std::slice::from_ref(&att));
        let parsed: serde_json::Value = serde_json::from_str(&msg).unwrap();
        let content = parsed["message"]["content"].as_array().unwrap();
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["type"], "image");
    }

    #[test]
    fn user_message_unreadable_image_is_skipped() {
        let att = ImageAttachment {
            path: std::path::PathBuf::from("/no/such/file.png"),
            media_type: None,
        };
        let (msg, _accepted) = build_user_message("hi", std::slice::from_ref(&att));
        let parsed: serde_json::Value = serde_json::from_str(&msg).unwrap();
        // Text block survives; the unreadable image drops out.
        let content = parsed["message"]["content"].as_array().unwrap();
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["type"], "text");
    }

    #[test]
    fn user_message_reports_accepted_image_count() {
        let dir = tempfile::tempdir().unwrap();
        let good1 = temp_attachment(dir.path(), "a.png", PNG_BYTES);
        let bad = temp_attachment(dir.path(), "b.png", b"not an image");
        let good2 = temp_attachment(dir.path(), "c.jpg", JPEG_BYTES);
        let (_msg, accepted) = build_user_message("hi", &[good1, bad, good2]);
        // Two valid images inlined; the non-image drops — the caller
        // computes dropped = 3 - 2 = 1.
        assert_eq!(accepted, 2);
    }

    #[test]
    fn user_message_non_image_file_is_not_inlined() {
        // A readable file that isn't an image (e.g. a secret a client
        // pointed `path` at) must never be base64-inlined.
        let dir = tempfile::tempdir().unwrap();
        let att = temp_attachment(dir.path(), "secret.png", b"-----BEGIN PRIVATE KEY-----");
        let (msg, _accepted) = build_user_message("hi", std::slice::from_ref(&att));
        let parsed: serde_json::Value = serde_json::from_str(&msg).unwrap();
        let content = parsed["message"]["content"].as_array().unwrap();
        assert_eq!(content.len(), 1);
        assert_eq!(content[0]["type"], "text");
    }

    #[test]
    fn codex_args_include_image_flag_before_stdin_marker() {
        let dir = tempfile::tempdir().unwrap();
        let att = temp_attachment(dir.path(), "a.png", b"x");
        let opts = InvokeOptions { attachments: vec![att.clone()], ..Default::default() };
        let args = build_codex_args(&opts, None);
        let image_idx = args.iter().position(|a| a == "--image").unwrap();
        assert_eq!(args[image_idx + 1], att.path.to_string_lossy());
        // The stdin `-` marker stays last, after the image flag.
        assert_eq!(args.last().unwrap(), "-");
        assert!(image_idx < args.len() - 1);
    }

    #[test]
    fn codex_args_skip_missing_image() {
        let opts = InvokeOptions {
            attachments: vec![ImageAttachment {
                path: std::path::PathBuf::from("/no/such/img.png"),
                media_type: None,
            }],
            ..Default::default()
        };
        let args = build_codex_args(&opts, None);
        assert!(!args.iter().any(|a| a == "--image"));
    }

    #[test]
    fn gemini_args_append_image_refs() {
        let args = build_gemini_args(
            &InvokeOptions::default(),
            "look",
            &["/tmp/stage/img0.png".to_string()],
        );
        assert_eq!(args[1], "look @/tmp/stage/img0.png");
    }

    #[test]
    fn stage_gemini_images_copies_to_safe_paths() {
        let src = tempfile::tempdir().unwrap();
        let att = temp_attachment(src.path(), "orig.png", PNG_BYTES);
        let (stage, refs) = stage_gemini_images(std::slice::from_ref(&att));
        assert!(stage.is_some());
        assert_eq!(refs.len(), 1);
        // Staged name is index-based, typed by sniffed content, and
        // space-free; the file exists with the original bytes.
        assert!(refs[0].ends_with("img0.png"));
        assert!(!refs[0].contains(' '));
        assert_eq!(std::fs::read(&refs[0]).unwrap(), PNG_BYTES);
    }

    #[test]
    fn stage_gemini_images_skips_non_image() {
        let src = tempfile::tempdir().unwrap();
        let att = temp_attachment(src.path(), "notes.png", b"just text");
        let (_stage, refs) = stage_gemini_images(std::slice::from_ref(&att));
        assert!(refs.is_empty());
    }

    #[test]
    fn invoke_options_deserializes_attachments() {
        let json = r#"{"attachments":[{"path":"/a/b.png","media_type":"image/png"}]}"#;
        let opts: InvokeOptions = serde_json::from_str(json).unwrap();
        assert_eq!(opts.attachments.len(), 1);
        assert_eq!(opts.attachments[0].path, std::path::PathBuf::from("/a/b.png"));
        assert_eq!(opts.attachments[0].media_type.as_deref(), Some("image/png"));
        // And the field is optional / defaults empty.
        let bare: InvokeOptions = serde_json::from_str("{}").unwrap();
        assert!(bare.attachments.is_empty());
    }
}