fix(lifecycle): don't split streaming messages on subagent tool calls (#1756)

* fix(lifecycle): don't split streaming messages on subagent tool calls A subagent's internal tool calls (ParentToolCallID set) stream on the same session concurrently with the parent agent's own text. Flushing the message buffer on those events closed the parent's in-flight streaming message, so each subagent tool call started a new message row — splitting one logical message mid-sentence and breaking markdown (e.g. **bold** spans rendered as literal asterisks) across the fragments. Only flush on top-level tool calls; subagent-internal ones leave the parent's streaming segment open. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor(lifecycle): drop dead flush-text branch in tool call handler flushMessageBuffer always publishes remaining content through the streaming path and returns "", so the event.Text / history-append branch could never execute. Call it directly instead of through a helper carrying dead code. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Kandev Agent <agent@kandev.dev> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-07-17 14:31:28 +01:00
parent 0df2bbac0a
commit 1806b87e7e
2 changed files with 91 additions and 7 deletions
@@ -322,14 +322,16 @@ func (m *Manager) handleCompleteEvent(execution *AgentExecution, event *agentctl
 // handleToolCallEvent processes the "tool_call" agent event: flushes the message buffer
 // and stores the tool call in session history.
 // Returns the (possibly updated) event.
+//
+// Subagent-internal tool calls (ParentToolCallID set) stream on the same session
+// concurrently with the parent agent's own text, so they must NOT flush the
+// buffer — flushing would split the parent's in-flight streaming message into
+// separate DB rows mid-sentence, breaking markdown that spans the boundary.
 func (m *Manager) handleToolCallEvent(execution *AgentExecution, event agentctl.AgentEvent) agentctl.AgentEvent {
-	if flushedText := m.flushMessageBuffer(execution); flushedText != "" {
-		event.Text = flushedText
-		if m.historyManager != nil && execution.historyEnabled && execution.SessionID != "" {
-			if err := m.historyManager.AppendAgentMessage(execution.SessionID, flushedText); err != nil {
-				m.logger.Warn("failed to store agent message to history", zap.Error(err))
-			}
-		}
+	if event.ParentToolCallID == "" {
+		// flushMessageBuffer publishes any remaining buffered content through
+		// the streaming path itself and always returns "".
+		m.flushMessageBuffer(execution)
 	}
 	if m.historyManager != nil && execution.historyEnabled && execution.SessionID != "" {
 		if err := m.historyManager.AppendToolCall(execution.SessionID, event); err != nil {
@@ -208,6 +208,88 @@ func TestHandleAgentEvent_StreamingThenToolCallThenComplete(t *testing.T) {
 	}
 }

+// TestHandleAgentEvent_SubagentToolCallDoesNotSplitStreaming is the regression test
+// for streaming messages being shattered into multiple DB rows: a subagent (Task tool)
+// streams its internal tool calls (tagged with ParentToolCallID) on the same session
+// while the parent agent is still streaming text. Those tool calls must NOT flush the
+// message buffer — otherwise every subagent tool call starts a new message row,
+// splitting the parent's message mid-sentence and breaking markdown across rows.
+func TestHandleAgentEvent_SubagentToolCallDoesNotSplitStreaming(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	if err := mgr.executionStore.Add(execution); err != nil {
+		t.Fatalf("add execution: %v", err)
+	}
+
+	// Parent agent starts streaming its message
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Here's the CI picture\n",
+	})
+
+	execution.messageMu.Lock()
+	msgIDBefore := execution.currentMessageID
+	execution.messageMu.Unlock()
+	if msgIDBefore == "" {
+		t.Fatal("currentMessageID should be set after message_chunk")
+	}
+
+	// A subagent's internal tool call arrives mid-stream (ParentToolCallID set)
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:             "tool_call",
+		ToolCallID:       "subagent-tool-1",
+		ParentToolCallID: "parent-task-tool",
+		ToolName:         "execute",
+	})
+
+	// The parent's streaming message must survive the subagent tool call
+	execution.messageMu.Lock()
+	msgIDAfter := execution.currentMessageID
+	execution.messageMu.Unlock()
+	if msgIDAfter != msgIDBefore {
+		t.Errorf("subagent tool_call must not close the streaming message: message ID changed from %q to %q",
+			msgIDBefore, msgIDAfter)
+	}
+
+	// More parent text — must APPEND to the same message, not create a new one
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "on the new PR.\n",
+	})
+
+	var streamingEvents []AgentStreamEventPayload
+	for _, e := range eventBus.getStreamEvents() {
+		if e.Data != nil && e.Data.Type == "message_streaming" {
+			streamingEvents = append(streamingEvents, e)
+		}
+	}
+	if len(streamingEvents) < 2 {
+		t.Fatalf("expected at least 2 message_streaming events, got %d", len(streamingEvents))
+	}
+	last := streamingEvents[len(streamingEvents)-1]
+	if !last.Data.IsAppend {
+		t.Error("text after a subagent tool_call should append to the existing message, got IsAppend=false")
+	}
+	if last.Data.MessageID != msgIDBefore {
+		t.Errorf("text after a subagent tool_call should keep message ID %q, got %q",
+			msgIDBefore, last.Data.MessageID)
+	}
+
+	// A top-level tool call (no ParentToolCallID) must still flush as before
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:       "tool_call",
+		ToolCallID: "tool-1",
+		ToolName:   "read_file",
+	})
+
+	execution.messageMu.Lock()
+	msgIDAfterTopLevel := execution.currentMessageID
+	execution.messageMu.Unlock()
+	if msgIDAfterTopLevel != "" {
+		t.Errorf("currentMessageID should be cleared after a top-level tool_call, got %q", msgIDAfterTopLevel)
+	}
+}
+
 // TestHandleAgentEvent_CompleteWithoutStreaming verifies that complete events are
 // properly handled when no streaming was used (buffer is empty).
 // All adapters now send text via message_chunk events, so this tests the empty buffer case.