//! ACP (Agent Client Protocol) agent backend: spawns and supervises an
//! external agent process speaking JSON-RPC 2.0 over NDJSON stdio (KRZ-301 —
//! the vendor-neutrality seam: any agent that speaks ACP can drive a kranz
//! mission role without a per-vendor CLI parser).
//!
//! **Schema version targeted:** ACP protocol version **1** — the stable
//! `schema/v1/schema.json` of `agentclientprotocol/agent-client-protocol`
//! (meta.json `"version": 1`, crate release v1.6.0, 2026-07-21). v2 shapes
//! (schema.unstable.json) are deliberately NOT used: unstable discriminators
//! would tie the seam to a moving target. Protocol versions are MAJOR-only,
//! so a peer answering `initialize` with anything other than 1 is rejected.
//!
//! ## Wire → event mapping
//!
//! | ACP message | maps to |
//! |---|---|
//! | `initialize` + `session/new` responses | synthesized [`AgentEvent::Init`] (the peer's `sessionId`; model recorded from config — ACP v1 has no standard model-reporting field) |
//! | `session/update` `agent_message_chunk` (text) | [`AgentEvent::Text`] per chunk (deltas) |
//! | `session/update` `tool_call` | [`AgentEvent::ToolUse`] (`tool` = ACP `kind`, `summary` = `title`) |
//! | `session/update` `tool_call_update` with terminal status (`completed`/`failed`) | [`AgentEvent::ToolResult`] (`denied: false` — a failed tool is a normal failure, not a denial, mirroring `backend_codex`) |
//! | `session/request_permission` refused at the seam | synthesized [`AgentEvent::ToolResult`] with `denied: true` (the refusal IS the event — the peer may report nothing itself) |
//! | `session/update` `usage_update` | remembered; its `cost` (USD only) lands on the terminal `Result.cost_usd` |
//! | `session/prompt` response (`stopReason`) | synthesized terminal [`AgentEvent::Result`]: text stitched from the LAST assistant message's chunks (mirrors `backend_codex` "last agent_message wins"; `messageId` changes delimit messages), `is_error` ⇔ `stopReason != "end_turn"` — ACP v1 names `end_turn` as the ONLY natural completion, so `refusal`, `max_tokens`, `max_turn_requests`, `cancelled`, and any missing/unknown reason all fail honestly (12th-pass review: a truncated or cancelled validator report must never read as a pass) |
//! | everything else (`plan`, `agent_thought_chunk`, `available_commands_update`, unknown kinds, unparseable lines) | [`AgentEvent::Other`] — kept for transcripts, never dropped |
//!
//! ## What is NOT on this wire (absent, never fabricated)
//!
//! ACP v1's `usage_update` reports context-window state (`used`/`size`
//! tokens) and an optional cumulative `cost` — NOT an input/output/cache
//! token split. [`AgentEvent::Result`] `usage` therefore stays the zero
//! default (a fabricated split would be invented data), and `cost_usd` is
//! `Some` only when the peer reported a USD amount. There is likewise no
//! client-side price-table fallback: kranz cannot price an arbitrary ACP
//! peer's model, so unreported cost stays `None`. Model selection is the
//! peer's own concern (encoded in the configured command/args); the
//! configured model string is recorded on `Init` for attribution only.
//!
//! ## Permission mapping (the allow/deny seam)
//!
//! The peer asks before acting via `session/request_permission`; the answer
//! is computed from the [`SessionSpec`] (see [`decide_permission`]):
//!
//! - **Disallowed patterns** (`spec.disallowed_tools`, claude-shaped strings
//! like `Bash(git push*)`) are matched against the ACP `kind` via
//! [`TOOL_NAME_KINDS`] (`Bash`↔`execute`, `Edit`/`Write`↔`edit`, …) and a
//! `*`-wildcard glob against the call's subject (command for `execute`,
//! path for the file kinds, title otherwise). A match refuses the call —
//! this is where the no-push/no-publish invariants land.
//! - **Read-only sessions** (`writable: false`) refuse the filesystem-
//! mutating kinds `edit`/`delete`/`move` outright; `execute` stays allowed
//! unless disallowed-matched, because a read-only role still runs
//! read-only commands (`git diff`, `cargo test`) and ACP's `execute` kind
//! does not split reads from writes. This is the ACP equivalent posture;
//! what it CANNOT do is constrain a peer that never asks permission —
//! there is no client-side fs proxy in v1, and `BackendKind::Acp` reports
//! `supports_sandbox_enforcement() == false`, so `config::validate`
//! refuses an enforced OS sandbox on this backend rather than letting the
//! gap go silent. `spec.allowed_tools` is not interpreted yet (the peer's
//! permission request IS the ask; auto-approving without one would weaken
//! the seam).
//!
//! - **Missing wire fields fail CLOSED.** ACP v1 leaves `title` optional and
//! `rawInput` free-form, so a call can arrive with no subject at all; every
//! glob then matches nothing and the deny list would silently vacate. When
//! a deny pattern's tool name covers the call's kind but the subject is
//! empty, the call is refused and the reason names the missing field. Same
//! for a read-only session when the peer omits `kind` (it arrives as
//! `"other"`, which `MUTATING_KINDS` cannot classify).
//!
//! A refusal picks the first `reject_once` (else `reject_always`) option the
//! peer offered, falling back to the `cancelled` outcome when it offered
//! none; an approval picks the first `allow_once` (else `allow_always`,
//! else first) option.
//!
//! ## Process supervision
//!
//! House discipline, mirroring `backend_kimi`/`backend_codex`: env-cleared
//! spawn ([`crate::agent_env::agent_session_env`] — ACP has no canonical
//! auth env var, so NO ambient credential crosses; the peer authenticates
//! from its own config), unix process-group + post-reap sweep /
//! windows Job Object tree kill, `kill_on_drop`, bounded stdout lines and a
//! bounded stderr tail ([`crate::stream_bounds`]). A killed peer's torn
//! final NDJSON line is never a parse failure: [`BoundedLines`] returns it
//! as one last unterminated line, which routes to [`AgentEvent::Other`]
//! like any unparseable line, so the events already surfaced stay complete
//! and the session simply ends `Aborted`/`Failed`.
//!
//! `resume` is rejected at the seam (`session/load` is an optional v1
//! capability this backend does not negotiate; `session/resume` is
//! unstable-v2 only). Client capabilities advertise `fs`/`terminal` as
//! unsupported, so a conformant peer never calls `fs/*`/`terminal/*`; one
//! that does gets a JSON-RPC `-32601` error response, not silent service.
use crate::backend::{
AgentBackend, AgentEvent, AgentSession, PromptMode, SessionExit, SessionSpec,
};
#[cfg(unix)]
use crate::backend_claude::kill_group;
#[cfg(windows)]
use crate::backend_claude::win_job;
use crate::error::{EngineError, Result};
use crate::stream_bounds::{drain_to_tail, BoundedLines, STDERR_TAIL_CAP};
use crate::types::TokenUsage;
use serde_json::{json, Value};
use std::collections::{HashMap, VecDeque};
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::{Arc, Mutex};
use tokio::process::{Child, ChildStdin, ChildStdout};
use tokio::task::JoinHandle;
/// Max characters kept in tool-use / tool-result summaries.
const SUMMARY_MAX_CHARS: usize = 200;
/// Max characters of captured stderr included in failure messages.
const STDERR_TAIL_CHARS: usize = 500;
/// The only ACP protocol version this backend speaks (stable schema v1; see
/// module docs). A peer negotiating anything else is refused at `initialize`.
const ACP_PROTOCOL_VERSION: u64 = 1;
/// Deadline for one handshake request (`initialize`, `session/new`). Both
/// are capability negotiation — no model call — so a peer that cannot answer
/// within this window is hung or not an ACP agent; bounding it keeps
/// `start()` from parking the run loop forever. The prompt turn itself is
/// deliberately unbounded here: turn/stall budgets are the engine's call
/// (runner-level), not the transport's.
const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
/// JSON-RPC method names, stable schema v1 (`meta.json`).
mod method {
pub(crate) const INITIALIZE: &str = "initialize";
pub(crate) const SESSION_NEW: &str = "session/new";
pub(crate) const SESSION_PROMPT: &str = "session/prompt";
pub(crate) const SESSION_CANCEL: &str = "session/cancel";
pub(crate) const SESSION_UPDATE: &str = "session/update";
pub(crate) const REQUEST_PERMISSION: &str = "session/request_permission";
}
/// Claude-style tool name → ACP `kind`s it matches, for interpreting
/// `SessionSpec::disallowed_tools` patterns against ACP tool calls. ACP
/// kinds are the only tool identity the wire carries (`title` is
/// free-form display text), so the mapping is necessarily coarse; it is
/// used ONLY for deny decisions, never to auto-approve.
const TOOL_NAME_KINDS: &[(&str, &[&str])] = &[
("Bash", &["execute"]),
("Read", &["read"]),
("Write", &["edit"]),
("Edit", &["edit"]),
("NotebookEdit", &["edit"]),
("Glob", &["search"]),
("Grep", &["search"]),
("WebSearch", &["search", "fetch"]),
("WebFetch", &["fetch"]),
];
/// ACP `kind`s that mutate the filesystem; refused outright in read-only
/// (`writable: false`) sessions. `execute` is deliberately not in this set —
/// see the module-docs permission section.
const MUTATING_KINDS: &[&str] = &["edit", "delete", "move"];
// ---------------------------------------------------------------------------
// JSON-RPC framing
// ---------------------------------------------------------------------------
/// One stdout line classified by JSON-RPC shape. The wire is symmetric
/// (both sides issue requests), so a line is one of: a response to a client
/// request, a peer request we must answer, or a notification.
#[derive(Debug)]
enum Frame {
/// `result`/`error` for a client-issued request id.
Response { id: u64, outcome: RpcOutcome },
/// Peer→client request (`session/request_permission`, or an unsupported
/// client method). The `id` is echoed back verbatim — it may be a string
/// or a number per JSON-RPC, so it is kept as a raw [`Value`].
Request {
id: Value,
method: String,
params: Value,
raw: Value,
},
/// Peer→client notification (`session/update`, or anything else).
Notification {
method: String,
params: Value,
raw: Value,
},
/// Not JSON-RPC-shaped (or not JSON at all) — transcript only.
Unrecognized(Value),
}
/// The payload of a JSON-RPC response: peer ids are always numbers in our
/// exchanges with the peer's client side, but the error path keeps the raw
/// object for the transcript.
#[derive(Debug)]
enum RpcOutcome {
Result(Value),
Error(Value),
}
/// Classify one stdout line. Unparseable lines become
/// [`Frame::Unrecognized`] with `raw = {"unparsed": <line>}` so nothing is
/// ever dropped from transcripts (mirrors `backend_codex::parse_codex_line`).
fn classify_line(line: &str) -> Frame {
let value = match serde_json::from_str::<Value>(line) {
Ok(value) => value,
Err(_) => return Frame::Unrecognized(json!({ "unparsed": line })),
};
classify_value(value)
}
fn classify_value(value: Value) -> Frame {
let obj = match value.as_object() {
Some(obj) => obj,
None => return Frame::Unrecognized(value),
};
let has_id = obj.contains_key("id");
let method = obj.get("method").and_then(Value::as_str);
match (method, has_id) {
// Request: method + id.
(Some(method), true) => Frame::Request {
id: obj.get("id").cloned().unwrap_or(Value::Null),
method: method.to_string(),
params: obj.get("params").cloned().unwrap_or(Value::Null),
raw: value,
},
// Notification: method, no id.
(Some(method), false) => Frame::Notification {
method: method.to_string(),
params: obj.get("params").cloned().unwrap_or(Value::Null),
raw: value,
},
// Response: no method, carries result or error. Our request ids are
// numbers; anything else is not a response to us.
(None, true) => {
let id = obj.get("id").and_then(Value::as_u64);
match (id, obj.get("result"), obj.get("error")) {
(Some(id), Some(result), _) => Frame::Response {
id,
outcome: RpcOutcome::Result(result.clone()),
},
(Some(id), None, Some(error)) => Frame::Response {
id,
outcome: RpcOutcome::Error(error.clone()),
},
_ => Frame::Unrecognized(value),
}
}
(None, false) => Frame::Unrecognized(value),
}
}
// ---------------------------------------------------------------------------
// Permission policy (pure; the seam the ticket pins)
// ---------------------------------------------------------------------------
/// What the seam decided about one `session/request_permission`.
#[derive(Debug, Clone, PartialEq, Eq)]
enum PermissionDecision {
Allow,
/// Human-readable reason; lands on the synthesized denial event's summary.
Deny(String),
}
/// Everything known about the tool call a permission request covers: the
/// `tool_call`/`tool_call_update` fields tracked by [`AcpSession`], merged
/// with whatever the request itself carries (the request's embedded
/// `ToolCallUpdate` may be the first place a title/kind appears).
#[derive(Debug, Clone, Default)]
struct ToolCallInfo {
kind: String,
title: String,
/// Command (`execute`) or path (file kinds) extracted from
/// `rawInput`/`locations`; the subject glob patterns match against.
subject: String,
}
/// Extract the matchable subject from a tool-call-shaped value
/// (`rawInput.command`/`cmd` for execute, `locations[0].path` or
/// `rawInput.path` for the file kinds, `title` as the last resort).
fn tool_call_subject(kind: &str, title: &str, call: &Value) -> String {
let raw_input = call.get("rawInput").cloned().unwrap_or(Value::Null);
let str_at = |value: &Value, keys: &[&str]| -> Option<String> {
keys.iter()
.find_map(|k| value.get(*k).and_then(Value::as_str).map(str::to_string))
};
if kind == "execute" {
if let Some(command) = str_at(&raw_input, &["command", "cmd"]) {
return command;
}
}
if let Some(path) = call
.get("locations")
.and_then(Value::as_array)
.and_then(|locs| locs.first())
.and_then(|loc| loc.get("path"))
.and_then(Value::as_str)
{
return path.to_string();
}
if let Some(path) = str_at(&raw_input, &["path", "filePath", "file_path"]) {
return path;
}
title.to_string()
}
/// `*`-wildcard match (a bare `*` spans any text including empty; every
/// other character matches literally, case-sensitively — shell commands and
/// paths are case-sensitive on the platforms this guards).
fn wildcard_match(pattern: &str, text: &str) -> bool {
let parts = pattern.split('*');
let anchored_start = !pattern.starts_with('*');
let anchored_end = !pattern.ends_with('*');
let mut rest = text;
let mut first = true;
for part in parts {
if part.is_empty() {
first = false;
continue;
}
match rest.find(part) {
Some(idx) if !first || !anchored_start || idx == 0 => {
rest = &rest[idx + part.len()..];
}
_ => return false,
}
first = false;
}
// After the last literal, a pattern not ending in `*` must end exactly
// there (nothing left over).
!anchored_end || rest.is_empty()
}
/// Split one claude-shaped permission pattern (`Bash(git push*)`, `Write`,
/// …) into its tool name and subject glob. A bare name carries the glob `*`.
fn split_pattern(pattern: &str) -> (&str, &str) {
match pattern.split_once('(') {
Some((name, rest)) => (name.trim(), rest.strip_suffix(')').unwrap_or(rest)),
None => (pattern.trim(), "*"),
}
}
/// Whether one permission pattern's tool name maps to this ACP `kind` at
/// all, regardless of subject. Separate from [`pattern_matches`] because a
/// pattern that COVERS a call but cannot be evaluated against it is a
/// refusal, not a pass.
fn pattern_covers_kind(pattern: &str, kind: &str) -> bool {
let (name, _) = split_pattern(pattern);
TOOL_NAME_KINDS
.iter()
.find(|(n, _)| n.eq_ignore_ascii_case(name))
.is_some_and(|(_, kinds)| kinds.contains(&kind))
}
/// Whether one claude-shaped permission pattern (`Bash(git push*)`,
/// `Write`, …) covers an ACP tool call of `kind` with `subject`.
fn pattern_matches(pattern: &str, kind: &str, subject: &str) -> bool {
let (_, glob) = split_pattern(pattern);
if pattern_covers_kind(pattern, kind) {
wildcard_match(glob, subject)
} else {
false
}
}
/// The seam: decide one permission request from the [`SessionSpec`]. Deny
/// rules are evaluated before the read-only posture so the recorded reason
/// names the most specific rule that fired.
///
/// Fails CLOSED on missing wire fields. ACP v1 leaves `title` optional and
/// `rawInput` free-form, so a peer can send a tool call with no subject at
/// all; every glob then matches nothing and the deny list silently vacates.
/// The same holds for the read-only posture when the peer omits `kind`
/// (it arrives as `"other"`, which `MUTATING_KINDS` cannot classify). In
/// both cases the guard cannot be evaluated, so the call is refused and the
/// reason names the field the peer left out.
fn decide_permission(spec: &SessionSpec, call: &ToolCallInfo) -> PermissionDecision {
if call.subject.trim().is_empty() {
if let Some(pattern) = spec
.disallowed_tools
.iter()
.find(|pattern| pattern_covers_kind(pattern, &call.kind))
{
return PermissionDecision::Deny(format!(
"tool call carries no subject (no rawInput command or path, no locations[0].path, \
no title), so deny pattern {pattern:?} for ACP kind {:?} cannot be evaluated",
call.kind
));
}
}
for pattern in &spec.disallowed_tools {
if pattern_matches(pattern, &call.kind, &call.subject) {
return PermissionDecision::Deny(format!(
"matches SessionSpec.disallowed_tools pattern {pattern:?}"
));
}
}
if !spec.writable {
let kind = call.kind.trim();
if kind.is_empty() || kind == "other" {
return PermissionDecision::Deny(format!(
"read-only session (writable: false): tool call carries no usable ACP kind \
({kind:?}), so whether it mutates the filesystem cannot be decided"
));
}
if MUTATING_KINDS.contains(&kind) {
return PermissionDecision::Deny(format!(
"read-only session (writable: false): ACP kind {:?} mutates the filesystem",
call.kind
));
}
}
PermissionDecision::Allow
}
/// Build the JSON-RPC result answering a permission request. `Allow` picks
/// the first `allow_once` option, then `allow_always`, then the first
/// option at all; `Deny` picks the first `reject_once`, then
/// `reject_always`, and falls back to the `cancelled` outcome when the peer
/// offered no reject option (the only refusal the stable schema guarantees).
fn permission_response(decision: &PermissionDecision, options: &[Value]) -> Value {
let option_kind = |opt: &Value| -> String {
opt.get("kind")
.and_then(Value::as_str)
.unwrap_or("")
.to_string()
};
let option_id = |opt: &Value| opt.get("optionId").cloned().unwrap_or(Value::Null);
let pick = |kinds: &[&str]| -> Option<Value> {
options
.iter()
.find(|opt| kinds.contains(&option_kind(opt).as_str()))
.map(option_id)
};
let selected = match decision {
PermissionDecision::Allow => pick(&["allow_once"])
.or_else(|| pick(&["allow_always"]))
.or_else(|| options.first().map(option_id)),
PermissionDecision::Deny(_) => pick(&["reject_once"]).or_else(|| pick(&["reject_always"])),
};
match selected {
Some(option_id) => json!({ "outcome": { "outcome": "selected", "optionId": option_id } }),
None => match decision {
PermissionDecision::Allow => {
// No allow option at all: the only safe answer is refusal.
json!({ "outcome": { "outcome": "cancelled" } })
}
PermissionDecision::Deny(_) => json!({ "outcome": { "outcome": "cancelled" } }),
},
}
}
// ---------------------------------------------------------------------------
// Event mapping
// ---------------------------------------------------------------------------
/// Keep at most `max` characters (not bytes — never splits a code point).
fn truncate_chars(text: &str, max: usize) -> String {
if text.chars().count() <= max {
text.to_string()
} else {
text.chars().take(max).collect()
}
}
/// Last `max` characters of `text` (for stderr tails in failure messages).
fn last_chars(text: &str, max: usize) -> String {
let chars: Vec<char> = text.chars().collect();
let start = chars.len().saturating_sub(max);
chars[start..].iter().collect()
}
/// Summarize a tool call's content for a [`AgentEvent::ToolResult`]: the
/// first text content block, else the updated title, else the status.
fn tool_result_summary(update: &Value, tracked: &ToolCallInfo, status: &str) -> String {
if let Some(content) = update.get("content").and_then(Value::as_array) {
for item in content {
if item.get("type").and_then(Value::as_str) == Some("content") {
if let Some(text) = item
.get("content")
.and_then(|c| c.get("text"))
.and_then(Value::as_str)
{
return truncate_chars(text, SUMMARY_MAX_CHARS);
}
}
}
}
if !tracked.title.is_empty() {
return truncate_chars(&tracked.title, SUMMARY_MAX_CHARS);
}
status.to_string()
}
// ---------------------------------------------------------------------------
// Backend
// ---------------------------------------------------------------------------
/// The [`AgentBackend`] for an ACP-speaking agent executable (KRZ-301).
///
/// There is no canonical ACP binary name and no `--version` convention, so
/// there is deliberately no discovery/probe: the configured command is
/// spawned directly and the `initialize` handshake IS the probe — a
/// non-ACP executable fails there, loudly, before any prompt turn.
#[derive(Debug, Clone)]
pub struct AcpBackend {
program: PathBuf,
args: Vec<String>,
}
impl AcpBackend {
/// Spawn `program args` as the ACP agent (no validation performed; the
/// handshake at session start is the validation).
pub fn new(program: impl Into<PathBuf>, args: Vec<String>) -> Self {
AcpBackend {
program: program.into(),
args,
}
}
/// The program this backend spawns.
pub fn program(&self) -> &std::path::Path {
&self.program
}
}
#[async_trait::async_trait]
impl AgentBackend for AcpBackend {
async fn start(&self, spec: SessionSpec) -> Result<Box<dyn AgentSession>> {
if spec.resume.is_some() {
return Err(EngineError::Backend(
"acp backend: resume is unsupported (session/load is an optional v1 \
capability this backend does not negotiate)"
.to_string(),
));
}
let model = spec.model.clone();
let mut command = tokio::process::Command::new(&self.program);
command
.args(&self.args)
.current_dir(&spec.cwd)
// agent-env-clear: CLEARED env from the minimal allowlist. ACP
// defines no canonical auth env var, so NO ambient credential is
// injected (auth_env_name None) — the peer authenticates from
// its own config or fails loudly, never by inheritance.
.env_clear()
.envs(crate::agent_env::agent_session_env(
&spec.env,
&spec.session_id,
None,
))
// ACP is bidirectional: stdin carries the client's requests.
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
// Unix: make the child the leader of a fresh process group so aborts
// can kill the whole tree, mirroring `backend_claude::ClaudeBackend`.
#[cfg(unix)]
command.process_group(0);
let mut child = command.spawn().map_err(|e| {
EngineError::Backend(format!(
"failed to spawn acp agent {}: {e}",
self.program.display()
))
})?;
// Windows: kill-on-close Job Object, mirroring `backend_claude`.
#[cfg(windows)]
let job = match child.raw_handle() {
Some(handle) => match win_job::JobHandle::create_and_assign(handle) {
Ok(job) => Some(job),
Err(e) => {
tracing::warn!(error = %e, "failed to create Job Object for acp child; \
tree-kill on abort will be unavailable");
None
}
},
None => None,
};
let stdin = child
.stdin
.take()
.ok_or_else(|| EngineError::Backend("acp child has no stdin pipe".to_string()))?;
let stdout = child
.stdout
.take()
.ok_or_else(|| EngineError::Backend("acp child has no stdout pipe".to_string()))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| EngineError::Backend("acp child has no stderr pipe".to_string()))?;
// Capture stderr concurrently so a chatty child never blocks on a
// full pipe and failure messages can include the tail. The stream is
// drained to EOF but only a bounded tail is retained — a noisy or
// malicious peer must not exhaust host memory (stream_bounds).
let stderr_buf = Arc::new(Mutex::new(String::new()));
let stderr_task = {
let buf = Arc::clone(&stderr_buf);
tokio::spawn(async move {
let tail = drain_to_tail(stderr, STDERR_TAIL_CAP).await;
*buf.lock().expect("stderr buffer lock") = tail;
})
};
let mut session = AcpSession {
session_id: spec.session_id.clone(),
acp_session_id: None,
model,
spec,
child,
#[cfg(windows)]
job,
stdin,
lines: BoundedLines::new(stdout),
stderr_buf,
stderr_task: Some(stderr_task),
queue: VecDeque::new(),
next_request_id: 1,
prompt_request_id: None,
tool_calls: HashMap::new(),
message_text: String::new(),
message_id: None,
last_usage: None,
saw_result: false,
saw_success_result: false,
exit: None,
};
// The handshake is eager: a non-ACP executable (or a hung peer)
// fails start() loudly rather than mid-run. A handshake failure
// kills the child before the error crosses back.
if let Err(e) = session.handshake().await {
session.kill_child().await;
return Err(e);
}
Ok(Box::new(session))
}
}
// ---------------------------------------------------------------------------
// Session
// ---------------------------------------------------------------------------
/// A live ACP session (the [`AgentSession`] impl).
///
/// Reading is INLINE in [`AcpSession::next_event`] (no reader task), which
/// keeps memory bounded by [`BoundedLines`] and is deadlock-free for this
/// protocol: the peer blocks waiting for each permission answer, and the
/// read loop answers permission requests synchronously as they arrive, so
/// stdin writes never wait on a peer that is itself waiting on a full
/// stdout pipe.
pub struct AcpSession {
session_id: String,
/// The peer-issued session id (`session/new` response).
acp_session_id: Option<String>,
model: String,
/// Kept for permission decisions (`disallowed_tools`, `writable`).
spec: SessionSpec,
child: Child,
#[cfg(windows)]
job: Option<win_job::JobHandle>,
stdin: ChildStdin,
lines: BoundedLines<ChildStdout>,
stderr_buf: Arc<Mutex<String>>,
stderr_task: Option<JoinHandle<()>>,
/// Converted events not yet surfaced; popped one per `next_event`.
queue: VecDeque<AgentEvent>,
next_request_id: u64,
/// The in-flight `session/prompt` request id; its response synthesizes
/// the terminal `Result`. `None` outside a prompt turn.
prompt_request_id: Option<u64>,
/// Tracked tool calls by `toolCallId` (kind/title/subject), so a
/// `tool_call_update` or permission request resolves to what is known
/// about the call.
tool_calls: HashMap<String, ToolCallInfo>,
/// Accumulated text of the CURRENT assistant message (chunks with the
/// same `messageId` concatenate; a changed/missing-`messageId` boundary
/// starts a new message, and the last message wins the terminal Result,
/// mirroring `backend_codex`'s last-agent_message rule).
message_text: String,
message_id: Option<String>,
/// Latest `usage_update` (context state + optional cumulative USD cost);
/// its cost lands on the terminal `Result` (see module docs).
last_usage: Option<Value>,
saw_result: bool,
saw_success_result: bool,
exit: Option<SessionExit>,
}
#[cfg(unix)]
impl Drop for AcpSession {
fn drop(&mut self) {
crate::backend_claude::kill_unreaped_group(&self.child);
}
}
impl AcpSession {
// -- wire helpers -------------------------------------------------------
/// Serialize one JSON-RPC message as a single NDJSON line on the peer's
/// stdin. Keys are inserted in sorted order by serde_json's default map,
/// which also puts `"id"` first — mock peers and debugging tools rely on
/// nothing more than JSON semantics, but stable key order keeps captured
/// transcripts diffable.
async fn write_message(&mut self, message: Value) -> Result<()> {
use tokio::io::AsyncWriteExt;
let mut line = serde_json::to_string(&message)
.map_err(|e| EngineError::Backend(format!("failed to encode acp message: {e}")))?;
line.push('\n');
self.stdin.write_all(line.as_bytes()).await.map_err(|e| {
EngineError::Backend(format!("failed to write to acp agent stdin: {e}"))
})?;
self.stdin
.flush()
.await
.map_err(|e| EngineError::Backend(format!("failed to flush acp agent stdin: {e}")))?;
Ok(())
}
/// Send a client request and return its id (the caller either awaits the
/// response via [`AcpSession::pump_until_response`] or, for
/// `session/prompt`, leaves it to the `next_event` loop).
async fn send_request(&mut self, method: &str, params: Value) -> Result<u64> {
let id = self.next_request_id;
self.next_request_id += 1;
self.write_message(json!({
"jsonrpc": "2.0",
"id": id,
"method": method,
"params": params,
}))
.await?;
Ok(id)
}
/// `initialize` + `session/new`, then the first `session/prompt` (its
/// response streams in through `next_event` like any later turn).
async fn handshake(&mut self) -> Result<()> {
let init_id = self
.send_request(
method::INITIALIZE,
json!({
"protocolVersion": ACP_PROTOCOL_VERSION,
"clientCapabilities": {
// fs/terminal unsupported: a conformant peer never
// calls fs/* or terminal/*; one that does is answered
// with -32601 rather than silently served.
"fs": { "readTextFile": false, "writeTextFile": false },
"terminal": false,
},
"clientInfo": {
"name": "kranz",
"title": "kranz mission engine",
"version": env!("CARGO_PKG_VERSION"),
},
}),
)
.await?;
let init_result = self.pump_until_response(init_id, HANDSHAKE_TIMEOUT).await?;
let peer_version = init_result
.get("protocolVersion")
.and_then(Value::as_u64)
.unwrap_or(0);
if peer_version != ACP_PROTOCOL_VERSION {
return Err(EngineError::Backend(format!(
"acp agent negotiated protocol version {peer_version}, but this backend speaks \
only stable version {ACP_PROTOCOL_VERSION} (schema v1)"
)));
}
let new_id = self
.send_request(
method::SESSION_NEW,
json!({
"cwd": self.spec.cwd.display().to_string(),
"mcpServers": [],
}),
)
.await?;
let new_result = self.pump_until_response(new_id, HANDSHAKE_TIMEOUT).await?;
let acp_session_id = new_result
.get("sessionId")
.and_then(Value::as_str)
.ok_or_else(|| {
EngineError::Backend("acp session/new response carried no sessionId".to_string())
})?
.to_string();
self.acp_session_id = Some(acp_session_id.clone());
self.session_id = acp_session_id.clone();
// Init is synthesized from the handshake (no wire line carries it),
// mirroring backend_kimi: the configured model is recorded for
// attribution — ACP v1 has no model-reporting field.
self.queue.push_back(AgentEvent::Init {
session_id: acp_session_id,
model: self.model.clone(),
raw: json!({
"initialize": init_result,
"sessionNew": new_result,
"synthesizedBy": "backend_acp",
}),
});
let prompt_text = match &self.spec.prompt {
PromptMode::SingleShot(text) | PromptMode::Streaming(text) => text.clone(),
};
self.send_prompt(&prompt_text).await
}
/// Send one `session/prompt` request and mark its id as the turn whose
/// response synthesizes the terminal `Result`.
async fn send_prompt(&mut self, text: &str) -> Result<()> {
let acp_session_id = self
.acp_session_id
.clone()
.ok_or_else(|| EngineError::Backend("acp session not established yet".to_string()))?;
let id = self
.send_request(
method::SESSION_PROMPT,
json!({
"sessionId": acp_session_id,
"prompt": [ { "type": "text", "text": text } ],
}),
)
.await?;
self.prompt_request_id = Some(id);
// A new turn begins: the terminal Result stitches only this turn's
// last message.
self.message_text.clear();
self.message_id = None;
Ok(())
}
/// Read frames until the response to `id` arrives, converting everything
/// else through the normal frame path (notifications become queued
/// events; permission requests are answered). Used by the handshake —
/// the only place a response is awaited synchronously.
async fn pump_until_response(
&mut self,
id: u64,
timeout: std::time::Duration,
) -> Result<Value> {
let pump = async {
loop {
let frame = match self.read_frame().await? {
Some(frame) => frame,
None => {
return Err(EngineError::Backend(format!(
"acp agent closed stdout before answering request id {id}; \
stderr tail: {}",
self.stderr_tail()
)))
}
};
match frame {
Frame::Response {
id: response_id,
outcome,
} if response_id == id => {
return match outcome {
RpcOutcome::Result(result) => Ok(result),
RpcOutcome::Error(error) => Err(EngineError::Backend(format!(
"acp request id {id} failed: {error}"
))),
};
}
other => self.handle_frame(other).await?,
}
}
};
match tokio::time::timeout(timeout, pump).await {
Ok(result) => result,
Err(_) => Err(EngineError::Backend(format!(
"acp agent did not answer request id {id} within {}s (handshake timeout)",
timeout.as_secs()
))),
}
}
/// Read and classify the next stdout line; `None` at EOF. Blank lines
/// are skipped (NDJSON tolerates them; a peer's pretty-printing or
/// keepalive must not fabricate events).
async fn read_frame(&mut self) -> Result<Option<Frame>> {
loop {
match self.lines.next_line().await {
Ok(Some(line)) if line.trim().is_empty() => continue,
Ok(Some(line)) => return Ok(Some(classify_line(&line))),
Ok(None) => return Ok(None),
Err(e) => {
return Err(EngineError::Backend(format!(
"error reading acp agent stdout: {e}; stderr tail: {}",
self.stderr_tail()
)))
}
}
}
}
/// Convert one frame into queued events and side effects (permission
/// answers, tool tracking, usage capture, terminal-Result synthesis).
async fn handle_frame(&mut self, frame: Frame) -> Result<()> {
match frame {
Frame::Notification {
method,
params,
raw,
} => {
if method == method::SESSION_UPDATE {
self.handle_session_update(¶ms, raw);
} else {
self.queue.push_back(AgentEvent::Other { raw });
}
}
Frame::Request {
id,
method,
params,
raw,
} => {
if method == method::REQUEST_PERMISSION {
self.handle_permission_request(id, ¶ms, raw).await?;
} else {
// A client capability we did not advertise (fs/*,
// terminal/*, elicitation/*): refuse with JSON-RPC
// -32601 rather than silently serving or hanging.
self.write_message(json!({
"jsonrpc": "2.0",
"id": id,
"error": {
"code": -32601,
"message": format!("kranz acp backend does not support {method:?}"),
},
}))
.await?;
self.queue.push_back(AgentEvent::Other { raw });
}
}
Frame::Response { id, outcome } => {
if Some(id) == self.prompt_request_id {
self.prompt_request_id = None;
self.synthesize_result(outcome, id);
} else {
// A response to nothing outstanding (a straggler from a
// cancelled turn, or a peer bug): transcript only.
self.queue.push_back(AgentEvent::Other {
raw: match outcome {
RpcOutcome::Result(result) => {
json!({ "unmatchedResponse": { "id": id, "result": result } })
}
RpcOutcome::Error(error) => {
json!({ "unmatchedResponse": { "id": id, "error": error } })
}
},
});
}
}
Frame::Unrecognized(raw) => {
self.queue.push_back(AgentEvent::Other { raw });
}
}
Ok(())
}
/// Map one `session/update` notification onto events (see module docs).
fn handle_session_update(&mut self, params: &Value, raw: Value) {
let update = params.get("update").cloned().unwrap_or(Value::Null);
match update.get("sessionUpdate").and_then(Value::as_str) {
Some("agent_message_chunk") => {
let text = update
.get("content")
.and_then(|c| c.get("text"))
.and_then(Value::as_str)
.unwrap_or("");
if text.is_empty() {
self.queue.push_back(AgentEvent::Other { raw });
return;
}
// Message boundaries: a changed messageId starts a new
// message; the LAST message wins the terminal Result.
let chunk_id = update
.get("messageId")
.and_then(Value::as_str)
.map(str::to_string);
if chunk_id.is_some() && chunk_id != self.message_id {
self.message_text.clear();
self.message_id = chunk_id;
}
self.message_text.push_str(text);
self.queue.push_back(AgentEvent::Text {
text: text.to_string(),
raw,
});
}
Some("tool_call") => {
let id = update
.get("toolCallId")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let kind = update
.get("kind")
.and_then(Value::as_str)
.unwrap_or("other")
.to_string();
let title = update
.get("title")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let subject = tool_call_subject(&kind, &title, &update);
self.tool_calls.insert(
id,
ToolCallInfo {
kind: kind.clone(),
title: title.clone(),
subject,
},
);
self.queue.push_back(AgentEvent::ToolUse {
tool: kind,
summary: truncate_chars(&title, SUMMARY_MAX_CHARS),
raw,
});
}
Some("tool_call_update") => {
let id = update
.get("toolCallId")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
let status = update
.get("status")
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
{
let tracked = self.tool_calls.entry(id).or_default();
if let Some(kind) = update.get("kind").and_then(Value::as_str) {
tracked.kind = kind.to_string();
}
if let Some(title) = update.get("title").and_then(Value::as_str) {
tracked.title = title.to_string();
}
}
match status.as_str() {
// Terminal statuses surface as first-class ToolResult
// events; progress updates (pending/in_progress) are
// transcript-only. `failed` is a normal failure, NOT a
// denial (mirrors backend_codex) — denials are
// synthesized at the permission seam.
"completed" | "failed" => {
let tracked = self
.tool_calls
.get(
update
.get("toolCallId")
.and_then(Value::as_str)
.unwrap_or_default(),
)
.cloned()
.unwrap_or_default();
self.queue.push_back(AgentEvent::ToolResult {
tool: Some(tracked.kind.clone()),
denied: false,
summary: tool_result_summary(&update, &tracked, &status),
raw,
});
}
_ => self.queue.push_back(AgentEvent::Other { raw }),
}
}
Some("usage_update") => {
// Remembered for the terminal Result's cost; the update
// itself is transcript-only (no AgentEvent kind for
// mid-stream usage, and `used`/`size` are context-window
// state, not a billable token split).
self.last_usage = Some(update.clone());
self.queue.push_back(AgentEvent::Other { raw });
}
_ => self.queue.push_back(AgentEvent::Other { raw }),
}
}
/// Answer one `session/request_permission` at the seam; a refusal is
/// synthesized into a `denied` ToolResult so the guardrail firing is a
/// first-class event (the peer may report nothing itself).
async fn handle_permission_request(
&mut self,
id: Value,
params: &Value,
raw: Value,
) -> Result<()> {
let call_update = params.get("toolCall").cloned().unwrap_or(Value::Null);
let call_id = call_update
.get("toolCallId")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
// Merge what the request carries with what the tool_call/update
// stream already told us about this call.
let mut info = self.tool_calls.get(&call_id).cloned().unwrap_or_default();
if let Some(kind) = call_update.get("kind").and_then(Value::as_str) {
info.kind = kind.to_string();
}
if info.kind.is_empty() {
info.kind = "other".to_string();
}
if let Some(title) = call_update.get("title").and_then(Value::as_str) {
info.title = title.to_string();
}
if info.subject.is_empty() {
info.subject = tool_call_subject(&info.kind, &info.title, &call_update);
}
self.tool_calls.insert(call_id.clone(), info.clone());
let decision = decide_permission(&self.spec, &info);
let options = params
.get("options")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let result = permission_response(&decision, &options);
self.write_message(json!({
"jsonrpc": "2.0",
"id": id,
"result": result,
}))
.await?;
if let PermissionDecision::Deny(reason) = &decision {
tracing::info!(
session_id = %self.session_id,
tool_call_id = %call_id,
kind = %info.kind,
decision = "deny",
reason = %reason,
"acp permission request refused at the kranz seam"
);
self.queue.push_back(AgentEvent::ToolResult {
tool: Some(info.kind.clone()),
denied: true,
summary: truncate_chars(
&format!("refused by kranz permission seam: {reason}"),
SUMMARY_MAX_CHARS,
),
raw,
});
} else {
// The approval itself is not an event — the peer's own
// tool_call_update reports the outcome. The request line stays
// in the transcript.
self.queue.push_back(AgentEvent::Other { raw });
}
Ok(())
}
/// Synthesize the terminal `Result` from a `session/prompt` response
/// (see module docs): text stitched from the turn's last assistant
/// message, `is_error` ⇔ `stopReason != "end_turn"`, cost only when the
/// peer reported a USD amount — absent data stays absent.
///
/// WHY non-`end_turn` is an error, not just `refusal` (12th-pass review):
/// ACP v1's stop reasons are `end_turn` (natural completion), `refusal`,
/// `max_tokens`, `max_turn_requests`, and `cancelled`. Mapping only
/// `refusal` to `is_error` let a turn cut short by `max_tokens`/
/// `max_turn_requests` — or answered `cancelled`, or carrying a missing
/// or unrecognized reason — surface as a SUCCESSFUL result, so a
/// truncated validator report could pass validation. Fail-closed is the
/// only honest mapping: anything but a natural completion is an error,
/// and the reason string rides the raw payload so the failure is
/// diagnosable (`null` when the peer omitted the field entirely).
fn synthesize_result(&mut self, outcome: RpcOutcome, request_id: u64) {
let last_cost_usd = self
.last_usage
.as_ref()
.and_then(|u| u.get("cost"))
.filter(|cost| {
cost.get("currency").and_then(Value::as_str) == Some("USD")
&& cost.get("amount").and_then(Value::as_f64).is_some()
})
.and_then(|cost| cost.get("amount").and_then(Value::as_f64));
let (text, is_error, raw) = match outcome {
RpcOutcome::Result(result) => {
let stop_reason = result.get("stopReason").and_then(Value::as_str);
(
std::mem::take(&mut self.message_text),
stop_reason != Some("end_turn"),
json!({
"promptResponse": result,
"usageUpdate": self.last_usage,
"synthesizedBy": "backend_acp",
// The classification input, verbatim: exactly what the
// peer sent, `null` when it sent nothing — the raw
// payload always explains WHY a non-end_turn failed.
"stopReason": stop_reason,
}),
)
}
RpcOutcome::Error(error) => (
format!("acp session/prompt failed: {error}"),
true,
json!({
"promptError": error,
"requestId": request_id,
"synthesizedBy": "backend_acp",
}),
),
};
let event = AgentEvent::Result {
text,
is_error,
usage: TokenUsage::default(),
cost_usd: last_cost_usd,
num_turns: Some(1),
raw,
};
self.observe(&event);
self.queue.push_back(event);
}
fn observe(&mut self, event: &AgentEvent) {
if let AgentEvent::Result { is_error, .. } = event {
self.saw_result = true;
if !is_error {
self.saw_success_result = true;
}
}
}
/// Kill the child and reap it, best-effort; also joins the stderr
/// capture task. Mirrors `backend_kimi::KimiSession::kill_child`
/// exactly: unix process-group SIGKILL (with a post-reap sweep for
/// stragglers that raced a mid-fork), windows kill-on-close Job Object.
async fn kill_child(&mut self) {
#[cfg(unix)]
{
let pgid = self
.child
.id()
.and_then(|pid| i32::try_from(pid).ok())
.filter(|pid| *pid > 0);
let group_killed = matches!(pgid, Some(pgid) if kill_group(pgid));
if !group_killed {
let _ = self.child.start_kill();
}
let _ = self.child.wait().await;
if group_killed {
if let Some(pgid) = pgid {
let _ = kill_group(pgid);
}
}
}
#[cfg(windows)]
{
match &self.job {
Some(job) => job.kill(),
None => {
let _ = self.child.start_kill();
}
}
let _ = self.child.wait().await;
}
#[cfg(all(not(unix), not(windows)))]
{
let _ = self.child.start_kill();
let _ = self.child.wait().await;
}
if let Some(task) = self.stderr_task.take() {
let _ = task.await;
}
}
async fn finish_at_eof(&mut self) {
let status = self.child.wait().await;
if let Some(task) = self.stderr_task.take() {
let _ = task.await;
}
let exit = match status {
Ok(status) if status.success() && self.saw_result => SessionExit::Completed,
Ok(status) => SessionExit::Failed(format!(
"acp agent exited with {status}{}; stderr tail: {}",
if self.saw_result {
""
} else {
" without answering session/prompt"
},
self.stderr_tail(),
)),
Err(e) => SessionExit::Failed(format!(
"failed to reap acp agent process: {e}; stderr tail: {}",
self.stderr_tail(),
)),
};
self.exit = Some(exit);
}
fn stderr_tail(&self) -> String {
let captured = self
.stderr_buf
.lock()
.map(|guard| guard.clone())
.unwrap_or_default();
last_chars(captured.trim_end(), STDERR_TAIL_CHARS)
}
}
#[async_trait::async_trait]
impl AgentSession for AcpSession {
fn session_id(&self) -> String {
self.session_id.clone()
}
async fn next_event(&mut self) -> Result<Option<AgentEvent>> {
loop {
if let Some(event) = self.queue.pop_front() {
return Ok(Some(event));
}
if self.exit.is_some() {
return Ok(None);
}
let frame = match self.read_frame().await {
Ok(Some(frame)) => frame,
Ok(None) => {
self.finish_at_eof().await;
return Ok(None);
}
Err(e) => {
self.kill_child().await;
self.exit = Some(SessionExit::Failed(e.to_string()));
return Ok(None);
}
};
if let Err(e) = self.handle_frame(frame).await {
// A transport error mid-session (stdin write failed — the
// peer is gone): fail the session honestly rather than hang.
self.kill_child().await;
self.exit = Some(SessionExit::Failed(e.to_string()));
return Ok(None);
}
}
}
async fn send_user_message(&mut self, text: &str) -> Result<()> {
if self.exit.is_some() {
return Err(EngineError::Backend(
"acp session is closed; cannot send further messages".to_string(),
));
}
if self.acp_session_id.is_none() {
return Err(EngineError::Backend(
"acp session not established yet; cannot send a message".to_string(),
));
}
self.send_prompt(text).await
}
async fn abort(&mut self) -> Result<()> {
// Best-effort graceful cancel first (the peer MAY stop its turn
// cleanly and answer the prompt with stopReason "cancelled"), then
// the house tree-kill regardless — abort must never depend on the
// peer honoring the notification.
if let Some(acp_session_id) = self.acp_session_id.clone() {
let _ = self
.write_message(json!({
"jsonrpc": "2.0",
"method": method::SESSION_CANCEL,
"params": { "sessionId": acp_session_id },
}))
.await;
}
let already_exited = matches!(self.child.try_wait(), Ok(Some(_)));
self.kill_child().await;
if self.saw_success_result && already_exited {
self.exit = Some(SessionExit::Completed);
} else {
self.exit = Some(SessionExit::Aborted);
}
Ok(())
}
fn exit_status(&self) -> Option<SessionExit> {
self.exit.clone()
}
}
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn backend_acp_classify_distinguishes_response_request_notification() {
let response =
classify_line(r#"{"jsonrpc":"2.0","id":3,"result":{"stopReason":"end_turn"}}"#);
assert!(matches!(
response,
Frame::Response {
id: 3,
outcome: RpcOutcome::Result(_)
}
));
let error =
classify_line(r#"{"jsonrpc":"2.0","id":4,"error":{"code":-32603,"message":"boom"}}"#);
assert!(matches!(
error,
Frame::Response {
id: 4,
outcome: RpcOutcome::Error(_)
}
));
let request = classify_line(
r#"{"jsonrpc":"2.0","id":100,"method":"session/request_permission","params":{}}"#,
);
assert!(matches!(
request,
Frame::Request { ref method, .. } if method == "session/request_permission"
));
let notification = classify_line(
r#"{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"s","update":{"sessionUpdate":"plan"}}}"#,
);
assert!(matches!(
notification,
Frame::Notification { ref method, .. } if method == "session/update"
));
// Garbage is never a hard failure — transcript only.
let torn = classify_line(r#"{"jsonrpc":"2.0","method":"session/upda"#);
assert!(matches!(torn, Frame::Unrecognized(_)));
}
#[test]
fn backend_acp_wildcard_match_anchors_like_a_shell_glob() {
assert!(wildcard_match("git push*", "git push origin main"));
assert!(wildcard_match("git push*", "git push"));
assert!(!wildcard_match("git push*", "git pull"));
assert!(wildcard_match("*", "anything"));
assert!(wildcard_match(
"cargo * --workspace",
"cargo test --workspace"
));
assert!(!wildcard_match(
"cargo * --workspace",
"cargo test --package x"
));
assert!(wildcard_match("*/etc/passwd", "/etc/passwd"));
assert!(!wildcard_match("*/etc/passwd", "/etc/passwd.bak"));
}
#[test]
fn backend_acp_pattern_matches_maps_claude_names_to_acp_kinds() {
assert!(pattern_matches(
"Bash(git push*)",
"execute",
"git push origin main"
));
assert!(!pattern_matches("Bash(git push*)", "execute", "git pull"));
assert!(!pattern_matches(
"Bash(git push*)",
"edit",
"git push origin main"
));
assert!(pattern_matches("Write", "edit", "/repo/src/main.rs"));
assert!(!pattern_matches("Write", "read", "/repo/src/main.rs"));
assert!(pattern_matches("Edit(/etc/*)", "edit", "/etc/hosts"));
assert!(pattern_matches("Read", "read", "/anywhere"));
// Unknown tool names match nothing (deny-only mapping; never widens).
assert!(!pattern_matches("NotAClaudeTool(*)", "execute", "x"));
}
fn spec_with(writable: bool, disallowed: &[&str]) -> SessionSpec {
SessionSpec {
cwd: PathBuf::from("."),
prompt: PromptMode::SingleShot("do the thing".to_string()),
append_system_prompt: None,
model: "acp-model".to_string(),
effort: "high".to_string(),
session_id: "sess-1".to_string(),
resume: None,
permission_mode: None,
allowed_tools: vec![],
disallowed_tools: disallowed.iter().map(|s| s.to_string()).collect(),
tools: vec![],
writable,
settings_json: None,
json_schema: None,
max_budget_usd: None,
max_turns: None,
env: Default::default(),
sandbox: None,
hook_status: None,
}
}
#[test]
fn backend_acp_permission_denies_disallowed_and_mutating_kinds() {
let spec = spec_with(true, &["Bash(git push*)"]);
let push = ToolCallInfo {
kind: "execute".to_string(),
title: "git push origin main".to_string(),
subject: "git push origin main".to_string(),
};
assert!(matches!(
decide_permission(&spec, &push),
PermissionDecision::Deny(reason) if reason.contains("Bash(git push*)")
));
let test = ToolCallInfo {
kind: "execute".to_string(),
title: "cargo test".to_string(),
subject: "cargo test".to_string(),
};
assert_eq!(decide_permission(&spec, &test), PermissionDecision::Allow);
// Read-only posture: mutating kinds refused, execute/read allowed.
let ro = spec_with(false, &[]);
let edit = ToolCallInfo {
kind: "edit".to_string(),
title: "write src/main.rs".to_string(),
subject: "/repo/src/main.rs".to_string(),
};
assert!(matches!(
decide_permission(&ro, &edit),
PermissionDecision::Deny(reason) if reason.contains("writable: false")
));
assert_eq!(decide_permission(&ro, &test), PermissionDecision::Allow);
let read = ToolCallInfo {
kind: "read".to_string(),
title: "read src/main.rs".to_string(),
subject: "/repo/src/main.rs".to_string(),
};
assert_eq!(decide_permission(&ro, &read), PermissionDecision::Allow);
}
/// The permission seam used to fail OPEN when the peer omitted the
/// subject: `wildcard_match("git push*", "")` is false, so no deny fired
/// and the decision was Allow. Every glob-carrying deny rule was
/// bypassable that way, by a peer that need not even be hostile.
#[test]
fn backend_acp_permission_denies_when_the_subject_is_missing() {
let spec = spec_with(true, &["Bash(git push*)"]);
let no_subject = ToolCallInfo {
kind: "execute".to_string(),
title: String::new(),
subject: String::new(),
};
assert!(
matches!(
decide_permission(&spec, &no_subject),
PermissionDecision::Deny(ref reason)
if reason.contains("no subject") && reason.contains("Bash(git push*)")
),
"got {:?}",
decide_permission(&spec, &no_subject)
);
// Whitespace is no subject either.
let blank_subject = ToolCallInfo {
subject: " ".to_string(),
..no_subject.clone()
};
assert!(matches!(
decide_permission(&spec, &blank_subject),
PermissionDecision::Deny(_)
));
// Precise: a deny list that does not cover this call's kind is not
// made to fire by a missing subject.
let read_no_subject = ToolCallInfo {
kind: "read".to_string(),
..no_subject.clone()
};
assert_eq!(
decide_permission(&spec_with(true, &["Bash(git push*)"]), &read_no_subject),
PermissionDecision::Allow
);
}
/// Read-only posture: `MUTATING_KINDS` cannot classify a call whose kind
/// the peer omitted (it arrives as `"other"`), so the containment claim
/// cannot be checked and the call is refused.
#[test]
fn backend_acp_read_only_denies_an_unclassifiable_kind() {
let ro = spec_with(false, &[]);
for kind in ["", "other"] {
let call = ToolCallInfo {
kind: kind.to_string(),
title: "do something".to_string(),
subject: "/repo/src/main.rs".to_string(),
};
assert!(
matches!(
decide_permission(&ro, &call),
PermissionDecision::Deny(ref reason) if reason.contains("kind")
),
"kind {kind:?} got {:?}",
decide_permission(&ro, &call)
);
}
// A writable session still allows an unclassified kind: the read-only
// posture is the only thing that turns on the kind.
let writable = spec_with(true, &[]);
let other = ToolCallInfo {
kind: "other".to_string(),
title: "think".to_string(),
subject: "think".to_string(),
};
assert_eq!(
decide_permission(&writable, &other),
PermissionDecision::Allow
);
}
#[test]
fn backend_acp_permission_response_picks_options_or_cancels() {
let options = vec![
json!({ "optionId": "allow-1", "name": "Allow", "kind": "allow_once" }),
json!({ "optionId": "reject-1", "name": "Reject", "kind": "reject_once" }),
];
let allow = permission_response(&PermissionDecision::Allow, &options);
assert_eq!(allow["outcome"]["optionId"], json!("allow-1"));
let deny = permission_response(&PermissionDecision::Deny("nope".to_string()), &options);
assert_eq!(deny["outcome"]["optionId"], json!("reject-1"));
// No reject option offered: refusal degrades to the cancelled
// outcome — never an invented selection.
let deny_no_reject = permission_response(
&PermissionDecision::Deny("nope".to_string()),
&[options[0].clone()],
);
assert_eq!(deny_no_reject["outcome"]["outcome"], json!("cancelled"));
}
}