lisone/ds2api
1
0
Fork 0
mirror of https://github.com/CJackHwang/ds2api.git synced 2026-05-13 12:47:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor streaming accumulation and chat history UI

Browse Source
This commit is contained in:
CJACK
2026-05-02 20:15:38 +08:00
parent 20d71f528a
commit c8f7b6b371
13 changed files with 1223 additions and 1037 deletions
Download Patch File Download Diff File Expand all files Collapse all files

165
internal/httpapi/openai/chat/chat_stream_runtime.go
View File

@@ -6,6 +6,7 @@ import (
"strings"
openaifmt "ds2api/internal/format/openai"
"ds2api/internal/httpapi/openai/shared"
"ds2api/internal/sse"
streamengine "ds2api/internal/stream"
"ds2api/internal/toolstream"
@@ -34,15 +35,11 @@ type chatStreamRuntime struct {
toolCallsEmitted bool
toolCallsDoneEmitted bool
toolSieve toolstream.State
streamToolCallIDs map[int]string
streamToolNames map[int]string
rawThinking strings.Builder
thinking strings.Builder
toolDetectionThinking strings.Builder
rawText strings.Builder
text strings.Builder
responseMessageID int
toolSieve toolstream.State
streamToolCallIDs map[int]string
streamToolNames map[int]string
accumulator shared.StreamAccumulator
responseMessageID int
finalThinking string
finalText string
@@ -112,6 +109,11 @@ func newChatStreamRuntime(
emitEarlyToolDeltas: emitEarlyToolDeltas,
streamToolCallIDs: map[int]string{},
streamToolNames: map[int]string{},
accumulator: shared.StreamAccumulator{
ThinkingEnabled: thinkingEnabled,
SearchEnabled: searchEnabled,
StripReferenceMarkers: stripReferenceMarkers,
},
}
}
@@ -177,8 +179,8 @@ func (s *chatStreamRuntime) markContextCancelled() {
s.finalErrorStatus = 499
s.finalErrorMessage = "request context cancelled"
s.finalErrorCode = string(streamengine.StopReasonContextCancelled)
s.finalThinking = s.thinking.String()
s.finalText = cleanVisibleOutput(s.text.String(), s.stripReferenceMarkers)
s.finalThinking = s.accumulator.Thinking.String()
s.finalText = cleanVisibleOutput(s.accumulator.Text.String(), s.stripReferenceMarkers)
s.finalFinishReason = string(streamengine.StopReasonContextCancelled)
}
@@ -191,12 +193,12 @@ func (s *chatStreamRuntime) finalize(finishReason string, deferEmptyOutput bool)
s.finalErrorStatus = 0
s.finalErrorMessage = ""
s.finalErrorCode = ""
finalThinking := s.thinking.String()
finalToolDetectionThinking := s.toolDetectionThinking.String()
finalText := cleanVisibleOutput(s.text.String(), s.stripReferenceMarkers)
finalThinking := s.accumulator.Thinking.String()
finalToolDetectionThinking := s.accumulator.ToolDetectionThinking.String()
finalText := cleanVisibleOutput(s.accumulator.Text.String(), s.stripReferenceMarkers)
s.finalThinking = finalThinking
s.finalText = finalText
detected := detectAssistantToolCalls(s.rawText.String(), finalText, s.rawThinking.String(), finalToolDetectionThinking, s.toolNames)
detected := detectAssistantToolCalls(s.accumulator.RawText.String(), finalText, s.accumulator.RawThinking.String(), finalToolDetectionThinking, s.toolNames)
if len(detected.Calls) > 0 && !s.toolCallsDoneEmitted {
finishReason = "tool_calls"
s.sendDelta(map[string]any{
@@ -265,7 +267,7 @@ func (s *chatStreamRuntime) onParsed(parsed sse.LineResult) streamengine.ParsedD
s.responseMessageID = parsed.ResponseMessageID
}
if parsed.ContentFilter {
if strings.TrimSpace(s.text.String()) == "" {
if strings.TrimSpace(s.accumulator.Text.String()) == "" {
return streamengine.ParsedDecision{Stop: true, StopReason: streamengine.StopReason("content_filter")}
}
return streamengine.ParsedDecision{Stop: true, StopReason: streamengine.StopReasonHandlerRequested}
@@ -277,98 +279,65 @@ func (s *chatStreamRuntime) onParsed(parsed sse.LineResult) streamengine.ParsedD
return streamengine.ParsedDecision{Stop: true, StopReason: streamengine.StopReasonHandlerRequested}
}
contentSeen := false
batch := chatDeltaBatch{runtime: s}
for _, p := range parsed.ToolDetectionThinkingParts {
trimmed := sse.TrimContinuationOverlap(s.toolDetectionThinking.String(), p.Text)
if trimmed != "" {
s.toolDetectionThinking.WriteString(trimmed)
}
}
for _, p := range parsed.Parts {
accumulated := s.accumulator.Apply(parsed)
for _, p := range accumulated.Parts {
if p.Type == "thinking" {
rawTrimmed := sse.TrimContinuationOverlap(s.rawThinking.String(), p.Text)
if rawTrimmed != "" {
s.rawThinking.WriteString(rawTrimmed)
contentSeen = true
}
if s.thinkingEnabled {
cleanedText := cleanVisibleOutput(rawTrimmed, s.stripReferenceMarkers)
if cleanedText == "" {
continue
}
trimmed := sse.TrimContinuationOverlap(s.thinking.String(), cleanedText)
if trimmed == "" {
continue
}
s.thinking.WriteString(trimmed)
batch.append("reasoning_content", trimmed)
}
batch.append("reasoning_content", p.VisibleText)
continue
}
if p.RawText == "" {
continue
}
if p.CitationOnly {
continue
}
if !s.bufferToolContent {
batch.append("content", p.VisibleText)
} else {
rawTrimmed := sse.TrimContinuationOverlap(s.rawText.String(), p.Text)
if rawTrimmed == "" {
continue
}
s.rawText.WriteString(rawTrimmed)
contentSeen = true
cleanedText := cleanVisibleOutput(rawTrimmed, s.stripReferenceMarkers)
if s.searchEnabled && sse.IsCitation(cleanedText) {
continue
}
trimmed := sse.TrimContinuationOverlap(s.text.String(), cleanedText)
if trimmed != "" {
s.text.WriteString(trimmed)
}
if !s.bufferToolContent {
if trimmed == "" {
events := toolstream.ProcessChunk(&s.toolSieve, p.RawText, s.toolNames)
for _, evt := range events {
if len(evt.ToolCallDeltas) > 0 {
if !s.emitEarlyToolDeltas {
continue
}
filtered := filterIncrementalToolCallDeltasByAllowed(evt.ToolCallDeltas, s.streamToolNames)
if len(filtered) == 0 {
continue
}
formatted := formatIncrementalStreamToolCallDeltas(filtered, s.streamToolCallIDs)
if len(formatted) == 0 {
continue
}
batch.flush()
tcDelta := map[string]any{
"tool_calls": formatted,
}
s.toolCallsEmitted = true
s.sendDelta(tcDelta)
continue
}
batch.append("content", trimmed)
} else {
events := toolstream.ProcessChunk(&s.toolSieve, rawTrimmed, s.toolNames)
for _, evt := range events {
if len(evt.ToolCallDeltas) > 0 {
if !s.emitEarlyToolDeltas {
continue
}
filtered := filterIncrementalToolCallDeltasByAllowed(evt.ToolCallDeltas, s.streamToolNames)
if len(filtered) == 0 {
continue
}
formatted := formatIncrementalStreamToolCallDeltas(filtered, s.streamToolCallIDs)
if len(formatted) == 0 {
continue
}
batch.flush()
tcDelta := map[string]any{
"tool_calls": formatted,
}
s.toolCallsEmitted = true
s.sendDelta(tcDelta)
if len(evt.ToolCalls) > 0 {
batch.flush()
s.toolCallsEmitted = true
s.toolCallsDoneEmitted = true
tcDelta := map[string]any{
"tool_calls": formatFinalStreamToolCallsWithStableIDs(evt.ToolCalls, s.streamToolCallIDs, s.toolsRaw),
}
s.sendDelta(tcDelta)
s.resetStreamToolCallState()
continue
}
if evt.Content != "" {
cleaned := cleanVisibleOutput(evt.Content, s.stripReferenceMarkers)
if cleaned == "" || (s.searchEnabled && sse.IsCitation(cleaned)) {
continue
}
if len(evt.ToolCalls) > 0 {
batch.flush()
s.toolCallsEmitted = true
s.toolCallsDoneEmitted = true
tcDelta := map[string]any{
"tool_calls": formatFinalStreamToolCallsWithStableIDs(evt.ToolCalls, s.streamToolCallIDs, s.toolsRaw),
}
s.sendDelta(tcDelta)
s.resetStreamToolCallState()
continue
}
if evt.Content != "" {
cleaned := cleanVisibleOutput(evt.Content, s.stripReferenceMarkers)
if cleaned == "" || (s.searchEnabled && sse.IsCitation(cleaned)) {
continue
}
batch.append("content", cleaned)
}
batch.append("content", cleaned)
}
}
}
}
batch.flush()
return streamengine.ParsedDecision{ContentSeen: contentSeen}
return streamengine.ParsedDecision{ContentSeen: accumulated.ContentSeen}
}

8
internal/httpapi/openai/chat/empty_retry_runtime.go
View File

@@ -237,7 +237,7 @@ func (h *Handler) consumeChatStreamAttempt(r *http.Request, resp *http.Response,
OnParsed: func(parsed sse.LineResult) streamengine.ParsedDecision {
decision := streamRuntime.onParsed(parsed)
if historySession != nil {
historySession.progress(streamRuntime.thinking.String(), streamRuntime.text.String())
historySession.progress(streamRuntime.accumulator.Thinking.String(), streamRuntime.accumulator.Text.String())
}
return decision
},
@@ -249,7 +249,7 @@ func (h *Handler) consumeChatStreamAttempt(r *http.Request, resp *http.Response,
OnContextDone: func() {
streamRuntime.markContextCancelled()
if historySession != nil {
historySession.stopped(streamRuntime.thinking.String(), streamRuntime.text.String(), string(streamengine.StopReasonContextCancelled))
historySession.stopped(streamRuntime.accumulator.Thinking.String(), streamRuntime.accumulator.Text.String(), string(streamengine.StopReasonContextCancelled))
}
},
})
@@ -269,7 +269,7 @@ func recordChatStreamHistory(streamRuntime *chatStreamRuntime, historySession *c
return
}
if streamRuntime.finalErrorMessage != "" {
historySession.error(streamRuntime.finalErrorStatus, streamRuntime.finalErrorMessage, streamRuntime.finalErrorCode, streamRuntime.thinking.String(), streamRuntime.text.String())
historySession.error(streamRuntime.finalErrorStatus, streamRuntime.finalErrorMessage, streamRuntime.finalErrorCode, streamRuntime.accumulator.Thinking.String(), streamRuntime.accumulator.Text.String())
return
}
historySession.success(http.StatusOK, streamRuntime.finalThinking, streamRuntime.finalText, streamRuntime.finalFinishReason, streamRuntime.finalUsage)
@@ -278,7 +278,7 @@ func recordChatStreamHistory(streamRuntime *chatStreamRuntime, historySession *c
func failChatStreamRetry(streamRuntime *chatStreamRuntime, historySession *chatHistorySession, status int, message, code string) {
streamRuntime.sendFailedChunk(status, message, code)
if historySession != nil {
historySession.error(status, message, code, streamRuntime.thinking.String(), streamRuntime.text.String())
historySession.error(status, message, code, streamRuntime.accumulator.Thinking.String(), streamRuntime.accumulator.Text.String())
}
}

6
internal/httpapi/openai/chat/handler_chat.go
View File

@@ -254,7 +254,7 @@ func (h *Handler) handleStream(w http.ResponseWriter, r *http.Request, resp *htt
OnParsed: func(parsed sse.LineResult) streamengine.ParsedDecision {
decision := streamRuntime.onParsed(parsed)
if historySession != nil {
historySession.progress(streamRuntime.thinking.String(), streamRuntime.text.String())
historySession.progress(streamRuntime.accumulator.Thinking.String(), streamRuntime.accumulator.Text.String())
}
return decision
},
@@ -268,14 +268,14 @@ func (h *Handler) handleStream(w http.ResponseWriter, r *http.Request, resp *htt
return
}
if streamRuntime.finalErrorMessage != "" {
historySession.error(streamRuntime.finalErrorStatus, streamRuntime.finalErrorMessage, streamRuntime.finalErrorCode, streamRuntime.thinking.String(), streamRuntime.text.String())
historySession.error(streamRuntime.finalErrorStatus, streamRuntime.finalErrorMessage, streamRuntime.finalErrorCode, streamRuntime.accumulator.Thinking.String(), streamRuntime.accumulator.Text.String())
return
}
historySession.success(http.StatusOK, streamRuntime.finalThinking, streamRuntime.finalText, streamRuntime.finalFinishReason, streamRuntime.finalUsage)
},
OnContextDone: func() {
if historySession != nil {
historySession.stopped(streamRuntime.thinking.String(), streamRuntime.text.String(), string(streamengine.StopReasonContextCancelled))
historySession.stopped(streamRuntime.accumulator.Thinking.String(), streamRuntime.accumulator.Text.String(), string(streamengine.StopReasonContextCancelled))
}
},
})

111
internal/httpapi/openai/responses/responses_stream_runtime_core.go
View File

@@ -7,6 +7,7 @@ import (
"ds2api/internal/config"
openaifmt "ds2api/internal/format/openai"
"ds2api/internal/httpapi/openai/shared"
"ds2api/internal/promptcompat"
"ds2api/internal/sse"
streamengine "ds2api/internal/stream"
@@ -36,31 +37,27 @@ type responsesStreamRuntime struct {
toolCallsEmitted bool
toolCallsDoneEmitted bool
sieve toolstream.State
rawThinking strings.Builder
thinking strings.Builder
toolDetectionThinking strings.Builder
rawText strings.Builder
text strings.Builder
visibleText strings.Builder
responseMessageID int
streamToolCallIDs map[int]string
functionItemIDs map[int]string
functionOutputIDs map[int]int
functionArgs map[int]string
functionDone map[int]bool
functionAdded map[int]bool
functionNames map[int]string
messageItemID string
messageOutputID int
nextOutputID int
messageAdded bool
messagePartAdded bool
sequence int
failed bool
finalErrorStatus int
finalErrorMessage string
finalErrorCode string
sieve toolstream.State
accumulator shared.StreamAccumulator
visibleText strings.Builder
responseMessageID int
streamToolCallIDs map[int]string
functionItemIDs map[int]string
functionOutputIDs map[int]int
functionArgs map[int]string
functionDone map[int]bool
functionAdded map[int]bool
functionNames map[int]string
messageItemID string
messageOutputID int
nextOutputID int
messageAdded bool
messagePartAdded bool
sequence int
failed bool
finalErrorStatus int
finalErrorMessage string
finalErrorCode string
persistResponse func(obj map[string]any)
}
@@ -108,6 +105,11 @@ func newResponsesStreamRuntime(
toolChoice: toolChoice,
traceID: traceID,
persistResponse: persistResponse,
accumulator: shared.StreamAccumulator{
ThinkingEnabled: thinkingEnabled,
SearchEnabled: searchEnabled,
StripReferenceMarkers: stripReferenceMarkers,
},
}
}
@@ -155,10 +157,10 @@ func (s *responsesStreamRuntime) finalize(finishReason string, deferEmptyOutput
s.processToolStreamEvents(toolstream.Flush(&s.sieve, s.toolNames), true, true)
}
finalThinking := s.thinking.String()
finalToolDetectionThinking := s.toolDetectionThinking.String()
finalText := cleanVisibleOutput(s.text.String(), s.stripReferenceMarkers)
textParsed := detectAssistantToolCalls(s.rawText.String(), finalText, s.rawThinking.String(), finalToolDetectionThinking, s.toolNames)
finalThinking := s.accumulator.Thinking.String()
finalToolDetectionThinking := s.accumulator.ToolDetectionThinking.String()
finalText := cleanVisibleOutput(s.accumulator.Text.String(), s.stripReferenceMarkers)
textParsed := detectAssistantToolCalls(s.accumulator.RawText.String(), finalText, s.accumulator.RawThinking.String(), finalToolDetectionThinking, s.toolNames)
detected := textParsed.Calls
s.logToolPolicyRejections(textParsed)
@@ -228,62 +230,27 @@ func (s *responsesStreamRuntime) onParsed(parsed sse.LineResult) streamengine.Pa
return streamengine.ParsedDecision{Stop: true}
}
contentSeen := false
batch := responsesDeltaBatch{runtime: s}
for _, p := range parsed.ToolDetectionThinkingParts {
trimmed := sse.TrimContinuationOverlap(s.toolDetectionThinking.String(), p.Text)
if trimmed != "" {
s.toolDetectionThinking.WriteString(trimmed)
}
}
for _, p := range parsed.Parts {
accumulated := s.accumulator.Apply(parsed)
for _, p := range accumulated.Parts {
if p.Type == "thinking" {
rawTrimmed := sse.TrimContinuationOverlap(s.rawThinking.String(), p.Text)
if rawTrimmed != "" {
s.rawThinking.WriteString(rawTrimmed)
contentSeen = true
}
if !s.thinkingEnabled {
continue
}
cleanedText := cleanVisibleOutput(rawTrimmed, s.stripReferenceMarkers)
if cleanedText == "" {
continue
}
trimmed := sse.TrimContinuationOverlap(s.thinking.String(), cleanedText)
if trimmed == "" {
continue
}
s.thinking.WriteString(trimmed)
batch.append("reasoning", trimmed)
batch.append("reasoning", p.VisibleText)
continue
}
rawTrimmed := sse.TrimContinuationOverlap(s.rawText.String(), p.Text)
if rawTrimmed == "" {
if p.RawText == "" {
continue
}
s.rawText.WriteString(rawTrimmed)
contentSeen = true
cleanedText := cleanVisibleOutput(rawTrimmed, s.stripReferenceMarkers)
if s.searchEnabled && sse.IsCitation(cleanedText) {
if p.CitationOnly {
continue
}
trimmed := sse.TrimContinuationOverlap(s.text.String(), cleanedText)
if trimmed != "" {
s.text.WriteString(trimmed)
}
if !s.bufferToolContent {
if trimmed == "" {
continue
}
batch.append("text", trimmed)
batch.append("text", p.VisibleText)
continue
}
batch.flush()
s.processToolStreamEvents(toolstream.ProcessChunk(&s.sieve, rawTrimmed, s.toolNames), true, true)
s.processToolStreamEvents(toolstream.ProcessChunk(&s.sieve, p.RawText, s.toolNames), true, true)
}
batch.flush()
return streamengine.ParsedDecision{ContentSeen: contentSeen}
return streamengine.ParsedDecision{ContentSeen: accumulated.ContentSeen}
}

104
internal/httpapi/openai/shared/stream_accumulator.go Normal file
View File

@@ -0,0 +1,104 @@
package shared
import (
"strings"
"ds2api/internal/sse"
)
type StreamAccumulator struct {
ThinkingEnabled bool
SearchEnabled bool
StripReferenceMarkers bool
RawThinking strings.Builder
Thinking strings.Builder
ToolDetectionThinking strings.Builder
RawText strings.Builder
Text strings.Builder
}
type StreamPartDelta struct {
Type string
RawText string
VisibleText string
CitationOnly bool
}
type StreamAccumulatorResult struct {
ContentSeen bool
Parts []StreamPartDelta
}
func (a *StreamAccumulator) Apply(parsed sse.LineResult) StreamAccumulatorResult {
out := StreamAccumulatorResult{}
for _, p := range parsed.ToolDetectionThinkingParts {
trimmed := sse.TrimContinuationOverlap(a.ToolDetectionThinking.String(), p.Text)
if trimmed != "" {
a.ToolDetectionThinking.WriteString(trimmed)
}
}
for _, p := range parsed.Parts {
if p.Type == "thinking" {
delta := a.applyThinkingPart(p.Text)
if delta.RawText != "" {
out.ContentSeen = true
}
if delta.RawText != "" || delta.VisibleText != "" {
out.Parts = append(out.Parts, delta)
}
continue
}
delta := a.applyTextPart(p.Text)
if delta.RawText != "" {
out.ContentSeen = true
}
if delta.RawText != "" || delta.VisibleText != "" || delta.CitationOnly {
out.Parts = append(out.Parts, delta)
}
}
return out
}
func (a *StreamAccumulator) applyThinkingPart(text string) StreamPartDelta {
rawTrimmed := sse.TrimContinuationOverlap(a.RawThinking.String(), text)
if rawTrimmed != "" {
a.RawThinking.WriteString(rawTrimmed)
}
delta := StreamPartDelta{Type: "thinking", RawText: rawTrimmed}
if !a.ThinkingEnabled || rawTrimmed == "" {
return delta
}
cleanedText := CleanVisibleOutput(rawTrimmed, a.StripReferenceMarkers)
if cleanedText == "" {
return delta
}
trimmed := sse.TrimContinuationOverlap(a.Thinking.String(), cleanedText)
if trimmed == "" {
return delta
}
a.Thinking.WriteString(trimmed)
delta.VisibleText = trimmed
return delta
}
func (a *StreamAccumulator) applyTextPart(text string) StreamPartDelta {
rawTrimmed := sse.TrimContinuationOverlap(a.RawText.String(), text)
if rawTrimmed == "" {
return StreamPartDelta{Type: "text"}
}
a.RawText.WriteString(rawTrimmed)
delta := StreamPartDelta{Type: "text", RawText: rawTrimmed}
cleanedText := CleanVisibleOutput(rawTrimmed, a.StripReferenceMarkers)
if a.SearchEnabled && sse.IsCitation(cleanedText) {
delta.CitationOnly = true
return delta
}
trimmed := sse.TrimContinuationOverlap(a.Text.String(), cleanedText)
if trimmed == "" {
return delta
}
a.Text.WriteString(trimmed)
delta.VisibleText = trimmed
return delta
}

97
internal/httpapi/openai/shared/stream_accumulator_test.go Normal file
View File

@@ -0,0 +1,97 @@
package shared
import (
"testing"
"ds2api/internal/sse"
)
func TestStreamAccumulatorAppliesThinkingAndTextDedupe(t *testing.T) {
acc := StreamAccumulator{ThinkingEnabled: true, StripReferenceMarkers: true}
thinkingPrefix := "this is a long thinking snapshot prefix used by DeepSeek continue replay"
textPrefix := "this is a long visible answer snapshot prefix used by DeepSeek continue replay"
first := acc.Apply(sse.LineResult{
Parsed: true,
Parts: []sse.ContentPart{
{Type: "thinking", Text: thinkingPrefix},
{Type: "text", Text: textPrefix},
},
})
second := acc.Apply(sse.LineResult{
Parsed: true,
Parts: []sse.ContentPart{
{Type: "thinking", Text: thinkingPrefix + " next"},
{Type: "text", Text: textPrefix + " world"},
},
})
if !first.ContentSeen || !second.ContentSeen {
t.Fatalf("expected both chunks to mark content seen")
}
if got := acc.RawThinking.String(); got != thinkingPrefix+" next" {
t.Fatalf("raw thinking = %q", got)
}
if got := acc.Thinking.String(); got != thinkingPrefix+" next" {
t.Fatalf("thinking = %q", got)
}
if got := acc.RawText.String(); got != textPrefix+" world" {
t.Fatalf("raw text = %q", got)
}
if got := acc.Text.String(); got != textPrefix+" world" {
t.Fatalf("text = %q", got)
}
if got := second.Parts[0].VisibleText; got != " next" {
t.Fatalf("thinking delta = %q", got)
}
if got := second.Parts[1].VisibleText; got != " world" {
t.Fatalf("text delta = %q", got)
}
}
func TestStreamAccumulatorKeepsHiddenThinkingForToolDetection(t *testing.T) {
acc := StreamAccumulator{ThinkingEnabled: false, StripReferenceMarkers: true}
result := acc.Apply(sse.LineResult{
Parsed: true,
Parts: []sse.ContentPart{
{Type: "thinking", Text: "<tool_calls></tool_calls>"},
},
ToolDetectionThinkingParts: []sse.ContentPart{
{Type: "thinking", Text: "detect"},
{Type: "thinking", Text: " tools"},
},
})
if !result.ContentSeen {
t.Fatalf("expected hidden thinking to count as upstream content")
}
if got := acc.RawThinking.String(); got != "<tool_calls></tool_calls>" {
t.Fatalf("raw thinking = %q", got)
}
if got := acc.Thinking.String(); got != "" {
t.Fatalf("visible thinking = %q", got)
}
if got := acc.ToolDetectionThinking.String(); got != "detect tools" {
t.Fatalf("tool detection thinking = %q", got)
}
}
func TestStreamAccumulatorSuppressesCitationTextWhenSearchEnabled(t *testing.T) {
acc := StreamAccumulator{SearchEnabled: true, StripReferenceMarkers: true}
result := acc.Apply(sse.LineResult{
Parsed: true,
Parts: []sse.ContentPart{{Type: "text", Text: "[citation:1]"}},
})
if !result.ContentSeen {
t.Fatalf("expected citation chunk to mark upstream content")
}
if len(result.Parts) != 1 || !result.Parts[0].CitationOnly {
t.Fatalf("expected citation-only delta, got %#v", result.Parts)
}
if got := acc.RawText.String(); got != "[citation:1]" {
t.Fatalf("raw text = %q", got)
}
if got := acc.Text.String(); got != "" {
t.Fatalf("visible text = %q", got)
}
}