embacle 0.22.5

// ABOUTME: CopilotHeadlessRunner wraps the copilot CLI via ACP (Agent Client Protocol) for LLM completions.
// ABOUTME: Keeps one copilot --acp subprocess alive across complete() calls so the GitHub→Copilot OAuth token exchange amortizes; per-call session/new is cheap on the warm transport.
//
// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 dravr.ai

use std::collections::VecDeque;
use std::env;
use std::fs;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::process::Stdio;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::LazyLock;
use std::sync::{Arc, Mutex as StdMutex, OnceLock, PoisonError};
use std::time::{Duration, Instant};

use tokio_stream::Stream;

use agent_client_protocol_schema as schema;
use async_trait::async_trait;
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufWriter};
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
use tokio::sync::{mpsc, Mutex as TokioMutex, OwnedMutexGuard, Semaphore};
use tokio::time;
use tokio_stream::wrappers::UnboundedReceiverStream;
use tracing::{debug, field, info, instrument, trace, warn, Span};

use crate::copilot_headless_config::{CopilotHeadlessConfig, PermissionPolicy};
use crate::copilot_models::catalog_ids;
use crate::types::{
    ChatMessage, ChatRequest, ChatResponse, ChatStream, LlmCapabilities, LlmProvider, McpHeader,
    McpServerConfig, McpTransport, MessageRole, RunnerError, StreamChunk, TokenUsage,
};

/// Default prompt timeout (5 minutes). Override with `EMBACLE_ACP_PROMPT_TIMEOUT_SECS`.
const DEFAULT_ACP_PROMPT_TIMEOUT_SECS: u64 = 300;

/// Read prompt timeout from env, falling back to [`DEFAULT_ACP_PROMPT_TIMEOUT_SECS`].
/// Default number of warm `copilot --acp` subprocesses kept in the pool.
///
/// One slot serves one completion at a time, because the ACP transport is
/// request/response by JSON-RPC id and cannot route two concurrent prompts.
/// The pool buys concurrency by holding N independent transports, never by
/// interleaving on one.
///
/// Two rather than one: the deployed backend accepts 80 concurrent requests
/// per instance (`backend_max_instance_request_concurrency`) inside a 2Gi
/// container, and every slot is a full `copilot` CLI subprocess competing for
/// that budget. Two lifts the hard serialization of a single slot while
/// leaving the memory envelope intact; raise it with the env var once the
/// per-slot resident cost is measured against real traffic.
const DEFAULT_ACP_POOL_SIZE: usize = 2;

/// Upper bound on `EMBACLE_ACP_POOL_SIZE`, so a typo cannot fork the container
/// into an unbounded number of CLI subprocesses.
///
/// Four, because [`ACP_PROCESS_BASE_MB`] alone is the binding constraint:
/// eight idle subprocesses would hold 1,704 MB before serving one session,
/// well past [`ACP_POOL_BUDGET_MB`]. The derived session ceiling keeps memory
/// safe above this too, but it does so by recycling almost every completion —
/// past about three slots the respawn cost eats the concurrency it bought.
const MAX_ACP_POOL_SIZE: usize = 4;

/// Resident memory a `copilot --acp` subprocess holds once `initialize`
/// completes, before it has served a single session.
///
/// Measured against CLI 1.0.81 on 2026-08-28: 213 MB.
const ACP_PROCESS_BASE_MB: usize = 213;

/// Resident memory one ACP session adds, and never gives back.
///
/// Measured against CLI 1.0.81 on 2026-08-28: ~26 MB per `session/new`, and
/// **independent of system-prompt size** — 0 B, 16 B and 27 KB prompts all cost
/// the same, so this is fixed per-session overhead inside the CLI rather than
/// anything we send.
///
/// It cannot be given back through the protocol. The agent advertises
/// `sessionCapabilities.close`, and `session/close` returns a result rather
/// than an error, but 20 new+close pairs grew resident memory by 515.8 MB
/// against 522.5 MB for 20 sessions never closed — the call is accepted and
/// frees nothing. Recycling the subprocess is the only remedy.
const ACP_SESSION_COST_MB: usize = 26;

/// Total resident memory every pooled subprocess may occupy between them.
///
/// Half of the deployed backend's 2Gi container, leaving the other half to the
/// server itself. The pool is sized and recycled against this, not against a
/// number that felt right.
const ACP_POOL_BUDGET_MB: usize = 1024;

/// Serializes `settings.json` write → spawn → `initialize` for every ACP child.
///
/// `copilot --acp` selects the served model from the `model` field of
/// `$HOME/.copilot/settings.json` and nothing else — measured 2026-08-30 by
/// asking the model to name itself under conflicting configuration:
///
/// | settings.json    | `--model`        | model answered   |
/// |------------------|------------------|------------------|
/// | `claude-sonnet-5`| `gpt-5.6-sol`    | "Claude Sonnet 5"|
/// | `gpt-5.6-sol`    | `claude-sonnet-5`| "GPT-5.6 Sol"    |
///
/// `--model` moves only the cosmetic `currentModelId` label. So the routing
/// input is a PROCESS-GLOBAL file, and the child reads it after exec — two
/// concurrent spawns for different models could interleave (A writes X, B
/// writes Y, A's child reads Y) and A's turn would silently run on the wrong
/// model. Nothing downstream can detect that: a `session/prompt` result carries
/// only `stopReason` and `usage`, with no model field, and the response model
/// is synthesized from what the caller asked for.
///
/// Holding this across the handshake — not merely across the write — is what
/// closes it. Once `initialize` has returned, the child has read its
/// configuration, so a later writer cannot change the model under it.
///
/// Spawns are rare (only on discard: exit, model change, session ceiling), so
/// serializing them costs little; a turn that reuses a warm subprocess never
/// touches this.
static MODEL_ROUTING_GATE: LazyLock<TokioMutex<()>> = LazyLock::new(|| TokioMutex::new(()));

/// Resident memory the STREAMING paths may occupy between them.
///
/// Separate from [`ACP_POOL_BUDGET_MB`] because a streamed turn does not use
/// the pool: its subprocess outlives the call inside a background task, so
/// borrowing a warm slot would hold it for the whole turn — up to the prompt
/// timeout — and make `checkout().await` block the first token, which is the
/// latency streaming exists to protect. A dedicated bound caps the memory
/// without that trade.
///
/// The two budgets are spent from the same container, so raising either eats
/// headroom the server itself needs. Deployed dev peaked at 45% of 2Gi
/// (~920 MB) on 2026-08-30, subprocesses included.
const ACP_STREAM_BUDGET_MB: usize = 512;

/// How many streamed turns may run at once.
///
/// Each holds a dedicated subprocess for the turn: [`ACP_PROCESS_BASE_MB`] to
/// exist plus one session at [`ACP_SESSION_COST_MB`], since a stream opens
/// exactly one and the child dies with the turn. Derived rather than chosen so
/// the bound moves with the measurements it rests on, and floored at 1 so a
/// budget smaller than one subprocess still serves turns one at a time rather
/// than deadlocking.
fn max_concurrent_streams() -> usize {
    (ACP_STREAM_BUDGET_MB / (ACP_PROCESS_BASE_MB + ACP_SESSION_COST_MB)).max(1)
}

/// Sessions a warm subprocess may serve before it is recycled.
///
/// Derived, not chosen: each slot gets an equal share of
/// [`ACP_POOL_BUDGET_MB`], spends [`ACP_PROCESS_BASE_MB`] of it on existing,
/// and buys sessions with the rest at [`ACP_SESSION_COST_MB`] each. Deriving it
/// is what makes `EMBACLE_ACP_POOL_SIZE` safe to raise: a fixed ceiling that
/// suited one slot would OOM the container at eight.
///
/// Floored at 1 so a pool sized past its own budget still makes progress,
/// recycling after every completion rather than deadlocking.
fn max_sessions_per_process(pool_size: usize) -> u32 {
    let per_slot = ACP_POOL_BUDGET_MB / pool_size.max(1);
    let headroom = per_slot.saturating_sub(ACP_PROCESS_BASE_MB);
    u32::try_from(headroom / ACP_SESSION_COST_MB)
        .unwrap_or(u32::MAX)
        .max(1)
}

/// Number of warm subprocesses to pool, from `EMBACLE_ACP_POOL_SIZE`.
///
/// Clamped to `1..=MAX_ACP_POOL_SIZE`; an unparseable or zero value falls back
/// to the default rather than disabling the pool.
fn acp_pool_size() -> usize {
    env::var("EMBACLE_ACP_POOL_SIZE")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|n| *n > 0)
        .unwrap_or(DEFAULT_ACP_POOL_SIZE)
        .min(MAX_ACP_POOL_SIZE)
}

fn acp_prompt_timeout() -> Duration {
    let secs = env::var("EMBACLE_ACP_PROMPT_TIMEOUT_SECS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(DEFAULT_ACP_PROMPT_TIMEOUT_SECS);
    Duration::from_secs(secs)
}

/// Build the JSON params for an ACP `session/prompt` request.
///
/// Always includes `sessionId` and `prompt` blocks. When `max_tokens` is
/// specified, forwards it as `maxTokens` so the ACP provider can respect
/// the caller's output length limit.
fn build_prompt_params(session_id: &str, prompt: &[Value], max_tokens: Option<u32>) -> Value {
    let mut params = json!({
        "sessionId": session_id,
        "prompt": prompt,
    });
    if let Some(mt) = max_tokens {
        params["maxTokens"] = Value::from(mt);
    }
    params
}

/// Pull the enabled model ids out of a `session/new` result.
///
/// ACP returns `models.availableModels[]` with `{modelId, name, description}`
/// and a `_meta` carrying `copilotEnablement` / `copilotUsage` /
/// `copilotPriceCategory`. Only ids the account may actually use are kept: a
/// list that includes disabled models is no better than the hardcoded one for
/// deciding whether a configured model will work.
///
/// `auto` is dropped — it is a selection strategy, not a model, and a caller
/// checking "is my configured model available" gets a false positive from it.
///
/// Returns `None` when the field is absent (older CLI) or holds nothing usable,
/// so the caller keeps the catalog rather than replacing it with emptiness.
fn models_from_session(result: &Value) -> Option<Vec<String>> {
    let listed = result.get("models")?.get("availableModels")?.as_array()?;
    let ids: Vec<String> = listed
        .iter()
        .filter(|m| {
            m.get("_meta")
                .and_then(|meta| meta.get("copilotEnablement"))
                .and_then(Value::as_str)
                // Absent enablement means the CLI did not say; treat as usable
                // rather than silently dropping a model that works.
                .is_none_or(|state| state == "enabled")
        })
        .filter_map(|m| m.get("modelId").and_then(Value::as_str))
        .filter(|id| *id != "auto")
        .map(ToOwned::to_owned)
        .collect();
    (!ids.is_empty()).then_some(ids)
}

/// Extra attempts when a turn comes back with no content at all.
///
/// One. A second empty turn is an answer about this prompt rather than a flake,
/// and every attempt is a full inference against a metered provider.
const DEGENERATE_TURN_RETRIES: u32 = 1;

/// Serialize embacle MCP server configs into the ACP `session/new`
/// `mcpServers` wire format.
///
/// Matches the Agent Client Protocol `McpServer` schema: HTTP/SSE are
/// `type`-tagged and carry `{name,url,headers:[{name,value}]}`; stdio is
/// untagged and carries `{name,command,args,env:[{name,value}]}`. The
/// `headers_to_json` shape pins the `{name,value}` pairs the schema's
/// `HttpHeader`/`EnvVariable` expect.
fn mcp_servers_to_acp_json(servers: &[McpServerConfig]) -> Vec<Value> {
    fn headers_to_json(headers: &[McpHeader]) -> Vec<Value> {
        headers
            .iter()
            .map(|h| json!({ "name": h.name, "value": h.value }))
            .collect()
    }

    servers
        .iter()
        .map(|server| match &server.transport {
            McpTransport::Http { url, headers } => json!({
                "type": "http",
                "name": server.name,
                "url": url,
                "headers": headers_to_json(headers),
            }),
            McpTransport::Sse { url, headers } => json!({
                "type": "sse",
                "name": server.name,
                "url": url,
                "headers": headers_to_json(headers),
            }),
            McpTransport::Stdio { command, args, env } => json!({
                "name": server.name,
                "command": command,
                "args": args,
                "env": headers_to_json(env),
            }),
        })
        .collect()
}

// ---------------------------------------------------------------------------
// NDJSON transport
// ---------------------------------------------------------------------------

/// Async NDJSON transport for ACP JSON-RPC communication.
///
/// Handles reading/writing newline-delimited JSON messages over stdio pipes.
/// Each message is a single JSON line terminated by `\n`.
struct AcpTransport {
    writer: BufWriter<ChildStdin>,
    reader: BufReader<ChildStdout>,
    next_id: i64,
}

impl AcpTransport {
    fn new(stdin: ChildStdin, stdout: ChildStdout) -> Self {
        Self {
            writer: BufWriter::new(stdin),
            reader: BufReader::new(stdout),
            next_id: 1,
        }
    }

    /// Send a JSON-RPC request and return its id.
    async fn send_request(&mut self, method: &str, params: Value) -> Result<i64, RunnerError> {
        let id = self.next_id;
        self.next_id += 1;

        let msg = json!({
            "jsonrpc": "2.0",
            "id": id,
            "method": method,
            "params": params,
        });

        self.write_message(&msg).await?;
        Ok(id)
    }

    /// Send a JSON-RPC response (for server-to-client requests like permission).
    async fn send_response(&mut self, id: &Value, result: Value) -> Result<(), RunnerError> {
        let msg = json!({
            "jsonrpc": "2.0",
            "id": id,
            "result": result,
        });
        self.write_message(&msg).await
    }

    /// Write a single NDJSON message.
    async fn write_message(&mut self, msg: &Value) -> Result<(), RunnerError> {
        let line = serde_json::to_string(msg)
            .map_err(|e| RunnerError::internal(format!("JSON serialization failed: {e}")))?;
        self.writer
            .write_all(line.as_bytes())
            .await
            .map_err(|e| RunnerError::internal(format!("Write failed: {e}")))?;
        self.writer
            .write_all(b"\n")
            .await
            .map_err(|e| RunnerError::internal(format!("Write newline failed: {e}")))?;
        self.writer
            .flush()
            .await
            .map_err(|e| RunnerError::internal(format!("Flush failed: {e}")))?;
        Ok(())
    }

    /// Default per-message read timeout (90 seconds).
    ///
    /// If the copilot process is alive but not sending any messages for this
    /// duration, the read is considered failed. This catches hung processes
    /// that don't produce output but haven't exited.
    /// Override with `EMBACLE_ACP_MESSAGE_TIMEOUT_SECS`.
    fn message_timeout() -> Duration {
        let secs = env::var("EMBACLE_ACP_MESSAGE_TIMEOUT_SECS")
            .ok()
            .and_then(|v| v.parse::<u64>().ok())
            .unwrap_or(90);
        Duration::from_secs(secs)
    }

    /// Read the next NDJSON message, skipping blank lines.
    ///
    /// Times out if no message arrives within [`Self::message_timeout`],
    /// detecting hung copilot processes that are alive but not responding.
    async fn read_message(&mut self) -> Result<Value, RunnerError> {
        let mut line = String::new();
        loop {
            line.clear();
            let read_result =
                time::timeout(Self::message_timeout(), self.reader.read_line(&mut line)).await;

            let n = match read_result {
                Ok(Ok(n)) => n,
                Ok(Err(e)) => {
                    return Err(RunnerError::internal(format!("Read failed: {e}")));
                }
                Err(_) => {
                    return Err(RunnerError::internal(format!(
                        "ACP message read timed out after {}s — copilot process may be hung",
                        Self::message_timeout().as_secs()
                    )));
                }
            };

            if n == 0 {
                return Err(RunnerError::internal("ACP connection closed unexpectedly"));
            }
            let trimmed = line.trim();
            if trimmed.is_empty() {
                continue;
            }
            return serde_json::from_str(trimmed)
                .map_err(|e| RunnerError::internal(format!("JSON parse failed: {e}")));
        }
    }

    /// Read messages until we get the response matching the given request id.
    ///
    /// Non-matching messages (notifications, other responses) are skipped.
    async fn read_response(&mut self, expected_id: i64) -> Result<Value, RunnerError> {
        loop {
            let msg = self.read_message().await?;
            if msg.get("id").and_then(Value::as_i64) == Some(expected_id) {
                if let Some(error) = msg.get("error") {
                    return Err(RunnerError::external_service(
                        "copilot-acp",
                        format!("RPC error: {error}"),
                    ));
                }
                return Ok(msg.get("result").cloned().unwrap_or(Value::Null));
            }
        }
    }
}

// ---------------------------------------------------------------------------
// ACP session lifecycle
// ---------------------------------------------------------------------------

/// Ensure the GitHub Copilot CLI config selects `model` for the next
/// `copilot --acp` session, returning the settings path on success.
///
/// `copilot --acp` routes by the `"model"` field in `$HOME/.copilot/settings.json`
/// ONLY — it ignores both the ACP `session/new` `model` field and the `--model`
/// flag (those set a cosmetic `currentModelId` label, not the served model). With
/// no settings.json the CLI falls through to the account's default / experiment
/// routing (e.g. GPT auto-selection), so a `claude-sonnet-4.6` request silently
/// runs GPT/Gemini. Read-modify-write that file so the session actually runs
/// `model`, preserving any unrelated keys (theme, effortLevel, ...) already there.
///
/// Errors are the caller's to downgrade to a warning: a config write failure must
/// never abort a turn — it just leaves the previous/default model in effect.
///
/// LIMITATION(registre#134): settings.json is process-global and this write is
/// unsynchronized, while the child reads it asynchronously AFTER exec. Two
/// spawns for different models can interleave — A writes X, B writes Y, A's
/// child reads Y — and A's turn then runs on the wrong model with nothing
/// downstream able to notice, since the response reports the model we asked
/// for rather than the one that served it. Serializing write-and-spawn narrows
/// the window without closing it; the real fix is a per-child config location,
/// which is entangled with credential storage under the same directory. The
/// pool reduces the exposure — spawns now happen only on discard rather than
/// once per call — but does not remove it.
fn ensure_copilot_settings_model(model: &str) -> Result<PathBuf, RunnerError> {
    let home = env::var_os("HOME").ok_or_else(|| {
        RunnerError::internal("HOME is not set; cannot locate copilot settings.json")
    })?;
    write_settings_model(&PathBuf::from(home).join(".copilot"), model)
}

/// Read-modify-write `<copilot_dir>/settings.json` so its `"model"` key is
/// `model`, preserving any other keys. Split from [`ensure_copilot_settings_model`]
/// so it is testable against a temp dir without mutating the process `HOME`.
fn write_settings_model(copilot_dir: &Path, model: &str) -> Result<PathBuf, RunnerError> {
    fs::create_dir_all(copilot_dir)
        .map_err(|e| RunnerError::internal(format!("create {}: {e}", copilot_dir.display())))?;
    let path = copilot_dir.join("settings.json");

    // Preserve existing keys; a missing or unparseable file starts empty.
    let mut settings = fs::read_to_string(&path)
        .ok()
        .and_then(|s| serde_json::from_str::<Value>(&s).ok())
        .filter(Value::is_object)
        .unwrap_or_else(|| json!({}));
    settings["model"] = Value::String(model.to_owned());

    let body = serde_json::to_string_pretty(&settings)
        .map_err(|e| RunnerError::internal(format!("serialize copilot settings: {e}")))?;
    fs::write(&path, body)
        .map_err(|e| RunnerError::internal(format!("write {}: {e}", path.display())))?;
    Ok(path)
}

/// Spawn the copilot --acp subprocess with piped stdio, pinning `model`.
///
/// Pinning happens two ways, because the `--acp` server resolves its model
/// differently from the rest of the CLI:
/// 1. [`ensure_copilot_settings_model`] writes `model` into
///    `~/.copilot/settings.json` — the field `copilot --acp` ACTUALLY routes by.
/// 2. The `--model` flag is also passed; for `--acp` it only sets the cosmetic
///    `currentModelId` label, but it keeps that label consistent with the route
///    and is the real selector for the non-ACP (`-p`) paths.
///
/// The model is fixed for the lifetime of the spawned subprocess — callers
/// reusing a warm process must respawn to change it.
fn spawn_copilot(
    cli_path: &PathBuf,
    github_token: Option<&str>,
    model: &str,
) -> Result<Child, RunnerError> {
    // The route-determining step. Non-fatal: on failure copilot keeps whatever
    // model its settings.json already selects rather than aborting the spawn.
    match ensure_copilot_settings_model(model) {
        Ok(path) => info!(
            model = %model,
            settings = %path.display(),
            "ACP: pinned routing model in copilot settings.json"
        ),
        Err(e) => warn!(
            model = %model,
            error = %e,
            "ACP: could not pin routing model in settings.json; copilot may use its default model"
        ),
    }

    let mut cmd = Command::new(cli_path);
    cmd.arg("--acp");
    if !model.is_empty() {
        cmd.arg("--model").arg(model);
    }
    cmd.stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());

    if let Some(token) = github_token {
        cmd.env("COPILOT_GITHUB_TOKEN", token);
    }

    info!(cli_path = %cli_path.display(), model = %model, "Spawning copilot --acp subprocess");

    let child = cmd
        .spawn()
        .map_err(|e| RunnerError::internal(format!("Failed to spawn copilot --acp: {e}")))?;

    info!(
        pid = child.id().unwrap_or(0),
        "copilot --acp subprocess started"
    );
    Ok(child)
}

/// Default session setup timeout (60 seconds).
///
/// Copilot CLI may need 20–25s for first-run package extraction in containers,
/// plus time for auth handshake. Override with `EMBACLE_ACP_SESSION_TIMEOUT_SECS`.
const DEFAULT_ACP_SESSION_TIMEOUT_SECS: u64 = 60;

/// Read session setup timeout from `EMBACLE_ACP_SESSION_TIMEOUT_SECS` env var,
/// falling back to [`DEFAULT_ACP_SESSION_TIMEOUT_SECS`].
fn acp_session_timeout() -> Duration {
    let secs = env::var("EMBACLE_ACP_SESSION_TIMEOUT_SECS")
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .unwrap_or(DEFAULT_ACP_SESSION_TIMEOUT_SECS);
    Duration::from_secs(secs)
}

/// ACP `mode` config value that runs the agent loop to completion.
const AUTOPILOT_MODE_VALUE: &str =
    "https://agentclientprotocol.com/protocol/session-modes#autopilot";

/// Switch a session to Autopilot mode so Copilot runs the tool loop to
/// completion (call tools → consume results → synthesize) within a single
/// prompt turn.
///
/// In the default Agent mode, Copilot ends the turn right after the tool
/// batch — the model never consumes the results, so callers get a bare
/// "I'll pull X…" preface. Autopilot ("runs until task completion without
/// user interaction") makes Copilot feed results back to the model and emit a
/// real answer. Only meaningful when the session has MCP tools to call.
async fn set_autopilot_mode(
    transport: &mut AcpTransport,
    session_id: &str,
) -> Result<(), RunnerError> {
    let req_id = transport
        .send_request(
            "session/set_config_option",
            json!({
                "sessionId": session_id,
                "configId": "mode",
                "value": AUTOPILOT_MODE_VALUE,
            }),
        )
        .await?;
    let _ = transport.read_response(req_id).await?;
    Ok(())
}

/// Initialize ACP connection and create a session.
///
/// Returns the transport and session id ready for prompting.
/// Times out after [`acp_session_timeout`] to detect hung copilot processes early.
async fn setup_session(
    cli_path: &PathBuf,
    github_token: Option<&str>,
    model: &str,
    system_prompt: Option<&str>,
    mcp_servers: &[McpServerConfig],
) -> Result<(AcpTransport, Child, StderrRing, String, Option<Vec<String>>), RunnerError> {
    let mut child = spawn_copilot(cli_path, github_token, model)?;
    let stderr_ring = StderrRing::drain(child.stderr.take());

    let stdin = child
        .stdin
        .take()
        .ok_or_else(|| RunnerError::internal("Failed to capture copilot stdin"))?;
    let stdout = child
        .stdout
        .take()
        .ok_or_else(|| RunnerError::internal("Failed to capture copilot stdout"))?;

    let mut transport = AcpTransport::new(stdin, stdout);

    // Wrap handshake + session creation in a timeout to detect hung processes early
    let session_result = time::timeout(acp_session_timeout(), async {
        // Initialize handshake
        info!("ACP: sending initialize handshake");
        let init_id = transport
            .send_request(
                "initialize",
                json!({
                    "protocolVersion": 1,
                    "clientInfo": {
                        "name": "embacle",
                        "version": env!("CARGO_PKG_VERSION"),
                    },
                    "capabilities": {},
                }),
            )
            .await?;
        let init_resp = transport.read_response(init_id).await?;
        info!("ACP: initialize handshake complete");
        debug!(response = %init_resp, "ACP initialize response");

        // Create session with model, optional system prompt, and any MCP
        // servers the model should call tools from.
        let mut session_params = json!({
            "model": model,
            "cwd": env::current_dir()
                .map_err(|e| RunnerError::internal(format!("Failed to get cwd: {e}")))?,
            "mcpServers": mcp_servers_to_acp_json(mcp_servers),
        });
        if let Some(sys) = system_prompt {
            session_params["systemPrompt"] = Value::String(sys.to_owned());
        }

        info!(
            model = %model,
            has_system_prompt = system_prompt.is_some(),
            mcp_servers = mcp_servers.len(),
            "ACP: creating session"
        );
        let session_id_req = transport
            .send_request("session/new", session_params)
            .await?;
        let session_result = transport.read_response(session_id_req).await?;

        let session_id = session_result
            .get("sessionId")
            .and_then(Value::as_str)
            .ok_or_else(|| {
                RunnerError::external_service("copilot-acp", "Missing sessionId in response")
            })?
            .to_owned();

        // The CLI tells us which models this account may use. Read it here
        // rather than trusting the compiled-in catalog, which cannot know.
        let observed = models_from_session(&session_result);

        info!(
            session_id = %session_id,
            model = %model,
            reported_models = observed.as_ref().map_or(0, Vec::len),
            "ACP session created"
        );

        // With MCP tools available, run the loop to completion (see
        // `set_autopilot_mode`). Best-effort: on failure we log and continue
        // in the default Agent mode rather than aborting the turn.
        if !mcp_servers.is_empty() {
            match set_autopilot_mode(&mut transport, &session_id).await {
                Ok(()) => info!(session_id = %session_id, "ACP: session mode set to Autopilot"),
                Err(e) => {
                    warn!(session_id = %session_id, error = %e, "ACP: failed to set Autopilot mode");
                }
            }
        }

        Ok::<_, RunnerError>((session_id, observed))
    })
    .await;

    match session_result {
        Ok(Ok((session_id, observed))) => Ok((transport, child, stderr_ring, session_id, observed)),
        Ok(Err(e)) => {
            warn!(
                error = %e,
                stderr = %stderr_ring.snapshot(),
                "ACP session setup failed"
            );
            let _ = child.kill().await;
            Err(e)
        }
        Err(_elapsed) => {
            warn!(
                stderr = %stderr_ring.snapshot(),
                timeout_secs = acp_session_timeout().as_secs(),
                "ACP session setup timed out — copilot process may be hung (auth issue?)"
            );
            let _ = child.kill().await;
            Err(RunnerError::timeout(format!(
                "copilot-acp: session setup timed out after {}s (check copilot auth)",
                acp_session_timeout().as_secs()
            )))
        }
    }
}

/// A live `copilot --acp` subprocess that has completed the ACP `initialize`
/// handshake and is ready to accept `session/new` calls without re-running
/// the GitHub→Copilot OAuth token exchange.
///
/// Held inside [`CopilotHeadlessRunner::process`] so successful chat calls
/// amortize the auth handshake across the subprocess lifetime. The transport
/// is request/response by JSON-RPC id and cannot interleave concurrent
/// prompts, so the parent always wraps this in a `tokio::sync::Mutex`.
struct AcpProcess {
    child: Child,
    transport: AcpTransport,
    /// Continuously drained stderr tail, sampled on failure paths. The drain
    /// also keeps the subprocess from ever blocking on a full stderr pipe.
    stderr_ring: StderrRing,
    /// The model this subprocess was spawned with via `copilot --acp --model`.
    /// Fixed for the subprocess lifetime — a request for a different model
    /// requires respawning (see [`CopilotHeadlessRunner::complete`]).
    model: String,
    /// How many `session/new` calls this subprocess has served.
    ///
    /// Nothing ever closes an ACP session: the protocol call is never sent
    /// (`session/close` appears nowhere in this crate) and the CLI holds each
    /// session for the subprocess lifetime. A warm subprocess therefore
    /// accumulates one dead session per completion, forever. Recycling on this
    /// count bounds that without depending on a protocol call the CLI may not
    /// implement.
    sessions_served: u32,
}

impl AcpProcess {
    /// Spawn the subprocess and complete the ACP `initialize` handshake.
    /// On success the returned process is ready for [`Self::new_session`].
    ///
    /// On any handshake failure the subprocess is killed before the error
    /// is returned so callers never leak a half-initialized child.
    async fn spawn_and_initialize(
        cli_path: &PathBuf,
        github_token: Option<&str>,
        model: &str,
    ) -> Result<Self, RunnerError> {
        // Held until the handshake completes: see MODEL_ROUTING_GATE.
        let routing_gate = MODEL_ROUTING_GATE.lock().await;
        let mut child = spawn_copilot(cli_path, github_token, model)?;
        let stderr_ring = StderrRing::drain(child.stderr.take());
        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| RunnerError::internal("Failed to capture copilot stdin"))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| RunnerError::internal("Failed to capture copilot stdout"))?;
        let mut transport = AcpTransport::new(stdin, stdout);

        let init_outcome = time::timeout(acp_session_timeout(), async {
            info!("ACP: sending initialize handshake");
            let init_id = transport
                .send_request(
                    "initialize",
                    json!({
                        "protocolVersion": 1,
                        "clientInfo": {
                            "name": "embacle",
                            "version": env!("CARGO_PKG_VERSION"),
                        },
                        "capabilities": {},
                    }),
                )
                .await?;
            let init_resp = transport.read_response(init_id).await?;
            info!("ACP: initialize handshake complete");
            debug!(response = %init_resp, "ACP initialize response");
            Ok::<_, RunnerError>(())
        })
        .await;

        // The child has read its configuration by now, so a later writer can no
        // longer change the model under it. Released here rather than at scope
        // exit so the point it stops mattering is stated, not inferred.
        drop(routing_gate);

        match init_outcome {
            Ok(Ok(())) => Ok(Self {
                child,
                transport,
                stderr_ring,
                model: model.to_owned(),
                sessions_served: 0,
            }),
            Ok(Err(e)) => {
                warn!(error = %e, stderr = %stderr_ring.snapshot(), "ACP initialize failed");
                let _ = child.kill().await;
                Err(e)
            }
            Err(_elapsed) => {
                warn!(
                    stderr = %stderr_ring.snapshot(),
                    timeout_secs = acp_session_timeout().as_secs(),
                    "ACP initialize timed out — copilot process may be hung (auth issue?)"
                );
                let _ = child.kill().await;
                Err(RunnerError::timeout(format!(
                    "copilot-acp: initialize timed out after {}s (check copilot auth)",
                    acp_session_timeout().as_secs()
                )))
            }
        }
    }

    /// Create a fresh ACP session on the already-initialized subprocess.
    ///
    /// Cheap relative to [`Self::spawn_and_initialize`]: no new subprocess,
    /// no new GitHub→Copilot token exchange — copilot reuses the in-process
    /// token cache it built during `initialize`.
    /// Open a session on this subprocess.
    ///
    /// `mcpServers` is declared per session, and the agent honours that: a warm
    /// subprocess reconnects to the declared servers on every `session/new`
    /// rather than caching a connection by name and url. Measured against CLI
    /// 1.0.81 on 2026-08-30 with a stub MCP server recording Authorization
    /// headers — two sessions on ONE subprocess, same server name and url,
    /// different bearers: session 2 issued three fresh requests carrying the
    /// SECOND bearer.
    ///
    /// That is what makes pooling safe for `converse()`, which carries a real
    /// per-turn tool surface whose credential rotates. Had the agent reused the
    /// first session's connection, a reused subprocess would have called tools
    /// with a revoked token, or worse, against the previous turn's
    /// tenant-scoped surface. Re-measure before assuming it still holds.
    async fn new_session(
        &mut self,
        model: &str,
        system_prompt: Option<&str>,
        mcp_servers: &[McpServerConfig],
    ) -> Result<(String, Option<Vec<String>>), RunnerError> {
        let outcome = time::timeout(acp_session_timeout(), async {
            let mut session_params = json!({
                "model": model,
                "cwd": env::current_dir()
                    .map_err(|e| RunnerError::internal(format!("Failed to get cwd: {e}")))?,
                "mcpServers": mcp_servers_to_acp_json(mcp_servers),
            });
            if let Some(sys) = system_prompt {
                session_params["systemPrompt"] = Value::String(sys.to_owned());
            }
            info!(
                model = %model,
                has_system_prompt = system_prompt.is_some(),
                mcp_servers = mcp_servers.len(),
                "ACP: creating session"
            );
            let req_id = self
                .transport
                .send_request("session/new", session_params)
                .await?;
            let resp = self.transport.read_response(req_id).await?;
            let session_id = resp
                .get("sessionId")
                .and_then(Value::as_str)
                .ok_or_else(|| {
                    RunnerError::external_service("copilot-acp", "Missing sessionId in response")
                })?
                .to_owned();

            // The CLI reports which models this account may actually use, on
            // every session/new. The pooled path dropped it, so a runner that
            // only ever went through `complete()` answered `available_models()`
            // from the compiled-in catalog forever — the same staleness that
            // once paged on a model that worked.
            let observed = models_from_session(&resp);

            info!(
                session_id = %session_id,
                model = %model,
                reported_models = observed.as_ref().map_or(0, Vec::len),
                "ACP session created"
            );

            // Run the tool loop to completion when MCP tools are available
            // (see `set_autopilot_mode`); best-effort.
            if !mcp_servers.is_empty() {
                match set_autopilot_mode(&mut self.transport, &session_id).await {
                    Ok(()) => info!(session_id = %session_id, "ACP: session mode set to Autopilot"),
                    Err(e) => {
                        warn!(session_id = %session_id, error = %e, "ACP: failed to set Autopilot mode");
                    }
                }
            }

            Ok::<_, RunnerError>((session_id, observed))
        })
        .await;

        match outcome {
            Ok(Ok(pair)) => Ok(pair),
            Ok(Err(e)) => Err(e),
            Err(_elapsed) => Err(RunnerError::timeout(format!(
                "copilot-acp: session/new timed out after {}s",
                acp_session_timeout().as_secs()
            ))),
        }
    }

    /// Returns true if the subprocess has not yet exited.
    ///
    /// Uses `try_wait` (non-blocking) so calling this on a healthy
    /// subprocess returns immediately without changing its state.
    fn is_alive(&mut self) -> bool {
        matches!(self.child.try_wait(), Ok(None))
    }
}

/// Newest stderr bytes kept for failure diagnostics.
const STDERR_RING_CAP: usize = 8_192;

/// Last-[`STDERR_RING_CAP`]-bytes ring of the subprocess's stderr, drained
/// continuously by a background task.
///
/// The pipe used to sit unread until a failure path sampled it: a subprocess
/// that writes more than the OS pipe buffer (~64KB) to stderr while healthy
/// then blocks on the write, and the whole ACP session freezes with no
/// diagnostic — indistinguishable from a hung model. The drain task consumes
/// the pipe for the child's entire lifetime and keeps only the newest bytes;
/// it exits on pipe EOF when the child dies. Snapshots are synchronous and
/// safe to take at any point, including after the child was killed.
#[derive(Clone)]
struct StderrRing {
    buf: Arc<StdMutex<VecDeque<u8>>>,
}

impl StderrRing {
    /// Start draining `stderr`; with `None` (pipe never captured) every
    /// snapshot reports "(empty)" rather than blocking.
    fn drain(stderr: Option<ChildStderr>) -> Self {
        let buf = Arc::new(StdMutex::new(VecDeque::new()));
        if let Some(mut pipe) = stderr {
            let ring = Arc::clone(&buf);
            tokio::spawn(async move {
                let mut chunk = [0u8; 1024];
                loop {
                    match pipe.read(&mut chunk).await {
                        Ok(0) | Err(_) => break,
                        Ok(n) => {
                            let mut bytes = ring.lock().unwrap_or_else(PoisonError::into_inner);
                            bytes.extend(&chunk[..n]);
                            while bytes.len() > STDERR_RING_CAP {
                                bytes.pop_front();
                            }
                        }
                    }
                }
            });
        }
        Self { buf }
    }

    /// The retained stderr tail as lossy UTF-8, for failure log lines.
    fn snapshot(&self) -> String {
        let bytes = self.buf.lock().unwrap_or_else(PoisonError::into_inner);
        if bytes.is_empty() {
            return "(empty)".to_owned();
        }
        let (front, back) = bytes.as_slices();
        let mut joined = Vec::with_capacity(bytes.len());
        joined.extend_from_slice(front);
        joined.extend_from_slice(back);
        String::from_utf8_lossy(&joined).into_owned()
    }
}

// ---------------------------------------------------------------------------
// Notification and permission handling
// ---------------------------------------------------------------------------

/// Accumulated state from ACP session notifications during a prompt turn.
struct TurnAccumulator {
    content: String,
    tool_calls: Vec<ObservedToolCall>,
}

impl TurnAccumulator {
    const fn new() -> Self {
        Self {
            content: String::new(),
            tool_calls: Vec::new(),
        }
    }
}

/// Process a session/update notification, accumulating content and tool calls.
fn process_notification(params: &Value, acc: &mut TurnAccumulator) {
    process_notification_inner(params, acc, None);
}

/// Process a session/update notification with optional streaming sink.
///
/// When `event_tx` is `Some`, every observed text delta and tool-call
/// observation is also forwarded to the channel as a [`HeadlessStreamEvent`].
/// When `None`, behaves identically to [`process_notification`].
fn process_notification_streaming(
    params: &Value,
    acc: &mut TurnAccumulator,
    event_tx: &mpsc::UnboundedSender<Result<HeadlessStreamEvent, RunnerError>>,
) {
    process_notification_inner(params, acc, Some(event_tx));
}

fn process_notification_inner(
    params: &Value,
    acc: &mut TurnAccumulator,
    event_tx: Option<&mpsc::UnboundedSender<Result<HeadlessStreamEvent, RunnerError>>>,
) {
    let Some(params) = params.get("params").or(Some(params)) else {
        return;
    };

    let Ok(notif) = serde_json::from_value::<schema::SessionNotification>(params.clone()) else {
        return;
    };

    match &notif.update {
        schema::SessionUpdate::AgentMessageChunk(chunk) => {
            if let schema::ContentBlock::Text(text) = &chunk.content {
                acc.content.push_str(&text.text);
                if let Some(tx) = event_tx {
                    let _ = tx.send(Ok(HeadlessStreamEvent::TextDelta(text.text.clone())));
                }
            }
        }
        schema::SessionUpdate::ToolCall(tc) => {
            let observed = ObservedToolCall {
                id: tc.tool_call_id.0.to_string(),
                title: tc.title.clone(),
                status: format!("{:?}", tc.status),
            };
            acc.tool_calls.push(observed.clone());
            if let Some(tx) = event_tx {
                let _ = tx.send(Ok(HeadlessStreamEvent::ToolCall(observed)));
            }
        }
        schema::SessionUpdate::ToolCallUpdate(update) => {
            let update_id = update.tool_call_id.0.to_string();
            if let Some(existing) = acc.tool_calls.iter_mut().find(|t| t.id == update_id) {
                if let Some(ref title) = update.fields.title {
                    existing.title.clone_from(title);
                }
                if let Some(ref status) = update.fields.status {
                    existing.status = format!("{status:?}");
                }
                if let Some(tx) = event_tx {
                    let _ = tx.send(Ok(HeadlessStreamEvent::ToolCall(existing.clone())));
                }
            }
        }
        _ => {}
    }
}

/// Serialize a permission outcome through the schema type, so the wire shape
/// is the protocol's — internally tagged: `{"outcome": {"outcome":
/// "cancelled"}}` / `{"outcome": {"outcome": "selected", "optionId": ...}}`.
///
/// The previous hand-rolled JSON emitted `{"outcome": "cancelled"}` (a bare
/// string where the protocol wants a tagged object) and dropped the
/// `"outcome": "selected"` discriminator on approvals. An agent that parses
/// strictly treats such a response as unanswered and keeps the permission
/// request pending — parking the whole session in silence, which under the
/// production `deny_all` policy means every permission prompt was a stall.
fn permission_response(outcome: schema::RequestPermissionOutcome) -> Value {
    serde_json::to_value(schema::RequestPermissionResponse::new(outcome))
        .unwrap_or_else(|_| json!({ "outcome": { "outcome": "cancelled" } }))
}

/// Build a permission response based on the configured policy.
///
/// With `AutoApprove`: selects `AllowAlways` over `AllowOnce`. If no allow option
/// exists, cancels the request instead of falling back to a reject option.
/// With `DenyAll`: always cancels the request.
fn build_permission_response(params: &Value, policy: PermissionPolicy) -> Value {
    if policy == PermissionPolicy::DenyAll {
        debug!("Permission policy is DenyAll, cancelling");
        return permission_response(schema::RequestPermissionOutcome::Cancelled);
    }

    let Ok(req) = serde_json::from_value::<schema::RequestPermissionRequest>(params.clone()) else {
        warn!("Failed to parse permission request, cancelling");
        return permission_response(schema::RequestPermissionOutcome::Cancelled);
    };

    // Prefer AllowAlways over AllowOnce for fewer repeated prompts
    let option_id = req
        .options
        .iter()
        .find(|o| matches!(o.kind, schema::PermissionOptionKind::AllowAlways))
        .or_else(|| {
            req.options
                .iter()
                .find(|o| matches!(o.kind, schema::PermissionOptionKind::AllowOnce))
        })
        .map(|o| &o.option_id);

    option_id.map_or_else(
        || {
            warn!("Permission request had no allow options, cancelling");
            permission_response(schema::RequestPermissionOutcome::Cancelled)
        },
        |id| {
            debug!(?id, "Auto-approving permission request");
            permission_response(schema::RequestPermissionOutcome::Selected(
                schema::SelectedPermissionOutcome::new(id.clone()),
            ))
        },
    )
}

/// Extract token usage from the prompt response JSON.
///
/// ACP returns usage at `/result/usage` with camelCase fields: `totalTokens`,
/// `inputTokens`, `outputTokens`, and — on agents that implement the unstable
/// session-usage capability — `cachedReadTokens`, `cachedWriteTokens` and
/// `thoughtTokens`.
///
/// Those last three were read past for a long time, and because a dropped field
/// is indistinguishable from an absent one, a downstream project concluded the
/// transport did not carry them and registered it as a limitation. It does.
/// Two consecutive live Copilot turns (2026-08-27, claude-opus-4.8):
///
/// ```text
/// {"cachedReadTokens":15320,"cachedWriteTokens":12540,"inputTokens":27862,
///  "outputTokens":4,"thoughtTokens":0,"totalTokens":27866}
/// ```
///
/// 55% of the prompt served from cache, reported on every call, discarded here.
/// Note the shape of a COLD turn — a large `cachedWriteTokens` with
/// `cachedReadTokens: 0` — which looks exactly like an agent that does not report
/// reads. Only the second, warm turn distinguishes them; see
/// `examples/acp_usage_probe.rs`, which is deliberately two-turn for that reason.
fn extract_usage(result: &Value) -> Option<TokenUsage> {
    let usage = result
        .pointer("/result/usage")
        .or_else(|| result.get("usage"))?;

    // The whole object, before we pick fields off it. ACP's `Usage` also defines
    // `cachedReadTokens`, `cachedWriteTokens` and `thoughtTokens`
    // (agent-client-protocol-schema, `Usage`), all optional on an unstable
    // capability — so whether an agent populates them is an empirical question,
    // not a schema one. Logging the raw object is how that gets answered without
    // guessing.
    debug!(usage = %usage, "ACP usage payload");

    let input = usage.get("inputTokens").and_then(Value::as_u64)?;
    let output = usage.get("outputTokens").and_then(Value::as_u64)?;
    let total = usage
        .get("totalTokens")
        .and_then(Value::as_u64)
        .unwrap_or(input + output);

    // Optional per the schema, so `None` means the agent said nothing — which
    // must stay distinct from `Some(0)`, "the agent measured zero". Collapsing
    // those is what made a hardcoded zero read as a measurement.
    let cached_read = usage.get("cachedReadTokens").and_then(Value::as_u64);
    let cached_write = usage.get("cachedWriteTokens").and_then(Value::as_u64);
    let thoughts = usage.get("thoughtTokens").and_then(Value::as_u64);

    #[allow(clippy::cast_possible_truncation)]
    Some(
        TokenUsage::new(input as u32, output as u32, total as u32)
            .with_cache(
                cached_read.map(|v| v as u32),
                cached_write.map(|v| v as u32),
            )
            .with_reasoning(thoughts.map(|v| v as u32)),
    )
}

fn map_stop_reason(reason: &str) -> &'static str {
    match reason {
        "max_tokens" => "length",
        "max_turn_requests" => "max_turns",
        "refusal" => "refusal",
        "cancelled" => "cancelled",
        _ => "stop",
    }
}

// ---------------------------------------------------------------------------
// Message collection loops
// ---------------------------------------------------------------------------

/// Read messages until prompt completes, collecting all content.
async fn collect_complete(
    transport: &mut AcpTransport,
    prompt_id: i64,
    model: String,
    policy: PermissionPolicy,
    session_id: &str,
) -> Result<(ChatResponse, Vec<ObservedToolCall>), RunnerError> {
    let mut acc = TurnAccumulator::new();
    let mut message_count: u32 = 0;

    loop {
        let msg = transport.read_message().await?;
        message_count += 1;

        if message_count == 1 {
            info!("ACP: receiving first message from copilot");
        }
        // Log method notifications for visibility (every 10th to avoid spam)
        if let Some(method) = msg.get("method").and_then(Value::as_str) {
            if message_count <= 5 || message_count.is_multiple_of(10) {
                debug!(method, message_count, "ACP notification received");
            }
        }

        // Prompt response — the turn is complete
        if msg.get("id").and_then(Value::as_i64) == Some(prompt_id) {
            if let Some(error) = msg.get("error") {
                return Err(RunnerError::external_service(
                    "copilot-acp",
                    format!("Prompt failed: {error}"),
                ));
            }

            let stop_reason = msg
                .pointer("/result/stopReason")
                .and_then(Value::as_str)
                .unwrap_or("end_turn");

            let usage = extract_usage(&msg);

            debug!(
                content_len = acc.content.len(),
                tool_calls = acc.tool_calls.len(),
                model = %model,
                has_usage = usage.is_some(),
                "Copilot Headless complete() response"
            );

            let response = ChatResponse {
                content: acc.content,
                model,
                usage,
                finish_reason: Some(map_stop_reason(stop_reason).to_owned()),
                warnings: None,
                tool_calls: None,
            };

            return Ok((response, acc.tool_calls));
        }

        // Server requests and notifications
        handle_server_message(&msg, transport, &mut acc, policy, session_id).await?;
    }
}

/// Read messages until prompt completes, streaming chunks via channel.
async fn collect_streaming(
    transport: &mut AcpTransport,
    prompt_id: i64,
    chunk_tx: &mpsc::UnboundedSender<Result<StreamChunk, RunnerError>>,
    policy: PermissionPolicy,
    session_id: &str,
) -> Result<(), RunnerError> {
    let mut acc = TurnAccumulator::new();

    loop {
        let msg = transport.read_message().await?;

        // Prompt response — the turn is complete
        if msg.get("id").and_then(Value::as_i64) == Some(prompt_id) {
            if let Some(error) = msg.get("error") {
                return Err(RunnerError::external_service(
                    "copilot-acp",
                    format!("Prompt failed: {error}"),
                ));
            }

            let stop_reason = msg
                .pointer("/result/stopReason")
                .and_then(Value::as_str)
                .unwrap_or("end_turn");

            let _ = chunk_tx.send(Ok(StreamChunk {
                delta: String::new(),
                is_final: true,
                finish_reason: Some(map_stop_reason(stop_reason).to_owned()),
            }));

            return Ok(());
        }

        // Server requests and notifications
        if let Some(method) = msg.get("method").and_then(Value::as_str) {
            match method {
                "session/update" => {
                    if let Some(params) = msg.get("params") {
                        if !notification_is_for_session(params, session_id) {
                            debug!("ACP: dropped a session/update from another session");
                            continue;
                        }
                        // Try to extract text delta for streaming
                        if let Ok(notif) =
                            serde_json::from_value::<schema::SessionNotification>(params.clone())
                        {
                            if let schema::SessionUpdate::AgentMessageChunk(chunk) = &notif.update {
                                if let schema::ContentBlock::Text(text) = &chunk.content {
                                    let _ = chunk_tx.send(Ok(StreamChunk {
                                        delta: text.text.clone(),
                                        is_final: false,
                                        finish_reason: None,
                                    }));
                                }
                            }
                        }
                        // Also track tool calls for internal accounting
                        process_notification(params, &mut acc);
                    }
                }
                "session/request_permission" => {
                    if let (Some(id), Some(params)) = (msg.get("id"), msg.get("params")) {
                        let response = build_permission_response(params, policy);
                        transport.send_response(id, response).await?;
                    }
                }
                _ => {}
            }
        }
    }
}

/// Handle a server-to-client message (notification or request).
/// Whether a `session/update` notification belongs to the turn being read.
///
/// A pooled subprocess serves many sessions over its life, one athlete after
/// another, and nothing in the transport separates their notifications: the
/// read loop folds every `session/update` it sees into the accumulator until
/// the prompt response arrives. A notification left in the pipe by an earlier
/// session would therefore be appended to a later athlete's reply — a silent
/// wrong answer, and the more dangerous half of it is that `converse()` output
/// is what the athlete actually reads.
///
/// Copilot tags every notification (verified on the wire, CLI 1.0.81:
/// `session/update` params are `["sessionId", "update"]`). An UNTAGGED
/// notification is accepted rather than dropped — an agent that does not tag
/// cannot be filtered, and silently discarding its content would be a worse
/// failure than the one this prevents.
fn notification_is_for_session(params: &Value, session_id: &str) -> bool {
    params
        .get("sessionId")
        .and_then(Value::as_str)
        .is_none_or(|id| id == session_id)
}

async fn handle_server_message(
    msg: &Value,
    transport: &mut AcpTransport,
    acc: &mut TurnAccumulator,
    policy: PermissionPolicy,
    session_id: &str,
) -> Result<(), RunnerError> {
    if let Some(method) = msg.get("method").and_then(Value::as_str) {
        match method {
            "session/update" => {
                if let Some(params) = msg.get("params") {
                    if notification_is_for_session(params, session_id) {
                        process_notification(params, acc);
                    } else {
                        debug!("ACP: dropped a session/update from another session");
                    }
                }
            }
            "session/request_permission" => {
                if let (Some(id), Some(params)) = (msg.get("id"), msg.get("params")) {
                    let response = build_permission_response(params, policy);
                    transport.send_response(id, response).await?;
                }
            }
            _ => {}
        }
    }
    Ok(())
}

/// Streaming variant of [`handle_server_message`].
///
/// Identical to the non-streaming form except that every observed
/// content/tool-call notification is also forwarded to `event_tx` as a
/// [`HeadlessStreamEvent`]. Permission-request handling is unchanged.
async fn handle_server_message_streaming(
    msg: &Value,
    transport: &mut AcpTransport,
    acc: &mut TurnAccumulator,
    policy: PermissionPolicy,
    event_tx: &mpsc::UnboundedSender<Result<HeadlessStreamEvent, RunnerError>>,
    session_id: &str,
) -> Result<(), RunnerError> {
    if let Some(method) = msg.get("method").and_then(Value::as_str) {
        match method {
            "session/update" => {
                if let Some(params) = msg.get("params") {
                    if notification_is_for_session(params, session_id) {
                        process_notification_streaming(params, acc, event_tx);
                    } else {
                        debug!("ACP: dropped a session/update from another session");
                    }
                }
            }
            "session/request_permission" => {
                if let (Some(id), Some(params)) = (msg.get("id"), msg.get("params")) {
                    let response = build_permission_response(params, policy);
                    transport.send_response(id, response).await?;
                }
            }
            _ => {}
        }
    }
    Ok(())
}

/// Read messages until prompt completes, emitting [`HeadlessStreamEvent`]s
/// as ACP notifications arrive while accumulating the same final state
/// that [`collect_complete`] produces.
///
/// On success returns the aggregated [`HeadlessToolResponse`] so the
/// caller can emit a final [`HeadlessStreamEvent::Done`] event.
async fn collect_streaming_with_tools(
    transport: &mut AcpTransport,
    prompt_id: i64,
    model: String,
    policy: PermissionPolicy,
    event_tx: &mpsc::UnboundedSender<Result<HeadlessStreamEvent, RunnerError>>,
    session_id: &str,
) -> Result<HeadlessToolResponse, RunnerError> {
    let mut acc = TurnAccumulator::new();

    loop {
        let msg = transport.read_message().await?;

        // Prompt response — the turn is complete
        if msg.get("id").and_then(Value::as_i64) == Some(prompt_id) {
            if let Some(error) = msg.get("error") {
                return Err(RunnerError::external_service(
                    "copilot-acp",
                    format!("Prompt failed: {error}"),
                ));
            }

            let stop_reason = msg
                .pointer("/result/stopReason")
                .and_then(Value::as_str)
                .unwrap_or("end_turn");
            let usage = extract_usage(&msg);

            return Ok(HeadlessToolResponse {
                content: acc.content,
                model,
                tool_calls: acc.tool_calls,
                usage,
                finish_reason: Some(map_stop_reason(stop_reason).to_owned()),
            });
        }

        handle_server_message_streaming(&msg, transport, &mut acc, policy, event_tx, session_id)
            .await?;
    }
}

// ---------------------------------------------------------------------------
// Public types
// ---------------------------------------------------------------------------

/// A tool call observed during an ACP session turn.
#[derive(Debug, Clone)]
pub struct ObservedToolCall {
    /// Tool call ID from the ACP protocol.
    pub id: String,
    /// Human-readable title describing the tool action.
    pub title: String,
    /// Execution status (e.g., "Pending", "`InProgress`", "Completed", "Failed").
    pub status: String,
}

/// Response from a headless conversation turn including tool execution metadata.
#[derive(Debug, Clone)]
pub struct HeadlessToolResponse {
    /// Final assistant response content.
    pub content: String,
    /// Model that generated the response.
    pub model: String,
    /// Tool calls observed during the turn.
    pub tool_calls: Vec<ObservedToolCall>,
    /// Token usage for this turn.
    pub usage: Option<TokenUsage>,
    /// Finish reason.
    pub finish_reason: Option<String>,
}

/// Event emitted by [`CopilotHeadlessRunner::converse_stream`] as the ACP
/// turn progresses.
///
/// Unlike [`StreamChunk`] (which only carries text deltas), this enum
/// surfaces tool-call observations alongside text — keeping the rich
/// metadata that [`CopilotHeadlessRunner::converse`] returns at the end
/// of the turn while delivering it incrementally.
#[derive(Debug, Clone)]
pub enum HeadlessStreamEvent {
    /// Partial assistant text — the next chunk to append to the
    /// in-flight assistant message.
    TextDelta(String),
    /// A tool call was observed (start or status update). Each event
    /// is a snapshot of the tool call's latest known state, so a
    /// consumer can either accumulate updates or replace by id.
    ToolCall(ObservedToolCall),
    /// The turn has finished. Carries the aggregated
    /// [`HeadlessToolResponse`] — same shape that
    /// [`CopilotHeadlessRunner::converse`] would have returned.
    /// Always emitted as the last event before the stream closes
    /// successfully.
    Done(HeadlessToolResponse),
}

/// Stream of [`HeadlessStreamEvent`]s for a single converse turn.
pub type HeadlessEventStream =
    Pin<Box<dyn Stream<Item = Result<HeadlessStreamEvent, RunnerError>> + Send>>;

/// Why a warm subprocess must not serve the turn in front of it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DiscardReason {
    /// The child exited between calls — crash, OOM kill, upstream disconnect.
    Exited,
    /// `--model` is fixed at spawn, so a process pinned to model A cannot
    /// serve a request for model B.
    ModelChanged,
    /// It has served its share of the memory budget. Nothing frees an ACP
    /// session, so this is the only bound on their accumulation.
    SessionCeiling,
}

/// Decide whether a warm subprocess may serve the next turn.
///
/// Pure, so the rule can be tested without a subprocess: every caller of the
/// pool asks this one question, and the answer is what separates "reuse the
/// warm process" from "kill it and spawn a fresh one".
const fn discard_reason(
    alive: bool,
    model_matches: bool,
    sessions_served: u32,
    max_sessions: u32,
) -> Option<DiscardReason> {
    if !alive {
        return Some(DiscardReason::Exited);
    }
    if !model_matches {
        return Some(DiscardReason::ModelChanged);
    }
    if sessions_served >= max_sessions {
        return Some(DiscardReason::SessionCeiling);
    }
    None
}

/// Everything one pooled turn needs, so the two entry points hand it over as a
/// single value rather than a seven-argument call.
struct TurnRequest<'a> {
    cli_path: &'a PathBuf,
    model: &'a str,
    system_prompt: Option<&'a str>,
    prompt_blocks: &'a [Value],
    max_tokens: Option<u32>,
    mcp_servers: &'a [McpServerConfig],
    /// Which entry point is running, so a log line says which path a warm
    /// subprocess was recycled or killed on.
    caller: &'static str,
}

/// A fixed set of warm `copilot --acp` subprocesses, each behind its own lock.
///
/// The ACP wire format carries no response routing, so two prompts must never
/// share one transport. The pool respects that by never multiplexing a slot:
/// concurrency comes from holding several independent subprocesses, each
/// serving one completion at a time.
///
/// A slot holds `None` until first use and returns to `None` whenever its
/// subprocess is discarded, so every slot self-heals on the next checkout.
struct AcpPool {
    slots: Vec<Arc<TokioMutex<Option<AcpProcess>>>>,
    /// Sessions a slot may serve before recycling, derived from the pool size
    /// so the whole pool stays inside [`ACP_POOL_BUDGET_MB`].
    max_sessions: u32,
    /// Round-robin cursor, so a burst of concurrent turns spreads across slots
    /// instead of contending on slot 0 and queueing behind it.
    next: AtomicUsize,
}

impl AcpPool {
    fn new(size: usize) -> Self {
        let size = size.max(1);
        Self {
            slots: (0..size).map(|_| Arc::new(TokioMutex::new(None))).collect(),
            next: AtomicUsize::new(0),
            max_sessions: max_sessions_per_process(size),
        }
    }

    /// Take exclusive use of one slot for the duration of a completion.
    ///
    /// Tries every slot once without blocking, so an idle subprocess is always
    /// preferred over waiting. When all slots are busy it queues on one rather
    /// than spinning or spawning past the configured size — the pool is a
    /// bound, and saturation degrades to the single-slot behaviour it replaced
    /// rather than to unbounded subprocess growth.
    async fn checkout(&self) -> OwnedMutexGuard<Option<AcpProcess>> {
        let len = self.slots.len();
        let start = self.next.fetch_add(1, Ordering::Relaxed);
        for offset in 0..len {
            let slot = &self.slots[(start.wrapping_add(offset)) % len];
            if let Ok(guard) = Arc::clone(slot).try_lock_owned() {
                return guard;
            }
        }
        Arc::clone(&self.slots[start % len]).lock_owned().await
    }
}

// ---------------------------------------------------------------------------
// Public runner
// ---------------------------------------------------------------------------

/// GitHub Copilot Headless (ACP) LLM provider.
///
/// Communicates with `copilot --acp` via the Agent Client Protocol (JSON-RPC over stdio).
/// Spawns a new copilot subprocess per request using NDJSON framing.
/// Uses types from `agent-client-protocol-schema` for protocol message deserialization.
///
/// Copilot manages its own tool execution internally (GitHub tools, code search),
/// but cannot execute external MCP tools. Tool calls are observed and reported
/// via [`HeadlessToolResponse`] from [`converse()`](Self::converse).
/// For custom tools, callers should use text-based tool calling (CLI tool loop).
pub struct CopilotHeadlessRunner {
    config: CopilotHeadlessConfig,
    /// Ranked catalog from [`crate::copilot_models`], used until the CLI tells
    /// us otherwise. A constant cannot track what the vendor ships, and cannot
    /// express per-account entitlement at all.
    available_models: Vec<String>,
    /// What `session/new` actually reported, once we have seen it.
    ///
    /// The catalog listed 21 models on 2026-08-24 while the CLI reported 28 for
    /// the account in front of it — `claude-sonnet-5` among the missing, which
    /// is the model every coaching turn runs on. A platform-side check against
    /// the stale list duly paged on a working model.
    ///
    /// `OnceLock` rather than a lock around a mutable field because
    /// [`EmbacleLlmProvider::available_models`] hands back a `&[String]` tied to
    /// `&self`, which a guard cannot outlive. First session wins: an account's
    /// entitlements do not change mid-process, and a restart re-reads them.
    observed_models: OnceLock<Vec<String>>,
    /// Long-lived `copilot --acp` subprocess + initialized transport.
    ///
    /// Lazily spawned on the first `complete()` call and kept warm across
    /// calls so the GitHub→Copilot OAuth token exchange amortizes across
    /// the subprocess lifetime instead of running per request. Cleared and
    /// respawned on subprocess death or any complete()-path error so the
    /// next call always starts from a known-good state.
    ///
    /// Each slot is wrapped in `tokio::sync::Mutex` because the NDJSON
    /// transport is request/response per JSON-RPC id and cannot interleave
    /// concurrent prompts without response routing — which the ACP wire
    /// format does not support. Concurrency comes from the number of slots,
    /// never from sharing one.
    pool: AcpPool,
    /// Admission control for the streaming paths, which spawn a dedicated
    /// subprocess per call rather than borrowing a pool slot.
    ///
    /// Without it nothing bounds how many `copilot --acp` children exist at
    /// once — 213 MB each, on an instance that accepts 80 concurrent requests
    /// inside 2Gi. A permit is held for the whole streamed turn and released
    /// when the background task ends, however it ends.
    stream_permits: Arc<Semaphore>,
}

impl CopilotHeadlessRunner {
    /// Create a new provider from environment configuration.
    ///
    /// The set of available models is taken from the ranked catalog in
    /// [`crate::copilot_models`]. Availability per account is resolved lazily
    /// by the Copilot CLI runner's self-heal loop, not at construction time.
    #[must_use]
    pub fn from_env() -> Self {
        Self {
            config: CopilotHeadlessConfig::from_env(),
            available_models: catalog_ids(),
            observed_models: OnceLock::new(),
            pool: AcpPool::new(acp_pool_size()),
            stream_permits: Arc::new(Semaphore::new(max_concurrent_streams())),
        }
    }

    /// Create a new provider with explicit configuration.
    #[must_use]
    pub fn with_config(config: CopilotHeadlessConfig) -> Self {
        Self {
            config,
            available_models: catalog_ids(),
            observed_models: OnceLock::new(),
            pool: AcpPool::new(acp_pool_size()),
            stream_permits: Arc::new(Semaphore::new(max_concurrent_streams())),
        }
    }

    /// Remember the model list the CLI reported, the first time it reports one.
    ///
    /// Later sessions are ignored rather than overwriting: an account's
    /// entitlements do not change mid-process, and `available_models()` hands
    /// out a slice borrowed from `&self`, which a value that can be swapped
    /// underneath it could not safely provide.
    fn record_observed_models(&self, observed: Option<Vec<String>>) {
        let Some(models) = observed else {
            return;
        };
        if self.observed_models.set(models).is_ok() {
            debug!(
                count = self.observed_models.get().map_or(0, Vec::len),
                catalog = self.available_models.len(),
                "ACP reported the account's model list; superseding the catalog"
            );
        }
    }

    /// Resolve the copilot CLI binary path.
    fn resolve_cli_path(&self) -> Result<PathBuf, RunnerError> {
        if let Some(ref path) = self.config.cli_path {
            return Ok(path.clone());
        }
        which::which("copilot").map_err(|_| RunnerError::binary_not_found("copilot"))
    }

    /// Resolve the model to use for a request.
    /// Provider-alias names (`copilot_headless`, `copilot`) are mapped to the
    /// configured default model because they are not valid Copilot model identifiers.
    /// Any other model name (e.g. `gpt-4.1`, `claude-opus-4.7`) is passed through.
    fn resolve_model(&self, requested: Option<&str>) -> String {
        match requested {
            Some(m) if m != "copilot_headless" && m != "copilot" => m.to_owned(),
            _ => self.config.model.clone(),
        }
    }

    /// Build ACP prompt content blocks from the conversation messages.
    ///
    /// ACP creates a fresh session per request with no built-in multi-turn memory.
    /// To provide conversation continuity, prior user/assistant exchanges are
    /// serialized into a `<conversation-history>` block prepended to the prompt.
    ///
    /// The system prompt is prepended as plain text to the prompt. It is also
    /// sent via ACP `session/new` `systemPrompt`, but Copilot CLI's request
    /// schema strips unknown keys, so that field never reaches the model —
    /// verified against the pinned CLI. The prompt-text copy is the delivery.
    ///
    /// The number of history messages is capped by `max_history_turns` from
    /// [`CopilotHeadlessConfig`]. Only the most recent turns are kept.
    ///
    /// Always includes a text block. When the last user message has images,
    /// appends image blocks with `type: "image"`, `data`, and `mimeType`.
    fn build_prompt_blocks(&self, request: &ChatRequest) -> Vec<Value> {
        // ALWAYS inlined. Copilot CLI's ACP `session/new` schema silently strips
        // unknown keys, so the `systemPrompt` field never reaches the model —
        // prompt-text inlining is the only delivery path this runner has. The
        // former `inject_system_in_prompt` knob had no correct `false` value: it
        // did not select an alternative mechanism, it selected none, shipping
        // every turn with no persona and no safety scaffolding, silently.
        let system = Self::extract_system_prompt(request);
        let max_turns = self.config.max_history_turns;

        // Separate non-system messages into history (all but last user) + last user
        let non_system: Vec<&ChatMessage> = request
            .messages
            .iter()
            .filter(|m| m.role != MessageRole::System)
            .collect();

        let (history, last_user) = if non_system.is_empty() {
            (Vec::new(), None)
        } else {
            let last_idx = non_system.iter().rposition(|m| m.role == MessageRole::User);
            match last_idx {
                Some(idx) => {
                    let hist = non_system[..idx].to_vec();
                    (hist, Some(non_system[idx]))
                }
                None => (non_system, None),
            }
        };

        let user_text = last_user.map(|m| m.content.as_str()).unwrap_or_default();

        // Apply max_history_turns limit — keep only the most recent turns
        let truncated_history = if max_turns == 0 || history.is_empty() {
            &[][..]
        } else if history.len() > max_turns {
            &history[history.len() - max_turns..]
        } else {
            &history
        };

        // Serialize prior turns into a conversation history block
        let history_block = if truncated_history.is_empty() {
            String::new()
        } else {
            let mut buf = String::from("<conversation-history>\n");
            for msg in truncated_history {
                let role_label = match msg.role {
                    MessageRole::User => "User",
                    MessageRole::Assistant => "Assistant",
                    MessageRole::Tool => "Tool",
                    MessageRole::System => continue,
                };
                buf.push_str(role_label);
                buf.push_str(": ");
                buf.push_str(&msg.content);
                buf.push('\n');
            }
            buf.push_str("</conversation-history>\n\n");
            buf
        };

        // Assemble: system prompt + conversation history + current user message
        let mut text = String::new();
        if let Some(sys) = system {
            text.push_str(sys);
            text.push_str("\n\n");
        }
        text.push_str(&history_block);
        text.push_str(user_text);

        let mut blocks = vec![json!({"type": "text", "text": text})];

        if let Some(images) = last_user.and_then(|m| m.images.as_ref()) {
            for img in images {
                blocks.push(json!({
                    "type": "image",
                    "data": img.data,
                    "mimeType": img.mime_type,
                }));
            }
        }

        blocks
    }

    /// Extract the system prompt if present.
    fn extract_system_prompt(request: &ChatRequest) -> Option<&str> {
        request
            .messages
            .iter()
            .find(|m| m.role == MessageRole::System)
            .map(|m| m.content.as_str())
    }
    /// Run one turn on a pooled subprocess.
    ///
    /// Checks out a slot, makes sure the process in it is fit to serve this
    /// request, opens a session, sends the prompt and collects the answer.
    /// On success the subprocess is left alive for the next turn; on ANY
    /// failure it is killed and its slot cleared, so the next caller starts
    /// from a known-good state rather than inheriting a desynced transport.
    ///
    /// Both entry points run through here. `complete()` adds the empty-turn
    /// retry and discards the tool calls; `converse()` keeps them. Duplicating
    /// the checkout/discard/kill logic into each was the alternative, and it is
    /// exactly the shape that lets one path quietly stop killing a child.
    /// Leave the slot holding a subprocess fit to serve this turn.
    ///
    /// Discards whatever is there when it has exited, is pinned to a different
    /// model, or has spent its share of the memory budget, then spawns a
    /// replacement if the slot ended up empty.
    async fn ensure_fit_process(
        &self,
        guard: &mut OwnedMutexGuard<Option<AcpProcess>>,
        cli_path: &PathBuf,
        model: &str,
        caller: &'static str,
    ) -> Result<(), RunnerError> {
        if let Some(p) = guard.as_mut() {
            if let Some(reason) = discard_reason(
                p.is_alive(),
                p.model == model,
                p.sessions_served,
                self.pool.max_sessions,
            ) {
                match reason {
                    DiscardReason::Exited => {
                        warn!(caller, "ACP subprocess exited between calls; respawning");
                    }
                    DiscardReason::ModelChanged => info!(
                        caller,
                        warm_model = %p.model,
                        requested_model = %model,
                        "ACP requested model changed; respawning warm subprocess to re-pin --model"
                    ),
                    DiscardReason::SessionCeiling => info!(
                        caller,
                        sessions_served = p.sessions_served,
                        "ACP subprocess reached its session ceiling; recycling"
                    ),
                }
                // Kill before dropping, for every reason except a child that has
                // already exited. Tokio's `Child` does not kill on drop unless
                // `kill_on_drop(true)` was set at spawn, which it is not, so
                // `**guard = None` alone orphans a live `copilot --acp`
                // subprocess for the lifetime of the server.
                //
                // The ceiling exists because an ACP session cannot be closed —
                // `session/close` is advertised and accepted but frees nothing
                // (measured) — so a warm subprocess holds ~26 MB for every
                // session it has ever served. Unbounded, that walks a 2Gi
                // container into an OOM.
                if reason != DiscardReason::Exited {
                    let _ = p.child.kill().await;
                }
                **guard = None;
            }
        }

        if guard.is_none() {
            let fresh = AcpProcess::spawn_and_initialize(
                cli_path,
                self.config.github_token.as_deref(),
                model,
            )
            .await?;
            **guard = Some(fresh);
        }
        Ok(())
    }

    /// Run one turn on a pooled subprocess.
    ///
    /// On success the subprocess is left alive for the next turn; on ANY
    /// failure it is killed and its slot cleared, so the next caller starts
    /// from a known-good state rather than inheriting a desynced transport.
    ///
    /// Both entry points run through here. `complete()` adds the empty-turn
    /// retry and discards the tool calls; `converse()` keeps them. Duplicating
    /// the checkout/discard/kill logic into each was the alternative, and it is
    /// exactly the shape that lets one path quietly stop killing a child.
    async fn run_pooled_turn(
        &self,
        turn: &TurnRequest<'_>,
    ) -> Result<(ChatResponse, Vec<ObservedToolCall>), RunnerError> {
        // Hold ONE POOL SLOT for the full RPC round-trip. ACP transport is
        // request/response by JSON-RPC id with no response-routing in the wire
        // format, so concurrent prompts would interleave reads unsafely — this
        // lock is what prevents that, and it is per subprocess, not server-wide.
        let mut guard = self.pool.checkout().await;
        self.ensure_fit_process(&mut guard, turn.cli_path, turn.model, turn.caller)
            .await?;

        let Some(process) = guard.as_mut() else {
            return Err(RunnerError::internal(
                "copilot-acp: pool slot empty after ensuring a process",
            ));
        };

        let session_id = match process
            .new_session(turn.model, turn.system_prompt, turn.mcp_servers)
            .await
        {
            Ok((id, observed)) => {
                process.sessions_served = process.sessions_served.saturating_add(1);
                self.record_observed_models(observed);
                id
            }
            Err(e) => {
                // session/new failed on a previously-healthy subprocess. Could
                // be the cached Copilot OAuth token expired and the CLI didn't
                // auto-refresh, or the process is wedged. Kill and clear so the
                // next call respawns and re-authenticates.
                let _ = process.child.kill().await;
                *guard = None;
                return Err(e);
            }
        };

        info!(
            caller = turn.caller,
            session_id = %session_id,
            prompt_blocks = turn.prompt_blocks.len(),
            "ACP: sending prompt"
        );
        if tracing::enabled!(tracing::Level::TRACE) {
            match serde_json::to_string(turn.prompt_blocks) {
                Ok(blocks_json) => trace!(prompt_blocks = %blocks_json, "ACP prompt blocks"),
                Err(e) => trace!(error = %e, "ACP prompt blocks serialization failed"),
            }
        }

        let prompt_id = match process
            .transport
            .send_request(
                "session/prompt",
                build_prompt_params(&session_id, turn.prompt_blocks, turn.max_tokens),
            )
            .await
        {
            Ok(id) => id,
            Err(e) => {
                // Writing to stdin failed — pipe is broken. Discard.
                let _ = process.child.kill().await;
                *guard = None;
                return Err(e);
            }
        };

        let started = Instant::now();
        let result = time::timeout(
            acp_prompt_timeout(),
            collect_complete(
                &mut process.transport,
                prompt_id,
                turn.model.to_owned(),
                self.config.permission_policy,
                &session_id,
            ),
        )
        .await;

        match result {
            Ok(Ok((response, tool_calls))) => {
                info!(
                    caller = turn.caller,
                    latency_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
                    content_len = response.content.len(),
                    tool_calls = tool_calls.len(),
                    finish_reason = response.finish_reason.as_deref().unwrap_or("none"),
                    "ACP: response received"
                );
                if tracing::enabled!(tracing::Level::TRACE) {
                    trace!(content = %response.content, "ACP response body");
                }
                // SUCCESS: leave the subprocess alive so the next turn reuses
                // it. This is the whole point of the pool.
                Ok((response, tool_calls))
            }
            Ok(Err(e)) => {
                // Prompt failed mid-stream. Transport may be desynced; the
                // conservative choice is to kill and respawn rather than risk a
                // corrupt session bleeding into the next call.
                warn!(caller = turn.caller, error = %e, stderr = %process.stderr_ring.snapshot(), "ACP turn failed");
                let _ = process.child.kill().await;
                *guard = None;
                Err(e)
            }
            Err(_elapsed) => {
                warn!(caller = turn.caller, stderr = %process.stderr_ring.snapshot(), "ACP turn timed out");
                let _ = process.child.kill().await;
                *guard = None;
                Err(RunnerError::timeout(format!(
                    "copilot-acp: prompt timed out after {}s",
                    acp_prompt_timeout().as_secs()
                )))
            }
        }
    }

    /// Run a conversation turn and return detailed results including tool call metadata.
    ///
    /// Unlike [`complete()`](LlmProvider::complete), this returns an [`HeadlessToolResponse`]
    /// with observed tool calls that copilot executed internally during the turn.
    pub async fn converse(
        &self,
        request: &ChatRequest,
    ) -> Result<HeadlessToolResponse, RunnerError> {
        let cli_path = self.resolve_cli_path()?;
        let model = self.resolve_model(request.model.as_deref());
        let system_prompt = Self::extract_system_prompt(request);
        let prompt_blocks = self.build_prompt_blocks(request);

        // Pooled, like `complete()`. This path used to spawn a dedicated
        // `copilot --acp` child per call and kill it at the end, which paid the
        // full handshake every turn (~3.2s cold against ~1.7s warm) and put no
        // bound at all on how many children could exist at once — on an
        // instance that accepts 80 concurrent requests inside 2Gi, with each
        // child holding 213 MB before serving a single session.
        let (response, tool_calls) = self
            .run_pooled_turn(&TurnRequest {
                cli_path: &cli_path,
                model: &model,
                system_prompt,
                prompt_blocks: &prompt_blocks,
                max_tokens: request.max_tokens,
                mcp_servers: &request.mcp_servers,
                caller: "converse",
            })
            .await?;

        Ok(HeadlessToolResponse {
            content: response.content,
            model: response.model,
            tool_calls,
            usage: response.usage,
            finish_reason: response.finish_reason,
        })
    }

    /// Streaming variant of [`converse()`](Self::converse).
    ///
    /// Returns a [`HeadlessEventStream`] that yields [`HeadlessStreamEvent`]s
    /// as the ACP turn progresses:
    ///
    /// - [`HeadlessStreamEvent::TextDelta`] — partial assistant text as
    ///   `AgentMessageChunk` notifications arrive.
    /// - [`HeadlessStreamEvent::ToolCall`] — every observed tool call (and
    ///   subsequent status updates), letting the consumer surface "calling
    ///   tool X..." progress to end users while the turn is still running.
    /// - [`HeadlessStreamEvent::Done`] — the final aggregated response,
    ///   identical in shape to what [`converse()`](Self::converse) would
    ///   return. Always emitted last on success.
    ///
    /// The underlying ACP session runs in a background tokio task that
    /// owns the spawned `copilot --acp` child process; it is killed when
    /// the turn completes (success, error, or timeout). Dropping the
    /// returned stream early does **not** abort the in-flight turn — the
    /// task will still drain the session and the child will be cleaned up
    /// when the turn finishes.
    ///
    /// This path spawns per call rather than borrowing a pool slot, and that is
    /// deliberate: pooling would hold a warm slot for the whole streamed turn —
    /// up to the prompt timeout, against a default pool of two — and make
    /// `checkout().await` block the first token, which is the latency streaming
    /// exists to protect. It pays the handshake per stream in exchange.
    ///
    /// What it does NOT do any more is spawn without limit. Admission is taken
    /// from [`CopilotHeadlessRunner::stream_permits`] before the spawn, so the
    /// number of concurrent children is bounded by
    /// [`max_concurrent_streams()`] rather than by how many athletes happen to
    /// be typing — 213 MB each, in a 2Gi container that accepts 80 concurrent
    /// requests.
    ///
    /// # Errors
    ///
    /// Returns a setup-time error before any events are emitted if the
    /// CLI cannot be located or the ACP handshake fails. After the stream
    /// starts, transport / protocol failures arrive as `Err` items in the
    /// stream itself, and the configured prompt timeout
    /// (`EMBACLE_ACP_PROMPT_TIMEOUT_SECS`, default 5 min) becomes a
    /// terminal `RunnerError::Timeout` event.
    #[instrument(skip(self, request), fields(model = field::Empty))]
    pub async fn converse_stream(
        &self,
        request: &ChatRequest,
    ) -> Result<HeadlessEventStream, RunnerError> {
        let cli_path = self.resolve_cli_path()?;
        let model = self.resolve_model(request.model.as_deref());
        Span::current().record("model", field::display(&model));
        let system_prompt = Self::extract_system_prompt(request);
        let prompt_blocks = self.build_prompt_blocks(request);

        // Admission BEFORE the spawn, not after: the point of the bound is to
        // stop the subprocess existing, and a permit taken after
        // `setup_session` would have already paid the 213 MB it exists to cap.
        // The permit rides into the background task below and is released when
        // that task ends, however it ends — done, error, timeout, or an
        // abandoned stream that still drains.
        let stream_permit = Arc::clone(&self.stream_permits)
            .acquire_owned()
            .await
            .map_err(|_| RunnerError::internal("copilot-acp: stream admission semaphore closed"))?;

        let (mut transport, mut child, stderr_ring, session_id, observed) = setup_session(
            &cli_path,
            self.config.github_token.as_deref(),
            &model,
            system_prompt,
            &request.mcp_servers,
        )
        .await?;
        self.record_observed_models(observed);

        info!(session_id = %session_id, "ACP: sending streaming prompt");
        let prompt_id = transport
            .send_request(
                "session/prompt",
                build_prompt_params(&session_id, &prompt_blocks, request.max_tokens),
            )
            .await?;

        let (event_tx, event_rx) = mpsc::unbounded_channel();
        let policy = self.config.permission_policy;
        let timeout = acp_prompt_timeout();
        let model_for_task = model.clone();
        let session_for_task = session_id.clone();

        // Drive the ACP session and emit events on a background task so
        // the caller can consume the stream incrementally. The task owns
        // the transport and child, and kills the child when it finishes
        // — matching the lifecycle of `converse()`.
        tokio::spawn(async move {
            let result = time::timeout(
                timeout,
                collect_streaming_with_tools(
                    &mut transport,
                    prompt_id,
                    model_for_task,
                    policy,
                    &event_tx,
                    &session_for_task,
                ),
            )
            .await;

            match result {
                Ok(Ok(response)) => {
                    info!(
                        content_len = response.content.len(),
                        tool_calls = response.tool_calls.len(),
                        "ACP converse_stream completed successfully"
                    );
                    let _ = event_tx.send(Ok(HeadlessStreamEvent::Done(response)));
                }
                Ok(Err(e)) => {
                    warn!(error = %e, stderr = %stderr_ring.snapshot(), "ACP converse_stream failed");
                    let _ = event_tx.send(Err(e));
                }
                Err(_) => {
                    warn!(
                        stderr = %stderr_ring.snapshot(),
                        timeout_secs = timeout.as_secs(),
                        "ACP converse_stream timed out"
                    );
                    let _ = event_tx.send(Err(RunnerError::timeout(format!(
                        "copilot-acp: prompt timed out after {}s",
                        timeout.as_secs()
                    ))));
                }
            }

            let _ = child.kill().await;

            // Release admission only now, with the child reaped. Dropping it
            // any earlier would let the next streamed turn spawn while this
            // subprocess is still resident, which is the overcommit the bound
            // exists to prevent.
            drop(stream_permit);
        });

        let stream = UnboundedReceiverStream::new(event_rx);
        Ok(Box::pin(stream))
    }
}

#[async_trait]
impl LlmProvider for CopilotHeadlessRunner {
    fn name(&self) -> &'static str {
        "copilot_headless"
    }

    fn display_name(&self) -> &str {
        "GitHub Copilot (Headless)"
    }

    fn capabilities(&self) -> LlmCapabilities {
        let base =
            LlmCapabilities::STREAMING | LlmCapabilities::SYSTEM_MESSAGES | LlmCapabilities::VISION;
        // SDK_TOOL_CALLING is opt-in via `mcp_tool_calling`. When set, the
        // caller passes `mcp_servers` per request and Copilot calls those tools
        // natively over ACP — the CLI advertises `mcpCapabilities {http,sse}` at
        // initialize, so it CAN execute external MCP tools. When unset, callers
        // fall through to text-based tool calling, where the host parses
        // <tool_call> blocks and executes tools itself.
        if self.config.mcp_tool_calling {
            base | LlmCapabilities::SDK_TOOL_CALLING
        } else {
            base
        }
    }

    fn default_model(&self) -> &str {
        &self.config.model
    }

    fn available_models(&self) -> &[String] {
        self.observed_models
            .get()
            .map_or(self.available_models.as_slice(), Vec::as_slice)
    }

    #[instrument(skip_all, fields(runner = "copilot_headless", model = field::Empty))]
    async fn complete(&self, request: &ChatRequest) -> Result<ChatResponse, RunnerError> {
        let cli_path = self.resolve_cli_path()?;
        let model = self.resolve_model(request.model.as_deref());
        Span::current().record("model", field::display(&model));
        let system_prompt = Self::extract_system_prompt(request);
        let prompt_blocks = self.build_prompt_blocks(request);

        // Copilot ends a turn with zero agent-message chunks often enough to
        // matter: 2 of 11 turns in a controlled corpus run (2026-08-24). The
        // subprocess is healthy and the protocol is satisfied — `finish_reason`
        // is `stop`, latency is a normal ~20s — there is simply no content.
        //
        // A caller cannot distinguish that from a model that legitimately had
        // nothing to say, so it surfaces as a lost turn: the platform's repair,
        // identity re-ask and verifier all ride this path, and an empty answer
        // there reaches an athlete as "je n'ai pas réussi à formuler une
        // réponse". Retrying once on a fresh session recovers it without the
        // caller ever seeing the gap.
        //
        // Bounded at one extra attempt: a second empty turn is a real answer
        // about this prompt, not a flake, and every retry is a full inference.
        let mut attempt: u32 = 0;
        loop {
            let (response, _tool_calls) = self
                .run_pooled_turn(&TurnRequest {
                    cli_path: &cli_path,
                    model: &model,
                    system_prompt,
                    prompt_blocks: &prompt_blocks,
                    max_tokens: request.max_tokens,
                    mcp_servers: &request.mcp_servers,
                    caller: "complete",
                })
                .await?;

            if response.content.trim().is_empty() {
                if attempt < DEGENERATE_TURN_RETRIES {
                    // Logged under a stable message so the rate is greppable —
                    // the defect was diagnosed 2026-08-23 and could not be
                    // measured, only noticed.
                    warn!(
                        attempt,
                        finish_reason = response.finish_reason.as_deref().unwrap_or("none"),
                        "ACP complete: empty turn, retrying on a fresh session"
                    );
                    attempt += 1;
                    continue;
                }
                warn!(
                    attempt,
                    "ACP complete: empty turn survived retry, returning it"
                );
            }

            return Ok(response);
        }
    }

    async fn complete_stream(&self, request: &ChatRequest) -> Result<ChatStream, RunnerError> {
        let cli_path = self.resolve_cli_path()?;
        let model = self.resolve_model(request.model.as_deref());
        let system_prompt = Self::extract_system_prompt(request).map(str::to_owned);
        let prompt_blocks = self.build_prompt_blocks(request);

        // Admission BEFORE the spawn, not after: the point of the bound is to
        // stop the subprocess existing, and a permit taken after
        // `setup_session` would have already paid the 213 MB it exists to cap.
        // The permit rides into the background task below and is released when
        // that task ends, however it ends — done, error, timeout, or an
        // abandoned stream that still drains.
        let stream_permit = Arc::clone(&self.stream_permits)
            .acquire_owned()
            .await
            .map_err(|_| RunnerError::internal("copilot-acp: stream admission semaphore closed"))?;

        let (mut transport, mut child, stderr_ring, session_id, observed) = setup_session(
            &cli_path,
            self.config.github_token.as_deref(),
            &model,
            system_prompt.as_deref(),
            &request.mcp_servers,
        )
        .await?;
        self.record_observed_models(observed);

        let prompt_id = transport
            .send_request(
                "session/prompt",
                build_prompt_params(&session_id, &prompt_blocks, request.max_tokens),
            )
            .await?;

        let (chunk_tx, chunk_rx) = mpsc::unbounded_channel();
        let policy = self.config.permission_policy;
        let session_for_task = session_id.clone();

        tokio::spawn(async move {
            let result = time::timeout(
                acp_prompt_timeout(),
                collect_streaming(
                    &mut transport,
                    prompt_id,
                    &chunk_tx,
                    policy,
                    &session_for_task,
                ),
            )
            .await;
            match result {
                Ok(Err(e)) => {
                    warn!(error = %e, stderr = %stderr_ring.snapshot(), "ACP prompt stream failed");
                    let _ = chunk_tx.send(Err(e));
                }
                Err(_) => {
                    warn!(
                        stderr = %stderr_ring.snapshot(),
                        timeout_secs = acp_prompt_timeout().as_secs(),
                        "ACP prompt stream timed out"
                    );
                    let _ = chunk_tx.send(Err(RunnerError::timeout(format!(
                        "copilot-acp: prompt timed out after {}s",
                        acp_prompt_timeout().as_secs()
                    ))));
                }
                Ok(Ok(())) => {}
            }
            let _ = child.kill().await;

            // Release admission only now, with the child reaped. Dropping it
            // any earlier would let the next streamed turn spawn while this
            // subprocess is still resident, which is the overcommit the bound
            // exists to prevent.
            drop(stream_permit);
        });

        let stream = UnboundedReceiverStream::new(chunk_rx);
        Ok(Box::pin(stream))
    }

    async fn health_check(&self) -> Result<bool, RunnerError> {
        self.resolve_cli_path().map_or(Ok(false), |path| {
            tracing::info!(cli_path = %path.display(), "Copilot Headless health check: binary found");
            Ok(true)
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::ChatMessage;
    use serde_json::json;

    /// A pool of two serves two completions at once.
    ///
    /// This is the whole point of the change: the runner is a process-wide
    /// singleton and the deployed backend accepts 80 concurrent requests per
    /// instance, so a single slot made every athlete on the instance wait out
    /// every other athlete's full inference.
    #[tokio::test]
    async fn pool_of_two_hands_out_two_slots_concurrently() {
        let pool = AcpPool::new(2);

        let first = pool.checkout().await;
        let second = time::timeout(Duration::from_millis(250), pool.checkout()).await;
        assert!(
            second.is_ok(),
            "a second slot must be available while the first is held"
        );

        // Both guards are alive here — that is the concurrency being asserted.
        drop(second);
        drop(first);
    }

    /// The per-slot lock really does exclude, so the pool is not handing out
    /// shared access to one transport.
    ///
    /// The ACP wire format has no response routing: two prompts on one
    /// subprocess would interleave reads unsafely. A pool that failed to
    /// exclude would be worse than the single slot it replaced, and would look
    /// identical to a working one until it corrupted a turn under load — so
    /// the exclusion is asserted by making it fire.
    #[tokio::test]
    async fn a_single_slot_still_excludes() {
        let pool = AcpPool::new(1);

        let held = pool.checkout().await;
        let blocked = time::timeout(Duration::from_millis(250), pool.checkout()).await;
        assert!(
            blocked.is_err(),
            "one slot must serialize: a second checkout cannot succeed while the slot is held"
        );

        drop(held);
        let reacquired = time::timeout(Duration::from_millis(250), pool.checkout()).await;
        assert!(
            reacquired.is_ok(),
            "the slot must be reusable once released"
        );
    }

    /// Concurrent checkouts land on distinct slots rather than queueing on
    /// slot 0, which is what the round-robin cursor buys.
    #[tokio::test]
    async fn checkout_spreads_across_every_slot() {
        let pool = AcpPool::new(4);

        let mut held = Vec::new();
        for slot in 0..4 {
            let guard = time::timeout(Duration::from_millis(250), pool.checkout()).await;
            assert!(guard.is_ok(), "slot {slot} of 4 must be reachable");
            // `Result` iterates its `Ok`, so this keeps the guard alive.
            held.extend(guard);
        }

        assert_eq!(held.len(), 4, "every slot must have been reachable");

        // All four are held simultaneously; the fifth has nowhere to go.
        let exhausted = time::timeout(Duration::from_millis(250), pool.checkout()).await;
        assert!(
            exhausted.is_err(),
            "a 4-slot pool must not hand out a 5th slot — the pool is a bound"
        );
    }

    /// Every pool the env var can produce stays inside the memory budget.
    ///
    /// This is the assertion that makes `EMBACLE_ACP_POOL_SIZE` safe to turn
    /// up. The ceiling is derived rather than chosen precisely so that raising
    /// the pool lowers the per-slot session count instead of walking the
    /// container into an OOM — if that derivation is ever replaced by a fixed
    /// number, this fails.
    #[test]
    fn every_allowed_pool_size_fits_the_memory_budget() {
        for size in 1..=MAX_ACP_POOL_SIZE {
            let sessions = max_sessions_per_process(size) as usize;
            let peak = size * (ACP_PROCESS_BASE_MB + sessions * ACP_SESSION_COST_MB);
            assert!(
                peak <= ACP_POOL_BUDGET_MB,
                "pool of {size} peaks at {peak} MB, over the {ACP_POOL_BUDGET_MB} MB budget \
                 ({sessions} sessions/slot)"
            );
            assert!(
                sessions >= 1,
                "a slot must serve at least one session, else no completion can finish"
            );
        }
    }

    /// A bigger pool must recycle sooner, never later.
    #[test]
    fn a_larger_pool_recycles_sooner() {
        let one = max_sessions_per_process(1);
        let two = max_sessions_per_process(2);
        let four = max_sessions_per_process(4);
        assert!(
            one > two && two >= four,
            "session ceiling must fall as the pool grows: {one} / {two} / {four}"
        );
        assert_eq!(
            AcpPool::new(2).max_sessions,
            two,
            "the pool carries its ceiling"
        );
    }

    /// A notification tagged with another session is not this turn's content.
    ///
    /// This is the cross-athlete case. A pooled subprocess serves one athlete
    /// after another, and the read loop folds every `session/update` it sees
    /// into the accumulator until the prompt response arrives — so a straggler
    /// left in the pipe by an earlier session would be appended to a later
    /// athlete's reply. On the `converse()` path that reply is what the athlete
    /// reads.
    #[test]
    fn a_notification_from_another_session_is_rejected() {
        let mine = json!({"sessionId": "s-1", "update": {}});
        let theirs = json!({"sessionId": "s-2", "update": {}});

        assert!(notification_is_for_session(&mine, "s-1"));
        assert!(
            !notification_is_for_session(&theirs, "s-1"),
            "another session's notification must never reach this turn's accumulator"
        );
    }

    /// An agent that does not tag its notifications is still heard.
    ///
    /// Copilot 1.0.81 always tags, but dropping content from an agent that does
    /// not would be a worse failure than the one the filter prevents: the turn
    /// would come back empty with nothing to explain it.
    #[test]
    fn an_untagged_notification_is_accepted() {
        let untagged = json!({"update": {}});
        assert!(notification_is_for_session(&untagged, "s-1"));

        let wrong_type = json!({"sessionId": 42, "update": {}});
        assert!(
            notification_is_for_session(&wrong_type, "s-1"),
            "a non-string id is unreadable, not a mismatch"
        );
    }

    /// The streaming bound is derived from the same measurements as the pool's.
    #[test]
    fn concurrent_streams_are_derived_from_the_budget() {
        let per_stream = ACP_PROCESS_BASE_MB + ACP_SESSION_COST_MB;
        assert_eq!(max_concurrent_streams(), ACP_STREAM_BUDGET_MB / per_stream);
        assert!(
            max_concurrent_streams() >= 1,
            "a budget below one subprocess must still serve turns one at a time, not deadlock"
        );
    }

    /// The pool and the streaming paths spend from ONE container, and together
    /// they must leave the server room to run.
    ///
    /// This is the assertion that makes either budget safe to edit. They were
    /// derived separately — the pool against recycling, the streams against
    /// admission — and nothing else notices when their sum grows past what the
    /// 2Gi container can hold. Deployed dev peaked at 45% (~920 MB) with
    /// subprocesses included, so the server's own share is real and not
    /// negligible.
    #[test]
    fn both_acp_budgets_together_leave_the_server_room() {
        const CONTAINER_MB: usize = 2048;
        const SERVER_HEADROOM_MB: usize = 384;

        let pool_peak = acp_pool_size()
            * (ACP_PROCESS_BASE_MB
                + max_sessions_per_process(acp_pool_size()) as usize * ACP_SESSION_COST_MB);
        let stream_peak = max_concurrent_streams() * (ACP_PROCESS_BASE_MB + ACP_SESSION_COST_MB);

        assert!(
            pool_peak + stream_peak + SERVER_HEADROOM_MB <= CONTAINER_MB,
            "pool {pool_peak} MB + streams {stream_peak} MB leaves under {SERVER_HEADROOM_MB} MB \
             of the {CONTAINER_MB} MB container for the server itself"
        );
    }

    /// A healthy, correctly-pinned, under-ceiling process is reused.
    ///
    /// The reuse case is the one worth pinning hardest: a rule that discarded
    /// too eagerly would still be correct, just slow, and would look identical
    /// to a working pool from the outside while paying a ~3.2s cold spawn on
    /// every turn.
    #[test]
    fn a_healthy_process_within_its_ceiling_is_reused() {
        assert_eq!(discard_reason(true, true, 0, 11), None);
        assert_eq!(discard_reason(true, true, 10, 11), None);
    }

    #[test]
    fn each_discard_condition_is_recognised() {
        assert_eq!(
            discard_reason(false, true, 0, 11),
            Some(DiscardReason::Exited)
        );
        assert_eq!(
            discard_reason(true, false, 0, 11),
            Some(DiscardReason::ModelChanged)
        );
        assert_eq!(
            discard_reason(true, true, 11, 11),
            Some(DiscardReason::SessionCeiling)
        );
    }

    /// The ceiling fires AT the limit, not one session past it.
    ///
    /// Off by one here is 26 MB of resident memory per slot per turn, which is
    /// the whole reason the ceiling is derived from a budget.
    #[test]
    fn the_session_ceiling_is_inclusive() {
        assert_eq!(discard_reason(true, true, 10, 11), None);
        assert_eq!(
            discard_reason(true, true, 11, 11),
            Some(DiscardReason::SessionCeiling)
        );
        assert_eq!(
            discard_reason(true, true, 12, 11),
            Some(DiscardReason::SessionCeiling)
        );
    }

    /// A dead child is reported as dead even when it also changed model and
    /// blew its ceiling — the caller must not try to kill it again.
    #[test]
    fn an_exited_child_outranks_every_other_reason() {
        assert_eq!(
            discard_reason(false, false, 99, 11),
            Some(DiscardReason::Exited)
        );
    }

    #[test]
    fn pool_size_is_clamped_and_never_zero() {
        assert_eq!(AcpPool::new(0).slots.len(), 1, "a pool must have a slot");
        assert_eq!(AcpPool::new(3).slots.len(), 3);
        const {
            assert!(
                DEFAULT_ACP_POOL_SIZE >= 1 && DEFAULT_ACP_POOL_SIZE <= MAX_ACP_POOL_SIZE,
                "the default must sit inside the allowed range"
            );
        }
    }

    /// A real `session/new` result, trimmed to the fields the parser reads.
    ///
    /// Shape captured from CLI 1.0.80 on 2026-08-24, including the disabled
    /// entry — the wire format carries entitlement, and dropping that would
    /// make this fixture agree with a parser that ignored it.
    fn session_result() -> Value {
        json!({
            "sessionId": "bc4841f4-88f8-4b46-9320-72dc509c48e0",
            "models": {
                "currentModelId": "gpt-5.6-sol",
                "availableModels": [
                    {"modelId": "auto", "name": "Auto"},
                    {"modelId": "claude-sonnet-5", "name": "Claude Sonnet 5",
                     "_meta": {"copilotEnablement": "enabled"}},
                    {"modelId": "gpt-5.6-sol", "name": "GPT-5.6 Sol",
                     "_meta": {"copilotEnablement": "enabled"}},
                    {"modelId": "some-locked-model", "name": "Locked",
                     "_meta": {"copilotEnablement": "disabled"}},
                    {"modelId": "no-meta-model", "name": "No meta"}
                ]
            }
        })
    }

    /// The account's real list comes off the wire, not out of the catalog.
    #[test]
    fn models_are_read_from_the_session_result() {
        let ids = models_from_session(&session_result()).unwrap_or_default();
        assert!(
            !ids.is_empty(),
            "the fixture carries models; parser returned none"
        );
        assert!(
            ids.contains(&"claude-sonnet-5".to_owned()),
            "claude-sonnet-5 is enabled on this account and is the model coaching \
             turns run on; the compiled-in catalog does not list it, which is the \
             whole reason to read the wire — got {ids:?}"
        );
        assert!(ids.contains(&"gpt-5.6-sol".to_owned()));
    }

    /// A model the account may not use is not "available".
    #[test]
    fn disabled_models_are_excluded() {
        let ids = models_from_session(&session_result()).unwrap_or_default();
        assert!(
            !ids.is_empty(),
            "the fixture carries models; parser returned none"
        );
        assert!(
            !ids.contains(&"some-locked-model".to_owned()),
            "a disabled model must not be reported available — a list that \
             includes them is no better than the hardcoded one for deciding \
             whether a configured model will work"
        );
    }

    /// `auto` is a strategy, not a model.
    #[test]
    fn auto_is_not_reported_as_a_model() {
        let ids = models_from_session(&session_result()).unwrap_or_default();
        assert!(
            !ids.is_empty(),
            "the fixture carries models; parser returned none"
        );
        assert!(
            !ids.contains(&"auto".to_owned()),
            "`auto` would make any configured-model check pass by accident"
        );
    }

    /// Absent entitlement means the CLI did not say, not that it is denied.
    #[test]
    fn a_model_without_meta_is_kept() {
        let ids = models_from_session(&session_result()).unwrap_or_default();
        assert!(
            !ids.is_empty(),
            "the fixture carries models; parser returned none"
        );
        assert!(
            ids.contains(&"no-meta-model".to_owned()),
            "dropping models the CLI said nothing about would silently shrink \
             the list on any CLI that omits _meta"
        );
    }

    /// An older CLI omits the field; the caller must keep its catalog.
    #[test]
    fn a_response_without_models_yields_none() {
        let bare = json!({ "sessionId": "abc" });
        assert!(
            models_from_session(&bare).is_none(),
            "None keeps the catalog; Some(vec![]) would replace it with nothing \
             and report every model as unavailable"
        );
        let empty = json!({ "sessionId": "abc", "models": { "availableModels": [] } });
        assert!(
            models_from_session(&empty).is_none(),
            "an empty list is indistinguishable from 'not reported' and must not \
             supersede the catalog either"
        );
    }

    /// The protocol's outcome is internally tagged: a cancel travels as
    /// `{"outcome": {"outcome": "cancelled"}}`. The old hand-rolled
    /// `{"outcome": "cancelled"}` put a bare string where the agent expects a
    /// tagged object — an agent parsing strictly treats that as unanswered and
    /// keeps the permission request pending, parking the session in silence.
    /// Under the production `deny_all` policy every permission prompt took
    /// this path.
    #[test]
    fn deny_all_permission_response_is_schema_shaped() {
        let resp = build_permission_response(&json!({}), PermissionPolicy::DenyAll);
        assert_eq!(resp, json!({ "outcome": { "outcome": "cancelled" } }));
    }

    /// An approval must carry the `"outcome": "selected"` discriminator the
    /// old shape dropped, and still prefer `AllowAlways` over `AllowOnce`.
    #[test]
    fn auto_approve_response_carries_the_selected_discriminator() {
        let params = json!({
            "sessionId": "s1",
            "toolCall": { "toolCallId": "tc1" },
            "options": [
                { "optionId": "opt-once", "name": "Allow once", "kind": "allow_once" },
                { "optionId": "opt-always", "name": "Always allow", "kind": "allow_always" },
            ],
        });
        let resp = build_permission_response(&params, PermissionPolicy::AutoApprove);
        assert_eq!(
            resp,
            json!({ "outcome": { "outcome": "selected", "optionId": "opt-always" } })
        );
    }

    #[test]
    fn write_settings_model_creates_file_when_absent() {
        let dir = tempfile::tempdir().unwrap(); // Safe: test setup, tempdir creation
        let copilot = dir.path().join(".copilot");
        let path = write_settings_model(&copilot, "claude-sonnet-4.6").unwrap(); // Safe: test assertion on function under test
        let raw = fs::read_to_string(&path).unwrap(); // Safe: test reads file it just created
        let v: Value = serde_json::from_str(&raw).unwrap(); // Safe: test parses JSON it just wrote
        assert_eq!(v["model"], "claude-sonnet-4.6");
    }

    #[test]
    fn write_settings_model_preserves_other_keys() {
        // The real settings.json carries theme/effortLevel/etc; only "model" must change.
        let dir = tempfile::tempdir().unwrap(); // Safe: test setup, tempdir creation
        let copilot = dir.path().join(".copilot");
        fs::create_dir_all(&copilot).unwrap(); // Safe: test setup, dir inside fresh tempdir
        fs::write(
            copilot.join("settings.json"),
            r#"{"theme":"auto","effortLevel":"high","model":"gemini-3.5-flash"}"#,
        )
        .unwrap(); // Safe: test setup, write into fresh tempdir
        write_settings_model(&copilot, "claude-sonnet-4.6").unwrap(); // Safe: test assertion on function under test
        let raw = fs::read_to_string(copilot.join("settings.json")).unwrap(); // Safe: test reads file it just wrote
        let v: Value = serde_json::from_str(&raw).unwrap(); // Safe: test parses JSON written by function under test
        assert_eq!(v["model"], "claude-sonnet-4.6"); // overwritten
        assert_eq!(v["theme"], "auto"); // preserved
        assert_eq!(v["effortLevel"], "high"); // preserved
    }

    #[test]
    fn write_settings_model_recovers_from_corrupt_file() {
        let dir = tempfile::tempdir().unwrap(); // Safe: test setup, tempdir creation
        let copilot = dir.path().join(".copilot");
        fs::create_dir_all(&copilot).unwrap(); // Safe: test setup, dir inside fresh tempdir
        fs::write(copilot.join("settings.json"), "not valid json {{").unwrap(); // Safe: test setup, write into fresh tempdir
        write_settings_model(&copilot, "claude-sonnet-4.6").unwrap(); // Safe: test assertion on function under test
        let raw = fs::read_to_string(copilot.join("settings.json")).unwrap(); // Safe: test reads file it just wrote
        let v: Value = serde_json::from_str(&raw).unwrap(); // Safe: test parses JSON written by function under test
        assert_eq!(v["model"], "claude-sonnet-4.6");
    }

    /// Build a valid ACP permission request JSON with the given option kinds.
    ///
    /// Uses camelCase field names matching the `agent-client-protocol-schema` serde config.
    /// `PermissionOptionKind` uses `snake_case`: `allow_once`, `allow_always`,
    /// `reject_once`, `reject_always`.
    fn make_permission_params(kinds: &[&str]) -> Value {
        let options: Vec<Value> = kinds
            .iter()
            .enumerate()
            .map(|(i, kind)| {
                json!({
                    "optionId": format!("opt_{i}"),
                    "name": format!("Option {i}"),
                    "kind": kind
                })
            })
            .collect();
        json!({
            "sessionId": "test-session",
            "toolCall": {
                "toolCallId": "tc_1"
            },
            "options": options
        })
    }

    #[test]
    fn permission_only_reject_options_cancels() {
        let params = make_permission_params(&["reject_once", "reject_always"]);
        let result = build_permission_response(&params, PermissionPolicy::AutoApprove);
        assert_eq!(result["outcome"]["outcome"], "cancelled");
    }

    #[test]
    fn permission_prefers_allow_always_over_allow_once() {
        let params = make_permission_params(&["allow_once", "allow_always", "reject_once"]);
        let result = build_permission_response(&params, PermissionPolicy::AutoApprove);
        // AllowAlways is at index 1 → opt_1
        let selected_id = result["outcome"]["optionId"].as_str().unwrap(); // Safe: test assertion
        assert_eq!(selected_id, "opt_1");
    }

    #[test]
    fn permission_selects_allow_once_when_no_allow_always() {
        let params = make_permission_params(&["allow_once", "reject_once"]);
        let result = build_permission_response(&params, PermissionPolicy::AutoApprove);
        let selected_id = result["outcome"]["optionId"].as_str().unwrap(); // Safe: test assertion
        assert_eq!(selected_id, "opt_0");
    }

    #[test]
    fn permission_empty_options_cancels() {
        let params = json!({
            "sessionId": "test-session",
            "toolCall": {
                "toolCallId": "tc_1"
            },
            "options": []
        });
        let result = build_permission_response(&params, PermissionPolicy::AutoApprove);
        assert_eq!(result["outcome"]["outcome"], "cancelled");
    }

    #[test]
    fn permission_deny_all_policy_always_cancels() {
        let params = make_permission_params(&["allow_once", "allow_always"]);
        let result = build_permission_response(&params, PermissionPolicy::DenyAll);
        assert_eq!(result["outcome"]["outcome"], "cancelled");
    }

    /// Create a test runner with configurable `max_history_turns`.
    fn test_runner(max_history_turns: usize) -> CopilotHeadlessRunner {
        CopilotHeadlessRunner {
            config: CopilotHeadlessConfig {
                max_history_turns,
                ..CopilotHeadlessConfig::default()
            },
            available_models: vec![],
            observed_models: OnceLock::new(),
            pool: AcpPool::new(acp_pool_size()),
            stream_permits: Arc::new(Semaphore::new(max_concurrent_streams())),
        }
    }

    /// Create a test runner with system prompt injection disabled.

    #[test]
    fn build_prompt_blocks_text_only_no_system() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![ChatMessage::user("Hello")]);
        let blocks = runner.build_prompt_blocks(&request);
        assert_eq!(blocks.len(), 1);
        assert_eq!(blocks[0]["type"], "text");
        assert_eq!(blocks[0]["text"], "Hello");
    }

    #[test]
    fn build_prompt_blocks_injects_system_prompt_as_plain_text() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::system("You are a fitness assistant"),
            ChatMessage::user("Hello"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        assert_eq!(blocks.len(), 1);
        assert_eq!(blocks[0]["type"], "text");
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion
                                                        // System prompt injected as plain text — no XML tags
        assert!(text.contains("You are a fitness assistant"));
        assert!(!text.contains("<system-instructions>"));
        assert!(text.contains("Hello"));
    }

    #[test]
    fn build_prompt_blocks_with_images() {
        use crate::types::ImagePart;

        let runner = test_runner(20);
        let img = ImagePart::new("aGVsbG8=", "image/png").unwrap(); // Safe: test assertion
        let request = ChatRequest::new(vec![ChatMessage::user_with_images(
            "Describe this image",
            vec![img],
        )]);
        let blocks = runner.build_prompt_blocks(&request);
        assert_eq!(blocks.len(), 2);
        assert_eq!(blocks[0]["type"], "text");
        assert!(blocks[0]["text"]
            .as_str()
            .unwrap() // Safe: test assertion
            .contains("Describe this image"));
        assert_eq!(blocks[1]["type"], "image");
        assert_eq!(blocks[1]["data"], "aGVsbG8=");
        assert_eq!(blocks[1]["mimeType"], "image/png");
    }

    #[test]
    fn build_prompt_blocks_uses_last_user_message() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::user("first"),
            ChatMessage::assistant("response"),
            ChatMessage::user("second"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion
        assert!(text.contains("second"));
        // The last user message should NOT be in the history section
        assert!(!text.ends_with("second\n</conversation-history>"));
    }

    #[test]
    fn build_prompt_blocks_includes_conversation_history() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::system("You are helpful"),
            ChatMessage::user("What is my pace?"),
            ChatMessage::assistant("Your average pace is 5:30/km"),
            ChatMessage::user("And my heart rate?"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        // System prompt injected as plain text
        assert!(text.contains("You are helpful"));
        assert!(!text.contains("<system-instructions>"));

        // Conversation history block present with prior turns
        assert!(text.contains("<conversation-history>"));
        assert!(text.contains("User: What is my pace?"));
        assert!(text.contains("Assistant: Your average pace is 5:30/km"));
        assert!(text.contains("</conversation-history>"));

        // Current user message at the end (outside history block)
        assert!(text.contains("And my heart rate?"));
        // Current message should NOT be inside the history block
        assert!(!text.contains("User: And my heart rate?"));
    }

    #[test]
    fn build_prompt_blocks_no_history_for_single_turn() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::system("Be helpful"),
            ChatMessage::user("Hello"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion
                                                        // No history block when there's only one user message
        assert!(!text.contains("<conversation-history>"));
        assert!(text.contains("Hello"));
    }

    #[test]
    fn build_prompt_blocks_truncates_history_to_max_turns() {
        let runner = test_runner(2); // Only keep 2 most recent history messages
        let request = ChatRequest::new(vec![
            ChatMessage::user("msg1"),
            ChatMessage::assistant("reply1"),
            ChatMessage::user("msg2"),
            ChatMessage::assistant("reply2"),
            ChatMessage::user("msg3"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        // Only the 2 most recent history messages should be included
        assert!(!text.contains("User: msg1"));
        assert!(!text.contains("Assistant: reply1"));
        assert!(text.contains("User: msg2"));
        assert!(text.contains("Assistant: reply2"));
        assert!(text.contains("msg3")); // Current message
    }

    #[test]
    fn build_prompt_blocks_zero_max_turns_disables_history() {
        let runner = test_runner(0);
        let request = ChatRequest::new(vec![
            ChatMessage::user("first"),
            ChatMessage::assistant("response"),
            ChatMessage::user("second"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion
        assert!(!text.contains("<conversation-history>"));
        assert!(text.contains("second"));
    }

    #[test]
    fn build_prompt_blocks_max_turns_one_keeps_last_history_message() {
        let runner = test_runner(1);
        let request = ChatRequest::new(vec![
            ChatMessage::user("msg1"),
            ChatMessage::assistant("reply1"),
            ChatMessage::user("msg2"),
            ChatMessage::assistant("reply2"),
            ChatMessage::user("current"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        // Only the single most recent history message (reply2)
        assert!(!text.contains("User: msg1"));
        assert!(!text.contains("reply1"));
        assert!(!text.contains("User: msg2"));
        assert!(text.contains("Assistant: reply2"));
        assert!(text.contains("current"));
    }

    #[test]
    fn build_prompt_blocks_tool_messages_included_in_history() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::user("Check my activities"),
            ChatMessage::tool("get_activities", "call_1", "{\"activities\": []}"),
            ChatMessage::assistant("No activities found"),
            ChatMessage::user("Try again"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        assert!(text.contains("<conversation-history>"));
        assert!(text.contains("User: Check my activities"));
        assert!(text.contains("Tool: "));
        assert!(text.contains("Assistant: No activities found"));
        assert!(text.contains("Try again"));
    }

    #[test]
    fn build_prompt_blocks_empty_messages() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![]);
        let blocks = runner.build_prompt_blocks(&request);
        assert_eq!(blocks.len(), 1);
        // Empty prompt — no crash
        assert_eq!(blocks[0]["text"], "");
    }

    #[test]
    fn build_prompt_blocks_only_system_message() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![ChatMessage::system("Be helpful")]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion
                                                        // System prompt as plain text, no XML tags, no history
        assert!(text.contains("Be helpful"));
        assert!(!text.contains("<system-instructions>"));
        assert!(!text.contains("<conversation-history>"));
    }

    #[test]
    fn build_prompt_blocks_long_conversation_keeps_most_recent() {
        let runner = test_runner(4);
        let mut messages = vec![ChatMessage::system("system")];
        for i in 1..=10 {
            messages.push(ChatMessage::user(format!("user_{i}")));
            messages.push(ChatMessage::assistant(format!("reply_{i}")));
        }
        messages.push(ChatMessage::user("current"));
        let request = ChatRequest::new(messages);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        // Only last 4 history messages kept (user_9, reply_9, user_10, reply_10)
        assert!(!text.contains("user_8"));
        assert!(!text.contains("reply_8"));
        assert!(text.contains("User: user_9"));
        assert!(text.contains("Assistant: reply_9"));
        assert!(text.contains("User: user_10"));
        assert!(text.contains("Assistant: reply_10"));
        assert!(text.contains("current"));
    }

    #[test]
    fn build_prompt_blocks_preserves_section_ordering() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::system("sys prompt"),
            ChatMessage::user("q1"),
            ChatMessage::assistant("a1"),
            ChatMessage::user("q2"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        // Verify ordering: system prompt < conversation-history < current message
        let sys_pos = text.find("sys prompt").unwrap(); // Safe: test assertion
        let hist_pos = text.find("<conversation-history>").unwrap(); // Safe: test assertion
        let current_pos = text.find("q2").unwrap(); // Safe: test assertion
        assert!(sys_pos < hist_pos, "system must come before history");
        assert!(
            hist_pos < current_pos,
            "history must come before current message"
        );
    }

    #[test]
    fn build_prompt_blocks_history_exact_at_max_turns() {
        let runner = test_runner(2);
        // Exactly 2 history messages — should include all, no truncation
        let request = ChatRequest::new(vec![
            ChatMessage::user("q1"),
            ChatMessage::assistant("a1"),
            ChatMessage::user("current"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        assert!(text.contains("User: q1"));
        assert!(text.contains("Assistant: a1"));
        assert!(text.contains("current"));
    }

    #[test]
    fn build_prompt_blocks_multiple_system_messages_uses_first() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::system("first system"),
            ChatMessage::system("second system"),
            ChatMessage::user("hello"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        // extract_system_prompt returns the first system message
        assert!(text.contains("first system"));
    }

    #[test]
    fn capabilities_omit_sdk_tool_calling_by_default() {
        let runner = CopilotHeadlessRunner::with_config(CopilotHeadlessConfig::default());
        let caps = runner.capabilities();
        assert!(caps.supports_vision());
        assert!(caps.supports_streaming());
        assert!(
            !caps.supports_sdk_tool_calling(),
            "default config falls through to text-based tool calling"
        );
    }

    #[test]
    fn capabilities_advertise_sdk_tool_calling_when_mcp_enabled() {
        let runner = CopilotHeadlessRunner::with_config(CopilotHeadlessConfig {
            mcp_tool_calling: true,
            ..CopilotHeadlessConfig::default()
        });
        assert!(
            runner.capabilities().supports_sdk_tool_calling(),
            "mcp_tool_calling=true routes tool turns through the ACP converse() loop"
        );
    }

    #[test]
    fn mcp_servers_to_acp_json_http_matches_wire_format() {
        let servers = vec![McpServerConfig {
            name: "dravr".to_owned(),
            transport: McpTransport::Http {
                url: "http://localhost:8081/mcp".to_owned(),
                headers: vec![McpHeader {
                    name: "Authorization".to_owned(),
                    value: "Bearer tok".to_owned(),
                }],
            },
        }];
        let json = mcp_servers_to_acp_json(&servers);
        assert_eq!(json.len(), 1);
        assert_eq!(json[0]["type"], "http");
        assert_eq!(json[0]["name"], "dravr");
        assert_eq!(json[0]["url"], "http://localhost:8081/mcp");
        assert_eq!(json[0]["headers"][0]["name"], "Authorization");
        assert_eq!(json[0]["headers"][0]["value"], "Bearer tok");
    }

    #[test]
    fn mcp_servers_to_acp_json_empty_is_empty_array() {
        assert!(mcp_servers_to_acp_json(&[]).is_empty());
    }

    #[test]
    fn build_prompt_params_without_max_tokens() {
        let blocks = vec![json!({"type": "text", "text": "hello"})];
        let params = build_prompt_params("sess-1", &blocks, None);
        assert_eq!(params["sessionId"], "sess-1");
        assert!(params["prompt"].is_array());
        assert!(params.get("maxTokens").is_none());
    }

    #[test]
    fn build_prompt_params_with_max_tokens() {
        let blocks = vec![json!({"type": "text", "text": "hello"})];
        let params = build_prompt_params("sess-2", &blocks, Some(1024));
        assert_eq!(params["sessionId"], "sess-2");
        assert_eq!(params["maxTokens"], 1024);
    }

    #[test]
    fn build_prompt_params_preserves_prompt_blocks() {
        let blocks = vec![
            json!({"type": "text", "text": "hello"}),
            json!({"type": "image", "data": "abc", "mimeType": "image/png"}),
        ];
        let params = build_prompt_params("s1", &blocks, Some(512));
        let prompt = params["prompt"].as_array().unwrap(); // Safe: test assertion
        assert_eq!(prompt.len(), 2);
        assert_eq!(prompt[0]["type"], "text");
        assert_eq!(prompt[1]["type"], "image");
    }

    #[test]
    fn default_mode_multi_turn_system_as_plain_text() {
        let runner = test_runner(20);
        let request = ChatRequest::new(vec![
            ChatMessage::system("Return JSON only"),
            ChatMessage::user("First question"),
            ChatMessage::assistant("{\"answer\": 1}"),
            ChatMessage::user("Second question"),
        ]);
        let blocks = runner.build_prompt_blocks(&request);
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion

        // System prompt as plain text — no XML tags
        assert!(text.contains("Return JSON only"));
        assert!(!text.contains("<system-instructions>"));

        // Conversation history and current message also present
        assert!(text.contains("<conversation-history>"));
        assert!(text.contains("User: First question"));
        assert!(text.contains("Second question"));
    }

    #[test]
    fn default_mode_with_images_includes_system() {
        use crate::types::ImagePart;

        let runner = test_runner(20);
        let img = ImagePart::new("aGVsbG8=", "image/png").unwrap(); // Safe: test assertion
        let request = ChatRequest::new(vec![
            ChatMessage::system("Analyze images precisely"),
            ChatMessage::user_with_images("What is this?", vec![img]),
        ]);
        let blocks = runner.build_prompt_blocks(&request);

        // System prompt present as plain text
        let text = blocks[0]["text"].as_str().unwrap(); // Safe: test assertion
        assert!(text.contains("Analyze images precisely"));
        assert!(!text.contains("<system-instructions>"));
        assert!(text.contains("What is this?"));

        // Image block still present
        assert_eq!(blocks.len(), 2);
        assert_eq!(blocks[1]["type"], "image");
    }

    #[test]
    fn resolve_model_uses_explicit_model() {
        let runner = test_runner(20);
        assert_eq!(runner.resolve_model(Some("gpt-4.1")), "gpt-4.1");
    }

    #[test]
    fn resolve_model_maps_copilot_headless_to_default() {
        let runner = test_runner(20);
        let result = runner.resolve_model(Some("copilot_headless"));
        assert_eq!(result, runner.config.model);
    }

    #[test]
    fn resolve_model_maps_copilot_to_default() {
        let runner = test_runner(20);
        let result = runner.resolve_model(Some("copilot"));
        assert_eq!(result, runner.config.model);
    }

    #[test]
    fn resolve_model_uses_default_when_none() {
        let runner = test_runner(20);
        let result = runner.resolve_model(None);
        assert_eq!(result, runner.config.model);
    }

    /// Build a `session/update` notification of an `AgentMessageChunk`
    /// carrying the given text. Mirrors the ACP wire format consumed by
    /// `process_notification`.
    fn make_text_chunk_notification(text: &str) -> Value {
        json!({
            "sessionId": "test-session",
            "update": {
                "sessionUpdate": "agent_message_chunk",
                "content": {
                    "type": "text",
                    "text": text
                }
            }
        })
    }

    /// Build a `session/update` notification of a `ToolCall` with the
    /// given id, title, and status.
    fn make_tool_call_notification(id: &str, title: &str, status: &str) -> Value {
        json!({
            "sessionId": "test-session",
            "update": {
                "sessionUpdate": "tool_call",
                "toolCallId": id,
                "title": title,
                "status": status,
                "kind": "other",
                "content": []
            }
        })
    }

    /// Helper: pull the next event from `rx` and assert it is the expected
    /// `TextDelta`. Uses the existing test-assertion safety convention so
    /// the architectural validator (which counts bare panics / expects in
    /// src/) stays green.
    fn expect_text_delta(
        rx: &mut mpsc::UnboundedReceiver<Result<HeadlessStreamEvent, RunnerError>>,
        expected: &str,
    ) {
        let event = rx.try_recv().unwrap().unwrap(); // Safe: test assertion
        let HeadlessStreamEvent::TextDelta(s) = event else {
            unreachable!("expected TextDelta event variant"); // Safe: test assertion
        };
        assert_eq!(s, expected);
    }

    fn expect_tool_call(
        rx: &mut mpsc::UnboundedReceiver<Result<HeadlessStreamEvent, RunnerError>>,
    ) -> ObservedToolCall {
        let event = rx.try_recv().unwrap().unwrap(); // Safe: test assertion
        let HeadlessStreamEvent::ToolCall(tc) = event else {
            unreachable!("expected ToolCall event variant"); // Safe: test assertion
        };
        tc
    }

    #[test]
    fn streaming_notification_forwards_text_delta_and_accumulates() {
        let mut acc = TurnAccumulator::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        process_notification_streaming(&make_text_chunk_notification("Hello, "), &mut acc, &tx);
        process_notification_streaming(&make_text_chunk_notification("world!"), &mut acc, &tx);

        // Accumulator behaves like the non-streaming path
        assert_eq!(acc.content, "Hello, world!");
        assert_eq!(acc.tool_calls.len(), 0);

        // Both chunks are emitted on the channel in order
        expect_text_delta(&mut rx, "Hello, ");
        expect_text_delta(&mut rx, "world!");
        assert!(rx.try_recv().is_err(), "expected no more events");
    }

    #[test]
    fn streaming_notification_forwards_tool_call() {
        let mut acc = TurnAccumulator::new();
        let (tx, mut rx) = mpsc::unbounded_channel();
        process_notification_streaming(
            &make_tool_call_notification("tc_1", "Reading file", "in_progress"),
            &mut acc,
            &tx,
        );

        assert_eq!(acc.tool_calls.len(), 1);
        assert_eq!(acc.tool_calls[0].id, "tc_1");
        assert_eq!(acc.tool_calls[0].title, "Reading file");

        let tc = expect_tool_call(&mut rx);
        assert_eq!(tc.id, "tc_1");
        assert_eq!(tc.title, "Reading file");
    }

    #[test]
    fn process_notification_no_channel_matches_streaming_state() {
        // The non-streaming entry point must produce the *same* accumulator
        // state as the streaming one — we just lose the per-event channel.
        let mut acc_plain = TurnAccumulator::new();
        process_notification(&make_text_chunk_notification("Hi"), &mut acc_plain);
        process_notification(
            &make_tool_call_notification("tc_a", "Tool", "completed"),
            &mut acc_plain,
        );

        let mut acc_stream = TurnAccumulator::new();
        let (tx, _rx) = mpsc::unbounded_channel();
        process_notification_streaming(&make_text_chunk_notification("Hi"), &mut acc_stream, &tx);
        process_notification_streaming(
            &make_tool_call_notification("tc_a", "Tool", "completed"),
            &mut acc_stream,
            &tx,
        );

        assert_eq!(acc_plain.content, acc_stream.content);
        assert_eq!(acc_plain.tool_calls.len(), acc_stream.tool_calls.len());
        assert_eq!(acc_plain.tool_calls[0].id, acc_stream.tool_calls[0].id);
        assert_eq!(
            acc_plain.tool_calls[0].title,
            acc_stream.tool_calls[0].title
        );
    }

    /// A real Copilot ACP usage object, captured 2026-08-27 from claude-opus-4.8.
    ///
    /// Verbatim from the wire — see `examples/acp_usage_probe.rs`. Pinned as a
    /// fixture because the question this answers ("does the agent populate the
    /// optional cache fields?") is empirical, and the previous answer was a guess
    /// that hardened into a registered limitation.
    const REAL_WARM_TURN_USAGE: &str = r#"{
        "result": {
            "usage": {
                "cachedReadTokens": 15320,
                "cachedWriteTokens": 12540,
                "inputTokens": 27862,
                "outputTokens": 4,
                "thoughtTokens": 0,
                "totalTokens": 27866
            }
        }
    }"#;

    #[test]
    fn extract_usage_keeps_the_cache_counts_copilot_actually_sends() {
        let Ok(value) = serde_json::from_str::<Value>(REAL_WARM_TURN_USAGE) else {
            unreachable!("the checked-in fixture must parse")
        };
        let Some(usage) = extract_usage(&value) else {
            unreachable!("usage is present in the fixture")
        };

        assert_eq!(usage.prompt_tokens, 27862);
        assert_eq!(usage.completion_tokens, 4);
        assert_eq!(usage.total_tokens, 27866);

        // The point of the test: these were on the wire and thrown away.
        assert_eq!(
            usage.cached_read_tokens,
            Some(15320),
            "cachedReadTokens is reported on every turn and must survive parsing"
        );
        assert_eq!(usage.cached_write_tokens, Some(12540));

        // Zero REPORTED, not absent. The distinction is the whole point of the
        // Option: `None` would mean the agent said nothing about reasoning.
        assert_eq!(usage.reasoning_tokens, Some(0));
    }

    #[test]
    fn extract_usage_reports_absent_cache_fields_as_none_not_zero() {
        // An agent implementing only the stable subset of the usage capability.
        let value = json!({
            "result": { "usage": { "inputTokens": 100, "outputTokens": 10, "totalTokens": 110 } }
        });
        let Some(usage) = extract_usage(&value) else {
            unreachable!("usage is present in the fixture")
        };

        assert_eq!(usage.prompt_tokens, 100);
        assert_eq!(
            usage.cached_read_tokens, None,
            "an agent that reports nothing must not be recorded as measuring zero — \
             collapsing those is how a hardcoded 0 passed for a measurement"
        );
        assert_eq!(usage.cached_write_tokens, None);
        assert_eq!(usage.reasoning_tokens, None);
    }
}