diff --git a/CHANGELOG.md b/CHANGELOG.md index 63b3e3f..9087e7f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,40 @@ Versioning and Keep a Changelog conventions. cleanup of its hidden Codex thread can neither delay a turn nor replace a completed turn's success with a write error. Code that constructs `AdapterOutput` with struct syntax must now set `optional_writes`. +- `Provider::Acp` and `providers::acp::Acp` (default feature `acp`): one + adapter for any agent that speaks the Agent Client Protocol over stdio — + Gemini CLI (`gemini --acp`), GitHub Copilot (`copilot --acp`), Cursor + (`cursor-agent acp`) and the rest of the ACP registry. Agents live in a + shared, updatable `AcpAgents` registry and a turn selects one with the + `agent` harness option, advertised as a `HarnessControlKind::Agent` + control. Turns run `initialize`, `session/new` (or `session/resume`, else + `session/load` without its replay), optional mode and model selection, then + `session/prompt`; `session/request_permission` reaches the interaction + handler unless `AcceptEdits`/`FullAccess` decide it, cancellation sends + `session/cancel`, and `-32000` fails as an authentication error before the + prompt. ACP requires an absolute working directory; a relative one is + refused before the agent starts. `Acp::probe_agent` runs the handshake (and + optionally opens a session) to report capabilities and sign-in state. See + `docs/how-to/run-acp-agents.md`. +- ACP agents appear in harness discovery and readiness. `discover_harnesses` + now includes `Provider::Acp` when the adapter is registered and + `AgentRuntime::readiness(Provider::Acp)` resolves each registered agent's + own command inside the configured transport instead of a placeholder `acp` + executable: the harness is `Ready` when at least one agent is installed, + `readiness.detail` names every agent, and the `Agent` control group lists + only installed agents. New `AgentAdapter::selectable_executables` (default + `None`, so existing adapters are unchanged) and `SelectableExecutable` + expose that per-turn choice to the runtime. +- `TurnEvent::PlanUpdated` with `PlanEntry`/`PlanEntryStatus`: a complete + replacement of an agent's execution plan, emitted for ACP `plan` updates. +- `AgentAdapter::interaction_answered`: called by the turn loop right after + an approval or question reply is encoded, so an adapter that tracks + unanswered requests stops refusing one the application already answered + when the turn is cancelled. The default does nothing. +- `AgentAdapter::terminal_exit_grace`: a stdio provider that keeps running + after its terminal frame and end-of-input is stopped after this grace + period instead of holding the finished turn until its deadline. The default + (`None`) keeps reading to end-of-output as before. - `AgentTask::model`: the model a Claude Code subagent runs on, as its own messages report it (for example `claude-haiku-5-5`), including a subagent that inherits the parent's model. It is `None` until the subagent's first @@ -146,6 +180,14 @@ Versioning and Keep a Changelog conventions. ### Fixed +- A finished turn, probe or metadata query no longer waits for a descendant + of the provider that kept its stderr open after the process ended, which + turned a completed ACP turn into a timeout when the agent's terminal grace + stopped it. The runtime waits a bounded 2 seconds for stderr to close and + keeps the tail that had arrived for diagnostics. A turn keeps draining the + pipe in the background past that point, so a tool left running by + `ToolProcessPolicy::PreserveOnCompletion` never gets a broken pipe; probes, + which have already stopped the process, stop reading. - Changing a Claude turn's model, effort or permission mode no longer replaces the retained process and stops its background subagents and shells. The process switches them in place with `set_model`, diff --git a/Cargo.toml b/Cargo.toml index 03f445b..a5a8353 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ repository = "https://github.com/gotempsh/agent-runtime-sdk" homepage = "https://github.com/gotempsh/agent-runtime-sdk" documentation = "https://docs.rs/temps-agent-runtime" readme = "README.md" -keywords = ["agents", "claude", "codex", "opencode", "sandbox"] +keywords = ["agents", "claude", "codex", "acp", "sandbox"] categories = ["api-bindings", "asynchronous", "development-tools"] include = [ "/src/**", @@ -29,11 +29,13 @@ include = [ ] [features] -default = ["claude", "codex", "opencode", "pi", "nono", "ssh", "docker", "tailnet"] +default = ["claude", "codex", "opencode", "pi", "acp", "nono", "ssh", "docker", "tailnet"] claude = [] codex = [] opencode = [] pi = [] +# Any agent that speaks the Agent Client Protocol over stdio. +acp = [] nono = [] ssh = [] # Run provider CLIs inside an existing container through `docker exec`. diff --git a/docs/how-to/run-acp-agents.md b/docs/how-to/run-acp-agents.md new file mode 100644 index 0000000..3f7e59b --- /dev/null +++ b/docs/how-to/run-acp-agents.md @@ -0,0 +1,123 @@ +# Run Agent Client Protocol agents + +`providers::acp::Acp` runs any agent that speaks the +[Agent Client Protocol](https://agentclientprotocol.com) (ACP) over stdio: +Gemini CLI, GitHub Copilot, Cursor, Kimi, goose, Qwen Code and the rest of the +[ACP registry](https://agentclientprotocol.com/registry). One adapter, +`Provider::Acp`, serves them all; the application registers each agent's +command and picks one per turn. + +```rust +use temps_agent_runtime::providers::acp::AGENT_OPTION; +use temps_agent_runtime::providers::{Acp, AcpAgent, AcpAgents}; +use temps_agent_runtime::{AgentRuntime, PermissionMode, Provider, TurnRequest}; + +let agents = AcpAgents::new(); +agents.insert(AcpAgent::new("gemini", "Gemini CLI", "gemini", ["--acp"]))?; + +let mut builder = AgentRuntime::builder(); +builder.register(Acp::new(agents.clone())); +let runtime = builder.build()?; + +let mut request = TurnRequest::new(Provider::Acp, "/path/to/project", "Fix the failing test"); +request.harness_options.insert(AGENT_OPTION.into(), "gemini".into()); +request.permission_mode = PermissionMode::Default; +let result = runtime.run(request, &events, Some(&approvals)).await?; +``` + +The working directory must be absolute: ACP sends it to the agent as `cwd`, +and the agent starts inside it, so a relative path would resolve a second time. +A relative one fails the turn before the agent starts. + +`AcpAgents` is shared: inserting, replacing or removing an agent on any clone +takes effect for the next turn of every runtime holding it. + +## What a turn does + +1. `initialize` with protocol version 1. The client offers no `fs` or + `terminal` capability, so the agent uses its own tools in the working + directory, inside whatever sandbox wraps the process. +2. `session/new`, or for `TurnRequest::session_id`: `session/resume` when the + agent advertises it, else `session/load` (its history replay is dropped). + An agent with neither starts a new session and emits a warning. A resumed + session the agent no longer knows fails with `acp::session_not_found`. +3. `PermissionMode::Custom(mode)` selects that session mode + (`session/set_mode`); `TurnRequest::model` selects a model through the + session's `model` config option (or `session/set_model`). An unknown mode + or model fails before the prompt, naming the choices. +4. `session/prompt`. `session/update` notifications become `TextDelta`, + `ReasoningDelta`, `ToolCall`, `PlanUpdated`, `Usage` and `Warning` events; + the prompt's `stopReason` ends the turn. + +`launch_context.system_prompt_append` is sent as a leading text block of a new +session's first prompt (ACP has no system prompt). Stdio and HTTP MCP servers +are passed in the session request with their referenced variables resolved +from `TurnRequest::environment`; HTTP servers are dropped with a warning when +the agent does not support them. + +## Permissions + +| `PermissionMode` | `session/request_permission` | +| --- | --- | +| `Default`, `Custom(_)` | Every request reaches the `InteractionHandler` | +| `AcceptEdits` | `read`, `edit`, `delete`, `move` and `search` kinds are allowed; others reach the handler | +| `FullAccess` | Every request is allowed | +| `Plan` | Rejected: ACP has no client-enforceable read-only mode | + +An agent asks only when it decides to; an agent configured to run tools +without asking never sends a request. + +## Offer the installed agents + +Register the adapter and `discover_harnesses` lists `Provider::Acp` like any +other harness. The runtime resolves each registered agent's own command inside +the configured transport (local, SSH, container or sandbox), concurrently: + +- `HarnessReadiness::status` is `Ready` when at least one agent is installed + and `NotInstalled` when none is (or none is registered). +- `readiness.detail` names every agent, installed or not found, with its + version when it reports one. +- The `agent` control group (`HarnessControlKind::Agent`) lists only the + installed agents, so its options are exactly what a turn can run. A group + with no installed agent is omitted. +- `runtime.readiness(Provider::Acp)` reports the same aggregate instead of + looking for a program named `acp`. + +Authentication stays `Unknown` here: opening a session can leave an empty +session in an agent's history, so the sign-in check is the explicit +`Acp::probe_agent` below. Registry changes apply to the next discovery without +rebuilding the runtime. + +Custom adapters that choose among several executables per turn opt in through +`AgentAdapter::selectable_executables`. + +## Check an agent + +```rust +let probe = Acp::probe_agent(&agent, &scratch_dir, true, Duration::from_secs(30)).await?; +match probe.authentication { + AcpAuthentication::Authenticated => { /* ready */ } + AcpAuthentication::Required { message } => { /* show the agent's sign-in hint */ } + _ => { /* unknown */ } +} +``` + +`probe_agent` starts the agent locally, runs `initialize` and, with +`check_session`, `session/new` in the given (absolute) directory, then stops +it. It +reports the protocol version, agent name and version, sign-in methods and +session capabilities. Agents may record that empty session in their history, +so run it on an explicit user action. + +## Limits + +- One process per turn; the process is not retained across turns. +- An agent that keeps running after the turn ends is stopped after + `terminal_exit_grace` (3 seconds), which also stops tool processes it left + running. A descendant that escaped the agent's process group and still + holds its stderr no longer delays the finished turn: the runtime waits at + most 2 seconds for stderr to close, keeps what had arrived, and goes on + draining the pipe in the background so a preserved tool is never given a + broken pipe. +- Questions (elicitation) and client-side file system and terminal methods + are not offered. diff --git a/docs/reference/events.md b/docs/reference/events.md index f363edd..b2f11f9 100644 --- a/docs/reference/events.md +++ b/docs/reference/events.md @@ -108,6 +108,7 @@ status transition. | Event | Meaning | Durable projection | | --- | --- | --- | | `TasksChanged` | Complete bounded task snapshot for the current run | Replace the materialized task collection | +| `PlanUpdated` | Complete execution plan an agent published (ACP `plan`) | Replace the materialized plan | | `TaskActivity` | One ordered native task transition | Append to the run timeline | Claude currently emits these events for native Task/Agent work. `AgentTask` diff --git a/src/adapter.rs b/src/adapter.rs index 2dc7a89..a017ff7 100644 --- a/src/adapter.rs +++ b/src/adapter.rs @@ -86,6 +86,36 @@ pub struct AuthenticationProbeSpec { pub command: CommandSpec, } +/// One executable an adapter may run, chosen per turn rather than fixed. +/// +/// See [`AgentAdapter::selectable_executables`]. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct SelectableExecutable { + /// Value that selects this executable. It is the id of the matching option + /// in the adapter's [`crate::HarnessControlKind::Agent`] control group. + pub id: String, + /// User-facing name, used in readiness diagnostics. + pub label: String, + /// Executable name or path meaningful inside the selected transport. + pub executable: PathBuf, +} + +impl SelectableExecutable { + /// Describe one selectable executable. + pub fn new( + id: impl Into, + label: impl Into, + executable: impl Into, + ) -> Self { + Self { + id: id.into(), + label: label.into(), + executable: executable.into(), + } + } +} + impl CommandSpec { /// Create a process specification for an executable. pub fn new(program: impl Into) -> Self { @@ -327,9 +357,24 @@ pub trait AgentAdapter: Send + Sync { Provider::Codex => "codex", Provider::OpenCode => "opencode", Provider::Pi => "pi", + Provider::Acp => "acp", }) } + /// Executables this adapter can run when a turn chooses among several, + /// such as the agents registered with the ACP adapter. + /// + /// `None`, the default, means [`Self::executable`] is the one program the + /// adapter runs. `Some` replaces it for readiness: the runtime resolves + /// every entry inside the configured transport, reports the provider as + /// installed when at least one resolves, names each entry in the + /// readiness detail, and keeps only the installed entries as options of + /// the [`crate::HarnessControlKind::Agent`] control group in discovery. + /// An empty `Some` means nothing is registered yet. + fn selectable_executables(&self) -> Option> { + None + } + /// Static and interactive permission behavior implemented by this adapter. fn permission_support(&self) -> PermissionSupport; @@ -657,6 +702,19 @@ pub trait AgentAdapter: Send + Sync { Ok(AdapterOutput::default()) } + /// How long a stdio provider may take to exit on its own after its + /// terminal frame and the closing of its stdin. + /// + /// `None` — the default — reads the provider to end-of-output and waits + /// for it to exit, because closing stdin is what makes it exit. `Some` + /// stops reading at the terminal frame and terminates a provider that is + /// still running when the grace period ends; the protocol already decided + /// the turn, so that stop is not a failure. Protocol servers such as ACP + /// agents need this: nothing obliges them to exit at end-of-input. + fn terminal_exit_grace(&self) -> Option { + None + } + /// Encode a provider-native cooperative interrupt for the running turn. /// /// Returning `Some` makes the runtime write the frame on cancellation and @@ -685,6 +743,21 @@ pub trait AgentAdapter: Send + Sync { original: &Value, answer: Option, ) -> Result>>; + + /// Record that the reply to the approval or question `original` was + /// encoded and is being written to the provider. + /// + /// Called by the turn loop right after + /// [`AgentAdapter::approval_response`] or + /// [`AgentAdapter::question_response`], with the same `original`. An + /// adapter that tracks requests still waiting for an answer — to refuse + /// them when it encodes an [`AgentAdapter::interrupt_request`] — drops + /// this one here, so a cancellation that arrives before the provider + /// reacts to the answer does not answer the same request twice. The + /// default does nothing. + fn interaction_answered(&self, original: &Value, state: &mut AdapterState) { + let _ = (original, state); + } } /// Resolve an executable override or search common user-local `PATH` roots. diff --git a/src/lib.rs b/src/lib.rs index 31e35cc..f5d7b71 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,7 +56,7 @@ pub mod tailnet; pub use adapter::{ AccountUsageProbeSpec, AdapterOutput, AdapterState, AgentAdapter, AuthenticationProbeSpec, CatalogProbeSpec, CommandSpec, IdleFrame, InteractionRequest, OversizedFrame, ProtocolStreams, - ProviderTerminalFailure, + ProviderTerminalFailure, SelectableExecutable, }; pub use chat::{ Chat, ChatApproval, ChatAttachment, ChatCommit, ChatEvent, ChatEventData, ChatMessage, @@ -111,10 +111,10 @@ pub use types::{ AgentWorkflowAgentState, AgentWorkflowPhase, ApprovalDecision, ApprovalRequest, AutoCompactionPolicy, CompactionTrigger, ContextCompaction, ContextWindowUsage, DenyAll, EventSink, InteractionHandler, LaunchContext, LaunchContextCapabilities, McpServerConfig, - NoopEventSink, PermissionMode, PermissionSupport, Provider, ProviderProcessStatus, - ProviderReadiness, QuestionAnswer, QuestionOption, QuestionPrompt, QuestionRequest, RunStatus, - SecretString, ToolCallStatus, ToolImage, ToolProcessPolicy, TurnCapabilities, TurnEvent, - TurnProvenance, TurnRequest, TurnResult, Usage, + NoopEventSink, PermissionMode, PermissionSupport, PlanEntry, PlanEntryStatus, Provider, + ProviderProcessStatus, ProviderReadiness, QuestionAnswer, QuestionOption, QuestionPrompt, + QuestionRequest, RunStatus, SecretString, ToolCallStatus, ToolImage, ToolProcessPolicy, + TurnCapabilities, TurnEvent, TurnProvenance, TurnRequest, TurnResult, Usage, }; pub use startup::{StartupObserver, StartupStage, StartupTiming}; diff --git a/src/process.rs b/src/process.rs index 202960c..151b546 100644 --- a/src/process.rs +++ b/src/process.rs @@ -1,6 +1,8 @@ use std::collections::VecDeque; use std::ffi::OsStr; use std::process::Stdio; +use std::sync::{Arc, Mutex}; +use std::time::Duration; use tokio::io::AsyncReadExt; use tokio::process::{Child, Command}; @@ -199,6 +201,115 @@ where Ok(String::from_utf8_lossy(&tail.into_iter().collect::>()).into_owned()) } +/// Reads a child's stderr into a bounded tail while the process runs. +/// +/// A descendant that outlives its parent can keep the pipe open, so the +/// stream may not reach end-of-file when the process does. [`Self::finish`] +/// therefore waits only a bounded time for it and keeps what arrived: a +/// finished process never waits on a stray descendant, and a failure +/// diagnostic still carries the stderr the process wrote. +/// +/// What happens to the reader once the wait is over, or the capture is +/// dropped, depends on who owns the process: +/// +/// - By default the reader stops. This is for probes, which have already +/// stopped the process they started. +/// - [`Self::detach`] leaves it draining in the background. A process +/// started with `ToolProcessPolicy::PreserveOnCompletion` may leave tools +/// running that inherited this pipe, and closing the read end would hand +/// them a broken pipe (`SIGPIPE`) on their next write. The reader holds +/// only the bounded tail and ends when the last writer closes the pipe. +pub(crate) struct StderrCapture { + task: Option>>, + tail: Arc>>, + detached: bool, +} + +impl StderrCapture { + pub(crate) fn spawn(mut stderr: R, capacity: usize) -> Self + where + R: tokio::io::AsyncRead + Unpin + Send + 'static, + { + let tail = Arc::new(Mutex::new(VecDeque::new())); + let shared = Arc::clone(&tail); + let task = tokio::spawn(async move { + let mut buffer = [0_u8; 4096]; + loop { + let read = stderr.read(&mut buffer).await?; + if read == 0 { + return Ok(()); + } + let mut tail = shared + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for byte in &buffer[..read] { + if tail.len() == capacity { + tail.pop_front(); + } + tail.push_back(*byte); + } + } + }); + Self { + task: Some(task), + tail, + detached: false, + } + } + + /// Keep reading in the background when the wait ends or this is dropped. + /// + /// Call this only once the owner has decided to leave descendants + /// running; until then an early return stops the reader with the process. + pub(crate) fn detach(&mut self) { + self.detached = true; + } + + /// Stop reading now, even when [`Self::detach`]ed. + pub(crate) fn abort(&self) { + if let Some(task) = &self.task { + task.abort(); + } + } + + /// Wait up to `grace` for end-of-file and return the tail captured so far. + pub(crate) async fn finish(mut self, grace: Duration) -> std::io::Result { + // The handle stays in `self` while waiting, so cancelling this future + // drops `self`, which applies the same policy as an elapsed grace. + let Some(task) = self.task.as_mut() else { + return Ok(String::new()); + }; + match tokio::time::timeout(grace, task).await { + Ok(Ok(Ok(()))) => { + self.task = None; + } + Ok(Ok(Err(error))) => return Err(error), + Ok(Err(error)) => { + return Err(std::io::Error::other(format!( + "stderr reader task failed: {error}" + ))) + } + // The reader is still running; dropping `self` below leaves it + // or stops it according to `detach`. + Err(_) => {} + } + let tail = self + .tail + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let bytes = tail.iter().copied().collect::>(); + Ok(String::from_utf8_lossy(&bytes).into_owned()) + } +} + +impl Drop for StderrCapture { + fn drop(&mut self) { + if !self.detached { + self.abort(); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -395,4 +506,84 @@ mod tests { .expect("termination status"); assert!(!status.success()); } + + #[tokio::test] + async fn stderr_capture_keeps_what_arrived_when_the_pipe_stays_open() { + let (reader, mut writer) = tokio::io::duplex(64); + let capture = StderrCapture::spawn(reader, 8); + writer.write_all(b"0123456789").await.unwrap(); + // The writer stays open, as a descendant holding stderr would. + let started = std::time::Instant::now(); + let tail = capture.finish(Duration::from_millis(200)).await.unwrap(); + assert_eq!(tail, "23456789"); + assert!(started.elapsed() < Duration::from_secs(5)); + drop(writer); + } + + #[tokio::test] + async fn a_detached_stderr_capture_keeps_draining_for_a_late_writer() { + let (reader, mut writer) = tokio::io::duplex(8); + let mut capture = StderrCapture::spawn(reader, 64); + capture.detach(); + writer.write_all(b"early").await.unwrap(); + let tail = capture.finish(Duration::from_millis(200)).await.unwrap(); + assert_eq!(tail, "early"); + // The reader is still there: a write far larger than the pipe's + // buffer completes instead of failing with a broken pipe. + tokio::time::timeout(Duration::from_secs(5), writer.write_all(&[b'x'; 1024])) + .await + .expect("the late writer was blocked") + .expect("the reader closed the pipe"); + } + + #[tokio::test] + async fn cancelling_a_detached_wait_keeps_draining_for_a_late_writer() { + let (reader, mut writer) = tokio::io::duplex(8); + let mut capture = StderrCapture::spawn(reader, 64); + capture.detach(); + let cancelled = tokio::time::timeout( + Duration::from_millis(50), + capture.finish(Duration::from_secs(30)), + ) + .await; + assert!(cancelled.is_err(), "the wait should still be pending"); + tokio::time::timeout(Duration::from_secs(5), writer.write_all(&[b'x'; 1024])) + .await + .expect("the late writer was blocked") + .expect("cancelling the wait closed the pipe"); + } + + #[tokio::test] + async fn cancelling_the_wait_stops_the_reader() { + let (reader, mut writer) = tokio::io::duplex(8); + let capture = StderrCapture::spawn(reader, 64); + let cancelled = tokio::time::timeout( + Duration::from_millis(50), + capture.finish(Duration::from_secs(30)), + ) + .await; + assert!(cancelled.is_err(), "the wait should still be pending"); + // Once the reader is gone the pipe is broken for the writer. + let mut broken = false; + for _ in 0..100 { + if writer.write_all(&[b'x'; 64]).await.is_err() { + broken = true; + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(broken, "the reader kept running after the wait was dropped"); + } + + #[tokio::test] + async fn stderr_capture_returns_at_end_of_file_without_waiting_out_the_grace() { + let (reader, mut writer) = tokio::io::duplex(64); + let capture = StderrCapture::spawn(reader, 32); + writer.write_all(b"boom").await.unwrap(); + drop(writer); + let started = std::time::Instant::now(); + let tail = capture.finish(Duration::from_secs(30)).await.unwrap(); + assert_eq!(tail, "boom"); + assert!(started.elapsed() < Duration::from_secs(5)); + } } diff --git a/src/providers/acp.rs b/src/providers/acp.rs new file mode 100644 index 0000000..65b95a4 --- /dev/null +++ b/src/providers/acp.rs @@ -0,0 +1,2796 @@ +//! Generic adapter for agents that speak the Agent Client Protocol. +//! +//! [ACP](https://agentclientprotocol.com) is JSON-RPC 2.0 over the agent's +//! stdio, one message per line. Any agent that implements it — Gemini CLI +//! (`gemini --acp`), GitHub Copilot (`copilot --acp`), Cursor +//! (`cursor-agent acp`), Kimi, goose, Qwen Code and many more — runs through +//! this one adapter. The application registers each agent's command in an +//! [`AcpAgents`] registry and selects one per turn with the +//! [`AGENT_OPTION`] harness option. +//! +//! One turn is one process: +//! +//! 1. `initialize` with protocol version 1. The client offers neither file +//! system nor terminal capabilities, so the agent uses its own tools in the +//! turn's working directory, inside whatever sandbox wraps the process. +//! 2. `session/new`, or — to continue a conversation — `session/resume` when +//! the agent supports it, else `session/load`, whose history replay is +//! dropped because the application already has it. An agent that can do +//! neither starts a new session and the turn says so in a warning. +//! 3. Optional `session/set_mode` (for [`PermissionMode::Custom`]) and model +//! selection through the session's model config option. +//! 4. `session/prompt`. `session/update` notifications stream text, thoughts, +//! tool calls, plans and usage until the prompt's response reports its stop +//! reason, which ends the turn. +//! +//! `session/request_permission` reaches the application as an approval unless +//! the permission mode already decides it. Cancellation sends `session/cancel` +//! and answers every open permission request as cancelled, as the protocol +//! requires. + +use std::collections::BTreeMap; +use std::fmt; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, RwLock}; +use std::time::Duration; + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; + +use crate::adapter::AdapterState; +use crate::lifecycle::DeliveryState; +use crate::{ + AdapterOutput, AgentAdapter, ApprovalDecision, ApprovalRequest, CommandSpec, + ContextWindowUsage, HarnessControlGroup, HarnessControlKind, HarnessControlOption, + InteractionRequest, LaunchContextCapabilities, McpServerConfig, PermissionMode, + PermissionSupport, PlanEntry, PlanEntryStatus, Provider, ProviderProcessErrorKind, + ProviderReadiness, ProviderTerminalFailure, QuestionAnswer, QuestionRequest, Result, RunStatus, + RuntimeError, SelectableExecutable, ToolCallStatus, TurnEvent, TurnRequest, Usage, +}; + +/// Harness option naming the registered [`AcpAgent::id`] a turn runs. +pub const AGENT_OPTION: &str = "agent"; +/// ACP protocol version this adapter implements. +pub const PROTOCOL_VERSION: u64 = 1; +/// JSON-RPC error code ACP agents return when they need authentication. +pub const AUTH_REQUIRED_CODE: i64 = -32000; +/// JSON-RPC error code ACP agents return for an unknown resource. +pub const RESOURCE_NOT_FOUND_CODE: i64 = -32002; + +const STATE_KEY: &str = "acp"; +/// The prompt waits here, outside the per-line state, until the session is +/// ready, so the state decoded on every line never carries a copy of it. +const PROMPT_KEY: &str = "acp_prompt"; + +const ID_INITIALIZE: u64 = 1; +const ID_SESSION: u64 = 2; +const ID_SET_MODE: u64 = 3; +const ID_SET_MODEL: u64 = 4; +const ID_PROMPT: u64 = 5; + +const MAX_TOOL_OUTPUT_CHARS: usize = 4096; +const MAX_TRACKED_TOOLS: usize = 256; +/// A tool input larger than this is not kept for the call's later updates. +const MAX_KEPT_INPUT_BYTES: usize = 16 * 1024; +const MAX_OPEN_PERMISSIONS: usize = 64; +const MAX_DIAGNOSTIC_CHARS: usize = 1000; +const MAX_TITLE_CHARS: usize = 200; +const MAX_LISTED_CHOICES: usize = 12; +const MAX_AGENT_ID_BYTES: usize = 64; +/// How long a finished agent may take to exit after its stdin closes before +/// the runtime stops it. Node-based agents commonly keep their event loop +/// alive after end-of-input. +const EXIT_GRACE: Duration = Duration::from_secs(3); +/// How long a probe waits for a stopped agent's stderr to close. +const STDERR_DRAIN: Duration = Duration::from_secs(2); + +/// One ACP agent command the application makes available. +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AcpAgent { + /// Stable identifier selected through [`AGENT_OPTION`]: lowercase ASCII + /// letters, digits, `.`, `_` and `-`. + pub id: String, + /// Human-readable name used in diagnostics. + pub label: String, + /// Executable meaningful inside the execution transport. + pub command: PathBuf, + /// Arguments that start the agent in ACP mode, such as `["--acp"]`. + pub args: Vec, + /// Explicit environment additions for the agent process. The runtime + /// otherwise passes only its safe allowlist, so provider API keys must be + /// supplied here or through [`TurnRequest::environment`]. + #[serde(default)] + pub environment: BTreeMap, +} + +impl fmt::Debug for AcpAgent { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("AcpAgent") + .field("id", &self.id) + .field("label", &self.label) + .field("command", &self.command) + .field("args", &self.args) + .field( + "environment_keys", + &self.environment.keys().collect::>(), + ) + .finish() + } +} + +impl AcpAgent { + /// Describe an agent started by `command` with `args`. + pub fn new( + id: impl Into, + label: impl Into, + command: impl Into, + args: impl IntoIterator>, + ) -> Self { + Self { + id: id.into(), + label: label.into(), + command: command.into(), + args: args.into_iter().map(Into::into).collect(), + environment: BTreeMap::new(), + } + } + + /// Check the identifier and command before the agent is registered. + pub fn validate(&self) -> Result<()> { + if !valid_agent_id(&self.id) { + return Err(invalid( + "acp_agent.id", + "ACP agent ids are 1-64 lowercase letters, digits, `.`, `_` or `-`, starting with a letter or digit", + )); + } + let command = self.command.as_os_str(); + if command.is_empty() || self.command.to_string_lossy().contains('\0') { + return Err(invalid("acp_agent.command", "the agent command is empty")); + } + if self.args.iter().any(|arg| arg.contains('\0')) { + return Err(invalid("acp_agent.args", "arguments cannot contain NUL")); + } + if self + .environment + .keys() + .any(|name| name.is_empty() || name.contains('=') || name.contains('\0')) + { + return Err(invalid( + "acp_agent.environment", + "environment names must be non-empty and cannot contain `=` or NUL", + )); + } + Ok(()) + } + + fn command_spec(&self) -> CommandSpec { + let mut spec = CommandSpec::new(self.command.clone()); + spec.args = self.args.iter().map(Into::into).collect(); + for (name, value) in &self.environment { + spec.environment.insert(name.into(), value.into()); + } + spec.interactive_stdin = true; + spec + } +} + +/// Whether `path` is absolute on the machine that will run the agent. +/// +/// ACP requires `cwd` to be absolute, and the agent starts inside the working +/// directory, so a relative one would resolve a second time from there +/// (`workspace` becoming `workspace/workspace`). The adapter cannot resolve it +/// for a remote transport, so it refuses one instead. The host and the +/// transport may use different path syntaxes, hence the textual check. +fn is_absolute_directory(path: &Path) -> bool { + if path.is_absolute() { + return true; + } + let text = path.to_string_lossy(); + let bytes = text.as_bytes(); + text.starts_with('/') + || text.starts_with("\\\\") + || (bytes.len() >= 3 + && bytes[0].is_ascii_alphabetic() + && bytes[1] == b':' + && matches!(bytes[2], b'\\' | b'/')) +} + +fn relative_directory_message(path: &Path) -> String { + format!( + "ACP agents need an absolute working directory, but `{}` is relative; pass its full path", + path.display() + ) +} + +fn valid_agent_id(id: &str) -> bool { + let bytes = id.as_bytes(); + !bytes.is_empty() + && bytes.len() <= MAX_AGENT_ID_BYTES + && bytes + .first() + .is_some_and(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit()) + && bytes.iter().all(|byte| { + byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'.' | b'_' | b'-') + }) +} + +/// Shared, updatable set of ACP agents. +/// +/// Cloning shares the registry, so an application can keep one handle and +/// change the available agents without rebuilding its runtimes. +#[derive(Debug, Clone, Default)] +pub struct AcpAgents { + inner: Arc>>, +} + +impl AcpAgents { + /// An empty registry. + pub fn new() -> Self { + Self::default() + } + + /// Replace every registered agent. Invalid agents are rejected and the + /// registry is left unchanged. + pub fn replace(&self, agents: impl IntoIterator) -> Result<()> { + let mut next = BTreeMap::new(); + for agent in agents { + agent.validate()?; + next.insert(agent.id.clone(), agent); + } + *self.write() = next; + Ok(()) + } + + /// Register or replace one agent. + pub fn insert(&self, agent: AcpAgent) -> Result<()> { + agent.validate()?; + self.write().insert(agent.id.clone(), agent); + Ok(()) + } + + /// Remove one agent. + pub fn remove(&self, id: &str) -> Option { + self.write().remove(id) + } + + /// Look up one agent. + pub fn get(&self, id: &str) -> Option { + self.inner + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(id) + .cloned() + } + + /// Every registered agent, ordered by id. + pub fn list(&self) -> Vec { + self.inner + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .values() + .cloned() + .collect() + } + + fn write(&self) -> std::sync::RwLockWriteGuard<'_, BTreeMap> { + self.inner + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +/// Adapter for any agent that speaks the Agent Client Protocol over stdio. +/// +/// Turns select an agent from the shared [`AcpAgents`] registry with the +/// [`AGENT_OPTION`] harness option. +#[derive(Debug, Clone, Default)] +pub struct Acp { + agents: AcpAgents, +} + +impl Acp { + /// Run agents from `agents`. + pub fn new(agents: AcpAgents) -> Self { + Self { agents } + } + + /// The registry this adapter reads. + pub fn agents(&self) -> &AcpAgents { + &self.agents + } + + fn agent_for(&self, request: &TurnRequest) -> Result { + let id = request + .harness_options + .get(AGENT_OPTION) + .map(|id| id.trim()) + .filter(|id| !id.is_empty()) + .ok_or_else(|| { + invalid( + "harness_options.agent", + "choose which ACP agent runs this turn", + ) + })?; + self.agents.get(id).ok_or_else(|| { + invalid( + "harness_options.agent", + format!("no ACP agent named `{id}` is registered"), + ) + }) + } + + /// Start `agent` locally, run the ACP handshake and report what it + /// supports and whether it can open a session without signing in. + /// + /// With `check_session` the probe also calls `session/new` in + /// `working_directory`, which must then be absolute because ACP requires + /// an absolute `cwd` ([`AcpProbeErrorKind::Invalid`] otherwise); agents + /// that require sign-in refuse it with + /// [`AUTH_REQUIRED_CODE`]. Agents may record that empty session in their + /// own history, so applications should run this only on an explicit user + /// action. The process is always stopped before this returns. + pub async fn probe_agent( + agent: &AcpAgent, + working_directory: &Path, + check_session: bool, + timeout: Duration, + ) -> std::result::Result { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + agent + .validate() + .map_err(|error| AcpProbeError::new(AcpProbeErrorKind::Invalid, error.to_string()))?; + if check_session && !is_absolute_directory(working_directory) { + return Err(AcpProbeError::new( + AcpProbeErrorKind::Invalid, + relative_directory_message(working_directory), + )); + } + let spec = agent.command_spec(); + let mut child = crate::process::spawn(&spec, working_directory).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + AcpProbeError::new( + AcpProbeErrorKind::NotInstalled, + format!("`{}` was not found on PATH", agent.command.display()), + ) + } else { + AcpProbeError::new( + AcpProbeErrorKind::Spawn, + format!("could not start `{}`: {error}", agent.command.display()), + ) + } + })?; + let mut guard = crate::process::ProcessTreeGuard::for_child(&child); + let mut stdin = child.stdin.take(); + let stdout = child.stdout.take(); + let stderr = child.stderr.take(); + let stderr_task = + stderr.map(|stderr| crate::process::StderrCapture::spawn(stderr, MAX_DIAGNOSTIC_CHARS)); + let exchange = async { + let mut lines = BufReader::new(stdout.ok_or_else(|| { + AcpProbeError::new(AcpProbeErrorKind::Spawn, "stdout was not piped") + })?) + .lines(); + let writer = stdin.as_mut().ok_or_else(|| { + AcpProbeError::new(AcpProbeErrorKind::Spawn, "stdin was not piped") + })?; + let send = |frame: Value| { + let mut bytes = serde_json::to_vec(&frame).unwrap_or_default(); + bytes.push(b'\n'); + bytes + }; + let initialize = send(initialize_request()); + writer + .write_all(&initialize) + .await + .map_err(io_probe_error)?; + writer.flush().await.map_err(io_probe_error)?; + let initialized = read_response(&mut lines, ID_INITIALIZE).await?; + if let Some(error) = initialized.get("error") { + return Err(AcpProbeError::new( + AcpProbeErrorKind::Protocol, + format!("initialize failed: {}", error_message(error)), + )); + } + let result = initialized.get("result").cloned().unwrap_or(Value::Null); + let mut probe = AcpAgentProbe::from_initialize(&result); + if !check_session { + return Ok(probe); + } + let session = send(request( + ID_SESSION, + "session/new", + json!({"cwd": working_directory, "mcpServers": []}), + )); + writer.write_all(&session).await.map_err(io_probe_error)?; + writer.flush().await.map_err(io_probe_error)?; + let opened = read_response(&mut lines, ID_SESSION).await?; + probe.authentication = match opened.get("error") { + None => AcpAuthentication::Authenticated, + Some(error) if error_code(error) == Some(AUTH_REQUIRED_CODE) => { + AcpAuthentication::Required { + message: bounded(&error_message(error), MAX_DIAGNOSTIC_CHARS), + } + } + Some(error) => AcpAuthentication::Unknown { + message: bounded(&error_message(error), MAX_DIAGNOSTIC_CHARS), + }, + }; + Ok(probe) + }; + let outcome = tokio::time::timeout(timeout, exchange).await; + drop(stdin); + guard.terminate(); + let _ = child.kill().await; + // Stopping the agent's process group does not close stderr when a + // descendant left the group and kept the pipe, so the read is bounded + // and abandoned instead of holding the probe past its timeout. + let stderr = match stderr_task { + Some(task) => task.finish(STDERR_DRAIN).await.ok(), + None => None, + }; + let with_stderr = |mut error: AcpProbeError| { + if let Some(stderr) = stderr.as_deref().map(str::trim).filter(|s| !s.is_empty()) { + error.message = format!("{} ({})", error.message, bounded(stderr, 400)); + } + error + }; + match outcome { + Ok(Ok(probe)) => Ok(probe), + Ok(Err(error)) => Err(with_stderr(error)), + Err(_) => Err(with_stderr(AcpProbeError::new( + AcpProbeErrorKind::Timeout, + format!( + "`{}` did not answer the ACP handshake within {}s; check that its arguments start ACP mode", + agent.command.display(), + timeout.as_secs() + ), + ))), + } + } +} + +fn io_probe_error(error: std::io::Error) -> AcpProbeError { + AcpProbeError::new( + AcpProbeErrorKind::Protocol, + format!("the agent closed its input: {error}"), + ) +} + +async fn read_response( + lines: &mut tokio::io::Lines, + id: u64, +) -> std::result::Result +where + R: tokio::io::AsyncBufRead + Unpin, +{ + while let Some(line) = lines.next_line().await.map_err(io_probe_error)? { + let Ok(value) = serde_json::from_str::(line.trim()) else { + continue; + }; + if value.get("method").is_none() && value.get("id").and_then(Value::as_u64) == Some(id) { + return Ok(value); + } + } + Err(AcpProbeError::new( + AcpProbeErrorKind::Protocol, + "the agent exited before answering the ACP handshake; check that its arguments start ACP mode and that the installed version supports ACP", + )) +} + +/// What a local ACP handshake reported about one agent. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AcpAgentProbe { + /// Protocol version the agent negotiated. + pub protocol_version: Option, + /// Agent-reported name. + pub agent_name: Option, + /// Agent-reported version. + pub agent_version: Option, + /// Sign-in methods the agent offers. + pub auth_methods: Vec, + /// The agent can replay a stored session (`session/load`). + pub load_session: bool, + /// The agent can continue a stored session without replay (`session/resume`). + pub resume_session: bool, + /// Whether a session could be opened without signing in. + pub authentication: AcpAuthentication, +} + +impl AcpAgentProbe { + fn from_initialize(result: &Value) -> Self { + let info = result.get("agentInfo"); + let capabilities = result.get("agentCapabilities"); + Self { + protocol_version: result.get("protocolVersion").and_then(Value::as_u64), + agent_name: info + .and_then(|info| info.get("title").or_else(|| info.get("name"))) + .and_then(Value::as_str) + .map(|name| bounded(name, MAX_TITLE_CHARS)), + agent_version: info + .and_then(|info| info.get("version")) + .and_then(Value::as_str) + .map(|version| bounded(version, 64)), + auth_methods: auth_methods(result), + load_session: capabilities + .and_then(|caps| caps.get("loadSession")) + .and_then(Value::as_bool) + .unwrap_or(false), + resume_session: capabilities + .and_then(|caps| caps.pointer("/sessionCapabilities/resume")) + .is_some_and(|resume| !resume.is_null()), + authentication: AcpAuthentication::Unknown { + message: "a session was not opened".into(), + }, + } + } +} + +/// A sign-in method an ACP agent advertises. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AcpAuthMethod { + /// Protocol identifier. + pub id: String, + /// Human-readable name. + pub name: String, + /// Optional description. + pub description: Option, +} + +/// Whether an ACP agent could open a session. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "status", rename_all = "snake_case")] +#[non_exhaustive] +pub enum AcpAuthentication { + /// `session/new` succeeded. + Authenticated, + /// The agent refused with [`AUTH_REQUIRED_CODE`]. + Required { + /// Agent diagnostic. + message: String, + }, + /// Not checked, or the agent failed for another reason. + Unknown { + /// Why authentication is unknown. + message: String, + }, +} + +/// Why an ACP probe could not complete. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("{message}")] +pub struct AcpProbeError { + /// Stable category. + pub kind: AcpProbeErrorKind, + /// Bounded diagnostic. + pub message: String, +} + +impl AcpProbeError { + fn new(kind: AcpProbeErrorKind, message: impl Into) -> Self { + Self { + kind, + message: message.into(), + } + } +} + +/// Category of an [`AcpProbeError`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[non_exhaustive] +pub enum AcpProbeErrorKind { + /// The agent definition is invalid. + Invalid, + /// The command was not found. + NotInstalled, + /// The command could not be started. + Spawn, + /// The agent did not answer in time. + Timeout, + /// The agent answered with something other than ACP. + Protocol, +} + +/// Per-turn protocol state retained between output lines. +#[derive(Debug, Default, Serialize, Deserialize)] +#[serde(default)] +struct TurnState { + agent_label: String, + working_directory: PathBuf, + /// Session the caller asked to continue. + resume: Option, + /// Session this turn runs in once opened. + session_id: Option, + /// `session/load` is replaying history that the application already has. + replaying: bool, + /// Instructions delivered ahead of the prompt of a new session. + instructions: Option, + /// `mcpServers` for session requests; HTTP entries are dropped when the + /// agent cannot connect to them. + mcp_servers: Vec, + /// Mode requested through [`PermissionMode::Custom`]. + requested_mode: Option, + /// Model requested for this turn. + requested_model: Option, + permission: Policy, + /// Requests still to send after the session opens, in order. + pending: Vec<(u64, Value)>, + prompt_sent: bool, + supports_images: bool, + /// Tool calls seen this turn. + tools: Vec, + /// Permission requests the client has not replied to, as + /// `(request id, tool call id)`. An interrupt refuses these; a request + /// leaves the list when its reply is written or its tool moves on. + open_permissions: Vec<(Value, Option)>, + warned_noise: bool, +} + +/// A tool call's stable name and, until it finishes, its latest input and +/// output: updates may omit fields they did not change, and every emitted +/// `ToolCall` event is a complete replacement. +#[derive(Debug, Default, Serialize, Deserialize)] +#[serde(default)] +struct TrackedTool { + id: String, + title: String, + input: Option, + /// Latest bounded output an update carried; a final update that only + /// reports the status completes the call with it. + output: Option, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +enum Policy { + /// Every permission request reaches the application. + #[default] + Ask, + /// File edits are allowed; everything else reaches the application. + AcceptEdits, + /// Every permission request is allowed. + Allow, +} + +fn load(state: &AdapterState) -> TurnState { + state + .extensions + .get(STATE_KEY) + .cloned() + .and_then(|value| serde_json::from_value(value).ok()) + .unwrap_or_default() +} + +fn store(state: &mut AdapterState, turn: &TurnState) { + if let Ok(value) = serde_json::to_value(turn) { + state.extensions.insert(STATE_KEY.to_string(), value); + } +} + +fn protocol(message: impl Into) -> RuntimeError { + RuntimeError::Protocol { + provider: Provider::Acp, + message: message.into(), + } +} + +fn invalid(field: &'static str, message: impl Into) -> RuntimeError { + RuntimeError::InvalidRequest { + field, + message: message.into(), + } +} + +fn encode(value: &Value) -> Result> { + serde_json::to_vec(value) + .map_err(|error| protocol(format!("could not encode an ACP message: {error}"))) +} + +fn request(id: u64, method: &str, params: Value) -> Value { + json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}) +} + +fn notification(method: &str, params: Value) -> Value { + json!({"jsonrpc": "2.0", "method": method, "params": params}) +} + +fn reply(id: &Value, result: Value) -> Value { + json!({"jsonrpc": "2.0", "id": id, "result": result}) +} + +fn reply_error(id: &Value, code: i64, message: &str) -> Value { + json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}}) +} + +fn initialize_request() -> Value { + request( + ID_INITIALIZE, + "initialize", + json!({ + "protocolVersion": PROTOCOL_VERSION, + "clientCapabilities": { + "fs": {"readTextFile": false, "writeTextFile": false}, + "terminal": false, + }, + "clientInfo": {"name": "temps-agent-runtime", "version": env!("CARGO_PKG_VERSION")}, + }), + ) +} + +fn bounded(text: &str, limit: usize) -> String { + if text.chars().count() <= limit { + return text.to_string(); + } + let mut bounded: String = text.chars().take(limit).collect(); + bounded.push_str("… [truncated]"); + bounded +} + +fn error_code(error: &Value) -> Option { + error.get("code").and_then(Value::as_i64) +} + +fn error_message(error: &Value) -> String { + let message = error + .get("message") + .and_then(Value::as_str) + .unwrap_or("the agent returned an error"); + match error.get("data") { + Some(Value::String(data)) if !data.trim().is_empty() => format!("{message}: {data}"), + Some(data) if data.is_object() => data + .get("message") + .or_else(|| data.get("details")) + .and_then(Value::as_str) + .map_or_else(|| message.to_string(), |data| format!("{message}: {data}")), + _ => message.to_string(), + } +} + +fn auth_methods(result: &Value) -> Vec { + result + .get("authMethods") + .and_then(Value::as_array) + .map(|methods| { + methods + .iter() + .take(MAX_LISTED_CHOICES) + .filter_map(|method| { + let id = method.get("id").and_then(Value::as_str)?; + Some(AcpAuthMethod { + id: bounded(id, 128), + name: bounded( + method.get("name").and_then(Value::as_str).unwrap_or(id), + MAX_TITLE_CHARS, + ), + description: method + .get("description") + .and_then(Value::as_str) + .map(|text| bounded(text, MAX_DIAGNOSTIC_CHARS)), + }) + }) + .collect() + }) + .unwrap_or_default() +} + +fn fail( + state: &mut AdapterState, + output: &mut AdapterOutput, + kind: ProviderProcessErrorKind, + diagnostic: impl Into, + code: &str, + delivery: DeliveryState, +) { + let diagnostic = bounded(&diagnostic.into(), MAX_DIAGNOSTIC_CHARS); + let kind = if kind == ProviderProcessErrorKind::Unknown { + crate::error::classify_provider_failure(&diagnostic) + } else { + kind + }; + state.terminal_failure = Some( + ProviderTerminalFailure::new(kind, diagnostic, delivery) + .with_provider_code(format!("acp::{code}")), + ); + state.result.status = RunStatus::Failed; + output.terminal = true; +} + +/// Resolve the turn's MCP servers into ACP's literal `mcpServers` entries. +/// +/// ACP sends variable values in the session request instead of naming +/// variables the agent should read, so every referenced variable must be part +/// of the turn's explicit environment; the runtime already rejects requests +/// that reference anything else. +fn mcp_servers(request: &TurnRequest) -> Result> { + let lookup = |source: &str, field: &'static str| { + request + .environment + .get(source) + .map(|value| value.expose().to_string()) + .ok_or_else(|| { + invalid( + field, + format!("references environment variable `{source}`, which was not supplied to the turn"), + ) + }) + }; + let mut servers = Vec::with_capacity(request.launch_context.mcp_servers.len()); + for (name, server) in &request.launch_context.mcp_servers { + servers.push(match server { + McpServerConfig::Stdio { + command, + args, + environment_from, + } => { + let mut env = Vec::with_capacity(environment_from.len()); + for (target, source) in environment_from { + env.push(json!({ + "name": target, + "value": lookup(source, "launch_context.mcp_servers")?, + })); + } + json!({"name": name, "command": command, "args": args, "env": env}) + } + McpServerConfig::Http { url, headers_from } => { + let mut headers = Vec::with_capacity(headers_from.len()); + for (header, source) in headers_from { + headers.push(json!({ + "name": header, + "value": lookup(source, "launch_context.mcp_servers")?, + })); + } + json!({"type": "http", "name": name, "url": url, "headers": headers}) + } + }); + } + Ok(servers) +} + +#[async_trait] +impl AgentAdapter for Acp { + fn provider(&self) -> Provider { + Provider::Acp + } + + fn executable(&self) -> PathBuf { + // Agents are chosen per turn; there is no single provider executable. + PathBuf::from("acp") + } + + fn permission_support(&self) -> PermissionSupport { + PermissionSupport { + default: true, + // Decided from the tool kind the agent reports when it asks. + accept_edits: true, + // ACP has no client-enforceable read-only mode. An agent's own + // planning mode is reachable through `Custom` when it offers one. + plan: false, + full_access: true, + // Agent-native session modes through `session/set_mode`. + custom: true, + live_approvals: true, + live_questions: false, + } + } + + fn launch_context_capabilities(&self) -> LaunchContextCapabilities { + LaunchContextCapabilities { + // Delivered as a leading text block of a new session's first + // prompt, because ACP has no system-prompt field. + system_prompt_append: true, + allowed_tools: false, + stdio_mcp: true, + // Dropped with a warning when the agent reports no HTTP support. + http_mcp: true, + strict_mcp_config: false, + } + } + + fn terminal_exit_grace(&self) -> Option { + Some(EXIT_GRACE) + } + + fn selectable_executables(&self) -> Option> { + // The agent is chosen per turn, so readiness is about the registered + // agents' own commands rather than the placeholder `acp` executable. + Some( + self.agents + .list() + .into_iter() + .map(|agent| SelectableExecutable::new(agent.id, agent.label, agent.command)) + .collect(), + ) + } + + fn control_groups(&self) -> Vec { + vec![HarnessControlGroup { + id: AGENT_OPTION.into(), + label: "Agent".into(), + kind: HarnessControlKind::Agent, + options: self + .agents + .list() + .into_iter() + .map(|agent| HarnessControlOption { + description: std::iter::once(agent.command.to_string_lossy().into_owned()) + .chain(agent.args.iter().cloned()) + .collect::>() + .join(" "), + id: agent.id, + label: agent.label, + is_default: false, + dangerous: false, + }) + .collect(), + }] + } + + async fn readiness(&self) -> ProviderReadiness { + let agents = self.agents.list(); + ProviderReadiness { + provider: Provider::Acp, + installed: !agents.is_empty(), + executable: None, + version: None, + detail: if agents.is_empty() { + "No ACP agent is registered.".into() + } else { + format!( + "{} ACP agent(s) registered; check each with Acp::probe_agent.", + agents.len() + ) + }, + } + } + + fn command(&self, request: &TurnRequest) -> Result { + let agent = self.agent_for(request)?; + let mut spec = agent.command_spec(); + spec.initial_stdin = Some(encode(&initialize_request())?); + Ok(spec) + } + + fn prepare_turn(&self, request: &TurnRequest, state: &mut AdapterState) -> Result<()> { + let agent = self.agent_for(request)?; + if !is_absolute_directory(&request.working_directory) { + return Err(invalid( + "working_directory", + relative_directory_message(&request.working_directory), + )); + } + let permission = match &request.permission_mode { + PermissionMode::Default | PermissionMode::Custom(_) => Policy::Ask, + PermissionMode::AcceptEdits => Policy::AcceptEdits, + PermissionMode::FullAccess => Policy::Allow, + mode => { + return Err(invalid( + "permission_mode", + format!("ACP agents support Default, AcceptEdits, FullAccess or an agent mode, not {mode:?}"), + )) + } + }; + let requested_mode = match &request.permission_mode { + PermissionMode::Custom(mode) if !mode.trim().is_empty() => Some(mode.trim().into()), + _ => None, + }; + let requested_model = request + .model + .as_deref() + .map(str::trim) + .filter(|model| !model.is_empty() && *model != "default") + .map(str::to_owned); + let turn = TurnState { + agent_label: agent.label.clone(), + working_directory: request.working_directory.clone(), + resume: request + .session_id + .as_deref() + .map(str::trim) + .filter(|id| !id.is_empty()) + .map(str::to_owned), + instructions: request + .launch_context + .system_prompt_append + .as_deref() + .map(str::trim) + .filter(|text| !text.is_empty()) + .map(str::to_owned), + mcp_servers: mcp_servers(request)?, + requested_mode, + requested_model, + permission, + ..TurnState::default() + }; + store(state, &turn); + state.extensions.insert( + PROMPT_KEY.to_string(), + Value::String(request.prompt.clone()), + ); + Ok(()) + } + + fn interrupt_request(&self, state: &AdapterState) -> Option> { + let turn = load(state); + let session = turn.session_id?; + let mut frames = vec![notification( + "session/cancel", + json!({"sessionId": session}), + )]; + frames.extend( + turn.open_permissions + .iter() + .map(|(id, _)| reply(id, json!({"outcome": {"outcome": "cancelled"}}))), + ); + let lines = frames + .iter() + .filter_map(|frame| serde_json::to_string(frame).ok()) + .collect::>() + .join("\n"); + Some(lines.into_bytes()) + } + + fn interaction_answered(&self, original: &Value, state: &mut AdapterState) { + let Some(id) = original.get("id") else { + return; + }; + let mut turn = load(state); + let before = turn.open_permissions.len(); + // The reply carries the user's decision, so a later cancellation must + // not answer the same request again. + turn.open_permissions.retain(|(open, _)| open != id); + if turn.open_permissions.len() != before { + store(state, &turn); + } + } + + fn parse_line(&self, line: &str, state: &mut AdapterState) -> Result { + let mut output = AdapterOutput::default(); + let mut turn = load(state); + let trimmed = line.trim(); + if trimmed.is_empty() { + return Ok(output); + } + let Ok(value) = serde_json::from_str::(trimmed) else { + // Some agents log to stdout. Say so once instead of failing a turn + // the protocol itself is still carrying. + if !turn.warned_noise { + turn.warned_noise = true; + output.events.push(TurnEvent::Warning { + message: format!( + "{} wrote non-ACP output to stdout; it was ignored", + turn.agent_label + ), + }); + store(state, &turn); + } + return Ok(output); + }; + let method = value.get("method").and_then(Value::as_str); + let id = value.get("id").filter(|id| !id.is_null()); + match (method, id) { + (Some(method), Some(id)) => agent_request(method, id, &value, &mut turn, &mut output)?, + (Some("session/update"), None) => { + session_update(&value, state, &mut turn, &mut output); + } + (None, Some(_)) => response(&value, state, &mut turn, &mut output)?, + (Some(_) | None, None) => {} + } + store(state, &turn); + Ok(output) + } + + fn approval_response( + &self, + _request: &ApprovalRequest, + original: &Value, + decision: ApprovalDecision, + ) -> Result>> { + let id = original + .get("id") + .ok_or_else(|| protocol("the permission request has no id"))?; + let options = original + .pointer("/params/options") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let preferences: &[&str] = match decision { + ApprovalDecision::Allow => &["allow_once", "allow_always"], + ApprovalDecision::AllowForSession => &["allow_always", "allow_once"], + _ => &["reject_once", "reject_always"], + }; + let outcome = option_for(&options, preferences).map_or_else( + || json!({"outcome": "cancelled"}), + |option| json!({"outcome": "selected", "optionId": option}), + ); + Ok(Some(encode(&reply(id, json!({"outcome": outcome})))?)) + } + + fn question_response( + &self, + _request: &QuestionRequest, + _original: &Value, + _answer: Option, + ) -> Result>> { + // This adapter never raises questions. + Ok(None) + } +} + +/// What a permission request asks to do, phrased to follow " wants +/// to": `run \`npm test\``, `edit src/main.rs`. Built from the ACP tool kind +/// and whatever target the request names. +fn permission_action(kind: &str, tool: &Value, input: Option<&Value>) -> String { + let text = |keys: &[&str]| { + keys.iter() + .find_map(|key| input?.get(key)?.as_str()) + .map(|text| bounded(text, MAX_TITLE_CHARS)) + }; + let path = || { + tool.get("locations") + .and_then(Value::as_array) + .and_then(|locations| locations.first()) + .and_then(|location| location.get("path")) + .and_then(Value::as_str) + .map(|path| bounded(path, MAX_TITLE_CHARS)) + .or_else(|| text(&["filePath", "file_path", "path"])) + }; + let on_path = |verb: &str, fallback: &str| { + path().map_or_else(|| fallback.to_owned(), |path| format!("{verb} {path}")) + }; + match kind { + "execute" => text(&["command", "cmd"]).map_or_else( + || "run a command".to_owned(), + |command| format!("run `{command}`"), + ), + "read" => on_path("read", "read a file"), + "edit" => on_path("edit", "edit files"), + "delete" => on_path("delete", "delete files"), + "move" => on_path("move", "move files"), + "search" => "search the workspace".to_owned(), + "fetch" => text(&["url"]).map_or_else( + || "fetch from the network".to_owned(), + |url| format!("fetch {url}"), + ), + _ => "use this tool".to_owned(), + } +} + +fn option_for(options: &[Value], kinds: &[&str]) -> Option { + kinds.iter().find_map(|kind| { + options + .iter() + .find(|option| option.get("kind").and_then(Value::as_str) == Some(kind)) + .and_then(|option| option.get("optionId")) + .and_then(Value::as_str) + .map(str::to_owned) + }) +} + +/// Answer a request the agent sent to the client. +fn agent_request( + method: &str, + id: &Value, + value: &Value, + turn: &mut TurnState, + output: &mut AdapterOutput, +) -> Result<()> { + if method != "session/request_permission" { + // Neither file system, terminal nor elicitation capabilities were + // offered, so a conforming agent never sends these. + output.writes.push(encode(&reply_error( + id, + -32601, + &format!("{method} is not supported by this client"), + ))?); + return Ok(()); + } + let params = value.get("params").unwrap_or(&Value::Null); + let tool = params.get("toolCall").unwrap_or(&Value::Null); + let options = params + .get("options") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let kind = tool.get("kind").and_then(Value::as_str).unwrap_or("other"); + let automatic = match turn.permission { + Policy::Allow => Some(&["allow_once", "allow_always"][..]), + Policy::AcceptEdits if matches!(kind, "edit" | "delete" | "move" | "read" | "search") => { + Some(&["allow_once", "allow_always"][..]) + } + _ => None, + }; + if let Some(option) = automatic.and_then(|kinds| option_for(&options, kinds)) { + output.writes.push(encode(&reply( + id, + json!({"outcome": {"outcome": "selected", "optionId": option}}), + ))?); + return Ok(()); + } + let tool_call_id = tool.get("toolCallId").and_then(Value::as_str); + let tracked = tool_call_id.and_then(|id| turn.tools.iter().find(|tool| tool.id == id)); + // The name the tool call already streamed under, so the request and its + // tool row read as the same thing. + let title = tracked + .map(|tool| tool.title.clone()) + .or_else(|| tool.get("title").and_then(Value::as_str).map(str::to_owned)) + .unwrap_or_else(|| format!("{kind} tool")); + let input = tool + .get("rawInput") + .filter(|input| !input.is_null()) + .cloned() + .or_else(|| tracked.and_then(|tool| tool.input.clone())); + let description = permission_action(kind, tool, input.as_ref()); + if turn.open_permissions.len() < MAX_OPEN_PERMISSIONS { + turn.open_permissions + .push((id.clone(), tool_call_id.map(str::to_owned))); + } + output.interaction = Some(InteractionRequest::Approval { + request: ApprovalRequest { + id: id.to_string(), + tool_name: bounded(&title, MAX_TITLE_CHARS), + input: input + .unwrap_or_else(|| json!({"kind": kind, "locations": tool.get("locations")})), + description: Some(description), + }, + original: value.clone(), + }); + Ok(()) +} + +/// Advance the handshake on a response to one of our requests. +fn response( + value: &Value, + state: &mut AdapterState, + turn: &mut TurnState, + output: &mut AdapterOutput, +) -> Result<()> { + let Some(id) = value.get("id").and_then(Value::as_u64) else { + return Ok(()); + }; + let error = value.get("error"); + let result = value.get("result").unwrap_or(&Value::Null); + match id { + ID_INITIALIZE => { + if let Some(error) = error { + fail( + state, + output, + ProviderProcessErrorKind::Unknown, + format!( + "{} failed to initialize: {}", + turn.agent_label, + error_message(error) + ), + "initialize", + DeliveryState::NotSent, + ); + return Ok(()); + } + let probe = AcpAgentProbe::from_initialize(result); + if probe.protocol_version != Some(PROTOCOL_VERSION) { + fail( + state, + output, + ProviderProcessErrorKind::Unknown, + format!( + "{} speaks ACP protocol version {}, but this client supports version {PROTOCOL_VERSION}", + turn.agent_label, + probe + .protocol_version + .map_or_else(|| "unknown".into(), |version| version.to_string()) + ), + "protocol_version", + DeliveryState::NotSent, + ); + return Ok(()); + } + turn.supports_images = result + .pointer("/agentCapabilities/promptCapabilities/image") + .and_then(Value::as_bool) + .unwrap_or(false); + let http = result + .pointer("/agentCapabilities/mcpCapabilities/http") + .and_then(Value::as_bool) + .unwrap_or(false); + if !http { + let before = turn.mcp_servers.len(); + turn.mcp_servers + .retain(|server| server.get("type").and_then(Value::as_str) != Some("http")); + if turn.mcp_servers.len() != before { + output.events.push(TurnEvent::Warning { + message: format!( + "{} cannot connect to HTTP MCP servers, so {} of them were left out of this turn", + turn.agent_label, + before - turn.mcp_servers.len() + ), + }); + } + } + let cwd = turn.working_directory.clone(); + let servers = Value::Array(turn.mcp_servers.clone()); + let open = match turn.resume.clone() { + Some(session) if probe.resume_session => request( + ID_SESSION, + "session/resume", + json!({"sessionId": session, "cwd": cwd, "mcpServers": servers}), + ), + Some(session) if probe.load_session => { + turn.replaying = true; + request( + ID_SESSION, + "session/load", + json!({"sessionId": session, "cwd": cwd, "mcpServers": servers}), + ) + } + Some(_) => { + output.events.push(TurnEvent::Warning { + message: format!( + "{} cannot resume ACP sessions, so this message starts a new session without the earlier conversation", + turn.agent_label + ), + }); + turn.resume = None; + request( + ID_SESSION, + "session/new", + json!({"cwd": cwd, "mcpServers": servers}), + ) + } + None => request( + ID_SESSION, + "session/new", + json!({"cwd": cwd, "mcpServers": servers}), + ), + }; + output.writes.push(encode(&open)?); + } + ID_SESSION => { + turn.replaying = false; + if let Some(error) = error { + let message = error_message(error); + let (kind, code, diagnostic) = match error_code(error) { + Some(AUTH_REQUIRED_CODE) => ( + ProviderProcessErrorKind::AuthenticationFailed, + "auth_required", + format!( + "{} needs you to sign in before it can start a session ({message}). Sign in with the agent's own CLI, then send the message again.", + turn.agent_label + ), + ), + Some(RESOURCE_NOT_FOUND_CODE) if turn.resume.is_some() => ( + ProviderProcessErrorKind::Unknown, + "session_not_found", + format!( + "{} session `{}` was not found: {message}", + turn.agent_label, + turn.resume.as_deref().unwrap_or_default() + ), + ), + _ => ( + ProviderProcessErrorKind::Unknown, + "session", + format!("{} could not open a session: {message}", turn.agent_label), + ), + }; + fail( + state, + output, + kind, + diagnostic, + code, + DeliveryState::NotSent, + ); + return Ok(()); + } + let session = match turn.resume.clone() { + Some(session) => session, + None => match result.get("sessionId").and_then(Value::as_str) { + Some(session) if !session.trim().is_empty() => session.to_string(), + _ => { + fail( + state, + output, + ProviderProcessErrorKind::Unknown, + format!("{} opened a session without an id", turn.agent_label), + "session", + DeliveryState::NotSent, + ); + return Ok(()); + } + }, + }; + state.result.session_id = Some(session.clone()); + if turn.resume.as_deref() != Some(session.as_str()) { + output.events.push(TurnEvent::SessionStarted { + session_id: session.clone(), + title: None, + }); + } + turn.session_id = Some(session.clone()); + if let Err(message) = configure_session(&session, result, state, turn) { + fail( + state, + output, + ProviderProcessErrorKind::ModelUnavailable, + message, + "configure", + DeliveryState::NotSent, + ); + return Ok(()); + } + advance(state, turn, output)?; + } + ID_SET_MODE | ID_SET_MODEL => { + if let Some(error) = error { + let what = if id == ID_SET_MODE { "mode" } else { "model" }; + fail( + state, + output, + if id == ID_SET_MODEL { + ProviderProcessErrorKind::ModelUnavailable + } else { + ProviderProcessErrorKind::Unknown + }, + format!( + "{} refused the requested {what}: {}", + turn.agent_label, + error_message(error) + ), + "configure", + DeliveryState::NotSent, + ); + return Ok(()); + } + advance(state, turn, output)?; + } + ID_PROMPT => { + output.terminal = true; + if let Some(error) = error { + let auth = error_code(error) == Some(AUTH_REQUIRED_CODE); + fail( + state, + output, + if auth { + ProviderProcessErrorKind::AuthenticationFailed + } else { + ProviderProcessErrorKind::Unknown + }, + format!("{}: {}", turn.agent_label, error_message(error)), + if auth { "auth_required" } else { "prompt" }, + DeliveryState::Accepted, + ); + return Ok(()); + } + match result.get("stopReason").and_then(Value::as_str) { + Some("end_turn") | None => {} + Some("max_tokens") => output.events.push(TurnEvent::Warning { + message: format!("{} stopped at its output token limit", turn.agent_label), + }), + Some("max_turn_requests") => output.events.push(TurnEvent::Warning { + message: format!( + "{} stopped after reaching its limit of model requests for one turn", + turn.agent_label + ), + }), + Some("refusal") => fail( + state, + output, + ProviderProcessErrorKind::Unknown, + format!("{} refused to continue this turn", turn.agent_label), + "refusal", + DeliveryState::Accepted, + ), + Some("cancelled") => fail( + state, + output, + ProviderProcessErrorKind::Unknown, + format!( + "{} cancelled the turn before it completed", + turn.agent_label + ), + "cancelled", + DeliveryState::Accepted, + ), + Some(other) => output.events.push(TurnEvent::Warning { + message: format!("{} stopped: {}", turn.agent_label, bounded(other, 64)), + }), + } + } + _ => {} + } + Ok(()) +} + +/// Queue mode and model selection for a freshly opened session. +fn configure_session( + session: &str, + result: &Value, + state: &mut AdapterState, + turn: &mut TurnState, +) -> std::result::Result<(), String> { + if let Some(mode) = turn.requested_mode.clone() { + let modes = result + .pointer("/modes/availableModes") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + let current = result + .pointer("/modes/currentModeId") + .and_then(Value::as_str); + let available = modes + .iter() + .filter_map(|entry| entry.get("id").and_then(Value::as_str)) + .collect::>(); + if !available.contains(&mode.as_str()) { + return Err(if available.is_empty() { + format!( + "{} has no session modes, so mode `{mode}` cannot be selected", + turn.agent_label + ) + } else { + format!( + "{} has no mode `{mode}`; it offers {}", + turn.agent_label, + available + .iter() + .take(MAX_LISTED_CHOICES) + .copied() + .collect::>() + .join(", ") + ) + }); + } + if current != Some(mode.as_str()) { + turn.pending.push(( + ID_SET_MODE, + request( + ID_SET_MODE, + "session/set_mode", + json!({"sessionId": session, "modeId": mode}), + ), + )); + } + } + let config_model = result + .get("configOptions") + .and_then(Value::as_array) + .and_then(|options| { + options.iter().find(|option| { + option.get("category").and_then(Value::as_str) == Some("model") + || option.get("id").and_then(Value::as_str) == Some("model") + }) + }) + .cloned(); + let unstable_models = result.get("models").cloned(); + state.result.model = config_model + .as_ref() + .and_then(|option| option.get("currentValue")) + .or_else(|| { + unstable_models + .as_ref() + .and_then(|models| models.get("currentModelId")) + }) + .and_then(Value::as_str) + .map(str::to_owned); + let Some(model) = turn.requested_model.clone() else { + return Ok(()); + }; + if let Some(option) = config_model { + let values = select_values(&option); + if !values.iter().any(|value| value == &model) { + return Err(choices_error(&turn.agent_label, &model, &values)); + } + state.result.model = Some(model.clone()); + if option.get("currentValue").and_then(Value::as_str) != Some(model.as_str()) { + let config_id = option.get("id").cloned().unwrap_or(Value::Null); + turn.pending.push(( + ID_SET_MODEL, + request( + ID_SET_MODEL, + "session/set_config_option", + json!({"sessionId": session, "configId": config_id, "value": model}), + ), + )); + } + return Ok(()); + } + if let Some(models) = unstable_models { + let values = models + .get("availableModels") + .and_then(Value::as_array) + .map(|models| { + models + .iter() + .filter_map(|model| model.get("modelId").and_then(Value::as_str)) + .map(str::to_owned) + .collect::>() + }) + .unwrap_or_default(); + if !values.iter().any(|value| value == &model) { + return Err(choices_error(&turn.agent_label, &model, &values)); + } + state.result.model = Some(model.clone()); + if models.get("currentModelId").and_then(Value::as_str) != Some(model.as_str()) { + turn.pending.push(( + ID_SET_MODEL, + request( + ID_SET_MODEL, + "session/set_model", + json!({"sessionId": session, "modelId": model}), + ), + )); + } + return Ok(()); + } + Err(format!( + "{} does not let clients choose a model, so `{model}` cannot be selected; choose the model in the agent's own settings", + turn.agent_label + )) +} + +fn select_values(option: &Value) -> Vec { + let mut values = Vec::new(); + for entry in option + .get("options") + .and_then(Value::as_array) + .into_iter() + .flatten() + { + if let Some(value) = entry.get("value").and_then(Value::as_str) { + values.push(value.to_string()); + } + for nested in entry + .get("options") + .and_then(Value::as_array) + .into_iter() + .flatten() + { + if let Some(value) = nested.get("value").and_then(Value::as_str) { + values.push(value.to_string()); + } + } + } + values +} + +fn choices_error(agent: &str, model: &str, values: &[String]) -> String { + format!( + "{agent} has no model `{model}`; it offers {}", + if values.is_empty() { + "none".to_string() + } else { + values + .iter() + .take(MAX_LISTED_CHOICES) + .cloned() + .collect::>() + .join(", ") + } + ) +} + +/// Send the next queued configuration request, or the prompt once none remain. +fn advance( + state: &mut AdapterState, + turn: &mut TurnState, + output: &mut AdapterOutput, +) -> Result<()> { + if !turn.pending.is_empty() { + let (_, next) = turn.pending.remove(0); + output.writes.push(encode(&next)?); + return Ok(()); + } + if turn.prompt_sent { + return Ok(()); + } + let Some(session) = turn.session_id.clone() else { + return Err(protocol("the ACP session is not open")); + }; + let prompt = state + .extensions + .remove(PROMPT_KEY) + .and_then(|value| value.as_str().map(str::to_owned)) + .unwrap_or_default(); + let mut blocks = Vec::with_capacity(2); + // A continued session already received the instructions. + if turn.resume.is_none() { + if let Some(instructions) = turn.instructions.take() { + blocks.push(json!({"type": "text", "text": instructions})); + } + } + blocks.push(json!({"type": "text", "text": prompt})); + output.writes.push(encode(&request( + ID_PROMPT, + "session/prompt", + json!({"sessionId": session, "prompt": blocks}), + ))?); + turn.prompt_sent = true; + output.turn_submitted = true; + Ok(()) +} + +fn session_update( + value: &Value, + state: &mut AdapterState, + turn: &mut TurnState, + output: &mut AdapterOutput, +) { + let params = value.get("params").unwrap_or(&Value::Null); + if turn.replaying { + return; + } + if let (Some(ours), Some(theirs)) = ( + turn.session_id.as_deref(), + params.get("sessionId").and_then(Value::as_str), + ) { + if ours != theirs { + return; + } + } + let update = params.get("update").unwrap_or(&Value::Null); + match update + .get("sessionUpdate") + .and_then(Value::as_str) + .unwrap_or_default() + { + "agent_message_chunk" => { + if let Some(text) = update.get("content").and_then(block_text) { + if !text.is_empty() { + state.saw_text_delta = true; + state.result.text.push_str(&text); + output.events.push(TurnEvent::TextDelta { text }); + } + } + } + "agent_thought_chunk" => { + if let Some(text) = update.get("content").and_then(block_text) { + if !text.is_empty() { + state + .result + .reasoning + .get_or_insert_with(String::new) + .push_str(&text); + output.events.push(TurnEvent::ReasoningDelta { text }); + } + } + } + "tool_call" => tool_call(update, turn, output, true), + "tool_call_update" => tool_call(update, turn, output, false), + "plan" => { + let entries = update + .get("entries") + .and_then(Value::as_array) + .map(|entries| { + entries + .iter() + .filter_map(|entry| { + let content = entry.get("content").and_then(Value::as_str)?; + Some(PlanEntry { + content: bounded(content, MAX_DIAGNOSTIC_CHARS), + status: match entry.get("status").and_then(Value::as_str) { + Some("completed") => PlanEntryStatus::Completed, + Some("in_progress") => PlanEntryStatus::InProgress, + _ => PlanEntryStatus::Pending, + }, + priority: entry + .get("priority") + .and_then(Value::as_str) + .map(|priority| bounded(priority, 16)), + }) + }) + .take(MAX_TRACKED_TOOLS) + .collect::>() + }) + .unwrap_or_default(); + output.events.push(TurnEvent::PlanUpdated { entries }); + } + "usage_update" => { + let used = update.get("used").and_then(Value::as_u64); + let size = update.get("size").and_then(Value::as_u64); + let cost_usd = update + .get("cost") + .filter(|cost| { + cost.get("currency") + .and_then(Value::as_str) + .is_some_and(|currency| currency.eq_ignore_ascii_case("usd")) + }) + .and_then(|cost| cost.get("amount")) + .and_then(Value::as_f64); + let usage = Usage { + context_window: Some(ContextWindowUsage { + used_tokens: used, + limit_tokens: size, + model: state.result.model.clone(), + estimated: false, + }), + cost_usd, + ..Usage::default() + }; + state + .result + .usage + .context_window + .clone_from(&usage.context_window); + if cost_usd.is_some() { + state.result.usage.cost_usd = cost_usd; + } + output.events.push(TurnEvent::Usage(usage)); + } + "session_info_update" => { + if let Some(title) = update + .get("title") + .and_then(Value::as_str) + .map(str::trim) + .filter(|title| !title.is_empty()) + { + state.result.session_title = Some(bounded(title, MAX_TITLE_CHARS)); + } + } + "notice" => { + if let Some(title) = update.get("title").and_then(Value::as_str) { + let description = update + .get("description") + .and_then(Value::as_str) + .map(|text| format!(": {}", bounded(text, MAX_DIAGNOSTIC_CHARS))) + .unwrap_or_default(); + output.events.push(TurnEvent::Warning { + message: format!("{}{description}", bounded(title, MAX_TITLE_CHARS)), + }); + } + } + // User echoes, command lists, mode and config changes carry nothing a + // turn result needs. + _ => {} + } +} + +fn tool_call(update: &Value, turn: &mut TurnState, output: &mut AdapterOutput, created: bool) { + let Some(id) = update.get("toolCallId").and_then(Value::as_str) else { + return; + }; + let reported_input = update + .get("rawInput") + .filter(|input| !input.is_null()) + .cloned(); + let index = if let Some(index) = turn.tools.iter().position(|tool| tool.id == id) { + index + } else { + // The first title names the tool for the rest of the turn; agents + // often retitle a call with its argument (a file path) later. + let title = update + .get("title") + .and_then(Value::as_str) + .map(|title| bounded(title, MAX_TITLE_CHARS)) + .or_else(|| { + update + .get("kind") + .and_then(Value::as_str) + .map(|kind| format!("{kind} tool")) + }) + .unwrap_or_else(|| "tool".into()); + if turn.tools.len() >= MAX_TRACKED_TOOLS { + turn.tools.remove(0); + } + turn.tools.push(TrackedTool { + id: id.to_string(), + title, + input: None, + output: None, + }); + turn.tools.len() - 1 + }; + if let Some(input) = &reported_input { + let small = + serde_json::to_vec(input).is_ok_and(|bytes| bytes.len() <= MAX_KEPT_INPUT_BYTES); + turn.tools[index].input = small.then(|| input.clone()); + } + // An agent may stream output in `in_progress` updates and finish with a + // bare `status: completed`; keep the latest so the final event is complete. + let reported_output = update + .get("content") + .and_then(tool_content_text) + .or_else(|| update.get("rawOutput").and_then(raw_output_text)) + .map(|text| bounded(&text, MAX_TOOL_OUTPUT_CHARS)); + if let Some(text) = &reported_output { + turn.tools[index].output = Some(text.clone()); + } + let title = turn.tools[index].title.clone(); + let input = reported_input.or_else(|| turn.tools[index].input.clone()); + let status = update.get("status").and_then(Value::as_str); + if status.is_some_and(|status| status != "pending") { + // A tool that moved on is no longer waiting for permission. + turn.open_permissions + .retain(|(_, tool)| tool.as_deref() != Some(id)); + } + match status { + Some("completed" | "failed") => { + turn.tools[index].input = None; + let failed = status == Some("failed"); + let text = turn.tools[index].output.take(); + output.events.push(TurnEvent::ToolCall { + id: Some(id.to_string()), + name: title, + status: if failed { + ToolCallStatus::Failed + } else { + ToolCallStatus::Succeeded + }, + input, + output: if failed { None } else { text.clone() }, + error: if failed { + Some(text.unwrap_or_else(|| "the tool reported an error".into())) + } else { + None + }, + task_id: None, + images: Vec::new(), + }); + } + _ if created || update.get("rawInput").is_some() || update.get("title").is_some() => { + output.events.push(TurnEvent::ToolCall { + id: Some(id.to_string()), + name: title, + status: ToolCallStatus::Started, + input, + output: None, + error: None, + task_id: None, + images: Vec::new(), + }); + } + _ => {} + } +} + +fn block_text(block: &Value) -> Option { + match block.get("type").and_then(Value::as_str)? { + "text" => block.get("text").and_then(Value::as_str).map(str::to_owned), + "resource_link" => block + .get("uri") + .or_else(|| block.get("name")) + .and_then(Value::as_str) + .map(str::to_owned), + "resource" => block + .pointer("/resource/text") + .or_else(|| block.pointer("/resource/uri")) + .and_then(Value::as_str) + .map(str::to_owned), + "image" => Some("[image]".into()), + "audio" => Some("[audio]".into()), + _ => None, + } +} + +fn tool_content_text(content: &Value) -> Option { + let parts = content + .as_array()? + .iter() + .filter_map(|item| match item.get("type").and_then(Value::as_str)? { + "content" => item.get("content").and_then(block_text), + "diff" => { + let path = item.get("path").and_then(Value::as_str).unwrap_or("a file"); + let new_text = item + .get("newText") + .and_then(Value::as_str) + .unwrap_or_default(); + Some(if item.get("oldText").is_some_and(|old| !old.is_null()) { + format!("Edited {path}\n{new_text}") + } else { + format!("Created {path}\n{new_text}") + }) + } + _ => None, + }) + .collect::>(); + (!parts.is_empty()).then(|| parts.join("\n")) +} + +fn raw_output_text(raw: &Value) -> Option { + match raw { + Value::Null => None, + Value::String(text) => Some(text.clone()), + other => serde_json::to_string(other).ok(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn agents() -> AcpAgents { + let agents = AcpAgents::new(); + agents + .insert(AcpAgent::new("gemini", "Gemini CLI", "gemini", ["--acp"])) + .unwrap(); + agents + } + + fn turn(prompt: &str) -> TurnRequest { + let mut request = TurnRequest::new(Provider::Acp, std::env::temp_dir(), prompt); + request + .harness_options + .insert(AGENT_OPTION.into(), "gemini".into()); + request + } + + fn started(adapter: &Acp, request: &TurnRequest) -> AdapterState { + let mut state = AdapterState::default(); + adapter.prepare_turn(request, &mut state).unwrap(); + state + } + + fn feed(adapter: &Acp, state: &mut AdapterState, frame: Value) -> AdapterOutput { + adapter.parse_line(&frame.to_string(), state).unwrap() + } + + fn written(output: &AdapterOutput) -> Vec { + output + .writes + .iter() + .flat_map(|bytes| { + String::from_utf8_lossy(bytes) + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect::>() + }) + .collect() + } + + fn initialized(capabilities: Value) -> Value { + json!({"jsonrpc": "2.0", "id": ID_INITIALIZE, "result": { + "protocolVersion": 1, "agentCapabilities": capabilities, "authMethods": [] + }}) + } + + #[test] + fn launches_the_selected_agent_and_starts_with_initialize() { + let adapter = Acp::new(agents()); + let spec = adapter.command(&turn("hi")).unwrap(); + assert_eq!(spec.program, PathBuf::from("gemini")); + assert_eq!(spec.args, vec![std::ffi::OsString::from("--acp")]); + assert!(spec.interactive_stdin); + let first: Value = serde_json::from_slice(spec.initial_stdin.as_ref().unwrap()).unwrap(); + assert_eq!(first["method"], "initialize"); + assert_eq!(first["params"]["protocolVersion"], 1); + assert_eq!(first["params"]["clientCapabilities"]["terminal"], false); + } + + #[test] + fn an_unknown_or_missing_agent_is_rejected_before_spawning() { + let adapter = Acp::new(agents()); + let mut request = turn("hi"); + request.harness_options.clear(); + assert!(adapter + .command(&request) + .unwrap_err() + .to_string() + .contains("choose which ACP agent")); + request + .harness_options + .insert(AGENT_OPTION.into(), "copilot".into()); + assert!(adapter + .command(&request) + .unwrap_err() + .to_string() + .contains("`copilot`")); + } + + #[test] + fn a_tool_keeps_its_name_and_input_when_later_updates_omit_them() { + // `opencode acp` names a call by its tool, then retitles it with the + // file path and completes it without `rawInput`. + let adapter = Acp::new(agents()); + let (mut state, _) = opened(&adapter, &turn("hi"), json!({"sessionId": "s"})); + let update = |update: Value| { + json!({"jsonrpc": "2.0", "method": "session/update", + "params": {"sessionId": "s", "update": update}}) + }; + feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call", "toolCallId": "w", "title": "write", "kind": "edit", "status": "pending", "rawInput": {}}), + ), + ); + feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call_update", "toolCallId": "w", "status": "in_progress", + "rawInput": {"filePath": "/repo/hello.txt", "content": "hi"}}), + ), + ); + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call_update", "toolCallId": "w", "status": "completed", + "title": "hello.txt", "content": [{"type": "content", "content": {"type": "text", "text": "Wrote file successfully."}}]}), + ), + ); + let Some(TurnEvent::ToolCall { + name, + input, + status, + .. + }) = out.events.first() + else { + panic!("expected the completed call"); + }; + assert_eq!(name, "write"); + assert_eq!(*status, ToolCallStatus::Succeeded); + assert_eq!(input.as_ref().unwrap()["filePath"], "/repo/hello.txt"); + } + + #[test] + fn registered_agents_are_advertised_as_the_agent_control() { + let adapter = Acp::new(agents()); + let groups = adapter.control_groups(); + assert_eq!(groups[0].id, AGENT_OPTION); + assert_eq!(groups[0].kind, HarnessControlKind::Agent); + assert_eq!(groups[0].options[0].id, "gemini"); + assert_eq!(groups[0].options[0].description, "gemini --acp"); + // The registry is shared: a later registration is advertised at once. + adapter + .agents() + .insert(AcpAgent::new("kimi", "Kimi CLI", "kimi", ["acp"])) + .unwrap(); + assert_eq!(adapter.control_groups()[0].options.len(), 2); + } + + #[test] + fn agent_ids_are_validated() { + assert!(AcpAgent::new("Gemini", "x", "gemini", ["--acp"]) + .validate() + .is_err()); + assert!(AcpAgent::new("", "x", "gemini", ["--acp"]) + .validate() + .is_err()); + assert!(AcpAgent::new("-x", "x", "gemini", ["--acp"]) + .validate() + .is_err()); + assert!(AcpAgent::new("kimi-cli.2", "x", "kimi", ["acp"]) + .validate() + .is_ok()); + assert!(AcpAgent::new("ok", "x", "", Vec::::new()) + .validate() + .is_err()); + } + + #[test] + fn a_new_session_streams_until_the_prompt_returns() { + let adapter = Acp::new(agents()); + let mut request = turn("fix it"); + request.launch_context.system_prompt_append = Some("Fleet rules".into()); + let mut state = started(&adapter, &request); + let out = feed(&adapter, &mut state, initialized(json!({}))); + let frames = written(&out); + assert_eq!(frames[0]["method"], "session/new"); + assert_eq!(frames[0]["params"]["mcpServers"], json!([])); + + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SESSION, "result": {"sessionId": "s-1"}}), + ); + assert!( + matches!(&out.events[0], TurnEvent::SessionStarted { session_id, .. } if session_id == "s-1") + ); + let frames = written(&out); + assert_eq!(frames[0]["method"], "session/prompt"); + let blocks = frames[0]["params"]["prompt"].as_array().unwrap(); + assert_eq!(blocks[0]["text"], "Fleet rules"); + assert_eq!(blocks[1]["text"], "fix it"); + assert!(out.turn_submitted); + + let update = |update: Value| { + json!({"jsonrpc": "2.0", "method": "session/update", + "params": {"sessionId": "s-1", "update": update}}) + }; + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "agent_thought_chunk", "content": {"type": "text", "text": "thinking"}}), + ), + ); + assert!(matches!(&out.events[0], TurnEvent::ReasoningDelta { text } if text == "thinking")); + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call", "toolCallId": "t1", "title": "Read main.rs", "kind": "read", "status": "pending", "rawInput": {"path": "main.rs"}}), + ), + ); + assert!( + matches!(&out.events[0], TurnEvent::ToolCall { status: ToolCallStatus::Started, name, .. } if name == "Read main.rs") + ); + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call_update", "toolCallId": "t1", "status": "completed", + "content": [{"type": "content", "content": {"type": "text", "text": "fn main() {}"}}]}), + ), + ); + assert!( + matches!(&out.events[0], TurnEvent::ToolCall { status: ToolCallStatus::Succeeded, name, output: Some(text), .. } if name == "Read main.rs" && text == "fn main() {}") + ); + let out = feed( + &adapter, + &mut state, + update(json!({"sessionUpdate": "plan", "entries": [ + {"content": "Read", "priority": "high", "status": "completed"}, + {"content": "Fix", "priority": "medium", "status": "in_progress"}]})), + ); + assert!( + matches!(&out.events[0], TurnEvent::PlanUpdated { entries } if entries.len() == 2 && entries[1].status == PlanEntryStatus::InProgress) + ); + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "Done."}}), + ), + ); + assert!(matches!(&out.events[0], TurnEvent::TextDelta { text } if text == "Done.")); + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "usage_update", "used": 1200, "size": 200_000, "cost": {"amount": 0.01, "currency": "USD"}}), + ), + ); + assert!( + matches!(&out.events[0], TurnEvent::Usage(usage) if usage.context_window.as_ref().unwrap().used_tokens == Some(1200)) + ); + // Another session's updates are not this turn's. + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "method": "session/update", + "params": {"sessionId": "other", "update": {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "x"}}}}), + ); + assert_eq!(out.events.len(), 0); + + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_PROMPT, "result": {"stopReason": "end_turn"}}), + ); + assert!(out.terminal); + assert!(state.terminal_failure.is_none()); + assert_eq!(state.result.text, "Done."); + assert_eq!(state.result.session_id.as_deref(), Some("s-1")); + assert_eq!(state.result.usage.cost_usd, Some(0.01)); + } + + #[test] + fn resume_prefers_session_resume_and_skips_instructions() { + let adapter = Acp::new(agents()); + let mut request = turn("again"); + request.session_id = Some("s-1".into()); + request.launch_context.system_prompt_append = Some("Fleet rules".into()); + let mut state = started(&adapter, &request); + let out = feed( + &adapter, + &mut state, + initialized(json!({"loadSession": true, "sessionCapabilities": {"resume": {}}})), + ); + assert_eq!(written(&out)[0]["method"], "session/resume"); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SESSION, "result": {}}), + ); + assert!( + out.events.is_empty(), + "a resumed session is not announced again" + ); + let prompt = &written(&out)[0]; + assert_eq!(prompt["params"]["sessionId"], "s-1"); + assert_eq!(prompt["params"]["prompt"].as_array().unwrap().len(), 1); + } + + #[test] + fn session_load_replay_is_dropped() { + let adapter = Acp::new(agents()); + let mut request = turn("again"); + request.session_id = Some("s-1".into()); + let mut state = started(&adapter, &request); + let out = feed( + &adapter, + &mut state, + initialized(json!({"loadSession": true})), + ); + assert_eq!(written(&out)[0]["method"], "session/load"); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "method": "session/update", + "params": {"sessionId": "s-1", "update": {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "old"}}}}), + ); + assert_eq!(out.events.len(), 0); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SESSION, "result": null}), + ); + assert_eq!(written(&out)[0]["method"], "session/prompt"); + assert_eq!(state.result.text.len(), 0); + } + + #[test] + fn an_agent_that_cannot_resume_starts_fresh_and_says_so() { + let adapter = Acp::new(agents()); + let mut request = turn("again"); + request.session_id = Some("s-1".into()); + request.launch_context.system_prompt_append = Some("Fleet rules".into()); + let mut state = started(&adapter, &request); + let out = feed(&adapter, &mut state, initialized(json!({}))); + assert!( + matches!(&out.events[0], TurnEvent::Warning { message } if message.contains("cannot resume")) + ); + assert_eq!(written(&out)[0]["method"], "session/new"); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SESSION, "result": {"sessionId": "s-2"}}), + ); + assert!( + matches!(&out.events[0], TurnEvent::SessionStarted { session_id, .. } if session_id == "s-2") + ); + assert_eq!( + written(&out)[0]["params"]["prompt"] + .as_array() + .unwrap() + .len(), + 2 + ); + } + + #[test] + fn auth_required_fails_before_the_prompt_with_a_sign_in_hint() { + let adapter = Acp::new(agents()); + let mut state = started(&adapter, &turn("hi")); + feed(&adapter, &mut state, initialized(json!({}))); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SESSION, + "error": {"code": -32000, "message": "Authentication required"}}), + ); + assert!(out.terminal); + let failure = state.terminal_failure.as_ref().unwrap(); + assert_eq!(failure.kind, ProviderProcessErrorKind::AuthenticationFailed); + assert_eq!(failure.delivery, DeliveryState::NotSent); + assert_eq!(failure.provider_code.as_deref(), Some("acp::auth_required")); + assert!(failure + .diagnostic + .contains("Gemini CLI needs you to sign in")); + } + + #[test] + fn a_missing_resumed_session_fails_instead_of_starting_over() { + let adapter = Acp::new(agents()); + let mut request = turn("again"); + request.session_id = Some("gone".into()); + let mut state = started(&adapter, &request); + feed( + &adapter, + &mut state, + initialized(json!({"loadSession": true})), + ); + feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SESSION, + "error": {"code": -32002, "message": "Resource not found"}}), + ); + let failure = state.terminal_failure.as_ref().unwrap(); + assert_eq!( + failure.provider_code.as_deref(), + Some("acp::session_not_found") + ); + } + + #[test] + fn a_newer_protocol_version_is_refused() { + let adapter = Acp::new(agents()); + let mut state = started(&adapter, &turn("hi")); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": 2}}), + ); + assert!(out.terminal); + assert!(state + .terminal_failure + .as_ref() + .unwrap() + .diagnostic + .contains("version 2")); + } + + fn opened( + adapter: &Acp, + request: &TurnRequest, + session: Value, + ) -> (AdapterState, AdapterOutput) { + let mut state = started(adapter, request); + feed(adapter, &mut state, initialized(json!({}))); + let out = feed( + adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SESSION, "result": session}), + ); + (state, out) + } + + #[test] + fn a_custom_mode_and_model_are_selected_before_the_prompt() { + let adapter = Acp::new(agents()); + let mut request = turn("hi"); + request.permission_mode = PermissionMode::Custom("architect".into()); + request.model = Some("pro".into()); + let session = json!({"sessionId": "s", + "modes": {"currentModeId": "code", "availableModes": [{"id": "code", "name": "Code"}, {"id": "architect", "name": "Architect"}]}, + "configOptions": [{"id": "model", "name": "Model", "category": "model", "type": "select", "currentValue": "flash", + "options": [{"value": "flash", "name": "Flash"}, {"value": "pro", "name": "Pro"}]}]}); + let (mut state, out) = opened(&adapter, &request, session); + assert_eq!(written(&out)[0]["method"], "session/set_mode"); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SET_MODE, "result": {}}), + ); + let frame = &written(&out)[0]; + assert_eq!(frame["method"], "session/set_config_option"); + assert_eq!(frame["params"]["value"], "pro"); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_SET_MODEL, "result": {"configOptions": []}}), + ); + assert_eq!(written(&out)[0]["method"], "session/prompt"); + assert_eq!(state.result.model.as_deref(), Some("pro")); + } + + #[test] + fn an_unknown_model_fails_naming_the_choices() { + let adapter = Acp::new(agents()); + let mut request = turn("hi"); + request.model = Some("ultra".into()); + let session = json!({"sessionId": "s", "models": {"currentModelId": "a", + "availableModels": [{"modelId": "a", "name": "A"}, {"modelId": "b", "name": "B"}]}}); + let (state, out) = opened(&adapter, &request, session); + assert!(out.terminal); + let failure = state.terminal_failure.as_ref().unwrap(); + assert_eq!(failure.kind, ProviderProcessErrorKind::ModelUnavailable); + assert!( + failure.diagnostic.contains("offers a, b"), + "{}", + failure.diagnostic + ); + } + + fn permission_request(kind: &str) -> Value { + json!({"jsonrpc": "2.0", "id": 7, "method": "session/request_permission", "params": { + "sessionId": "s", + "toolCall": {"toolCallId": "t1", "title": "Run npm test", "kind": kind, "rawInput": {"command": "npm test"}}, + "options": [ + {"optionId": "yes", "name": "Allow", "kind": "allow_once"}, + {"optionId": "always", "name": "Always allow", "kind": "allow_always"}, + {"optionId": "no", "name": "Reject", "kind": "reject_once"}]}}) + } + + #[test] + fn permission_requests_become_approvals_and_round_trip() { + let adapter = Acp::new(agents()); + let (mut state, _) = opened(&adapter, &turn("hi"), json!({"sessionId": "s"})); + let out = feed(&adapter, &mut state, permission_request("execute")); + let Some(InteractionRequest::Approval { request, original }) = out.interaction else { + panic!("expected an approval"); + }; + assert_eq!(request.tool_name, "Run npm test"); + assert_eq!(request.input["command"], "npm test"); + // Reads after " wants to". + assert_eq!(request.description.as_deref(), Some("run `npm test`")); + let answer = |decision| { + let bytes = adapter + .approval_response(&request, &original, decision) + .unwrap() + .unwrap(); + serde_json::from_slice::(&bytes).unwrap() + }; + assert_eq!( + answer(ApprovalDecision::Allow)["result"]["outcome"]["optionId"], + "yes" + ); + assert_eq!( + answer(ApprovalDecision::AllowForSession)["result"]["outcome"]["optionId"], + "always" + ); + let denied = answer(ApprovalDecision::Deny { reason: None }); + assert_eq!(denied["id"], 7); + assert_eq!(denied["result"]["outcome"]["optionId"], "no"); + } + + #[test] + fn a_permission_request_says_what_it_would_do() { + let located = json!({"locations": [{"path": "/repo/src/main.rs"}]}); + assert_eq!( + permission_action("edit", &located, None), + "edit /repo/src/main.rs" + ); + assert_eq!( + permission_action("read", &Value::Null, Some(&json!({"filePath": "a.txt"}))), + "read a.txt" + ); + assert_eq!( + permission_action("delete", &Value::Null, None), + "delete files" + ); + assert_eq!( + permission_action("execute", &Value::Null, None), + "run a command" + ); + assert_eq!( + permission_action( + "fetch", + &Value::Null, + Some(&json!({"url": "https://example.com"})) + ), + "fetch https://example.com" + ); + assert_eq!( + permission_action("think", &Value::Null, None), + "use this tool" + ); + } + + #[test] + fn permission_modes_answer_what_they_cover() { + let adapter = Acp::new(agents()); + let mut request = turn("hi"); + request.permission_mode = PermissionMode::AcceptEdits; + let (mut state, _) = opened(&adapter, &request, json!({"sessionId": "s"})); + let out = feed(&adapter, &mut state, permission_request("edit")); + assert!(out.interaction.is_none()); + assert_eq!(written(&out)[0]["result"]["outcome"]["optionId"], "yes"); + let out = feed(&adapter, &mut state, permission_request("execute")); + assert!( + out.interaction.is_some(), + "commands still ask under AcceptEdits" + ); + + request.permission_mode = PermissionMode::FullAccess; + let (mut state, _) = opened(&adapter, &request, json!({"sessionId": "s"})); + let out = feed(&adapter, &mut state, permission_request("execute")); + assert!(out.interaction.is_none()); + assert_eq!(written(&out)[0]["result"]["outcome"]["optionId"], "yes"); + + request.permission_mode = PermissionMode::Plan; + let mut state = AdapterState::default(); + assert!(adapter.prepare_turn(&request, &mut state).is_err()); + } + + #[test] + fn cancellation_cancels_the_session_and_open_permissions() { + let adapter = Acp::new(agents()); + let (mut state, _) = opened(&adapter, &turn("hi"), json!({"sessionId": "s"})); + feed(&adapter, &mut state, permission_request("execute")); + let bytes = adapter.interrupt_request(&state).unwrap(); + let frames = String::from_utf8(bytes) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(frames[0]["method"], "session/cancel"); + assert_eq!(frames[0]["params"]["sessionId"], "s"); + assert_eq!(frames[1]["id"], 7); + assert_eq!(frames[1]["result"]["outcome"]["outcome"], "cancelled"); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": ID_PROMPT, "result": {"stopReason": "cancelled"}}), + ); + assert!(out.terminal); + assert_eq!(state.result.status, RunStatus::Failed); + } + + #[test] + fn unsupported_client_methods_get_an_error_reply() { + let adapter = Acp::new(agents()); + let (mut state, _) = opened(&adapter, &turn("hi"), json!({"sessionId": "s"})); + let out = feed( + &adapter, + &mut state, + json!({"jsonrpc": "2.0", "id": "x", "method": "fs/read_text_file", + "params": {"sessionId": "s", "path": "/etc/hosts"}}), + ); + let frame = &written(&out)[0]; + assert_eq!(frame["id"], "x"); + assert_eq!(frame["error"]["code"], -32601); + } + + #[test] + fn stdout_noise_warns_once_without_failing() { + let adapter = Acp::new(agents()); + let mut state = started(&adapter, &turn("hi")); + let out = adapter + .parse_line("Loaded cached credentials.", &mut state) + .unwrap(); + assert_eq!(out.events.len(), 1); + let out = adapter.parse_line("more noise", &mut state).unwrap(); + assert_eq!(out.events.len(), 0); + assert!(!out.terminal); + } + + #[test] + fn mcp_servers_are_sent_with_resolved_values() { + let adapter = Acp::new(agents()); + let mut request = turn("hi"); + request + .environment + .insert("FLEET_TOKEN".into(), crate::SecretString::new("secret")); + request.launch_context.mcp_servers.insert( + "fleet".into(), + McpServerConfig::Stdio { + command: "fleet-mcp".into(), + args: vec!["serve".into()], + environment_from: BTreeMap::from([("TOKEN".into(), "FLEET_TOKEN".into())]), + }, + ); + request.launch_context.mcp_servers.insert( + "web".into(), + McpServerConfig::Http { + url: "http://127.0.0.1:1/mcp".into(), + headers_from: BTreeMap::new(), + }, + ); + let mut state = started(&adapter, &request); + let out = feed(&adapter, &mut state, initialized(json!({}))); + assert!( + matches!(&out.events[0], TurnEvent::Warning { message } if message.contains("HTTP MCP")) + ); + let servers = written(&out)[0]["params"]["mcpServers"].clone(); + assert_eq!(servers.as_array().unwrap().len(), 1); + assert_eq!(servers[0]["name"], "fleet"); + assert_eq!( + servers[0]["env"][0], + json!({"name": "TOKEN", "value": "secret"}) + ); + } + + #[test] + fn a_tool_completed_without_content_keeps_the_output_it_streamed() { + // Some agents stream the result in `in_progress` updates and finish + // with a bare status. + let adapter = Acp::new(agents()); + let (mut state, _) = opened(&adapter, &turn("hi"), json!({"sessionId": "s"})); + let update = |update: Value| { + json!({"jsonrpc": "2.0", "method": "session/update", + "params": {"sessionId": "s", "update": update}}) + }; + feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call", "toolCallId": "t", "title": "Run tests", "kind": "execute", "status": "pending"}), + ), + ); + feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call_update", "toolCallId": "t", "status": "in_progress", + "content": [{"type": "content", "content": {"type": "text", "text": "first"}}]}), + ), + ); + feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call_update", "toolCallId": "t", "status": "in_progress", + "content": [{"type": "content", "content": {"type": "text", "text": "12 passed, 1 failed"}}]}), + ), + ); + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call_update", "toolCallId": "t", "status": "failed"}), + ), + ); + let Some(TurnEvent::ToolCall { + status, + output, + error, + .. + }) = out.events.first() + else { + panic!("expected the failed call"); + }; + assert_eq!(*status, ToolCallStatus::Failed); + assert_eq!(output, &None); + assert_eq!(error.as_deref(), Some("12 passed, 1 failed")); + + // The kept output belongs to the call that finished. + feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call", "toolCallId": "u", "title": "Read", "kind": "read", "status": "pending"}), + ), + ); + let out = feed( + &adapter, + &mut state, + update( + json!({"sessionUpdate": "tool_call_update", "toolCallId": "u", "status": "completed"}), + ), + ); + assert!(matches!( + out.events.first(), + Some(TurnEvent::ToolCall { + status: ToolCallStatus::Succeeded, + output: None, + .. + }) + )); + } + + #[test] + fn a_relative_working_directory_is_refused_before_spawning() { + let adapter = Acp::new(agents()); + let mut request = turn("hi"); + request.working_directory = PathBuf::from("workspace"); + let error = adapter + .prepare_turn(&request, &mut AdapterState::default()) + .unwrap_err() + .to_string(); + assert!(error.contains("absolute working directory"), "{error}"); + assert!(error.contains("`workspace`"), "{error}"); + + // Remote transports use their own path syntax. + for absolute in [ + std::env::temp_dir(), + "/workspace/a".into(), + "C:\\repo".into(), + "\\\\host\\share".into(), + ] { + assert!(is_absolute_directory(&absolute), "{}", absolute.display()); + } + for relative in [".", "workspace", "../repo", "C:repo", ""] { + assert!(!is_absolute_directory(Path::new(relative)), "{relative}"); + } + } + + #[test] + fn an_answered_permission_is_not_cancelled_again() { + let adapter = Acp::new(agents()); + let (mut state, _) = opened(&adapter, &turn("hi"), json!({"sessionId": "s"})); + let out = feed(&adapter, &mut state, permission_request("execute")); + let Some(InteractionRequest::Approval { original, .. }) = out.interaction else { + panic!("expected an approval"); + }; + // Cancelling while the user is still deciding refuses the request. + let frames = |state: &AdapterState| { + String::from_utf8(adapter.interrupt_request(state).unwrap()) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>() + }; + assert_eq!(frames(&state).len(), 2); + // Once the decision was sent, the agent has its answer; only the + // session is cancelled. + adapter.interaction_answered(&original, &mut state); + let frames = frames(&state); + assert_eq!(frames.len(), 1); + assert_eq!(frames[0]["method"], "session/cancel"); + } +} diff --git a/src/providers/mod.rs b/src/providers/mod.rs index 0927cba..f9e4450 100644 --- a/src/providers/mod.rs +++ b/src/providers/mod.rs @@ -1,5 +1,7 @@ //! Built-in coding-agent CLI adapters. +#[cfg(feature = "acp")] +pub mod acp; #[cfg(feature = "claude")] mod claude; #[cfg(feature = "codex")] @@ -15,6 +17,8 @@ mod opencode_serve; #[cfg(feature = "pi")] mod pi; +#[cfg(feature = "acp")] +pub use acp::{Acp, AcpAgent, AcpAgents}; #[cfg(feature = "claude")] pub use claude::Claude; #[cfg(feature = "codex")] diff --git a/src/runtime.rs b/src/runtime.rs index 756b225..e15e068 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -30,6 +30,10 @@ use crate::{ const DEFAULT_MAX_PROMPT_BYTES: usize = 256 * 1024; const DEFAULT_MAX_EVENT_LINE_BYTES: usize = 2 * 1024 * 1024; const STDERR_TAIL_BYTES: usize = 32 * 1024; +/// How long a finished process's stderr may stay open before the reader is +/// stopped. A descendant that outlived the process can hold the pipe, and the +/// result of a finished turn or probe must not wait for it. +const STDERR_DRAIN_GRACE: Duration = Duration::from_secs(2); const MAX_PROVIDER_CODE_CHARS: usize = 256; const CATALOG_MAX_BYTES: usize = 1024 * 1024; const CATALOG_TIMEOUT: Duration = Duration::from_secs(15); @@ -2237,6 +2241,30 @@ fn validate_attachments(request: &TurnRequest) -> Result<()> { Ok(()) } +/// Readiness of one adapter, with the selectable executables that resolved. +struct AdapterInspection { + readiness: crate::ProviderReadiness, + /// Ids of the installed [`crate::SelectableExecutable`]s, or `None` for an + /// adapter that runs one fixed executable. + installed_choices: Option>, +} + +/// Keep only the installed agents in `Agent` control groups, so an +/// application offers what can actually run, and drop a group left empty. +fn retain_installed_agents(groups: &mut Vec, installed: &[String]) { + for group in groups + .iter_mut() + .filter(|group| group.kind == crate::HarnessControlKind::Agent) + { + group + .options + .retain(|option| installed.contains(&option.id)); + } + groups.retain(|group| { + group.kind != crate::HarnessControlKind::Agent || !group.options.is_empty() + }); +} + /// Bounded process runtime shared by all provider adapters. #[derive(Clone)] pub struct AgentRuntime { @@ -2317,20 +2345,135 @@ impl AgentRuntime { } /// Inspect one compiled adapter inside the configured execution transport. + /// + /// An adapter that runs one of several registered executables, such as + /// the ACP adapter, is installed when at least one of them resolves; + /// [`crate::ProviderReadiness::detail`] names each. pub async fn readiness(&self, provider: Provider) -> Result { let adapter = self .adapters .get(&provider) .ok_or(RuntimeError::AdapterUnavailable { provider })?; - self.transport - .readiness(TransportReadinessRequest { - provider, - program: adapter.executable(), - }) + self.inspect_adapter(provider, adapter.as_ref()) .await + .map(|inspection| inspection.readiness) .map_err(|source| RuntimeError::Transport { provider, source }) } + /// Resolve an adapter's executable, or each executable it can select + /// among, inside the configured transport. + async fn inspect_adapter( + &self, + provider: Provider, + adapter: &dyn AgentAdapter, + ) -> crate::TransportResult { + let Some(choices) = adapter.selectable_executables() else { + let readiness = self + .transport + .readiness(TransportReadinessRequest { + provider, + program: adapter.executable(), + }) + .await?; + return Ok(AdapterInspection { + readiness, + installed_choices: None, + }); + }; + if choices.is_empty() { + return Ok(AdapterInspection { + readiness: crate::ProviderReadiness { + provider, + installed: false, + executable: None, + version: None, + detail: format!( + "No {provider} agent is registered; register one to make this provider available." + ), + }, + installed_choices: Some(Vec::new()), + }); + } + let mut tasks = tokio::task::JoinSet::new(); + for (index, choice) in choices.iter().enumerate() { + let transport = Arc::clone(&self.transport); + let request = TransportReadinessRequest { + provider, + program: choice.executable.clone(), + }; + tasks.spawn(async move { (index, transport.readiness(request).await) }); + } + let mut outcomes = choices.iter().map(|_| None).collect::>(); + while let Some(joined) = tasks.join_next().await { + if let Ok((index, outcome)) = joined { + outcomes[index] = Some(outcome); + } + } + let mut installed = Vec::new(); + let mut notes = Vec::with_capacity(choices.len()); + let mut first_error = None; + let mut inspected = 0usize; + for (choice, outcome) in choices.iter().zip(outcomes) { + match outcome { + Some(Ok(readiness)) if readiness.installed => { + inspected += 1; + let version = readiness + .version + .as_deref() + .and_then(|version| version.lines().next()) + .map(|version| version.chars().take(60).collect::()) + .filter(|version| !version.is_empty()); + notes.push(match version { + Some(version) => format!("{}: installed ({version})", choice.label), + None => format!("{}: installed", choice.label), + }); + installed.push((choice.id.clone(), readiness)); + } + Some(Ok(_)) => { + inspected += 1; + notes.push(format!( + "{}: `{}` was not found", + choice.label, + choice.executable.display() + )); + } + Some(Err(error)) => { + notes.push(format!( + "{}: could not be inspected ({})", + choice.label, + error.message.chars().take(160).collect::() + )); + first_error.get_or_insert(error); + } + None => notes.push(format!("{}: inspection did not finish", choice.label)), + } + } + if inspected == 0 { + if let Some(error) = first_error { + return Err(error); + } + } + let installed_choices = installed.iter().map(|(id, _)| id.clone()).collect(); + let single = (installed.len() == 1).then(|| installed[0].1.clone()); + Ok(AdapterInspection { + readiness: crate::ProviderReadiness { + provider, + installed: !installed.is_empty(), + // One resolved path and version identify an executable only + // when exactly one of the registered ones is installed. + executable: single.as_ref().and_then(|found| found.executable.clone()), + version: single.and_then(|found| found.version), + detail: format!( + "{} of {} registered {provider} agent(s) installed. {}.", + installed.len(), + choices.len(), + notes.join("; ") + ), + }, + installed_choices: Some(installed_choices), + }) + } + /// Autocomplete working directories inside this runtime's transport. pub async fn suggest_working_directories( &self, @@ -2678,6 +2821,15 @@ impl AgentRuntime { adapter: Arc, working_directory: &std::path::Path, ) -> crate::TransportResult> { + if provider == Provider::Acp { + return Err(TransportError::new( + TransportErrorKind::Unsupported, + self.transport.name(), + "discover_mcp", + "ACP agents have no common MCP listing command", + false, + )); + } if provider == Provider::Pi { // `pi mcp list` connects to every configured server and exits // nonzero when any of them is unreachable, so it cannot serve as a @@ -2764,7 +2916,7 @@ impl AgentRuntime { }; let stderr_task = process .take_stderr() - .map(|stderr| tokio::spawn(crate::process::bounded_stderr(stderr, STDERR_TAIL_BYTES))); + .map(|stderr| crate::process::StderrCapture::spawn(stderr, STDERR_TAIL_BYTES)); let run = async { let read_output = async { let mut output = Vec::new(); @@ -2790,17 +2942,13 @@ impl AgentRuntime { match result { Ok(Ok((output, status))) if status.success => { if let Some(task) = stderr_task { - let _ = task.await; + task.abort(); } Ok(output) } Ok(Ok((_, _))) => { let stderr = match stderr_task { - Some(task) => task - .await - .ok() - .and_then(std::result::Result::ok) - .unwrap_or_default(), + Some(task) => task.finish(STDERR_DRAIN_GRACE).await.unwrap_or_default(), None => String::new(), }; Err(extension_transport_error( @@ -2911,7 +3059,7 @@ impl AgentRuntime { }; let stderr_task = process .take_stderr() - .map(|stderr| tokio::spawn(crate::process::bounded_stderr(stderr, STDERR_TAIL_BYTES))); + .map(|stderr| crate::process::StderrCapture::spawn(stderr, STDERR_TAIL_BYTES)); let read = async { let mut reader = BufReader::new(stdout).lines(); let mut lines = Vec::new(); @@ -2995,6 +3143,7 @@ impl AgentRuntime { Provider::Codex, Provider::OpenCode, Provider::Pi, + Provider::Acp, ] { if self.adapters.contains_key(&provider) { let runtime = self.clone(); @@ -3037,31 +3186,30 @@ impl AgentRuntime { if !capabilities.process_tree_termination { limitations.push(HarnessLimitation::ProcessTreeTerminationUnavailable); } - match self - .transport - .readiness(TransportReadinessRequest { - provider, - program: adapter.executable(), - }) - .await - { - Ok(readiness) => { - let (authentication, control_groups, models, account_usage) = if readiness.installed - { - let authentication = - self.probe_harness_authentication(Arc::clone(adapter), context.clone()); - let metadata = self.probe_harness_metadata(Arc::clone(adapter), context); - let (authentication, metadata) = tokio::join!(authentication, metadata); - let (control_groups, models, account_usage) = metadata; - (authentication, control_groups, models, account_usage) - } else { - ( - HarnessAuthentication::unknown("executable_not_installed"), - adapter.control_groups(), - HarnessModelCatalog::unsupported("executable_not_installed"), - None, - ) - }; + match self.inspect_adapter(provider, adapter.as_ref()).await { + Ok(AdapterInspection { + readiness, + installed_choices, + }) => { + let (authentication, mut control_groups, models, account_usage) = + if readiness.installed { + let authentication = + self.probe_harness_authentication(Arc::clone(adapter), context.clone()); + let metadata = self.probe_harness_metadata(Arc::clone(adapter), context); + let (authentication, metadata) = tokio::join!(authentication, metadata); + let (control_groups, models, account_usage) = metadata; + (authentication, control_groups, models, account_usage) + } else { + ( + HarnessAuthentication::unknown("executable_not_installed"), + adapter.control_groups(), + HarnessModelCatalog::unsupported("executable_not_installed"), + None, + ) + }; + if let Some(installed) = &installed_choices { + retain_installed_agents(&mut control_groups, installed); + } HarnessReadiness { provider, status: if !readiness.installed { @@ -3352,7 +3500,7 @@ impl AgentRuntime { }; let stderr_task = process .take_stderr() - .map(|stderr| tokio::spawn(crate::process::bounded_stderr(stderr, STDERR_TAIL_BYTES))); + .map(|stderr| crate::process::StderrCapture::spawn(stderr, STDERR_TAIL_BYTES)); let mut seen = Vec::with_capacity(expected.len()); let read_lines = async { let mut reader = BufReader::new(stdout).lines(); @@ -3424,11 +3572,7 @@ impl AgentRuntime { process.wait().await.ok() }; let stderr = match stderr_task { - Some(task) => task - .await - .ok() - .and_then(std::result::Result::ok) - .unwrap_or_default(), + Some(task) => task.finish(STDERR_DRAIN_GRACE).await.unwrap_or_default(), None => String::new(), }; if status.is_some_and(|status| !status.success) { @@ -5354,7 +5498,9 @@ impl AgentRuntime { let decision = decision.unwrap_or_else(|_| crate::ApprovalDecision::Deny { reason: Some("Approval timed out".to_string()), }); - adapter.approval_response(&approval, &original, decision)? + let response = adapter.approval_response(&approval, &original, decision)?; + adapter.interaction_answered(&original, state); + response } InteractionRequest::Question { request: question, @@ -5377,7 +5523,10 @@ impl AgentRuntime { else { return Ok(true); }; - adapter.question_response(&question, &original, answer.ok().flatten())? + let response = + adapter.question_response(&question, &original, answer.ok().flatten())?; + adapter.interaction_answered(&original, state); + response } }; if let Some(response) = response { @@ -5611,7 +5760,7 @@ impl AgentRuntime { stream: "stderr setup", source: std::io::Error::new(std::io::ErrorKind::BrokenPipe, "stderr was not piped"), })?; - let stderr_task = tokio::spawn(crate::process::bounded_stderr(stderr, STDERR_TAIL_BYTES)); + let mut stderr_task = crate::process::StderrCapture::spawn(stderr, STDERR_TAIL_BYTES); // A provider whose protocol is not carried by its own stdio replaces // both halves here. The child stays spawned, supervised and // stderr-drained exactly as before; only the frame carrier differs, @@ -5619,6 +5768,7 @@ impl AgentRuntime { // interaction timeouts, line bounding — is shared by both kinds of // provider instead of growing a second turn loop. let mut attached = false; + let exit_grace = adapter.terminal_exit_grace(); let reader: crate::TransportReader = match adapter.attach(request, &state).await { Ok(Some(streams)) => { attached = true; @@ -5717,7 +5867,9 @@ impl AgentRuntime { } }), }; - adapter.approval_response(&approval, &original, decision)? + let response = adapter.approval_response(&approval, &original, decision)?; + adapter.interaction_answered(&original, &mut state); + response } InteractionRequest::Question { request: question, @@ -5741,7 +5893,9 @@ impl AgentRuntime { interactions.answer(question.clone()), ) => answer.ok().flatten(), }; - adapter.question_response(&question, &original, answer)? + let response = adapter.question_response(&question, &original, answer)?; + adapter.interaction_answered(&original, &mut state); + response } }; if let Some(response) = response { @@ -5757,7 +5911,7 @@ impl AgentRuntime { if output.terminal { protocol_completed = true; stdin.take(); - if attached { + if attached || exit_grace.is_some() { // A stdio provider is read to end-of-output because // closing its stdin is what makes it exit, and trailing // lines can still arrive. An adapter-supplied carrier has @@ -5802,6 +5956,23 @@ impl AgentRuntime { code: None, }, } + } else if let Some(grace) = exit_grace.filter(|_| protocol_completed) { + // The protocol decided the turn; an agent that keeps running + // after end-of-input is stopped instead of holding the turn open + // until its deadline. + if let Ok(status) = tokio::time::timeout(grace, process.wait()).await { + status.map_err(|source| RuntimeError::Transport { provider, source })?; + } else { + process + .terminate() + .await + .map_err(|source| RuntimeError::Transport { provider, source })?; + let _ = tokio::time::timeout(ATTACHED_SHUTDOWN_GRACE, process.wait()).await; + } + TransportExitStatus { + success: true, + code: None, + } } else { process .wait() @@ -5818,12 +5989,18 @@ impl AgentRuntime { } crate::ToolProcessPolicy::TerminateOnCompletion => {} } + // The process is gone, but a descendant it left behind can still hold + // stderr open; the finished turn must not wait for it. When the policy + // leaves descendants running, the reader keeps draining in the + // background past the grace or a cancelled wait: stopping it would + // break the pipe under a tool that `PreserveOnCompletion` meant to + // keep running. Every earlier return stops the reader with the turn. + if request.tool_process_policy == crate::ToolProcessPolicy::PreserveOnCompletion { + stderr_task.detach(); + } let stderr = stderr_task + .finish(STDERR_DRAIN_GRACE) .await - .map_err(|error| RuntimeError::Protocol { - provider, - message: format!("stderr reader task failed: {error}"), - })? .map_err(|source| RuntimeError::ProcessIo { provider, stream: "stderr read", @@ -5932,6 +6109,7 @@ const fn provider_order(provider: Provider) -> u8 { Provider::Codex => 1, Provider::OpenCode => 2, Provider::Pi => 3, + Provider::Acp => 4, } } @@ -5941,6 +6119,7 @@ fn provider_argument(provider: Provider) -> &'static str { Provider::Codex => "codex", Provider::OpenCode => "open_code", Provider::Pi => "pi", + Provider::Acp => "acp", } } @@ -6266,6 +6445,13 @@ fn mcp_management_access(provider: Provider) -> (HarnessExtensionAccess, Harness ); (denied.clone(), denied) } + Provider::Acp => { + let denied = HarnessExtensionAccess::denied( + HarnessExtensionDenialKind::ProviderUnsupported, + "ACP agents receive MCP servers per session; configure persistent servers in each agent's own settings.", + ); + (denied.clone(), denied) + } } } @@ -6430,6 +6616,15 @@ fn build_mcp_management_command( false, )); } + (Provider::Acp, _) => { + return Err(TransportError::new( + TransportErrorKind::Unsupported, + transport, + "manage_mcp_server", + "ACP agents have no common MCP management command", + false, + )); + } } Ok(command) } diff --git a/src/types.rs b/src/types.rs index 51a14f3..7a6bbdf 100644 --- a/src/types.rs +++ b/src/types.rs @@ -177,6 +177,8 @@ pub enum Provider { OpenCode, /// pi coding agent CLI. Pi, + /// Any agent speaking the Agent Client Protocol, selected per turn. + Acp, } impl fmt::Display for Provider { @@ -186,6 +188,7 @@ impl fmt::Display for Provider { Self::Codex => "Codex", Self::OpenCode => "OpenCode", Self::Pi => "pi", + Self::Acp => "ACP agent", }) } } @@ -1032,6 +1035,31 @@ pub(crate) fn is_standard_base64(data: &str) -> bool { } } +/// One step of an agent's execution plan. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct PlanEntry { + /// What the step does. + pub content: String, + /// Progress of the step. + pub status: PlanEntryStatus, + /// Agent-reported priority such as `high`, `medium` or `low`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub priority: Option, +} + +/// Progress of a [`PlanEntry`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[non_exhaustive] +pub enum PlanEntryStatus { + /// Not started. + Pending, + /// Being worked on. + InProgress, + /// Finished. + Completed, +} + /// Provider-neutral streaming event. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] @@ -1090,6 +1118,14 @@ pub enum TurnEvent { #[serde(default, skip_serializing_if = "Vec::is_empty")] images: Vec, }, + /// Full replacement of the agent's execution plan. + /// + /// Agents that publish a plan (ACP `plan` updates) send every entry on + /// each change, so consumers replace the previous plan. + PlanUpdated { + /// Plan entries in the agent's order. + entries: Vec, + }, /// Replaceable snapshot of provider-native tasks and subagents. TasksChanged { /// Complete current task set, bounded by the adapter. diff --git a/tests/acp_stdio.rs b/tests/acp_stdio.rs new file mode 100644 index 0000000..eabb1b1 --- /dev/null +++ b/tests/acp_stdio.rs @@ -0,0 +1,687 @@ +//! End-to-end ACP coverage against a scripted agent process. +//! +//! The fixture is a POSIX shell script that speaks ACP's JSON-RPC over its +//! real stdio: it answers `initialize`, `session/new` and `session/load`, +//! streams `session/update` notifications, asks for permission, honors +//! `session/cancel`, and can keep running after end-of-input the way some +//! Node-based agents do. Nothing here needs a real agent installed. + +#![cfg(all(feature = "acp", unix))] + +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use async_trait::async_trait; +use temps_agent_runtime::providers::acp::{ + AcpAgentProbe, AcpAuthentication, AcpProbeErrorKind, AGENT_OPTION, +}; +use temps_agent_runtime::providers::{Acp, AcpAgent, AcpAgents}; +use temps_agent_runtime::{ + AgentRuntime, ApprovalDecision, ApprovalRequest, EventSink, HarnessControlKind, HarnessStatus, + InteractionHandler, PermissionMode, PlanEntryStatus, Provider, ProviderProcessErrorKind, + QuestionAnswer, QuestionRequest, Result, RuntimeError, ToolCallStatus, TurnEvent, TurnRequest, +}; + +const AGENT: &str = r#"#!/bin/sh +# Scripted ACP agent. MOCK_ACP_MODE: ok, auth, hang, linger, stall (never +# reacts to a permission reply), split (streams a tool result, then completes +# with a bare status), orphan (never answers, and leaves a child of its own +# process group that keeps stderr open), finish_orphan (leaves a child that +# writes to stderr after 8 seconds, then runs the turn to completion and +# exits) or late_orphan (the same child after 2 seconds, and never answers). +if [ "$1" = --version ]; then echo "mock-agent 0.1.0"; exit 0; fi +log="$MOCK_ACP_LOG" +say() { printf '%s\n' "$1"; } +update() { say "{\"jsonrpc\":\"2.0\",\"method\":\"session/update\",\"params\":{\"sessionId\":\"mock-session\",\"update\":$1}}"; } +if [ "$MOCK_ACP_MODE" = finish_orphan ] || [ "$MOCK_ACP_MODE" = late_orphan ]; then + # Outlives the turn, then writes to the inherited stderr and records that + # the write succeeded: a closed pipe would kill it with SIGPIPE first. + delay=8 + if [ "$MOCK_ACP_MODE" = late_orphan ]; then delay=2; fi + child="sleep $delay; echo late diagnostic >&2; echo wrote > \"\$1\"" + if command -v perl >/dev/null 2>&1; then + perl -MPOSIX -e 'POSIX::setsid(); exec @ARGV' sh -c "$child" sh "$log.late" >/dev/null & + else + setsid sh -c "$child" sh "$log.late" >/dev/null & + fi + echo $! > "$log.orphan" +elif [ "$MOCK_ACP_MODE" = orphan ]; then + if command -v perl >/dev/null 2>&1; then + perl -MPOSIX -e 'POSIX::setsid(); exec "sleep", "20"' >/dev/null & + else + setsid sleep 20 >/dev/null & + fi + echo $! > "$log.orphan" +fi +while IFS= read -r line; do + printf '%s\n' "$line" >> "$log" + case "$line" in + *'"method":"initialize"'*) + if [ "$MOCK_ACP_MODE" = orphan ] || [ "$MOCK_ACP_MODE" = late_orphan ]; then continue; fi + say '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true},"authMethods":[{"id":"login","name":"Log in with Mock"}],"agentInfo":{"name":"mock-agent","version":"0.1.0"}}}' ;; + *'"method":"session/new"'*) + if [ "$MOCK_ACP_MODE" = auth ]; then + say '{"jsonrpc":"2.0","id":2,"error":{"code":-32000,"message":"Authentication required"}}' + else + echo 'Loaded cached credentials.' + say '{"jsonrpc":"2.0","id":2,"result":{"sessionId":"mock-session"}}' + fi ;; + *'"method":"session/load"'*) + update '{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"replayed"}}' + say '{"jsonrpc":"2.0","id":2,"result":null}' ;; + *'"method":"session/prompt"'*) + update '{"sessionUpdate":"plan","entries":[{"content":"Run the tests","priority":"high","status":"in_progress"}]}' + update '{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"Running tests."}}' + if [ "$MOCK_ACP_MODE" = hang ]; then continue; fi + update '{"sessionUpdate":"tool_call","toolCallId":"call-1","title":"Run npm test","kind":"execute","status":"pending","rawInput":{"command":"npm test"}}' + say '{"jsonrpc":"2.0","id":7,"method":"session/request_permission","params":{"sessionId":"mock-session","toolCall":{"toolCallId":"call-1","title":"Run npm test","kind":"execute"},"options":[{"optionId":"allow","name":"Allow","kind":"allow_once"},{"optionId":"reject","name":"Reject","kind":"reject_once"}]}}' ;; + *'"id":7'*) + if [ "$MOCK_ACP_MODE" = stall ]; then continue; fi + case "$line" in + *'"optionId":"allow"'*) + if [ "$MOCK_ACP_MODE" = split ]; then + update '{"sessionUpdate":"tool_call_update","toolCallId":"call-1","status":"in_progress","content":[{"type":"content","content":{"type":"text","text":"12 passed"}}]}' + update '{"sessionUpdate":"tool_call_update","toolCallId":"call-1","status":"completed"}' + else + update '{"sessionUpdate":"tool_call_update","toolCallId":"call-1","status":"completed","content":[{"type":"content","content":{"type":"text","text":"12 passed"}}]}' + fi + update '{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":" All green."}}' ;; + *) + update '{"sessionUpdate":"tool_call_update","toolCallId":"call-1","status":"failed","content":[{"type":"content","content":{"type":"text","text":"rejected by the user"}}]}' + update '{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":" Skipped."}}' ;; + esac + update '{"sessionUpdate":"plan","entries":[{"content":"Run the tests","priority":"high","status":"completed"}]}' + say '{"jsonrpc":"2.0","id":5,"result":{"stopReason":"end_turn"}}' ;; + *'"method":"session/cancel"'*) + say '{"jsonrpc":"2.0","id":5,"result":{"stopReason":"cancelled"}}' ;; + esac +done +if [ "$MOCK_ACP_MODE" = linger ]; then + while :; do sleep 1; done +fi +"#; + +struct Fixture { + _dir: tempfile::TempDir, + script: PathBuf, + log: PathBuf, + workspace: PathBuf, +} + +impl Fixture { + fn new() -> Self { + let dir = tempfile::tempdir().unwrap(); + let script = dir.path().join("mock-acp-agent"); + std::fs::write(&script, AGENT).unwrap(); + std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap(); + let workspace = dir.path().join("workspace"); + std::fs::create_dir(&workspace).unwrap(); + Self { + log: dir.path().join("frames.log"), + script, + workspace, + _dir: dir, + } + } + + fn agent(&self, mode: &str) -> AcpAgent { + let mut agent = AcpAgent::new("mock", "Mock Agent", &self.script, Vec::::new()); + agent + .environment + .insert("MOCK_ACP_MODE".into(), mode.into()); + agent + .environment + .insert("MOCK_ACP_LOG".into(), self.log.to_string_lossy().into()); + agent + } + + fn methods(&self) -> Vec { + std::fs::read_to_string(&self.log) + .unwrap_or_default() + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .map(|frame| { + frame + .get("method") + .and_then(serde_json::Value::as_str) + .map_or_else(|| format!("reply:{}", frame["id"]), str::to_owned) + }) + .collect() + } + + fn runtime(&self, mode: &str) -> AgentRuntime { + let agents = AcpAgents::new(); + agents.insert(self.agent(mode)).unwrap(); + let mut builder = AgentRuntime::builder(); + builder.register(Acp::new(agents)); + builder.build().unwrap() + } + + fn request(&self, prompt: &str) -> TurnRequest { + let mut request = TurnRequest::new(Provider::Acp, &self.workspace, prompt); + request + .harness_options + .insert(AGENT_OPTION.into(), "mock".into()); + request.timeout = Duration::from_secs(30); + request.interaction_timeout = Duration::from_secs(10); + request + } +} + +#[derive(Default)] +struct Collector { + events: Arc>>, +} + +#[async_trait] +impl EventSink for Collector { + async fn emit(&self, event: TurnEvent) -> Result<()> { + self.events.lock().unwrap().push(event); + Ok(()) + } +} + +impl Collector { + fn events(&self) -> Vec { + self.events.lock().unwrap().clone() + } +} + +struct Responder { + decision: ApprovalDecision, + approvals: Arc>>, +} + +impl Responder { + fn new(decision: ApprovalDecision) -> Self { + Self { + decision, + approvals: Arc::new(Mutex::new(Vec::new())), + } + } +} + +#[async_trait] +impl InteractionHandler for Responder { + async fn approve(&self, request: ApprovalRequest) -> ApprovalDecision { + self.approvals.lock().unwrap().push(request); + self.decision.clone() + } + + async fn answer(&self, _request: QuestionRequest) -> Option { + None + } +} + +#[tokio::test] +async fn runs_a_turn_with_an_approved_tool_call() { + let fixture = Fixture::new(); + let events = Collector::default(); + let responder = Responder::new(ApprovalDecision::Allow); + let result = fixture + .runtime("ok") + .run(fixture.request("run the tests"), &events, Some(&responder)) + .await + .unwrap(); + + assert_eq!(result.text, "Running tests. All green."); + assert_eq!(result.session_id.as_deref(), Some("mock-session")); + assert_eq!( + fixture.methods(), + ["initialize", "session/new", "session/prompt", "reply:7"] + ); + let approvals = responder.approvals.lock().unwrap().clone(); + assert_eq!(approvals.len(), 1); + assert_eq!(approvals[0].tool_name, "Run npm test"); + + let events = events.events(); + assert!(events.iter().any(|event| matches!( + event, + TurnEvent::SessionStarted { session_id, .. } if session_id == "mock-session" + ))); + assert!(events.iter().any(|event| matches!( + event, + TurnEvent::ToolCall { status: ToolCallStatus::Succeeded, output: Some(text), .. } if text == "12 passed" + ))); + let plans = events + .iter() + .filter_map(|event| match event { + TurnEvent::PlanUpdated { entries } => Some(entries[0].status), + _ => None, + }) + .collect::>(); + assert_eq!( + plans, + [PlanEntryStatus::InProgress, PlanEntryStatus::Completed] + ); + assert!(events.iter().any(|event| matches!( + event, + TurnEvent::Warning { message } if message.contains("non-ACP output") + ))); +} + +#[tokio::test] +async fn a_denied_permission_is_reported_to_the_agent() { + let fixture = Fixture::new(); + let events = Collector::default(); + let responder = Responder::new(ApprovalDecision::Deny { reason: None }); + let result = fixture + .runtime("ok") + .run(fixture.request("run the tests"), &events, Some(&responder)) + .await + .unwrap(); + assert_eq!(result.text, "Running tests. Skipped."); + assert!(events.events().iter().any(|event| matches!( + event, + TurnEvent::ToolCall { + status: ToolCallStatus::Failed, + .. + } + ))); +} + +#[tokio::test] +async fn full_access_answers_permission_requests_itself() { + let fixture = Fixture::new(); + let events = Collector::default(); + let mut request = fixture.request("run the tests"); + request.permission_mode = PermissionMode::FullAccess; + let result = fixture + .runtime("ok") + .run(request, &events, None) + .await + .unwrap(); + assert_eq!(result.text, "Running tests. All green."); +} + +#[tokio::test] +async fn resuming_loads_the_session_without_replaying_it() { + let fixture = Fixture::new(); + let events = Collector::default(); + let mut request = fixture.request("again"); + request.session_id = Some("mock-session".into()); + request.permission_mode = PermissionMode::FullAccess; + let result = fixture + .runtime("ok") + .run(request, &events, None) + .await + .unwrap(); + assert_eq!(result.text, "Running tests. All green."); + assert_eq!(fixture.methods()[1], "session/load"); + assert!(!events + .events() + .iter() + .any(|event| matches!(event, TurnEvent::SessionStarted { .. }))); +} + +#[tokio::test] +async fn an_agent_that_needs_sign_in_fails_with_an_authentication_error() { + let fixture = Fixture::new(); + let error = fixture + .runtime("auth") + .run(fixture.request("hi"), &Collector::default(), None) + .await + .unwrap_err(); + match error { + RuntimeError::ProcessFailed { kind, .. } => { + assert_eq!(kind, ProviderProcessErrorKind::AuthenticationFailed); + } + other => panic!("expected an authentication failure, got {other:?}"), + } + assert!(!fixture + .methods() + .iter() + .any(|method| method == "session/prompt")); +} + +#[tokio::test] +async fn an_agent_that_keeps_running_after_its_turn_is_stopped() { + let fixture = Fixture::new(); + let mut request = fixture.request("run the tests"); + request.permission_mode = PermissionMode::FullAccess; + let started = Instant::now(); + let result = fixture + .runtime("linger") + .run(request, &Collector::default(), None) + .await + .unwrap(); + assert_eq!(result.text, "Running tests. All green."); + assert!( + started.elapsed() < Duration::from_secs(15), + "the finished turn waited {:?} for the agent to exit", + started.elapsed() + ); +} + +#[tokio::test] +async fn cancellation_sends_session_cancel() { + let fixture = Fixture::new(); + let events = Collector::default(); + let request = fixture.request("run the tests"); + let cancellation = request.cancellation.clone(); + let runtime = fixture.runtime("hang"); + let turn = runtime.run(request, &events, None); + let cancel = async { + for _ in 0..200 { + if events + .events() + .iter() + .any(|event| matches!(event, TurnEvent::TextDelta { .. })) + { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + cancellation.cancel(); + }; + let (result, ()) = tokio::join!(turn, cancel); + assert!( + matches!(result, Err(RuntimeError::Cancelled { .. })), + "{result:?}" + ); + assert!(fixture + .methods() + .iter() + .any(|method| method == "session/cancel")); +} + +#[tokio::test] +async fn probing_reports_capabilities_and_sign_in_state() { + let fixture = Fixture::new(); + let probe: AcpAgentProbe = Acp::probe_agent( + &fixture.agent("ok"), + &fixture.workspace, + true, + Duration::from_secs(10), + ) + .await + .unwrap(); + assert_eq!(probe.protocol_version, Some(1)); + assert_eq!(probe.agent_name.as_deref(), Some("mock-agent")); + assert!(probe.load_session); + assert_eq!(probe.auth_methods[0].name, "Log in with Mock"); + assert_eq!(probe.authentication, AcpAuthentication::Authenticated); + + let probe = Acp::probe_agent( + &fixture.agent("auth"), + &fixture.workspace, + true, + Duration::from_secs(10), + ) + .await + .unwrap(); + assert!(matches!( + probe.authentication, + AcpAuthentication::Required { .. } + )); +} + +#[tokio::test] +async fn probing_a_missing_or_silent_command_says_what_is_wrong() { + let fixture = Fixture::new(); + let missing = AcpAgent::new("missing", "Missing", "temps-acp-not-installed", ["--acp"]); + let error = Acp::probe_agent(&missing, &fixture.workspace, false, Duration::from_secs(5)) + .await + .unwrap_err(); + assert_eq!(error.kind, AcpProbeErrorKind::NotInstalled); + + // `sleep` never speaks ACP: the probe times out instead of hanging. + let silent = AcpAgent::new("silent", "Silent", "sleep", ["30"]); + let started = Instant::now(); + let error = Acp::probe_agent(&silent, Path::new("."), false, Duration::from_secs(1)) + .await + .unwrap_err(); + assert_eq!(error.kind, AcpProbeErrorKind::Timeout); + assert!(started.elapsed() < Duration::from_secs(10)); +} + +#[tokio::test] +async fn a_tool_result_streamed_before_the_final_status_is_kept() { + let fixture = Fixture::new(); + let events = Collector::default(); + let responder = Responder::new(ApprovalDecision::Allow); + let result = fixture + .runtime("split") + .run(fixture.request("run the tests"), &events, Some(&responder)) + .await + .unwrap(); + assert_eq!(result.text, "Running tests. All green."); + let completed = events + .events() + .into_iter() + .filter_map(|event| match event { + TurnEvent::ToolCall { + status: ToolCallStatus::Succeeded, + output, + .. + } => Some(output), + _ => None, + }) + .collect::>(); + assert_eq!(completed, [Some("12 passed".to_string())]); +} + +#[tokio::test] +async fn a_relative_working_directory_is_refused_before_the_agent_starts() { + let fixture = Fixture::new(); + // `tests` exists relative to the test's directory, so the refusal is the + // adapter's, not a missing directory. + let mut request = TurnRequest::new(Provider::Acp, "tests", "hi"); + request + .harness_options + .insert(AGENT_OPTION.into(), "mock".into()); + let error = fixture + .runtime("ok") + .run(request, &Collector::default(), None) + .await + .unwrap_err(); + assert!( + error.to_string().contains("absolute working directory"), + "{error}" + ); + assert_eq!(fixture.methods().len(), 0, "the agent must not be started"); + + let error = Acp::probe_agent( + &fixture.agent("ok"), + Path::new("workspace"), + true, + Duration::from_secs(10), + ) + .await + .unwrap_err(); + assert_eq!(error.kind, AcpProbeErrorKind::Invalid); + assert_eq!(fixture.methods().len(), 0, "the agent must not be started"); +} + +#[tokio::test] +async fn cancelling_after_the_decision_does_not_answer_the_request_twice() { + let fixture = Fixture::new(); + let events = Collector::default(); + let responder = Responder::new(ApprovalDecision::Allow); + let request = fixture.request("run the tests"); + let cancellation = request.cancellation.clone(); + let runtime = fixture.runtime("stall"); + let turn = runtime.run(request, &events, Some(&responder)); + let cancel = async { + // The agent never reacts to the decision, so the permission stays + // open in the adapter until the decision's reply is on the wire. + for _ in 0..400 { + if fixture.methods().iter().any(|method| method == "reply:7") { + break; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + cancellation.cancel(); + }; + let (result, ()) = tokio::join!(turn, cancel); + assert!( + matches!(result, Err(RuntimeError::Cancelled { .. })), + "{result:?}" + ); + let methods = fixture.methods(); + assert!(methods.iter().any(|method| method == "session/cancel")); + assert_eq!( + methods + .iter() + .filter(|method| method.as_str() == "reply:7") + .count(), + 1, + "request 7 was answered more than once: {methods:?}" + ); +} + +#[tokio::test] +async fn a_probe_does_not_wait_for_a_descendant_that_kept_stderr_open() { + let fixture = Fixture::new(); + let started = Instant::now(); + let error = Acp::probe_agent( + &fixture.agent("orphan"), + &fixture.workspace, + false, + Duration::from_secs(1), + ) + .await + .unwrap_err(); + let elapsed = started.elapsed(); + // The agent left its process group, so stopping the group does not close + // the stderr pipe the descendant still holds. + if let Ok(pid) = std::fs::read_to_string(format!("{}.orphan", fixture.log.display())) { + let _ = std::process::Command::new("kill").arg(pid.trim()).status(); + } + assert_eq!(error.kind, AcpProbeErrorKind::Timeout); + assert!( + elapsed < Duration::from_secs(10), + "the probe waited {elapsed:?} for stderr to close" + ); +} + +#[tokio::test] +async fn a_finished_turn_does_not_wait_for_a_descendant_that_kept_stderr_open() { + let fixture = Fixture::new(); + let mut request = fixture.request("run the tests"); + request.permission_mode = PermissionMode::FullAccess; + // The descendant holds stderr for 8 seconds; a turn that waited for it + // would hit this deadline instead of returning its result. + request.timeout = Duration::from_secs(6); + let started = Instant::now(); + let result = fixture + .runtime("finish_orphan") + .run(request, &Collector::default(), None) + .await; + let elapsed = started.elapsed(); + let result = result.unwrap_or_else(|error| { + panic!("the finished turn failed after {elapsed:?}: {error:?}"); + }); + assert_eq!(result.text, "Running tests. All green."); + assert!( + elapsed < Duration::from_secs(6), + "the finished turn waited {elapsed:?} for stderr to close" + ); + + // The default policy preserves what the agent left running, so the + // runtime must not have closed the pipe under it: its late write has to + // succeed rather than raise SIGPIPE. + let marker = format!("{}.late", fixture.log.display()); + let mut wrote = false; + for _ in 0..300 { + if std::fs::read_to_string(&marker).is_ok_and(|text| text.trim() == "wrote") { + wrote = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + if let Ok(pid) = std::fs::read_to_string(format!("{}.orphan", fixture.log.display())) { + let _ = std::process::Command::new("kill").arg(pid.trim()).status(); + } + assert!(wrote, "the preserved descendant lost its stderr pipe"); +} + +#[tokio::test] +async fn discovery_lists_registered_acp_agents_that_are_installed() { + let fixture = Fixture::new(); + let agents = AcpAgents::new(); + agents.insert(fixture.agent("ok")).unwrap(); + agents + .insert(AcpAgent::new( + "ghost", + "Ghost Agent", + "temps-acp-not-installed", + ["--acp"], + )) + .unwrap(); + let mut builder = AgentRuntime::builder(); + builder.register(Acp::new(agents.clone())); + let runtime = builder.build().unwrap(); + + let inventory = runtime.discover_harnesses().await; + let harness = inventory + .harnesses + .iter() + .find(|harness| harness.provider == Provider::Acp) + .expect("a registered ACP adapter appears in the inventory"); + assert_eq!(harness.status, HarnessStatus::Ready); + let readiness = harness.readiness.as_ref().unwrap(); + assert!(readiness.installed); + assert_eq!(readiness.version.as_deref(), Some("mock-agent 0.1.0")); + assert!( + readiness.detail.contains("Ghost Agent") && readiness.detail.contains("not found"), + "{}", + readiness.detail + ); + let offered = harness + .control_groups + .iter() + .filter(|group| group.kind == HarnessControlKind::Agent) + .flat_map(|group| group.options.iter().map(|option| option.id.as_str())) + .collect::>(); + assert_eq!(offered, ["mock"]); + assert!( + runtime.readiness(Provider::Acp).await.unwrap().installed, + "readiness must follow the registered agents, not the placeholder `acp` executable" + ); + + // Registry changes apply to the next probe without rebuilding the runtime. + agents.remove("mock"); + let inventory = runtime.discover_harnesses().await; + let harness = inventory + .harnesses + .iter() + .find(|harness| harness.provider == Provider::Acp) + .unwrap(); + assert_eq!(harness.status, HarnessStatus::NotInstalled); + assert_eq!(harness.control_groups.len(), 0); + assert!(!runtime.readiness(Provider::Acp).await.unwrap().installed); + + agents.remove("ghost"); + let readiness = runtime.readiness(Provider::Acp).await.unwrap(); + assert!(!readiness.installed); + assert!(readiness.detail.contains("agent is registered")); +} + +#[tokio::test] +async fn a_failed_turn_stops_reading_stderr() { + let fixture = Fixture::new(); + let mut request = fixture.request("run the tests"); + request.permission_mode = PermissionMode::FullAccess; + // The agent never answers, so the turn fails at its deadline, before the + // descendant that escaped the process group writes to stderr. + request.timeout = Duration::from_secs(1); + let result = fixture + .runtime("late_orphan") + .run(request, &Collector::default(), None) + .await; + assert!( + matches!(result, Err(RuntimeError::Timeout { .. })), + "{result:?}" + ); + // A failed turn has nothing to preserve, so the reader goes with it and + // the descendant's later write finds a closed pipe. + tokio::time::sleep(Duration::from_secs(4)).await; + let wrote = std::fs::read_to_string(format!("{}.late", fixture.log.display())).is_ok(); + if let Ok(pid) = std::fs::read_to_string(format!("{}.orphan", fixture.log.display())) { + let _ = std::process::Command::new("kill").arg(pid.trim()).status(); + } + assert!(!wrote, "a failed turn left its stderr reader running"); +}