From 30f91176ae287f8f8314452e1518b6a9ee54dea0 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sun, 6 Sep 2026 14:10:11 +0100 Subject: [PATCH] fix(daemon): coalesce anonymous ACP text chunks --- crates/maestro-daemon/src/projection.rs | 96 ++++++++++++++++++++++--- docs/STRUCTURED-VIEW.md | 4 +- 2 files changed, 88 insertions(+), 12 deletions(-) diff --git a/crates/maestro-daemon/src/projection.rs b/crates/maestro-daemon/src/projection.rs index fb0a47f..0209d4d 100644 --- a/crates/maestro-daemon/src/projection.rs +++ b/crates/maestro-daemon/src/projection.rs @@ -40,6 +40,7 @@ pub struct Projection { turn_failed: bool, next_turn: u64, turn_tool_calls: HashSet, + anonymous_content_row: Option<(String, String)>, } impl Projection { @@ -60,6 +61,7 @@ impl Projection { SessionUpdate::ToolCall(tool) => self.tool_call(tool), SessionUpdate::ToolCallUpdate(update) => self.tool_update(update), SessionUpdate::Plan(plan) => { + self.anonymous_content_row = None; let data = to_value(plan); self.state.plan = Some(data.clone()); Some(self.replace("plan".into(), TranscriptRowKind::Plan, data)) @@ -82,11 +84,14 @@ impl Projection { // ACP deliberately adds notification variants over time. Known but // non-transcript updates become a generic notice instead of being // dropped or forcing clients to understand ACP's enum. - _ => Some(self.append( - format!("notice:{seq}"), - TranscriptRowKind::Notice, - to_value(¬ification.update), - )), + _ => { + self.anonymous_content_row = None; + Some(self.append( + format!("notice:{seq}"), + TranscriptRowKind::Notice, + to_value(¬ification.update), + )) + } }; ProjectionUpdate { transcript, @@ -122,6 +127,7 @@ impl Projection { /// Start a prompt turn and return its monotonically increasing token. pub fn begin_turn(&mut self) -> (u64, ProjectionUpdate) { + self.anonymous_content_row = None; self.next_turn += 1; let turn = self.next_turn; self.active_turn = Some(turn); @@ -245,12 +251,32 @@ impl Projection { kind: TranscriptRowKind, prefix: &str, ) -> Option { - let id = chunk - .message_id - .as_ref() - .map(|id| format!("{prefix}:{id}")) - .unwrap_or_else(|| format!("{prefix}:{seq}")); let data = to_value(chunk); + let id = if let Some(message_id) = chunk.message_id.as_ref() { + self.anonymous_content_row = None; + format!("{prefix}:{message_id}") + } else { + let text = data.pointer("/content/text").and_then(Value::as_str); + let previous = self + .anonymous_content_row + .as_ref() + .filter(|(previous_prefix, _)| previous_prefix == prefix) + .map(|(_, id)| id.clone()); + if text.is_some_and(|text| !text.is_empty()) { + if let Some(id) = previous { + let index = *self + .row_index + .get(&id) + .expect("anonymous content row remains addressable"); + let row = &mut self.rows[index]; + append_text(&mut row.data, &data); + return Some(TranscriptDelta::Patch { row: row.clone() }); + } + } + let id = format!("{prefix}:{seq}"); + self.anonymous_content_row = text.map(|_| (prefix.to_owned(), id.clone())); + id + }; if let Some(index) = self.row_index.get(&id).copied() { let row = &mut self.rows[index]; append_text(&mut row.data, &data); @@ -260,6 +286,7 @@ impl Projection { } fn tool_call(&mut self, tool: &ToolCall) -> Option { + self.anonymous_content_row = None; let id = format!("tool:{}", tool.tool_call_id); let tool_id = tool.tool_call_id.to_string(); let mut canonical = tool.clone(); @@ -290,6 +317,7 @@ impl Projection { &mut self, update: &maestro_acp::schema::ToolCallUpdate, ) -> Option { + self.anonymous_content_row = None; let tool_id = update.tool_call_id.to_string(); let Some(tool) = self.tools.get_mut(&tool_id) else { self.pending_tool_updates @@ -496,6 +524,54 @@ mod tests { ); } + #[test] + fn coalesces_anonymous_stream_chunks_without_joining_turns() { + let mut projection = Projection::default(); + let (first_turn, _) = projection.begin_turn(); + for (seq, text) in [(1, ""), (2, "p"), (3, "ong")] { + projection.apply( + seq, + &event(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": text}, + })), + ); + } + projection.turn_lifecycle(TurnLifecycle::Finish, first_turn); + projection.begin_turn(); + projection.apply( + 4, + &event(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": "next"}, + })), + ); + + let rows = projection.snapshot(4).rows; + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].data.pointer("/content/text").unwrap(), "pong"); + assert_eq!(rows[1].data.pointer("/content/text").unwrap(), "next"); + } + + #[test] + fn anonymous_empty_chunk_delimits_turns_during_replay() { + let mut projection = Projection::default(); + for (seq, text) in [(1, ""), (2, "visible"), (3, ""), (4, "p"), (5, "ong")] { + projection.apply( + seq, + &event(serde_json::json!({ + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": text}, + })), + ); + } + + let rows = projection.snapshot(5).rows; + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].data.pointer("/content/text").unwrap(), "visible"); + assert_eq!(rows[1].data.pointer("/content/text").unwrap(), "pong"); + } + #[test] fn tool_update_patches_the_original_stable_row() { let mut projection = Projection::default(); diff --git a/docs/STRUCTURED-VIEW.md b/docs/STRUCTURED-VIEW.md index f6ac2f2..69f0876 100644 --- a/docs/STRUCTURED-VIEW.md +++ b/docs/STRUCTURED-VIEW.md @@ -107,8 +107,8 @@ crate. | Kind | Carries | Row identity | | --- | --- | --- | | `User` | An echoed `user_message_chunk` — the prompt text as ACP saw it. | `user:`, or `user:` when the chunk carries no id. | -| `AgentText` | An `agent_message_chunk` — the agent's visible reply, streamed. | `agent:` / `agent:`. Later chunks for the same id are appended into `data.content.text` in place, not appended as new rows. | -| `AgentThought` | An `agent_thought_chunk` — the model's reasoning stream. | `thought:` / `thought:`, coalesced the same way as `AgentText`. | +| `AgentText` | An `agent_message_chunk` — the agent's visible reply, streamed. | `agent:`, or `agent:` from the first chunk when no id is supplied. Later chunks for the same id—or consecutive anonymous text chunks—are appended into `data.content.text` in place. An empty anonymous chunk starts a fresh row, matching the turn marker emitted by adapters that omit message ids. | +| `AgentThought` | An `agent_thought_chunk` — the model's reasoning stream. | `thought:` / first anonymous `thought:`, coalesced the same way as `AgentText`. | | `ToolCall` | A `tool_call` / `tool_call_update` pair whose content carries no diff. | `tool:` — one row for the call's whole life; updates patch it in place, they never create a second row. | | `ToolDiff` | The same tool-call machinery, when the tool's content includes a `Diff` block. | Same `tool:` identity as `ToolCall`; the *kind* flips between the two as the tool's own content changes, not the row. | | `Approval` | A daemon-originated permission request (`RequestPermissionRequest`), serialized without its nonce. | `approval:` — minted fresh per request, unrelated to the ACP request's own id. |