tkmind/tkmind_go
1
0
Fork 0
Code Issues Pull Requests Actions 5 Packages Projects Releases Wiki Activity

fix(claude-acp): surface tool calls and results to the session bus (#8408)

Browse Source
This commit is contained in:
walid
2026-04-08 23:13:40 -05:00
committed by GitHub
parent f682035bb8
commit e200a896a2
4 changed files with 431 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/goose/src/acp/provider.rs
+325 -11
View File
@@ -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<serde_json::Value>,
},
ToolCallComplete {
id: String,
raw_output: Option<serde_json::Value>,
content: Option<Vec<ToolCallContent>>,
is_error: bool,
},
PermissionRequest {
request: Box<RequestPermissionRequest>,
@@ -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<serde_json::Value>,
content: Vec<ToolCallContent>,
}
/// The single ACP session backing this provider instance.
#[derive(Clone)]
struct AcpSession {
@@ -117,6 +131,7 @@ pub struct AcpProvider {
pending_confirmations:
Arc<TokioMutex<HashMap<String, oneshot::Sender<PermissionConfirmation>>>>,
pending_tool_updates: Arc<Mutex<HashMap<String, AccumulatedToolCall>>>,
tx: Option<mpsc::Sender<ClientRequest>>,
loop_thread: Option<JoinHandle<()>>,
@@ -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<Mutex<HashMap<String, AccumulatedToolCall>>> =
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<Mutex<GooseMode>>,
prompt_response_tx: Arc<Mutex<Option<mpsc::Sender<AcpUpdate>>>>,
pending_tool_updates: Arc<Mutex<HashMap<String, AccumulatedToolCall>>>,
}
impl AcpClientLoop {
fn new(config: AcpProviderConfig, goose_mode: Arc<Mutex<GooseMode>>) -> Self {
fn new(
config: AcpProviderConfig,
goose_mode: Arc<Mutex<GooseMode>>,
pending_tool_updates: Arc<Mutex<HashMap<String, AccumulatedToolCall>>>,
) -> 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<ContentBlock> {
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<Vec<ToolCallContent>>,
raw_output: Option<serde_json::Value>,
) -> Vec<RmcpContent> {
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<Message> {
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<String> = 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:?}"
);
}
}
}
crates/goose/src/agents/reply_parts.rs
+86 -2
View File
@@ -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<serde_json::Value>) -> Tool {
let mut tool = Tool::new("test_tool", "a test tool", object!({ "type": "object" }));
if let Some(v) = meta_json {
crates/goose/src/conversation/message.rs
+15
View File
@@ -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)]
crates/goose/tests/providers.rs
+5
View File
@@ -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(),