acp_server.rs - CodeWhale

acp_server.rs

1	//! Minimal Agent Client Protocol stdio adapter.
2	//!
3	//! This starts from the ACP baseline: initialize, new session, prompt, and
4	//! cancel. It keeps stdout protocol-clean for editor clients and routes
5	//! prompts through the same configured provider route as one-shot CLI mode.
6	//!
7	//! `session/new` and `session/load` expose mode/model configuration. Standard
8	//! setters change only the addressed in-memory session between turns. Plan
9	//! uses the shared read-only tool authority and sandbox; changing mode is an
10	//! explicit prompt-prefix invalidation for the next turn. In-flight turns
11	//! keep their frozen prefix and return the existing busy error for setters.
12	//! Configuration is connection-local, not a new durable preference store;
13	//! legacy unscoped `selectModel` changes defaults for future sessions.
14	//!
15	//! `session/prompt` streams the provider response: each text delta is emitted
16	//! as a `session/update` agent_message_chunk as it arrives, instead of buffering
17	//! the whole turn and sending one chunk at the end. The stream is consumed
18	//! concurrently with the input reader so that a `session/cancel` for the same
19	//! session can interrupt the turn mid-stream (returning `stopReason: "cancelled"`)
20	//! instead of being queued behind it. A single writer task is preserved so
21	//! stdout stays protocol-clean.
22	//!
23	//! Each ACP session owns a [`crate::tools::ToolRegistry`] built from the same
24	//! file/search/git/patch/shell tools the CLI `exec` agent and the MCP server
25	//! adapter (`crate::mcp_server`) already use. When the model emits a tool call,
26	//! the turn driver executes it locally through that registry (no duplicate
27	//! filesystem/shell implementation), reports progress to the client as
28	//! `tool_call` / `tool_call_update` session updates, feeds the result back as
29	//! a `tool_result` content block, and re-opens the provider stream so the
30	//! model can keep going until it produces a final answer with no further tool
31	//! calls.
32
33	use std::collections::{HashMap, VecDeque};
34	use std::future::Future;
35	use std::path::PathBuf;
36	use std::sync::Arc;
37	use std::sync::atomic::{AtomicU64, Ordering};
38	use std::time::Duration;
39
40	use anyhow::{Result, anyhow};
41	use futures_util::StreamExt;
42	use serde_json::{Value, json};
43	use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader, Lines};
44	use tokio_util::sync::CancellationToken;
45
46	use crate::client::CodewhaleClient;
47	use crate::config::{ApiProvider, Config};
48	use crate::core::engine::turn_loop::run_tool_call_before_hooks;
49	use crate::core::engine::{
50	AutoReviewPlanDecision, ToolAskRuleDecision, auto_review_plan_decision_for_context,
51	exec_shell_ask_rule_decision_for_policy, file_tool_ask_rule_decision_for_policy,
52	};
53	use crate::llm_client::{LlmClient, StreamEventBox};
54	use crate::tools::spec::{ApprovalRequirement, PreparedToolCall, RichToolResult, ToolError};
55	use crate::tools::{ToolContext, ToolRegistry, ToolRegistryBuilder};
56	use crate::worker_profile::ShellPolicy;
57	use codewhale_config::AppMode;
58	use codewhale_execpolicy::ApprovalMode;
59	use codewhale_models::Role;
60	use codewhale_models::{
61	ContentBlock, ContentBlockStart, Delta, Message, MessageRequest, StreamEvent, SystemPrompt,
62	};
63
64	const ACP_PROTOCOL_VERSION: u64 = 1;
65
66	/// Hard cap on LLM <-> tool round-trips within a single `session/prompt`
67	/// turn. Guards against a model that never stops calling tools; each round is
68	/// one provider stream plus zero or more tool executions.
69	const MAX_ACP_TOOL_ROUNDS: usize = 50;
70
71	/// Maximum number of concurrent sessions kept in memory. When this limit is
72	/// exceeded, the oldest session with no in-flight prompt is evicted.
73	const MAX_ACP_SESSIONS: usize = 64;
74
75	/// A conforming ACP client answers every pending permission request with a
76	/// `cancelled` outcome when the prompt is cancelled. Bound that hand-off so a
77	/// broken client cannot strand the stdio server forever after cancellation.
78	const ACP_PERMISSION_CANCEL_GRACE: Duration = Duration::from_secs(2);
79
80	/// Agent-originated JSON-RPC request ids have their own namespace. Strings
81	/// avoid the client-specific numeric response-id compatibility shim used for
82	/// replies to client-originated requests.
83	static NEXT_ACP_PERMISSION_REQUEST_ID: AtomicU64 = AtomicU64::new(1);
84
85	/// Content is streamed to the model in full (no truncation); this cap only
86	/// bounds how much of a tool's output is echoed into the `tool_call_update`
87	/// notification the editor renders, so a large `File`/`Bash`
88	/// result does not flood the client UI.
89	const TOOL_CALL_CONTENT_PREVIEW_CHARS: usize = 4_000;
90
91	pub async fn run_acp_server(config: Config, model: String, default_cwd: PathBuf) -> Result<()> {
92	let stdin = tokio::io::stdin();
93	let stdout = tokio::io::stdout();
94	let mut reader = BufReader::new(stdin).lines();
95	let mut writer = tokio::io::BufWriter::new(stdout);
96	let mut server = AcpServer::new(config, model, default_cwd);
97
98	while let Some(line) = reader.next_line().await? {
99	if line.trim().is_empty() {
100	continue;
101	}
102
103	let message: Value = match serde_json::from_str(&line) {
104	Ok(value) => value,
105	Err(err) => {
106	write_jsonrpc_error(&mut writer, None, -32700, format!("invalid json: {err}"))
107	.await?;
108	continue;
109	}
110	};
111
112	if message.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
113	write_jsonrpc_error(
114	&mut writer,
115	message
116	.get("id")
117	.cloned()
118	.map(\|id\| server.response_id_policy.response_id(id)),
119	-32600,
120	"jsonrpc version must be 2.0",
121	)
122	.await?;
123	continue;
124	}
125
126	let id = message.get("id").cloned();
127	let method = match message.get("method").and_then(Value::as_str) {
128	Some(method) => method,
129	None if is_jsonrpc_response(&message) => {
130	// A late response to an agent-originated request (most notably
131	// permission after cancellation) has no request semantics and
132	// must not be answered with another JSON-RPC error.
133	continue;
134	}
135	None => {
136	write_jsonrpc_error(
137	&mut writer,
138	id.map(\|id\| server.response_id_policy.response_id(id)),
139	-32600,
140	"missing method",
141	)
142	.await?;
143	continue;
144	}
145	};
146	let params = message.get("params").cloned().unwrap_or_else(\|\| json!({}));
147
148	// `session/prompt` is driven concurrently with the reader so a
149	// `session/cancel` can interrupt the in-flight provider call or a
150	// running tool. Every other method is request/response and handled
151	// synchronously below.
152	if method == "session/prompt" {
153	match server.begin_prompt(params) {
154	Ok(prepared) => {
155	let PreparedPrompt {
156	session_id,
157	messages,
158	cwd,
159	config,
160	model,
161	} = prepared;
162	let response_id_policy = server.response_id_policy;
163	let Some(tool_registry) = server.session_tool_registry(&session_id) else {
164	let id = id.map(\|id\| response_id_policy.response_id(id));
165	write_jsonrpc_error(&mut writer, id, -32603, "unknown sessionId").await?;
166	continue;
167	};
168	// Freeze the first round's fully composed system prompt
169	// for this entire `session/prompt`. Tool calls may edit
170	// AGENTS.md, memory, or configured instruction files, but
171	// self-authored content cannot become same-turn system
172	// authority on a later provider round.
173	let frozen_system_prompt =
174	Arc::new(std::sync::Mutex::new(None::<SystemPrompt>));
175	// The stream-opening closure borrows `&server` only
176	// briefly per round; each returned `StreamEventBox` is
177	// `'static`, so it can be raced against the reader
178	// without holding a borrow on the server across an
179	// await, and the main task keeps exclusive ownership of
180	// stdout.
181	let outcome = run_agentic_prompt_turn(
182	AcpTurnContext {
183	config: &config,
184	model: &model,
185	session_id: &session_id,
186	tool_registry: &tool_registry,
187	response_id_policy,
188	},
189	messages,
190	&mut reader,
191	&mut writer,
192	\|msgs\| {
193	// Rebind to references before the `async move`
194	// block: `async move` moves every path it
195	// touches, and these are already-`Copy`
196	// references, so only `msgs` (the per-round
197	// owned clone) is actually moved in — `server`,
198	// `cwd`, and `tool_registry` stay borrowed from
199	// the enclosing scope across every call this
200	// `FnMut` closure makes.
201	let server = &server;
202	let cwd = &cwd;
203	let config = &config;
204	let model = &model;
205	let tool_registry = &tool_registry;
206	let frozen_system_prompt = Arc::clone(&frozen_system_prompt);
207	async move {
208	server
209	.open_prompt_stream(
210	config,
211	model,
212	&msgs,
213	cwd,
214	tool_registry,
215	&frozen_system_prompt,
216	)
217	.await
218	}
219	},
220	)
221	.await;
222	match outcome {
223	Ok((PromptOutcome::Completed(_text), full_messages)) => {
224	// Chunks were already streamed; record the full
225	// conversation (including any tool rounds) for
226	// the next prompt.
227	server.commit_turn_messages(&session_id, full_messages);
228	if let Some(id) = id {
229	let id = response_id_policy.response_id(id);
230	write_jsonrpc_result(
231	&mut writer,
232	id,
233	json!({ "stopReason": "end_turn" }),
234	)
235	.await?;
236	}
237	}
238	Ok((PromptOutcome::Cancelled, partial_messages)) => {
239	// The turn driver keeps complete receipts for every
240	// proposed tool call, including calls cancelled
241	// before execution, so partial side effects remain
242	// visible and no dangling tool_use block is stored.
243	server.commit_turn_messages(&session_id, partial_messages);
244	if let Some(id) = id {
245	let id = response_id_policy.response_id(id);
246	write_jsonrpc_result(
247	&mut writer,
248	id,
249	json!({ "stopReason": "cancelled" }),
250	)
251	.await?;
252	}
253	}
254	Ok((PromptOutcome::MaxRounds(_text), full_messages)) => {
255	// Max rounds reached — commit what we have
256	// (unlike cancel, this is a normal completion).
257	server.commit_turn_messages(&session_id, full_messages);
258	if let Some(id) = id {
259	let id = response_id_policy.response_id(id);
260	write_jsonrpc_result(
261	&mut writer,
262	id,
263	json!({ "stopReason": "max_turn_requests" }),
264	)
265	.await?;
266	}
267	}
268	Err(err) => {
269	if let Some(partial_messages) = err.partial_messages {
270	// A later provider round failed after one or
271	// more tools completed. Preserve those
272	// side-effect receipts in session history.
273	server.commit_turn_messages(&session_id, partial_messages);
274	} else {
275	// The user message was already pushed into
276	// session history by `begin_prompt`; roll it
277	// back when no tool receipt exists yet.
278	server.rollback_user_message(&session_id);
279	}
280	let id = id.map(\|id\| response_id_policy.response_id(id));
281	write_jsonrpc_error(&mut writer, id, -32603, err.source.to_string())
282	.await?;
283	}
284	}
285	}
286	Err(err) => {
287	let id = id.map(\|id\| server.response_id_policy.response_id(id));
288	write_jsonrpc_error(&mut writer, id, err.code, err.message).await?;
289	}
290	}
291	continue;
292	}
293
294	match server.handle_request(method, params).await {
295	Ok(AcpDispatch::Response(result)) => {
296	if let Some(id) = id {
297	let id = server.response_id_policy.response_id(id);
298	write_jsonrpc_result(&mut writer, id, result).await?;
299	}
300	}
301	Ok(AcpDispatch::Shutdown) => {
302	if let Some(id) = id {
303	let id = server.response_id_policy.response_id(id);
304	write_jsonrpc_result(&mut writer, id, json!(null)).await?;
305	}
306	break;
307	}
308	Err(err) => {
309	let id = id.map(\|id\| server.response_id_policy.response_id(id));
310	write_jsonrpc_error(&mut writer, id, err.code, err.message).await?;
311	}
312	}
313	}
314
315	Ok(())
316	}
317
318	fn is_jsonrpc_response(message: &Value) -> bool {
319	message.get("id").is_some()
320	&& (message.get("result").is_some() \|\| message.get("error").is_some())
321	}
322
323	/// Outcome of a `session/prompt` turn driven against the input stream.
324	#[derive(Debug, PartialEq, Eq)]
325	enum PromptOutcome {
326	/// The provider call finished first; carries the assistant text.
327	Completed(String),
328	/// A matching `session/cancel` arrived before the call finished.
329	Cancelled,
330	/// The turn reached the maximum number of tool-call round-trips.
331	/// Carries whatever text the model produced in the final round.
332	MaxRounds(String),
333	}
334
335	#[derive(Debug)]
336	struct AgenticPromptError {
337	source: anyhow::Error,
338	/// Present once at least one complete tool-result batch has been appended.
339	/// Those receipts may describe real side effects and must survive a later
340	/// provider failure.
341	partial_messages: Option<Vec<Message>>,
342	}
343
344	impl std::fmt::Display for AgenticPromptError {
345	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
346	self.source.fmt(f)
347	}
348	}
349
350	impl AgenticPromptError {
351	fn new(source: anyhow::Error, messages: &[Message], has_tool_receipts: bool) -> Self {
352	Self {
353	source,
354	partial_messages: has_tool_receipts.then(\|\| messages.to_vec()),
355	}
356	}
357	}
358
359	/// A tool call the model requested, assembled from streamed
360	/// `content_block_start` / `content_block_delta` / `content_block_stop`
361	/// events. `parse_error` is set when the accumulated `input_json_delta`
362	/// bytes did not parse as JSON — the call is still surfaced (rather than
363	/// silently dropped) so the model gets a clear tool-result error instead of
364	/// the turn hanging.
365	#[derive(Debug, Clone)]
366	struct PendingToolCall {
367	id: String,
368	name: String,
369	input: Value,
370	parse_error: Option<String>,
371	}
372
373	/// Accumulates one streamed `tool_use` content block until its
374	/// `content_block_stop`.
375	#[derive(Debug, Default)]
376	struct ToolUseAccumulator {
377	id: String,
378	name: String,
379	initial_input: Value,
380	buffer: String,
381	}
382
383	impl ToolUseAccumulator {
384	fn finalize(self) -> PendingToolCall {
385	if self.buffer.trim().is_empty() {
386	return PendingToolCall {
387	id: self.id,
388	name: self.name,
389	input: self.initial_input,
390	parse_error: None,
391	};
392	}
393	match serde_json::from_str(&self.buffer) {
394	Ok(input) => PendingToolCall {
395	id: self.id,
396	name: self.name,
397	input,
398	parse_error: None,
399	},
400	Err(_) => PendingToolCall {
401	id: self.id,
402	name: self.name,
403	input: json!({}),
404	parse_error: Some(self.buffer),
405	},
406	}
407	}
408	}
409
410	/// The text payload an ACP client should see for a given stream event, if any.
411	/// ACP baseline is text-only, so thinking/tool/control events carry no chunk.
412	fn stream_text_chunk(event: &StreamEvent) -> Option<&str> {
413	match event {
414	StreamEvent::ContentBlockDelta {
415	delta: Delta::TextDelta { text },
416	..
417	} => Some(text),
418	StreamEvent::ContentBlockStart {
419	content_block: ContentBlockStart::Text { text },
420	..
421	} => Some(text),
422	_ => None,
423	}
424	}
425
426	/// Consume a provider response `stream`, emitting each text delta as a
427	/// `session/update` chunk, while concurrently watching `reader` for a
428	/// `session/cancel` targeting `session_id`.
429	///
430	/// This is the streaming + cancellation control point. It is generic over the
431	/// reader/writer and takes the boxed stream, so it is unit-tested with canned
432	/// in-memory streams and readers — no real provider call required. The caller
433	/// keeps the only writer, so streamed chunks and acknowledgements all stay on
434	/// the single protocol-clean stdout stream.
435	///
436	/// Returns [`PromptOutcome::Completed`] with the full accumulated text once the
437	/// stream ends (or emits `message_stop`), plus any `tool_use` blocks the model
438	/// emitted during the round, so the caller can execute them and continue the
439	/// turn. A matching `session/cancel` (request or notification form) ends it
440	/// early with [`PromptOutcome::Cancelled`] — dropping the stream aborts the
441	/// underlying provider connection. The turn is single-flight: a cancel for a
442	/// different session is acknowledged and ignored; any other concurrent request
443	/// is rejected with a clear error so the client is not left waiting;
444	/// notifications without an id are ignored.
445	async fn drive_prompt_stream<R, W>(
446	mut stream: StreamEventBox,
447	session_id: &str,
448	response_id_policy: JsonRpcResponseIdPolicy,
449	reader: &mut Lines<R>,
450	writer: &mut W,
451	) -> Result<(PromptOutcome, Vec<PendingToolCall>)>
452	where
453	R: AsyncBufRead + Unpin,
454	W: AsyncWrite + Unpin,
455	{
456	let mut accumulated = String::new();
457	let mut tool_calls: Vec<PendingToolCall> = Vec::new();
458	let mut pending_tool_uses: HashMap<u32, ToolUseAccumulator> = HashMap::new();
459	// Once input closes mid-turn we stop selecting on the reader and just drain
460	// the stream to completion, rather than spinning on repeated EOFs.
461	let mut reader_open = true;
462	loop {
463	tokio::select! {
464	event = stream.next() => {
465	match event {
466	// Stream exhausted without an explicit stop: turn is done.
467	None => return Ok((PromptOutcome::Completed(accumulated), tool_calls)),
468	Some(Ok(event)) => {
469	if let Some(text) = stream_text_chunk(&event)
470	&& !text.is_empty() {
471	accumulated.push_str(text);
472	write_session_update(writer, session_id, text.to_string()).await?;
473	}
474	match event {
475	StreamEvent::ContentBlockStart {
476	index,
477	content_block: ContentBlockStart::ToolUse { id, name, input, ..},
478	} => {
479	pending_tool_uses.insert(
480	index,
481	ToolUseAccumulator {
482	id,
483	name,
484	initial_input: input,
485	buffer: String::new(),
486	},
487	);
488	}
489	StreamEvent::ContentBlockDelta {
490	index,
491	delta: Delta::InputJsonDelta { partial_json },
492	} => {
493	if let Some(acc) = pending_tool_uses.get_mut(&index) {
494	acc.buffer.push_str(&partial_json);
495	}
496	}
497	StreamEvent::ContentBlockStop { index } => {
498	if let Some(acc) = pending_tool_uses.remove(&index) {
499	tool_calls.push(acc.finalize());
500	}
501	}
502	StreamEvent::MessageStop => {
503	return Ok((PromptOutcome::Completed(accumulated), tool_calls));
504	}
505	StreamEvent::Error { error } => {
506	return Err(anyhow!("provider stream error: {error}"));
507	}
508	_ => {}
509	}
510	}
511	Some(Err(err)) => return Err(err),
512	}
513	}
514	line = reader.next_line(), if reader_open => {
515	let line = match line? {
516	Some(line) => line,
517	// Input closed mid-turn: stop watching it, keep draining.
518	None => {
519	reader_open = false;
520	continue;
521	}
522	};
523	if line.trim().is_empty() {
524	continue;
525	}
526	let message: Value = match serde_json::from_str(&line) {
527	Ok(value) => value,
528	Err(err) => {
529	write_jsonrpc_error(writer, None, -32700, format!("invalid json: {err}"))
530	.await?;
531	continue;
532	}
533	};
534	let id = message.get("id").cloned();
535	match message.get("method").and_then(Value::as_str) {
536	Some("session/cancel") => {
537	let target = message.pointer("/params/sessionId").and_then(Value::as_str);
538	// A cancel with no sessionId is treated as targeting the
539	// single in-flight turn.
540	if target.is_none() \|\| target == Some(session_id) {
541	if let Some(id) = id {
542	let id = response_id_policy.response_id(id);
543	write_jsonrpc_result(writer, id, json!(null)).await?;
544	}
545	// Dropping `stream` on return aborts the provider call.
546	return Ok((PromptOutcome::Cancelled, tool_calls));
547	}
548	// Cancel for some other session: acknowledge, keep going.
549	if let Some(id) = id {
550	let id = response_id_policy.response_id(id);
551	write_jsonrpc_result(writer, id, json!(null)).await?;
552	}
553	}
554	_ => {
555	// The turn is single-flight; do not silently drop a
556	// request the client expects a response to.
557	if let Some(id) = id {
558	let id = response_id_policy.response_id(id);
559	write_jsonrpc_error(
560	writer,
561	Some(id),
562	-32603,
563	"a session/prompt turn is already in progress",
564	)
565	.await?;
566	}
567	}
568	}
569	}
570	}
571	}
572	}
573
574	/// Outcome of executing one batch of tool calls from a single round.
575	enum ToolBatchOutcome {
576	/// Every tool call ran to completion; carries the `tool_result` messages
577	/// to append to the conversation, in call order.
578	Completed(Vec<Message>),
579	/// A matching `session/cancel` arrived while a tool was awaiting approval
580	/// or running. Carries receipts for every proposed call so completed or
581	/// partially completed side effects never disappear from history.
582	Cancelled(Vec<Message>),
583	}
584
585	#[derive(Debug, Clone, PartialEq, Eq)]
586	enum AcpToolAdmission {
587	Auto,
588	RequestPermission(String),
589	Block(String),
590	}
591
592	#[derive(Debug, Clone, PartialEq, Eq)]
593	enum AcpPermissionDecision {
594	Allow,
595	Reject(String),
596	Cancelled,
597	}
598
599	fn acp_shell_command_requests_detach(command: &str) -> bool {
600	let mut single_quoted = false;
601	let mut double_quoted = false;
602	let mut escaped = false;
603	let chars = command.chars().collect::<Vec<_>>();
604	for (index, ch) in chars.iter().copied().enumerate() {
605	if escaped {
606	escaped = false;
607	continue;
608	}
609	if ch == '\\' && !single_quoted {
610	escaped = true;
611	continue;
612	}
613	if ch == '\'' && !double_quoted {
614	single_quoted = !single_quoted;
615	continue;
616	}
617	if ch == '"' && !single_quoted {
618	double_quoted = !double_quoted;
619	continue;
620	}
621	if ch != '&' \|\| single_quoted \|\| double_quoted {
622	continue;
623	}
624	let previous = index.checked_sub(1).and_then(\|i\| chars.get(i)).copied();
625	let next = chars.get(index + 1).copied();
626	// `&&`, `&>`/`&>>`, and `>&` are chaining/redirection rather than a
627	// detached child. Any other unquoted ampersand is a background
628	// control operator and is unavailable in ACP.
629	if previous != Some('&') && next != Some('&') && next != Some('>') && previous != Some('>')
630	{
631	return true;
632	}
633	}
634
635	shell_words::split(command).is_ok_and(\|words\| {
636	words.iter().any(\|word\| {
637	matches!(
638	word.to_ascii_lowercase().as_str(),
639	"nohup" \| "disown" \| "setsid" \| "daemonize"
640	)
641	})
642	})
643	}
644
645	#[derive(Debug)]
646	struct PreparedAcpTool {
647	call: PreparedToolCall,
648	admission: AcpToolAdmission,
649	additional_context: Option<String>,
650	}
651
652	/// Prepare one registered call and fold every policy layer that can tighten
653	/// its admission. ACP deliberately does not use the TUI's workspace-write
654	/// carve-out: a remembered exact allow rule may clear the ordinary tool hold,
655	/// but the built-in safety floor and repository law can always re-add a prompt
656	/// or hard block afterwards.
657	fn prepare_acp_tool_admission(
658	config: &Config,
659	registry: &ToolRegistry,
660	call: &PendingToolCall,
661	) -> std::result::Result<(PreparedToolCall, AcpToolAdmission), ToolError> {
662	let spec = registry.get(&call.name).ok_or_else(\|\| {
663	ToolError::not_available(format!("tool '{}' is not registered", call.name))
664	})?;
665	let prepared = spec.prepare(call.input.clone(), registry.context())?;
666	let canonical_name =
667	crate::tools::canonical_action::canonical_action_alias(&call.name, &prepared.input);
668	if matches!(call.name.as_str(), "bash" \| "Bash" \| "exec_shell")
669	\|\| canonical_name.starts_with("exec_shell")
670	{
671	let action = prepared
672	.input
673	.get("action")
674	.and_then(Value::as_str)
675	.unwrap_or("run");
676	let requests_stateful_shell = action != "run"
677	\|\| prepared.starts_detached
678	\|\| prepared.input.get("interactive").and_then(Value::as_bool) == Some(true)
679	\|\| prepared.input.get("persist").and_then(Value::as_bool) == Some(true)
680	\|\| prepared
681	.input
682	.get("command")
683	.and_then(Value::as_str)
684	.is_some_and(acp_shell_command_requests_detach);
685	if requests_stateful_shell {
686	return Ok((
687	prepared,
688	AcpToolAdmission::Block(
689	"ACP v0.9.6 exposes foreground Bash runs only; background, TTY, interactive, persistent, and background-task control actions are unavailable."
690	.to_string(),
691	),
692	));
693	}
694	}
695	let mut permission_reason =
696	(prepared.approval != ApprovalRequirement::Auto).then(\|\| prepared.description.clone());
697	let approval_mode = acp_approval_mode(config);
698	let workspace = registry.context().workspace.as_path();
699
700	let typed_rule = exec_shell_ask_rule_decision_for_policy(
701	&config.exec_policy_engine,
702	&call.name,
703	&prepared.input,
704	workspace,
705	approval_mode,
706	)
707	.or_else(\|\| {
708	file_tool_ask_rule_decision_for_policy(
709	&config.exec_policy_engine,
710	&call.name,
711	&prepared.input,
712	workspace,
713	approval_mode,
714	)
715	});
716	match typed_rule {
717	Some(ToolAskRuleDecision::Allow) => permission_reason = None,
718	Some(ToolAskRuleDecision::Prompt(reason)) => permission_reason = Some(reason),
719	Some(ToolAskRuleDecision::Block(reason)) => {
720	return Ok((prepared, AcpToolAdmission::Block(reason)));
721	}
722	None => {}
723	}
724
725	let run_origin = if prepared.starts_detached {
726	crate::tui::auto_review::RunOrigin::Background
727	} else {
728	crate::tui::auto_review::RunOrigin::Headless
729	};
730	let review_context = crate::tui::auto_review::AutoReviewContext::from_tool_call(
731	&call.name,
732	&prepared.input,
733	run_origin,
734	approval_mode,
735	crate::config::is_workspace_trusted(workspace),
736	Some(workspace),
737	);
738	let (auto_review, _audit) =
739	auto_review_plan_decision_for_context(&config.auto_review_policy(), &review_context);
740	match auto_review {
741	AutoReviewPlanDecision::NoChange \| AutoReviewPlanDecision::Allow => {}
742	AutoReviewPlanDecision::ForcePrompt(reason) => permission_reason = Some(reason),
743	// Headless adapters keep the deterministic-only tier: a fallback hold
744	// that interactive Auto posture would send to the model guardian is
745	// a hard block here (the reviewer is an interactive-session feature).
746	AutoReviewPlanDecision::ConsultReviewer(reason) \| AutoReviewPlanDecision::Block(reason) => {
747	return Ok((prepared, AcpToolAdmission::Block(reason)));
748	}
749	}
750
751	if let Some(repo_law) =
752	crate::repo_law::repo_law_plan_decision(workspace, &call.name, &prepared.input)
753	{
754	match repo_law {
755	crate::repo_law::RepoLawPlanDecision::ForcePrompt(reason) => {
756	permission_reason = Some(reason);
757	}
758	crate::repo_law::RepoLawPlanDecision::Block(reason) => {
759	return Ok((prepared, AcpToolAdmission::Block(reason)));
760	}
761	}
762	}
763
764	let admission = permission_reason
765	.map(AcpToolAdmission::RequestPermission)
766	.unwrap_or(AcpToolAdmission::Auto);
767	// #6337: Bypass pre-approves prompts so an unattended `--yolo` session
768	// executes instead of stalling on permission requests no client answers.
769	// Hard blocks above (safety floor, repo law, reviewer consult) return
770	// early and are never downgraded.
771	let admission = if approval_mode == ApprovalMode::Bypass
772	&& matches!(admission, AcpToolAdmission::RequestPermission(_))
773	{
774	AcpToolAdmission::Auto
775	} else {
776	admission
777	};
778	Ok((prepared, admission))
779	}
780
781	/// Run the same strict pre-tool hook gate as the native turn loop, then
782	/// prepare and evaluate policy from the hook's final input. The initial
783	/// preparation is deliberately side-effect free and catches malformed input
784	/// before an operator hook is asked to reason about it; any rewrite is fully
785	/// re-prepared and all policy layers run again from that rewritten value.
786	async fn prepare_acp_tool_with_hooks(
787	config: &Config,
788	model: &str,
789	registry: &ToolRegistry,
790	call: &PendingToolCall,
791	) -> std::result::Result<PreparedAcpTool, ToolError> {
792	let spec = registry.get(&call.name).ok_or_else(\|\| {
793	ToolError::not_available(format!("tool '{}' is not registered", call.name))
794	})?;
795	// Initial validation mirrors the native prepare-before-hooks contract.
796	spec.prepare(call.input.clone(), registry.context())?;
797
798	let hook_outcome = run_tool_call_before_hooks(
799	registry.context().runtime.hook_executor.as_ref(),
800	&call.name,
801	&call.id,
802	&call.input,
803	acp_mode(config),
804	registry.context().workspace.as_path(),
805	model,
806	)
807	.await?;
808
809	let mut final_call = call.clone();
810	if let Some(updated_input) = hook_outcome.updated_input {
811	final_call.input = updated_input;
812	}
813	let (prepared, mut admission) = prepare_acp_tool_admission(config, registry, &final_call)?;
814	if hook_outcome.requires_approval && matches!(admission, AcpToolAdmission::Auto) {
815	admission = AcpToolAdmission::RequestPermission(
816	"A ToolCallBefore hook requires explicit approval for this call.".to_string(),
817	);
818	}
819
820	Ok(PreparedAcpTool {
821	call: prepared,
822	admission,
823	additional_context: hook_outcome.additional_context,
824	})
825	}
826
827	fn next_acp_permission_request_id() -> Value {
828	Value::String(format!(
829	"codewhale-permission-{}",
830	NEXT_ACP_PERMISSION_REQUEST_ID.fetch_add(1, Ordering::Relaxed)
831	))
832	}
833
834	async fn write_tool_permission_request<W>(
835	writer: &mut W,
836	request_id: &Value,
837	session_id: &str,
838	call: &PendingToolCall,
839	reason: &str,
840	) -> Result<()>
841	where
842	W: AsyncWrite + Unpin,
843	{
844	write_json_line(
845	writer,
846	json!({
847	"jsonrpc": "2.0",
848	"id": request_id,
849	"method": "session/request_permission",
850	"params": {
851	"sessionId": session_id,
852	"toolCall": {
853	"toolCallId": call.id,
854	"title": tool_call_title(call),
855	"kind": tool_call_kind(call),
856	"status": "pending",
857	"rawInput": call.input,
858	"content": [{
859	"type": "content",
860	"content": { "type": "text", "text": reason }
861	}]
862	},
863	"options": [
864	{
865	"optionId": "allow-once",
866	"name": "Allow once",
867	"kind": "allow_once"
868	},
869	{
870	"optionId": "reject-once",
871	"name": "Reject",
872	"kind": "reject_once"
873	}
874	]
875	}
876	}),
877	)
878	.await
879	}
880
881	/// Ask the ACP client to approve one sensitive call. The client owns the UI
882	/// and ACP v1 requires it to answer a pending request with `cancelled` when the
883	/// prompt turn is cancelled. Unknown, malformed, or errored responses all fail
884	/// closed as rejection; only the exact offered `allow-once` id authorizes work.
885	async fn request_tool_permission<R, W>(
886	reader: &mut Lines<R>,
887	writer: &mut W,
888	response_id_policy: JsonRpcResponseIdPolicy,
889	session_id: &str,
890	call: &PendingToolCall,
891	reason: &str,
892	) -> Result<AcpPermissionDecision>
893	where
894	R: AsyncBufRead + Unpin,
895	W: AsyncWrite + Unpin,
896	{
897	let request_id = next_acp_permission_request_id();
898	write_tool_permission_request(writer, &request_id, session_id, call, reason).await?;
899	let mut cancel_deadline = None;
900
901	loop {
902	let line = if let Some(deadline) = cancel_deadline {
903	match tokio::time::timeout_at(deadline, reader.next_line()).await {
904	Ok(line) => line?,
905	Err(_) => return Ok(AcpPermissionDecision::Cancelled),
906	}
907	} else {
908	reader.next_line().await?
909	};
910	let Some(line) = line else {
911	return Ok(AcpPermissionDecision::Reject(
912	"Permission denied: ACP client disconnected before answering.".to_string(),
913	));
914	};
915	if line.trim().is_empty() {
916	continue;
917	}
918	let message: Value = match serde_json::from_str(&line) {
919	Ok(value) => value,
920	Err(err) => {
921	write_jsonrpc_error(writer, None, -32700, format!("invalid json: {err}")).await?;
922	continue;
923	}
924	};
925
926	if let Some(method) = message.get("method").and_then(Value::as_str) {
927	let message_id = message.get("id").cloned();
928	if method == "session/cancel" {
929	let target = message.pointer("/params/sessionId").and_then(Value::as_str);
930	if target.is_none() \|\| target == Some(session_id) {
931	if let Some(message_id) = message_id {
932	let message_id = response_id_policy.response_id(message_id);
933	write_jsonrpc_result(writer, message_id, json!(null)).await?;
934	}
935	cancel_deadline.get_or_insert_with(\|\| {
936	tokio::time::Instant::now() + ACP_PERMISSION_CANCEL_GRACE
937	});
938	continue;
939	}
940	if let Some(message_id) = message_id {
941	let message_id = response_id_policy.response_id(message_id);
942	write_jsonrpc_result(writer, message_id, json!(null)).await?;
943	}
944	continue;
945	}
946
947	if let Some(message_id) = message_id {
948	let message_id = response_id_policy.response_id(message_id);
949	write_jsonrpc_error(
950	writer,
951	Some(message_id),
952	-32603,
953	"a session/prompt turn is already in progress",
954	)
955	.await?;
956	}
957	continue;
958	}
959
960	if message.get("id") != Some(&request_id) {
961	// This is a response, not a request; there is nothing valid to send
962	// back. Ignore stale/unrelated agent-response traffic without
963	// allowing it to satisfy the permission gate.
964	continue;
965	}
966	if cancel_deadline.is_some() {
967	return Ok(AcpPermissionDecision::Cancelled);
968	}
969	if message.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
970	return Ok(AcpPermissionDecision::Reject(
971	"Permission denied: malformed ACP response.".to_string(),
972	));
973	}
974	if message.get("error").is_some() {
975	return Ok(AcpPermissionDecision::Reject(
976	"Permission denied: ACP client returned an error.".to_string(),
977	));
978	}
979	match message
980	.pointer("/result/outcome/outcome")
981	.and_then(Value::as_str)
982	{
983	Some("cancelled") => return Ok(AcpPermissionDecision::Cancelled),
984	Some("selected") => {
985	let option_id = message
986	.pointer("/result/outcome/optionId")
987	.and_then(Value::as_str);
988	return Ok(match option_id {
989	Some("allow-once") => AcpPermissionDecision::Allow,
990	Some("reject-once") => AcpPermissionDecision::Reject(
991	"Permission denied by the user; the tool was not executed.".to_string(),
992	),
993	_ => AcpPermissionDecision::Reject(
994	"Permission denied: ACP client selected an unknown option.".to_string(),
995	),
996	});
997	}
998	_ => {
999	return Ok(AcpPermissionDecision::Reject(
1000	"Permission denied: malformed ACP response.".to_string(),
1001	));
1002	}
1003	}
1004	}
1005	}
1006
1007	async fn record_tool_execution_result<W>(
1008	writer: &mut W,
1009	session_id: &str,
1010	call: &PendingToolCall,
1011	result: std::result::Result<RichToolResult, ToolError>,
1012	) -> Result<Message>
1013	where
1014	W: AsyncWrite + Unpin,
1015	{
1016	let (content, is_error, rich_blocks) = match result {
1017	Ok(tool_result) => (
1018	tool_result.result.content,
1019	!tool_result.result.success,
1020	tool_result.content_blocks,
1021	),
1022	Err(err) => (format!("Error: {err}"), true, Vec::new()),
1023	};
1024	let status = if is_error { "failed" } else { "completed" };
1025	write_tool_call_update_with_blocks(
1026	writer,
1027	session_id,
1028	call,
1029	status,
1030	Some(&content),
1031	&rich_blocks,
1032	)
1033	.await?;
1034	Ok(tool_result_message_with_blocks(
1035	&call.id,
1036	content,
1037	is_error,
1038	rich_blocks
1039	.iter()
1040	.filter_map(\|block\| serde_json::to_value(block).ok())
1041	.collect(),
1042	))
1043	}
1044
1045	async fn record_unstarted_cancelled_calls<W, I>(
1046	writer: &mut W,
1047	session_id: &str,
1048	calls: I,
1049	) -> Result<Vec<Message>>
1050	where
1051	W: AsyncWrite + Unpin,
1052	I: IntoIterator<Item = PendingToolCall>,
1053	{
1054	let mut messages = Vec::new();
1055	for call in calls {
1056	write_tool_call_start(writer, session_id, &call).await?;
1057	let content = "Cancelled before execution; the tool was not run.";
1058	write_tool_call_update(writer, session_id, &call, "failed", Some(content)).await?;
1059	messages.push(tool_result_message(&call.id, content.to_string(), true));
1060	}
1061	Ok(messages)
1062	}
1063
1064	#[derive(Clone, Copy)]
1065	struct AcpTurnContext<'a> {
1066	config: &'a Config,
1067	model: &'a str,
1068	session_id: &'a str,
1069	tool_registry: &'a ToolRegistry,
1070	response_id_policy: JsonRpcResponseIdPolicy,
1071	}
1072
1073	/// Execute `tool_calls` in order against `registry`, reporting each one to
1074	/// the client as `tool_call` / `tool_call_update` session updates, while
1075	/// racing every execution against the reader for a `session/cancel`
1076	/// targeting `session_id`. On cancel, the tool's [`CancellationToken`] is
1077	/// signalled and the in-flight call is awaited to completion (so a
1078	/// cancel-aware tool like `Bash` gets a chance to kill its child
1079	/// process) before returning [`ToolBatchOutcome::Cancelled`].
1080	async fn execute_tool_calls_with_cancellation<R, W>(
1081	context: AcpTurnContext<'_>,
1082	tool_calls: Vec<PendingToolCall>,
1083	reader: &mut Lines<R>,
1084	writer: &mut W,
1085	) -> Result<ToolBatchOutcome>
1086	where
1087	R: AsyncBufRead + Unpin,
1088	W: AsyncWrite + Unpin,
1089	{
1090	let AcpTurnContext {
1091	config,
1092	model,
1093	session_id,
1094	tool_registry: registry,
1095	response_id_policy,
1096	} = context;
1097	let mut result_messages = Vec::with_capacity(tool_calls.len());
1098	let mut reader_open = true;
1099	let mut calls = tool_calls.into_iter();
1100
1101	while let Some(mut call) = calls.next() {
1102	write_tool_call_start(writer, session_id, &call).await?;
1103
1104	if let Some(parse_error) = call.parse_error.clone() {
1105	let content = format!("Error: tool arguments were not valid JSON: {parse_error}");
1106	write_tool_call_update(writer, session_id, &call, "failed", Some(&content)).await?;
1107	result_messages.push(tool_result_message(&call.id, content, true));
1108	continue;
1109	}
1110
1111	let prepared = match prepare_acp_tool_with_hooks(config, model, registry, &call).await {
1112	Ok(prepared) => prepared,
1113	Err(err) => {
1114	let content = format!("Error: {err}");
1115	write_tool_call_update(writer, session_id, &call, "failed", Some(&content)).await?;
1116	result_messages.push(tool_result_message(&call.id, content, true));
1117	continue;
1118	}
1119	};
1120	call.input = prepared.call.input;
1121
1122	match prepared.admission {
1123	AcpToolAdmission::Auto => {}
1124	AcpToolAdmission::Block(reason) => {
1125	let content = format!("Blocked by Codewhale policy: {reason}");
1126	write_tool_call_update(writer, session_id, &call, "failed", Some(&content)).await?;
1127	result_messages.push(tool_result_message(&call.id, content, true));
1128	continue;
1129	}
1130	AcpToolAdmission::RequestPermission(reason) => {
1131	match request_tool_permission(
1132	reader,
1133	writer,
1134	response_id_policy,
1135	session_id,
1136	&call,
1137	&reason,
1138	)
1139	.await?
1140	{
1141	AcpPermissionDecision::Allow => {}
1142	AcpPermissionDecision::Reject(content) => {
1143	write_tool_call_update(writer, session_id, &call, "failed", Some(&content))
1144	.await?;
1145	result_messages.push(tool_result_message(&call.id, content, true));
1146	continue;
1147	}
1148	AcpPermissionDecision::Cancelled => {
1149	let content = "Cancelled while awaiting permission; the tool was not run.";
1150	write_tool_call_update(writer, session_id, &call, "failed", Some(content))
1151	.await?;
1152	result_messages.push(tool_result_message(
1153	&call.id,
1154	content.to_string(),
1155	true,
1156	));
1157	result_messages.extend(
1158	record_unstarted_cancelled_calls(writer, session_id, calls).await?,
1159	);
1160	return Ok(ToolBatchOutcome::Cancelled(result_messages));
1161	}
1162	}
1163	}
1164	}
1165
1166	write_tool_call_update(writer, session_id, &call, "in_progress", None).await?;
1167
1168	let cancel_token = CancellationToken::new();
1169	let mut turn_context = registry.context().clone();
1170	turn_context.cancel_token = Some(cancel_token.clone());
1171	let exec_fut = registry.execute_rich_full_with_context(
1172	&call.name,
1173	call.input.clone(),
1174	Some(&turn_context),
1175	);
1176	tokio::pin!(exec_fut);
1177
1178	let mut cancelled = false;
1179	let exec_result = loop {
1180	tokio::select! {
1181	result = &mut exec_fut => break result,
1182	line = reader.next_line(), if reader_open => {
1183	let line = match line? {
1184	Some(line) => line,
1185	None => {
1186	reader_open = false;
1187	continue;
1188	}
1189	};
1190	if line.trim().is_empty() {
1191	continue;
1192	}
1193	let message: Value = match serde_json::from_str(&line) {
1194	Ok(value) => value,
1195	Err(err) => {
1196	write_jsonrpc_error(writer, None, -32700, format!("invalid json: {err}"))
1197	.await?;
1198	continue;
1199	}
1200	};
1201	let msg_id = message.get("id").cloned();
1202	match message.get("method").and_then(Value::as_str) {
1203	Some("session/cancel") => {
1204	let target = message.pointer("/params/sessionId").and_then(Value::as_str);
1205	if target.is_none() \|\| target == Some(session_id) {
1206	if let Some(msg_id) = msg_id {
1207	let msg_id = response_id_policy.response_id(msg_id);
1208	write_jsonrpc_result(writer, msg_id, json!(null)).await?;
1209	}
1210	cancel_token.cancel();
1211	// Give the tool a chance to observe the token and
1212	// wind down (e.g. kill a running child process)
1213	// before we drop it.
1214	cancelled = true;
1215	break (&mut exec_fut).await;
1216	}
1217	if let Some(msg_id) = msg_id {
1218	let msg_id = response_id_policy.response_id(msg_id);
1219	write_jsonrpc_result(writer, msg_id, json!(null)).await?;
1220	}
1221	}
1222	_ => {
1223	if let Some(msg_id) = msg_id {
1224	let msg_id = response_id_policy.response_id(msg_id);
1225	write_jsonrpc_error(
1226	writer,
1227	Some(msg_id),
1228	-32603,
1229	"a session/prompt turn is already in progress",
1230	)
1231	.await?;
1232	}
1233	}
1234	}
1235	}
1236	}
1237	};
1238
1239	let exec_result = exec_result.map(\|mut result\| {
1240	if let Some(context) = prepared.additional_context.as_deref() {
1241	result.result.content =
1242	format!("{}\n\n[hook context] {context}", result.result.content);
1243	}
1244	result
1245	});
1246	result_messages
1247	.push(record_tool_execution_result(writer, session_id, &call, exec_result).await?);
1248	if cancelled {
1249	result_messages
1250	.extend(record_unstarted_cancelled_calls(writer, session_id, calls).await?);
1251	return Ok(ToolBatchOutcome::Cancelled(result_messages));
1252	}
1253	}
1254
1255	Ok(ToolBatchOutcome::Completed(result_messages))
1256	}
1257
1258	fn tool_result_message(tool_use_id: &str, content: String, is_error: bool) -> Message {
1259	tool_result_message_with_blocks(tool_use_id, content, is_error, Vec::new())
1260	}
1261
1262	fn tool_result_message_with_blocks(
1263	tool_use_id: &str,
1264	content: String,
1265	is_error: bool,
1266	content_blocks: Vec<Value>,
1267	) -> Message {
1268	Message {
1269	role: Role::User,
1270	content: vec![ContentBlock::ToolResult {
1271	tool_use_id: tool_use_id.to_string(),
1272	content,
1273	is_error: Some(is_error),
1274	content_blocks: (!content_blocks.is_empty()).then_some(content_blocks),
1275	}],
1276	}
1277	}
1278
1279	/// Drive one `session/prompt` turn to completion, looping through as many
1280	/// LLM <-> tool round-trips as the model requests (bounded by
1281	/// [`MAX_ACP_TOOL_ROUNDS`]).
1282	///
1283	/// Recorded interim exception to the one-turn-loop rule (#6088, named in
1284	/// `crates/core/tests/single_turn_loop.rs`): ACP IDE sessions do not run on
1285	/// the full thread/turn runtime yet. #5835 converges them onto
1286	/// `Engine::run_turn` and deletes this loop along with the exception.
1287	///
1288	/// `open_stream` opens a fresh provider stream for the given message
1289	/// history; production callers wire it to [`AcpServer::open_prompt_stream`],
1290	/// while tests supply canned per-round streams so the loop can be exercised
1291	/// without a real provider. Returns the outcome of the final round plus the
1292	/// full message history (including any tool-call/tool-result rounds), which
1293	/// the caller commits to session history only when the turn completed
1294	/// normally.
1295	///
1296	/// `open_stream` takes the message history by value (a clone per round)
1297	/// rather than `&[Message]`: an `async fn`'s returned future captures the
1298	/// lifetime of every reference parameter, so a borrowed slice here would
1299	/// force `Fut` to depend on each call's borrow lifetime — which `FnMut`'s
1300	/// single associated `Fut` type cannot express. Taking ownership sidesteps
1301	/// that; production callers move the clone into an `async move` block.
1302	async fn run_agentic_prompt_turn<R, W, F, Fut>(
1303	context: AcpTurnContext<'_>,
1304	mut messages: Vec<Message>,
1305	reader: &mut Lines<R>,
1306	writer: &mut W,
1307	mut open_stream: F,
1308	) -> std::result::Result<(PromptOutcome, Vec<Message>), AgenticPromptError>
1309	where
1310	R: AsyncBufRead + Unpin,
1311	W: AsyncWrite + Unpin,
1312	F: FnMut(Vec<Message>) -> Fut,
1313	Fut: Future<Output = Result<StreamEventBox>>,
1314	{
1315	let AcpTurnContext {
1316	session_id,
1317	response_id_policy,
1318	..
1319	} = context;
1320	let mut has_tool_receipts = false;
1321	for _round in 0..MAX_ACP_TOOL_ROUNDS {
1322	let stream = open_stream(messages.clone())
1323	.await
1324	.map_err(\|error\| AgenticPromptError::new(error, &messages, has_tool_receipts))?;
1325	let (outcome, tool_calls) =
1326	drive_prompt_stream(stream, session_id, response_id_policy, reader, writer)
1327	.await
1328	.map_err(\|error\| AgenticPromptError::new(error, &messages, has_tool_receipts))?;
1329
1330	let text = match outcome {
1331	PromptOutcome::Cancelled => return Ok((PromptOutcome::Cancelled, messages)),
1332	PromptOutcome::Completed(text) => text,
1333	PromptOutcome::MaxRounds(text) => text,
1334	};
1335
1336	let mut assistant_content = Vec::new();
1337	if !text.is_empty() {
1338	assistant_content.push(ContentBlock::Text {
1339	text: text.clone(),
1340	cache_control: None,
1341	});
1342	}
1343	for call in &tool_calls {
1344	assistant_content.push(ContentBlock::ToolUse {
1345	id: call.id.clone(),
1346	name: call.name.clone(),
1347	input: call.input.clone(),
1348	caller: None,
1349	thought_signature: None,
1350	});
1351	}
1352	if !assistant_content.is_empty() {
1353	messages.push(Message {
1354	role: Role::Assistant,
1355	content: assistant_content,
1356	});
1357	}
1358
1359	if tool_calls.is_empty() {
1360	return Ok((PromptOutcome::Completed(text), messages));
1361	}
1362
1363	let batch = execute_tool_calls_with_cancellation(context, tool_calls, reader, writer)
1364	.await
1365	.map_err(\|error\| AgenticPromptError::new(error, &messages, has_tool_receipts))?;
1366	match batch {
1367	ToolBatchOutcome::Cancelled(tool_result_messages) => {
1368	messages.extend(tool_result_messages);
1369	return Ok((PromptOutcome::Cancelled, messages));
1370	}
1371	ToolBatchOutcome::Completed(tool_result_messages) => {
1372	messages.extend(tool_result_messages);
1373	has_tool_receipts = true;
1374	}
1375	}
1376	}
1377
1378	// Max rounds reached: return the text accumulated in the final round
1379	// rather than an error, so the client gets a structured completion
1380	// with a clear stop reason.
1381	let final_text = messages
1382	.iter()
1383	.rev()
1384	.find(\|m\| m.role == "assistant")
1385	.and_then(\|m\| {
1386	m.content.iter().find_map(\|b\| match b {
1387	ContentBlock::Text { text, .. } => Some(text.clone()),
1388	_ => None,
1389	})
1390	})
1391	.unwrap_or_default();
1392	Ok((PromptOutcome::MaxRounds(final_text), messages))
1393	}
1394
1395	struct AcpServer {
1396	config: Config,
1397	model: String,
1398	default_cwd: PathBuf,
1399	sessions: HashMap<String, AcpSession>,
1400	/// Insertion-order tracking of session ids. Used to evict the oldest
1401	/// session (by insertion order, not arbitrary HashMap iteration) when
1402	/// the session cap is reached.
1403	insertion_order: VecDeque<String>,
1404	/// Whether the connected client accepts `terminal` tool calls, from
1405	/// `initialize` params `clientCapabilities.terminal`. Defaults to `false`
1406	/// (restrictive): clients that omit the field get no shell access. Older
1407	/// ACP clients predating the `terminal` capability get a working agent
1408	/// without shell, which is safe; the client can re-declare support when it
1409	/// reconnects.
1410	client_supports_terminal: bool,
1411	response_id_policy: JsonRpcResponseIdPolicy,
1412	}
1413
1414	struct AcpSession {
1415	cwd: PathBuf,
1416	messages: Vec<Message>,
1417	config: Config,
1418	model: String,
1419	/// Built once per session over the session `cwd`, then reused for every
1420	/// prompt turn: `to_api_tools()` memoises the serialised catalog, and
1421	/// `file_read_tracker` / the shell manager need to persist across turns.
1422	tool_registry: Arc<ToolRegistry>,
1423	}
1424
1425	/// The `&mut self` result of validating a `session/prompt`: the user turn is
1426	/// already recorded, and the cloned conversation + cwd are ready for the
1427	/// borrow-free provider call that the prompt driver races against cancellation.
1428	struct PreparedPrompt {
1429	session_id: String,
1430	messages: Vec<Message>,
1431	cwd: PathBuf,
1432	config: Config,
1433	model: String,
1434	}
1435
1436	enum AcpDispatch {
1437	Response(Value),
1438	Shutdown,
1439	}
1440
1441	#[derive(Debug)]
1442	struct AcpError {
1443	code: i32,
1444	message: String,
1445	}
1446
1447	impl AcpServer {
1448	fn new(config: Config, model: String, default_cwd: PathBuf) -> Self {
1449	Self {
1450	config,
1451	model,
1452	default_cwd,
1453	sessions: HashMap::new(),
1454	insertion_order: VecDeque::new(),
1455	client_supports_terminal: false,
1456	response_id_policy: JsonRpcResponseIdPolicy::Preserve,
1457	}
1458	}
1459
1460	// `session/prompt` is handled in the main loop (it needs to run concurrently
1461	// with the reader for cancellation); every other method is request/response.
1462	async fn handle_request(
1463	&mut self,
1464	method: &str,
1465	params: Value,
1466	) -> std::result::Result<AcpDispatch, AcpError> {
1467	match method {
1468	"initialize" => {
1469	if let Some(terminal) = params
1470	.pointer("/clientCapabilities/terminal")
1471	.and_then(Value::as_bool)
1472	{
1473	self.client_supports_terminal = terminal;
1474	}
1475	self.response_id_policy = JsonRpcResponseIdPolicy::from_initialize_params(&params);
1476	Ok(AcpDispatch::Response(initialize_result(
1477	params.get("protocolVersion").and_then(Value::as_u64),
1478	&self.config,
1479	)))
1480	}
1481	"session/new" => Ok(AcpDispatch::Response(self.new_session(params)?)),
1482	"session/list" => Ok(AcpDispatch::Response(self.list_sessions(params)?)),
1483	"session/load" => Ok(AcpDispatch::Response(self.load_session(params)?)),
1484	"session/listProviders" => Ok(AcpDispatch::Response(self.list_providers())),
1485	"session/currentModel" => Ok(AcpDispatch::Response(self.current_model())),
1486	"session/selectModel" => Ok(AcpDispatch::Response(self.select_model(params)?)),
1487	"session/set_config_option" => {
1488	Ok(AcpDispatch::Response(self.set_session_config(params)?))
1489	}
1490	"session/set_mode" \| "session/set_model" => {
1491	let (config_id, field) = if method == "session/set_mode" {
1492	("mode", "modeId")
1493	} else {
1494	("model", "modelId")
1495	};
1496	self.set_session_config(json!({
1497	"sessionId": params.get("sessionId"),
1498	"configId": config_id,
1499	"value": params.get(field),
1500	}))?;
1501	Ok(AcpDispatch::Response(json!({})))
1502	}
1503	// A cancel that arrives with no prompt in flight is an idempotent
1504	// no-op (the in-flight case is handled by the prompt driver).
1505	"session/cancel" => Ok(AcpDispatch::Response(json!(null))),
1506	"shutdown" => Ok(AcpDispatch::Shutdown),
1507	_ => Err(AcpError::method_not_found(method)),
1508	}
1509	}
1510
1511	fn new_session(&mut self, params: Value) -> std::result::Result<Value, AcpError> {
1512	let cwd = params
1513	.get("cwd")
1514	.and_then(Value::as_str)
1515	.map(PathBuf::from)
1516	.unwrap_or_else(\|\| self.default_cwd.clone());
1517	// A bare uuid, the same shape `create_saved_session` produces and the
1518	// same shape `session/list` advertises. The old `codewhale-` prefix put
1519	// this id in a namespace no other method understood, so a client that
1520	// replayed it into `session/load` — the normal thing to do — got
1521	// `-32602` for an id we had just handed it (#6174).
1522	let session_id = uuid::Uuid::new_v4().to_string();
1523	let tool_registry = Arc::new(build_acp_tool_registry(
1524	&self.config,
1525	&cwd,
1526	self.client_supports_terminal,
1527	));
1528
1529	// Evict oldest session when at capacity.
1530	if self.sessions.len() >= MAX_ACP_SESSIONS {
1531	// `VecDeque` preserves true insertion order; HashMap iteration
1532	// does not. Pop from the front to evict the session created
1533	// earliest.
1534	if let Some(oldest) = self.insertion_order.pop_front() {
1535	self.sessions.remove(&oldest);
1536	}
1537	}
1538
1539	self.insertion_order.push_back(session_id.clone());
1540	self.sessions.insert(
1541	session_id.clone(),
1542	AcpSession {
1543	cwd,
1544	messages: Vec::new(),
1545	config: self.config.clone(),
1546	model: self.model.clone(),
1547	tool_registry,
1548	},
1549	);
1550	Ok(self.session_configuration(&session_id))
1551	}
1552
1553	/// Durable Codewhale sessions an ACP client can resume (#5864).
1554	///
1555	/// ACP sessions are in-memory and capped; Codewhale's own sessions are the
1556	/// durable record, and an IDE that offers "resume" means those. A store
1557	/// that cannot be read is an empty list, not a failed request: enumeration
1558	/// is discovery, and a client asking what exists should not be broken by a
1559	/// missing sessions directory.
1560	fn list_sessions(&self, params: Value) -> std::result::Result<Value, AcpError> {
1561	let cwd = match params.get("cwd") {
1562	None \| Some(Value::Null) => None,
1563	Some(Value::String(path)) if std::path::Path::new(path).is_absolute() => {
1564	Some(PathBuf::from(path))
1565	}
1566	Some(_) => {
1567	return Err(AcpError::invalid_params(
1568	"session/list cwd must be an absolute path",
1569	));
1570	}
1571	};
1572	let sessions = Self::session_manager()
1573	.and_then(\|manager\| manager.list_sessions().ok())
1574	.unwrap_or_default();
1575	let sessions: Vec<Value> = sessions
1576	.into_iter()
1577	.filter(\|meta\| {
1578	cwd.as_ref().is_none_or(\|cwd\| {
1579	crate::session_manager::paths_equivalent(&meta.workspace, cwd)
1580	})
1581	})
1582	.map(\|meta\| {
1583	json!({
1584	"sessionId": meta.id,
1585	"title": meta.title,
1586	"cwd": meta.workspace.to_string_lossy(),
1587	"createdAt": meta.created_at.to_rfc3339(),
1588	"updatedAt": meta.updated_at.to_rfc3339(),
1589	"messageCount": meta.message_count,
1590	})
1591	})
1592	.collect();
1593	Ok(json!({ "sessions": sessions }))
1594	}
1595
1596	/// Rehydrate a durable Codewhale session as this connection's ACP session.
1597	///
1598	/// The loaded session keeps its own id so a client can list, load, and
1599	/// prompt against one identity. Its `cwd` comes from the saved workspace,
1600	/// not the server default, because the tool registry is built over it.
1601	fn load_session(&mut self, params: Value) -> std::result::Result<Value, AcpError> {
1602	let session_id = params
1603	.get("sessionId")
1604	.and_then(Value::as_str)
1605	.ok_or_else(\|\| AcpError::invalid_params("session/load requires sessionId"))?
1606	.to_string();
1607	// Sessions this connection already holds resolve from memory. `session/new`
1608	// sessions live only here — nothing on the ACP path writes them to the
1609	// durable store — so consulting the store first would fail every id we
1610	// minted ourselves. This also makes reloading an already-loaded durable
1611	// session cheap and free of store side effects.
1612	if self.sessions.contains_key(&session_id) {
1613	return Ok(self.session_configuration(&session_id));
1614	}
1615	let manager = Self::session_manager()
1616	.ok_or_else(\|\| AcpError::internal("no Codewhale session store is available"))?;
1617	let saved = manager
1618	.resume_session_by_prefix(&session_id)
1619	.map_err(\|error\| {
1620	AcpError::invalid_params(format!("could not load session {session_id}: {error}"))
1621	})?
1622	.session;
1623
1624	let cwd = saved.metadata.workspace.clone();
1625	let tool_registry = Arc::new(build_acp_tool_registry(
1626	&self.config,
1627	&cwd,
1628	self.client_supports_terminal,
1629	));
1630	let resolved_id = saved.metadata.id.clone();
1631	// A short prefix can resolve to an id this connection already
1632	// tracks: the in-memory fast path above checked the prefix, not the
1633	// resolved id. Pushing again would duplicate the id in
1634	// `insertion_order` while `sessions.insert` merely overwrites, and a
1635	// later capacity eviction would then pop the stale front copy and
1636	// remove a live, recently reloaded session (#6245).
1637	if self.sessions.contains_key(&resolved_id) {
1638	return Ok(self.session_configuration(&resolved_id));
1639	}
1640	if self.sessions.len() >= MAX_ACP_SESSIONS
1641	&& let Some(oldest) = self.insertion_order.pop_front()
1642	{
1643	self.sessions.remove(&oldest);
1644	}
1645	self.insertion_order.push_back(resolved_id.clone());
1646	self.sessions.insert(
1647	resolved_id.clone(),
1648	AcpSession {
1649	cwd,
1650	messages: saved.messages,
1651	config: self.config.clone(),
1652	model: self.model.clone(),
1653	tool_registry,
1654	},
1655	);
1656	Ok(self.session_configuration(&resolved_id))
1657	}
1658
1659	fn session_models(session: &AcpSession) -> Vec<String> {
1660	let provider = session.config.api_provider();
1661	let mut models =
1662	crate::provider_lake::models_for_provider(&session.config, provider, provider);
1663	if !models.contains(&session.model) {
1664	models.push(session.model.clone());
1665	}
1666	models
1667	}
1668
1669	fn session_configuration(&self, session_id: &str) -> Value {
1670	use codewhale_localization::{MessageId, resolve_locale, tr};
1671	let settings = crate::settings::Settings::load().unwrap_or_default();
1672	let locale = resolve_locale(&settings.locale);
1673	let session = &self.sessions[session_id];
1674	let models = Self::session_models(session);
1675	let mut modes = vec![
1676	json!({"id": "plan", "name": tr(locale, MessageId::AppModePlan), "description": tr(locale, MessageId::AppModePlanHint)}),
1677	];
1678	if acp_mode(&self.config) != AppMode::Plan {
1679	modes.insert(0, json!({"id": "agent", "name": tr(locale, MessageId::AppModeAgent), "description": tr(locale, MessageId::AppModeAgentHint)}));
1680	}
1681	let current_mode = if acp_mode(&session.config) == AppMode::Plan {
1682	"plan"
1683	} else {
1684	"agent"
1685	};
1686	json!({
1687	"sessionId": session_id,
1688	"modes": {"currentModeId": current_mode, "availableModes": modes},
1689	"models": {
1690	"currentModelId": session.model,
1691	"availableModels": models.iter().map(\|model\| json!({"modelId": model, "name": model})).collect::<Vec<_>>()
1692	},
1693	"configOptions": [
1694	{"id": "mode", "name": tr(locale, MessageId::SettingSubjectMode), "category": "mode", "type": "select", "currentValue": current_mode,
1695	"options": modes.iter().map(\|mode\| json!({"value": mode["id"], "name": mode["name"], "description": mode["description"]})).collect::<Vec<_>>()},
1696	{"id": "model", "name": tr(locale, MessageId::SettingSubjectModel), "category": "model", "type": "select", "currentValue": session.model,
1697	"options": models.iter().map(\|model\| json!({"value": model, "name": model})).collect::<Vec<_>>()}
1698	]
1699	})
1700	}
1701
1702	fn set_session_config(&mut self, params: Value) -> std::result::Result<Value, AcpError> {
1703	let session_id = params
1704	.get("sessionId")
1705	.and_then(Value::as_str)
1706	.ok_or_else(\|\| AcpError::invalid_params("sessionId is required"))?;
1707	let config_id = params
1708	.get("configId")
1709	.and_then(Value::as_str)
1710	.ok_or_else(\|\| AcpError::invalid_params("configId is required"))?;
1711	let value = params
1712	.get("value")
1713	.and_then(Value::as_str)
1714	.ok_or_else(\|\| AcpError::invalid_params("value must be an offered string option"))?;
1715	if !self.sessions.contains_key(session_id) {
1716	return Err(AcpError::invalid_params("unknown sessionId"));
1717	}
1718	let state = self.session_configuration(session_id);
1719	let offered = state["configOptions"]
1720	.as_array()
1721	.unwrap()
1722	.iter()
1723	.any(\|option\| {
1724	option["id"] == config_id
1725	&& option["options"]
1726	.as_array()
1727	.unwrap()
1728	.iter()
1729	.any(\|choice\| choice["value"] == value)
1730	});
1731	if !offered {
1732	return Err(AcpError::invalid_params(
1733	"unknown configuration option or value",
1734	));
1735	}
1736	let session = self.sessions.get_mut(session_id).unwrap();
1737	match config_id {
1738	"model" => session.model = value.to_string(),
1739	"mode" => {
1740	session.config.sandbox_mode = if value == "plan" {
1741	Some("read-only".to_string())
1742	} else {
1743	self.config.sandbox_mode.clone()
1744	};
1745	// Rebuild at this explicit between-turn authority boundary.
1746	// Both the next prompt prefix and tools reflect the new mode;
1747	// an in-flight turn keeps its frozen prefix and rejects setters.
1748	session.tool_registry = Arc::new(build_acp_tool_registry(
1749	&session.config,
1750	&session.cwd,
1751	self.client_supports_terminal,
1752	));
1753	}
1754	_ => unreachable!("validated offered option"),
1755	}
1756	Ok(json!({"configOptions": self.session_configuration(session_id)["configOptions"]}))
1757	}
1758
1759	fn session_manager() -> Option<crate::session_manager::SessionManager> {
1760	let dir = crate::session_manager::default_sessions_dir().ok()?;
1761	crate::session_manager::SessionManager::new(dir).ok()
1762	}
1763
1764	fn session_tool_registry(&self, session_id: &str) -> Option<Arc<ToolRegistry>> {
1765	self.sessions
1766	.get(session_id)
1767	.map(\|session\| session.tool_registry.clone())
1768	}
1769
1770	fn list_providers(&self) -> Value {
1771	let mut providers = ApiProvider::sorted_for_display()
1772	.into_iter()
1773	.map(\|provider\| {
1774	json!({
1775	"id": provider.as_str(),
1776	"displayName": provider.display_name(),
1777	"defaultModel": provider.metadata().map(\|metadata\| metadata.default_model())
1778	})
1779	})
1780	.collect::<Vec<_>>();
1781
1782	// Include user-defined `[providers.<name>]` custom entries so ACP
1783	// clients can discover and round-trip the provider names that
1784	// `session/selectModel` now accepts (#1519).
1785	if let Some(custom) = self.config.providers.as_ref().map(\|p\| &p.custom) {
1786	let mut names = custom.keys().collect::<Vec<_>>();
1787	names.sort();
1788	for name in names {
1789	providers.push(json!({
1790	"id": name,
1791	"displayName": name,
1792	"defaultModel": custom.get(name).and_then(\|cfg\| cfg.model.clone())
1793	}));
1794	}
1795	}
1796
1797	json!({ "providers": providers })
1798	}
1799
1800	fn current_model(&self) -> Value {
1801	// Prefer the raw configured provider key so a custom `[providers.<name>]`
1802	// entry round-trips through ACP instead of canonicalizing to "custom".
1803	let provider = match self.config.provider.as_deref() {
1804	Some(name) if !name.trim().is_empty() => name.to_string(),
1805	_ => self.config.api_provider().as_str().to_string(),
1806	};
1807	json!({
1808	"provider": provider,
1809	"model": self.model.as_str()
1810	})
1811	}
1812
1813	fn select_model(&mut self, params: Value) -> std::result::Result<Value, AcpError> {
1814	let model = params
1815	.get("model")
1816	.and_then(Value::as_str)
1817	.ok_or_else(\|\| AcpError::invalid_params("model is required"))?
1818	.to_string();
1819
1820	if let Some(provider_value) = params.get("provider") {
1821	let provider_name = provider_value
1822	.as_str()
1823	.ok_or_else(\|\| AcpError::invalid_params("provider must be a string"))?;
1824	// Accept either a built-in provider id/alias or a user-defined
1825	// custom provider name that has a `[providers.<name>]` table. For
1826	// custom providers, preserve the raw key so routing can still find
1827	// the configured base URL / auth / model (#1519); canonicalizing to
1828	// "custom" would lose that table key.
1829	let is_custom = self
1830	.config
1831	.providers
1832	.as_ref()
1833	.and_then(\|providers\| providers.custom_provider_config(provider_name))
1834	.is_some();
1835	if !is_custom && ApiProvider::parse(provider_name).is_none() {
1836	return Err(AcpError::invalid_params(format!(
1837	"unknown provider: {provider_name}"
1838	)));
1839	}
1840	self.config.provider = Some(provider_name.to_string());
1841	}
1842
1843	self.model = model;
1844	Ok(self.current_model())
1845	}
1846
1847	/// Validate a `session/prompt` request and append the user turn to history,
1848	/// returning the cloned conversation for the (borrow-free) provider call.
1849	///
1850	/// This is the `&mut self` half of a prompt turn; the streaming provider
1851	/// call lives in [`AcpServer::open_prompt_stream`] (which borrows `&self`
1852	/// only and returns a `'static` stream) so it can be raced against the
1853	/// reader for cancellation.
1854	fn begin_prompt(&mut self, params: Value) -> std::result::Result<PreparedPrompt, AcpError> {
1855	let session_id = params
1856	.get("sessionId")
1857	.and_then(Value::as_str)
1858	.ok_or_else(\|\| AcpError::invalid_params("sessionId is required"))?
1859	.to_string();
1860	let prompt = extract_prompt_text(params.get("prompt"))
1861	.filter(\|text\| !text.trim().is_empty())
1862	.ok_or_else(\|\| AcpError::invalid_params("prompt must include text content"))?;
1863
1864	let (messages, cwd, config, model) = {
1865	let session = self
1866	.sessions
1867	.get_mut(&session_id)
1868	.ok_or_else(\|\| AcpError::invalid_params("unknown sessionId"))?;
1869	session.messages.push(Message {
1870	role: Role::User,
1871	content: vec![ContentBlock::Text {
1872	text: prompt,
1873	cache_control: None,
1874	}],
1875	});
1876	(
1877	session.messages.clone(),
1878	session.cwd.clone(),
1879	session.config.clone(),
1880	session.model.clone(),
1881	)
1882	};
1883
1884	Ok(PreparedPrompt {
1885	session_id,
1886	messages,
1887	cwd,
1888	config,
1889	model,
1890	})
1891	}
1892
1893	/// Commit the full message list produced by a completed turn into the
1894	/// session's history — the original history plus every assistant/tool-
1895	/// call/tool-result round the turn drove.
1896	///
1897	/// Called on all outcomes: normal completion, max-rounds, AND cancel.
1898	/// (The caller strips dangling assistant tool_use blocks on cancel before
1899	/// committing, which produces a clean partial history the next prompt can
1900	/// continue from instead of leaving the pre-turn state untouched.)
1901	fn commit_turn_messages(&mut self, session_id: &str, messages: Vec<Message>) {
1902	if let Some(session) = self.sessions.get_mut(session_id) {
1903	session.messages = messages;
1904	}
1905	}
1906
1907	/// Remove the last user message from the session history. Used to unwind
1908	/// the `begin_prompt` push when the turn itself fails (e.g. provider
1909	/// stream error), so the next prompt doesn't start with two consecutive
1910	/// `user` messages.
1911	fn rollback_user_message(&mut self, session_id: &str) {
1912	if let Some(session) = self.sessions.get_mut(session_id)
1913	&& session.messages.last().map(\|m\| m.role.as_str()) == Some("user")
1914	{
1915	session.messages.pop();
1916	}
1917	}
1918
1919	/// Resolve the route, build the streaming request, and open the provider
1920	/// response stream. Borrows `&self` only to read config/model; the returned
1921	/// [`StreamEventBox`] is `'static`, so the caller can race it against the
1922	/// reader without holding any borrow on the server. The cwd guard only needs
1923	/// to cover route resolution and client construction, not stream
1924	/// consumption, so it is dropped here.
1925	async fn open_prompt_stream(
1926	&self,
1927	config: &Config,
1928	selected_model: &str,
1929	messages: &[Message],
1930	cwd: &PathBuf,
1931	tool_registry: &ToolRegistry,
1932	frozen_system_prompt: &std::sync::Mutex<Option<SystemPrompt>>,
1933	) -> Result<StreamEventBox> {
1934	let _cwd_guard = ScopedCurrentDir::new(cwd)?;
1935	let last_user_text = messages
1936	.iter()
1937	.rev()
1938	.find_map(\|m\| {
1939	if m.role == "user" {
1940	m.content.iter().find_map(\|b\| match b {
1941	ContentBlock::Text { text, .. } => Some(text.as_str()),
1942	_ => None,
1943	})
1944	} else {
1945	None
1946	}
1947	})
1948	.unwrap_or("");
1949	let route = crate::resolve_cli_auto_route(config, selected_model, last_user_text).await?;
1950	let execution_config = crate::config_for_cli_route(config, &route);
1951	let client = CodewhaleClient::new(&execution_config)?;
1952	let model = route.model;
1953	let request_route = client.effective_route_envelope(&model, chrono::Utc::now());
1954	let reasoning_effort = route
1955	.reasoning_effort
1956	.and_then(\|effort\| {
1957	effort.api_value_for_route(
1958	execution_config.api_provider(),
1959	&execution_config.active_route_base_url(),
1960	&model,
1961	)
1962	})
1963	.map(str::to_string);
1964
1965	let tools = tool_registry.to_api_tools();
1966	let (route_limits, image_input) =
1967	resolve_acp_route_facts(&execution_config, request_route.provider, &model);
1968	let system = frozen_acp_system_prompt(
1969	frozen_system_prompt,
1970	&execution_config,
1971	cwd,
1972	request_route.provider,
1973	&request_route.model,
1974	route_limits,
1975	);
1976
1977	let mut outbound_messages = messages.to_vec();
1978	crate::image_attach::strip_images_when_unsupported(
1979	&mut outbound_messages,
1980	image_input,
1981	&request_route.model,
1982	);
1983	let request = MessageRequest {
1984	model,
1985	messages: outbound_messages,
1986	max_tokens: crate::route_budget::effective_max_output_tokens_for_route(
1987	request_route.provider,
1988	&request_route.model,
1989	route_limits,
1990	),
1991	system: Some(system),
1992	tools: Some(tools.clone()),
1993	tool_choice: if tools.is_empty() {
1994	None
1995	} else {
1996	Some(json!({ "type": "auto" }))
1997	},
1998	metadata: None,
1999	thinking: None,
2000	reasoning_effort,
2001	stream: Some(true),
2002	temperature: None,
2003	top_p: None,
2004	};
2005
2006	client.create_message_stream(request).await
2007	}
2008	}
2009
2010	fn resolve_acp_route_facts(
2011	config: &Config,
2012	provider: ApiProvider,
2013	model: &str,
2014	) -> (
2015	Option<codewhale_config::route::RouteLimits>,
2016	crate::model_profile::SupportState,
2017	) {
2018	let Ok(route) = crate::route_runtime::resolve_runtime_route(config, provider, Some(model))
2019	else {
2020	return (None, crate::model_profile::SupportState::Unknown);
2021	};
2022	(
2023	crate::route_budget::known_route_limits(route.candidate.limits()),
2024	route.candidate.capabilities().image_input,
2025	)
2026	}
2027
2028	/// Return the first fully composed ACP system prompt for this user turn.
2029	/// Later tool rounds clone that exact value instead of re-reading mutable
2030	/// instruction sources from disk.
2031	fn frozen_acp_system_prompt(
2032	slot: &std::sync::Mutex<Option<SystemPrompt>>,
2033	config: &Config,
2034	workspace: &std::path::Path,
2035	provider: ApiProvider,
2036	model: &str,
2037	route_limits: Option<codewhale_config::route::RouteLimits>,
2038	) -> SystemPrompt {
2039	let mut slot = match slot.lock() {
2040	Ok(slot) => slot,
2041	Err(poisoned) => poisoned.into_inner(),
2042	};
2043	if let Some(system) = slot.as_ref() {
2044	return system.clone();
2045	}
2046	let system = build_acp_system_prompt(config, workspace, provider, model, route_limits);
2047	*slot = Some(system.clone());
2048	system
2049	}
2050
2051	/// Compose ACP's stable prompt through the same headless host seam as
2052	/// `codewhale exec`. Tool availability remains owned by the request catalog;
2053	/// this function supplies the shared constitution, project instructions,
2054	/// configured instruction files, memory, locale, and route context.
2055	fn build_acp_system_prompt(
2056	config: &Config,
2057	workspace: &std::path::Path,
2058	provider: ApiProvider,
2059	model: &str,
2060	route_limits: Option<codewhale_config::route::RouteLimits>,
2061	) -> SystemPrompt {
2062	let settings = crate::settings::Settings::load().unwrap_or_default();
2063	let locale_tag = codewhale_localization::resolve_locale(&settings.locale)
2064	.tag()
2065	.to_string();
2066	let instructions = config
2067	.instructions_paths()
2068	.into_iter()
2069	.map(crate::prompts::InstructionSource::from)
2070	.collect::<Vec<_>>();
2071	let skills_dir = config.skills_dir();
2072	let user_memory_block = crate::native_memory::native_prompt_block(
2073	config.memory_enabled(),
2074	&config.memory_path(),
2075	workspace,
2076	);
2077
2078	crate::prompts::system_prompt_for_mode_with_context_skills_session_and_approval_for_host(
2079	workspace,
2080	None,
2081	Some(&skills_dir),
2082	Some(&instructions),
2083	crate::prompts::PromptSessionContext {
2084	user_memory_block: user_memory_block.as_deref(),
2085	goal_objective: None,
2086	project_context_pack_enabled: config.project_context_pack_enabled(),
2087	locale_tag: &locale_tag,
2088	translation_enabled: false,
2089	model_id: model,
2090	context_window_override: Some(crate::route_budget::route_context_window_tokens(
2091	provider,
2092	model,
2093	route_limits,
2094	)),
2095	verbosity: config.verbosity.as_deref(),
2096	skills_scan_codewhale_only: config.skills_config().scan_codewhale_only(),
2097	plugin_registry: None,
2098	recovery_hint: None,
2099	mode: acp_mode(config),
2100	},
2101	crate::prompts::PromptHost::Headless,
2102	)
2103	}
2104
2105	fn acp_mode(config: &Config) -> AppMode {
2106	if config.sandbox_mode.as_deref() == Some("read-only") {
2107	AppMode::Plan
2108	} else {
2109	AppMode::Agent
2110	}
2111	}
2112
2113	/// Approval posture for ACP turns, derived from server config instead of
2114	/// hardcoded: `--yolo` resolves to Bypass so an unattended headless session
2115	/// actually executes tools (#6337); otherwise the configured approval policy,
2116	/// else the Suggest default. Plan mode still pins read-only downstream
2117	/// regardless of posture.
2118	fn acp_approval_mode(config: &Config) -> ApprovalMode {
2119	if config.yolo.unwrap_or(false) {
2120	ApprovalMode::Bypass
2121	} else {
2122	config
2123	.approval_policy
2124	.as_deref()
2125	.and_then(ApprovalMode::from_config_value)
2126	.unwrap_or_default()
2127	}
2128	}
2129
2130	/// Build the tool registry for one ACP session, rooted at the session's
2131	/// `cwd`. Reuses the shared registry builders used by headless `exec` and the
2132	/// MCP adapter — no ACP-specific tool implementations.
2133	///
2134	/// Outside Plan mode, `Bash` is registered only when all three gates allow it: the
2135	/// client declares `clientCapabilities.terminal`, headless shell access is
2136	/// explicitly enabled in config, and the stable shell feature is enabled.
2137	/// Omitting any gate fails closed. The context also inherits the current
2138	/// mode-derived/configured sandbox boundary.
2139	/// `ToolContext::new` leaves `auto_approve` at its default (`false`), so the
2140	/// shell's own last-line safety check remains active after ACP's shared
2141	/// prepared-call, typed-policy, auto-review, repository-law, and explicit
2142	/// `session/request_permission` gates have admitted the call.
2143	fn build_acp_tool_registry(
2144	config: &Config,
2145	workspace: &std::path::Path,
2146	client_supports_terminal: bool,
2147	) -> ToolRegistry {
2148	let features = config.features();
2149	let external_sandbox_requested = config.sandbox_backend.as_deref().is_some_and(\|kind\| {
2150	let kind = kind.trim();
2151	!kind.is_empty() && !kind.eq_ignore_ascii_case("none")
2152	});
2153	let sandbox_backend = match crate::sandbox::backend::create_backend(config) {
2154	Ok(backend) => backend
2155	.filter(\|backend\| backend.kind() != crate::sandbox::backend::SandboxKind::Unsupported)
2156	.map(std::sync::Arc::from),
2157	Err(error) => {
2158	tracing::warn!("Failed to create ACP sandbox backend: {error}");
2159	None
2160	}
2161	};
2162	// A requested external sandbox is an execution boundary, not a hint. If
2163	// it cannot be constructed, omit Bash instead of silently running the
2164	// command on the local host.
2165	let sandbox_backend_ready = !external_sandbox_requested \|\| sandbox_backend.is_some();
2166	let allow_shell = acp_mode(config) != AppMode::Plan
2167	&& client_supports_terminal
2168	&& config.allow_shell()
2169	&& features.enabled(crate::features::Feature::ShellTool)
2170	&& sandbox_backend_ready;
2171	let shell_policy = if allow_shell {
2172	ShellPolicy::Full
2173	} else {
2174	ShellPolicy::None
2175	};
2176	let sandbox_policy = crate::core::authority::sandbox_policy_for_turn(
2177	acp_mode(config),
2178	acp_approval_mode(config),
2179	config.sandbox_mode.as_deref(),
2180	workspace,
2181	crate::core::authority::SandboxNetworkAccess::from_config(config.sandbox_network_access),
2182	);
2183	let mut context = ToolContext::new(workspace)
2184	.with_shell_policy(shell_policy)
2185	.with_elevated_sandbox_policy(sandbox_policy);
2186	if acp_mode(config) == AppMode::Plan {
2187	// Use the shared headless authority cap for file/Git dispatch too:
2188	// an OS shell sandbox alone cannot prevent in-process tool writes.
2189	context.tool_authority = Some(Arc::new(crate::tools::spec::ToolAuthorityEnvelope {
2190	schema_version: 1,
2191	owner: "acp-plan".to_string(),
2192	authority: crate::tools::spec::ToolMutationAuthority::ReadOnly,
2193	network_access: Some(false),
2194	shell: crate::tools::spec::ToolShellAuthority::None,
2195	verification: crate::tools::spec::ToolVerificationAuthority::None,
2196	writable_roots: Vec::new(),
2197	writable_files: Vec::new(),
2198	coordination_contracts: Vec::new(),
2199	}));
2200	}
2201	match context.shell_manager.lock() {
2202	Ok(mut manager) => manager.set_prefer_bwrap(config.prefer_bwrap.unwrap_or(false)),
2203	Err(poisoned) => poisoned
2204	.into_inner()
2205	.set_prefer_bwrap(config.prefer_bwrap.unwrap_or(false)),
2206	}
2207	if let Some(backend) = sandbox_backend {
2208	context = context.with_sandbox_backend(backend);
2209	}
2210	let hooks_config =
2211	crate::hooks::HooksConfig::load_with_project(config.hooks_config(), workspace);
2212	context.runtime.hook_executor = Some(Arc::new(crate::hooks::HookExecutor::new(
2213	hooks_config,
2214	workspace.to_path_buf(),
2215	)));
2216
2217	let mut builder = ToolRegistryBuilder::new()
2218	.with_file_tools()
2219	.with_search_tools()
2220	.with_git_tools();
2221	if features.enabled(crate::features::Feature::ApplyPatch) {
2222	builder = builder.with_patch_tools();
2223	}
2224	if allow_shell {
2225	builder = builder.with_foreground_shell_tools();
2226	}
2227
2228	let mut registry = builder.build(context);
2229	// ACP does not load arbitrary plugin replacements in v0.9.6, but it must
2230	// never fall through to a built-in the operator disabled or replaced.
2231	if let Some(overrides) = config
2232	.tools
2233	.as_ref()
2234	.and_then(\|tools\| tools.overrides.as_ref())
2235	{
2236	for tool_name in overrides.keys() {
2237	remove_acp_overridden_builtin(&mut registry, tool_name);
2238	}
2239	}
2240	registry
2241	}
2242
2243	/// ACP does not load executable tool replacements in v0.9.6. Remove the
2244	/// built-in compatibility family for every configured override so neither a
2245	/// hidden legacy alias nor a newly canonical lowercase name can fall through
2246	/// to the original implementation.
2247	fn remove_acp_overridden_builtin(registry: &mut ToolRegistry, tool_name: &str) {
2248	let aliases: &[&str] = match tool_name {
2249	"bash" \| "Bash" \| "exec_shell" => &["bash", "Bash", "exec_shell"],
2250	"read" \| "write" \| "edit" \| "File" \| "read_file" \| "write_file" \| "edit_file" => &[
2251	"read",
2252	"write",
2253	"edit",
2254	"File",
2255	"read_file",
2256	"write_file",
2257	"edit_file",
2258	],
2259	"apply_patch" => &["apply_patch"],
2260	_ => std::slice::from_ref(&tool_name),
2261	};
2262	for alias in aliases {
2263	registry.remove_tool(alias);
2264	}
2265	}
2266
2267	/// ACP `kind` hint for a tool call, used by the client to pick an icon/label.
2268	/// Falls back to `"other"` for tools without an obvious category.
2269	///
2270	/// `File` is a single canonical tool covering read/list/search/write/edit/
2271	/// patch (#4625), so its kind depends on the `action` argument rather than
2272	/// the tool name alone.
2273	fn tool_call_kind(call: &PendingToolCall) -> &'static str {
2274	match call.name.as_str() {
2275	"File" => match call.input.get("action").and_then(Value::as_str) {
2276	Some("write" \| "edit" \| "patch") => "edit",
2277	_ => "read",
2278	},
2279	"apply_patch" => "edit",
2280	"Git" => "read",
2281	"bash" \| "Bash" \| "terminal/run" \| "terminal/send" \| "terminal/wait"
2282	\| "terminal/cancel" \| "terminal/reset" => "execute",
2283	_ => "other",
2284	}
2285	}
2286
2287	/// Human-readable title for a tool call: the tool name plus its primary
2288	/// argument (path/command/pattern) when present, so the client's tool-call
2289	/// card is legible without expanding raw input.
2290	fn tool_call_title(call: &PendingToolCall) -> String {
2291	let detail = call
2292	.input
2293	.get("path")
2294	.or_else(\|\| call.input.get("command"))
2295	.or_else(\|\| call.input.get("pattern"))
2296	.or_else(\|\| call.input.get("task_id"))
2297	.and_then(Value::as_str);
2298	match detail {
2299	Some(detail) => format!("{}: {}", call.name, detail),
2300	None => call.name.clone(),
2301	}
2302	}
2303
2304	fn truncate_for_acp(content: &str) -> String {
2305	if content.chars().count() <= TOOL_CALL_CONTENT_PREVIEW_CHARS {
2306	return content.to_string();
2307	}
2308	let truncated: String = content
2309	.chars()
2310	.take(TOOL_CALL_CONTENT_PREVIEW_CHARS)
2311	.collect();
2312	format!("{truncated}\n… [truncated for display; the full result was sent to the model]")
2313	}
2314
2315	async fn write_tool_call_start<W>(
2316	writer: &mut W,
2317	session_id: &str,
2318	call: &PendingToolCall,
2319	) -> Result<()>
2320	where
2321	W: AsyncWrite + Unpin,
2322	{
2323	let notification = json!({
2324	"jsonrpc": "2.0",
2325	"method": "session/update",
2326	"params": {
2327	"sessionId": session_id,
2328	"update": {
2329	"sessionUpdate": "tool_call",
2330	"toolCallId": call.id,
2331	"title": tool_call_title(call),
2332	"kind": tool_call_kind(call),
2333	"status": "pending",
2334	"rawInput": call.input,
2335	}
2336	}
2337	});
2338	write_json_line(writer, notification).await
2339	}
2340
2341	async fn write_tool_call_update<W>(
2342	writer: &mut W,
2343	session_id: &str,
2344	call: &PendingToolCall,
2345	status: &str,
2346	content: Option<&str>,
2347	) -> Result<()>
2348	where
2349	W: AsyncWrite + Unpin,
2350	{
2351	write_tool_call_update_with_blocks(writer, session_id, call, status, content, &[]).await
2352	}
2353
2354	async fn write_tool_call_update_with_blocks<W>(
2355	writer: &mut W,
2356	session_id: &str,
2357	call: &PendingToolCall,
2358	status: &str,
2359	content: Option<&str>,
2360	rich_blocks: &[codewhale_tools::ToolResultContentBlock],
2361	) -> Result<()>
2362	where
2363	W: AsyncWrite + Unpin,
2364	{
2365	let mut update = json!({
2366	"sessionUpdate": "tool_call_update",
2367	"toolCallId": call.id,
2368	"status": status,
2369	});
2370	if content.is_some() \|\| !rich_blocks.is_empty() {
2371	let mut blocks = Vec::with_capacity(rich_blocks.len() + usize::from(content.is_some()));
2372	if let Some(content) = content {
2373	blocks.push(json!({
2374	"type": "content",
2375	"content": { "type": "text", "text": truncate_for_acp(content) }
2376	}));
2377	}
2378	blocks.extend(rich_blocks.iter().map(\|block\| match block {
2379	codewhale_tools::ToolResultContentBlock::Image { mime_type, data } => json!({
2380	"type": "content",
2381	"content": { "type": "image", "data": data, "mimeType": mime_type }
2382	}),
2383	}));
2384	update["content"] = json!(blocks);
2385	}
2386	let notification = json!({
2387	"jsonrpc": "2.0",
2388	"method": "session/update",
2389	"params": {
2390	"sessionId": session_id,
2391	"update": update
2392	}
2393	});
2394	write_json_line(writer, notification).await
2395	}
2396
2397	struct ScopedCurrentDir {
2398	prior: PathBuf,
2399	}
2400
2401	impl ScopedCurrentDir {
2402	fn new(cwd: &PathBuf) -> Result<Self> {
2403	let prior = std::env::current_dir()?;
2404	if cwd.as_os_str().is_empty() {
2405	return Ok(Self { prior });
2406	}
2407	std::env::set_current_dir(cwd)
2408	.map_err(\|err\| anyhow!("failed to enter ACP session cwd {}: {err}", cwd.display()))?;
2409	Ok(Self { prior })
2410	}
2411	}
2412
2413	impl Drop for ScopedCurrentDir {
2414	fn drop(&mut self) {
2415	let _ = std::env::set_current_dir(&self.prior);
2416	}
2417	}
2418
2419	impl AcpError {
2420	fn invalid_params(message: impl Into<String>) -> Self {
2421	Self {
2422	code: -32602,
2423	message: message.into(),
2424	}
2425	}
2426
2427	/// JSON-RPC internal error: the request was well-formed and the agent
2428	/// could not serve it.
2429	fn internal(message: impl Into<String>) -> Self {
2430	Self {
2431	code: -32603,
2432	message: message.into(),
2433	}
2434	}
2435
2436	fn method_not_found(method: &str) -> Self {
2437	Self {
2438	code: -32601,
2439	message: format!("method not found: {method}"),
2440	}
2441	}
2442	}
2443
2444	fn initialize_result(client_protocol_version: Option<u64>, config: &Config) -> Value {
2445	json!({
2446	"protocolVersion": client_protocol_version
2447	.map(\|version\| version.min(ACP_PROTOCOL_VERSION))
2448	.unwrap_or(ACP_PROTOCOL_VERSION),
2449	"agentCapabilities": {
2450	"loadSession": true,
2451	"modelSelection": true,
2452	"promptCapabilities": {
2453	"image": false,
2454	"audio": false,
2455	"embeddedContext": true
2456	},
2457	"mcpCapabilities": {
2458	"http": false,
2459	"sse": false
2460	},
2461	// ACP `SessionCapabilities` fields are objects, never booleans:
2462	// `{}` means "supported", absent/null means "not supported"
2463	// (#5969 — a boolean here made JetBrains' strictly-typed client
2464	// fail the handshake and kill the agent). `session/load` support
2465	// is advertised by the top-level `loadSession` above; it is not a
2466	// field of `sessionCapabilities`. We only claim `list` because
2467	// `session/list` is the only one of the optional session methods
2468	// this server dispatches.
2469	"sessionCapabilities": {
2470	"list": {}
2471	}
2472	},
2473	"agentInfo": {
2474	"name": "codewhale",
2475	"title": "codewhale",
2476	"version": env!("CARGO_PKG_VERSION")
2477	},
2478	"authMethods": acp_auth_methods(config)
2479	})
2480	}
2481
2482	fn acp_auth_methods(config: &Config) -> Value {
2483	let provider = config.api_provider().as_str();
2484	json!([
2485	{
2486	"id": "codewhale-terminal-auth",
2487	"name": "Set Codewhale API key",
2488	"description": format!("Run Codewhale's terminal credential setup for the {provider} provider."),
2489	"type": "terminal",
2490	"args": ["auth", "set", "--provider", provider],
2491	"env": {}
2492	}
2493	])
2494	}
2495
2496	fn extract_prompt_text(prompt: Option<&Value>) -> Option<String> {
2497	match prompt? {
2498	Value::String(text) => Some(text.clone()),
2499	Value::Array(blocks) => {
2500	let parts = blocks
2501	.iter()
2502	.filter_map(content_block_text)
2503	.collect::<Vec<_>>();
2504	(!parts.is_empty()).then(\|\| parts.join("\n\n"))
2505	}
2506	_ => None,
2507	}
2508	}
2509
2510	fn content_block_text(block: &Value) -> Option<String> {
2511	match block.get("type").and_then(Value::as_str)? {
2512	"text" => block
2513	.get("text")
2514	.and_then(Value::as_str)
2515	.map(str::to_string),
2516	"resource" => resource_text(block),
2517	"resource_link" \| "resourceLink" => resource_link_text(block),
2518	_ => None,
2519	}
2520	}
2521
2522	fn resource_text(block: &Value) -> Option<String> {
2523	let resource = block.get("resource").unwrap_or(block);
2524	if let Some(text) = resource.get("text").and_then(Value::as_str) {
2525	return Some(text.to_string());
2526	}
2527	resource_link_text(resource)
2528	}
2529
2530	fn resource_link_text(block: &Value) -> Option<String> {
2531	let uri = block
2532	.get("uri")
2533	.or_else(\|\| block.pointer("/resource/uri"))
2534	.and_then(Value::as_str)?;
2535	Some(format!("@{uri}"))
2536	}
2537
2538	async fn write_session_update<W>(writer: &mut W, session_id: &str, text: String) -> Result<()>
2539	where
2540	W: AsyncWrite + Unpin,
2541	{
2542	let notification = json!({
2543	"jsonrpc": "2.0",
2544	"method": "session/update",
2545	"params": {
2546	"sessionId": session_id,
2547	"update": {
2548	"sessionUpdate": "agent_message_chunk",
2549	"content": {
2550	"type": "text",
2551	"text": text
2552	}
2553	}
2554	}
2555	});
2556	write_json_line(writer, notification).await
2557	}
2558
2559	async fn write_jsonrpc_result<W>(writer: &mut W, id: Value, result: Value) -> Result<()>
2560	where
2561	W: AsyncWrite + Unpin,
2562	{
2563	write_json_line(
2564	writer,
2565	json!({
2566	"jsonrpc": "2.0",
2567	"id": id,
2568	"result": result
2569	}),
2570	)
2571	.await
2572	}
2573
2574	async fn write_jsonrpc_error<W>(
2575	writer: &mut W,
2576	id: Option<Value>,
2577	code: i32,
2578	message: impl Into<String>,
2579	) -> Result<()>
2580	where
2581	W: AsyncWrite + Unpin,
2582	{
2583	write_json_line(
2584	writer,
2585	json!({
2586	"jsonrpc": "2.0",
2587	"id": id,
2588	"error": {
2589	"code": code,
2590	"message": message.into()
2591	}
2592	}),
2593	)
2594	.await
2595	}
2596
2597	async fn write_json_line<W>(writer: &mut W, value: Value) -> Result<()>
2598	where
2599	W: AsyncWrite + Unpin,
2600	{
2601	writer.write_all(value.to_string().as_bytes()).await?;
2602	writer.write_all(b"\n").await?;
2603	writer.flush().await?;
2604	Ok(())
2605	}
2606
2607	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2608	enum JsonRpcResponseIdPolicy {
2609	/// JSON-RPC's normal contract: echo the request id without changing type.
2610	Preserve,
2611	/// Zed's ACP client currently decodes response ids as strings even when it
2612	/// sent a number. Keep this narrow compatibility mode client-identified.
2613	StringifyNumeric,
2614	}
2615
2616	impl JsonRpcResponseIdPolicy {
2617	fn from_initialize_params(params: &Value) -> Self {
2618	let client_name = params
2619	.pointer("/clientInfo/name")
2620	.and_then(Value::as_str)
2621	.unwrap_or_default();
2622	if client_name.eq_ignore_ascii_case("zed") {
2623	Self::StringifyNumeric
2624	} else {
2625	Self::Preserve
2626	}
2627	}
2628
2629	fn response_id(self, id: Value) -> Value {
2630	match (self, id) {
2631	(Self::StringifyNumeric, Value::Number(number)) => Value::String(number.to_string()),
2632	(_, id) => id,
2633	}
2634	}
2635	}
2636
2637	#[cfg(test)]
2638	mod tests {
2639	use super::*;
2640	use std::cell::RefCell;
2641	use std::collections::VecDeque;
2642
2643	#[tokio::test]
2644	async fn tool_update_emits_typed_acp_image_content() {
2645	let mut output = Vec::new();
2646	let call = PendingToolCall {
2647	id: "call_image_1".to_string(),
2648	name: "read".to_string(),
2649	input: json!({"path": "shot.png"}),
2650	parse_error: None,
2651	};
2652	write_tool_call_update_with_blocks(
2653	&mut output,
2654	"session_1",
2655	&call,
2656	"completed",
2657	Some("screenshot captured"),
2658	&[codewhale_tools::ToolResultContentBlock::Image {
2659	mime_type: "image/png".to_string(),
2660	data: "QUJD".to_string(),
2661	}],
2662	)
2663	.await
2664	.expect("ACP update");
2665
2666	let lines = parse_lines(output);
2667	let content = lines[0]["params"]["update"]["content"]
2668	.as_array()
2669	.expect("ACP content blocks");
2670	assert_eq!(content[0]["content"]["type"], "text");
2671	assert_eq!(content[1]["content"]["type"], "image");
2672	assert_eq!(content[1]["content"]["mimeType"], "image/png");
2673	assert_eq!(content[1]["content"]["data"], "QUJD");
2674	}
2675
2676	/// #5864: `serve --acp` implemented `initialize` and `session/new` and
2677	/// nothing else, so ACP clients that offer session history could not
2678	/// enumerate or resume anything. ACP sessions are in-memory and capped;
2679	/// the durable Codewhale sessions are what "resume" means.
2680	#[tokio::test]
2681	async fn session_list_and_load_reach_the_durable_codewhale_sessions() {
2682	let _guard = crate::test_support::lock_test_env();
2683	let home = tempfile::TempDir::new().expect("isolated codewhale home");
2684	let _home_guard =
2685	crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", home.path().as_os_str());
2686
2687	let workspace = home.path().join("workspace");
2688	std::fs::create_dir_all(&workspace).expect("workspace");
2689	let saved = crate::session_manager::create_saved_session(
2690	&[Message {
2691	role: Role::User,
2692	content: vec![ContentBlock::Text {
2693	text: "what did we decide?".to_string(),
2694	cache_control: None,
2695	}],
2696	}],
2697	"deepseek-v4-flash",
2698	&workspace,
2699	42,
2700	None,
2701	);
2702	let saved_id = saved.metadata.id.clone();
2703	let manager = crate::session_manager::SessionManager::new(
2704	crate::session_manager::default_sessions_dir().expect("sessions dir"),
2705	)
2706	.expect("session manager");
2707	manager.save_session(&saved).expect("save fixture session");
2708
2709	let mut server = AcpServer::new(
2710	Config::default(),
2711	"deepseek-v4-flash".to_string(),
2712	workspace.clone(),
2713	);
2714
2715	let listed = server.list_sessions(json!({})).expect("session/list");
2716	let ids: Vec<&str> = listed["sessions"]
2717	.as_array()
2718	.expect("sessions array")
2719	.iter()
2720	.filter_map(\|entry\| entry["sessionId"].as_str())
2721	.collect();
2722	assert!(
2723	ids.contains(&saved_id.as_str()),
2724	"session/list must enumerate durable sessions: {ids:?}"
2725	);
2726
2727	let other_workspace = home.path().join("other-workspace");
2728	std::fs::create_dir_all(&other_workspace).unwrap();
2729	let other = crate::session_manager::create_saved_session(
2730	&saved.messages,
2731	"deepseek-v4-flash",
2732	&other_workspace,
2733	0,
2734	None,
2735	);
2736	manager.save_session(&other).unwrap();
2737	for filter in [workspace.clone(), workspace.join(".")] {
2738	let AcpDispatch::Response(filtered) = server
2739	.handle_request("session/list", json!({"cwd": filter}))
2740	.await
2741	.expect("filtered session/list")
2742	else {
2743	panic!("session/list returned shutdown");
2744	};
2745	let entries = filtered["sessions"].as_array().unwrap();
2746	assert_eq!(
2747	entries.len(),
2748	1,
2749	"workspace filter must not include another directory"
2750	);
2751	assert_eq!(entries[0]["sessionId"], saved_id);
2752	}
2753	let AcpDispatch::Response(all) = server
2754	.handle_request("session/list", json!({}))
2755	.await
2756	.unwrap()
2757	else {
2758	panic!("session/list returned shutdown");
2759	};
2760	assert_eq!(all["sessions"].as_array().unwrap().len(), 2);
2761	let AcpDispatch::Response(empty) = server
2762	.handle_request(
2763	"session/list",
2764	json!({"cwd": home.path().join("missing-workspace")}),
2765	)
2766	.await
2767	.unwrap()
2768	else {
2769	panic!("session/list returned shutdown");
2770	};
2771	assert!(empty["sessions"].as_array().unwrap().is_empty());
2772	for invalid in [json!("relative"), json!(""), json!(42)] {
2773	let error = server
2774	.handle_request("session/list", json!({"cwd": invalid}))
2775	.await
2776	.err()
2777	.expect("invalid cwd must be rejected");
2778	assert_eq!(error.code, -32602);
2779	}
2780
2781	let loaded = server
2782	.load_session(json!({ "sessionId": saved_id }))
2783	.expect("session/load");
2784	assert_eq!(loaded["sessionId"], saved_id);
2785	assert!(
2786	loaded["configOptions"]
2787	.as_array()
2788	.is_some_and(\|options\| options.len() == 2)
2789	);
2790	let session = server
2791	.sessions
2792	.get(&saved_id)
2793	.expect("loaded session is addressable by its own id");
2794	assert_eq!(session.cwd, workspace, "cwd comes from the saved workspace");
2795	assert_eq!(
2796	session.messages.len(),
2797	1,
2798	"the conversation is rehydrated, not started empty"
2799	);
2800
2801	// A session that does not exist is a client error, not a panic.
2802	let missing = server.load_session(json!({ "sessionId": "codewhale-nope" }));
2803	assert_eq!(missing.expect_err("unknown session").code, -32602);
2804	let no_id = server.load_session(json!({}));
2805	assert_eq!(no_id.expect_err("missing sessionId").code, -32602);
2806	}
2807
2808	/// #6245: reloading a tracked session by a short prefix must not push a
2809	/// duplicate `insertion_order` entry. The duplicate made the deque
2810	/// disagree with `sessions`, so a later capacity eviction popped the
2811	/// stale front copy of a just-reloaded session and removed a live
2812	/// conversation.
2813	#[tokio::test]
2814	async fn loading_a_tracked_session_by_prefix_does_not_duplicate_ordering() {
2815	let _guard = crate::test_support::lock_test_env();
2816	let home = tempfile::TempDir::new().expect("isolated codewhale home");
2817	let _home_guard =
2818	crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", home.path().as_os_str());
2819
2820	let workspace = home.path().join("workspace");
2821	std::fs::create_dir_all(&workspace).expect("workspace");
2822	let saved = crate::session_manager::create_saved_session(
2823	&[Message {
2824	role: Role::User,
2825	content: vec![ContentBlock::Text {
2826	text: "reload me by prefix".to_string(),
2827	cache_control: None,
2828	}],
2829	}],
2830	"deepseek-v4-flash",
2831	&workspace,
2832	42,
2833	None,
2834	);
2835	let saved_id = saved.metadata.id.clone();
2836	let manager = crate::session_manager::SessionManager::new(
2837	crate::session_manager::default_sessions_dir().expect("sessions dir"),
2838	)
2839	.expect("session manager");
2840	manager.save_session(&saved).expect("save fixture session");
2841
2842	let mut server = AcpServer::new(
2843	Config::default(),
2844	"deepseek-v4-flash".to_string(),
2845	workspace.clone(),
2846	);
2847
2848	// Load by the full id first: the session becomes tracked exactly once.
2849	let loaded = server
2850	.load_session(json!({ "sessionId": saved_id }))
2851	.expect("load by full id");
2852	assert_eq!(loaded["sessionId"], saved_id);
2853
2854	// A prefix resolving to the same tracked id must be idempotent, not
2855	// a second insertion.
2856	let prefix: String = saved_id.chars().take(8).collect();
2857	let reloaded = server
2858	.load_session(json!({ "sessionId": prefix }))
2859	.expect("load by prefix");
2860	assert_eq!(reloaded["sessionId"], saved_id);
2861
2862	assert_eq!(server.sessions.len(), 1);
2863	assert_eq!(
2864	server.insertion_order.len(),
2865	server.sessions.len(),
2866	"a prefix reload of a tracked session must not duplicate the ordering entry"
2867	);
2868	assert_eq!(
2869	server
2870	.insertion_order
2871	.iter()
2872	.filter(\|id\| *id == &saved_id)
2873	.count(),
2874	1,
2875	"the ordering deque holds the tracked id exactly once"
2876	);
2877	}
2878
2879	#[tokio::test]
2880	async fn standard_session_configuration_is_offered_and_scoped_to_one_session() {
2881	let workspace = tempfile::tempdir().unwrap();
2882	let mut server = AcpServer::new(
2883	Config::default(),
2884	"deepseek-v4-flash".into(),
2885	workspace.path().into(),
2886	);
2887	let first = server.new_session(json!({})).unwrap();
2888	let second = server.new_session(json!({})).unwrap();
2889	let first_id = first["sessionId"].as_str().unwrap();
2890	let second_id = second["sessionId"].as_str().unwrap();
2891	assert_eq!(first["modes"]["currentModeId"], "agent");
2892	assert_eq!(first["models"]["currentModelId"], "deepseek-v4-flash");
2893	let alternative = first["models"]["availableModels"]
2894	.as_array()
2895	.unwrap()
2896	.iter()
2897	.find_map(\|row\| {
2898	row["modelId"]
2899	.as_str()
2900	.filter(\|model\| *model != "deepseek-v4-flash")
2901	})
2902	.expect("shared active-provider catalog offers another model");
2903
2904	server
2905	.handle_request(
2906	"session/set_model",
2907	json!({"sessionId": first_id, "modelId": alternative}),
2908	)
2909	.await
2910	.unwrap();
2911	let AcpDispatch::Response(configured) = server
2912	.handle_request(
2913	"session/set_config_option",
2914	json!({"sessionId": first_id, "configId": "mode", "value": "plan"}),
2915	)
2916	.await
2917	.unwrap()
2918	else {
2919	panic!("configuration response")
2920	};
2921	assert_eq!(configured["configOptions"].as_array().unwrap().len(), 2);
2922	assert_eq!(configured["configOptions"][0]["currentValue"], "plan");
2923	assert_eq!(configured["configOptions"][1]["currentValue"], alternative);
2924	assert_eq!(
2925	server.session_configuration(second_id),
2926	second,
2927	"another session is unchanged"
2928	);
2929	let prepared = server
2930	.begin_prompt(
2931	json!({"sessionId": first_id, "prompt": [{"type": "text", "text": "review this"}]}),
2932	)
2933	.unwrap();
2934	assert_eq!(
2935	prepared.model, alternative,
2936	"provider request receives the session model"
2937	);
2938	assert_eq!(acp_mode(&prepared.config), AppMode::Plan);
2939	server
2940	.handle_request(
2941	"session/set_mode",
2942	json!({"sessionId": first_id, "modeId": "agent"}),
2943	)
2944	.await
2945	.unwrap();
2946	assert_eq!(
2947	server.sessions[first_id].messages.len(),
2948	1,
2949	"configuration retains history"
2950	);
2951	assert_eq!(
2952	server.sessions[first_id].config.sandbox_mode,
2953	server.config.sandbox_mode
2954	);
2955	assert_eq!(
2956	server.model, "deepseek-v4-flash",
2957	"session setters do not alter defaults"
2958	);
2959	}
2960
2961	#[tokio::test]
2962	async fn plan_configuration_enforces_shared_read_only_tools() {
2963	let workspace = tempfile::tempdir().unwrap();
2964	let config = Config {
2965	allow_shell: Some(true),
2966	sandbox_mode: Some("danger-full-access".into()),
2967	..Config::default()
2968	};
2969	let mut server =
2970	AcpServer::new(config, "deepseek-v4-flash".into(), workspace.path().into());
2971	server.client_supports_terminal = true;
2972	let new = server.new_session(json!({})).unwrap();
2973	let id = new["sessionId"].as_str().unwrap();
2974	server
2975	.set_session_config(json!({"sessionId": id, "configId": "mode", "value": "plan"}))
2976	.unwrap();
2977	let registry = server.session_tool_registry(id).unwrap();
2978	assert!(
2979	registry.get("Bash").is_none(),
2980	"Plan does not offer shell execution"
2981	);
2982	let target = workspace.path().join("must-not-exist.txt");
2983	assert!(
2984	registry.get("write").is_some(),
2985	"exercise the real shared file writer"
2986	);
2987	let outcome = registry
2988	.execute_full(
2989	"write",
2990	json!({"path": target, "content": "unauthorized write"}),
2991	)
2992	.await;
2993	assert!(
2994	matches!(outcome, Err(ToolError::PermissionDenied { .. })),
2995	"the shared authority must reject mutation: {outcome:?}"
2996	);
2997	assert!(!target.exists());
2998	}
2999
3000	#[test]
3001	fn acp_approval_mode_derives_from_server_config() {
3002	// #6337: `--yolo --danger-full-access` must not silently run as Ask.
3003	let yolo = Config {
3004	yolo: Some(true),
3005	..Config::default()
3006	};
3007	assert_eq!(acp_approval_mode(&yolo), ApprovalMode::Bypass);
3008	let policy = Config {
3009	approval_policy: Some("never".into()),
3010	..Config::default()
3011	};
3012	assert_eq!(acp_approval_mode(&policy), ApprovalMode::Never);
3013	assert_eq!(acp_approval_mode(&Config::default()), ApprovalMode::Suggest);
3014	}
3015
3016	#[test]
3017	fn yolo_admission_auto_executes_write_without_permission_round_trip() {
3018	// #6337: an unattended `--yolo` session must execute tools instead of
3019	// stalling on permission requests no headless client answers.
3020	let (dir, registry) = workspace_registry();
3021	let config = Config {
3022	yolo: Some(true),
3023	..Config::default()
3024	};
3025	let call = pending_call(
3026	"File",
3027	json!({"action": "write", "path": "yolo.txt", "content": "yolo"}),
3028	);
3029	let (_, admission) = prepare_acp_tool_admission(&config, &registry, &call).unwrap();
3030	assert_eq!(admission, AcpToolAdmission::Auto);
3031	assert_eq!(registry.context().workspace, dir.path());
3032	}
3033
3034	#[test]
3035	fn default_admission_still_requests_permission_for_write() {
3036	// Pins the Ask default the yolo test above contrasts with: without
3037	// `--yolo`, a write surfaces a permission request to the client.
3038	let (_dir, registry) = workspace_registry();
3039	let call = pending_call(
3040	"File",
3041	json!({"action": "write", "path": "ask.txt", "content": "ask"}),
3042	);
3043	let (_, admission) =
3044	prepare_acp_tool_admission(&Config::default(), &registry, &call).unwrap();
3045	assert!(matches!(admission, AcpToolAdmission::RequestPermission(_)));
3046	}
3047
3048	#[tokio::test]
3049	async fn plan_mode_stays_read_only_under_yolo() {
3050	// The posture derivation must never loosen the Plan guardrail.
3051	let dir = tempfile::tempdir().expect("tempdir");
3052	let config = Config {
3053	yolo: Some(true),
3054	sandbox_mode: Some("read-only".into()),
3055	..Config::default()
3056	};
3057	let registry = build_acp_tool_registry(&config, dir.path(), false);
3058	let target = dir.path().join("must-not-exist.txt");
3059	let outcome = registry
3060	.execute_full("write", json!({"path": target, "content": "x"}))
3061	.await;
3062	assert!(
3063	matches!(outcome, Err(ToolError::PermissionDenied { .. })),
3064	"Plan stays read-only under yolo: {outcome:?}"
3065	);
3066	assert!(!target.exists());
3067	}
3068
3069	#[test]
3070	fn session_configuration_cannot_relax_a_configured_floor_or_invent_values() {
3071	let workspace = tempfile::tempdir().unwrap();
3072	let config = Config {
3073	sandbox_mode: Some("read-only".into()),
3074	..Config::default()
3075	};
3076	let mut server =
3077	AcpServer::new(config, "deepseek-v4-flash".into(), workspace.path().into());
3078	let before = server.new_session(json!({})).unwrap();
3079	let id = before["sessionId"].as_str().unwrap();
3080	assert_eq!(before["modes"]["currentModeId"], "plan");
3081	for params in [
3082	json!({"sessionId": id, "configId": "mode", "value": "agent"}),
3083	json!({"sessionId": id, "configId": "model", "value": "unknown-model"}),
3084	json!({"sessionId": id, "configId": "permission", "value": "bypass"}),
3085	json!({"sessionId": id, "configId": "mode", "value": true}),
3086	json!({"sessionId": "missing", "configId": "mode", "value": "plan"}),
3087	json!({"configId": "mode", "value": "plan"}),
3088	] {
3089	assert_eq!(server.set_session_config(params).unwrap_err().code, -32602);
3090	assert_eq!(server.session_configuration(id), before);
3091	}
3092	}
3093
3094	#[test]
3095	fn initialize_advertises_baseline_acp_agent() {
3096	let result = initialize_result(Some(1), &Config::default());
3097
3098	assert_eq!(result["protocolVersion"], 1);
3099	assert_eq!(result["agentInfo"]["name"], "codewhale");
3100	// #5864: enumerating and resuming durable Codewhale sessions is now
3101	// served, so the capability says so rather than declining it.
3102	// #5969: this test used to assert `list == true` and `load == true`,
3103	// certifying the wire format that broke every strictly-typed client.
3104	// `loadSession` is the top-level boolean that advertises
3105	// `session/load`; inside `sessionCapabilities` every field is a
3106	// capability object, and `load` is not a field at all. Comparing the
3107	// whole object pins both halves: a boolean `list` or a resurrected
3108	// `load` key fails here.
3109	assert_eq!(result["agentCapabilities"]["loadSession"], true);
3110	assert_eq!(
3111	result["agentCapabilities"]["sessionCapabilities"],
3112	json!({"list": {}})
3113	);
3114	assert!(
3115	result["agentCapabilities"]["sessionCapabilities"]["list"].is_object(),
3116	"sessionCapabilities.list must be a SessionListCapabilities object, got {}",
3117	result["agentCapabilities"]["sessionCapabilities"]["list"]
3118	);
3119	assert_eq!(
3120	result["agentCapabilities"]["promptCapabilities"]["embeddedContext"],
3121	true
3122	);
3123	assert_eq!(result["authMethods"][0]["type"], "terminal");
3124	assert_eq!(
3125	result["authMethods"][0]["args"],
3126	json!(["auth", "set", "--provider", "deepseek"])
3127	);
3128	}
3129
3130	#[test]
3131	fn initialize_advertises_model_selection_capability() {
3132	let result = initialize_result(Some(1), &Config::default());
3133
3134	assert_eq!(result["agentCapabilities"]["modelSelection"], true);
3135	}
3136
3137	#[test]
3138	fn list_providers_returns_provider_set() {
3139	let server = AcpServer::new(
3140	Config::default(),
3141	"deepseek-chat".into(),
3142	PathBuf::from("/tmp"),
3143	);
3144	let result = server.list_providers();
3145	let providers = result["providers"].as_array().expect("providers array");
3146
3147	assert!(!providers.is_empty());
3148	assert!(
3149	providers
3150	.iter()
3151	.any(\|provider\| provider["id"] == "deepseek")
3152	);
3153	}
3154
3155	#[test]
3156	fn current_model_reflects_constructor_default() {
3157	let config = Config::default();
3158	let expected_provider = config.api_provider().as_str();
3159	let server = AcpServer::new(config, "deepseek-reasoner".into(), PathBuf::from("/tmp"));
3160	let result = server.current_model();
3161
3162	assert_eq!(result["provider"], expected_provider);
3163	assert_eq!(result["model"], "deepseek-reasoner");
3164	}
3165
3166	#[test]
3167	fn select_model_updates_active_selection() {
3168	let mut server = AcpServer::new(
3169	Config::default(),
3170	"deepseek-chat".into(),
3171	PathBuf::from("/tmp"),
3172	);
3173
3174	let result = server
3175	.select_model(json!({ "provider": "openai", "model": "gpt-4o" }))
3176	.expect("select model");
3177
3178	assert_eq!(result["provider"], "openai");
3179	assert_eq!(result["model"], "gpt-4o");
3180	assert_eq!(server.current_model()["provider"], "openai");
3181	assert_eq!(server.current_model()["model"], "gpt-4o");
3182	}
3183
3184	#[test]
3185	fn select_model_rejects_unknown_provider() {
3186	let mut server = AcpServer::new(
3187	Config::default(),
3188	"deepseek-chat".into(),
3189	PathBuf::from("/tmp"),
3190	);
3191	let before = server.current_model();
3192
3193	let err = server
3194	.select_model(json!({ "provider": "unknown-provider", "model": "gpt-4o" }))
3195	.expect_err("unknown provider rejected");
3196
3197	assert_eq!(err.code, -32602);
3198	assert_eq!(server.current_model(), before);
3199	}
3200
3201	#[test]
3202	fn select_model_rejects_missing_model() {
3203	let mut server = AcpServer::new(
3204	Config::default(),
3205	"deepseek-chat".into(),
3206	PathBuf::from("/tmp"),
3207	);
3208
3209	let err = server
3210	.select_model(json!({ "provider": "openai" }))
3211	.expect_err("missing model rejected");
3212
3213	assert_eq!(err.code, -32602);
3214	}
3215
3216	#[test]
3217	fn extract_prompt_text_accepts_text_and_resource_blocks() {
3218	let prompt = json!([
3219	{ "type": "text", "text": "Review this file" },
3220	{
3221	"type": "resource",
3222	"resource": {
3223	"uri": "file:///tmp/app.rs",
3224	"mimeType": "text/rust",
3225	"text": "fn main() {}"
3226	}
3227	},
3228	{ "type": "resource_link", "uri": "file:///tmp/lib.rs" }
3229	]);
3230
3231	let text = extract_prompt_text(Some(&prompt)).expect("prompt text");
3232
3233	assert!(text.contains("Review this file"));
3234	assert!(text.contains("fn main() {}"));
3235	assert!(text.contains("@file:///tmp/lib.rs"));
3236	}
3237
3238	#[tokio::test]
3239	async fn session_update_is_protocol_clean_single_line_json() {
3240	let mut out = Vec::new();
3241
3242	write_session_update(&mut out, "sess_1", "hello\nworld".to_string())
3243	.await
3244	.expect("write update");
3245
3246	let line = String::from_utf8(out).expect("utf8");
3247	assert_eq!(line.lines().count(), 1);
3248	let value: Value = serde_json::from_str(line.trim()).expect("json");
3249	assert_eq!(value["method"], "session/update");
3250	assert_eq!(value["params"]["sessionId"], "sess_1");
3251	assert_eq!(value["params"]["update"]["content"]["text"], "hello\nworld");
3252	}
3253
3254	#[tokio::test]
3255	async fn jsonrpc_result_preserves_numeric_ids_for_avante_acp() {
3256	let mut out = Vec::new();
3257
3258	let params = json!({
3259	"protocolVersion": 1,
3260	"clientCapabilities": {}
3261	});
3262	let id = JsonRpcResponseIdPolicy::from_initialize_params(&params).response_id(json!(1));
3263	write_jsonrpc_result(&mut out, id, json!({"ok": true}))
3264	.await
3265	.expect("write result");
3266
3267	let line = String::from_utf8(out).expect("utf8");
3268	let value: Value = serde_json::from_str(line.trim()).expect("json");
3269	// Numeric ID must stay numeric — avante.nvim's Lua client uses
3270	// strict table keys (callbacks[1] ≠ callbacks["1"]).
3271	assert!(
3272	value["id"].is_number(),
3273	"numeric id must stay numeric, got {:?}",
3274	value["id"]
3275	);
3276	assert_eq!(value["result"], json!({"ok": true}));
3277	}
3278
3279	#[tokio::test]
3280	async fn jsonrpc_result_stringifies_numeric_ids_for_zed_acp() {
3281	let mut out = Vec::new();
3282
3283	let params = json!({
3284	"protocolVersion": 1,
3285	"clientCapabilities": {},
3286	"clientInfo": {
3287	"name": "zed",
3288	"version": "1.2.6"
3289	}
3290	});
3291	let id = JsonRpcResponseIdPolicy::from_initialize_params(&params).response_id(json!(1));
3292	write_jsonrpc_result(&mut out, id, json!({"ok": true}))
3293	.await
3294	.expect("write result");
3295
3296	let line = String::from_utf8(out).expect("utf8");
3297	let value: Value = serde_json::from_str(line.trim()).expect("json");
3298	assert_eq!(value["id"], "1");
3299	assert_eq!(value["result"], json!({"ok": true}));
3300	}
3301
3302	#[tokio::test]
3303	async fn jsonrpc_error_keeps_absent_id_null() {
3304	let mut out = Vec::new();
3305
3306	write_jsonrpc_error(&mut out, None, -32700, "invalid json")
3307	.await
3308	.expect("write error");
3309
3310	let line = String::from_utf8(out).expect("utf8");
3311	let value: Value = serde_json::from_str(line.trim()).expect("json");
3312	assert_eq!(value["id"], Value::Null);
3313	assert_eq!(value["error"]["code"], -32700);
3314	}
3315
3316	#[test]
3317	fn new_session_starts_with_empty_messages() {
3318	let mut server = AcpServer::new(
3319	Config::default(),
3320	"test-model".to_string(),
3321	PathBuf::from("/tmp"),
3322	);
3323	let result = server
3324	.new_session(json!({ "cwd": "/tmp" }))
3325	.expect("new session");
3326	let session_id = result["sessionId"].as_str().expect("session id");
3327	let session = server.sessions.get(session_id).expect("session exists");
3328	assert!(session.messages.is_empty());
3329	}
3330
3331	/// #6174: an ACP client has no id for a session it just created other than
3332	/// the one `session/new` returned, so that id must be loadable. It used to
3333	/// come back `codewhale-<uuid>` — a namespace `session/load` did not
3334	/// understand and the durable store never held — and replaying it, which is
3335	/// the normal client behaviour, failed with `-32602`.
3336	#[test]
3337	fn session_new_returns_an_id_that_session_load_resolves() {
3338	let mut server = AcpServer::new(
3339	Config::default(),
3340	"test-model".to_string(),
3341	PathBuf::from("/tmp"),
3342	);
3343	let created = server
3344	.new_session(json!({ "cwd": "/tmp" }))
3345	.expect("new session");
3346	let session_id = created["sessionId"]
3347	.as_str()
3348	.expect("session id")
3349	.to_string();
3350
3351	// The id is in the one namespace every method understands: the bare
3352	// uuid shape `session/list` advertises for durable sessions.
3353	assert!(
3354	!session_id.starts_with("codewhale-"),
3355	"session/new must not mint a prefixed id, got {session_id}"
3356	);
3357	uuid::Uuid::parse_str(&session_id)
3358	.unwrap_or_else(\|e\| panic!("session/new must mint a bare uuid, got {session_id}: {e}"));
3359
3360	// Replaying that exact id resolves, and resolves to the same session.
3361	let loaded = server
3362	.load_session(json!({ "sessionId": session_id }))
3363	.expect("session/load must resolve an id session/new returned");
3364	assert_eq!(loaded["sessionId"].as_str(), Some(session_id.as_str()));
3365	}
3366
3367	/// The memory hit must not paper over a genuinely unknown id: that still
3368	/// has to reach the durable store and fail there.
3369	#[test]
3370	fn session_load_still_rejects_an_id_no_one_minted() {
3371	let mut server = AcpServer::new(
3372	Config::default(),
3373	"test-model".to_string(),
3374	PathBuf::from("/tmp"),
3375	);
3376	let unknown = uuid::Uuid::new_v4().to_string();
3377	assert!(
3378	server
3379	.load_session(json!({ "sessionId": unknown }))
3380	.is_err(),
3381	"an id from no namespace must not resolve"
3382	);
3383	}
3384
3385	#[test]
3386	fn prompt_appends_user_and_assistant_messages_to_history() {
3387	let mut server = AcpServer::new(
3388	Config::default(),
3389	"test-model".to_string(),
3390	PathBuf::from("/tmp"),
3391	);
3392	let result = server
3393	.new_session(json!({ "cwd": "/tmp" }))
3394	.expect("new session");
3395	let session_id = result["sessionId"].as_str().unwrap().to_string();
3396
3397	// Simulate adding a user message (same logic as prompt() but without LLM call)
3398	{
3399	let session = server.sessions.get_mut(&session_id).unwrap();
3400	session.messages.push(Message {
3401	role: Role::User,
3402	content: vec![ContentBlock::Text {
3403	text: "1+1".to_string(),
3404	cache_control: None,
3405	}],
3406	});
3407	}
3408
3409	// Simulate assistant response
3410	{
3411	let session = server.sessions.get_mut(&session_id).unwrap();
3412	session.messages.push(Message {
3413	role: Role::Assistant,
3414	content: vec![ContentBlock::Text {
3415	text: "2".to_string(),
3416	cache_control: None,
3417	}],
3418	});
3419	}
3420
3421	// Second user message
3422	{
3423	let session = server.sessions.get_mut(&session_id).unwrap();
3424	session.messages.push(Message {
3425	role: Role::User,
3426	content: vec![ContentBlock::Text {
3427	text: "add one more".to_string(),
3428	cache_control: None,
3429	}],
3430	});
3431	}
3432
3433	// Verify full conversation history
3434	let session = server.sessions.get(&session_id).unwrap();
3435	assert_eq!(session.messages.len(), 3);
3436	assert_eq!(session.messages[0].role, "user");
3437	assert_eq!(session.messages[1].role, "assistant");
3438	assert_eq!(session.messages[2].role, "user");
3439
3440	// Verify text content
3441	assert_eq!(
3442	match &session.messages[0].content[0] {
3443	ContentBlock::Text { text, .. } => text.clone(),
3444	_ => String::new(),
3445	},
3446	"1+1"
3447	);
3448	assert_eq!(
3449	match &session.messages[1].content[0] {
3450	ContentBlock::Text { text, .. } => text.clone(),
3451	_ => String::new(),
3452	},
3453	"2"
3454	);
3455	assert_eq!(
3456	match &session.messages[2].content[0] {
3457	ContentBlock::Text { text, .. } => text.clone(),
3458	_ => String::new(),
3459	},
3460	"add one more"
3461	);
3462	}
3463
3464	fn lines_from(input: &'static str) -> Lines<BufReader<&'static [u8]>> {
3465	BufReader::new(input.as_bytes()).lines()
3466	}
3467
3468	fn text_delta(text: &str) -> StreamEvent {
3469	StreamEvent::ContentBlockDelta {
3470	index: 0,
3471	delta: Delta::TextDelta {
3472	text: text.to_string(),
3473	},
3474	}
3475	}
3476
3477	/// Simulate one streamed `tool_use` content block at `index`: a start
3478	/// event carrying the id/name, an `input_json_delta` with the full
3479	/// arguments JSON, and the closing stop event — matching the real
3480	/// provider's per-index streaming shape closely enough to exercise
3481	/// [`drive_prompt_stream`]'s accumulator.
3482	fn tool_use_events(index: u32, id: &str, name: &str, input_json: &str) -> Vec<StreamEvent> {
3483	vec![
3484	StreamEvent::ContentBlockStart {
3485	index,
3486	content_block: ContentBlockStart::ToolUse {
3487	id: id.to_string(),
3488	name: name.to_string(),
3489	input: json!({}),
3490	caller: None,
3491	thought_signature: None,
3492	},
3493	},
3494	StreamEvent::ContentBlockDelta {
3495	index,
3496	delta: Delta::InputJsonDelta {
3497	partial_json: input_json.to_string(),
3498	},
3499	},
3500	StreamEvent::ContentBlockStop { index },
3501	]
3502	}
3503
3504	/// A stream that yields the given events immediately, then ends.
3505	fn ready_stream(events: Vec<StreamEvent>) -> StreamEventBox {
3506	Box::pin(futures_util::stream::iter(
3507	events.into_iter().map(Ok::<_, anyhow::Error>),
3508	))
3509	}
3510
3511	fn error_stream(message: &'static str) -> StreamEventBox {
3512	Box::pin(futures_util::stream::iter(vec![Err(anyhow!(message))]))
3513	}
3514
3515	/// A stream that never yields, so a concurrent cancel always wins.
3516	fn pending_stream() -> StreamEventBox {
3517	Box::pin(futures_util::stream::pending::<Result<StreamEvent>>())
3518	}
3519
3520	/// A stream that yields `events` immediately, then emits `message_stop`
3521	/// after a short delay — long enough that an already-buffered reader line is
3522	/// processed first, making the ordering deterministic in tests.
3523	fn events_then_delayed_stop(events: Vec<StreamEvent>) -> StreamEventBox {
3524	let head = futures_util::stream::iter(events.into_iter().map(Ok::<_, anyhow::Error>));
3525	let tail = futures_util::stream::once(async {
3526	tokio::time::sleep(std::time::Duration::from_millis(20)).await;
3527	Ok(StreamEvent::MessageStop)
3528	});
3529	Box::pin(head.chain(tail))
3530	}
3531
3532	fn parse_lines(out: Vec<u8>) -> Vec<Value> {
3533	String::from_utf8(out)
3534	.expect("utf8")
3535	.lines()
3536	.filter(\|line\| !line.trim().is_empty())
3537	.map(\|line\| serde_json::from_str(line).expect("json"))
3538	.collect()
3539	}
3540
3541	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3542	enum PermissionClientScript {
3543	Allow,
3544	Reject,
3545	WrongIdThenReject,
3546	CancelThenLateAllow,
3547	AllowThenCancelRunning,
3548	}
3549
3550	async fn write_client_message(writer: &mut tokio::io::DuplexStream, message: Value) {
3551	writer
3552	.write_all(format!("{message}\n").as_bytes())
3553	.await
3554	.expect("write simulated ACP client message");
3555	}
3556
3557	async fn drive_permission_client(
3558	output: tokio::io::DuplexStream,
3559	mut input: tokio::io::DuplexStream,
3560	script: PermissionClientScript,
3561	must_not_exist_before_response: Option<PathBuf>,
3562	) -> Vec<Value> {
3563	let mut output = BufReader::new(output).lines();
3564	let mut seen = Vec::new();
3565	let mut sent_running_cancel = false;
3566	while let Some(line) = output.next_line().await.expect("read agent output") {
3567	let message: Value = serde_json::from_str(&line).expect("agent output json");
3568	seen.push(message.clone());
3569
3570	if message.get("method").and_then(Value::as_str) == Some("session/request_permission") {
3571	if let Some(path) = must_not_exist_before_response.as_ref() {
3572	assert!(
3573	!path.exists(),
3574	"sensitive tool ran before the permission response: {}",
3575	path.display()
3576	);
3577	}
3578	let request_id = message["id"].clone();
3579	let selected = \|option_id: &str\| {
3580	json!({
3581	"jsonrpc": "2.0",
3582	"id": request_id.clone(),
3583	"result": {
3584	"outcome": {
3585	"outcome": "selected",
3586	"optionId": option_id
3587	}
3588	}
3589	})
3590	};
3591	match script {
3592	PermissionClientScript::Allow
3593	\| PermissionClientScript::AllowThenCancelRunning => {
3594	write_client_message(&mut input, selected("allow-once")).await;
3595	}
3596	PermissionClientScript::Reject => {
3597	write_client_message(&mut input, selected("reject-once")).await;
3598	}
3599	PermissionClientScript::WrongIdThenReject => {
3600	write_client_message(
3601	&mut input,
3602	json!({
3603	"jsonrpc": "2.0",
3604	"id": "wrong-agent-request-id",
3605	"result": {
3606	"outcome": {
3607	"outcome": "selected",
3608	"optionId": "allow-once"
3609	}
3610	}
3611	}),
3612	)
3613	.await;
3614	write_client_message(&mut input, selected("reject-once")).await;
3615	}
3616	PermissionClientScript::CancelThenLateAllow => {
3617	write_client_message(
3618	&mut input,
3619	json!({
3620	"jsonrpc": "2.0",
3621	"method": "session/cancel",
3622	"params": { "sessionId": "sess_1" }
3623	}),
3624	)
3625	.await;
3626	write_client_message(
3627	&mut input,
3628	json!({
3629	"jsonrpc": "2.0",
3630	"id": request_id.clone(),
3631	"result": { "outcome": { "outcome": "cancelled" } }
3632	}),
3633	)
3634	.await;
3635	write_client_message(&mut input, selected("allow-once")).await;
3636	}
3637	}
3638	}
3639
3640	let update_status = message
3641	.pointer("/params/update/status")
3642	.and_then(Value::as_str);
3643	if script == PermissionClientScript::AllowThenCancelRunning
3644	&& update_status == Some("in_progress")
3645	&& !sent_running_cancel
3646	{
3647	sent_running_cancel = true;
3648	tokio::time::sleep(Duration::from_millis(100)).await;
3649	write_client_message(
3650	&mut input,
3651	json!({
3652	"jsonrpc": "2.0",
3653	"id": 7,
3654	"method": "session/cancel",
3655	"params": { "sessionId": "sess_1" }
3656	}),
3657	)
3658	.await;
3659	}
3660	if matches!(update_status, Some("completed" \| "failed")) {
3661	break;
3662	}
3663	}
3664	seen
3665	}
3666
3667	async fn execute_one_with_permission_client(
3668	config: &Config,
3669	registry: &ToolRegistry,
3670	call: PendingToolCall,
3671	script: PermissionClientScript,
3672	response_id_policy: JsonRpcResponseIdPolicy,
3673	must_not_exist_before_response: Option<PathBuf>,
3674	) -> (ToolBatchOutcome, Vec<Value>, Option<Value>) {
3675	let (client_input, agent_input) = tokio::io::duplex(64 * 1024);
3676	let (agent_output, client_output) = tokio::io::duplex(64 * 1024);
3677	let client = tokio::spawn(drive_permission_client(
3678	client_output,
3679	client_input,
3680	script,
3681	must_not_exist_before_response,
3682	));
3683	let mut reader = BufReader::new(agent_input).lines();
3684	let mut writer = agent_output;
3685
3686	let outcome = execute_tool_calls_with_cancellation(
3687	AcpTurnContext {
3688	config,
3689	model: "test-model",
3690	session_id: "sess_1",
3691	tool_registry: registry,
3692	response_id_policy,
3693	},
3694	vec![call],
3695	&mut reader,
3696	&mut writer,
3697	)
3698	.await
3699	.expect("execute ACP tool batch");
3700	let late_response = if script == PermissionClientScript::CancelThenLateAllow {
3701	let line = tokio::time::timeout(Duration::from_secs(1), reader.next_line())
3702	.await
3703	.expect("late permission response arrived")
3704	.expect("read late permission response")
3705	.expect("late permission response line");
3706	Some(serde_json::from_str(&line).expect("late response json"))
3707	} else {
3708	None
3709	};
3710	drop(writer);
3711	let seen = client.await.expect("simulated ACP client joins");
3712	(outcome, seen, late_response)
3713	}
3714
3715	#[tokio::test]
3716	async fn drive_prompt_streams_each_delta_as_a_chunk_then_completes() {
3717	let stream = ready_stream(vec![
3718	text_delta("hello"),
3719	text_delta(" world"),
3720	StreamEvent::MessageStop,
3721	]);
3722	let mut reader = lines_from("");
3723	let mut out = Vec::new();
3724
3725	let (outcome, tool_calls) = drive_prompt_stream(
3726	stream,
3727	"sess_1",
3728	JsonRpcResponseIdPolicy::Preserve,
3729	&mut reader,
3730	&mut out,
3731	)
3732	.await
3733	.expect("driver ok");
3734
3735	// Full text is accumulated for history...
3736	assert_eq!(outcome, PromptOutcome::Completed("hello world".to_string()));
3737	assert!(tool_calls.is_empty());
3738	// ...and each delta was emitted as its own session/update chunk.
3739	let updates = parse_lines(out);
3740	assert_eq!(updates.len(), 2);
3741	assert!(updates.iter().all(\|u\| u["method"] == "session/update"));
3742	assert_eq!(updates[0]["params"]["update"]["content"]["text"], "hello");
3743	assert_eq!(updates[1]["params"]["update"]["content"]["text"], " world");
3744	}
3745
3746	#[tokio::test]
3747	async fn drive_prompt_cancels_when_matching_cancel_arrives() {
3748	// A provider stream that never finishes within the test.
3749	let stream = pending_stream();
3750	let mut reader = lines_from(
3751	r#"{"jsonrpc":"2.0","method":"session/cancel","params":{"sessionId":"sess_1"}}"#,
3752	);
3753	let mut out = Vec::new();
3754
3755	let (outcome, tool_calls) = drive_prompt_stream(
3756	stream,
3757	"sess_1",
3758	JsonRpcResponseIdPolicy::Preserve,
3759	&mut reader,
3760	&mut out,
3761	)
3762	.await
3763	.expect("driver ok");
3764
3765	assert_eq!(outcome, PromptOutcome::Cancelled);
3766	assert!(tool_calls.is_empty());
3767	// Notification-form cancel (no id) is acknowledged by acting, not writing.
3768	assert!(out.is_empty());
3769	}
3770
3771	#[tokio::test]
3772	async fn drive_prompt_ignores_cancel_for_a_different_session() {
3773	// The unrelated cancel line is buffered and ready; the delayed stop makes
3774	// it process first, proving it does not abort the turn.
3775	let stream = events_then_delayed_stop(vec![text_delta("kept")]);
3776	let mut reader = lines_from(
3777	r#"{"jsonrpc":"2.0","id":7,"method":"session/cancel","params":{"sessionId":"other"}}"#,
3778	);
3779	let mut out = Vec::new();
3780
3781	let (outcome, _tool_calls) = drive_prompt_stream(
3782	stream,
3783	"sess_1",
3784	JsonRpcResponseIdPolicy::StringifyNumeric,
3785	&mut reader,
3786	&mut out,
3787	)
3788	.await
3789	.expect("driver ok");
3790
3791	assert_eq!(outcome, PromptOutcome::Completed("kept".to_string()));
3792	// The other-session cancel carried an id, so it was acknowledged with null.
3793	let lines = parse_lines(out);
3794	assert!(
3795	lines
3796	.iter()
3797	.any(\|v\| v["id"] == "7" && v["result"] == Value::Null),
3798	"expected a null ack for the other-session cancel, got {lines:?}"
3799	);
3800	}
3801
3802	#[tokio::test]
3803	async fn drive_prompt_rejects_a_concurrent_request_but_keeps_running() {
3804	let stream = events_then_delayed_stop(vec![text_delta("done")]);
3805	// A non-cancel request arrives mid-turn.
3806	let mut reader =
3807	lines_from(r#"{"jsonrpc":"2.0","id":9,"method":"session/new","params":{}}"#);
3808	let mut out = Vec::new();
3809
3810	let (outcome, _tool_calls) = drive_prompt_stream(
3811	stream,
3812	"sess_1",
3813	JsonRpcResponseIdPolicy::StringifyNumeric,
3814	&mut reader,
3815	&mut out,
3816	)
3817	.await
3818	.expect("driver ok");
3819
3820	assert_eq!(outcome, PromptOutcome::Completed("done".to_string()));
3821	let lines = parse_lines(out);
3822	assert!(
3823	lines
3824	.iter()
3825	.any(\|v\| v["id"] == "9" && v["error"]["code"] == -32603),
3826	"expected a prompt-in-progress error for the concurrent request, got {lines:?}"
3827	);
3828	}
3829
3830	#[tokio::test]
3831	async fn drive_prompt_assembles_a_single_streamed_tool_call() {
3832	let mut events = tool_use_events(0, "call_1", "read_file", r#"{"path":"src/lib.rs"}"#);
3833	events.push(StreamEvent::MessageStop);
3834	let stream = ready_stream(events);
3835	let mut reader = lines_from("");
3836	let mut out = Vec::new();
3837
3838	let (outcome, tool_calls) = drive_prompt_stream(
3839	stream,
3840	"sess_1",
3841	JsonRpcResponseIdPolicy::Preserve,
3842	&mut reader,
3843	&mut out,
3844	)
3845	.await
3846	.expect("driver ok");
3847
3848	assert_eq!(outcome, PromptOutcome::Completed(String::new()));
3849	assert_eq!(tool_calls.len(), 1);
3850	assert_eq!(tool_calls[0].id, "call_1");
3851	assert_eq!(tool_calls[0].name, "read_file");
3852	assert_eq!(tool_calls[0].input, json!({"path": "src/lib.rs"}));
3853	assert!(tool_calls[0].parse_error.is_none());
3854	}
3855
3856	#[tokio::test]
3857	async fn drive_prompt_assembles_multiple_parallel_tool_calls_in_order() {
3858	let mut events = tool_use_events(0, "call_1", "read_file", r#"{"path":"a.rs"}"#);
3859	events.extend(tool_use_events(
3860	1,
3861	"call_2",
3862	"read_file",
3863	r#"{"path":"b.rs"}"#,
3864	));
3865	events.push(StreamEvent::MessageStop);
3866	let stream = ready_stream(events);
3867	let mut reader = lines_from("");
3868	let mut out = Vec::new();
3869
3870	let (_outcome, tool_calls) = drive_prompt_stream(
3871	stream,
3872	"sess_1",
3873	JsonRpcResponseIdPolicy::Preserve,
3874	&mut reader,
3875	&mut out,
3876	)
3877	.await
3878	.expect("driver ok");
3879
3880	assert_eq!(tool_calls.len(), 2);
3881	assert_eq!(tool_calls[0].id, "call_1");
3882	assert_eq!(tool_calls[1].id, "call_2");
3883	}
3884
3885	#[tokio::test]
3886	async fn drive_prompt_reports_malformed_tool_arguments_instead_of_dropping_the_call() {
3887	let mut events = tool_use_events(0, "call_1", "read_file", "{not json");
3888	events.push(StreamEvent::MessageStop);
3889	let stream = ready_stream(events);
3890	let mut reader = lines_from("");
3891	let mut out = Vec::new();
3892
3893	let (_outcome, tool_calls) = drive_prompt_stream(
3894	stream,
3895	"sess_1",
3896	JsonRpcResponseIdPolicy::Preserve,
3897	&mut reader,
3898	&mut out,
3899	)
3900	.await
3901	.expect("driver ok");
3902
3903	assert_eq!(tool_calls.len(), 1);
3904	assert!(tool_calls[0].parse_error.is_some());
3905	}
3906
3907	#[test]
3908	fn different_sessions_have_independent_history() {
3909	let mut server = AcpServer::new(
3910	Config::default(),
3911	"test-model".to_string(),
3912	PathBuf::from("/tmp"),
3913	);
3914	let result1 = server
3915	.new_session(json!({ "cwd": "/tmp" }))
3916	.expect("session 1");
3917	let result2 = server
3918	.new_session(json!({ "cwd": "/tmp" }))
3919	.expect("session 2");
3920	let sid1 = result1["sessionId"].as_str().unwrap().to_string();
3921	let sid2 = result2["sessionId"].as_str().unwrap().to_string();
3922
3923	// Add messages to session 1
3924	{
3925	let session = server.sessions.get_mut(&sid1).unwrap();
3926	session.messages.push(Message {
3927	role: Role::User,
3928	content: vec![ContentBlock::Text {
3929	text: "hello".to_string(),
3930	cache_control: None,
3931	}],
3932	});
3933	}
3934
3935	// Session 2 should remain empty
3936	let session2 = server.sessions.get(&sid2).unwrap();
3937	assert!(session2.messages.is_empty());
3938
3939	// Session 1 should have the message
3940	let session1 = server.sessions.get(&sid1).unwrap();
3941	assert_eq!(session1.messages.len(), 1);
3942	}
3943
3944	#[test]
3945	fn concurrent_sessions_each_get_their_own_tool_registry() {
3946	let mut server = AcpServer::new(
3947	Config {
3948	allow_shell: Some(true),
3949	..Config::default()
3950	},
3951	"test-model".to_string(),
3952	PathBuf::from("/tmp"),
3953	);
3954	// Both config opt-in and the client terminal capability are present.
3955	server.client_supports_terminal = true;
3956	let s1 = server.new_session(json!({ "cwd": "/tmp" })).unwrap();
3957	let s2 = server.new_session(json!({ "cwd": "/tmp" })).unwrap();
3958	let id1 = s1["sessionId"].as_str().unwrap();
3959	let id2 = s2["sessionId"].as_str().unwrap();
3960
3961	let reg1 = server.session_tool_registry(id1).expect("registry 1");
3962	let reg2 = server.session_tool_registry(id2).expect("registry 2");
3963	assert!(!Arc::ptr_eq(&reg1, &reg2));
3964	// Both sessions expose the same reusable tool surface: the
3965	// canonical `File` action tool (read/list/search/write/edit),
3966	// `Git`, the `apply_patch` back-compat alias, and `Bash` (#4625
3967	// consolidated the old per-action tool names).
3968	assert!(reg1.contains("File"));
3969	assert!(reg1.contains("Git"));
3970	assert!(reg1.contains("apply_patch"));
3971	assert!(reg1.contains("bash"));
3972	assert!(reg1.contains("Bash"));
3973	assert!(
3974	reg1.names()
3975	.into_iter()
3976	.all(\|name\| !name.starts_with("terminal/")),
3977	"ACP must not expose stateful terminal tools"
3978	);
3979	assert!(reg1.context().runtime.hook_executor.is_some());
3980	}
3981
3982	#[test]
3983	fn shell_tool_omitted_when_client_declares_no_terminal_support() {
3984	let workspace = std::env::temp_dir();
3985	let config = Config {
3986	allow_shell: Some(true),
3987	..Config::default()
3988	};
3989	let registry = build_acp_tool_registry(&config, &workspace, false);
3990	assert!(!registry.contains("Bash"));
3991	assert!(registry.contains("File"));
3992	}
3993
3994	#[test]
3995	fn shell_tool_omitted_without_headless_config_opt_in() {
3996	let workspace = std::env::temp_dir();
3997	let registry = build_acp_tool_registry(&Config::default(), &workspace, true);
3998	assert!(!registry.contains("Bash"));
3999	assert_eq!(registry.context().shell_policy, ShellPolicy::None);
4000	assert!(!registry.context().auto_approve);
4001	}
4002
4003	#[test]
4004	fn acp_shell_uses_configured_external_sandbox_or_fails_closed() {
4005	let workspace = std::env::temp_dir();
4006	let configured = Config {
4007	allow_shell: Some(true),
4008	sandbox_backend: Some("opensandbox".to_string()),
4009	sandbox_url: Some("http://127.0.0.1:8080".to_string()),
4010	..Config::default()
4011	};
4012	let registry = build_acp_tool_registry(&configured, &workspace, true);
4013	assert!(registry.contains("bash"));
4014	assert!(registry.context().sandbox_backend.is_some());
4015
4016	let unsupported = Config {
4017	allow_shell: Some(true),
4018	sandbox_backend: Some("unsupported-backend".to_string()),
4019	..Config::default()
4020	};
4021	let registry = build_acp_tool_registry(&unsupported, &workspace, true);
4022	assert!(!registry.contains("bash"));
4023	assert!(!registry.contains("Bash"));
4024	assert!(registry.context().sandbox_backend.is_none());
4025	}
4026
4027	#[test]
4028	fn acp_tool_override_removes_every_builtin_compatibility_alias() {
4029	let mut overrides = std::collections::HashMap::new();
4030	overrides.insert("Bash".to_string(), crate::config::ToolOverride::Disabled);
4031	let config = Config {
4032	allow_shell: Some(true),
4033	tools: Some(crate::config::ToolsConfig {
4034	overrides: Some(overrides),
4035	..crate::config::ToolsConfig::default()
4036	}),
4037	..Config::default()
4038	};
4039	let registry = build_acp_tool_registry(&config, &std::env::temp_dir(), true);
4040	assert!(!registry.contains("bash"));
4041	assert!(!registry.contains("Bash"));
4042	assert!(
4043	registry
4044	.names()
4045	.into_iter()
4046	.all(\|name\| !name.starts_with("terminal/"))
4047	);
4048	}
4049
4050	#[test]
4051	fn acp_prompt_uses_the_stable_headless_composer() {
4052	let workspace = tempfile::tempdir().expect("workspace");
4053	std::fs::write(
4054	workspace.path().join("AGENTS.md"),
4055	"# ACP project law\n\nKeep the acp-project-marker visible.",
4056	)
4057	.expect("write project instructions");
4058	let extra = workspace.path().join("maintainer-instructions.md");
4059	std::fs::write(&extra, "Keep the acp-config-marker visible.")
4060	.expect("write configured instructions");
4061	let config = Config {
4062	instructions: Some(vec![extra.to_string_lossy().into_owned()]),
4063	..Config::default()
4064	};
4065
4066	let prompt = build_acp_system_prompt(
4067	&config,
4068	workspace.path(),
4069	ApiProvider::Deepseek,
4070	"deepseek-v4-pro",
4071	None,
4072	);
4073	let text = crate::prompts::system_prompt_flat_text(&prompt);
4074
4075	assert!(text.contains(crate::prompts::text::BASE_PROMPT.trim()));
4076	assert!(text.contains("acp-project-marker"));
4077	assert!(text.contains("acp-config-marker"));
4078	assert!(!text.contains("You are a coding assistant inside an ACP-compatible editor."));
4079	}
4080
4081	#[test]
4082	fn acp_system_prompt_is_byte_stable_after_round_one_agents_write() {
4083	let workspace = tempfile::tempdir().expect("workspace");
4084	let agents = workspace.path().join("AGENTS.md");
4085	std::fs::write(&agents, "round-one-authority").expect("write initial AGENTS");
4086	let config = Config::default();
4087	let slot = std::sync::Mutex::new(None);
4088
4089	let round_one = frozen_acp_system_prompt(
4090	&slot,
4091	&config,
4092	workspace.path(),
4093	ApiProvider::Deepseek,
4094	"deepseek-v4-pro",
4095	None,
4096	);
4097	// Simulate a model tool changing project instructions during round 1.
4098	std::fs::write(&agents, "round-two-self-authored-authority")
4099	.expect("mutate AGENTS between rounds");
4100	let round_two = frozen_acp_system_prompt(
4101	&slot,
4102	&config,
4103	workspace.path(),
4104	ApiProvider::Deepseek,
4105	"deepseek-v4-pro",
4106	None,
4107	);
4108
4109	assert_eq!(
4110	serde_json::to_vec(&round_one).unwrap(),
4111	serde_json::to_vec(&round_two).unwrap(),
4112	"later rounds must receive the byte-identical first-round system prompt"
4113	);
4114	let freshly_composed = build_acp_system_prompt(
4115	&config,
4116	workspace.path(),
4117	ApiProvider::Deepseek,
4118	"deepseek-v4-pro",
4119	None,
4120	);
4121	assert!(
4122	crate::prompts::system_prompt_flat_text(&freshly_composed)
4123	.contains("round-two-self-authored-authority"),
4124	"fixture must prove the mutable source really changed"
4125	);
4126	assert!(
4127	!crate::prompts::system_prompt_flat_text(&round_two)
4128	.contains("round-two-self-authored-authority")
4129	);
4130	}
4131
4132	fn workspace_registry() -> (tempfile::TempDir, ToolRegistry) {
4133	let dir = tempfile::tempdir().expect("tempdir");
4134	let config = Config {
4135	allow_shell: Some(true),
4136	..Config::default()
4137	};
4138	let registry = build_acp_tool_registry(&config, dir.path(), true);
4139	(dir, registry)
4140	}
4141
4142	fn pending_call(name: &str, input: Value) -> PendingToolCall {
4143	PendingToolCall {
4144	id: "call_1".to_string(),
4145	name: name.to_string(),
4146	input,
4147	parse_error: None,
4148	}
4149	}
4150
4151	fn config_with_policy_rule(rule: codewhale_execpolicy::ToolAskRule) -> Config {
4152	Config {
4153	exec_policy_engine: codewhale_execpolicy::ExecPolicyEngine::with_rulesets(vec![
4154	codewhale_execpolicy::Ruleset::user(vec![], vec![]).with_ask_rules(vec![rule]),
4155	]),
4156	..Config::default()
4157	}
4158	}
4159
4160	fn tool_call_hook_command(payload: &Value) -> String {
4161	let payload = payload.to_string();
4162	if cfg!(windows) {
4163	format!("echo {payload}")
4164	} else {
4165	format!("printf '%s\\n' '{payload}'")
4166	}
4167	}
4168
4169	fn config_with_tool_call_hook(mut config: Config, payload: Value, strict: bool) -> Config {
4170	let mut hook = crate::hooks::Hook::new(
4171	crate::hooks::HookEvent::ToolCallBefore,
4172	&tool_call_hook_command(&payload),
4173	);
4174	hook.continue_on_error = !strict;
4175	config.hooks = Some(crate::hooks::HooksConfig {
4176	enabled: true,
4177	hooks: vec![hook],
4178	..crate::hooks::HooksConfig::default()
4179	});
4180	config
4181	}
4182
4183	#[tokio::test]
4184	async fn acp_strict_tool_call_before_deny_blocks_before_execution() {
4185	let dir = tempfile::tempdir().expect("tempdir");
4186	let config = config_with_tool_call_hook(
4187	Config::default(),
4188	json!({"decision": "deny", "reason": "release gate"}),
4189	true,
4190	);
4191	let registry = build_acp_tool_registry(&config, dir.path(), false);
4192	let error = prepare_acp_tool_with_hooks(
4193	&config,
4194	"test-model",
4195	&registry,
4196	&pending_call("File", json!({"action": "read", "path": "safe.txt"})),
4197	)
4198	.await
4199	.expect_err("strict hook must deny");
4200
4201	assert!(error.to_string().contains("release gate"));
4202	}
4203
4204	#[tokio::test]
4205	async fn acp_hook_rewrite_is_reprepared_and_policy_is_re_evaluated() {
4206	let dir = tempfile::tempdir().expect("tempdir");
4207	let deny_rewritten_write = codewhale_execpolicy::ToolAskRule {
4208	action: codewhale_execpolicy::PermissionAction::Deny,
4209	..codewhale_execpolicy::ToolAskRule::file_path("write_file", "rewritten.txt")
4210	};
4211	let config = config_with_tool_call_hook(
4212	config_with_policy_rule(deny_rewritten_write),
4213	json!({
4214	"updatedInput": {
4215	"action": "write",
4216	"path": "rewritten.txt",
4217	"content": "rewritten by hook"
4218	}
4219	}),
4220	true,
4221	);
4222	let registry = build_acp_tool_registry(&config, dir.path(), false);
4223	let raw = pending_call("File", json!({"action": "read", "path": "safe.txt"}));
4224	let (_, raw_admission) = prepare_acp_tool_admission(&config, &registry, &raw).unwrap();
4225	assert_eq!(raw_admission, AcpToolAdmission::Auto);
4226
4227	let prepared = prepare_acp_tool_with_hooks(&config, "test-model", &registry, &raw)
4228	.await
4229	.expect("hook rewrite prepares");
4230	assert_eq!(
4231	prepared.call.input.get("action").and_then(Value::as_str),
4232	Some("write")
4233	);
4234	assert!(matches!(prepared.admission, AcpToolAdmission::Block(_)));
4235	assert!(!dir.path().join("rewritten.txt").exists());
4236	}
4237
4238	#[test]
4239	fn acp_admission_is_input_specific_and_has_no_workspace_write_carve_out() {
4240	let (dir, registry) = workspace_registry();
4241	let config = Config::default();
4242	let read = pending_call("File", json!({"action": "read", "path": "src/lib.rs"}));
4243	let write = pending_call(
4244	"File",
4245	json!({"action": "write", "path": "src/lib.rs", "content": "new"}),
4246	);
4247
4248	let (_, read_admission) = prepare_acp_tool_admission(&config, &registry, &read).unwrap();
4249	let (_, write_admission) = prepare_acp_tool_admission(&config, &registry, &write).unwrap();
4250
4251	assert_eq!(read_admission, AcpToolAdmission::Auto);
4252	assert!(matches!(
4253	write_admission,
4254	AcpToolAdmission::RequestPermission(_)
4255	));
4256	assert_eq!(registry.context().workspace, dir.path());
4257	}
4258
4259	#[test]
4260	fn acp_admission_folds_typed_rules_then_headless_safety_floor() {
4261	let (dir, registry) = workspace_registry();
4262	let workspace = dir.path().to_string_lossy().into_owned();
4263	let input = json!({"action": "write", "path": "allowed.txt", "content": "new"});
4264	let call = pending_call("File", input.clone());
4265
4266	let allow = codewhale_execpolicy::ToolAskRule::file_path("write_file", "allowed.txt")
4267	.into_exact_workspace_allow(workspace.clone());
4268	let (_, admission) =
4269	prepare_acp_tool_admission(&config_with_policy_rule(allow), &registry, &call).unwrap();
4270	assert_eq!(admission, AcpToolAdmission::Auto);
4271
4272	let ask = codewhale_execpolicy::ToolAskRule::file_path("write_file", "allowed.txt");
4273	let (_, admission) =
4274	prepare_acp_tool_admission(&config_with_policy_rule(ask), &registry, &call).unwrap();
4275	assert!(matches!(
4276	admission,
4277	AcpToolAdmission::RequestPermission(reason) if reason.contains("requires approval")
4278	));
4279
4280	let deny = codewhale_execpolicy::ToolAskRule {
4281	action: codewhale_execpolicy::PermissionAction::Deny,
4282	..codewhale_execpolicy::ToolAskRule::file_path("write_file", "allowed.txt")
4283	};
4284	let (_, admission) =
4285	prepare_acp_tool_admission(&config_with_policy_rule(deny), &registry, &call).unwrap();
4286	assert!(matches!(admission, AcpToolAdmission::Block(_)));
4287
4288	let command = "rm -rf ~/";
4289	let shell_allow = codewhale_execpolicy::ToolAskRule::exec_shell(command)
4290	.into_exact_workspace_allow(workspace);
4291	let shell_call = pending_call("Bash", json!({"command": command}));
4292	let (_, admission) = prepare_acp_tool_admission(
4293	&config_with_policy_rule(shell_allow),
4294	&registry,
4295	&shell_call,
4296	)
4297	.unwrap();
4298	assert!(matches!(
4299	admission,
4300	AcpToolAdmission::RequestPermission(reason)
4301	if reason.contains("Built-in safety gate")
4302	));
4303	}
4304
4305	#[test]
4306	fn acp_admission_blocks_detached_and_stateful_bash_inputs() {
4307	let (_dir, registry) = workspace_registry();
4308	for input in [
4309	json!({"command": "sleep 30", "background": true}),
4310	json!({"command": "sleep 30", "tty": true}),
4311	json!({"command": "echo hi", "interactive": true}),
4312	json!({"command": "serve", "background": true, "persist": true}),
4313	json!({"command": "sleep 30 &"}),
4314	json!({"command": "nohup sleep 30"}),
4315	json!({"action": "wait", "task_id": "shell-1"}),
4316	json!({"action": "cancel", "task_id": "shell-1"}),
4317	] {
4318	let call = pending_call("Bash", input);
4319	let (_, admission) =
4320	prepare_acp_tool_admission(&Config::default(), &registry, &call).unwrap();
4321	assert!(matches!(
4322	admission,
4323	AcpToolAdmission::Block(reason)
4324	if reason.contains("foreground Bash runs only")
4325	));
4326	}
4327	assert!(!acp_shell_command_requests_detach("echo '&' && echo done"));
4328	}
4329
4330	#[test]
4331	fn acp_admission_auto_review_and_repo_law_override_typed_allow() {
4332	let (dir, registry) = workspace_registry();
4333	let workspace = dir.path().to_string_lossy().into_owned();
4334	let command = "cargo test";
4335	let shell_allow = codewhale_execpolicy::ToolAskRule::exec_shell(command)
4336	.into_exact_workspace_allow(workspace.clone());
4337	let mut auto_review_block = config_with_policy_rule(shell_allow);
4338	auto_review_block.auto_review = Some(crate::config::AutoReviewConfig {
4339	block: vec![crate::config::AutoReviewRuleConfig {
4340	id: Some("acp-shell-block".to_string()),
4341	action_kind: Some("shell".to_string()),
4342	reason: Some("ACP shell is disabled by policy".to_string()),
4343	..Default::default()
4344	}],
4345	..Default::default()
4346	});
4347	let (_, admission) = prepare_acp_tool_admission(
4348	&auto_review_block,
4349	&registry,
4350	&pending_call("Bash", json!({"command": command})),
4351	)
4352	.unwrap();
4353	assert!(matches!(
4354	admission,
4355	AcpToolAdmission::Block(reason) if reason.contains("ACP shell is disabled by policy")
4356	));
4357
4358	let law_dir = dir.path().join(".codewhale");
4359	std::fs::create_dir_all(&law_dir).unwrap();
4360	std::fs::write(
4361	law_dir.join("constitution.json"),
4362	r#"{
4363	"protected_invariants": [
4364	{ "text": "Never rewrite the wire", "paths": ["wire.rs"], "action": "block" },
4365	{ "text": "Review release notes", "paths": ["CHANGELOG.md"] }
4366	]
4367	}"#,
4368	)
4369	.unwrap();
4370
4371	for (path, expected_block) in [("wire.rs", true), ("CHANGELOG.md", false)] {
4372	let allow = codewhale_execpolicy::ToolAskRule::file_path("write_file", path)
4373	.into_exact_workspace_allow(workspace.clone());
4374	let config = config_with_policy_rule(allow);
4375	let call = pending_call(
4376	"File",
4377	json!({"action": "write", "path": path, "content": "new"}),
4378	);
4379	let (_, admission) = prepare_acp_tool_admission(&config, &registry, &call).unwrap();
4380	if expected_block {
4381	assert!(matches!(
4382	admission,
4383	AcpToolAdmission::Block(reason) if reason.contains("Never rewrite the wire")
4384	));
4385	} else {
4386	assert!(matches!(
4387	admission,
4388	AcpToolAdmission::RequestPermission(reason)
4389	if reason.contains("Review release notes")
4390	));
4391	}
4392	}
4393	}
4394
4395	#[tokio::test]
4396	async fn acp_read_runs_without_permission_but_reports_pending_before_in_progress() {
4397	let (dir, registry) = workspace_registry();
4398	std::fs::write(dir.path().join("read.txt"), "safe").unwrap();
4399	let mut reader = lines_from("");
4400	let mut out = Vec::new();
4401
4402	let outcome = execute_tool_calls_with_cancellation(
4403	AcpTurnContext {
4404	config: &Config::default(),
4405	model: "test-model",
4406	session_id: "sess_1",
4407	tool_registry: &registry,
4408	response_id_policy: JsonRpcResponseIdPolicy::Preserve,
4409	},
4410	vec![pending_call(
4411	"File",
4412	json!({"action": "read", "path": "read.txt"}),
4413	)],
4414	&mut reader,
4415	&mut out,
4416	)
4417	.await
4418	.unwrap();
4419
4420	assert!(matches!(outcome, ToolBatchOutcome::Completed(_)));
4421	let messages = parse_lines(out);
4422	assert!(!messages.iter().any(\|message\| {
4423	message.get("method").and_then(Value::as_str) == Some("session/request_permission")
4424	}));
4425	let statuses = messages
4426	.iter()
4427	.filter_map(\|message\| {
4428	message
4429	.pointer("/params/update/status")
4430	.and_then(Value::as_str)
4431	})
4432	.collect::<Vec<_>>();
4433	assert_eq!(statuses, vec!["pending", "in_progress", "completed"]);
4434	}
4435
4436	#[tokio::test]
4437	async fn acp_permission_allow_executes_write_once_after_response() {
4438	let (dir, registry) = workspace_registry();
4439	let target = dir.path().join("allowed.txt");
4440	let (outcome, messages, late) = execute_one_with_permission_client(
4441	&Config::default(),
4442	&registry,
4443	pending_call(
4444	"File",
4445	json!({"action": "write", "path": "allowed.txt", "content": "written once"}),
4446	),
4447	PermissionClientScript::Allow,
4448	JsonRpcResponseIdPolicy::Preserve,
4449	Some(target.clone()),
4450	)
4451	.await;
4452
4453	assert!(late.is_none());
4454	assert!(matches!(outcome, ToolBatchOutcome::Completed(_)));
4455	assert_eq!(std::fs::read_to_string(target).unwrap(), "written once");
4456	let request = messages
4457	.iter()
4458	.find(\|message\| message["method"] == "session/request_permission")
4459	.expect("permission request");
4460	assert_eq!(request["params"]["toolCall"]["status"], "pending");
4461	assert_eq!(request["params"]["options"][0]["kind"], "allow_once");
4462	assert_eq!(request["params"]["options"][1]["kind"], "reject_once");
4463	let statuses = messages
4464	.iter()
4465	.filter_map(\|message\| {
4466	message
4467	.pointer("/params/update/status")
4468	.and_then(Value::as_str)
4469	})
4470	.collect::<Vec<_>>();
4471	assert_eq!(statuses, vec!["pending", "in_progress", "completed"]);
4472	}
4473
4474	#[tokio::test]
4475	async fn acp_permission_reject_and_wrong_id_fail_closed_without_write() {
4476	for script in [
4477	PermissionClientScript::Reject,
4478	PermissionClientScript::WrongIdThenReject,
4479	] {
4480	let (dir, registry) = workspace_registry();
4481	let target = dir.path().join("denied.txt");
4482	let (outcome, messages, _) = execute_one_with_permission_client(
4483	&Config::default(),
4484	&registry,
4485	pending_call(
4486	"File",
4487	json!({"action": "write", "path": "denied.txt", "content": "forbidden"}),
4488	),
4489	script,
4490	JsonRpcResponseIdPolicy::Preserve,
4491	Some(target.clone()),
4492	)
4493	.await;
4494
4495	assert!(!target.exists(), "{script:?} must not authorize the write");
4496	let ToolBatchOutcome::Completed(results) = outcome else {
4497	panic!("rejection should complete with a failed tool result");
4498	};
4499	assert_eq!(results.len(), 1);
4500	let ContentBlock::ToolResult { is_error, .. } = &results[0].content[0] else {
4501	panic!("expected tool result");
4502	};
4503	assert_eq!(*is_error, Some(true));
4504	assert!(!messages.iter().any(\|message\| {
4505	message
4506	.pointer("/params/update/status")
4507	.and_then(Value::as_str)
4508	== Some("in_progress")
4509	}));
4510	}
4511	}
4512
4513	#[tokio::test]
4514	async fn acp_permission_cancel_ignores_late_allow_and_never_runs_tool() {
4515	let (dir, registry) = workspace_registry();
4516	let target = dir.path().join("cancelled.txt");
4517	let (outcome, messages, late) = execute_one_with_permission_client(
4518	&Config::default(),
4519	&registry,
4520	pending_call(
4521	"File",
4522	json!({"action": "write", "path": "cancelled.txt", "content": "forbidden"}),
4523	),
4524	PermissionClientScript::CancelThenLateAllow,
4525	JsonRpcResponseIdPolicy::Preserve,
4526	Some(target.clone()),
4527	)
4528	.await;
4529
4530	assert!(!target.exists());
4531	assert!(matches!(outcome, ToolBatchOutcome::Cancelled(_)));
4532	assert!(messages.iter().any(\|message\| {
4533	message
4534	.pointer("/params/update/status")
4535	.and_then(Value::as_str)
4536	== Some("failed")
4537	}));
4538	let late = late.expect("late allow remains queued for the outer dispatcher");
4539	assert!(is_jsonrpc_response(&late));
4540	assert_eq!(
4541	late.pointer("/result/outcome/optionId")
4542	.and_then(Value::as_str),
4543	Some("allow-once")
4544	);
4545	}
4546
4547	#[tokio::test]
4548	async fn tool_registry_read_file_returns_real_contents() {
4549	let (dir, registry) = workspace_registry();
4550	std::fs::write(dir.path().join("hello.txt"), "hi there").unwrap();
4551
4552	let result = registry
4553	.execute_full("File", json!({"action": "read", "path": "hello.txt"}))
4554	.await
4555	.expect("read_file succeeds");
4556
4557	assert!(result.success);
4558	assert!(result.content.contains("hi there"));
4559	}
4560
4561	#[tokio::test]
4562	async fn tool_registry_write_file_creates_a_real_file() {
4563	let (dir, registry) = workspace_registry();
4564
4565	let result = registry
4566	.execute_full(
4567	"File",
4568	json!({"action": "write", "path": "created.txt", "content": "new content"}),
4569	)
4570	.await
4571	.expect("write_file succeeds");
4572
4573	assert!(result.success);
4574	let on_disk = std::fs::read_to_string(dir.path().join("created.txt")).unwrap();
4575	assert_eq!(on_disk, "new content");
4576	}
4577
4578	#[tokio::test]
4579	async fn tool_registry_list_dir_reports_real_directory_contents() {
4580	let (dir, registry) = workspace_registry();
4581	std::fs::write(dir.path().join("a.txt"), "a").unwrap();
4582	std::fs::write(dir.path().join("b.txt"), "b").unwrap();
4583
4584	let result = registry
4585	.execute_full("File", json!({"action": "list", "path": "."}))
4586	.await
4587	.expect("list_dir succeeds");
4588
4589	assert!(result.success);
4590	assert!(result.content.contains("a.txt"));
4591	assert!(result.content.contains("b.txt"));
4592	}
4593
4594	#[tokio::test]
4595	async fn tool_registry_bash_runs_a_real_command() {
4596	let (_dir, registry) = workspace_registry();
4597
4598	let result = registry
4599	.execute_full("Bash", json!({"command": "echo acp-terminal-check"}))
4600	.await
4601	.expect("Bash succeeds");
4602
4603	assert!(result.content.contains("acp-terminal-check"));
4604	}
4605
4606	#[tokio::test]
4607	async fn tool_registry_read_file_reports_failure_for_missing_path() {
4608	let (_dir, registry) = workspace_registry();
4609
4610	let err = registry
4611	.execute_full(
4612	"File",
4613	json!({"action": "read", "path": "does-not-exist.txt"}),
4614	)
4615	.await
4616	.expect_err("missing file is a tool error");
4617
4618	assert!(!err.to_string().is_empty());
4619	}
4620
4621	/// Feeds [`run_agentic_prompt_turn`] a fixed sequence of canned per-round
4622	/// streams (no real provider), so the multi-round tool loop — including
4623	/// nested tool calls across several rounds — is exercised end-to-end
4624	/// against the real file-tool registry. A plain struct + inherent async
4625	/// method (rather than a boxed closure) lets each test's `\|msgs\| scripted.next()`
4626	/// closure return the async method's own anonymous future type directly,
4627	/// so `Fut` is inferred without needing a `dyn Future` trait object.
4628	struct ScriptedStreams(RefCell<VecDeque<StreamEventBox>>);
4629
4630	impl ScriptedStreams {
4631	fn new(streams: Vec<StreamEventBox>) -> Self {
4632	Self(RefCell::new(VecDeque::from(streams)))
4633	}
4634
4635	async fn next(&self) -> Result<StreamEventBox> {
4636	Ok(self
4637	.0
4638	.borrow_mut()
4639	.pop_front()
4640	.expect("test provided enough scripted rounds"))
4641	}
4642	}
4643
4644	#[tokio::test]
4645	async fn agentic_turn_executes_a_tool_call_then_streams_the_final_answer() {
4646	let (dir, registry) = workspace_registry();
4647	std::fs::write(dir.path().join("VERSION"), "9.9.9").unwrap();
4648
4649	let round1 = ready_stream({
4650	let mut events =
4651	tool_use_events(0, "call_1", "File", r#"{"action":"read","path":"VERSION"}"#);
4652	events.push(StreamEvent::MessageStop);
4653	events
4654	});
4655	let round2 = ready_stream(vec![
4656	text_delta("The version is 9.9.9"),
4657	StreamEvent::MessageStop,
4658	]);
4659
4660	let scripted = ScriptedStreams::new(vec![round1, round2]);
4661	let mut reader = lines_from("");
4662	let mut out = Vec::new();
4663
4664	let (outcome, messages) = run_agentic_prompt_turn(
4665	AcpTurnContext {
4666	config: &Config::default(),
4667	model: "test-model",
4668	session_id: "sess_1",
4669	tool_registry: &registry,
4670	response_id_policy: JsonRpcResponseIdPolicy::Preserve,
4671	},
4672	vec![Message {
4673	role: Role::User,
4674	content: vec![ContentBlock::Text {
4675	text: "What version is this?".to_string(),
4676	cache_control: None,
4677	}],
4678	}],
4679	&mut reader,
4680	&mut out,
4681	\|_msgs\| scripted.next(),
4682	)
4683	.await
4684	.expect("turn completes");
4685
4686	assert_eq!(
4687	outcome,
4688	PromptOutcome::Completed("The version is 9.9.9".to_string())
4689	);
4690	// user -> assistant(tool_use) -> user(tool_result) -> assistant(text)
4691	assert_eq!(messages.len(), 4);
4692	assert!(matches!(
4693	messages[1].content[0],
4694	ContentBlock::ToolUse { .. }
4695	));
4696	let ContentBlock::ToolResult {
4697	content, is_error, ..
4698	} = &messages[2].content[0]
4699	else {
4700	panic!("expected a tool_result message");
4701	};
4702	assert!(content.contains("9.9.9"));
4703	assert_eq!(*is_error, Some(false));
4704
4705	// The client saw a tool_call start, a completed update, and the
4706	// streamed final-answer chunk.
4707	let lines = parse_lines(out);
4708	assert!(
4709	lines
4710	.iter()
4711	.any(\|v\| v["params"]["update"]["sessionUpdate"] == "tool_call")
4712	);
4713	assert!(lines.iter().any(
4714	\|v\| v["params"]["update"]["sessionUpdate"] == "tool_call_update"
4715	&& v["params"]["update"]["status"] == "completed"
4716	));
4717	assert!(lines.iter().any(\|v\| v["params"]["update"]["sessionUpdate"]
4718	== "agent_message_chunk"
4719	&& v["params"]["update"]["content"]["text"] == "The version is 9.9.9"));
4720	}
4721
4722	#[tokio::test]
4723	async fn agentic_turn_preserves_tool_receipts_when_later_provider_round_fails() {
4724	let (dir, registry) = workspace_registry();
4725	std::fs::write(dir.path().join("receipt.txt"), "observed").unwrap();
4726	let round1 = ready_stream({
4727	let mut events = tool_use_events(
4728	0,
4729	"call_receipt",
4730	"File",
4731	r#"{"action":"read","path":"receipt.txt"}"#,
4732	);
4733	events.push(StreamEvent::MessageStop);
4734	events
4735	});
4736	let scripted = ScriptedStreams::new(vec![round1, error_stream("provider unavailable")]);
4737	let mut reader = lines_from("");
4738	let mut out = Vec::new();
4739
4740	let error = run_agentic_prompt_turn(
4741	AcpTurnContext {
4742	config: &Config::default(),
4743	model: "test-model",
4744	session_id: "sess_1",
4745	tool_registry: &registry,
4746	response_id_policy: JsonRpcResponseIdPolicy::Preserve,
4747	},
4748	vec![Message {
4749	role: Role::User,
4750	content: vec![ContentBlock::Text {
4751	text: "Read receipt.txt".to_string(),
4752	cache_control: None,
4753	}],
4754	}],
4755	&mut reader,
4756	&mut out,
4757	\|_msgs\| scripted.next(),
4758	)
4759	.await
4760	.expect_err("second provider round fails");
4761
4762	assert!(error.source.to_string().contains("provider unavailable"));
4763	let messages = error
4764	.partial_messages
4765	.expect("completed tool receipt must be returned for commit");
4766	assert_eq!(messages.len(), 3);
4767	assert!(matches!(
4768	&messages[2].content[0],
4769	ContentBlock::ToolResult { tool_use_id, .. } if tool_use_id == "call_receipt"
4770	));
4771	}
4772
4773	#[tokio::test]
4774	async fn agentic_turn_chains_nested_tool_calls_across_rounds() {
4775	let (dir, registry) = workspace_registry();
4776	std::fs::write(dir.path().join("a.txt"), "contents-of-a").unwrap();
4777	std::fs::write(dir.path().join("b.txt"), "contents-of-b").unwrap();
4778
4779	let round1 = ready_stream({
4780	let mut events =
4781	tool_use_events(0, "call_1", "File", r#"{"action":"read","path":"a.txt"}"#);
4782	events.push(StreamEvent::MessageStop);
4783	events
4784	});
4785	// After seeing a.txt's contents, the model asks for b.txt too.
4786	let round2 = ready_stream({
4787	let mut events =
4788	tool_use_events(0, "call_2", "File", r#"{"action":"read","path":"b.txt"}"#);
4789	events.push(StreamEvent::MessageStop);
4790	events
4791	});
4792	let round3 = ready_stream(vec![
4793	text_delta("Both files read"),
4794	StreamEvent::MessageStop,
4795	]);
4796
4797	let scripted = ScriptedStreams::new(vec![round1, round2, round3]);
4798	let mut reader = lines_from("");
4799	let mut out = Vec::new();
4800
4801	let (outcome, messages) = run_agentic_prompt_turn(
4802	AcpTurnContext {
4803	config: &Config::default(),
4804	model: "test-model",
4805	session_id: "sess_1",
4806	tool_registry: &registry,
4807	response_id_policy: JsonRpcResponseIdPolicy::Preserve,
4808	},
4809	vec![Message {
4810	role: Role::User,
4811	content: vec![ContentBlock::Text {
4812	text: "Read both files".to_string(),
4813	cache_control: None,
4814	}],
4815	}],
4816	&mut reader,
4817	&mut out,
4818	\|_msgs\| scripted.next(),
4819	)
4820	.await
4821	.expect("turn completes");
4822
4823	assert_eq!(
4824	outcome,
4825	PromptOutcome::Completed("Both files read".to_string())
4826	);
4827	// user, assistant(tool_use a), user(result a), assistant(tool_use b),
4828	// user(result b), assistant(text)
4829	assert_eq!(messages.len(), 6);
4830	let ContentBlock::ToolResult {
4831	content: a_content, ..
4832	} = &messages[2].content[0]
4833	else {
4834	panic!("expected tool_result for a.txt");
4835	};
4836	assert!(a_content.contains("contents-of-a"));
4837	let ContentBlock::ToolResult {
4838	content: b_content, ..
4839	} = &messages[4].content[0]
4840	else {
4841	panic!("expected tool_result for b.txt");
4842	};
4843	assert!(b_content.contains("contents-of-b"));
4844	}
4845
4846	#[tokio::test]
4847	async fn agentic_turn_reports_a_tool_failure_back_to_the_model_and_keeps_going() {
4848	let (_dir, registry) = workspace_registry();
4849
4850	let round1 = ready_stream({
4851	let mut events = tool_use_events(
4852	0,
4853	"call_1",
4854	"File",
4855	r#"{"action":"read","path":"missing.txt"}"#,
4856	);
4857	events.push(StreamEvent::MessageStop);
4858	events
4859	});
4860	let round2 = ready_stream(vec![
4861	text_delta("That file does not exist"),
4862	StreamEvent::MessageStop,
4863	]);
4864
4865	let scripted = ScriptedStreams::new(vec![round1, round2]);
4866	let mut reader = lines_from("");
4867	let mut out = Vec::new();
4868
4869	let (outcome, messages) = run_agentic_prompt_turn(
4870	AcpTurnContext {
4871	config: &Config::default(),
4872	model: "test-model",
4873	session_id: "sess_1",
4874	tool_registry: &registry,
4875	response_id_policy: JsonRpcResponseIdPolicy::Preserve,
4876	},
4877	vec![Message {
4878	role: Role::User,
4879	content: vec![ContentBlock::Text {
4880	text: "Read missing.txt".to_string(),
4881	cache_control: None,
4882	}],
4883	}],
4884	&mut reader,
4885	&mut out,
4886	\|_msgs\| scripted.next(),
4887	)
4888	.await
4889	.expect("turn completes even though the tool failed");
4890
4891	assert_eq!(
4892	outcome,
4893	PromptOutcome::Completed("That file does not exist".to_string())
4894	);
4895	let ContentBlock::ToolResult { is_error, .. } = &messages[2].content[0] else {
4896	panic!("expected a tool_result message");
4897	};
4898	assert_eq!(*is_error, Some(true));
4899
4900	let lines = parse_lines(out);
4901	assert!(lines.iter().any(
4902	\|v\| v["params"]["update"]["sessionUpdate"] == "tool_call_update"
4903	&& v["params"]["update"]["status"] == "failed"
4904	));
4905	}
4906
4907	#[cfg(windows)]
4908	const SLOW_SHELL_COMMAND: &str = "ping -n 6 127.0.0.1 >NUL";
4909	#[cfg(not(windows))]
4910	const SLOW_SHELL_COMMAND: &str = "sleep 5";
4911
4912	#[tokio::test]
4913	async fn tool_batch_cancels_a_running_bash_and_applies_response_id_policy() {
4914	let (_dir, registry) = workspace_registry();
4915	let started = std::time::Instant::now();
4916
4917	let (outcome, messages, _) = execute_one_with_permission_client(
4918	&Config::default(),
4919	&registry,
4920	pending_call("Bash", json!({ "command": SLOW_SHELL_COMMAND })),
4921	PermissionClientScript::AllowThenCancelRunning,
4922	JsonRpcResponseIdPolicy::StringifyNumeric,
4923	None,
4924	)
4925	.await;
4926
4927	assert!(matches!(outcome, ToolBatchOutcome::Cancelled(_)));
4928	assert!(
4929	started.elapsed() < std::time::Duration::from_secs(4),
4930	"Bash cancellation must preempt the five-second command"
4931	);
4932	assert!(
4933	messages
4934	.iter()
4935	.any(\|value\| value["id"] == "7" && value["result"].is_null()),
4936	"Zed-compatible cancellation response id was not stringified: {messages:?}"
4937	);
4938	}
4939
4940	#[tokio::test]
4941	async fn concurrent_acp_sessions_execute_tools_independently() {
4942	let (dir1, registry1) = workspace_registry();
4943	let (dir2, registry2) = workspace_registry();
4944	std::fs::write(dir1.path().join("f.txt"), "session-one").unwrap();
4945	std::fs::write(dir2.path().join("f.txt"), "session-two").unwrap();
4946
4947	let (result1, result2) = tokio::join!(
4948	registry1.execute_full("File", json!({"action": "read", "path": "f.txt"})),
4949	registry2.execute_full("File", json!({"action": "read", "path": "f.txt"})),
4950	);
4951
4952	assert!(
4953	result1
4954	.expect("session 1 read")
4955	.content
4956	.contains("session-one")
4957	);
4958	assert!(
4959	result2
4960	.expect("session 2 read")
4961	.content
4962	.contains("session-two")
4963	);
4964	}
4965	}
4966

4966 lines RUST