// 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::fmt::Write as _;
use std::fs;
use std::path::{self, Path, PathBuf};
use std::process::{self, 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 agent_client_protocol_schema::v1 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_common::{
render_turn, runtime_env_keys, system_prompt, HeadlessEventStream, HeadlessStreamEvent,
HeadlessToolResponse, HeadlessTurnProvider, ObservedToolCall, PermissionPolicy,
ALLOW_ALL_MCP_SERVER_INSTRUCTIONS_FLAG,
};
use crate::copilot_headless_config::CopilotHeadlessConfig;
use crate::copilot_models::catalog_ids;
use crate::sandbox::{apply_sandbox, SandboxPolicy};
use crate::types::{
ChatRequest, ChatResponse, ChatStream, LlmCapabilities, LlmProvider, McpHeader,
McpServerConfig, McpTransport, 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.
///
/// Both spawn paths take it: [`AcpProcess::spawn_and_initialize`] for the
/// pool and [`setup_session`] for streamed turns. Pooled spawns are rare (only
/// on discard: exit, model change, session ceiling), and a turn that reuses a
/// warm subprocess never touches this. A streamed turn spawns every time, so
/// concurrent streams queue here for the length of one handshake each — the
/// price of each one running on the model it asked for.
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
}
/// The directory every `copilot --acp` subprocess runs in and the `cwd` of
/// every session it serves.
///
/// Copilot treats the session `cwd` as the project it is working on: it reads
/// `.mcp.json` / `.github/mcp.json` from it and spawns every MCP server listed
/// there for each session, and folds `AGENTS.md` and
/// `.github/copilot-instructions.md` into the model's instructions. A coaching
/// turn has no project, so nothing in whatever directory the host happens to
/// run from — a git checkout, on a developer's machine — may reach the model's
/// tool surface. Handing copilot the host's cwd does exactly that: every
/// session spawns the checkout's stdio MCP servers, and a warm subprocess holds
/// them until it is recycled — measured at two 11-hour subprocesses holding 140
/// orphaned `node` processes and 20 GB of swap.
///
/// The host's choice wins when it made one, made absolute against the host's
/// cwd because the ACP schema requires an absolute session path. Otherwise the
/// directory is a scratch directory under the system temp dir, keyed by process
/// id so two servers on one machine cannot see what the other's subprocess
/// wrote there. The directory is created at spawn, not here: the runner
/// constructors are infallible, and the spawn path is where a filesystem error
/// already propagates. `path::absolute` fails only for an empty path or an
/// unreadable cwd; the configured path is then kept as given and the spawn's
/// `create_dir_all` reports the real error.
fn session_cwd(configured: Option<&Path>) -> PathBuf {
configured.map_or_else(
|| env::temp_dir().join(format!("embacle-copilot-{}", process::id())),
|dir| path::absolute(dir).unwrap_or_else(|_| dir.to_path_buf()),
)
}
/// Build the JSON params for an ACP `session/new` request.
///
/// `cwd` is the session working directory from [`session_cwd`]; it is sent as
/// a string because the ACP schema requires an absolute path there and a
/// non-UTF-8 path has no JSON form. `systemPrompt` is present only when the
/// caller supplied one, so an absent prompt is an absent key rather than an
/// empty string copilot would treat as an instruction.
fn build_session_params(
model: &str,
cwd: &Path,
system_prompt: Option<&str>,
mcp_servers: &[McpServerConfig],
) -> Value {
let mut params = json!({
"model": model,
"cwd": cwd.to_string_lossy(),
"mcpServers": mcp_servers_to_acp_json(mcp_servers),
});
if let Some(sys) = system_prompt {
params["systemPrompt"] = Value::String(sys.to_owned());
}
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)
}
/// Warn when the model the caller asked for is not one the account may use.
///
/// `copilot --acp` routes by `~/.copilot/settings.json` alone. Handed a model
/// id it does not recognise or may not use, it does not refuse — it silently
/// serves the turn from its own default and reports the requested id back as a
/// cosmetic label. That is not hypothetical: production ran on GPT and Gemini
/// for roughly two months while every log line said `claude-sonnet-4.6`.
///
/// `session/new` carries the enabled list, so the mismatch is knowable at
/// session setup, before a single token is spent. Absent list (`None`, an older
/// CLI that does not report) means no information, which is not the same as a
/// mismatch — say nothing rather than cry wolf on every turn.
///
/// This warns rather than fails: a model retired upstream would otherwise take
/// chat down entirely, and serving a turn on the wrong model beats serving no
/// turn. The warning is what makes the substitution visible.
fn model_is_unavailable(model: &str, observed: Option<&Vec<String>>) -> bool {
observed.is_some_and(|available| !available.iter().any(|id| id == model))
}
/// Emit the warning [`model_is_unavailable`] decides on.
fn warn_on_unavailable_model(model: &str, observed: Option<&Vec<String>>) {
if !model_is_unavailable(model, observed) {
return;
}
// `observed` is Some here: an absent list can never be unavailable.
let available = observed.map_or_else(String::new, |ids| ids.join(","));
warn!(
requested_model = %model,
available_models = %available,
"ACP: requested model is not enabled on this account; copilot will serve this turn \
from its own default and report the requested id as a label"
);
}
/// 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.
///
/// settings.json is process-global and the child reads it AFTER exec, so this
/// write is safe only under [`MODEL_ROUTING_GATE`], held until the child's
/// `initialize` returns. Every caller of [`spawn_copilot`] takes 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)
}
/// Build the `copilot --acp` command line without spawning it.
///
/// `--allow-all-mcp-server-instructions` is passed unconditionally: on ACP the
/// MCP servers arrive per session in `session/new`, after the warm subprocess is
/// already running, so whether a turn will carry a server is unknown at spawn.
/// Without the flag a caller's server — never on the CLI's allowlist — has its
/// `initialize` instructions dropped instead of folded into the system prompt.
/// The scratch [`session_cwd`] a host gets when it configures none keeps a
/// project's `.mcp.json` servers out of the session, so the fold reaches the
/// caller's servers and the CLI's own builtins, whose instructions the
/// allowlist already folds. The cwd does not scope servers the CLI reads from
/// the host user's own Copilot config under `$HOME`; the flag folds those too.
///
/// The child starts from a cleared environment ([`acp_sandbox`]), as every
/// `-p` runner's does: the model it runs can be steered by its prompt, and
/// its tools read their own environment.
fn acp_command(
cli_path: &Path,
github_token: Option<&str>,
model: &str,
session_cwd: &Path,
) -> Command {
let mut cmd = Command::new(cli_path);
cmd.arg("--acp");
if !model.is_empty() {
cmd.arg("--model").arg(model);
}
cmd.arg(ALLOW_ALL_MCP_SERVER_INSTRUCTIONS_FLAG);
cmd.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped());
apply_sandbox(&mut cmd, &acp_sandbox(session_cwd, github_token));
cmd
}
/// The environment and directory the `copilot --acp` child runs in: the host
/// variables [`runtime_env_keys`] allows and nothing else, in `session_cwd`,
/// with `github_token` (when the host configured one) as
/// `COPILOT_GITHUB_TOKEN`, ahead of whatever the host's own variables say.
fn acp_sandbox(session_cwd: &Path, github_token: Option<&str>) -> SandboxPolicy {
let mut policy =
SandboxPolicy::new(session_cwd.to_path_buf()).with_env_keys(runtime_env_keys());
if let Some(token) = github_token {
policy
.env_overrides
.push((GITHUB_TOKEN_ENV.to_owned(), token.to_owned()));
}
policy
}
/// The variable the Copilot runtime reads its GitHub token from first.
const GITHUB_TOKEN_ENV: &str = "COPILOT_GITHUB_TOKEN";
/// 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.
///
/// The subprocess runs in `session_cwd` (see [`session_cwd`]), created here if
/// it does not exist; unset, that is a scratch directory, not the host's cwd.
fn spawn_copilot(
cli_path: &Path,
github_token: Option<&str>,
model: &str,
session_cwd: &Path,
) -> 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"
),
}
fs::create_dir_all(session_cwd).map_err(|e| {
RunnerError::internal(format!("create session cwd {}: {e}", session_cwd.display()))
})?;
let mut cmd = acp_command(cli_path, github_token, model, session_cwd);
info!(
cli_path = %cli_path.display(),
model = %model,
cwd = %session_cwd.display(),
"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: &Path,
github_token: Option<&str>,
model: &str,
session_cwd: &Path,
system_prompt: Option<&str>,
mcp_servers: &[McpServerConfig],
) -> Result<(AcpTransport, Child, StderrRing, String, Option<Vec<String>>), RunnerError> {
// Held until the handshake completes: see MODEL_ROUTING_GATE. A streamed
// turn spawns a child every time, so skipping the gate here would race
// every concurrent stream, and every pooled spawn, for settings.json.
let routing_gate = MODEL_ROUTING_GATE.lock().await;
let mut child = spawn_copilot(cli_path, github_token, model, session_cwd)?;
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");
// The child has read its configuration, so a later writer can no longer
// change the model under it. An error or timeout before this point
// drops the guard with this future.
drop(routing_gate);
// Create session with model, working directory, optional system
// prompt, and any MCP servers the model should call tools from.
let session_params = build_session_params(model, session_cwd, system_prompt, mcp_servers);
info!(
model = %model,
cwd = %session_cwd.display(),
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"
);
warn_on_unavailable_model(model, observed.as_ref());
// 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: &Path,
github_token: Option<&str>,
model: &str,
session_cwd: &Path,
) -> 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, session_cwd)?;
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,
session_cwd: &Path,
system_prompt: Option<&str>,
mcp_servers: &[McpServerConfig],
) -> Result<(String, Option<Vec<String>>), RunnerError> {
let outcome = time::timeout(acp_session_timeout(), async {
let session_params =
build_session_params(model, session_cwd, system_prompt, mcp_servers);
info!(
model = %model,
cwd = %session_cwd.display(),
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"
);
warn_on_unavailable_model(model, observed.as_ref());
// 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>,
/// What the CLI itself wrote into the reply stream; never part of
/// `content`.
notices: Vec<CliNotice>,
/// Where the turn stands in Copilot's Autopilot loop.
run: AutopilotRun,
}
/// Where a turn stands in Copilot's Autopilot loop, read from the ACP stream.
///
/// Autopilot keeps one ACP turn open across several model requests. A
/// response that calls a tool is followed by another request carrying the
/// tool's result; a response that calls none ends the run, and if the model
/// never called `task_complete` the runtime starts a *continuation*: it sends
/// the model an internal control message ("You have not yet marked the task
/// as complete…") inside the same turn. ACP marks none of this. The wire
/// carries no frame for the control message and no run boundary; every
/// `agent_message_chunk` of every request lands in one stream.
///
/// What the wire does carry is a `usage_update` at the start of every model
/// request: the first one of the turn, one after each tool batch, and one when
/// a continuation starts (recorded on CLI 1.0.90: a reply with no tool call,
/// then `usage_update`, then the continuation's `task_complete`). So a request
/// that begins after a response with text and no tool call is a continuation.
///
/// The model's reply to the athlete is the text of the run that answered. A
/// continuation is the model reacting to the runtime's control message, which
/// tells it outright not to acknowledge that message in visible text. When it
/// does anyway, it narrates its own turn in the third person — "the 404
/// persists, I already told the athlete" — and that text, appended with no
/// separator, reached the athlete as part of the coach's reply. So once a
/// reply exists, text written in a continuation is withheld. A run that ended
/// with no reply leaves its continuation's text in place: that text is then
/// the only answer there is.
#[derive(Debug, Default)]
struct AutopilotRun {
/// The current model request has produced reply text.
request_wrote_text: bool,
/// The current model request has called a tool.
request_called_tool: bool,
/// A continuation started after the model had already replied.
continuing_after_reply: bool,
/// Bytes of continuation text withheld from the reply.
withheld_bytes: usize,
}
impl TurnAccumulator {
fn new() -> Self {
Self {
content: String::new(),
tool_calls: Vec::new(),
notices: Vec::new(),
run: AutopilotRun::default(),
}
}
/// A model request has started (`usage_update`).
///
/// When the request before it wrote text and called no tool, the run
/// ended there and this request is an Autopilot continuation.
fn begin_model_request(&mut self) {
let run = &mut self.run;
let previous_ended_run = run.request_wrote_text && !run.request_called_tool;
if previous_ended_run && !run.continuing_after_reply && !self.content.trim().is_empty() {
debug!(
reply_len = self.content.len(),
"ACP: Autopilot continued after the model replied; its text is withheld"
);
run.continuing_after_reply = true;
}
run.request_wrote_text = false;
run.request_called_tool = false;
}
/// The current model request called a tool.
fn record_tool_call(&mut self) {
self.run.request_called_tool = true;
}
/// Take a chunk of model text, returning it when it belongs in the reply.
///
/// Text written in a continuation after the model replied is withheld
/// (see [`AutopilotRun`]) and returns `None`, so no caller streams it.
fn accept_text(&mut self, text: &str) -> Option<String> {
if self.run.continuing_after_reply {
if self.run.withheld_bytes == 0 {
warn!(
"ACP: the model wrote text in an Autopilot continuation after its reply; \
withheld from the reply"
);
}
self.run.withheld_bytes += text.len();
return None;
}
if !text.trim().is_empty() {
self.run.request_wrote_text = true;
}
self.content.push_str(text);
Some(text.to_owned())
}
/// The turn as an error, when the CLI answered it and the model did not.
///
/// A turn with model text keeps it: a notice beside a real answer is the
/// CLI commenting on the turn, not replacing it.
fn failure(&self) -> Option<RunnerError> {
if !self.content.trim().is_empty() {
return None;
}
let errors: Vec<&str> = self
.notices
.iter()
.filter(|n| n.level == NoticeLevel::Error)
.map(|n| n.message.as_str())
.collect();
if errors.is_empty() {
return None;
}
let mut message = format!(
"copilot-acp: the CLI answered the turn itself, with no model text: {}",
errors.join(" / ")
);
for other in self
.notices
.iter()
.filter(|n| n.level != NoticeLevel::Error)
{
let _ = write!(message, " ({}: {})", other.level.label(), other.message);
}
Some(notice_error(&errors.join(" "), message))
}
}
/// How severe a [`CliNotice`] is, from the prefix the CLI gave it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum NoticeLevel {
Error,
Warning,
Info,
}
impl NoticeLevel {
const fn label(self) -> &'static str {
match self {
Self::Error => "error",
Self::Warning => "warning",
Self::Info => "info",
}
}
}
/// A notice the Copilot CLI wrote into the reply stream.
///
/// The CLI's ACP adapter renders the runtime's `session.error`,
/// `session.warning` and `session.info` events as ordinary
/// `agent_message_chunk` text — `Error: …`, `Warning: …`, `Info: …` — and
/// still ends the turn with `end_turn` (read from its adapter, and observed on
/// the wire with CLI 1.0.86/1.0.87). Nothing else marks them as the CLI's.
/// Taken as model text they reached the athlete as the coach's reply — "Error:
/// No response was returned. Send your message again to retry." — and a turn
/// the runtime had failed (spent quota, rate limit, lost login, a prompt over
/// the model's context budget) looked like a success, so no fallback tier was
/// ever asked.
#[derive(Debug, Clone, PartialEq, Eq)]
struct CliNotice {
level: NoticeLevel,
message: String,
}
/// The notice a chunk carries, if the chunk is one.
///
/// Recognised by the adapter's rendering: the chunk opens with the whole
/// prefix. The adapter sends each notice as one chunk, while a model's reply
/// streams as deltas, so a reply that mentions an error mid-sentence is left
/// alone.
fn cli_notice(text: &str) -> Option<CliNotice> {
[
("Error: ", NoticeLevel::Error),
("Warning: ", NoticeLevel::Warning),
("Info: ", NoticeLevel::Info),
]
.into_iter()
.find_map(|(prefix, level)| {
text.strip_prefix(prefix).map(|message| CliNotice {
level,
message: message.trim().to_owned(),
})
})
}
/// The error a CLI-answered turn becomes.
///
/// Every kind returned here is a provider fault, so a fallback chain asks its
/// next tier. The text is all ACP carries — the runtime's own error type does
/// not survive the adapter — so quota and login failures are told apart by
/// their wording, which matters only for how long a router rests the tier.
fn notice_error(errors: &str, message: String) -> RunnerError {
let lower = errors.to_lowercase();
let mentions = |needles: &[&str]| needles.iter().any(|n| lower.contains(n));
if mentions(&[
"rate limit",
"rate-limit",
"quota",
"premium request",
"too many requests",
]) {
RunnerError::rate_limit("copilot-acp", message)
} else if mentions(&[
"authenticat",
"/login",
"copilot access",
"token may be expired",
"sign in",
]) {
RunnerError::auth_failure(message)
} else {
RunnerError::external_service("copilot-acp", message)
}
}
/// Process a session/update notification, accumulating content and tool calls.
///
/// Returns the reply text the notification added, if any — what a streaming
/// caller forwards. A CLI notice and withheld continuation text return `None`.
fn process_notification(params: &Value, acc: &mut TurnAccumulator) -> Option<String> {
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>>>,
) -> Option<String> {
let params = params.get("params").unwrap_or(params);
let Ok(notif) = serde_json::from_value::<schema::SessionNotification>(params.clone()) else {
return None;
};
match ¬if.update {
schema::SessionUpdate::AgentMessageChunk(chunk) => {
let schema::ContentBlock::Text(text) = &chunk.content else {
return None;
};
if let Some(notice) = cli_notice(&text.text) {
warn!(
level = notice.level.label(),
message = %notice.message,
"ACP: the CLI wrote a notice into the reply stream; kept out of the reply"
);
acc.notices.push(notice);
return None;
}
let accepted = acc.accept_text(&text.text)?;
if let Some(tx) = event_tx {
let _ = tx.send(Ok(HeadlessStreamEvent::TextDelta(accepted.clone())));
}
return Some(accepted);
}
schema::SessionUpdate::UsageUpdate(_) => acc.begin_model_request(),
schema::SessionUpdate::ToolCall(tc) => {
acc.record_tool_call();
// ACP carries a title and a status; the name, arguments and
// result stay `None` because the adapter never sends them.
let observed = ObservedToolCall {
id: tc.tool_call_id.0.to_string(),
title: tc.title.clone(),
status: format!("{:?}", tc.status),
..ObservedToolCall::default()
};
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())));
}
}
}
_ => {}
}
None
}
/// 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");
if let Some(err) = acc.failure() {
return Err(err);
}
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}"),
));
}
if let Some(err) = acc.failure() {
return Err(err);
}
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;
}
// Only text the accumulator takes into the reply is
// streamed: never a CLI notice, never a continuation's
// text after the model replied.
if let Some(delta) = process_notification(params, &mut acc) {
let _ = chunk_tx.send(Ok(StreamChunk {
delta,
is_final: false,
finish_reason: None,
}));
}
}
}
"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) {
let _ = 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}"),
));
}
if let Some(err) = acc.failure() {
return Err(err);
}
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
// ---------------------------------------------------------------------------
/// 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,
/// Working directory of every subprocess this runner spawns and `cwd` of
/// every session they serve, resolved once from `config.working_directory`
/// by [`session_cwd`] so the pooled and the streamed paths agree.
session_cwd: PathBuf,
/// 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::with_config(CopilotHeadlessConfig::from_env())
}
/// Create a new provider with explicit configuration.
#[must_use]
pub fn with_config(config: CopilotHeadlessConfig) -> Self {
let session_cwd = session_cwd(config.working_directory.as_deref());
info!(cwd = %session_cwd.display(), "copilot --acp subprocesses run in this directory");
Self {
config,
session_cwd,
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.
///
/// A configured override is made absolute against the host's cwd: the
/// subprocess runs in [`Self::session_cwd`], and `Command` leaves a relative
/// program path with a changed `current_dir` platform-specific.
fn resolve_cli_path(&self) -> Result<PathBuf, RunnerError> {
// `which` resolves every form the override can take — a bare name
// through PATH, a relative path against the host's cwd, an absolute
// path as is — and returns an absolute path, which matters because the
// subprocess runs in `session_cwd`, not where the host was started.
let requested = self
.config
.cli_path
.as_deref()
.unwrap_or_else(|| Path::new("copilot"));
which::which(requested)
.map_err(|_| RunnerError::binary_not_found(requested.display().to_string()))
}
/// 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 turn = render_turn(request, self.config.max_history_turns, true);
let mut blocks = vec![json!({"type": "text", "text": turn.text})];
if let Some(images) = turn.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
}
/// 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: &Path,
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,
&self.session_cwd,
)
.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,
&self.session_cwd,
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 = 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 = 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,
&self.session_cwd,
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 HeadlessTurnProvider for CopilotHeadlessRunner {
async fn converse(&self, request: &ChatRequest) -> Result<HeadlessToolResponse, RunnerError> {
Self::converse(self, request).await
}
async fn converse_stream(
&self,
request: &ChatRequest,
) -> Result<HeadlessEventStream, RunnerError> {
Self::converse_stream(self, request).await
}
}
#[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 = 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() {
// Size the prompt so the NEXT question is answerable from the
// logs. The rate is now known — 18 of 99 turn-attempts on
// 2026-09-01, matching the 2 of 11 measured when this was first
// diagnosed, so it is stable rather than drifting — but nothing
// recorded WHICH prompts do it. The retry recovering only 2 of
// 18 says it is a property of the prompt, not a flake, and
// these fields are what let that be tested against a hypothesis
// (size, model, block count) instead of guessed at.
let prompt_bytes: usize = prompt_blocks.iter().map(|b| b.to_string().len()).sum();
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,
model = %model,
blocks = prompt_blocks.len(),
prompt_bytes,
system_prompt_bytes = system_prompt.map_or(0, str::len),
finish_reason = response.finish_reason.as_deref().unwrap_or("none"),
"ACP complete: empty turn, retrying on a fresh session"
);
attempt += 1;
continue;
}
warn!(
attempt,
model = %model,
blocks = prompt_blocks.len(),
prompt_bytes,
system_prompt_bytes = system_prompt.map_or(0, str::len),
"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 = 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,
&self.session_cwd,
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, ErrorKind};
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"
);
}
#[test]
fn the_published_list_carries_the_5_5_sonnet_and_haiku() {
// Until a session reports the account's own list, this runner
// publishes the catalog, the same one copilot_sdk publishes.
let runner = CopilotHeadlessRunner::with_config(CopilotHeadlessConfig::default());
let published = runner.available_models();
for model in ["claude-sonnet-5.5", "claude-haiku-5.5"] {
assert!(
published.iter().any(|id| id == model),
"{model} missing from {published:?}"
);
}
}
/// 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 the_configured_model_being_absent_is_the_incident_condition() {
let observed = models_from_session(&session_result());
assert!(
model_is_unavailable("claude-sonnet-4.6", observed.as_ref()),
"a retired id the account cannot use is exactly the silent substitution that ran \
production on GPT and Gemini for two months while the logs said sonnet"
);
}
#[test]
fn a_model_the_account_may_use_raises_nothing() {
let observed = models_from_session(&session_result());
assert!(
!model_is_unavailable("claude-sonnet-5", observed.as_ref()),
"the configured model is enabled on this account; warning here would train the \
operator to ignore the warning"
);
}
#[test]
fn a_disabled_model_is_treated_as_unavailable() {
let observed = models_from_session(&session_result());
assert!(
model_is_unavailable("some-locked-model", observed.as_ref()),
"the id appears on the wire but carries copilotEnablement=disabled, so the account \
may not use it and copilot will substitute"
);
}
#[test]
fn an_unreported_list_never_warns() {
assert!(
!model_is_unavailable("claude-sonnet-5", None),
"an older CLI reports no list at all; absent information is not a mismatch, and \
warning on every turn would be noise that hides the real one"
);
}
#[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(¶ms, 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(¶ms, 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(¶ms, 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(¶ms, 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(¶ms, 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(¶ms, 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()
},
session_cwd: session_cwd(None),
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"
);
}
fn acp_args(model: &str) -> Vec<String> {
acp_command(
Path::new("copilot"),
None,
model,
Path::new("/tmp/embacle-acp-test"),
)
.as_std()
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect()
}
/// The ACP spawn opts in to MCP server instructions, so the loopback tool
/// host's `initialize` instructions reach the system prompt as they do on
/// the `-p` runner (carnet#507).
#[test]
fn acp_command_allows_all_mcp_server_instructions() {
let args = acp_args("gpt-5.4");
assert_eq!(args.first().map(String::as_str), Some("--acp"));
assert!(
args.iter()
.any(|a| a == "--allow-all-mcp-server-instructions"),
"ACP spawn must pass --allow-all-mcp-server-instructions, got {args:?}"
);
let idx = args
.iter()
.position(|a| a == "--model")
.expect("--model flag should be present"); // Safe: test assertion
assert_eq!(args.get(idx + 1).map(String::as_str), Some("gpt-5.4"));
}
/// The `copilot --acp` child gets the allowlisted host variables, its
/// token and nothing else of the host's environment: a prompt-steered
/// model that reads its environment finds no host secret there.
#[cfg(unix)]
#[tokio::test]
async fn the_acp_child_inherits_only_the_allowlisted_environment() {
use std::os::unix::fs::PermissionsExt;
const HOST_SECRET: &str = "EMBACLE_TEST_ACP_HOST_SECRET";
env::set_var(HOST_SECRET, "leak");
let dir = tempfile::tempdir().unwrap(); // Safe: test setup
let script = dir.path().join("copilot");
fs::write(&script, "#!/bin/sh\nenv > env.dump\n").unwrap(); // Safe: test setup
fs::set_permissions(&script, fs::Permissions::from_mode(0o755)).unwrap(); // Safe: test setup
// A sibling test's fork can hold the just-written script's write
// descriptor open, and exec then fails with ETXTBSY (os error 26).
let mut attempts = 0;
let status = loop {
attempts += 1;
let mut cmd = acp_command(&script, Some("tok-acp-1"), "", dir.path());
match cmd.status().await {
Err(e) if e.raw_os_error() == Some(26) && attempts < 10 => {
time::sleep(Duration::from_millis(50)).await;
}
outcome => break outcome.unwrap(), // Safe: test assertion
}
};
env::remove_var(HOST_SECRET);
assert!(status.success(), "the stand-in exits 0: {status:?}");
let dump = fs::read_to_string(dir.path().join("env.dump")).unwrap(); // Safe: the stand-in wrote it in its cwd
let names: Vec<&str> = dump
.lines()
.filter_map(|line| line.split_once('=').map(|(name, _)| name))
.collect();
assert!(
dump.lines()
.any(|line| line == "COPILOT_GITHUB_TOKEN=tok-acp-1"),
"the configured token reaches the child: {names:?}"
);
assert!(names.contains(&"PATH"), "{names:?}");
assert!(
!dump.contains(HOST_SECRET),
"a host variable outside the allowlist leaked: {names:?}"
);
let allowed = runtime_env_keys();
let unexpected: Vec<&&str> = names
.iter()
.filter(|name| !allowed.iter().any(|key| key == *name))
.filter(|name| !matches!(**name, "PWD" | "SHLVL" | "_" | "OLDPWD"))
.collect();
assert!(
unexpected.is_empty(),
"only allowlisted variables reach the child, and the shell's own: {unexpected:?}"
);
}
/// An unpinned model drops `--model` but keeps the instructions opt-in.
#[test]
fn acp_command_without_model_still_allows_mcp_server_instructions() {
let args = acp_args("");
assert!(!args.iter().any(|a| a == "--model"));
assert!(args
.iter()
.any(|a| a == "--allow-all-mcp-server-instructions"));
}
#[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");
}
/// The names copilot reads from a session cwd. None of them may exist in
/// the scratch directory, or the model would inherit a project.
const PROJECT_FILES: [&str; 3] = [".mcp.json", ".github/mcp.json", "AGENTS.md"];
#[test]
fn an_unconfigured_runner_gets_a_scratch_cwd_not_the_host_cwd() {
let host_cwd = env::current_dir().unwrap(); // Safe: test setup
let dir = session_cwd(None);
assert_ne!(
dir, host_cwd,
"the host's cwd is the directory being isolated from"
);
assert!(dir.is_absolute(), "ACP requires an absolute session cwd");
assert!(
dir.starts_with(env::temp_dir()),
"{} is not under the system temp dir",
dir.display()
);
assert!(
dir.file_name()
.and_then(|n| n.to_str())
.is_some_and(|n| n == format!("embacle-copilot-{}", process::id())),
"scratch dir is keyed by process id: {}",
dir.display()
);
// What spawn_copilot does before handing the directory to the CLI.
fs::create_dir_all(&dir).unwrap(); // Safe: test assertion on the resolved path being creatable
assert!(dir.is_dir());
for name in PROJECT_FILES {
assert!(
!dir.join(name).exists(),
"{name} in the scratch dir would reach the model"
);
}
fs::remove_dir_all(&dir).unwrap(); // Safe: test cleanup of the directory this test created
}
#[test]
fn a_configured_working_directory_is_used_verbatim() {
let tmp = tempfile::tempdir().unwrap(); // Safe: test setup, tempdir creation
let dir = session_cwd(Some(tmp.path()));
assert_eq!(dir, tmp.path());
assert_ne!(dir, env::current_dir().unwrap()); // Safe: test setup
}
#[test]
fn a_relative_working_directory_is_anchored_to_the_host_cwd() {
let dir = session_cwd(Some(Path::new("scratch/copilot")));
assert!(dir.is_absolute(), "ACP requires an absolute session cwd");
assert_eq!(
dir,
env::current_dir().unwrap().join("scratch/copilot") // Safe: test setup
);
}
#[test]
fn a_bare_cli_name_is_found_on_path_and_a_missing_one_is_named() {
let runner = CopilotHeadlessRunner::with_config(CopilotHeadlessConfig {
cli_path: Some(PathBuf::from("sh")),
..CopilotHeadlessConfig::default()
});
let found = runner.resolve_cli_path().unwrap(); // Safe: test assertion, sh is on every PATH
assert!(found.is_absolute(), "{} must be absolute", found.display());
// The stem, not the last component: a Windows PATH search finds `sh.exe`.
assert_eq!(
found.file_stem().and_then(|stem| stem.to_str()),
Some("sh"),
"{} is not the sh that PATH names",
found.display()
);
let runner = CopilotHeadlessRunner::with_config(CopilotHeadlessConfig {
cli_path: Some(PathBuf::from("embacle-no-such-binary")),
..CopilotHeadlessConfig::default()
});
let err = runner.resolve_cli_path().unwrap_err();
assert!(
err.to_string().contains("embacle-no-such-binary"),
"the error names the override: {err}"
);
}
#[test]
fn runner_resolves_session_cwd_from_config() {
let tmp = tempfile::tempdir().unwrap(); // Safe: test setup, tempdir creation
let configured = CopilotHeadlessRunner::with_config(CopilotHeadlessConfig {
working_directory: Some(tmp.path().to_path_buf()),
..CopilotHeadlessConfig::default()
});
assert_eq!(configured.session_cwd, tmp.path());
let unconfigured = CopilotHeadlessRunner::with_config(CopilotHeadlessConfig::default());
assert_ne!(
unconfigured.session_cwd,
env::current_dir().unwrap(), // Safe: test setup
"a default config must not run copilot in the host's cwd"
);
assert!(unconfigured.session_cwd.starts_with(env::temp_dir()));
}
#[test]
fn session_params_carry_the_session_cwd() {
let dir = session_cwd(None);
let host_cwd = env::current_dir().unwrap(); // Safe: test setup
let params = build_session_params("claude-sonnet-4.6", &dir, Some("be brief"), &[]);
assert_eq!(params["cwd"], json!(dir.to_string_lossy()));
assert_ne!(params["cwd"], json!(host_cwd.to_string_lossy()));
assert_eq!(params["model"], "claude-sonnet-4.6");
assert_eq!(params["systemPrompt"], "be brief");
assert_eq!(params["mcpServers"], json!([]));
}
#[test]
fn session_params_without_system_prompt_omit_the_key() {
let tmp = tempfile::tempdir().unwrap(); // Safe: test setup, tempdir creation
let servers = vec![McpServerConfig {
name: "dravr".to_owned(),
transport: McpTransport::Http {
url: "http://localhost:8081/mcp".to_owned(),
headers: vec![],
},
}];
let params = build_session_params("gpt-4.1", tmp.path(), None, &servers);
assert_eq!(params["cwd"], json!(tmp.path().to_string_lossy()));
assert!(params.get("systemPrompt").is_none());
assert_eq!(params["mcpServers"][0]["name"], "dravr");
assert_eq!(params["mcpServers"][0]["url"], "http://localhost:8081/mcp");
}
#[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");
}
/// The two chunks the CLI sent, verbatim, for a prompt over the model's
/// context budget (CLI 1.0.87, provider mode): a warning, then the
/// runtime's error, and no model text at all.
const CONTEXT_BUDGET_WARNING: &str = "Warning: Static system messages and tool definitions \
exceed the model's usable context budget. Reduce static context or switch to a \
larger-context model.";
const NO_RESPONSE_ERROR: &str =
"Error: No response was returned. Send your message again to retry.";
#[test]
fn a_turn_the_cli_answered_alone_is_an_error_not_a_reply() {
let mut acc = TurnAccumulator::new();
process_notification(
&make_text_chunk_notification(CONTEXT_BUDGET_WARNING),
&mut acc,
);
process_notification(&make_text_chunk_notification(NO_RESPONSE_ERROR), &mut acc);
assert_eq!(acc.content, "", "the CLI's text is not the model's reply");
let err = acc.failure().unwrap(); // Safe: test assertion — a CLI-answered turn fails
assert_eq!(err.kind, ErrorKind::ExternalService, "{err}");
assert!(
err.kind.is_provider_fault(),
"a fallback chain asks its next tier"
);
let text = err.to_string();
assert!(text.contains("No response was returned"), "{text}");
assert!(
text.contains("usable context budget"),
"the warning is kept for diagnosis: {text}"
);
}
#[test]
fn a_notice_beside_a_real_answer_is_dropped_and_the_answer_kept() {
let mut acc = TurnAccumulator::new();
process_notification(
&make_text_chunk_notification(
"Info: Rate limited. Switching to auto mode to continue.",
),
&mut acc,
);
process_notification(&make_text_chunk_notification("Your FTP is "), &mut acc);
process_notification(&make_text_chunk_notification("250 W."), &mut acc);
assert_eq!(acc.content, "Your FTP is 250 W.");
assert_eq!(acc.notices.len(), 1);
assert!(acc.failure().is_none(), "the model answered");
}
#[test]
fn a_reply_that_mentions_an_error_is_the_model_speaking() {
let mut acc = TurnAccumulator::new();
process_notification(
&make_text_chunk_notification("The watch logged an Error: sensor dropout at km 12."),
&mut acc,
);
assert_eq!(
acc.content,
"The watch logged an Error: sensor dropout at km 12."
);
assert!(acc.notices.is_empty());
}
#[test]
fn a_cli_error_is_classified_by_its_wording() {
let cases = [
("Error: You have exceeded your premium request quota.", ErrorKind::RateLimit),
("Error: Rate limit exceeded, retry later.", ErrorKind::RateLimit),
(
"Error: Your authentication token may be expired. Start 'copilot' and run the '/login' command to re-authenticate",
ErrorKind::AuthFailure,
),
(NO_RESPONSE_ERROR, ErrorKind::ExternalService),
];
for (chunk, expected) in cases {
let mut acc = TurnAccumulator::new();
process_notification(&make_text_chunk_notification(chunk), &mut acc);
let err = acc.failure().unwrap(); // Safe: test assertion — an error-only turn fails
assert_eq!(err.kind, expected, "{chunk}");
assert!(err.kind.is_provider_fault(), "{chunk}: falls through");
}
}
#[test]
fn a_cli_notice_is_never_streamed_as_reply_text() {
let mut acc = TurnAccumulator::new();
let (tx, mut rx) = mpsc::unbounded_channel();
process_notification_streaming(
&make_text_chunk_notification(CONTEXT_BUDGET_WARNING),
&mut acc,
&tx,
);
process_notification_streaming(
&make_text_chunk_notification(NO_RESPONSE_ERROR),
&mut acc,
&tx,
);
assert!(rx.try_recv().is_err(), "no delta reached the stream");
assert!(acc.failure().is_some());
}
#[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");
}
// -- Autopilot continuations ------------------------------------------
/// The athlete-facing reply of the 2026-10-01 incident turn (dravr-carnet
/// #690), verbatim from the persisted message.
const INCIDENT_REPLY: &str = "Je n'arrive pas à récupérer le détail complet de cette sortie \
depuis Strava en ce moment (petit pépin technique côté fournisseur) — mais j'ai les \
chiffres de base.\n\nEasy Run #30, dimanche 27 septembre : environ 9,9 km en 1h38, avec \
seulement 54 m de dénivelé — donc un parcours plat. Ça donne un rythme d'environ \
9:54/km, plutôt tranquille pour une sortie \"easy\", ce qui colle bien avec le nom. Par \
contre, je n'ai pas la fréquence cardiaque ni les splits pour cette activité, donc je ne \
peux pas confirmer que c'était vraiment en zone facile niveau effort — juste que le \
rythme l'était.\n\nSi tu veux, je peux réessayer un peu plus tard pour aller chercher le \
détail complet.";
/// What the model wrote in the Autopilot continuation of that turn, and
/// what the athlete received glued to the reply.
const INCIDENT_CONTINUATION: &str = "Le retour persiste (404 côté Strava) — je l'ai déjà \
signalé à l'athlète avec une analyse basée sur les données disponibles (distance, \
durée, dénivelé, allure) et une mention honnête de l'absence de fréquence cardiaque et \
de splits. J'ai répondu de façon complète compte tenu des données accessibles.";
const INCIDENT_SESSION: &str = "5796d5e9-7ff0-4d44-935d-c5ee9911b310";
/// The prompt request id the recorded turns answer.
const RECORDED_PROMPT_ID: i64 = 4;
fn wire_update(session_id: &str, update: &Value) -> String {
json!({
"jsonrpc": "2.0",
"method": "session/update",
"params": { "sessionId": session_id, "update": update },
})
.to_string()
}
fn wire_usage(session_id: &str, used: u64) -> String {
wire_update(
session_id,
&json!({ "sessionUpdate": "usage_update", "used": used, "size": 200_000 }),
)
}
fn wire_text(session_id: &str, text: &str) -> String {
wire_update(
session_id,
&json!({
"sessionUpdate": "agent_message_chunk",
"content": { "type": "text", "text": text },
}),
)
}
/// A tool call and its completion, as the CLI reports one.
fn wire_tool(session_id: &str, id: &str, title: &str, status: &str) -> [String; 2] {
[
wire_update(
session_id,
&json!({
"sessionUpdate": "tool_call", "toolCallId": id, "title": title,
"kind": "other", "status": "pending",
}),
),
wire_update(
session_id,
&json!({ "sessionUpdate": "tool_call_update", "toolCallId": id, "status": status }),
),
]
}
fn wire_end_turn(prompt_id: i64) -> String {
json!({
"jsonrpc": "2.0",
"id": prompt_id,
"result": { "stopReason": "end_turn" },
})
.to_string()
}
/// Split text into the small deltas a model streams.
fn wire_text_chunks(session_id: &str, text: &str) -> Vec<String> {
let chars: Vec<char> = text.chars().collect();
chars
.chunks(37)
.map(|c| wire_text(session_id, &c.iter().collect::<String>()))
.collect()
}
/// The incident turn's shape on the wire: three tool rounds, the reply in
/// a response that calls no tool, then the continuation Autopilot started
/// — which retried the activity fetch, narrated the turn, and called
/// `task_complete`. Each model request opens with a `usage_update`, as
/// recorded on CLI 1.0.90.
fn incident_wire() -> String {
let s = INCIDENT_SESSION;
let mut lines = vec![wire_usage(s, 15_600)];
for (n, (id, title)) in [
("t1", "dravr-get_activities"),
("t2", "dravr-analyze_activity"),
("t3", "dravr-get_activity_intelligence"),
]
.into_iter()
.enumerate()
{
lines.extend(wire_tool(s, id, title, "completed"));
lines.push(wire_usage(s, 16_000 + n as u64 * 1_000));
}
lines.extend(wire_text_chunks(s, INCIDENT_REPLY));
// The reply's response called no tool: Autopilot continues.
lines.push(wire_usage(s, 20_000));
lines.extend(wire_tool(s, "t4", "dravr-analyze_activity", "failed"));
lines.push(wire_usage(s, 21_000));
lines.extend(wire_text_chunks(s, INCIDENT_CONTINUATION));
lines.extend(wire_tool(s, "t5", "task_complete", "completed"));
lines.push(wire_end_turn(RECORDED_PROMPT_ID));
lines.join("\n") + "\n"
}
/// A recorded `copilot --acp` turn (CLI 1.0.90, claude-sonnet-5,
/// Autopilot): a preamble, a file read, the answer, then a continuation
/// that only called `task_complete`. Trimmed to `session/update` and the
/// prompt result; the read's local path is replaced with `/work/note.txt`
/// and its diff output dropped. Nothing else is edited.
const RECORDED_PREAMBLE_TOOL_ANSWER: &str = r#"{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"usage_update","used":15614,"size":200000}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"Je vais lire le"}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":" fichier note.txt d"}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"ans le répertoire courant"}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"."}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"tool_call","toolCallId":"toolu_01RqDfTVJJQeGkLxK2GSYHJu","title":"Viewing /work/note.txt","kind":"read","status":"pending","rawInput":{"path":"/work/note.txt"},"locations":[{"path":"/work/note.txt"}]}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"tool_call_update","toolCallId":"toolu_01RqDfTVJJQeGkLxK2GSYHJu","status":"completed","rawOutput":{"content":"Le mot secret est: érable.\n"}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"usage_update","used":15807,"size":200000}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"Le mot secret est «"}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":" érable »."}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"usage_update","used":15982,"size":200000}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"tool_call","toolCallId":"toolu_01HPouXQQpyd51v1UhjHQt5P","title":"task_complete","kind":"other","status":"pending","rawInput":{"summary":"J'ai lu note.txt et communiqué le mot secret : « érable »."}}}}
{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"6b6bb20c-010d-4f6e-9d83-a3b7107143aa","update":{"sessionUpdate":"tool_call_update","toolCallId":"toolu_01HPouXQQpyd51v1UhjHQt5P","status":"completed","content":[{"type":"content","content":{"type":"text","text":"J'ai lu note.txt et communiqué le mot secret : « érable »."}}],"rawOutput":{"content":"J'ai lu note.txt et communiqué le mot secret : « érable ».","detailedContent":"✓ Task completed: J'ai lu note.txt et communiqué le mot secret : « érable »."}}}}
{"jsonrpc":"2.0","id":4,"result":{"stopReason":"end_turn","usage":{"inputTokens":62755,"outputTokens":241,"totalTokens":62996,"thoughtTokens":0,"cachedReadTokens":53945,"cachedWriteTokens":8804}}}
"#;
const RECORDED_PREAMBLE_SESSION: &str = "6b6bb20c-010d-4f6e-9d83-a3b7107143aa";
/// Hand a recorded wire to a collector over a real pipe, as `copilot
/// --acp` would write it. `cat` echoes the bytes written to its stdin.
async fn replay(wire: &str) -> (AcpTransport, Child) {
let mut child = Command::new("cat")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.kill_on_drop(true)
.spawn()
.unwrap(); // Safe: test setup — `cat` is on every unix runner
let mut stdin = child.stdin.take().unwrap(); // Safe: test setup — piped above
stdin.write_all(wire.as_bytes()).await.unwrap(); // Safe: test setup
stdin.flush().await.unwrap(); // Safe: test setup
let stdout = child.stdout.take().unwrap(); // Safe: test setup — piped above
(AcpTransport::new(stdin, stdout), child)
}
/// What `complete()` returns for a recorded turn.
async fn replay_complete(
wire: &str,
session_id: &str,
) -> (ChatResponse, Vec<ObservedToolCall>) {
let (mut transport, _child) = replay(wire).await;
collect_complete(
&mut transport,
RECORDED_PROMPT_ID,
"claude-sonnet-5".to_owned(),
PermissionPolicy::DenyAll,
session_id,
)
.await
.unwrap() // Safe: test assertion — the recorded turn ends normally
}
/// Every delta `complete_stream()` emits for a recorded turn, joined.
async fn replay_complete_stream(wire: &str, session_id: &str) -> String {
let (mut transport, _child) = replay(wire).await;
let (tx, mut rx) = mpsc::unbounded_channel();
collect_streaming(
&mut transport,
RECORDED_PROMPT_ID,
&tx,
PermissionPolicy::DenyAll,
session_id,
)
.await
.unwrap(); // Safe: test assertion — the recorded turn ends normally
drop(tx);
let mut streamed = String::new();
while let Some(chunk) = rx.recv().await {
streamed.push_str(&chunk.unwrap().delta); // Safe: test assertion
}
streamed
}
/// Every text delta `converse_stream()` emits for a recorded turn, joined,
/// and the content of the response it ends with.
async fn replay_converse_stream(wire: &str, session_id: &str) -> (String, String) {
let (mut transport, _child) = replay(wire).await;
let (tx, mut rx) = mpsc::unbounded_channel();
let response = collect_streaming_with_tools(
&mut transport,
RECORDED_PROMPT_ID,
"claude-sonnet-5".to_owned(),
PermissionPolicy::DenyAll,
&tx,
session_id,
)
.await
.unwrap(); // Safe: test assertion — the recorded turn ends normally
drop(tx);
let mut streamed = String::new();
while let Some(event) = rx.recv().await {
let event = event.unwrap(); // Safe: test assertion
if let HeadlessStreamEvent::TextDelta(delta) = event {
streamed.push_str(&delta);
}
}
(streamed, response.content)
}
#[tokio::test]
async fn a_continuation_after_the_reply_is_not_delivered_by_complete() {
let (response, tool_calls) = replay_complete(&incident_wire(), INCIDENT_SESSION).await;
assert_eq!(response.content, INCIDENT_REPLY);
assert_eq!(
tool_calls.len(),
5,
"the continuation's tool calls are still reported"
);
}
#[tokio::test]
async fn a_continuation_after_the_reply_is_never_streamed_by_complete_stream() {
let streamed = replay_complete_stream(&incident_wire(), INCIDENT_SESSION).await;
assert_eq!(streamed, INCIDENT_REPLY);
}
#[tokio::test]
async fn a_continuation_after_the_reply_is_never_streamed_by_converse_stream() {
let (streamed, content) = replay_converse_stream(&incident_wire(), INCIDENT_SESSION).await;
assert_eq!(streamed, INCIDENT_REPLY);
assert_eq!(content, INCIDENT_REPLY);
}
/// A preamble the model writes before a tool call is part of the run that
/// answers, so it stays: only a continuation is withheld.
#[tokio::test]
async fn text_before_and_after_a_tool_call_is_one_reply_on_every_path() {
let expected = "Je vais lire le fichier note.txt dans le répertoire courant.\
Le mot secret est « érable ».";
let wire = RECORDED_PREAMBLE_TOOL_ANSWER;
let session = RECORDED_PREAMBLE_SESSION;
let (response, tool_calls) = replay_complete(wire, session).await;
assert_eq!(response.content, expected);
assert_eq!(tool_calls.len(), 2);
assert_eq!(replay_complete_stream(wire, session).await, expected);
let (streamed, content) = replay_converse_stream(wire, session).await;
assert_eq!(streamed, expected);
assert_eq!(content, expected);
}
/// The continuation's own wording, when the model wrote no reply before
/// it, is the only answer there is.
#[test]
fn a_continuation_after_a_run_with_no_reply_is_kept() {
let s = "test-session";
let mut acc = TurnAccumulator::new();
let (tx, mut rx) = mpsc::unbounded_channel();
let mut feed = |line: &str| {
let msg: Value = serde_json::from_str(line).unwrap(); // Safe: test fixture
process_notification_streaming(&msg, &mut acc, &tx);
};
feed(&wire_usage(s, 1));
feed(&wire_text(s, " \n"));
feed(&wire_usage(s, 2));
feed(&wire_text(s, "Your FTP is 250 W."));
for line in wire_tool(s, "tc", "task_complete", "completed") {
feed(&line);
}
assert_eq!(acc.content.trim(), "Your FTP is 250 W.");
let mut streamed = String::new();
while let Ok(Ok(event)) = rx.try_recv() {
if let HeadlessStreamEvent::TextDelta(delta) = event {
streamed.push_str(&delta);
}
}
assert_eq!(streamed.trim(), "Your FTP is 250 W.");
}
/// A continuation that writes text and only then calls `task_complete` —
/// no retried tool — is withheld the same way.
#[test]
fn a_text_then_task_complete_continuation_is_withheld() {
let s = "test-session";
let mut acc = TurnAccumulator::new();
let mut delivered = String::new();
let mut feed = |line: String| {
let msg: Value = serde_json::from_str(&line).unwrap(); // Safe: test fixture
if let Some(delta) = process_notification(&msg, &mut acc) {
delivered.push_str(&delta);
}
};
feed(wire_usage(s, 1));
feed(wire_text(s, "Your FTP is 250 W."));
feed(wire_usage(s, 2));
feed(wire_text(s, "I already gave the athlete their FTP."));
for line in wire_tool(s, "tc", "task_complete", "completed") {
feed(line);
}
assert_eq!(delivered, "Your FTP is 250 W.");
assert_eq!(acc.content, "Your FTP is 250 W.");
}
#[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);
}
}