fix: filter streamed thinking tags (#1300)

Co-authored-by: GoClaw Operator <operator@goclaw>

Author: nguyenha935

internal/channels/events.go

@@ -98,10 +98,10 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
 			currentStream := rc.stream
 			rc.stream = nil
 			rc.inToolPhase = true
-			rc.thinkingDone = false    // allow new thinking in next iteration
-			rc.thinkingBuffer = ""     // reset thinking buffer for new iteration
-			rc.hasThinking = false     // new iteration starts fresh
-			rc.tagParseSkipped = false // re-enable tag parsing for next iteration
+			rc.thinkingDone = false // allow new thinking in next iteration
+			rc.thinkingBuffer = ""  // reset thinking buffer for new iteration
+			rc.hasThinking = false  // new iteration starts fresh
+			rc.tagParsePending = "" // discard any incomplete tag from the previous iteration
 			rc.mu.Unlock()
 			if currentStream != nil {
 				if err := currentStream.Stop(ctx); err != nil {
@@ -135,19 +135,22 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
 				// Fallback <think> tag parsing: for providers that embed thinking
 				// in the content stream (DeepSeek-via-OpenRouter, Qwen, some Ollama models).
 				// Only activates when no native ChatEventThinking was received.
-				if !rc.hasThinking && !rc.thinkingDone && !rc.tagParseSkipped {
-					candidate := rc.streamBuffer + content
+				if !rc.hasThinking && !rc.thinkingDone {
+					previousAnswer := rc.streamBuffer
+					candidate := rc.streamBuffer + rc.tagParsePending + content
 					split := SplitThinkTags(candidate)
-					if split.Thinking != "" {
+					if split.Found || split.Pending != "" {
 						// Found think tags — commit to buffer and route or suppress reasoning
 						// before any tagged content can leak into the answer lane.
 						displayReasoningInStream := rc.ReasoningDelivery.ShowInChannel && !rc.ReasoningDelivery.BubbleDelivery && sc.ReasoningStreamEnabled()
 						previousThinking := rc.thinkingBuffer
-						rc.streamBuffer = candidate
 						rc.thinkingBuffer = split.Thinking
 						thinkText := rc.thinkingBuffer
 						currentStream := rc.stream
-						if split.Partial {
+						if split.Partial || split.Pending != "" {
+							rc.streamBuffer = split.Answer
+							rc.tagParsePending = split.Pending
+							answerText := rc.streamBuffer
 							// Still inside <think> — wait for the close tag before streaming
 							// answer content. Native thinking uses hasThinking; tag parsing
 							// keeps it false until close so later chunks continue parsing.
@@ -158,9 +161,12 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
 								}
 							} else if displayReasoningInStream && currentStream != nil {
 								currentStream.Update(ctx, formatReasoningPreview(thinkText))
+							} else if currentStream != nil {
+								currentStream.Update(ctx, answerText)
 							}
 							break
 						}
+						rc.tagParsePending = ""
 						// Tag closed — transition to answer, or strip reasoning entirely
 						// when Show Reasoning is off.
 						answerText := split.Answer
@@ -177,7 +183,7 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
 							m.flushReasoningBubbles(runID)
 						}
 
-						if !displayReasoningInStream {
+						if previousAnswer != "" || !displayReasoningInStream {
 							if reasoningStream != nil && answerText != "" {
 								reasoningStream.Update(ctx, answerText)
 							}
@@ -209,9 +215,6 @@ func (m *Manager) HandleAgentEvent(eventType, runID string, payload any) {
 						}
 						break
 					}
-					// No think tags found — mark as skipped so we don't re-parse.
-					// Don't commit to streamBuffer here — the normal flow below appends content.
-					rc.tagParseSkipped = true
 				}
 
 				// Reasoning→answer transition: first chunk after native thinking events.

internal/channels/manager.go

@@ -54,7 +54,7 @@ type RunContext struct {
 	thinkingBuffer       string        // accumulated thinking/reasoning text
 	hasThinking          bool          // true if any thinking events received this iteration
 	thinkingDone         bool          // true after first chunk arrives (reasoning→answer transition complete)
-	tagParseSkipped      bool          // true after first chunk with no <think> tags (skip re-parsing)
+	tagParsePending      string        // raw trailing text withheld because it may be a split <think> tag
 	reasoningBubbles     *reasoningBubbleBuffer
 	reasoningBubbleTimer *time.Timer
 }

internal/channels/reasoning_bubbles.go

@@ -105,25 +105,36 @@ func (m *Manager) appendReasoningBubbleFromThinkTagChunk(runID string, rc *RunCo
 	var done bool
 
 	rc.mu.Lock()
-	if rc.thinkingDone || rc.tagParseSkipped {
+	if rc.thinkingDone {
 		rc.mu.Unlock()
 		return
 	}
 
-	candidate := rc.streamBuffer + content
+	candidate := rc.streamBuffer + rc.tagParsePending + content
 	split := SplitThinkTags(candidate)
-	if split.Thinking == "" {
-		rc.tagParseSkipped = true
+	if !split.Found && split.Pending == "" {
+		rc.streamBuffer = split.Answer
+		rc.mu.Unlock()
+		return
+	}
+	if !split.Found && split.Pending != "" {
+		rc.streamBuffer = split.Answer
+		rc.tagParsePending = split.Pending
 		rc.mu.Unlock()
 		return
 	}
+	if split.Pending != "" {
+		rc.streamBuffer = split.Answer
+		rc.tagParsePending = split.Pending
+	}
 
 	previousThinking := rc.thinkingBuffer
-	rc.streamBuffer = candidate
+	rc.streamBuffer = split.Answer
+	rc.tagParsePending = split.Pending
 	rc.thinkingBuffer = split.Thinking
 	if !split.Partial {
 		rc.thinkingDone = true
-		rc.streamBuffer = split.Answer
+		rc.tagParsePending = ""
 		done = true
 	}
 	delta = reasoningDelta(previousThinking, split.Thinking)

internal/channels/reasoning_delivery_test.go

@@ -203,6 +203,57 @@ func TestHandleAgentEvent_ReasoningOffStripsThinkTagsFromStreamingAnswer(t *test
 	}
 }
 
+func TestHandleAgentEvent_ReasoningOffStripsLateThinkTagsFromStreamingAnswer(t *testing.T) {
+	mb := bus.New()
+	mgr := NewManager(mb)
+	ch := &reasoningStreamingTestChannel{name: "test", reasoningEnabled: true}
+	mgr.RegisterChannel("test", ch)
+	delivery := ResolveReasoningDelivery(ReasoningDeliveryOff, nil)
+	mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, true, false, true, ResolvedChatBehavior{}, delivery)
+
+	mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil)
+	mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "Visible prefix "})
+	mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "<thinking>hidden reasoning</thinking> visible answer"})
+
+	if len(ch.streams) != 1 {
+		t.Fatalf("streams = %d, want answer stream only", len(ch.streams))
+	}
+	got := ch.streams[0].lastUpdate()
+	if strings.Contains(got, "<thinking>") || strings.Contains(got, "hidden reasoning") {
+		t.Fatalf("stream leaked thinking tag content: %q", got)
+	}
+	if got != "Visible prefix  visible answer" {
+		t.Fatalf("stream updates = %q, want clean answer", got)
+	}
+}
+
+func TestHandleAgentEvent_ReasoningOffStripsSplitThinkTagsFromStreamingAnswer(t *testing.T) {
+	mb := bus.New()
+	mgr := NewManager(mb)
+	ch := &reasoningStreamingTestChannel{name: "test", reasoningEnabled: true}
+	mgr.RegisterChannel("test", ch)
+	delivery := ResolveReasoningDelivery(ReasoningDeliveryOff, nil)
+	mgr.RegisterRunWithBehavior("run-1", "test", "chat-1", "msg-1", nil, uuid.Nil, true, false, true, ResolvedChatBehavior{}, delivery)
+
+	mgr.HandleAgentEvent(protocol.AgentEventRunStarted, "run-1", nil)
+	mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "Visible "})
+	mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "<think"})
+
+	if got := ch.streams[0].lastUpdate(); got != "Visible " {
+		t.Fatalf("partial tag name leaked before completion: %q", got)
+	}
+
+	mgr.HandleAgentEvent(protocol.ChatEventChunk, "run-1", map[string]string{"content": "ing>hidden reasoning</thinking> answer"})
+
+	got := ch.streams[0].lastUpdate()
+	if strings.Contains(got, "<thinking>") || strings.Contains(got, "hidden reasoning") {
+		t.Fatalf("stream leaked split thinking tag content: %q", got)
+	}
+	if got != "Visible  answer" {
+		t.Fatalf("stream updates = %q, want clean answer", got)
+	}
+}
+
 func TestHandleAgentEvent_AlwaysBubblesExtractsThinkTagsFromStreamingChunk(t *testing.T) {
 	mb := bus.New()
 	mgr := NewManager(mb)
@@ -267,3 +318,10 @@ func (s *reasoningRecordingStream) Stop(context.Context) error {
 }
 
 func (s *reasoningRecordingStream) MessageID() int { return 1 }
+
+func (s *reasoningRecordingStream) lastUpdate() string {
+	if len(s.updates) == 0 {
+		return ""
+	}
+	return s.updates[len(s.updates)-1]
+}

internal/channels/think_tag_parser.go

@@ -17,6 +17,8 @@ type ThinkTagSplit struct {
 	Thinking string // content inside <think> tags (empty if no tags found)
 	Answer   string // content outside <think> tags
 	Partial  bool   // true if an unclosed <think> tag is present (buffer for more)
+	Found    bool   // true if an opening think tag was found
+	Pending  string // trailing text withheld because it may be a split opening tag
 }
 
 // SplitThinkTags extracts thinking content from <think>...</think> tags.
@@ -28,11 +30,13 @@ func SplitThinkTags(text string) ThinkTagSplit {
 	if !strings.Contains(lower, "<think") &&
 		!strings.Contains(lower, "<thought") &&
 		!strings.Contains(lower, "<antthinking") {
-		return ThinkTagSplit{Answer: text}
+		answer, pending := splitTrailingThinkTagPrefix(text)
+		return ThinkTagSplit{Answer: answer, Pending: pending}
 	}
 
 	var thinking, answer strings.Builder
 	remaining := text
+	found := false
 
 	for remaining != "" {
 		// Find opening tag
@@ -47,6 +51,7 @@ func SplitThinkTags(text string) ThinkTagSplit {
 		if openLoc[0] > 0 {
 			answer.WriteString(remaining[:openLoc[0]])
 		}
+		found = true
 
 		// Find closing tag after the opening tag
 		afterOpen := remaining[openLoc[1]:]
@@ -58,6 +63,8 @@ func SplitThinkTags(text string) ThinkTagSplit {
 				Thinking: thinking.String(),
 				Answer:   answer.String(),
 				Partial:  true,
+				Found:    true,
+				Pending:  remaining[openLoc[0]:],
 			}
 		}
 
@@ -66,8 +73,46 @@ func SplitThinkTags(text string) ThinkTagSplit {
 		remaining = afterOpen[closeLoc[1]:]
 	}
 
+	answerText := answer.String()
+	pending := ""
+	if !found {
+		answerText, pending = splitTrailingThinkTagPrefix(answerText)
+	}
+
 	return ThinkTagSplit{
 		Thinking: thinking.String(),
-		Answer:   answer.String(),
+		Answer:   answerText,
+		Found:    found,
+		Pending:  pending,
+	}
+}
+
+func splitTrailingThinkTagPrefix(text string) (answer, pending string) {
+	idx := strings.LastIndex(text, "<")
+	if idx < 0 {
+		return text, ""
+	}
+	suffix := text[idx:]
+	if strings.Contains(suffix, ">") || !isPossibleThinkOpenPrefix(suffix) {
+		return text, ""
+	}
+	return text[:idx], suffix
+}
+
+func isPossibleThinkOpenPrefix(suffix string) bool {
+	lower := strings.ToLower(suffix)
+	if !strings.HasPrefix(lower, "<") {
+		return false
+	}
+	rest := strings.TrimLeft(lower[1:], " \t\r\n")
+	normalized := "<" + rest
+	if normalized == "<" {
+		return true
+	}
+	for _, tag := range []string{"<think", "<thinking", "<thought", "<antthinking"} {
+		if strings.HasPrefix(tag, normalized) || strings.HasPrefix(normalized, tag) {
+			return true
+		}
 	}
+	return false
 }

internal/pipeline/observe_stage.go

@@ -55,16 +55,17 @@ func (s *ObserveStage) drainInjectedMessages() []providers.Message {
 }
 
 func (s *ObserveStage) observeFinalResponse(state *RunState, resp *providers.ChatResponse, injected []providers.Message) {
+	content, thinking := splitTaggedThinkingContent(resp.Content, resp.Thinking)
 	if len(injected) == 0 {
-		state.Observe.FinalContent = resp.Content
-		state.Observe.FinalThinking = resp.Thinking
+		state.Observe.FinalContent = content
+		state.Observe.FinalThinking = thinking
 		return
 	}
 
 	state.Messages.AppendPending(providers.Message{
 		Role:      "assistant",
-		Content:   resp.Content,
-		Thinking:  resp.Thinking,
+		Content:   content,
+		Thinking:  thinking,
 		Transient: true,
 	})
 	appendPendingMessages(state, injected)

internal/pipeline/stages_test.go

@@ -1981,6 +1981,27 @@ func TestObserveStage_LateInjectionAfterFinalContinuesRun(t *testing.T) {
 	}
 }
 
+func TestObserveStage_MovesTaggedThinkingIntoFinalThinking(t *testing.T) {
+	t.Parallel()
+	stage := NewObserveStage(&PipelineDeps{})
+	state := defaultState()
+	state.Think.LastResponse = &providers.ChatResponse{
+		Content:  "Visible <thinking>hidden reasoning</thinking> answer",
+		Thinking: "native",
+	}
+
+	err := stage.Execute(context.Background(), state)
+	if err != nil {
+		t.Fatalf("Execute() error: %v", err)
+	}
+	if state.Observe.FinalContent != "Visible  answer" {
+		t.Errorf("FinalContent = %q, want %q", state.Observe.FinalContent, "Visible  answer")
+	}
+	if state.Observe.FinalThinking != "native\nhidden reasoning" {
+		t.Errorf("FinalThinking = %q, want %q", state.Observe.FinalThinking, "native\nhidden reasoning")
+	}
+}
+
 func TestObserveStage_FinalContent_SetWhenNoToolCalls(t *testing.T) {
 	t.Parallel()
 	deps := &PipelineDeps{}
@@ -2440,6 +2461,35 @@ func TestFinalizeStage_SanitizesContent(t *testing.T) {
 	}
 }
 
+func TestFinalizeStage_SuppressesSilentReplyAfterFlush(t *testing.T) {
+	t.Parallel()
+	finalContent := "This is directed at Bảo Ly Content, not me. I should stay silent.NO_REPLY"
+	var flushed []providers.Message
+	deps := &PipelineDeps{
+		IsSilentReply: func(content string) bool {
+			return content == finalContent
+		},
+		FlushMessages: func(_ context.Context, _ string, messages []providers.Message) error {
+			flushed = append(flushed, messages...)
+			return nil
+		},
+	}
+	stage := NewFinalizeStage(deps)
+	state := defaultState()
+	state.Observe.FinalContent = finalContent
+
+	err := stage.Execute(context.Background(), state)
+	if err != nil {
+		t.Fatalf("Execute() error: %v", err)
+	}
+	if state.Observe.FinalContent != "" {
+		t.Fatalf("FinalContent = %q, want empty after silent reply suppression", state.Observe.FinalContent)
+	}
+	if len(flushed) == 0 || flushed[len(flushed)-1].Content != finalContent {
+		t.Fatalf("flushed final content = %#v, want original silent reply persisted before suppression", flushed)
+	}
+}
+
 func TestFinalizeStage_DeduplicatesMediaByPath(t *testing.T) {
 	t.Parallel()
 	deps := &PipelineDeps{}
@@ -3332,4 +3382,3 @@ func TestToolStage_Parallel_DefersNonToolMessages(t *testing.T) {
 		t.Errorf("pending[3].Content = %q, want nudge", pending[3].Content)
 	}
 }
-