diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index beac83ceb9..400e0f4c6b 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -1351,11 +1351,17 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age zap.Bool("is_error", isError)) // Flush the message buffer to publish any remaining content as a streaming message. - // All adapters now handle duplicate prevention at the adapter layer: - // - They send text via message_chunk events (which accumulate in the buffer) - // - They send complete events WITHOUT text - // So we just flush the buffer here - no need to track streamingUsedThisTurn. - m.flushMessageBuffer(execution) + // If the buffer never triggered streaming (no newlines), we emit the text on the + // complete event so the orchestrator can persist the final message. + flushedText := m.flushMessageBuffer(execution) + if flushedText != "" { + event.Text = flushedText + if m.historyManager != nil && execution.SessionID != "" { + if err := m.historyManager.AppendAgentMessage(execution.SessionID, flushedText); err != nil { + m.logger.Warn("failed to store final agent message to history", zap.Error(err)) + } + } + } m.logger.Info("complete event processed", zap.String("execution_id", execution.ID), diff --git a/apps/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go index dc851e03e2..75c92c856e 100644 --- a/apps/backend/internal/agent/lifecycle/manager_test.go +++ b/apps/backend/internal/agent/lifecycle/manager_test.go @@ -839,6 +839,52 @@ func TestHandleAgentEvent_CompleteWithoutStreaming(t *testing.T) { } } +// TestHandleAgentEvent_CompleteWithBufferedText verifies that buffered text +// without streaming is emitted on complete for persistence. +func TestHandleAgentEvent_CompleteWithBufferedText(t *testing.T) { + mgr, eventBus := createTestManagerWithTracking() + execution := createTestExecution("exec-1", "task-1", "session-1") + mgr.executionStore.Add(execution) + + // Buffer text without newlines (no streaming event should be emitted) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "message_chunk", + Text: "Final message without newline", + }) + + // Complete event should flush buffer into complete text + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "complete", + }) + + events := eventBus.getStreamEvents() + + var completeEvents []AgentStreamEventPayload + var streamingEvents []AgentStreamEventPayload + for _, e := range events { + if e.Data != nil { + switch e.Data.Type { + case "complete": + completeEvents = append(completeEvents, e) + case "message_streaming": + streamingEvents = append(streamingEvents, e) + } + } + } + + if len(completeEvents) != 1 { + t.Errorf("expected 1 complete event, got %d", len(completeEvents)) + } + + if len(completeEvents) > 0 && completeEvents[0].Data.Text != "Final message without newline" { + t.Errorf("expected complete event to carry buffered text, got %q", completeEvents[0].Data.Text) + } + + if len(streamingEvents) != 0 { + t.Errorf("expected 0 message_streaming events when no newlines, got %d", len(streamingEvents)) + } +} + // TestHandleAgentEvent_CompleteThenMessageChunk tests the scenario where // message_chunk arrives after complete. This documents the behavior when // an adapter incorrectly sends text after the turn has completed.