From 60abcec9934610d08c79e1fc9906738411cb9207 Mon Sep 17 00:00:00 2001 From: Shpat Date: Tue, 12 May 2026 14:15:22 -0400 Subject: [PATCH] fix(acp): coalesce streaming chunks under one message id (#8788) Signed-off-by: Douwe Osinga Co-authored-by: Douwe Osinga --- crates/goose/src/acp/provider.rs | 31 ++++++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/crates/goose/src/acp/provider.rs b/crates/goose/src/acp/provider.rs index 142a3195..a2bf2ca4 100644 --- a/crates/goose/src/acp/provider.rs +++ b/crates/goose/src/acp/provider.rs @@ -356,6 +356,13 @@ impl AcpProvider { } } +fn fresh_text_run() -> (String, i64) { + ( + uuid::Uuid::new_v4().to_string(), + chrono::Utc::now().timestamp(), + ) +} + #[async_trait::async_trait] impl Provider for AcpProvider { fn get_name(&self) -> &str { @@ -451,22 +458,36 @@ impl Provider for AcpProvider { Ok(Box::pin(try_stream! { let mut suppress_text = false; let mut rejected_tool_calls: HashSet = HashSet::new(); + // Stable id+timestamp per contiguous run so Desktop coalesces chunks into one bubble. + let mut text_run: Option<(String, i64)> = None; + let mut thought_run: Option<(String, i64)> = None; while let Some(update) = rx.recv().await { match update { AcpUpdate::Text(text) => { if !suppress_text { - let message = Message::assistant().with_text(text); + let (id, ts) = text_run + .get_or_insert_with(fresh_text_run) + .clone(); + let message = Message::new(Role::Assistant, ts, vec![]) + .with_text(text) + .with_id(id); yield (Some(message), None); } } AcpUpdate::Thought(text) => { - let message = Message::assistant() + let (id, ts) = thought_run + .get_or_insert_with(fresh_text_run) + .clone(); + let message = Message::new(Role::Assistant, ts, vec![]) .with_thinking(text, "") - .with_visibility(true, false); + .with_visibility(true, false) + .with_id(id); yield (Some(message), None); } AcpUpdate::ToolCallStart { id, name, kind, raw_input } => { + text_run = None; + thought_run = None; if reject_all_tools { suppress_text = true; rejected_tool_calls.insert(id); @@ -498,6 +519,8 @@ impl Provider for AcpProvider { content, is_error, } => { + text_run = None; + thought_run = None; if rejected_tool_calls.remove(&id) { // In chat mode no tool_request was emitted (suppressed at // ToolCallStart), so surface a plain text message. In other @@ -527,6 +550,8 @@ impl Provider for AcpProvider { } } AcpUpdate::PermissionRequest { request, response_tx } => { + text_run = None; + thought_run = None; if let Some(decision) = permission_decision_from_mode(goose_mode) { if decision.should_record_rejection() { rejected_tool_calls.insert(request.tool_call.tool_call_id.0.to_string());