From e200a896a24978d7aacb63f1b5e3666d9e4eda8d Mon Sep 17 00:00:00 2001 From: walid <34014558+wkhori@users.noreply.github.com> Date: Wed, 8 Apr 2026 23:13:40 -0500 Subject: [PATCH] fix(claude-acp): surface tool calls and results to the session bus (#8408) --- crates/goose/src/acp/provider.rs | 336 ++++++++++++++++++++++- crates/goose/src/agents/reply_parts.rs | 88 +++++- crates/goose/src/conversation/message.rs | 15 + crates/goose/tests/providers.rs | 5 + 4 files changed, 431 insertions(+), 13 deletions(-) diff --git a/crates/goose/src/acp/provider.rs b/crates/goose/src/acp/provider.rs index 575fe631..30751a18 100644 --- a/crates/goose/src/acp/provider.rs +++ b/crates/goose/src/acp/provider.rs @@ -2,7 +2,7 @@ use agent_client_protocol_schema::AGENT_METHOD_NAMES; use anyhow::{Context, Result}; use async_stream::try_stream; use futures::future::BoxFuture; -use rmcp::model::{Role, Tool}; +use rmcp::model::{CallToolRequestParams, CallToolResult, Content as RmcpContent, Role, Tool}; use sacp::schema::{ ClientCapabilities, CloseSessionRequest, ContentBlock, ContentChunk, EnvVariable, HttpHeader, ImageContent, InitializeRequest, InitializeResponse, McpCapabilities, McpServer, McpServerHttp, @@ -11,7 +11,7 @@ use sacp::schema::{ SessionConfigKind, SessionConfigOption, SessionConfigOptionCategory, SessionConfigSelectOptions, SessionId, SessionNotification, SessionUpdate, SetSessionConfigOptionRequest, SetSessionModeRequest, SetSessionModeResponse, StopReason, - TextContent, ToolCallContent, + TextContent, ToolCallContent, ToolCallStatus, ToolKind, }; use sacp::{Agent, Client, ConnectionTo}; use std::collections::{HashMap, HashSet}; @@ -26,7 +26,7 @@ use tokio_util::compat::{TokioAsyncReadCompatExt as _, TokioAsyncWriteCompatExt use crate::acp::{map_permission_response, PermissionDecision}; use crate::config::{ExtensionConfig, GooseMode}; -use crate::conversation::message::{Message, MessageContent}; +use crate::conversation::message::{Message, MessageContent, TOOL_META_EXTERNAL_DISPATCH_KEY}; use crate::model::ModelConfig; use crate::permission::permission_confirmation::PrincipalType; use crate::permission::{Permission, PermissionConfirmation}; @@ -88,9 +88,15 @@ enum AcpUpdate { Thought(String), ToolCallStart { id: String, + name: String, + kind: ToolKind, + raw_input: Option, }, ToolCallComplete { id: String, + raw_output: Option, + content: Option>, + is_error: bool, }, PermissionRequest { request: Box, @@ -100,6 +106,14 @@ enum AcpUpdate { Error(String), } +/// Per-tool-call buffer for accumulating ACP ToolCallUpdate fields across +/// non-terminal updates, drained on the terminal status update. +#[derive(Debug, Default)] +struct AccumulatedToolCall { + raw_output: Option, + content: Vec, +} + /// The single ACP session backing this provider instance. #[derive(Clone)] struct AcpSession { @@ -117,6 +131,7 @@ pub struct AcpProvider { pending_confirmations: Arc>>>, + pending_tool_updates: Arc>>, tx: Option>, loop_thread: Option>, @@ -193,7 +208,13 @@ impl AcpProvider { let (init_tx, init_rx) = oneshot::channel(); let mode_mapping = config.mode_mapping.clone(); let goose_mode_shared = Arc::new(Mutex::new(goose_mode)); - let client_loop = AcpClientLoop::new(config, goose_mode_shared.clone()); + let pending_tool_updates: Arc>> = + Arc::new(Mutex::new(HashMap::new())); + let client_loop = AcpClientLoop::new( + config, + goose_mode_shared.clone(), + pending_tool_updates.clone(), + ); let loop_thread = spawn_client_loop(run(client_loop, rx, init_tx)); let _init_response = init_rx @@ -238,6 +259,7 @@ impl AcpProvider { mode_mapping, session, pending_confirmations: Arc::new(TokioMutex::new(HashMap::new())), + pending_tool_updates, tx: Some(tx), loop_thread: Some(loop_thread), }) @@ -378,6 +400,11 @@ impl Provider for AcpProvider { let session_id = self.acp_session_id(); let prompt_blocks = messages_to_prompt(messages); + // Drop any tool-call buffer state left over from a prior prompt + // (e.g. cancelled or interrupted before its terminal status arrived). + if let Ok(mut buffer) = self.pending_tool_updates.lock() { + buffer.clear(); + } let mut rx = self .prompt(session_id, prompt_blocks) .await @@ -408,15 +435,63 @@ impl Provider for AcpProvider { .with_visibility(true, false); yield (Some(message), None); } - AcpUpdate::ToolCallStart { id, .. } => { + AcpUpdate::ToolCallStart { id, name, kind, raw_input } => { if reject_all_tools { suppress_text = true; rejected_tool_calls.insert(id); + } else { + let mut params = CallToolRequestParams::new(name); + if let Some(serde_json::Value::Object(map)) = raw_input { + params = params.with_arguments(map); + } + // external_dispatch tells the agent loop not to redispatch this + // call. goose.acp.kind preserves ACP's stable categorization for + // downstream consumers (metrics, observability, icon selection) + // independent of the display title we put in `name`. + let tool_meta = Some(serde_json::json!({ + TOOL_META_EXTERNAL_DISPATCH_KEY: true, + "goose.acp.kind": kind, + })); + let message = Message::assistant().with_tool_request_with_metadata( + id, + Ok(params), + None, + tool_meta, + ); + yield (Some(message), None); } } - AcpUpdate::ToolCallComplete { id, .. } => { + AcpUpdate::ToolCallComplete { + id, + raw_output, + content, + is_error, + } => { if rejected_tool_calls.remove(&id) { - let message = Message::assistant().with_text("Tool call was denied."); + // In chat mode no tool_request was emitted (suppressed at + // ToolCallStart), so surface a plain text message. In other + // modes a tool_request WAS emitted, so pair it with an error + // tool_response so downstream consumers see the rejection. + if reject_all_tools { + let message = Message::assistant() + .with_text("Tool call was denied."); + yield (Some(message), None); + } else { + let denial = vec![RmcpContent::text("Tool call was denied.")]; + let result = CallToolResult::error(denial); + let message = + Message::user().with_tool_response(id, Ok(result)); + yield (Some(message), None); + } + } else { + let result_content = + acp_tool_call_content_to_rmcp(content, raw_output); + let result = if is_error { + CallToolResult::error(result_content) + } else { + CallToolResult::success(result_content) + }; + let message = Message::user().with_tool_response(id, Ok(result)); yield (Some(message), None); } } @@ -486,14 +561,20 @@ struct AcpClientLoop { config: AcpProviderConfig, goose_mode: Arc>, prompt_response_tx: Arc>>>, + pending_tool_updates: Arc>>, } impl AcpClientLoop { - fn new(config: AcpProviderConfig, goose_mode: Arc>) -> Self { + fn new( + config: AcpProviderConfig, + goose_mode: Arc>, + pending_tool_updates: Arc>>, + ) -> Self { Self { config, goose_mode, prompt_response_tx: Arc::new(Mutex::new(None)), + pending_tool_updates, } } @@ -539,6 +620,7 @@ impl AcpClientLoop { config, goose_mode, prompt_response_tx, + pending_tool_updates, } = self; let notification_callback = config.notification_callback.clone(); let reverse_modes = reverse_mode_mapping(&config.mode_mapping); @@ -550,6 +632,7 @@ impl AcpClientLoop { let prompt_response_tx = prompt_response_tx.clone(); let reverse_modes = reverse_modes.clone(); let goose_mode = goose_mode.clone(); + let pending_tool_updates = pending_tool_updates.clone(); async move |notification: SessionNotification, _cx| { if let Some(ref cb) = notification_callback { cb(notification.clone()); @@ -605,14 +688,98 @@ impl AcpClientLoop { let _ = tx.try_send(AcpUpdate::Thought(text)); } SessionUpdate::ToolCall(tool_call) => { + let id = tool_call.tool_call_id.0.to_string(); + let initial_status = tool_call.status; + let synchronous_terminal = matches!( + initial_status, + ToolCallStatus::Completed | ToolCallStatus::Failed + ); + // Seed the buffer; drain immediately if the call is + // already terminal (synchronous tool, no follow-up). + let synchronous_accumulated = + if let Ok(mut buffer) = pending_tool_updates.lock() { + let entry = buffer.entry(id.clone()).or_default(); + if let Some(raw_output) = tool_call.raw_output.clone() { + entry.raw_output = Some(raw_output); + } + entry.content.extend(tool_call.content.clone()); + if synchronous_terminal { + buffer.remove(&id) + } else { + None + } + } else { + None + }; + // ACP carries no canonical tool name to clients — only + // `title` (display) and `kind` (category). We pass `title` + // for renderer affordance, surface `kind` separately via + // tool_meta for stable categorization, and the + // goose.external_dispatch marker keeps `name` off the + // agent loop's routing/auth paths. let _ = tx.try_send(AcpUpdate::ToolCallStart { - id: tool_call.tool_call_id.0.to_string(), + id: id.clone(), + name: tool_call.title.clone(), + kind: tool_call.kind, + raw_input: tool_call.raw_input.clone(), }); + if let Some(accumulated) = synchronous_accumulated { + let content = if accumulated.content.is_empty() { + None + } else { + Some(accumulated.content) + }; + let _ = tx.try_send(AcpUpdate::ToolCallComplete { + id, + raw_output: accumulated.raw_output, + content, + is_error: matches!( + initial_status, + ToolCallStatus::Failed + ), + }); + } } SessionUpdate::ToolCallUpdate(update) => { - if update.fields.status.is_some() { + let id = update.tool_call_id.0.to_string(); + // Merge patch-like fields; only emit on terminal status. + let terminal_status = update.fields.status.filter(|s| { + matches!( + s, + ToolCallStatus::Completed | ToolCallStatus::Failed + ) + }); + let accumulated = if let Ok(mut buffer) = + pending_tool_updates.lock() + { + let entry = buffer.entry(id.clone()).or_default(); + if let Some(raw_output) = update.fields.raw_output.clone() { + entry.raw_output = Some(raw_output); + } + if let Some(content) = update.fields.content.clone() { + entry.content.extend(content); + } + if terminal_status.is_some() { + buffer.remove(&id) + } else { + None + } + } else { + None + }; + if let (Some(accumulated), Some(status)) = + (accumulated, terminal_status) + { + let content = if accumulated.content.is_empty() { + None + } else { + Some(accumulated.content) + }; let _ = tx.try_send(AcpUpdate::ToolCallComplete { - id: update.tool_call_id.0.to_string(), + id, + raw_output: accumulated.raw_output, + content, + is_error: matches!(status, ToolCallStatus::Failed), }); } } @@ -978,6 +1145,63 @@ fn messages_to_prompt(messages: &[Message]) -> Vec { content_blocks } +/// Convert ACP `ToolCallContent` blocks into the rmcp `Content` shape goose's +/// `Message::with_tool_response` consumes. Handles `Content` (text/image/other), +/// `Diff`, and `Terminal` variants; falls back to a JSON serialization of +/// `raw_output` when no blocks are present so the renderer always has something. +fn acp_tool_call_content_to_rmcp( + content: Option>, + raw_output: Option, +) -> Vec { + let mut out = Vec::new(); + if let Some(blocks) = content { + for block in blocks { + match block { + ToolCallContent::Content(val) => match val.content { + ContentBlock::Text(text) => { + out.push(RmcpContent::text(text.text)); + } + ContentBlock::Image(image) => { + out.push(RmcpContent::image(image.data, image.mime_type)); + } + other => { + if let Ok(json) = serde_json::to_string(&other) { + out.push(RmcpContent::text(json)); + } + } + }, + ToolCallContent::Diff(diff) => { + let path = diff.path.display(); + let body = match diff.old_text.as_deref() { + Some(old) => { + format!("--- {path}\n{old}\n+++ {path}\n{}", diff.new_text) + } + None => format!("+++ {path}\n{}", diff.new_text), + }; + out.push(RmcpContent::text(body)); + } + ToolCallContent::Terminal(terminal) => { + out.push(RmcpContent::text(format!( + "[terminal {}]", + terminal.terminal_id.0 + ))); + } + _ => {} + } + } + } + if out.is_empty() { + if let Some(raw) = raw_output { + let text = match raw { + serde_json::Value::String(s) => s, + other => other.to_string(), + }; + out.push(RmcpContent::text(text)); + } + } + out +} + fn build_action_required_message(request: &RequestPermissionRequest) -> Option { let tool_title = request .tool_call @@ -1371,4 +1595,94 @@ mod tests { assert_eq!(result, expected); } } + + #[test] + fn acp_tool_call_content_handles_text_diff_terminal_and_image() { + use sacp::schema::{Diff, Terminal, TerminalId, TextContent}; + + let diff_block = ToolCallContent::Diff( + Diff::new(std::path::PathBuf::from("/tmp/file.txt"), "new\n").old_text("old\n"), + ); + let terminal_block = ToolCallContent::Terminal(Terminal::new(TerminalId::new("term-7"))); + let text_block = ToolCallContent::Content(sacp::schema::Content::new(ContentBlock::Text( + TextContent::new("hello"), + ))); + let image_block = ToolCallContent::Content(sacp::schema::Content::new( + ContentBlock::Image(ImageContent::new("base64data", "image/png")), + )); + + let out = acp_tool_call_content_to_rmcp( + Some(vec![text_block, diff_block, terminal_block, image_block]), + None, + ); + + assert_eq!(out.len(), 4, "all four block kinds should produce output"); + let serialized: Vec = out + .iter() + .map(|c| serde_json::to_string(c).unwrap()) + .collect(); + assert!( + serialized[0].contains("hello"), + "text block lost: {serialized:?}" + ); + assert!( + serialized[1].contains("/tmp/file.txt"), + "diff path lost: {serialized:?}" + ); + assert!( + serialized[1].contains("new"), + "diff body lost: {serialized:?}" + ); + assert!( + serialized[2].contains("term-7"), + "terminal id lost: {serialized:?}" + ); + assert!( + serialized[3].contains("base64data"), + "image data lost: {serialized:?}" + ); + } + + #[test] + fn acp_tool_call_content_falls_back_to_raw_output_when_blocks_empty() { + let out = + acp_tool_call_content_to_rmcp(Some(vec![]), Some(serde_json::json!({"key": "value"}))); + assert_eq!(out.len(), 1); + let serialized = serde_json::to_string(&out[0]).unwrap(); + assert!( + serialized.contains("key"), + "fallback raw_output lost: {serialized}" + ); + } + + /// Pins the tool_meta shape that the `AcpUpdate::ToolCallStart` consumer + /// emits onto the synthesized `ToolRequest`. ACP doesn't expose a canonical + /// tool name to clients, so we surface `kind` here as a stable categorization + /// signal alongside the `external_dispatch` marker that bypasses agent-loop + /// routing. + #[test] + fn tool_meta_pairs_external_dispatch_marker_with_acp_kind() { + let cases = [ + (ToolKind::Execute, "execute"), + (ToolKind::Read, "read"), + (ToolKind::Edit, "edit"), + (ToolKind::Other, "other"), + ]; + for (kind, expected) in cases { + let tool_meta = serde_json::json!({ + TOOL_META_EXTERNAL_DISPATCH_KEY: true, + "goose.acp.kind": kind, + }); + assert_eq!( + tool_meta[TOOL_META_EXTERNAL_DISPATCH_KEY], + serde_json::Value::Bool(true), + "external_dispatch marker missing for kind={kind:?}" + ); + assert_eq!( + tool_meta["goose.acp.kind"], + serde_json::Value::String(expected.to_string()), + "goose.acp.kind serialized wrong for kind={kind:?}" + ); + } + } } diff --git a/crates/goose/src/agents/reply_parts.rs b/crates/goose/src/agents/reply_parts.rs index f6805262..9b39b9d6 100644 --- a/crates/goose/src/agents/reply_parts.rs +++ b/crates/goose/src/agents/reply_parts.rs @@ -356,7 +356,24 @@ impl Agent { coerce_tool_arguments(tool_call.arguments.clone(), &schema_value); if let Some(ref meta) = tool.meta { - coerced_req.tool_meta = serde_json::to_value(meta).ok(); + // Merge registry meta into existing tool_meta; + // existing keys win so provider markers (e.g. + // goose.external_dispatch) survive coercion. + let new_meta = serde_json::to_value(meta).ok(); + coerced_req.tool_meta = + match (coerced_req.tool_meta.take(), new_meta) { + ( + Some(Value::Object(mut existing)), + Some(Value::Object(new)), + ) => { + for (k, v) in new { + existing.entry(k).or_insert(v); + } + Some(Value::Object(existing)) + } + (None, new) => new, + (existing, _) => existing, + }; } } } @@ -388,7 +405,12 @@ impl Agent { let coerced_req = &tool_requests[tool_request_index]; tool_request_index += 1; - let should_include = if let Ok(tool_call) = &coerced_req.tool_call { + // Always keep externally-dispatched requests visible, even if + // their name happens to overlap a registered frontend tool — + // they're observation-only and must not be removed from history. + let should_include = if coerced_req.is_externally_dispatched() { + true + } else if let Ok(tool_call) = &coerced_req.tool_call { !self.is_frontend_tool(&tool_call.name).await } else { true @@ -420,6 +442,11 @@ impl Agent { let mut other_requests = Vec::new(); for request in tool_requests { + // Skip externally-dispatched requests (e.g. claude-acp); the + // provider already executed the tool. Stays in filtered_message. + if request.is_externally_dispatched() { + continue; + } if let Ok(tool_call) = &request.tool_call { if self.is_frontend_tool(&tool_call.name).await { frontend_requests.push(request); @@ -720,6 +747,63 @@ mod tests { )); } + #[tokio::test] + async fn categorize_tool_requests_skips_externally_dispatched_and_preserves_marker() { + // External requests must (1) survive coercion with goose.external_dispatch + // intact, (2) be excluded from both dispatch buckets, (3) stay in + // filtered_message. + use crate::conversation::message::TOOL_META_EXTERNAL_DISPATCH_KEY; + + let agent = crate::agents::Agent::new(); + + let registry_tool = Tool::new("test_tool", "a test tool", object!({ "type": "object" })) + .with_meta(rmcp::model::Meta( + serde_json::json!({ "ui": { "visibility": ["model"] } }) + .as_object() + .unwrap() + .clone(), + )); + + let response = Message::assistant().with_tool_request_with_metadata( + "tool-1", + Ok(rmcp::model::CallToolRequestParams::new("test_tool")), + None, + Some(serde_json::json!({ TOOL_META_EXTERNAL_DISPATCH_KEY: true })), + ); + + let (frontend_requests, other_requests, filtered_message) = agent + .categorize_tool_requests(&response, &[registry_tool], false) + .await; + + assert!( + frontend_requests.is_empty(), + "external request leaked into frontend_requests: {frontend_requests:?}" + ); + assert!( + other_requests.is_empty(), + "external request leaked into other_requests: {other_requests:?}" + ); + assert_eq!(filtered_message.content.len(), 1); + let tool_req = match &filtered_message.content[0] { + MessageContent::ToolRequest(req) => req, + other => panic!("expected ToolRequest, got {other:?}"), + }; + assert!( + tool_req.is_externally_dispatched(), + "goose.external_dispatch marker was clobbered by coercion; merged tool_meta = {:?}", + tool_req.tool_meta + ); + let merged = tool_req + .tool_meta + .as_ref() + .and_then(|v| v.as_object()) + .expect("tool_meta should be an object after merge"); + assert!( + merged.contains_key("ui"), + "registry tool meta keys were dropped; merged tool_meta = {merged:?}" + ); + } + fn make_tool_with_meta(meta_json: Option) -> Tool { let mut tool = Tool::new("test_tool", "a test tool", object!({ "type": "object" })); if let Some(v) = meta_json { diff --git a/crates/goose/src/conversation/message.rs b/crates/goose/src/conversation/message.rs index db5ef206..3490ca9c 100644 --- a/crates/goose/src/conversation/message.rs +++ b/crates/goose/src/conversation/message.rs @@ -105,8 +105,23 @@ impl ToolRequest { Err(e) => format!("Invalid tool call: {}", e), } } + + /// Returns true if this tool request was already executed externally + /// (e.g. by an ACP provider's underlying SDK) and the agent loop must + /// not redispatch it. See [`TOOL_META_EXTERNAL_DISPATCH_KEY`]. + pub fn is_externally_dispatched(&self) -> bool { + self.tool_meta + .as_ref() + .and_then(|v| v.get(TOOL_META_EXTERNAL_DISPATCH_KEY)) + .and_then(|v| v.as_bool()) + .unwrap_or(false) + } } +/// Marker key under `ToolRequest.tool_meta` indicating the tool was already +/// executed externally; the agent loop must skip redispatch. +pub const TOOL_META_EXTERNAL_DISPATCH_KEY: &str = "goose.external_dispatch"; + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] #[derive(ToSchema)] diff --git a/crates/goose/tests/providers.rs b/crates/goose/tests/providers.rs index 36b55d50..f6ad47cb 100644 --- a/crates/goose/tests/providers.rs +++ b/crates/goose/tests/providers.rs @@ -335,6 +335,11 @@ impl ProviderFixture { None => return Ok(response1), }; + // Provider already executed an externally-dispatched tool — don't redispatch. + if tool_req.is_externally_dispatched() { + return Ok(response1); + } + let params = tool_req.tool_call.as_ref().unwrap().clone(); let ctx = goose::agents::ToolCallContext::new( self.session_id.to_string(),