diff --git a/internal/channels/events.go b/internal/channels/events.go index 9b3e4a33..2a567c98 100644 --- a/internal/channels/events.go +++ b/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 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 — 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. diff --git a/internal/channels/manager.go b/internal/channels/manager.go index d2f0a03c..484f45ca 100644 --- a/internal/channels/manager.go +++ b/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 tags (skip re-parsing) + tagParsePending string // raw trailing text withheld because it may be a split tag reasoningBubbles *reasoningBubbleBuffer reasoningBubbleTimer *time.Timer } diff --git a/internal/channels/reasoning_bubbles.go b/internal/channels/reasoning_bubbles.go index e82f5f10..177151bc 100644 --- a/internal/channels/reasoning_bubbles.go +++ b/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) diff --git a/internal/channels/reasoning_delivery_test.go b/internal/channels/reasoning_delivery_test.go index eb187fe2..9ab48c72 100644 --- a/internal/channels/reasoning_delivery_test.go +++ b/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": "hidden reasoning 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, "") || 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": "hidden reasoning answer"}) + + got := ch.streams[0].lastUpdate() + if strings.Contains(got, "") || 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] +} diff --git a/internal/channels/think_tag_parser.go b/internal/channels/think_tag_parser.go index f0a5a6d0..73cfaab1 100644 --- a/internal/channels/think_tag_parser.go +++ b/internal/channels/think_tag_parser.go @@ -17,6 +17,8 @@ type ThinkTagSplit struct { Thinking string // content inside tags (empty if no tags found) Answer string // content outside tags Partial bool // true if an unclosed 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 ... tags. @@ -28,11 +30,13 @@ func SplitThinkTags(text string) ThinkTagSplit { if !strings.Contains(lower, " 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{"hidden reasoning 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) } } - diff --git a/internal/pipeline/think_tag_split.go b/internal/pipeline/think_tag_split.go new file mode 100644 index 00000000..d17c8f2e --- /dev/null +++ b/internal/pipeline/think_tag_split.go @@ -0,0 +1,56 @@ +package pipeline + +import ( + "regexp" + "strings" +) + +var taggedThinkingOpenRe = regexp.MustCompile(`(?is)<\s*(?:redacted_thinking|think(?:ing)?|thought|antthinking)\b[^>]*>`) +var taggedThinkingCloseRe = regexp.MustCompile(`(?is)`) + +func splitTaggedThinkingContent(content, existingThinking string) (string, string) { + lower := strings.ToLower(content) + if !strings.Contains(lower, " { + it("moves legacy tagged thinking from assistant content into thinking", () => { + const messages: Message[] = [ + { role: "assistant", content: "Visible hidden reasoning answer" }, + ]; + + const [msg] = transformHistoryMessages(messages); + + expect(msg?.content).toBe("Visible answer"); + expect(msg?.thinking).toBe("hidden reasoning"); + }); + + it("preserves existing thinking and appends tagged thinking", () => { + const messages: Message[] = [ + { role: "assistant", content: "Visible tagged answer", thinking: "native" }, + ]; + + const [msg] = transformHistoryMessages(messages); + + expect(msg?.content).toBe("Visible answer"); + expect(msg?.thinking).toBe("native\ntagged"); + }); +}); diff --git a/ui/web/src/adapters/chat-message.adapter.ts b/ui/web/src/adapters/chat-message.adapter.ts index 8fdfc7f5..952fe9df 100644 --- a/ui/web/src/adapters/chat-message.adapter.ts +++ b/ui/web/src/adapters/chat-message.adapter.ts @@ -2,6 +2,7 @@ import type { Message } from "@/types/session"; import type { ChatMessage, ToolStreamEntry, MediaItem } from "@/types/chat"; import { toFileUrl } from "@/lib/file-helpers"; import { messageToTimestamp } from "@/lib/message-utils"; +import { splitThinkingTagsFromText } from "@/lib/think-tag-stream"; /** * Transform raw Message[] from history RPC into ChatMessage[] for display. @@ -21,8 +22,13 @@ export function transformHistoryMessages( } const msgs: ChatMessage[] = allMsgs.map((m: Message, i: number) => { + const normalized = m.role === "assistant" + ? splitThinkingTagsFromText(m.content ?? "", m.thinking ?? "") + : { content: m.content, thinking: m.thinking }; const chatMsg: ChatMessage = { ...m, + content: normalized.content, + thinking: normalized.thinking, timestamp: messageToTimestamp(m, i, allMsgs.length), }; diff --git a/ui/web/src/lib/think-tag-stream.test.ts b/ui/web/src/lib/think-tag-stream.test.ts new file mode 100644 index 00000000..9abe40f3 --- /dev/null +++ b/ui/web/src/lib/think-tag-stream.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it } from "vitest"; +import { appendFilteredThinkingChunk, createThinkTagStreamFilterState } from "./think-tag-stream"; + +describe("think tag stream filter", () => { + it("strips late thinking tags from streamed answer text", () => { + let state = createThinkTagStreamFilterState(); + + state = appendFilteredThinkingChunk(state, "Visible prefix "); + state = appendFilteredThinkingChunk(state, "hidden reasoning visible answer"); + + expect(state.text).toBe("Visible prefix visible answer"); + expect(state.thinking).toBe("hidden reasoning"); + expect(state.thinkingDelta).toBe("hidden reasoning"); + }); + + it("strips thinking tags split across stream chunks", () => { + let state = createThinkTagStreamFilterState(); + + state = appendFilteredThinkingChunk(state, "Visible "); + state = appendFilteredThinkingChunk(state, "<"); + state = appendFilteredThinkingChunk(state, "thinking>hidden reasoning"); + state = appendFilteredThinkingChunk(state, " answer"); + + expect(state.text).toBe("Visible answer"); + expect(state.thinking).toBe("hidden reasoning"); + }); + + it("streams thinking text while the closing tag is still pending", () => { + let state = createThinkTagStreamFilterState(); + + state = appendFilteredThinkingChunk(state, "Visible hidden "); + expect(state.text).toBe("Visible "); + expect(state.thinking).toBe("hidden "); + expect(state.thinkingDelta).toBe("hidden "); + + state = appendFilteredThinkingChunk(state, "reasoning answer"); + expect(state.text).toBe("Visible answer"); + expect(state.thinking).toBe("hidden reasoning"); + }); + + it("holds a partial think tag name until the tag resolves", () => { + let state = createThinkTagStreamFilterState(); + + state = appendFilteredThinkingChunk(state, "Visible "); + state = appendFilteredThinkingChunk(state, "hidden reasoning answer"); + + expect(state.text).toBe("Visible answer"); + expect(state.thinking).toBe("hidden reasoning"); + }); +}); diff --git a/ui/web/src/lib/think-tag-stream.ts b/ui/web/src/lib/think-tag-stream.ts new file mode 100644 index 00000000..b69b3a6a --- /dev/null +++ b/ui/web/src/lib/think-tag-stream.ts @@ -0,0 +1,99 @@ +export type ThinkTagStreamFilterState = { + text: string; + thinking: string; + thinkingDelta: string; + pending: string; + inThinking: boolean; +}; + +const openThinkTagRe = /<\s*(?:redacted_thinking|think(?:ing)?|thought|antthinking)\b[^>]*>/i; +const closeThinkTagRe = /<\/\s*(?:redacted_thinking|think(?:ing)?|thought|antthinking)\s*>/i; +const thinkOpenPrefixes = [" 0) { + if (inThinking) { + const close = closeThinkTagRe.exec(input); + if (!close) { + const split = splitTrailingThinkTagPrefix(input, thinkClosePrefixes); + thinking += split.answer; + thinkingDelta += split.answer; + pending = split.pending; + return { text, thinking, thinkingDelta, pending, inThinking: true }; + } + + const thinkingPart = input.slice(0, close.index); + thinking += thinkingPart; + thinkingDelta += thinkingPart; + input = input.slice(close.index + close[0].length); + inThinking = false; + continue; + } + + const open = openThinkTagRe.exec(input); + if (!open) { + const split = splitTrailingThinkTagPrefix(input, thinkOpenPrefixes); + text += split.answer; + pending = split.pending; + break; + } + + text += input.slice(0, open.index); + input = input.slice(open.index + open[0].length); + inThinking = true; + } + + return { text, thinking, thinkingDelta, pending, inThinking }; +} + +export function splitThinkingTagsFromText(content: string, existingThinking = ""): { content: string; thinking?: string } { + const state = appendFilteredThinkingChunk(createThinkTagStreamFilterState(), content); + const thinking = appendThinking(existingThinking, state.thinking); + return { content: state.text, thinking: thinking || undefined }; +} + +function appendThinking(existing: string, addition: string): string { + const cleanExisting = existing.trim(); + const cleanAddition = addition.trim(); + if (!cleanExisting) return cleanAddition; + if (!cleanAddition) return cleanExisting; + return `${cleanExisting}\n${cleanAddition}`; +} + +function splitTrailingThinkTagPrefix(input: string, prefixes: string[]): { answer: string; pending: string } { + const idx = input.lastIndexOf("<"); + if (idx < 0) return { answer: input, pending: "" }; + + const suffix = input.slice(idx); + if (suffix.includes(">") || !isPossibleThinkTagPrefix(suffix, prefixes)) { + return { answer: input, pending: "" }; + } + return { answer: input.slice(0, idx), pending: suffix }; +} + +function isPossibleThinkTagPrefix(suffix: string, prefixes: string[]): boolean { + const lower = suffix.toLowerCase(); + if (!lower.startsWith("<")) return false; + + const normalized = lower.startsWith(" tag.startsWith(normalized) || normalized.startsWith(tag)); +} diff --git a/ui/web/src/pages/chat/hooks/use-chat-messages.ts b/ui/web/src/pages/chat/hooks/use-chat-messages.ts index f59fa420..d71c8f81 100644 --- a/ui/web/src/pages/chat/hooks/use-chat-messages.ts +++ b/ui/web/src/pages/chat/hooks/use-chat-messages.ts @@ -8,6 +8,7 @@ import { toFileUrl, mediaKindFromMime } from "@/lib/file-helpers"; import { transformHistoryMessages } from "@/adapters/chat-message.adapter"; import { useChatTeamTasks } from "./use-chat-team-tasks"; import { useChatMessagesStore } from "@/stores/use-chat-messages-store"; +import { appendFilteredThinkingChunk, createThinkTagStreamFilterState } from "@/lib/think-tag-stream"; // Stable empty array — avoids creating a new reference on every render inside // Zustand selectors, which would trigger an infinite re-render loop (React #185). @@ -41,6 +42,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { const expectingRunRef = useRef(false); const streamRef = useRef(""); const thinkingRef = useRef(""); + const streamFilterRef = useRef(createThinkTagStreamFilterState()); const toolStreamRef = useRef([]); const agentIdRef = useRef(agentId); agentIdRef.current = agentId; @@ -96,6 +98,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { expectingRunRef.current = false; streamRef.current = ""; thinkingRef.current = ""; + streamFilterRef.current = createThinkTagStreamFilterState(); toolStreamRef.current = []; activityRef.current = null; blockRepliesRef.current = []; @@ -158,6 +161,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { setToolStream([]); streamRef.current = ""; thinkingRef.current = ""; + streamFilterRef.current = createThinkTagStreamFilterState(); toolStreamRef.current = []; } return; @@ -179,7 +183,11 @@ export function useChatMessages(sessionKey: string, agentId: string) { break; } case "chunk": { - streamRef.current += event.payload?.content ?? ""; + streamFilterRef.current = appendFilteredThinkingChunk(streamFilterRef.current, event.payload?.content ?? ""); + if (streamFilterRef.current.thinkingDelta) { + thinkingRef.current += streamFilterRef.current.thinkingDelta; + } + streamRef.current = streamFilterRef.current.text; if (!rafPendingRef.current) { rafPendingRef.current = true; rafHandleRef.current = requestAnimationFrame(() => { @@ -246,6 +254,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { setToolStream([]); streamRef.current = ""; thinkingRef.current = ""; + streamFilterRef.current = createThinkTagStreamFilterState(); toolStreamRef.current = []; activityRef.current = null; setActivity(null); @@ -269,6 +278,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { setToolStream([]); streamRef.current = ""; thinkingRef.current = ""; + streamFilterRef.current = createThinkTagStreamFilterState(); activityRef.current = null; setActivity(null); blockRepliesRef.current = []; @@ -286,6 +296,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { setToolStream([]); streamRef.current = ""; thinkingRef.current = ""; + streamFilterRef.current = createThinkTagStreamFilterState(); toolStreamRef.current = []; activityRef.current = null; setActivity(null);